self.handle_new_sync_comp_message(message, ctx)

                    },

                    Message::SyncPort(message) => {

                        self.handle_new_sync_port_message(message, ctx);

                        None

                    },

                    Message::SyncControl(message) => {

                        self.handle_new_sync_control_message(message, ctx)

                    },

                    Message::Control(_) => unreachable!("control message in component"),

                }

            };

            if let Some(result) = immediate_result {

                return Some(result);

            }

        }

        return None;

    }

    pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {

        // Go through all branches that are awaiting new messages and see if

        // there is one that can receive this message.

            // Message should not be handled now

            return;

        }

        let message = ctx.read_message_using_ticket(ticket).as_data();

        let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);

        while let Some(branch_id) = iter_id {

            iter_id = self.tree.get_queue_next(branch_id);

            let branch = &self.tree[branch_id];

            if branch.awaiting_port != message.data_header.target_port { continue; }

            if !self.consensus.branch_can_receive(branch_id, &message) { continue; }

            // This branch can receive, so fork and given it the message

            let receiving_branch_id = self.tree.fork_branch(branch_id);

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

            let receiving_branch = &mut self.tree[receiving_branch_id];

            debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);

            receiving_branch.awaiting_port = PortIdLocal::new_invalid();

            receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.clone());

            self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

        }

    }

    pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {

        if let Some(round_conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            return Some(self.enter_non_sync_mode(round_conclusion, ctx));

        }

        return None;

    }

    pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {

        self.consensus.handle_new_sync_port_message(message, ctx);

    }

    pub fn handle_new_sync_control_message(&mut self, message: SyncControlMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {

        if let Some(round_conclusion) = self.consensus.handle_new_sync_control_message(message, ctx) {

            return Some(self.enter_non_sync_mode(round_conclusion, ctx));

        }

        return None;

    }

    // --- Running code

    pub fn run_in_sync_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        // Check if we have any branch that needs running

        debug_assert!(self.tree.is_in_sync() && self.consensus.is_in_sync());

    // --- State that is cleared after each round

    // Local component's state

    highest_connector_id: ConnectorId,

    branch_annotations: Vec<BranchAnnotation>, // index is branch ID

    branch_markers: Vec<BranchId>, // index is branch marker, maps to branch

    // Gathered state from communication

    encountered_ports: VecSet<PortIdLocal>, // to determine if we should send "port remains silent" messages.

    solution_combiner: SolutionCombiner,

    handled_wave: bool, // encountered notification wave in this round

    conclusion: Option<RoundConclusion>,

    ack_remaining: u32,

    // --- Persistent state

    peers: Vec<Peer>,

    sync_round: u32,

    // --- Workspaces

    workspace_ports: Vec<PortIdLocal>,

}

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

pub(crate) enum Consistency {

    Valid,

    Inconsistent,

}

pub(crate) enum MessageOrigin {

    Past,

    Present,

    Future

}

impl Consensus {

