// connector.rs

//

// Represents a component. A component (and the scheduler that is running it)

// has many properties that are not easy to subdivide into aspects that are

// conceptually handled by particular data structures. That is to say: the code

// that we run governs: running PDL code, keeping track of ports, instantiating

// new components and transports (i.e. interacting with the runtime), running

// a consensus algorithm, etc. But on the other hand, our data is rather

// simple: we have a speculative execution tree, a set of ports that we own,

// and a bit of code that we should run.

//

// So currently the code is organized as following:

// - The scheduler that is running the component is the authoritative source on

//     ports during *non-sync* mode. The consensus algorithm is the

//     authoritative source during *sync* mode. They retrieve each other's

//     state during the transitions. Hence port data exists duplicated between

//     these two datastructures.

// - The execution tree is where executed branches reside. But the execution

//     tree is only aware of the tree shape itself (and keeps track of some

//     queues of branches that are in a particular state), and tends to store

//     the PDL program state. The consensus algorithm is also somewhat aware

//     of the execution tree, but only in terms of what is needed to complete

//     a sync round (for now, that means the port mapping in each branch).

//     Hence once more we have properties conceptually associated with branches

//     in two places.

// - TODO: Write about handling messages, consensus wrapping data

// - TODO: Write about way information is exchanged between PDL/component and scheduler through ctx

use std::collections::HashMap;

use std::sync::atomic::AtomicBool;

use crate::{PortId, ProtocolDescription};

use crate::common::ComponentState;

use crate::protocol::eval::{EvalContinuation, EvalError, Prompt, Value, ValueGroup};

use crate::protocol::{RunContext, RunResult};

use super::native::Connector;

use super::port::{PortKind, PortIdLocal};

use super::scheduler::{ComponentCtx, SchedulerCtx};

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

}

impl ConnectorPublic {

    pub fn new(initialize_as_sleeping: bool) -> Self {

        ConnectorPublic{

            inbox: PublicInbox::new(),

            sleeping: AtomicBool::new(initialize_as_sleeping),

        }

    }

}

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

enum Mode {

    NonSync,    // running non-sync code

    Sync,       // running sync code (in potentially multiple branches)

    SyncError,  // encountered an unrecoverable error in sync mode

    Error,      // encountered an error in non-sync mode (or finished handling the sync mode error).

}

#[derive(Debug)]

pub(crate) enum ConnectorScheduling {

    Immediate,          // Run again, immediately

    Later,              // Schedule for running, at some later point in time

    NotNow,             // Do not reschedule for running

    Exit,               // Connector has exited

    Error(EvalError),   // Connector has experienced a fatal error

}

pub(crate) struct ConnectorPDL {

    mode: Mode,

    eval_error: Option<EvalError>,

    tree: ExecTree,

    consensus: Consensus,

    last_finished_handled: Option<BranchId>,

}

