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

        let mapped_idx = self.mapped_index(branch_idx, port_idx);

        return &self.port_mapping[mapped_idx];

    }

    #[inline]

    fn get_port_mut(&mut self, branch_idx: u32, port_idx: usize) -> &mut PortAssignment {

        let mapped_idx = self.mapped_index(branch_idx, port_idx);

        return &mut self.port_mapping[mapped_idx];

    }

    #[inline]

    fn num_ports(&self) -> usize {

        return self.owned_ports.len();

    }

    // Function for internal use: retrieve index in flattened port mapping array

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

    }

}

struct BranchQueue {

    first: u32,

    last: u32,

}

impl BranchQueue {

    #[inline]

    fn new() -> Self {

        Self{ first: 0, last: 0 }

    }

    #[inline]

    fn is_empty(&self) -> bool {

        debug_assert!((self.first == 0) == (self.last == 0));

        return self.first == 0;

    }

    /// Creates a new channel.

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

        let (getter_port, putter_port) = self.runtime.create_channel(self.connector_id);

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

            lock.push_back(ApplicationJob::NewChannel((getter_port, putter_port)));

        }

        // Add to owned ports for error checking while creating a connector

        self.owned_ports.reserve(2);

        self.owned_ports.push(putter_id);

        self.owned_ports.push(getter_id);

        return Channel{ putter_id, getter_id };

    }

    /// Creates a new connector. Note that it is not scheduled immediately, but

    /// depends on the `ApplicationConnector` to run, followed by the created

    /// connector being scheduled.

    pub fn create_connector(&mut self, module: &str, routine: &str, arguments: ValueGroup) -> Result<(), ComponentCreationError> {

        // Retrieve ports and make sure that we own the ones that are currently

        // specified. This is also checked by the scheduler, but that is done

        // asynchronously.

        let mut initial_ports = Vec::new();

        find_ports_in_value_group(&arguments, &mut initial_ports);

        for port_to_remove in &initial_ports {

            match self.owned_ports.iter().position(|v| v == port_to_remove) {

                Some(index_to_remove) => {

                    // We own the port, so continue

                    self.owned_ports.remove(index_to_remove);

                },

                None => {

                    // We don't own the port

                    return Err(ComponentCreationError::UnownedPort);

                }

            }

        }

        let state = self.runtime.protocol_description.new_component_v2(module.as_bytes(), routine.as_bytes(), arguments)?;

        let connector = ConnectorPDL::new(Branch::new_initial_branch(state), initial_ports);

        // Put on job queue

        {

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

    }

}

// Because it contains pointers we're going to do a copy by value on this one

#[derive(Clone, Copy)]

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

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

        // connector that we run

        let mut delta_state = RunDeltaState::new();

        '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 cur_schedule == ConnectorScheduling::Immediate {

                // Check all the message that are in the shared inbox

                while let Some(message) = scheduled.public.inbox.take_message() {

                    // Check for rerouting

                    if let Some(other_connector_id) = scheduled.router.should_reroute(message.sending_connector, message.receiving_port) {

                        self.runtime.send_message(other_connector_id, message);

                        continue;

                    }

                    // Check for messages that requires special action from the

                    // scheduler.

                    if let MessageContents::Control(content) = message.contents {

                        match content.content {

                            ControlMessageVariant::ChangePortPeer(port_id, new_target_connector_id) => {

                                // Need to change port target

                                let port = scheduled.context.get_port_mut(port_id);

                                port.peer_connector = new_target_connector_id;

                                // Note: for simplicity we program the scheduler to always finish

                                // running a connector with an empty outbox. If this ever changes

                                // then accepting the "port peer changed" message implies we need

                                // to change the recipient of the message in the outbox.

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

                                // And respond with an Ack

                                self.runtime.send_message(

                                    message.sending_connector,

                                    Message{

                                        sending_connector: connector_id,

                                        receiving_port: PortIdLocal::new_invalid(),

                                        contents: MessageContents::Control(ControlMessage{

                                            id: content.id,

                                            content: ControlMessageVariant::Ack,

        synchronous {

            print(\"putting!\");

            put(sender, true);

        }

        index += 1;

    }

}

primitive getter(in<bool> receiver, u32 loops) {

    u32 index = 0;

    while (index < loops) {

        synchronous {

            print(\"getting!\");

            auto result = get(receiver);

            assert(result);

        }

        index += 1;

    }

}

    ");

    let mut api = rt.create_interface();

    let channel = api.create_channel();

    api.create_connector("", "putter", ValueGroup::new_stack(vec![

        Value::Output(PortId(Id{ connector_id: 0, u32_suffix: channel.putter_id.index })),

        Value::UInt32(num_loops)

    ])).expect("create putter");

    api.create_connector("", "getter", ValueGroup::new_stack(vec![

        Value::Input(PortId(Id{ connector_id: 0, u32_suffix: channel.getter_id.index })),

        Value::UInt32(num_loops)

    ])).expect("create getter");

}