    pub fn new() -> Self {

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

            branch_markers: Vec::new(),

            encountered_ports: VecSet::new(),

            solution_combiner: SolutionCombiner::new(),

            handled_wave: false,

            conclusion: None,

            ack_remaining: 0,

            peers: Vec::new(),

            sync_round: 0,

            workspace_ports: Vec::new(),

        }

    }

    // --- Controlling sync round and branches

                channel_id,

                port.registered_id.unwrap_or(BranchMarker::new_invalid())

            ));

        }

        let local_solution = LocalSolution{

            component: ctx.id,

            final_branch_id: branch_id,

            port_mapping: target_mapping,

        };

        let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalSolution(local_solution), ctx);

        return maybe_conclusion;

    }

    /// Notifies the consensus algorithm about the chosen branch to commit to

    /// memory (may be the invalid "start" branch)

    pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<ComponentPortChange>) {

        debug_assert!(self.is_in_sync());

        // TODO: Handle sending and receiving ports

        // Set final ports

        let branch = &self.branch_annotations[branch_id.index as usize];

        // Clear out internal storage to defaults

        self.highest_connector_id = ConnectorId::new_invalid();

        self.branch_annotations.clear();

        self.branch_markers.clear();

        self.encountered_ports.clear();

        self.solution_combiner.clear();

        self.handled_wave = false;

        self.conclusion = None;

        self.ack_remaining = 0;

        // And modify persistent storage

        self.sync_round += 1;

        for peer in self.peers.iter_mut() {

            peer.encountered_this_round = false;

            peer.expected_sync_round += 1;

        }

    }

    // --- Handling messages

    /// Prepares a message for sending. Caller should have made sure that

    /// sending the message is consistent with the speculative state.

    pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtx) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

        let annotation = &self.branch_annotations[branch_id.index as usize];

        for expected in &message.data_header.expected_mapping {

            // If we own the port, then we have an entry in the

            // annotation, check if the current mapping matches

            for current in &annotation.channel_mapping {

                if expected.channel_id == current.channel_id {

                    if expected.registered_id != current.registered_id {

                        // IDs do not match, we cannot receive the

                        // message in this branch

                        return false;

                    }

                }

            }

        }

        return true;

    }

    // --- Internal helpers

    fn handle_received_sync_header(&mut self, sync_header: SyncHeader, ctx: &mut ComponentCtx) -> MessageOrigin {

        debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves

        let origin = self.handle_peer(&sync_header);

        if origin != MessageOrigin::Present {

            // We do not have to handle it now

            return origin;

        }

        if sync_header.highest_component_id > self.highest_connector_id {

            // Sender has higher component ID. So should be the target of our

            // messages. We should also let all of our peers know

            self.highest_connector_id = sync_header.highest_component_id;

            for peer in self.peers.iter() {

                if peer.id == sync_header.sending_component_id || !peer.encountered_this_round {

                    // Don't need to send it to this one

                    continue

                }

                let message = SyncCompMessage {

                    sync_header: self.create_sync_header(ctx),

                    target_component_id: peer.id,

                    content: SyncCompContent::Notification,

                };

                ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel

            }

            // But also send our locally combined solution

                // We should exit

                self.debug(" ... No more work, quitting");

                break 'thread_loop;

            }

            // We have something to do

            let connector_key = connector_key.unwrap();

            let connector_id = connector_key.downcast();

            self.debug_conn(connector_id, &format!(" ... Got work, running {}", connector_key.index));

            let scheduled = self.runtime.get_component_private(&connector_key);

            // Keep running until we should no longer immediately schedule the

            // connector.

            let mut cur_schedule = ConnectorScheduling::Immediate;

            while let ConnectorScheduling::Immediate = cur_schedule {

                self.handle_inbox_messages(scheduled);

                // Run the main behaviour of the connector, depending on its

                // current state.

                if scheduled.shutting_down {

                    // Nothing to do. But we're stil waiting for all our pending

                    // control messages to be answered.

                    self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));

                    if scheduled.router.num_pending_acks() == 0 {

                        // We're actually done, we can safely destroy the

                        // currently running connector

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    } else {

                        cur_schedule = ConnectorScheduling::NotNow;

                    }

                } else {

                    self.debug_conn(connector_id, "Running ...");

                    let scheduler_ctx = SchedulerCtx{ runtime: &*self.runtime };

                    let new_schedule = scheduled.connector.run(scheduler_ctx, &mut scheduled.ctx);

                    self.debug_conn(connector_id, &format!("Finished running (new scheduling is {:?})", new_schedule));

                    // Handle all of the output from the current run: messages to

                    // send and connectors to instantiate.

                    self.handle_changes_in_context(scheduled);

                    cur_schedule = new_schedule;

                }

            }

            // If here then the connector does not require immediate execution.

            // So enqueue it if requested, and otherwise put it in a sleeping

                    // Simply queue it again later

                    self.runtime.push_work(connector_key);

                },

                ConnectorScheduling::NotNow => {

                    // Need to sleep, note that we are the only ones which are

                    // allows to set the sleeping state to `true`, and since

                    // we're running it must currently be `false`.

                    self.try_go_to_sleep(connector_key, scheduled);

                },

                ConnectorScheduling::Exit => {

                    // Prepare for exit. Set the shutdown flag and broadcast

                    // messages to notify peers of closing channels

                    scheduled.shutting_down = true;

                    for port in &scheduled.ctx.ports {

                        if port.state != PortState::Closed {

                            let message = scheduled.router.prepare_closing_channel(

                                port.self_id, port.peer_id,

                                connector_id

                            );

                            self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));

                            self.runtime.send_message(port.peer_connector, Message::Control(message));

                        }

                    }

                    if scheduled.router.num_pending_acks() == 0 {

                        // All ports (if any) already closed

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                },

            }

        }

    }

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

    fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx.id;

        while let Some(message) = scheduled.public.inbox.take_message() {

            // Check if the message has to be rerouted because we have moved the

            // target port to another component.

            self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));

            if let Some(target_port) = message.target_port() {

                if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {

                    self.debug_conn(connector_id, " ... Rerouting the message");

                    self.runtime.send_message(other_component_id, message);

                    continue;

                }

            }

            // If here, then we should handle the message

            self.debug_conn(connector_id, " ... Handling the message");

                    scheduled.ctx.inbox.insert_new(message);

                }

            }

        }

    }

        }

    }

    /// Handles changes to the context that were made by the component. This is

    /// the way (due to Rust's borrowing rules) that we bubble up changes in the

    /// component's state that the scheduler needs to know about (e.g. a message

    /// that the component wants to send, a port that has been added).

    fn handle_changes_in_context(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx.id;

        // Handling any messages that were sent

        while let Some(message) = scheduled.ctx.outbox.pop_front() {

            self.debug_conn(connector_id, &format!("Sending message [outbox] \n --- {:#?}", message));

            let target_component_id = match &message {

                Message::Data(content) => {

                    // Data messages are always sent to a particular port, and

                    // may end up being rerouted.

                    let port_desc = scheduled.ctx.get_port_by_id(content.data_header.sending_port).unwrap();

                    debug_assert_eq!(port_desc.peer_id, content.data_header.target_port);

                    if port_desc.state == PortState::Closed {

                        todo!("handle sending over a closed port")

                    }

                ComponentStateChange::CreatedPort(port) => {

                    scheduled.ctx.ports.push(port);

                },

                ComponentStateChange::ChangedPort(port_change) => {

                    if port_change.is_acquired {

                        scheduled.ctx.ports.push(port_change.port);

                    } else {

                        let index = scheduled.ctx.ports

                            .iter()

                            .position(|v| v.self_id == port_change.port.self_id)

                            .unwrap();

                        scheduled.ctx.ports.remove(index);

                    }

                }

            }

        }

        // Finally, check if we just entered or just left a sync region

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                // Just left sync region. So prepare inbox for the next sync

                // round

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

        }

    }

    fn try_go_to_sleep(&self, connector_key: ConnectorKey, connector: &mut ScheduledConnector) {

        debug_assert_eq!(connector_key.index, connector.ctx.id.0);

        debug_assert_eq!(connector.public.sleeping.load(Ordering::Acquire), false);

        // This is the running connector, and only the running connector may

        // decide it wants to sleep again.

        connector.public.sleeping.store(true, Ordering::Release);

        // But due to reordering we might have received messages from peers who

        // did not consider us sleeping. If so, then we wake ourselves again.

        if !connector.public.inbox.is_empty() {

            // Try to wake ourselves up (needed because someone might be trying

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

                .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

                .is_ok();

            if should_wake_up_again {

                // Unreachable because:

                //  1. We only iterate over messages that were previously retrieved by `read_next_message`.

                //  2. Inbox does not contain control/ping messages.

                //  3. If `read_next_message` encounters anything else than a data message, it is removed from the inbox.

                unreachable!();

            }

            self.next_index += 1;

        }

        // No more messages

        return None;

    }

}

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

