debug_assert!(additional > 0);

            let new_cap = self.len.checked_add(additional).unwrap();

            let new_cap = cmp::max(new_cap, self.cap * Self::GROWTH_RATE);

            let layout = Layout::array::<T>(new_cap)

                .map_err(|_| AllocError::CapacityOverflow)?;

            debug_assert_eq!(new_cap * Self::T_SIZE, layout.size());

                    let old_base = self.base as *mut u8;

                    let (old_size, old_layout) = self.current_layout();

                    dealloc(old_base, old_layout);

                }

        return self.inner.iter();

    }

    #[inline]

    pub fn is_empty(&self) -> bool {

        self.inner.is_empty()

    pub fn fork_branch(&mut self, parent_branch_id: BranchId) -> BranchId {

        debug_assert!(self.is_in_sync());

        let parent_branch = &self[parent_branch_id];

        let new_branch_id = new_branch.id;

        return new_branch_id;

    }

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

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

use crate::runtime2::consensus::find_ports_in_value_group;

use crate::runtime2::port::PortKind;

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

        if self.tree.is_in_sync() {

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

            if let Some(solution_branch_id) = self.consensus.handle_new_finished_sync_branches(&self.tree, comp_ctx) {

            }

        } else {

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

            return scheduling;

            self.consensus.notify_of_new_branch(branch_id, receiving_branch_id);

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

            // And prepare the branch for running

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

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

        if let Some(solution_branch_id) = self.consensus.handle_new_sync_message(message, ctx) {

        }

    }

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

                    let mut any_branch_received = false;

                    for message in comp_ctx.get_read_data_messages(port_id) {

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

                            // fork-and-receive dance

                            any_branch_received = true;

                        }

                    branch.sync_state = SpeculativeState::ReachedSyncEnd;

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

                } else if consistency == Consistency::Inconsistent {

                }

            },

            RunResult::BranchPut(port_id, content) => {

                    // `put()` is valid.

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

                    comp_ctx.submit_message(MessageFancy::Data(DataMessageFancy{

                    }));

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

                comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                self.consensus.start_sync(comp_ctx);

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

                return ConnectorScheduling::Immediate;

use crate::collections::VecSet;

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

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

use super::ConnectorId;

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

//  and sending "I have a higher ID" messages.

pub(crate) struct Consensus {

    // Local component's state

    highest_connector_id: ConnectorId,

    branch_annotations: Vec<BranchAnnotation>,

    last_finished_handled: Option<BranchId>,

    solution_combiner: SolutionCombiner,

    workspace_ports: Vec<PortIdLocal>,

}

            branch_annotations: Vec::new(),

            last_finished_handled: None,

            encountered_peers: VecSet::new(),

            solution_combiner: SolutionCombiner::new(),

            workspace_ports: Vec::new(),

        }

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

            for port in source_mapping {

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

                target_mapping.push((

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

                ));

            }

        self.branch_annotations.clear();

        self.last_finished_handled = None;

        self.encountered_peers.clear();

        self.solution_combiner.clear();

    }

    /// 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 ComponentCtxFancy) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

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

        if cfg!(debug_assertions) {

            let port = branch.port_mapping.iter()

                .find(|v| v.port_id == source_port_id)

                .unwrap();

            self.workspace_ports.clear();

        }

        // TODO: Handle multiple firings. Right now we just assign the current

        //  branch to the `None` value because we know we can only send once.

        debug_assert!(branch.port_mapping.iter().find(|v| v.port_id == source_port_id).unwrap().registered_id.is_none());

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        let data_header = DataHeader{

            expected_mapping: branch.port_mapping.clone(),

            target_port: port_info.peer_id,

            new_mapping: branch_id

        };

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == source_port_id {

                mapping.expected_firing = Some(true);

            }

        }

    }

    /// Handles a new data message by handling the data and sync header, and

    /// 2. We return the branches that *can* receive the message, you still

    ///     have to explicitly call `notify_of_received_message`.

    pub fn handle_new_data_message(&mut self, exec_tree: &ExecTree, message: &DataMessageFancy, ctx: &mut ComponentCtxFancy, target_ids: &mut Vec<BranchId>) {

        self.handle_received_sync_header(&message.sync_header, ctx);

    }

        }

    }

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

        for mapping in &mut branch.port_mapping {

            if mapping.port_id == data_header.target_port {

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

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

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

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

        for expected in &data_header.expected_mapping {

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

    /// Checks data header and consults the stored port mapping and the

    /// execution tree to see which branches may receive the data message's

    /// contents.

        for branch in exec_tree.iter_queue(QueueKind::AwaitingMessage, None) {

            if branch.awaiting_port == data_header.target_port {

                // Found a branch awaiting the message, but we need to make sure

                // the mapping is correct

                    target_ids.push(branch.id);

                }

            }

    }

    fn send_or_store_local_solution(&mut self, solution: LocalSolution, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {

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

    pub new_mapping: BranchId,

}

/// 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 DataMessageFancy {

    pub sync_header: SyncHeader,

    pub data_header: DataHeader,

}

#[derive(Debug)]

    /// Retrieves public part of connector - accessible by many threads at once.

    fn get_public(&self, id: ConnectorId) -> &'static ConnectorPublic {

        unsafe {

            let connector = self.connectors.get(id.0 as usize);

            debug_assert!(!connector.is_null());

            return &(**connector).public;

    /// time.

    fn get_private(&self, key: &ConnectorKey) -> &'static mut ScheduledConnector {

        unsafe {

            let connector = self.connectors.get_mut(key.index as usize);

            debug_assert!(!connector.is_null());

            return &mut (**connector);

        connector.inbox.insert_message(MessageFancy::Control(ControlMessageFancy{

            id: 0,

            sending_component_id: self.connector_id,

        }));

        let should_wake_up = connector.sleeping

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

                    // the sender must make sure it actually wants to send to

                    // the specified component (and is not using an inconsistent

                    // component ID associated with a port).

                },

                MessageFancy::Control(_) => {

                    unreachable!("component sending control messages directly");

                        let mut message_idx = 0;

                        while message_idx < scheduled.ctx_fancy.inbox_messages.len() {

                            let message = &scheduled.ctx_fancy.inbox_messages[message_idx];

                                // Need to transfer this message

                                let message = scheduled.ctx_fancy.inbox_messages.remove(message_idx);

                                new_connector.ctx_fancy.inbox_messages.push(message);

    }

    #[inline]

        }

        return None

use crate::common::Id;

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

const NUM_INSTANCES: u32 = 10;  // number of test instances constructed

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

                print(\"getting!\");

                auto result = get(receiver);

                assert(result);

            }

            index += 1;

        }