// TODO: Remove remaining fields once 'fires()' is removed from language.

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    prepared: PreparedStatement,

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

            last_finished_handled: None,

        }

    }

    // --- Handling messages

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

        while let Some(message) = ctx.read_next_message() {

            match message {

                Message::Data(message) => self.handle_new_data_message(message, ctx),

                Message::SyncComp(message) => {

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

                        return Some(result);

                    }

                },

                Message::SyncPort(message) => self.handle_new_sync_port_message(message, ctx),

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

            }

        }

        return None;

    }

    pub fn handle_new_data_message(&mut self, message: DataMessage, ctx: &mut ComponentCtx) {

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

        // there is one that can receive this message.

        if !self.consensus.handle_new_data_message(&message, ctx) {

            // Old message, so drop it

            return;

        }

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

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

    }

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

        let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        // Retrieve the branch and run it

        let branch_id = branch_id.unwrap();

        let branch = &mut self.tree[branch_id];

        let mut run_context = ConnectorRunContext{

            branch_id,

            consensus: &self.consensus,

            prepared: branch.prepared.take(),

        };

        let run_result = Self::run_prompt(&mut branch.code_state, &sched_ctx.runtime.protocol_description, &mut run_context);

        if let Err(eval_error) = run_result {

            return ConnectorScheduling::Error(eval_error);

        }

        let run_result = run_result.unwrap();

        // Handle the returned result. Note that this match statement contains

        // explicit returns in case the run result requires that the component's

        // code is ran again immediately

            EvalContinuation::BranchInconsistent => {

                // Branch became inconsistent

                branch.sync_state = SpeculativeState::Inconsistent;

            },

            EvalContinuation::BlockFires(port_id) => {

                // Branch called `fires()` on a port that has not been used yet.

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                // Create two forks, one that assumes the port will fire, and

                // one that assumes the port remains silent

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

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

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

                self.consensus.notify_of_new_branch(branch_id, firing_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(firing_branch_id, port_id, true, comp_ctx);

                debug_assert_eq!(_result, Consistency::Valid);

                self.consensus.notify_of_new_branch(branch_id, silent_branch_id);

                let _result = self.consensus.notify_of_speculative_mapping(silent_branch_id, port_id, false, comp_ctx);

                debug_assert_eq!(_result, Consistency::Valid);

                // Somewhat important: we push the firing one first, such that

                // that branch is ran again immediately.

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

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

                return ConnectorScheduling::Immediate;

            },

            EvalContinuation::BlockGet(port_id) => {

                // Branch performed a `get()` on a port that does not have a

                // received message on that port.

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                branch.sync_state = SpeculativeState::HaltedAtBranchPoint;

                branch.awaiting_port = port_id;

                self.tree.push_into_queue(QueueKind::AwaitingMessage, branch_id);

                // Note: we only know that a branch is waiting on a message when

                // it reaches the `get` call. But we might have already received

                // a message that targets this branch, so check now.

                let mut any_message_received = false;

                for message in comp_ctx.get_read_data_messages(port_id) {

                    if self.consensus.branch_can_receive(branch_id, &message) {

                        // This branch can receive the message, so we do the

                        // fork-and-receive dance

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

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

                        branch.awaiting_port = PortIdLocal::new_invalid();

                        self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

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

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

                        any_message_received = true;

                    }

                }

                if any_message_received {

                    return ConnectorScheduling::Immediate;

                }

            }

            EvalContinuation::SyncBlockEnd => {

                let consistency = self.consensus.notify_of_finished_branch(branch_id);

                if consistency == Consistency::Valid {

                    branch.sync_state = SpeculativeState::ReachedSyncEnd;

                    self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

            EvalContinuation::NewFork => {

                // Like the `NewChannel` result. This means we're setting up

                // a branch and putting a marker inside the RunContext for the

                // next time we run the PDL code

                let left_id = branch_id;

                let right_id = self.tree.fork_branch(left_id);

                self.consensus.notify_of_new_branch(left_id, right_id);

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

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

                let left_branch = &mut self.tree[left_id];

                left_branch.prepared = PreparedStatement::ForkedExecution(true);

                let right_branch = &mut self.tree[right_id];

                right_branch.prepared = PreparedStatement::ForkedExecution(false);

            }

            EvalContinuation::Put(port_id, content) => {

                // Branch is attempting to send data

                let port_id = PortIdLocal::new(port_id.0.u32_suffix);

                let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                if let Err(_) = comp_ctx.submit_message(Message::Data(DataMessage {

                    sync_header, data_header,

                })) {

                    // We don't own the port

                    let pd = &sched_ctx.runtime.protocol_description;

                    let eval_error = branch.code_state.new_error_at_expr(

                        &pd.modules, &pd.heap,

                        String::from("attempted to 'put' on port that is no longer owned")

                    );

                    self.eval_error = Some(eval_error);

                    self.mode = Mode::SyncError;

                }

                branch.prepared = PreparedStatement::PerformedPut;

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

                return ConnectorScheduling::Immediate;

            },

            _ => unreachable!("unexpected run result {:?} in sync mode", run_result),

        }

        // If here then the run result did not require a particular action. We

        // return whether we have more active branches to run or not.

        if self.tree.queue_is_empty(QueueKind::Runnable) {

            return ConnectorScheduling::NotNow;

        } else {

            return ConnectorScheduling::Later;

        }

    }

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

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

        let branch = self.tree.base_branch_mut();

        debug_assert!(branch.sync_state == SpeculativeState::RunningNonSync);

        let mut run_context = ConnectorRunContext{

            branch_id: branch.id,

            consensus: &self.consensus,

            prepared: branch.prepared.take(),

        };

        let run_result = Self::run_prompt(&mut branch.code_state, &sched_ctx.runtime.protocol_description, &mut run_context);

        if let Err(eval_error) = run_result {

            return ConnectorScheduling::Error(eval_error);

        }

        let run_result = run_result.unwrap();

        match run_result {

            EvalContinuation::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

                branch.prepared = PreparedStatement::CreatedChannel((

                    Value::Output(PortId::new(putter.self_id.index)),

                    Value::Input(PortId::new(getter.self_id.index)),

                ));

                comp_ctx.push_port(putter);

                comp_ctx.push_port(getter);

                return ConnectorScheduling::Immediate;

            },

            _ => unreachable!("unexpected run result '{:?}' while running in non-sync mode", run_result),

        }

    }

    /// Helper that moves the component's state back into non-sync mode, using

    /// the provided solution branch ID as the branch that should be comitted to

    /// memory. If this function returns false, then the component is supposed

    /// to exit.

    fn enter_non_sync_mode(&mut self, conclusion: RoundConclusion, ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncError);

        // Depending on local state decide what to do

        let final_branch_id = match conclusion {

            RoundConclusion::Success(branch_id) => Some(branch_id),

            RoundConclusion::Failure => if self.mode == Mode::SyncError {

                // We experienced an error, so exit now

                None

            } else {

                // We didn't experience an error, so retry

                // TODO: Decide what to do with sync errors

                Some(BranchId::new_invalid())

            }

        };

        if let Some(solution_branch_id) = final_branch_id {

            let mut fake_vec = Vec::new();

            self.tree.end_sync(solution_branch_id);

            self.consensus.end_sync(solution_branch_id, &mut fake_vec);

            debug_assert!(fake_vec.is_empty());

            ctx.notify_sync_end(&[]);

            self.last_finished_handled = None;

            self.eval_error = None; // in case we came from the SyncError mode

            self.mode = Mode::NonSync;

            return ConnectorScheduling::Immediate;

        } else {

            // No final branch, because we're supposed to exit!

            self.last_finished_handled = None;

            self.mode = Mode::Error;

            return ConnectorScheduling::Exit;

        }

    }

    /// Runs the prompt repeatedly until some kind of execution-blocking

    /// condition appears.

    #[inline]

    fn run_prompt(prompt: &mut Prompt, pd: &ProtocolDescription, ctx: &mut ConnectorRunContext) -> Result<EvalContinuation, EvalError> {

        loop {

            let result = prompt.step(&pd.types, &pd.heap, &pd.modules, ctx);

            if let Ok(EvalContinuation::Stepping) = result {

                continue;

            }

            return result;

        }

    }

}

