iter_id = self.tree.get_queue_next(branch_id);

                    self.last_finished_handled = Some(branch_id);

                    if let Some(round_conclusion) = self.consensus.handle_new_finished_sync_branch(branch_id, comp_ctx) {

                        // Actually found a solution

                        return self.enter_non_sync_mode(round_conclusion, comp_ctx);

                    }

                    self.last_finished_handled = Some(branch_id);

                }

                return scheduling;

            },

            Mode::NonSync => {

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

                return scheduling;

            },

            Mode::SyncError => {

                todo!("write");

                return ConnectorScheduling::Exit;

            },

            Mode::Error => {

                // This shouldn't really be called. Because when we reach exit

                // mode the scheduler should not run the component anymore

                unreachable!("called component run() during error-mode");

            },

        }

    }

}

impl ConnectorPDL {

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

        Self{

            mode: Mode::NonSync,

            eval_error: None,

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

                },

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

            receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().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);

    }

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

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

                        branch.prepared = PreparedStatement::PerformedGet(message.content.as_message().unwrap().clone());

                        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,

                    content: DataContent::Message(content),

                })) {

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

            },

            EvalContinuation::SyncBlockStart => {

                comp_ctx.notify_sync_start();

                let sync_branch_id = self.tree.start_sync();

                debug_assert!(self.last_finished_handled.is_none());

                self.consensus.start_sync(comp_ctx);

                self.consensus.notify_of_new_branch(BranchId::new_invalid(), sync_branch_id);

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

                self.mode = Mode::Sync;

                return ConnectorScheduling::Immediate;

            },

            EvalContinuation::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_prompt = Prompt::new(

                    &sched_ctx.runtime.protocol_description.types,

                    &sched_ctx.runtime.protocol_description.heap,

                    definition_id, monomorph_idx, arguments

                );

                let new_component = ConnectorPDL::new(new_prompt);

                comp_ctx.push_component(new_component, comp_ctx.workspace_ports.clone());

                comp_ctx.workspace_ports.clear();

                return ConnectorScheduling::Later;

            },

            EvalContinuation::NewChannel => {

                let (getter, putter) = sched_ctx.runtime.create_channel(comp_ctx.id);

                debug_assert!(getter.kind == PortKind::Getter && putter.kind == PortKind::Putter);

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

        }

    }

    }

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

    }

    /// TODO: Remove this once multi-fire is in place

    #[deprecated]

    pub fn get_annotation(&self, branch_id: BranchId, channel_id: PortIdLocal) -> &ChannelAnnotation {

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

        let port = branch.channel_mapping.iter().find(|v| v.channel_id.index == channel_id.index).unwrap();

        return port;

    }

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

    pub fn start_sync(&mut self, ctx: &ComponentCtx) {

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

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

        debug_assert!(self.solution_combiner.local.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.

        self.branch_annotations.push(BranchAnnotation{

            channel_mapping: ctx.get_ports().iter()

                .map(|v| ChannelAnnotation {

                    channel_id: v.channel_id,

                    registered_id: None,

                    expected_firing: None,

                })

                .collect(),

            cur_marker: BranchMarker::new_invalid(),

        });

        self.branch_markers.push(BranchId::new_invalid());

        self.highest_connector_id = ctx.id;

    }

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

    /// called for each forked branch in the execution tree.

    pub fn notify_of_new_branch(&mut self, parent_branch_id: BranchId, new_branch_id: BranchId) {

        // If called correctly. Then each time we are notified the new branch's

        // index is the length in `branch_annotations`.

        debug_assert!(self.branch_annotations.len() == new_branch_id.index as usize);

        let parent_branch_annotations = &self.branch_annotations[parent_branch_id.index as usize];

        let new_marker = BranchMarker::new(self.branch_markers.len() as u32);

        let new_branch_annotations = BranchAnnotation{

            channel_mapping: parent_branch_annotations.channel_mapping.clone(),

            cur_marker: new_marker,

        };

        self.branch_annotations.push(new_branch_annotations);

        self.branch_markers.push(new_branch_id);

    }

    /// Notifies the consensus algorithm that a particular branch has

    /// encountered an unrecoverable error.

    pub fn notify_of_fatal_branch(&mut self, failed_branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        debug_assert!(self.is_in_sync());

        // Check for trivial case, where branch has not yet communicated within

        // the consensus algorithm

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

        if branch.channel_mapping.iter().all(|v| v.registered_id.is_none()) {

            return Some(RoundConclusion::Failure);

        }

        // We need to go through the hassle of notifying all participants in the

        // sync round that we've encountered an error.

        // --- notify leader

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

        // --- initiate discovery wave (to let leader know about all components)

        self.handled_wave = true;

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

            let channel_id = mapping.channel_id;

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

            let message = SyncPortMessage{

                sync_header: self.create_sync_header(ctx),

                source_port: port_info.self_id,

                target_port: port_info.peer_id,

                content: SyncPortContent::NotificationWave,

            };

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

        }

        return maybe_conclusion;

    }

    /// Notifies the consensus algorithm that a branch has reached the end of

    /// the sync block. A final check for consistency will be performed that the

    /// caller has to handle. Note that

    pub fn new(runtime: Arc<RuntimeInner>, scheduler_id: u32) -> Self {

        return Self{ runtime, scheduler_id };

    }

    pub fn run(&mut self) {

        // Setup global storage and workspaces that are reused for every

        // connector that we run

        'thread_loop: loop {

            // Retrieve a unit of work

            self.debug("Waiting for work");

            let connector_key = self.runtime.wait_for_work();

            if connector_key.is_none() {

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

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

            // state.

            match cur_schedule {

                ConnectorScheduling::Immediate => unreachable!(),

                ConnectorScheduling::Later => {

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

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                },

                ConnectorScheduling::Error(eval_error) => {

                    // Display error. Then exit

                    if port_desc.state == PortState::Closed {

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

                    }

                    port_desc.peer_connector

                },

                Message::SyncComp(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).

                    content.target_component_id

                },

                Message::SyncPort(content) => {

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

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

                    if port_desc.state == PortState::Closed {

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

                    }

                    port_desc.peer_connector

                }

                Message::Control(_) => {

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

                }

            };

            self.runtime.send_message(target_component_id, message);

        }

        while let Some(state_change) = scheduled.ctx.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.inbox_messages.len() {

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

                            if Self::get_message_target_port(message) == Some(port_id) {

                                // Need to transfer this message

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

                                new_connector.ctx.inbox_messages.push(message);

                            } else {

                                message_idx += 1;

                            }

                        }

                        // Transfer the port itself

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

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

                            .unwrap();

                        let port = scheduled.ctx.ports.remove(port_index);

                        new_connector.ctx.ports.push(port.clone());

                        // Notify the peer that the port has changed

                        let reroute_message = scheduled.router.prepare_reroute(

                            port.self_id, port.peer_id, scheduled.ctx.id,

                            port.peer_connector, new_connector.ctx.id

                        );

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

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

                    }

                    // Schedule new connector to run

                    self.runtime.push_work(new_key);

                },

                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 {

            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 {

                self.runtime.push_work(connector_key)

            }

        }

    }

    #[inline]

    fn get_message_target_port(message: &Message) -> Option<PortIdLocal> {

        match message {

            Message::Data(data) => return Some(data.data_header.target_port),

            Message::SyncComp(_) => {},

            Message::SyncPort(content) => return Some(content.target_port),

            Message::Control(control) => {

                match &control.content {

                    ControlContent::PortPeerChanged(port_id, _) => return Some(*port_id),

                    ControlContent::CloseChannel(port_id) => return Some(*port_id),

                    ControlContent::Ping | ControlContent::Ack => {},

                }

            },

        }

        return None

    }

    // TODO: Remove, this is debugging stuff

    fn debug(&self, message: &str) {

    }

    fn debug_conn(&self, conn: ConnectorId, message: &str) {

    }

}

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

