#[repr(transparent)]

pub struct PortId(pub(crate) Id);

/// A safely aliasable heap-allocated payload of message bytes

#[derive(Default, Eq, PartialEq, Clone, Ord, PartialOrd)]

pub struct Payload(pub Arc<Vec<u8>>);

    parent_index: BranchId,

    // Code execution state

    code_state: ComponentState,

    sync_state: SpeculativeState,

    next_branch_in_queue: Option<u32>,

    // Message/port state

            index: BranchId::new_invalid(),

            parent_index: BranchId::new_invalid(),

            code_state: component_state,

            sync_state: SpeculativeState::RunningNonSync,

            next_branch_in_queue: None,

            received: HashMap::new(),

            (parent_branch.sync_state == SpeculativeState::RunningNonSync && !parent_branch.parent_index.is_valid()) ||

            (parent_branch.sync_state == SpeculativeState::HaltedAtBranchPoint)

        );

        Branch{

            index: BranchId::new(new_index),

            parent_index: parent_branch.index,

            code_state: parent_branch.code_state.clone(),

            sync_state: SpeculativeState::RunningInSync,

            next_branch_in_queue: None,

            received: parent_branch.received.clone(),

        let branch_idx = branch_idx as usize;

        let num_ports = self.owned_ports.len();

        debug_assert!(port_idx < num_ports);

        debug_assert!((branch_idx + 1) * num_ports <= self.port_mapping.len());

    pub ports: ConnectorPorts,

}

struct ConnectorRunContext<'a> {

    ports: &'a ConnectorPorts,

    scheduler: SchedulerCtx<'a>,

}

impl<'a> RunContext for ConnectorRunContext<'a> {

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

            // Either acquired or released, must be silent

            return false;

        }

        let port_index = self.ports.get_port_index(PortIdLocal::new(port.0.u32_suffix)).unwrap();

        return mapping.is_assigned;

    }

    fn get(&mut self, port: PortId) -> Option<ValueGroup> {

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

            Some(message) => Some(message.message.clone()),

            None => None,

        }

    fn fires(&mut self, port: PortId) -> Option<Value> {

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

            return None

        }

        let port_index = self.ports.get_port_index(port_id).unwrap();

        if mapping.is_assigned {

            return Some(Value::Bool(mapping.num_times_fired != 0));

    }

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

    }

}

        }

    }

        if self.in_sync {

            // When in speculative mode we might have generated new sync

            // solutions, we need to turn them into proposed solutions here.

            return scheduling;

        } else {

            return scheduling;

        }

    }

    /// where it is the caller's responsibility to immediately take care of

    /// those changes. The return value indicates when (and if) the connector

    /// needs to be scheduled again.

        debug_assert!(self.in_sync);

        let branch = Self::pop_branch_from_queue(&mut self.branches, &mut self.sync_active);

        // Run the branch to the next blocking point

        // Match statement contains `return` statements only if the particular

        // run result behind handled requires an immediate re-run of the

                    results.ports.clear();

                    results.outbox.push(MessageContents::Data(message));

                    return ConnectorScheduling::Immediate

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

    }

    /// Runs the connector in non-synchronous mode.

        debug_assert!(!self.in_sync);

        debug_assert!(self.sync_active.is_empty() && self.sync_pending_get.is_empty() && self.sync_finished.is_empty());

        debug_assert!(self.branches.len() == 1);

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

        let mut run_context = ConnectorRunContext{

            ports: &self.ports,

        };

        match run_result {

            RunResult::ComponentTerminated => {

                self.in_sync = true;

                let first_sync_branch = Branch::new_sync_branching_from(1, branch);

                let first_sync_branch_id = first_sync_branch.index;

                self.branches.push(first_sync_branch);

                Self::push_branch_into_queue(&mut self.branches, &mut self.sync_active, first_sync_branch_id);

                // Add connector for later execution

                let new_connector_state = ComponentState {

                };

                let new_connector_ports = results.ports.clone(); // TODO: Do something with this

                let new_connector_branch = Branch::new_initial_branch(new_connector_state);

            },

            RunResult::NewChannel => {

                // Need to prepare a new channel

            },

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

        }

    /// Releasing ownership of ports during a sync-session. Will provide an

    /// error if the port was already used during a sync block.

    fn release_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {

        todo!("unfinished: add port properties during final solution-commit msgs");

        debug_assert!(branch.index.is_valid()); // branch in sync mode

    /// Acquiring ports during a sync-session.

    fn acquire_ports_during_sync(ports: &mut ConnectorPorts, branch: &mut Branch, port_ids: &[PortIdLocal]) -> Result<(), PortOwnershipError> {

        todo!("unfinished: add port properties during final solution-commit msgs");

        debug_assert!(branch.index.is_valid()); // branch in sync mode

use inbox::Message;

use connector::{ConnectorPDL, ConnectorPublic, ConnectorScheduling, RunDeltaState};

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

use crate::runtime2::port::Port;

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

        }

    }

        match self {

        }

    }

}

    pub connector: ConnectorVariant, // access by connector

    pub context: ConnectorCtx, // mutable access by scheduler, immutable by connector

    pub public: ConnectorPublic, // accessible by all schedulers and connectors

}

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

        self.scheduler_notifier.notify_one();

    }

    /// Creates a new port pair. Note that these are stored globally like the

    /// connectors are. Ports stored by components belong to those components.

        use port::{PortIdLocal, PortKind};

        let getter_id = self.port_counter.fetch_add(2, Ordering::SeqCst);

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

        };

        let putter_port = Port{

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

        };

        return (getter_port, putter_port);

    }

    // --- Creating/retrieving/destroying components

    pub(crate) fn create_interface_component(&self, component: ConnectorApplication) -> ConnectorKey {

        // Create as not sleeping, as we'll schedule it immediately

        let key = {

            let mut lock = self.connectors.write().unwrap();

        };

        // Transfer the ports

        return key;

    }

    pub(crate) fn get_component_private(&self, connector_key: &ConnectorKey) -> &'static mut ScheduledConnector {

        let lock = self.connectors.read().unwrap();

        return lock.get_private(connector_key);

    }

    pub(crate) fn get_component_public(&self, connector_id: ConnectorId) -> &'static ConnectorPublic {

        let lock = self.connectors.read().unwrap();

        return lock.get_public(connector_id);

            connector,

            context: ConnectorCtx::new(),

            public: ConnectorPublic::new(initially_sleeping),

        };

        let index;

    fn handle_message(&mut self, message: Message, ctx: &ConnectorCtx, delta_state: &mut RunDeltaState);

    /// Should run the connector's behaviour up until the next blocking point.

}

