class Syskit::NetworkGeneration::DataFlowDynamics

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 157
def self.compute_connection_policies(plan)
    engine = DataFlowDynamics.new(plan)
    engine.compute_connection_policies
    engine.result
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 152
def initialize(plan)
    @plan = plan
    super()
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 211
def add_port_trigger(task, port_name, name, period, burst)
    if has_information_for_port?(task, port_name)
        @result[task][port_name].add_trigger(name, period, burst)
    else
        info = create_port_info(task, port_name)
        info.add_trigger(name, period, burst)
        info
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 184
def add_task_info(task, info)
    add_port_info(task, nil, info)
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 180
def add_task_trigger(task, name, period, burst)
    add_port_trigger(task, nil, name, period, burst)
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 491
def compute_connection_policies
    # We only act on deployed tasks, as we need to know how the
    # tasks are triggered (what activity / priority / ...)
    deployed_tasks = plan.find_local_tasks(TaskContext).
        find_all(&:execution_agent)

    propagate(deployed_tasks)

    DataFlowDynamics.debug do
        DataFlowDynamics.debug "computing connections"
        deployed_tasks.each do |t|
            DataFlowDynamics.debug "  #{t}"
        end

        DataFlowDynamics.debug "available information for"
        result.each do |task, ports|
            DataFlowDynamics.debug "  #{task}: #{ports.keys.join(", ")}"
        end
        break
    end

    deployed_tasks.each do |source_task|
        source_task.each_concrete_output_connection do |source_port_name, sink_port_name, sink_task, policy|
            new_policy = policy_for(source_task, source_port_name, sink_port_name, sink_task, policy)
            policy.merge!(new_policy)
            # TODO: Announce that the policy changed to the relation
            # management code
        end
    end
    result
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 433
def compute_info_for(task, port_name)
    triggers = @triggers[[task, port_name]].map do |trigger_task, trigger_port|
        if has_final_information_for_port?(trigger_task, trigger_port)
            port_info(trigger_task, trigger_port)
        else
            DataFlowDynamics.debug do
                DataFlowDynamics.debug "  missing info on "                                 "#{trigger_task}.#{trigger_port} to compute "                                 "#{task}.#{port_name}"
                break
            end
            return false
        end
    end

    if (period = find_period_of(task))
        triggers = triggers.map do |trigger_info|
            trigger_info.sampled_at(period)
        end
    end

    triggers.each do |trigger_info|
        add_port_info(task, port_name, trigger_info)
    end
    done_port_info(task, port_name)
    return true
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 202
def create_port_info(task, port_name)
    port_model = task.model.find_port(port_name)
    dynamics = PortDynamics.new("#{task.orocos_name}.#{port_model.name}",
                            port_model.sample_size)
    dynamics.add_trigger("burst", port_model.burst_period, port_model.burst_size)
    set_port_info(task, port_name, dynamics)
    dynamics
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 192
def done_task_info(task)
    task.orogen_model.slaves.each do |slave_task|
        if slave_task = task_from_name[slave_task.name]
            add_task_info(slave_task, task_info(task))
            done_task_info(slave_task)
        end
    end
    done_port_info(task, nil)
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 420
def find_period_of(task)
    orogen_model = task.orogen_model
    while (master = orogen_model.master)
        orogen_model = master
    end
    if orogen_model.activity_type.name == "Periodic"
        orogen_model.period
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 176
def has_final_information_for_task?(task)
    has_final_information_for_port?(task, nil)
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 172
def has_information_for_task?(task)
    has_information_for_port?(task, nil)
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 296
def initial_combus_information(task)
    handled_ports = Set.new
    task.each_attached_device do |dev|
        srv = task.find_data_service(dev.name)
        srv.each_input_port do |port|
            handled_ports << port.name
            port = port.to_component_port
            dynamics = PortDynamics.new("#{task.orocos_name}.#{port.name}", dev.sample_size)
            if dev.period
                dynamics.add_trigger(dev.name, dev.period, 1)
                dynamics.add_trigger(dev.name, dev.period * dev.burst, dev.burst)
            end
            add_port_info(task, port.name, dynamics)
        end
    end
    handled_ports.each do |port_name|
        done_port_info(task, port_name)
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 222
def initial_device_information(task)
    triggering_devices = task.model.each_master_driver_service.map do |srv|
        [srv, task.find_device_attached_to(srv)]
    end
    DataFlowDynamics.debug do
        DataFlowDynamics.debug "initial port dynamics on #{task} (device)"
        DataFlowDynamics.debug "  attached devices: #{triggering_devices.map { |srv, dev| "#{dev.name} on #{srv.name}" }.join(", ")}"
        break
    end

    activity_type = task.orogen_model.activity_type.name
    case activity_type
    when "Periodic"
        initial_device_information_periodic_triggering(
            task, triggering_devices.to_a, task.orogen_model.period)
    else
        initial_device_information_internal_triggering(
            task, triggering_devices.to_a)
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 247
def initial_device_information_common(task, triggering_devices)
    triggering_devices.each do |service, device|
        DataFlowDynamics.debug { "  #{device.name}: #{device.period} #{device.burst}" }
        device_dynamics = PortDynamics.new(device.name, 1)
        if device.period
            device_dynamics.add_trigger(device.name, device.period, 1)
        end
        device_dynamics.add_trigger(device.name + "-burst", 0, device.burst)

        if !device_dynamics.empty?
            yield(service, device, device_dynamics)
        end
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 264
def initial_device_information_internal_triggering(task, triggering_devices)
    DataFlowDynamics.debug "  is triggered internally"

    initial_device_information_common(task, triggering_devices) do |service, device, device_dynamics|
        add_task_info(task, device_dynamics)
        service.each_output_port do |out_port|
            out_port = out_port.to_component_port
            out_port.orogen_model.triggered_on_update = false
            add_port_info(task, out_port.name, device_dynamics)
            done_port_info(task, out_port.name)
        end
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 280
def initial_device_information_periodic_triggering(task, triggering_devices, period)
    DataFlowDynamics.debug { "  is triggered with a period of #{period} seconds" }

    initial_device_information_common(task, triggering_devices) do |service, device, device_dynamics|
        service.each_output_port do |out_port|
            out_port = out_port.to_component_port
            out_port.orogen_model.triggered_on_update = false
            add_port_trigger(task, out_port.name,
                device.name, period, device_dynamics.queue_size(period))
            done_port_info(task, out_port.name)
        end
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 318
def initial_information(task)
    # Master tasks resolve their children recursively, so we need
    # to guard against initial_information being called twice for
    # the slaves
    #
    # The recursive call is required in order to make sure that
    # we have resolved the info *before* the call done_task_info
    # (which forbids any change later on)
    return if has_final_information_for_task?(task)

    task.orogen_model.slaves.each do |orogen_slave_task|
        if slave_task = task_from_name[orogen_slave_task.name]
            if !has_information_for_task?(slave_task)
                initial_information(slave_task)
            end
        end
    end

    set_port_info(task, nil, PortDynamics.new("#{task.orocos_name}.main"))
    task.model.each_output_port do |port|
        create_port_info(task, port.name)
    end

    add_task_info(task, task.requirements.dynamics.task)
    task.requirements.dynamics.ports.each do |port_name, dynamics|
        add_port_info(task, port_name, dynamics)
        done_port_info(task, port_name)
    end

    if task.kind_of?(Device)
        initial_device_information(task)
    end
    if task.kind_of?(ComBus)
        initial_combus_information(task)
    end

    activity_type = task.orogen_model.activity_type.name
    if activity_type == "Periodic"
        DataFlowDynamics.debug { "  adding periodic trigger #{task.orogen_model.period} 1" }
        add_task_trigger(task, "#{task.orocos_name}.main-period", task.orogen_model.period, 1)
        done_task_info(task)

    elsif activity_type == "SlaveActivity"
        # The master's main trigger is propagated in #done_task_info

    elsif !task.model.each_event_port.find { true }
        done_task_info(task)
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 529
def policy_for(source_task, source_port_name, sink_port_name, sink_task, policy)
    policy = policy.dup
    fallback_policy = policy.delete(:fallback_policy)

    # Don't do anything if the policy has already been set
    if !policy.empty?
        DataFlowDynamics.debug " #{source_task}:#{source_port_name} => #{sink_task}:#{sink_port_name} already connected with #{policy}"
        return policy
    end

    source_port = source_task.model.find_output_port(source_port_name)
    sink_port   = sink_task.model.find_input_port(sink_port_name)
    if !source_port
        raise InternalError, "#{source_port_name} is not a port of #{source_task.model}"
    elsif !sink_port
        raise InternalError, "#{sink_port_name} is not a port of #{sink_task.model}"
    end
    DataFlowDynamics.debug { "   #{source_task}:#{source_port.name} => #{sink_task}:#{sink_port.name}" }

    if !sink_port.needs_reliable_connection?
        if sink_port.required_connection_type == :data
            policy = Orocos::Port.prepare_policy(:type => :data)
            DataFlowDynamics.debug { "     result: #{policy}" }
            return policy
        elsif sink_port.required_connection_type == :buffer
            policy = Orocos::Port.prepare_policy(:type => :buffer, :size => 1)
            DataFlowDynamics.debug { "     result: #{policy}" }
            return policy
        end
    end

    # Compute the buffer size
    input_dynamics =
        if has_final_information_for_port?(source_task, source_port.name)
            port_info(source_task, source_port.name)
        end

    sink_task_dynamics  =
        if has_final_information_for_task?(sink_task)
            task_info(sink_task)
        end

    reading_latency =
        if sink_port.trigger_port?
            sink_task.trigger_latency
        elsif sink_task_dynamics && sink_task_dynamics.minimal_period
            sink_task_dynamics.minimal_period + sink_task.trigger_latency
        end

    if !input_dynamics || !reading_latency
        if fallback_policy
            if !input_dynamics
                DataFlowDynamics.warn do
                    DataFlowDynamics.warn "Cannot compute the period information for the output port"
                    DataFlowDynamics.warn "   #{source_task}:#{source_port.name}"
                    DataFlowDynamics.warn "   This is needed to compute the policy to connect to"
                    DataFlowDynamics.warn "   #{sink_task}:#{sink_port_name}"
                    DataFlowDynamics.warn "   The fallback policy #{fallback_policy} will be used"
                    break
                end

            else
                DataFlowDynamics.warn "#{sink_task} has no minimal period"
                DataFlowDynamics.warn "This is needed to compute the reading latency on #{sink_port.name}"
                DataFlowDynamics.warn "The fallback policy #{fallback_policy} will be used"
            end
            policy = fallback_policy
        elsif !input_dynamics
            raise SpecError, "the period information for output port #{source_task}:#{source_port.name} cannot be computed. This is needed to compute the policy to connect to #{sink_task}:#{sink_port_name}"
        else
            raise SpecError, "#{sink_task} has no minimal period, needed to compute reading latency on #{sink_port.name}"
        end
    else
        policy[:type] = :buffer
        size = (1.0 + Syskit.conf.buffer_size_margin) * input_dynamics.queue_size(reading_latency)
        policy[:size] = Integer(size) + 1
        DataFlowDynamics.debug do
            DataFlowDynamics.debug "     input_period:#{input_dynamics.minimal_period} => reading_latency:#{reading_latency}"
            DataFlowDynamics.debug "     sample_size:#{input_dynamics.sample_size}"
            input_dynamics.triggers.each do |tr|
                DataFlowDynamics.debug "     trigger(#{tr.name}): period=#{tr.period} count=#{tr.sample_count}"
            end
            break
        end
        DataFlowDynamics.debug { "     result: #{policy}" }
    end

    policy
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 461
def propagate_task(task)
    if !missing_ports.has_key?(task)
        return true
    end

    done = true
    required = missing_ports[task].dup
    DataFlowDynamics.debug do
        DataFlowDynamics.debug "trying to compute dataflow dynamics for #{task}"
        DataFlowDynamics.debug "  requires information on: #{required.map(&:to_s).join(", ")}"
        break
    end

    required.each do |missing|
        if !compute_info_for(task, missing)
            DataFlowDynamics.debug do
                DataFlowDynamics.debug "  cannot compute information on #{missing}"
                break
            end
            done = false
        end
    end
    done
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 408
def required_information(tasks)
    result = Hash.new
    tasks.each do |t|
        ports = t.model.each_output_port.to_a
        if !ports.empty?
            result[t] = ports.map(&:name).to_set
            result[t] << nil
        end
    end
    result
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 163
def reset(tasks = Array.new)
    super
    @triggers = Hash.new { |h, k| h[k] = Set.new }
    @task_from_name = Hash.new
    tasks.each do |t|
        task_from_name[t.orocos_name] = t
    end
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 188
def task_info(task)
    port_info(task, nil)
end

# File lib/syskit/network_generation/dataflow_dynamics.rb, line 370
def triggering_inputs(task)
    all_triggers = Set.new
    @triggers[[task, nil]] = Set.new
    task.model.each_event_port do |port|
        if task.has_concrete_input_connection?(port.name)
            all_triggers << port
            @triggers[[task, nil]] << [task, port.name]
        end
    end
    task.model.each_output_port do |port|
        if port.triggered_on_update?
            @triggers[[task, port.name]] << [task, nil]
        end
        port.port_triggers.each do |trigger_port|
            if task.has_concrete_input_connection?(trigger_port.name)
                @triggers[[task, port.name]] << [task, trigger_port.name]
                all_triggers << trigger_port
            end
        end
    end
    task.model.each_output_port do |port|
        if !@triggers.has_key?([task, port.name])
            done_port_info(task, port.name)
        end
    end

    all_triggers
end