// ComponentCtx

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

enum ComponentStateChange {

    CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

#[derive(Clone)]

pub(crate) struct ComponentPortChange {

    pub is_acquired: bool, // otherwise: released

    pub port: Port,

}

/// The component context (better name may be invented). This was created

/// because part of the component's state is managed by the scheduler, and part

/// of it by the component itself. When the component starts a sync block or

/// exits a sync block the partially managed state by both component and

/// scheduler need to be exchanged.

pub(crate) struct ComponentCtx {

    // Mostly managed by the scheduler

    pub(crate) id: ConnectorId,

    ports: Vec<Port>,

    inbox_messages: Vec<Message>,

    inbox_len_read: usize,

    // Submitted by the component

    is_in_sync: bool,

    changed_in_sync: bool,

    outbox: VecDeque<Message>,

    state_changes: VecDeque<ComponentStateChange>,

    // Workspaces that may be used by components to (generally) prevent

    // allocations. Be a good scout and leave it empty after you've used it.

    // TODO: Move to scheduler ctx, this is the wrong place

    pub workspace_ports: Vec<PortIdLocal>,

    pub workspace_branches: Vec<BranchId>,

}

impl ComponentCtx {

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

        return Self{

            id: ConnectorId::new_invalid(),

            ports: Vec::new(),

mod network_shapes;

mod api_component;

mod speculation_basic;

mod basics;

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

fn create_runtime(pdl: &str) -> Runtime {

    let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    return runtime;

}

fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {

    let protocol = ProtocolDescription::parse(pdl.as_bytes())

        .expect("parse PDL");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    let mut api = runtime.create_interface();

    for _ in 0..NUM_INSTANCES {

        constructor(&mut api);

    }

}

pub(crate) struct TestTimer {

    name: &'static str,

    started: std::time::Instant

}

impl TestTimer {

    pub(crate) fn new(name: &'static str) -> Self {

        Self{ name, started: std::time::Instant::now() }

    }

}

impl Drop for TestTimer {

    fn drop(&mut self) {

        let delta = std::time::Instant::now() - self.started;

        let nanos = (delta.as_secs_f64() * 1_000_000.0) as u64;

        let millis = nanos / 1000;

        let nanos = nanos % 1000;

        println!("[{}] Took {:>4}.{:03} ms", self.name, millis, nanos);

    }

}