}

    #[inline]

        return self.inner.iter();

    }

    /// using the provided branch as the final sync result.

    pub fn end_sync(&mut self, branch_id: BranchId) {

        debug_assert!(self.is_in_sync());

        // Swap indicated branch into the first position

        self.branches.swap(0, branch_id.index as usize);

        branch.id = BranchId::new_invalid();

        branch.parent_id = BranchId::new_invalid();

        branch.sync_state = SpeculativeState::RunningNonSync;

        branch.next_in_queue = BranchId::new_invalid();

        // Clear out all the queues

        for queue_idx in 0..NUM_QUEUES {

use super::branch::{Branch, BranchId, ExecTree, QueueKind, SpeculativeState};

use super::consensus::{Consensus, Consistency};

use super::native::Connector;

use super::port::PortIdLocal;

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

    prepared_channel: Option<(Value, Value)>,

}

    fn did_put(&mut self, port: PortId) -> bool {

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

        let annotation = self.consensus.get_annotation(self.branch_id, port_id);

        self.handle_new_messages(comp_ctx);

        if self.tree.is_in_sync() {

            let scheduling = self.run_in_sync_mode(sched_ctx, comp_ctx);

            return scheduling;

        } else {

            let scheduling = self.run_in_deterministic_mode(sched_ctx, comp_ctx);

    pub fn handle_new_messages(&mut self, ctx: &mut ComponentCtxFancy) {

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

            match message {

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

            }

        }

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

        // there is one that can receive this message.

        debug_assert!(ctx.workspace_branches.is_empty());

        for branch_id in ctx.workspace_branches.drain(..) {

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

    }

    pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) {

    }

    // --- Running code

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

        let mut run_context = ConnectorRunContext{

            consensus: &self.consensus,

            received: &branch.inbox,

            scheduler: *sched_ctx,

            RunResult::ComponentAtSyncStart => {

                let current_ports = comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

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

                return ConnectorScheduling::Immediate;

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

        }

    }

}

use crate::collections::VecSet;

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

struct BranchAnnotation {

    port_mapping: Vec<PortAnnotation>,

#[derive(Clone)]

pub(crate) struct GlobalSolution {

}

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

    /// Sets up the consensus algorithm for a new synchronous round. The

    /// provided ports should be the ports the component owns at the start of

    /// the sync round.

        debug_assert!(!self.highest_connector_id.is_valid());

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

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

        // We'll use the first "branch" (the non-sync one) to store our ports,

        // this allows cloning if we created a new branch.

                })

                .collect(),

        });

    }

    /// Notifies the consensus algorithm that a new branch has appeared. Must be

    /// Generates sync messages for any branches that are at the end of the

    /// sync block. To find these branches, they should've been put in the

    /// "finished" queue in the execution tree.

        debug_assert!(self.is_in_sync());

        let mut last_branch_id = self.last_finished_handled;

                final_branch_id: branch.id,

                port_mapping: target_mapping,

            };

            last_branch_id = Some(branch.id);

        }

        self.last_finished_handled = last_branch_id;

    }

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

            final_ports.push(port.port_id);

        }

    }

    // --- Handling messages

            SyncContent::LocalSolution(solution) => {

                // We might be the leader, or earlier messages caused us to not

                // be the leader anymore.

            },

            SyncContent::GlobalSolution(solution) => {

                // Take branch of interest and return it.

                    .find(|(connector_id, _)| connector_id == ctx.id)

                    .unwrap();

                return Some(*branch_id);

        } // else: exactly equal, so do nothing

    }

        if self.highest_connector_id == ctx.id {

            // We are the leader

            if let Some(global_solution) = self.solution_combiner.add_solution_and_check_for_global_solution(solution) {

                    let message = SyncMessageFancy{

                        sync_header: self.create_sync_header(ctx),

                        target_component_id: connector_id,

                    };

                    ctx.submit_message(MessageFancy::Sync(message));

                }

            }

        } else {

            // Someone else is the leader

                content: SyncContent::LocalSolution(solution),

            };

            ctx.submit_message(MessageFancy::Sync(message));

        }

    }

struct MatchedLocalSolution {

    final_branch_id: BranchId,

