/// Identifier of a port in a session

#[derive(Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize)]

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

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

/// "Orientation" of a port, determining whether they can send or receive messages with `put` and `get` respectively.

pub(crate) struct Branch {

    index: BranchId,

    parent_index: BranchId,

    // Code execution state

    code_state: ComponentState,

    sync_state: SpeculativeState,

    next_branch_in_queue: Option<u32>,

    // Message/port state

    received: HashMap<PortIdLocal, DataMessage>, // TODO: @temporary, remove together with fires()

    ports_delta: Vec<PortOwnershipDelta>,

}

    /// first instruction

    pub(crate) fn new_initial_branch(component_state: ComponentState) -> Self {

        Branch{

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

            ports_delta: Vec::new(),

        }

    }

    /// parent of the new branch within the execution tree.

    fn new_sync_branching_from(new_index: u32, parent_branch: &Branch) -> Self {

        debug_assert!(

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

            ports_delta: parent_branch.ports_delta.clone(),

        }

    }

    // based on branch/port index.

    #[inline]

    fn mapped_index(&self, branch_idx: u32, port_idx: usize) -> usize {

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

        return branch_idx * num_ports + port_idx;

    }

}

    // Port/message management

    pub committed_to: Option<(ConnectorId, u64)>,

    pub inbox: PrivateInbox,

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

        } else {

            return None;

        }

    }

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

    }

}

impl Connector for ConnectorPDL {

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

        use MessageContents as MC;

            MC::RequestCommit(content) => self.handle_request_commit_message(content, ctx, delta_state),

            MC::ConfirmCommit(content) => self.handle_confirm_commit_message(content, ctx, delta_state),

            MC::Control(_) | MC::Ping => {},

        }

    }

        if self.in_sync {

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

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

            if self.sync_finished_last_handled != self.sync_finished.last {

                // Retrieve first element in queue

                let mut next_id;

                self.sync_finished_last_handled = next_id;

            }

            return scheduling;

        } else {

            return scheduling;

        }

    }

}

impl ConnectorPDL {

    /// Runs the connector in synchronous mode. Potential changes to the global

    /// system's state are added to the `RunDeltaState` object by the connector,

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

        // connector.

