However, there are some properties that we can take advantage of:

1. When a component sends a message, it knows for certain what its own component ID and port ID is. So a transmitting port always sends the correct source component/port ID.

2. When a component receives a message, it knows for certain what its own component ID and port ID is. So once a receiving port receives a properly annotated message from a transmitted port, the receiving end can be certain about the component IDs and port IDs that make up the channel.

Note that although the message transmitter may not be certain about its final recipient, the components along the way *are* aware of the routing that is necessary to make the message arrive at the intended target. Combing back to the case where we have a creator `C`, new component `N` and peer `P`. Then `P` will send a message intended for `N`, but arriving at `C`. Here `C` can change the target port and component to `N` (as it is in the process of transferring that port, so knows both its original and new port ID). Once the message arrives and is accepted by the recipient then it is certain about the component and port IDs.

## Sending Sync Messages

Sync messages have the purpose of making sure that consensus is reached about the interactions that took place in all of the participating components' sync blocks. The previous runtime featured speculative execution and a branching execution model: a component could exhibit multiple behaviours, and at the end all components decide which combination of local behaviours constitute a satisfying single global behaviour (if one exists at all). Without speculative execution the model can be a lot simpler.

We'll only shortly discuss the mechanisms that are present in the synchronization algorithm. A component has a local counter, that is reset for each synchronous interaction, that is used when transmitting data messages. Such a message will be annotated with the counter at `N`, after which the component sends the next message with annotation `N+1`. At the same time the component will keep track of a local mapping from port ID to port annotation, we'll call this the port mapping. Likewise, when a component receives a data message it will assign the received annotation in its own port mapping. If two components assign the same annotation to the ports that constitute a channel, then there is an agreeable interaction there.

And so at the end of the synchronous round a component will somehow broadcast its port mapping. Note from the discussion above that a transmitting port's annotation is only associated with that transmitting port, since a transmitting port can only truly ever know its own component/port ID. While the receiving port's annotation knows about the peer's component/port ID as well. And so a component can broadcast `(component ID, port ID, mapping)` for each of its transmitting ports, while it can broadcast `(own component ID, own port ID, peer component ID, peer port ID, mapping)` for each receiving port. Then a recipient of these mappings can match them up and make sure that the mappings agree.

Note that this broadcasting of synchronous messages is essentially a component-to-component operation. However these messages must still be sent over ports anyway (and any port that was used to transmit messages to a particular receiving component will do). There are two reasons:

1. The sync message may need to be rerouted (e.g. a sender quickly fires both a data message and a subsequent sync message while the receiving port is being transferred to a new component), but needs to arrive at the same target as the data message. This is essentially restating that a transmitter never knows about the component ID of the recipient.

2. The sync message must not be taken into account by the recipient if it has not accepted any messages from the sender yet. Ofcourse this can be achieved in various ways but a simple way to achieve this is to send the sync message over ports.

## Annotating Data Messages

These port mappings are also sent along when sending data messages. We will not go into details but here the mapping makes sure that messages arrive in the right order, and certain kinds of deadlock or inconsistent protocol behaviour may be detected. This port mapping is checked for consistency by the recipient and, when consistent, the target port is updated with its new mapping.

use crate::protocol::eval::*;

use super::runtime::*;

use super::component::*;

// -----------------------------------------------------------------------------

// Generic types

// -----------------------------------------------------------------------------

#[derive(Debug, Copy, Clone, PartialEq, Eq)]

pub struct PortId(pub u32);

impl PortId {

    /// This value is not significant, it is chosen to make debugging easier: a

    /// very large port number is more likely to shine a light on bugs.

    pub fn new_invalid() -> Self {

        return Self(u32::MAX);

    }

}

#[derive(Debug, PartialEq, Eq, Clone, Copy)]

pub enum PortKind {

    Putter,

    Getter,

}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]

pub enum PortState {

    Open,

    BlockedDueToPeerChange,

    BlockedDueToFullBuffers,

    Closed,

}

impl PortState {

    pub fn is_blocked(&self) -> bool {

        match self {

            PortState::BlockedDueToPeerChange | PortState::BlockedDueToFullBuffers => true,

            PortState::Open | PortState::Closed => false,

        }

    }

}

pub struct Channel {

    pub putter_id: PortId,

    pub getter_id: PortId,

}

// -----------------------------------------------------------------------------

// Data messages

// -----------------------------------------------------------------------------

#[derive(Debug)]

pub struct DataMessage {

    pub data_header: MessageDataHeader,

    pub sync_header: MessageSyncHeader,

    pub content: ValueGroup,

}

#[derive(Debug)]

pub struct MessageDataHeader {

    pub new_mapping: u32,

    pub source_port: PortId,