    matches: Vec<ComponentMatches>,

}

        let component_id = solution.component;

        let solution = MatchedLocalSolution{

            final_branch_id: solution.final_branch_id,

            matches: Vec::new(),

        };

        // in the stored solutions which other components are peers of the new

        // one.

        if new_component {

            let mut component_peers = Vec::new();

            // Find the matching components

                    continue;

                }

                            // We have a shared port

                        }

                    }

                }

                    // We share some ports

                    component_peers.push(ComponentPeer{

                        target_id: other_component.component,

                        target_index: other_index,

                    });

                }

            }

                    num_ports_in_peers += existing_peer.involved_channels.len();

                }

                    other_component.all_peers_present = true;

                }

                // Check the port mappings between the pair of solutions.

                let mut all_matched = true;

                        if cur_port == other_port {

                            if cur_branch == other_branch {

                                // Same port mapping, go to next port

        // all (all components have their peers), and the exit condition of the

        // while loop: if we're here, then we have a global solution

        debug_assert_eq!(check_stack.len(), self.local.len());

        for (component_index, solution_index) in check_stack.iter().copied() {

            let component = &self.local[component_index];

            let solution = &component.solutions[solution_index];

        }

        // Just debugging here, TODO: @remove

        for (component_index, _) in check_stack.iter().copied() {

            let component = &self.local[component_index];

        }

        let mut total_num_checked = 0;

        for (component_index, solution_index) in check_stack.iter().copied() {

            let component = &self.local[component_index];

            let solution = &component.solutions[solution_index];

                match final_mapping.iter().find(|(v, _)| *v == channel_id) {

                    Some((_, encountered_branch_id)) => {

                        debug_assert_eq!(encountered_branch_id, branch_id);

            }

        }

        return Some(GlobalSolution{

        });

    }

                solutions.push(LocalSolution{

                    component: component.component,

                    final_branch_id: solution.final_branch_id,

                });

            }

        }

        return solutions;

    }

}

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

    pub sending_connector: ConnectorId,

    pub receiving_port: PortIdLocal, // may be invalid (in case of messages targeted at the connector)

    pub contents: MessageContents,

}

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

use crate::runtime2::branch::BranchId;

use crate::runtime2::ConnectorId;

    Data(DataMessageFancy),

    Sync(SyncMessageFancy),

    Control(ControlMessageFancy),

}

mod runtime;

mod messages;

mod branch;

mod native;

mod port;

mod scheduler;

mod consensus;

mod inbox2;

use crate::collections::RawVec;

use crate::ProtocolDescription;

use connector2::{ConnectorPDL, ConnectorPublic, ConnectorScheduling};

use native::{Connector, ConnectorApplication, ApplicationInterface};

/// A kind of token that, once obtained, allows mutable access to a connector.

/// We're trying to use move semantics as much as possible: the owner of this

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

use super::{ConnectorKey, ConnectorId, RuntimeInner};

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

/// Generic connector interface from the scheduler's point of view.

pub(crate) trait Connector {

enum ApplicationJob {

    NewChannel((Port, Port)),

    Shutdown,

}

        // Handle any incoming messages if we're participating in a round

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

            match message {

            }

        }

                        comp_ctx.push_port(endpoint_a);

                        comp_ctx.push_port(endpoint_b);

                    }

                        println!("DEBUG: API creating connector");

                    },

                    ApplicationJob::Shutdown => {

                        debug_assert!(queue.is_empty());

        // asynchronously.

        let mut initial_ports = Vec::new();

        find_ports_in_value_group(&arguments, &mut initial_ports);

            }

        }

        let state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;

        // Put on job queue

        {

            let mut queue = self.job_queue.lock().unwrap();

        }

        self.wake_up_connector_with_ping();

    fn wake_up_connector_with_ping(&self) {

        let connector = self.runtime.get_component_public(self.connector_id);

        let should_wake_up = connector.sleeping

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

                            // And respond with an Ack

                            let ack_message = MessageFancy::Control(ControlMessageFancy{

                                sending_component_id: connector_id,

                                content: ControlContent::Ack,

                            });

                            // Send an Ack

                            let ack_message = MessageFancy::Control(ControlMessageFancy{

                                sending_component_id: connector_id,

                                content: ControlContent::Ack,

                            });

                            self.runtime.send_message(message.sending_component_id, ack_message);

                        },

                        ControlContent::Ack => {

                        },

                        ControlContent::Ping => {},

                    }

                if message.data_header.target_port == self.match_port_id {

                    // Found a match

                    self.next_index += 1;

                }

            } else {

                // Unreachable because: