return self.len;

    }

    fn ensure_space(&mut self, additional: usize) -> Result<(), AllocError>{

        debug_assert!(Self::T_SIZE != 0);

        debug_assert!(self.cap >= self.len);

        if self.cap - self.len < additional {

            // Need to resize. Note that due to all checked conditions we have

            // that new_cap >= 1.

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

            unsafe {

                // Allocate new storage, transfer bits, deallocate old store

                let new_base = alloc(layout);

                if self.cap > 0 {

                    let old_base = self.base as *mut u8;

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

                    dealloc(old_base, old_layout);

                }

                self.base = new_base as *mut T;

                self.cap = new_cap;

            }

        } // else: still enough space

        return Ok(());

    }

    #[inline]

    }

    #[inline]

    pub fn clear(&mut self) {

        self.inner.clear();

    }

    #[inline]

    pub fn iter(&self) -> impl Iterator<Item=&T> {

        return self.inner.iter();

    }

    #[inline]

    pub fn is_empty(&self) -> bool {

        self.inner.is_empty()

    }

    #[inline]

    pub fn into_vec(self) -> Vec<T> {

        return self.inner;

    }

}

        let sync_branch = Branch::new_sync(1, &self.branches[0]);

        let sync_branch_id = sync_branch.id;

        self.branches.push(sync_branch);

        return sync_branch_id;

    }

    /// Creates a new speculative branch based on the provided one. The index to

    /// retrieve this new branch will be returned.

    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;

    }

    /// Collapses the speculative execution tree back into a deterministic one,

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

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

        debug_assert!(self.is_in_sync());

        debug_assert!(self.iter_queue(QueueKind::FinishedSync, None).any(|v| v.id == branch_id));

        // Swap indicated branch into the first position

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

///     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::sync::atomic::AtomicBool;

use crate::PortId;

use crate::common::ComponentState;

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

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

use super::inbox2::{DataMessageFancy, MessageFancy, SyncMessageFancy, PublicInbox};

use super::native::Connector;

use super::port::PortIdLocal;

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

pub(crate) struct ConnectorPublic {

    pub inbox: PublicInbox,

    pub sleeping: AtomicBool,

    fn get_channel(&mut self) -> Option<(Value, Value)> {

        return self.prepared_channel.take();

    }

}

impl Connector for ConnectorPDL {

    fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> ConnectorScheduling {

        self.handle_new_messages(comp_ctx);

        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;

        }

    }

}

impl ConnectorPDL {

    pub fn new(initial: ComponentState) -> Self {

        Self{

            tree: ExecTree::new(initial),

            consensus: Consensus::new(),

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

        let mut branches = Vec::new(); // TODO: @Remove

        self.consensus.handle_new_data_message(&self.tree, &message, ctx, &mut branches);

        for branch_id in branches.drain(..) {

            // 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];

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

        }

    }

    // --- Running code

    pub fn run_in_sync_mode(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtxFancy) -> 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;

        }

                let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);

                if consistency == Consistency::Valid {

                    // `get()` is valid, so mark the branch as awaiting a message

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

                        }

                    }

                    if any_branch_received {

                        return ConnectorScheduling::Immediate;

                    }

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            }

            RunResult::BranchAtSyncEnd => {

                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 if consistency == Consistency::Inconsistent {

                }

            },

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

                // Branch is attempting to send data

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

                let consistency = self.consensus.notify_of_speculative_mapping(branch_id, port_id, true);

                if consistency == Consistency::Valid {

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

                    return ConnectorScheduling::Immediate;

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

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

        }

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

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

        match run_result {

            RunResult::ComponentTerminated => {

                branch.sync_state = SpeculativeState::Finished;

                return ConnectorScheduling::Exit;

            },

            RunResult::ComponentAtSyncStart => {

                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;

            },

            RunResult::NewComponent(definition_id, monomorph_idx, arguments) => {

                // Note: we're relinquishing ownership of ports. But because

                // we are in non-sync mode the scheduler will handle and check

                // port ownership transfer.

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

                find_ports_in_value_group(&arguments, &mut comp_ctx.workspace_ports);

                let new_state = ComponentState {

use crate::collections::VecSet;

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

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

use super::ConnectorId;

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

use super::inbox2::{

    DataHeader, DataMessageFancy, MessageFancy,

    SyncContent, SyncHeader, SyncMessageFancy, PortAnnotation

};

use super::scheduler::ComponentCtxFancy;

struct BranchAnnotation {

    port_mapping: Vec<PortAnnotation>,

/// 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.

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>,

}

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

pub(crate) enum Consistency {

    Valid,

    Inconsistent,

}

impl Consensus {

    pub fn new() -> Self {

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

            last_finished_handled: None,

            encountered_peers: VecSet::new(),

            solution_combiner: SolutionCombiner::new(),

            workspace_ports: Vec::new(),

        }

    }

    // --- Controlling sync round and branches

    /// Returns whether the consensus algorithm is running in sync mode

    pub fn is_in_sync(&self) -> bool {

        return !self.branch_annotations.is_empty();

    }

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

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

    pub fn handle_new_finished_sync_branches(&mut self, tree: &ExecTree, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {

        debug_assert!(self.is_in_sync());

        let mut last_branch_id = self.last_finished_handled;

        for branch in tree.iter_queue(QueueKind::FinishedSync, last_branch_id) {

            // Turn the port mapping into a local solution

            let source_mapping = &self.branch_annotations[branch.id.index as usize].port_mapping;

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

                ));

            }

            let local_solution = LocalSolution{

                component: ctx.id,

                final_branch_id: branch.id,

                port_mapping: target_mapping,

            };

            let solution_branch = self.send_or_store_local_solution(local_solution, ctx);

            if solution_branch.is_some() {

                // No need to continue iterating, we've found the solution

        // Set final ports

        final_ports.clear();

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

        for port in &branch.port_mapping {

            final_ports.push(port.port_id);

        }

        // Clear out internal storage to defaults

        self.highest_connector_id = ConnectorId::new_invalid();

        self.branch_annotations.clear();

        self.last_finished_handled = None;

        self.encountered_peers.clear();

        self.solution_combiner.clear();

    }

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

            debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));

        }

        // Check for ports that are being sent

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

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

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

                mapping.registered_id = Some(branch_id);

            }

        }

    }

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

    /// checking which *existing* branches *can* receive the message. So two

    /// cautionary notes:

    /// 1. A future branch might also be able to receive this message, see the

    ///     `branch_can_receive` function.

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

    }

    /// Handles a new sync message by handling the sync header and the contents

    /// of the message. Returns `Some` with the branch ID of the global solution

    /// if the sync solution has been found.

    pub fn handle_new_sync_message(&mut self, message: SyncMessageFancy, ctx: &mut ComponentCtxFancy) -> Option<BranchId> {

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

        // And handle the contents

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

        match message.content {

                return self.send_or_store_local_solution(solution, ctx);

            },

            SyncContent::GlobalSolution(solution) => {

                // Take branch of interest and return it.

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

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

                    .unwrap();

                return Some(*branch_id);

            }

        }

    }

        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 {

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

                mapping.registered_id = Some(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.

        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

            // annotation, check if the current mapping matches

            for current in &annotation.port_mapping {

                if expected.port_id == current.port_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

    /// 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 handle_received_sync_header(&mut self, sync_header: &SyncHeader, ctx: &mut ComponentCtxFancy) {

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

        self.encountered_peers.push(sync_header.sending_component_id);

        if sync_header.highest_component_id > self.highest_connector_id {

            // Sender has lower leader ID, so it should know about our higher

            // one.

            let message = SyncMessageFancy{

                sync_header: self.create_sync_header(ctx),

                target_component_id: sync_header.sending_component_id,

                content: SyncContent::Notification

            };

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

        } // else: exactly equal, so do nothing

    }

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

                let mut my_final_branch_id = BranchId::new_invalid();

                for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {

                    if connector_id == ctx.id {

                        // This is our solution branch

                        my_final_branch_id = branch_id;

                        continue;

                    }

                    let message = SyncMessageFancy{

    pub highest_component_id: ConnectorId,

}

/// The header added to data messages

#[derive(Debug, Clone)]

pub(crate) struct DataHeader {

    pub expected_mapping: Vec<PortAnnotation>,

    pub sending_port: PortIdLocal,

    pub target_port: PortIdLocal,

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

pub(crate) enum SyncContent {

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

    GlobalSolution(GlobalSolution), // broadcasting to everyone

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

}

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

/// algorithm.

#[derive(Debug)]

impl ConnectorStore {

    fn with_capacity(capacity: usize) -> Self {

        Self {

            connectors: RawVec::with_capacity(capacity),

            free: Vec::with_capacity(capacity),

        }

    }

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

        }

    }

    /// Retrieves private part of connector - accessible by one thread at a

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

        }

    }

    /// Creates a new connector. Caller should ensure ports are set up correctly

    /// and the connector is queued for execution if needed.

    fn create(&mut self, connector: ConnectorVariant, initially_sleeping: bool) -> ConnectorKey {

        let mut connector = ScheduledConnector {

            connector,

            ctx_fancy: ComponentCtxFancy::new_empty(),

    }

    /// Called by runtime to set associated connector's ID.

    pub(crate) fn set_connector_id(&mut self, id: ConnectorId) {

        self.connector_id = id;

    }

    fn wake_up_connector_with_ping(&self) {

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

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

            id: 0,

            sending_component_id: self.connector_id,

        }));

        let should_wake_up = connector.sleeping

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

            .is_ok();

        if should_wake_up {

            println!("DEBUG: Waking up connector");

            let key = unsafe{ ConnectorKey::from_id(self.connector_id) };

            self.runtime.push_work(key);

        } else {

            println!("DEBUG: NOT waking up connector");

        }

    }

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

            match message {

                MessageFancy::Control(message) => {

                    if port_desc.state == PortState::Closed {

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

                    }

                    port_desc.peer_connector

                },

                MessageFancy::Sync(content) => {

                    // Sync messages are always sent to a particular component,

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

                }

            };

            self.runtime.send_message(target_component_id, message);

        }

        while let Some(state_change) = scheduled.ctx_fancy.state_changes.pop_front() {

            match state_change {

                ComponentStateChange::CreatedComponent(component, initial_ports) => {

                    // Creating a new component. The creator needs to relinquish

                    // ownership of the ports that are given to the new

                    // component. All data messages that were intended for that

                    // port also needs to be transferred.

                    let new_key = self.runtime.create_pdl_component(component, false);

                    let new_connector = self.runtime.get_component_private(&new_key);

                    for port_id in initial_ports {

                        // Transfer messages associated with the transferred port

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

                            } else {

                                message_idx += 1;

                            }

                        }

                        // Transfer the port itself

                        let port_index = scheduled.ctx_fancy.ports.iter()

                            .position(|v| v.self_id == port_id)

                            .unwrap();

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

                self.runtime.push_work(connector_key)

            }

        }

    }

    #[inline]