// Private Inbox

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

/// A structure that contains inbox messages. Some messages are left inside and

/// continuously re-read. Others are taken out, but may potentially be put back

/// for later reading. Later reading in this case implies that they are put back

/// for reading in the next sync round.

struct Inbox {

    messages: RawVec<Message>,

    next_delay_idx: u32,

    start_read_idx: u32,

    next_read_idx: u32,

    generation: u32,

}

#[derive(Clone, Copy)]

pub(crate) struct MessageTicket {

    index: u32,

    generation: u32,

}

impl Inbox {

    fn new() -> Self {

        return Inbox {

            messages: RawVec::new(),

            next_delay_idx: 0,

            start_read_idx: 0,

            next_read_idx: 0,

            generation: 0,

        }

    }

    fn insert_new(&mut self, message: Message) {

        assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size

        self.messages.push(message);

    }

    fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

        let cur_read_idx = self.next_read_idx as usize;

        if cur_read_idx >= self.messages.len() {

            return None;

        }

        self.generation += 1;

        self.next_read_idx += 1;

        return Some(MessageTicket{

            index: cur_read_idx as u32,

            generation: self.generation

        });

    }

    fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {

        debug_assert_eq!(self.generation, ticket.generation);

        return unsafe{ &*self.messages.get(ticket.index as usize) }

    }

    fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {

        debug_assert_eq!(self.generation, ticket.generation);

        unsafe {

            let take_idx = ticket.index as usize;

            let val = std::ptr::read(self.messages.get(take_idx));

            // Move messages to the right, clearing up space in the

            // front.

            let num_move_right = take_idx - self.start_read_idx as usize;

            self.messages.move_range(

                self.start_read_idx as usize,

                self.start_read_idx as usize + 1,

                num_move_right

            );

            self.start_read_idx += 1;

            return val;

        }

    }

    fn put_back_message(&mut self, message: Message) {

        // We have space in front of the array because we've taken out a message

        // before.

        debug_assert!(self.next_delay_idx < self.start_read_idx);

        unsafe {

            // Write to front of the array

            std::ptr::write(self.messages.get_mut(self.next_delay_idx as usize), message);

            self.next_delay_idx += 1;

        }

    }

    fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {

        return MessagesIter{

            messages: self.messages.as_slice(),

            next_index: self.start_read_idx as usize,

            max_index: self.next_read_idx as usize,

            match_port_id

        };

    }

        // Deallocate everything that was read

        self.destroy_range(self.start_read_idx, self.next_read_idx);

        self.generation += 1;

        // Join up all remaining values with the delayed ones in the front

        let num_to_move = self.messages.len() - self.next_read_idx as usize;

        self.messages.move_range(

            self.next_read_idx as usize,

            self.next_delay_idx as usize,

            num_to_move

        );

        // Set all indices (and the RawVec len) to make sense in this new state

        let new_len = self.next_delay_idx as usize + num_to_move;

        self.next_delay_idx = 0;

        self.start_read_idx = 0;

        self.next_read_idx = 0;

        self.messages.len = new_len;

    }

    fn transfer_messages_for_port(&mut self, port: PortIdLocal, new_inbox: &mut Inbox) {

        let mut idx = 0;

        while idx < self.messages.len() {

            let message = unsafe{ &*self.messages.get(idx) };

            if let Some(target_port) = message.target_port() {

                if target_port == port {

                    // Transfer port

                    unsafe {

                        let message = std::ptr::read(message as *const _);

                        }

                        self.messages.len -= 1;

                        new_inbox.insert_new(message);

                    }

                }

            }

        }

    }

    #[inline]

    fn destroy_range(&mut self, start_idx: u32, end_idx: u32) {

        for idx in (start_idx as usize)..(end_idx as usize) {

            unsafe {

                let msg = self.messages.get_mut(idx);

                std::ptr::drop_in_place(msg);

            }

        }

    }

}

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

// Control messages

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

struct ControlEntry {

    id: u32,

    variant: ControlVariant,

}

enum ControlVariant {

    ChangedPort(ControlChangedPort),

    ClosedChannel(ControlClosedChannel),

    ReroutePending,

}

struct ControlChangedPort {

    target_port: PortIdLocal,       // if send to this port, then reroute

    source_connector: ConnectorId,  // connector we expect messages from

    target_connector: ConnectorId,  // connector we need to reroute to

    id_of_ack_after_confirmation: u32, // control message ID we need to send to the target upon receiving an ack

}

struct ControlClosedChannel {

mod network_shapes;

mod api_component;

mod speculation;

mod data_transmission;

mod sync_failure;

use super::*;

use crate::{PortId, ProtocolDescription};

use crate::common::Id;

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

use crate::runtime2::native::{ApplicationSyncAction};

// Generic testing constants, use when appropriate to simplify stress-testing

// pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime

// pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed

// pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)