type SyncDone = Arc<(Mutex<bool>, Condvar)>;

        }

    }

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

        while let Some(job) = queue.pop_front() {

            match job {

    /// Creates a new channel.

    pub fn create_channel(&mut self) -> Channel {

        debug_assert_eq!(getter_port.kind, PortKind::Getter);

        let getter_id = getter_port.self_id;

        let putter_id = putter_port.self_id;

    }

}

pub enum PortKind {

    Putter,

    Getter,

    }

}

pub(crate) struct SchedulerCtx<'a> {

    pub(crate) runtime: &'a RuntimeInner

}

    scheduler_id: u32,

}

impl Scheduler {

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

        return Self{ runtime, scheduler_id };

                }

                // Actually run the connector

                let new_schedule = scheduled.connector.run(

                );

                // Handle all of the output from the current run: messages to

                // send and connectors to instantiate.

                    }

                },

                ConnectorScheduling::Exit => {

                }

            }

        }

                                receiving_port: PortIdLocal::new_invalid(),

                                contents: MessageContents::ConfirmCommit(contents.clone()),

                            };

                        }

                        (ConnectorId::new_invalid(), PortIdLocal::new_invalid())

                    },

                        receiving_port: peer_port,

                        contents: message,

                    };

                }

            }

        }

                // let the other end of the channel know that the port has

                // changed location.

                for port in &new_connector.context.ports {

                    let reroute_message = cur_connector.router.prepare_reroute(

                        port.self_id, port.peer_id, cur_connector.context.id,

                        port.peer_connector, new_connector.context.id

                    );

                }

                // Schedule new connector to run

        debug_assert!(delta_state.new_ports.is_empty());

        debug_assert!(delta_state.new_connectors.is_empty());

    }

}

    source_connector: ConnectorId,  // connector we expect messages from

}

    id_counter: u32,

}

    pub fn new() -> Self {

            id_counter: 0,

            active: Vec::new(),

        }

        return Message{

            sending_connector: self_connector_id,

            contents: MessageContents::Control(ControlMessage{

                id,

                content: ControlMessageVariant::ChangePortPeer(peer_port_id, new_owner_connector_id),

#[test]

fn test_put_and_get() {

primitive putter(out<bool> sender, u32 loops) {

    u32 index = 0;

    while (index < loops) {