Bringing these systems together: apart from data messages there might still be control messages in transit, or the exiting component might still have some control/sync work to do. And so we need to modify something we said earlier: instead of a component removing its self-reference the moment it terminates, we do this the moment we have received all the `Acknowledge` messages we were expecting. This way if a component is busy creating another component, we're certain that the appropriate protocols are observed. So:

Concluding everything described above, two separate mechanisms will act in conjunction:

- A. The exiting component `E` waiting until it has finished notifying all peers `P` of its termination:

  1. Component `E` sends a `ClosePort` message to each peer `P` for each of the shared channels (except when the port is already closed because `P` is also in the process of shutting down).

  2. The peer `P` receives the `ClosePort` message and closes the port as a result. This is a change of port state that will cause any subsequent `put` operations on that port to become a fatal error for component `P`. In response the peer `P` sends an `Acknowledge` message to component `E`, unless component `P` exited itself (that is: sending a `ClosePort` message to `E` before the `ClosePort` message from `P` arrived). After closing the port the component `P` will remove the reference to `E`.

  3. Component `E` waits until all of its pending control operations are finished (i.e. waiting for the `Acknowledge` messages following `ClosePort` messages, `PeerChangeBlockPort` messages, etc.). Once all of these are finished, note that we can no longer participate in any future control actions: component `E` will not create channels/components itself. Since all of its ports are closed, the peers `P` will also not send any data or control messages.

  4. Component `E` now checks its inbox for any remaining messages. It will respond to any data messages that arrived after `E` sending `ClosePort` and before `E` receiving `Acknowledge` with a `DataMessageFailed` message (except for ports that were closed before the `Acknowledge`). Then it removes the reference to itself, therefore decrementing the reference counter for the component by 1.

- B. The reference counting mechanism. Any sync round the exiting component `E` is participating in will conceptually hold a reference to `E`. The component `E` will always respond to sync messages as if it were alive (albeit trying to indicate to everyone that it is actually exiting). The component that removes the last reference to the component `E` (which may be `E` itself, but also a peer `P`) will truly remove the associated memory from the runtime. 

**Note**: We did not consider machine termination. That is to say: once we reach the runtime maturity where communication occurs over different machines, then we have to consider that machines encounter fatal errors. However these can only be resolved by embedding the possibility of failure inside the protocol over which these machines communicate. 

## Sending Data Messages

So far we've discussed the following properties associated with sending data messages:

1. Port IDs are decided locally. So a peer may have an ID that is outdated for the intended recipient.

2. Ports can move from owner to owner. So a peer might have a component ID that is outdated for the intended recipient.

3. Ports may be closed.

4. Message intended for specific ports may end up at an intermediate component that is passing that message along.

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

/// its peer is.

#[derive(Debug)]

pub struct SyncSolutionGetterPort {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

    pub peer_comp_id: CompId,

    pub peer_port_id: PortId,

    pub mapping: u32,

}

/// Putter port in a solution. A putter may not be certain about who its peer

/// component/port is.

#[derive(Debug)]

pub struct SyncSolutionPutterPort {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

    pub mapping: u32,

}

#[derive(Debug)]

pub struct SyncSolutionChannel {

    pub putter: Option<SyncSolutionPutterPort>,

    pub getter: Option<SyncSolutionGetterPort>,

}

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,

        }

    }

    #[inline]

    fn has_contributions(&self) -> bool {

        return !self.solution.channel_mapping.is_empty();

    }

    /// Returns a decision for the current round. If there is no decision (yet)

    /// then `RoundDecision::None` is returned.

    fn get_decision(&self) -> SyncRoundDecision {

        if self.matched_channels == self.solution.channel_mapping.len() {

            debug_assert_ne!(self.solution.decision, SyncRoundDecision::None);

            return self.solution.decision;

        }

        return SyncRoundDecision::None; // even in case of failure: wait for everyone.

    }

    fn combine_with_partial_solution(&mut self, partial: SyncPartialSolution) {

        debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);

        debug_assert_ne!(partial.decision, SyncRoundDecision::Solution);

        if partial.decision == SyncRoundDecision::Failure {

            self.solution.decision = SyncRoundDecision::Failure;

                        peer_comp_id: port.peer_comp_id,

                        peer_port_id: port.peer_port_id,

                        mapping

                    })

                };

                local_solution.push(new_entry);

            }

        }

        let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution);

        return decision;

    }

    /// Notifies that a decision has been reached. Note that the caller should

    /// still take the appropriate actions based on the decision it is supplying

    /// to the consensus layer.

    pub(crate) fn notify_sync_decision(&mut self, _decision: SyncRoundDecision) {

        // Reset everything for the next round

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

        self.mode = Mode::NonSync;

        self.round_index = self.round_index.wrapping_add(1);

        for port in self.ports.iter_mut() {

            port.mapping = None;

        }

        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));

        // Make sure the expected mapping matches the currently stored mapping

        for (expected_id, expected_annotation) in &message.data_header.expected_mapping {

            let got_annotation = self.get_annotation(*expected_id);

            if got_annotation != *expected_annotation {

                return false;

            }

        }

        // Expected mapping matches current mapping, so we will receive the message

        self.set_annotation(message.sync_header.sending_id, &message.data_header);

        // Handle the sync header embedded within the data message

        self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);

        return true;

    }

    /// Receives the sync message and updates the consensus state appropriately.

    pub(crate) fn receive_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> SyncRoundDecision {

        // Whatever happens: handle the sync header (possibly changing the

        // currently registered leader)

        self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);

            SyncRoundDecision::None => unreachable!(),

            SyncRoundDecision::Solution => true,

            SyncRoundDecision::Failure => false,

        };

        let mut peers = Vec::with_capacity(self.solution.solution.channel_mapping.len()); // TODO: @Performance

        for channel in self.solution.solution.channel_mapping.iter() {

            let getter = channel.getter.as_ref().unwrap();

            if getter.self_comp_id != comp_ctx.id && !peers.contains(&getter.self_comp_id) {

                peers.push(getter.self_comp_id);

            }

            if getter.peer_comp_id != comp_ctx.id && !peers.contains(&getter.peer_comp_id) {

                peers.push(getter.peer_comp_id);

            }

        }

        for peer in peers {

            let mut handle = sched_ctx.runtime.get_component_public(peer);

            let message = Message::Sync(SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: if is_success { SyncMessageContent::GlobalSolution } else { SyncMessageContent::GlobalFailure },

            });

            handle.send_message(sched_ctx, message, true);

            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,

        };

    }

    #[inline]

    fn take_mapping(&mut self) -> u32 {

        let mapping = self.mapping_counter;

        self.mapping_counter = self.mapping_counter.wrapping_add(1);

        return mapping;

    }

}