use crate::collections::VecSet;

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

use super::ConnectorId;

use super::branch::BranchId;

use super::port::{ChannelId, PortIdLocal};

use super::inbox::{

    SyncCompMessage, SyncCompContent, SyncPortMessage, SyncPortContent, SyncHeader,

};

struct BranchAnnotation {

    channel_mapping: Vec<ChannelAnnotation>,

    cur_marker: BranchMarker,

}

#[derive(Debug)]

pub(crate) struct LocalSolution {

    component: ConnectorId,

    final_branch_id: BranchId,

    port_mapping: Vec<(ChannelId, BranchMarker)>,

}

#[derive(Debug, Clone)]

pub(crate) struct GlobalSolution {

    component_branches: Vec<(ConnectorId, BranchId)>,

    channel_mapping: Vec<(ChannelId, BranchMarker)>, // TODO: This can go, is debugging info

}

#[derive(Debug, PartialEq, Eq)]

pub enum RoundConclusion {

    Failure,

    Success(BranchId),

}

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

// Consensus

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

struct Peer {

    id: ConnectorId,

    encountered_this_round: bool,

    expected_sync_round: u32,

}

/// The consensus algorithm. Currently only implemented to find the component

/// with the highest ID within the sync region and letting it handle all the

/// local solutions.

///

/// The type itself serves as an experiment to see how code should be organized.

// TODO: Flatten all datastructures

// TODO: Have a "branch+port position hint" in case multiple operations are

//  performed on the same port to prevent repeated lookups

// TODO: A lot of stuff should be batched. Like checking all the sync headers

//  and sending "I have a higher ID" messages. Should reduce locking by quite a