    pub target_port: PortId,

}

// -----------------------------------------------------------------------------

// Sync messages

// -----------------------------------------------------------------------------

#[derive(Debug)]

pub struct SyncMessage {

    pub sync_header: MessageSyncHeader,

    pub content: SyncMessageContent,

}

#[derive(Debug)]

pub enum SyncLocalSolutionEntry {

    Putter(SyncSolutionPutterPort),

    Getter(SyncSolutionGetterPort),

}

pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;

/// Getter port in a solution. Upon receiving a message it is certain about who

use crate::protocol::eval::ValueGroup;

use crate::runtime2::scheduler::*;

use crate::runtime2::runtime::*;

use crate::runtime2::communication::*;

use super::component_context::*;

pub struct PortAnnotation {

    self_comp_id: CompId,

    self_port_id: PortId,

    peer_comp_id: CompId, // only valid for getter ports

    peer_port_id: PortId, // only valid for getter ports

    mapping: Option<u32>,

}

impl PortAnnotation {

        return Self{

            self_comp_id: comp_id,

            self_port_id: port_id,

            peer_comp_id: CompId::new_invalid(),

            peer_port_id: PortId::new_invalid(),

        }

    }

}

#[derive(Debug, Eq, PartialEq)]

enum Mode {

    NonSync,

    SyncBusy,

    SyncAwaitingSolution,

    SelectBusy,

    SelectWait,

}

struct SolutionCombiner {

    solution: SyncPartialSolution,

    matched_channels: usize,

}

impl SolutionCombiner {

    fn new() -> Self {

        return Self {

            solution: SyncPartialSolution::default(),

            matched_channels: 0,

        }

        }

        self.solution.clear();

    }

    fn make_ports_consistent_with_ctx(&mut self, comp_ctx: &CompCtx) {

        let mut needs_setting_ports = false;

        if comp_ctx.num_ports() != self.ports.len() {

            needs_setting_ports = true;

        } else {

            for (idx, port) in comp_ctx.iter_ports().enumerate() {

                let comp_port_id = port.self_id;

                let cons_port_id = self.ports[idx].self_port_id;

                if comp_port_id != cons_port_id {

                    needs_setting_ports = true;

                    break;

                }

            }

        }

        if needs_setting_ports {

            self.ports.clear();

            self.ports.reserve(comp_ctx.num_ports());

            for port in comp_ctx.iter_ports() {

            }

        }

    }

    // -------------------------------------------------------------------------

    // Handling inbound and outbound messages

    // -------------------------------------------------------------------------

    pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {

        debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == port_info.self_id));

        let data_header = self.create_data_header_and_update_mapping(port_info);

        let sync_header = self.create_sync_header(comp_ctx);

        return DataMessage{ data_header, sync_header, content };

    }

    /// Checks if the data message can be received (due to port annotations), if

    /// it can then `true` is returned and the caller is responsible for handing

    /// the message of to the PDL code. Otherwise the message cannot be

    /// received.

    pub(crate) fn try_receive_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: &DataMessage) -> bool {

        debug_assert_eq!(self.mode, Mode::SyncBusy);

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == message.data_header.target_port));

            let _should_remove = handle.decrement_users();

            debug_assert!(_should_remove.is_none());

        }

    }

    fn send_to_leader(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, message: Message) {

        debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader

        let mut leader_info = sched_ctx.runtime.get_component_public(self.highest_id);

        leader_info.send_message(sched_ctx, message, true);

        let should_remove = leader_info.decrement_users();

        if let Some(key) = should_remove {

            sched_ctx.runtime.destroy_component(key);

        }

    }

    // -------------------------------------------------------------------------

    // Creating message headers

    // -------------------------------------------------------------------------

    fn create_data_header_and_update_mapping(&mut self, port_info: &Port) -> MessageDataHeader {

        let mut expected_mapping = Vec::with_capacity(self.ports.len());

        let mut port_index = usize::MAX;

        for (index, port) in self.ports.iter().enumerate() {

            if port.self_port_id == port_info.self_id {

            }

            expected_mapping.push((port.self_port_id, port.mapping));

        }

        let new_mapping = self.take_mapping();

        self.ports[port_index].mapping = Some(new_mapping);

        debug_assert_eq!(port_info.kind, PortKind::Putter);

        return MessageDataHeader{

            expected_mapping,

            new_mapping,

            source_port: port_info.self_id,

            target_port: port_info.peer_port_id,

        };

    }

    #[inline]

    fn create_sync_header(&self, comp_ctx: &CompCtx) -> MessageSyncHeader {

        return MessageSyncHeader{

            sync_round: self.round_index,

            sending_id: comp_ctx.id,

            highest_id: self.highest_id,

        };

    }