        match run_result {

            RunResult::BranchInconsistent => {

                    debug_assert!(results.ports.is_empty());

                    find_ports_in_value_group(&message.message, &mut results.ports);

                    Self::release_ports_during_sync(&mut self.ports, branch, &results.ports).unwrap();

                    results.ports.clear();

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

                    return ConnectorScheduling::Immediate

                } else {

                    branch.sync_state = SpeculativeState::Inconsistent;

                }

            },

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

        } else {

            return ConnectorScheduling::Later;

        }

    }

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

        let branch = &mut self.branches[0];

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

        let mut run_context = ConnectorRunContext{

            ports: &self.ports,

        };

        match run_result {

            RunResult::ComponentTerminated => {

                // Need to wait until all children are terminated

                // TODO: Think about how to do this?

                branch.sync_state = SpeculativeState::Finished;

            },

            RunResult::ComponentAtSyncStart => {

                // Prepare for sync execution and reschedule immediately

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

                return ConnectorScheduling::Later;

            },

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

                        todo!("fatal error handling");

                    }

                }

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

                let new_connector = ConnectorPDL::new(new_connector_branch, new_connector_ports);

                results.new_connectors.push(new_connector);

                return ConnectorScheduling::Later;

            },

            RunResult::NewChannel => {

                // Need to prepare a new channel

            },

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

        }

    }

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

        return Ok(())

    }

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

        for port_id in port_ids {

            match ports.get_port_index(*port_id) {

                Some(port_index) => {

        return Ok(())

    }

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

        'port_loop: for port_id in port_ids {

            for (delta_idx, delta) in branch.ports_delta.iter().enumerate() {

                if delta.port_id == *port_id {

use crate::collections::RawVec;

use crate::ProtocolDescription;

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

/// key is the only one that may execute the connector's code.

pub(crate) struct ConnectorKey {

    pub index: u32, // of connector

        match self {

            ConnectorVariant::UserDefined(c) => c.handle_message(message, ctx, delta_state),

            ConnectorVariant::Native(c) => c.handle_message(message, ctx, delta_state),

        }

    }

        match self {

        }

    }

}

pub(crate) struct ScheduledConnector {

    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

}

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

// Runtime

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

    pub(crate) fn push_work(&self, key: ConnectorKey) {

        let mut lock = self.connector_queue.lock().unwrap();

        lock.push_back(key);

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

        let putter_id = PortIdLocal::new(getter_id + 1);

        let getter_id = PortIdLocal::new(getter_id);

        let getter_port = Port{

            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 {

        // Initialize as sleeping, as it will be scheduled by the programmer.

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

        let key = lock.create(ConnectorVariant::Native(Box::new(component)), true);

    /// connector.

    // TODO: Nicer code, not forcing the caller to schedule, perhaps?

    pub(crate) fn create_pdl_component(&self, created_by: &mut ScheduledConnector, connector: ConnectorPDL) -> ConnectorKey {

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

        let key = {

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

        };

        // Transfer the ports

        {

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

            let created = lock.get_private(&key);

        }

        self.increment_active_components();

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

    }

    pub(crate) fn destroy_component(&self, connector_key: ConnectorKey) {

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

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

        let mut connector = ScheduledConnector {

            connector,

            context: ConnectorCtx::new(),

            public: ConnectorPublic::new(initially_sleeping),

        };

        let index;

        let key;

        if self.free.is_empty() {

    /// Handle a new message (preprocessed by the scheduler). You probably only

    /// want to handle `Data`, `Sync`, and `Solution` messages. The others are

    /// intended for the scheduler itself.

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

type JobQueue = Arc<Mutex<VecDeque<ApplicationJob>>>;

enum ApplicationJob {

            },

            MC::Control(_) => {},

            MC::Ping => {},

        }

    }

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

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

            match job {

                ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {

                    println!("DEBUG: API adopting ports");

                    delta_state.new_ports.reserve(2);

            owned_ports: Vec::new(),

        }

    }

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

        {

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

    pub fn is_valid(&self) -> bool {

        return self.index != u32::MAX;

    }

}

pub enum PortKind {

    Putter,

    Getter,

}

/// Represents a port inside of the runtime. May be without owner if it is

        }

        panic!("port {:?}, not owned by connector", id);

    }

}

pub(crate) struct SchedulerCtx<'a> {

    pub(crate) runtime: &'a RuntimeInner

}

pub(crate) struct Scheduler {

    runtime: Arc<RuntimeInner>,

    scheduler_id: u32,

}

impl Scheduler {

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

        return Self{ runtime, scheduler_id };

    }

    pub fn run(&mut self) {

                        // Let connector handle message

                        scheduled.connector.handle_message(message, &scheduled.context, &mut delta_state);

                    }

                }

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

                self.handle_delta_state(&connector_key, &mut scheduled.context, &mut delta_state);

                cur_schedule = new_schedule;

                        if should_reschedule_self {

                            self.runtime.push_work(connector_key);

                        }

                    }

                },

                ConnectorScheduling::Exit => {

                }

            }

        }

    }

    fn handle_delta_state(&mut self, connector_key: &ConnectorKey, context: &mut ConnectorCtx, delta_state: &mut RunDeltaState) {

                        for to_visit in &contents.to_visit {

                            let message = Message{

                                sending_connector: connector_id,

                                receiving_port: PortIdLocal::new_invalid(),

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

                            };

                        }

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

                    },

                    MessageContents::Control(_) | MessageContents::Ping => {

                        // Never generated by the user's code

                        unreachable!();

                if peer_connector.is_valid() {

                    let message = Message {

                        sending_connector: connector_id,

                        receiving_port: peer_port,

                        contents: message,

                    };

                }

            }

        }

        if !delta_state.new_ports.is_empty() {

            for port in delta_state.new_ports.drain(..) {

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

                // Call above changed ownership of ports, but we still have to

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

                self.runtime.push_work(new_key);

            }

        }

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

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

        }

    }

    /// Prepares rerouting messages due to changed ownership of a port. The

            source_connector: peer_connector_id,

            target_connector: new_owner_connector_id,

        });

        return Message{

            sending_connector: self_connector_id,

            contents: MessageContents::Control(ControlMessage{

                id,

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

            })

        };

    }

    return runtime;

}

#[test]

fn test_put_and_get() {

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

    u32 index = 0;

    while (index < loops) {

        synchronous {

            print(\"putting!\");

            put(sender, true);