//  bit.

pub(crate) struct Consensus {

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

        debug_assert!(self.is_in_sync());

        let port_desc = ctx.get_port_by_id(port_id).unwrap();

        let channel_id = port_desc.channel_id;

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

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == channel_id {

                match mapping.expected_firing {

                    None => {

                        // Not yet mapped, perform speculative mapping

                        mapping.expected_firing = Some(does_fire);

                        return Consistency::Valid;

                    },

                    Some(current) => {

                        // Already mapped

                        if current == does_fire {

                            return Consistency::Valid;

                        } else {

                            return Consistency::Inconsistent;

                        }

                    }

                }

            }

        }

        unreachable!("notify_of_speculative_mapping called with unowned port");

    }

    /// Generates a new local solution from a finished branch. If the component

    /// is not the leader of the sync region then it will be sent to the

    /// appropriate component. If it is the leader then there is a chance that

    /// this solution completes a global solution. In that case the solution

    /// branch ID will be returned.

    pub(crate) fn handle_new_finished_sync_branch(&mut self, branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        // Turn the port mapping into a local solution

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

        let mut target_mapping = Vec::with_capacity(source_mapping.len());

        for port in source_mapping {

            // Note: if the port is silent, and we've never communicated

            // over the port, then we need to do so now, to let the peer

            // component know about our sync leader state.

            let port_desc = ctx.get_port_by_channel_id(port.channel_id).unwrap();

            let self_port_id = port_desc.self_id;

            let peer_port_id = port_desc.peer_id;

            let channel_id = port_desc.channel_id;

            if !self.encountered_ports.contains(&self_port_id) {

                    sync_header: SyncHeader{

                        sending_component_id: ctx.id,

                        highest_component_id: self.highest_connector_id,

                        sync_round: self.sync_round

                    },

                }));

                self.encountered_ports.push(self_port_id);

            }

            target_mapping.push((

                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

    /// if the sync solution has been found.

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

        if !self.handle_received_sync_header(&message.sync_header, ctx) {

            return None;

        }

        // And handle the contents

        debug_assert_eq!(message.target_component_id, ctx.id);

        match &message.content {

            SyncCompContent::LocalFailure |

            SyncCompContent::LocalSolution(_) |

            SyncCompContent::PartialSolution(_) |

            SyncCompContent::AckFailure |

            SyncCompContent::Presence(_, _) => {

                // Needs to be handled by the leader

                return self.send_to_leader_or_handle_as_leader(message.content, ctx);

            },

            SyncCompContent::GlobalSolution(solution) => {

                // Found a global solution

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                let (_, branch_id) = solution.component_branches.iter()

                    .find(|(component_id, _)| *component_id == ctx.id)

                    .unwrap();

                return Some(RoundConclusion::Success(*branch_id));

            },

            SyncCompContent::GlobalFailure => {

                // Global failure of round, send Ack to leader

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                let _result = self.send_to_leader_or_handle_as_leader(SyncCompContent::AckFailure, ctx);

                debug_assert!(_result.is_none());

                return Some(RoundConclusion::Failure);

            },

            SyncCompContent::Notification => {

                // We were just interested in the sync header we handled above

                return None;

            }

        }

    }

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

        if !self.handle_received_sync_header(&message.sync_header, ctx) {

            return;

        }

        debug_assert!(self.is_in_sync());

        debug_assert!(ctx.get_port_by_id(message.target_port).is_some());

        match message.content {

            SyncPortContent::NotificationWave => {

                // Wave to discover everyone in the network, handling sync

                // header takes care of leader discovery, here we need to make

                // sure we propagate the wave

                if self.handled_wave {

                    return;

                }

                self.handled_wave = true;

                // Propagate wave to all peers except the one that has sent us

                // the wave.

                for mapping in &self.branch_annotations[0].channel_mapping {

                    let channel_id = mapping.channel_id;

                    let port_desc = ctx.get_port_by_channel_id(channel_id).unwrap();

                    if port_desc.self_id == message.target_port {

                        // Wave came from this port, no need to send one back

                        continue;

                    }

                    let message = SyncPortMessage{

                        sync_header: self.create_sync_header(ctx),

                        source_port: port_desc.self_id,

                        target_port: port_desc.peer_id,

                        content: SyncPortContent::NotificationWave,

                    };

                    ctx.submit_message(Message::SyncPort(message)).unwrap();

                }

            }

        }

    }

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, message: &DataMessage, ctx: &ComponentCtx) {

        debug_assert!(self.branch_can_receive(branch_id, message));

        let target_port = ctx.get_port_by_id(message.data_header.target_port).unwrap();

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

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == target_port.channel_id {

                // Found the port in which the message should be inserted

                mapping.registered_id = Some(message.data_header.new_mapping);

                // Check for sent ports

                debug_assert!(self.workspace_ports.is_empty());

                if !self.workspace_ports.is_empty() {

                    todo!("handle received ports");

                    self.workspace_ports.clear();

                }

                return;

            }

        }

        // If here, then the branch didn't actually own the port? Means the

        // caller made a mistake

        unreachable!("incorrect notify_of_received_message");

    }

    /// Matches the mapping between the branch and the data message. If they

    /// match then the branch can receive the message.

    pub fn branch_can_receive(&self, branch_id: BranchId, message: &DataMessage) -> bool {

        if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {

            if message.sync_header.sync_round < peer.expected_sync_round {

                return false;

            }

        }

        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) -> bool {

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

        if !self.handle_peer(sync_header) {

            // We can drop this package

            return false;

        }

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

            }

            // But also send our locally combined solution

            self.forward_local_data_to_new_leader(ctx);

        } else if sync_header.highest_component_id < self.highest_connector_id {

// TODO: Remove Debug derive from all types

#[derive(Debug, Copy, Clone)]

pub(crate) struct ChannelAnnotation {

    pub channel_id: ChannelId,

    pub registered_id: Option<BranchMarker>,

    pub expected_firing: Option<bool>,

}

/// Marker for a branch in a port mapping. A marker is, like a branch ID, a

/// unique identifier for a branch, but differs in that a branch only has one

/// branch ID, but might have multiple associated markers (i.e. one branch

/// performing a `put` three times will generate three markers.

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

pub(crate) struct BranchMarker{

    marker: u32,

}

impl BranchMarker {

    #[inline]

    pub(crate) fn new(marker: u32) -> Self {

        debug_assert!(marker != 0);

        return Self{ marker };

    }

    #[inline]

    pub(crate) fn new_invalid() -> Self {

        return Self{ marker: 0 }

    }

}

/// The header added by the synchronization algorithm to all.

#[derive(Debug, Clone)]

pub(crate) struct SyncHeader {

    pub sending_component_id: ConnectorId,

    pub highest_component_id: ConnectorId,

    pub sync_round: u32,

}

/// The header added to data messages

#[derive(Debug, Clone)]

pub(crate) struct DataHeader {

    pub expected_mapping: Vec<ChannelAnnotation>,

    pub sending_port: PortIdLocal,

    pub target_port: PortIdLocal,

    pub new_mapping: BranchMarker,

}

/// A data message is a message that is intended for the receiver's PDL code,

/// but will also be handled by the consensus algorithm

#[derive(Debug, Clone)]

pub(crate) struct DataMessage {

    pub sync_header: SyncHeader,

    pub data_header: DataHeader,

}

#[derive(Debug)]

pub(crate) enum SyncCompContent {

    LocalFailure, // notifying leader that component has failed (e.g. timeout, whatever)

    LocalSolution(LocalSolution), // sending a local solution to the leader

    PartialSolution(SolutionCombiner), // when new leader is detected, forward all local results

    GlobalSolution(GlobalSolution), // broadcasting to everyone

    GlobalFailure, // broadcasting to everyone

    AckFailure, // acknowledgement of failure to leader

    Notification, // just a notification (so purpose of message is to send the SyncHeader)

    Presence(ConnectorId, Vec<ChannelId>), // notifying leader of component presence (needed to ensure failing a round involves all components in a sync round)

}

/// A sync message is a message that is intended only for the consensus

/// algorithm. The message goes directly to a component.

#[derive(Debug)]

pub(crate) struct SyncCompMessage {

    pub sync_header: SyncHeader,

    pub target_component_id: ConnectorId,

    pub content: SyncCompContent,

}

#[derive(Debug)]

pub(crate) enum SyncPortContent {

    NotificationWave,

}