/// associated key out of the execution queue should (be able to) call this.

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

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

        unsafe {

            let connector = lock.connectors.get_mut(key.index as usize);

            debug_assert!(!connector.is_null());

            return *connector as &mut _;

        }

    }

    /// Create a new connector, returning the key that can be used to retrieve

        // Creation of the connector in the global store, requires a lock

        {

            let lock = self.inner.write().unwrap();

            let connector = ScheduledConnector {

                connector,

                context: ConnectorCtx::new(self.port_counter.clone()),

                public: ConnectorPublic::new(),

                router: Router::new(),

            };

            let index;

            if lock.free.is_empty() {

        // Setting of new connector's ID

        let key = ConnectorKey{ index: index as u32 };

        let new_connector = self.get_mut(&key);

        new_connector.context.id = key.downcast();

        // Transferring ownership of ports (and crashing if there is a

        // programmer's mistake in port management)

        match &new_connector.connector {

            ConnectorVariant::UserDefined(connector) => {

                for port_id in &connector.ports.owned_ports {

                    let mut port = created_by.context.remove_port(*port_id);

                    new_connector.context.add_port(port);

                }

            },

            ConnectorVariant::Native(_) => {}, // no initial ports (yet!)

        }

        return key;

    }

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

        let lock = self.inner.write().unwrap();

        return ConnectorScheduling::NotNow;

    }

}

/// The interface to a `ApplicationConnector`. This allows setting up the

/// interactions the `ApplicationConnector` performs within a synchronous round.

pub struct ApplicationInterface {

    sync_done: SyncDone,

    job_queue: JobQueue,

    runtime: Arc<RuntimeInner>,

    connector_id: ConnectorId,

}

impl ApplicationInterface {

    pub(crate) fn new(sync_done: SyncDone, job_queue: JobQueue, runtime: Arc<RuntimeInner1>) -> Self {

        return Self{

            sync_done, job_queue, runtime,

            connector_id: ConnectorId::new_invalid(),

            owned_ports: Vec::new(),

        }

    }

    /// Creates a new channel.

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

        // TODO: Duplicated logic in scheduler

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

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

        let getter_id = PortIdLocal::new(getter_id);

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            peer_connector: self.connector_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.

    // TODO: Optimize by yanking out scheduler logic for common use.

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

pub enum PortKind {

    Putter,

    Getter,

}

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

/// created by the application interfacing with the runtime, instead of being

/// created by a connector.

pub struct Port {

    pub self_id: PortIdLocal,

    pub peer_id: PortIdLocal,

    pub kind: PortKind,

    pub peer_connector: ConnectorId, // might be temporarily inconsistent while peer port is sent around in non-sync phase.

}

// TODO: Turn port ID into its own type

pub struct Channel {

    pub putter_id: PortIdLocal, // can put on it, so from the connector's point of view, this is an output

    pub getter_id: PortIdLocal, // vice versa: can get on it, so an input for the connector

}

/// accessed by the connector

pub(crate) struct ConnectorCtx {

    pub(crate) id: ConnectorId,

    port_counter: Arc<AtomicU32>,

    pub(crate) ports: Vec<Port>,

}

impl ConnectorCtx {

    pub(crate) fn new(port_counter: Arc<AtomicU32>) -> ConnectorCtx {

        Self{

            id: ConnectorId::new_invalid(),

            port_counter,

        }

    }

    /// Creates a (putter, getter) port pair belonging to the same channel. The

    /// port will be implicitly owned by the connector.

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

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

        self.ports.push(Port{

            self_id: getter_id,

            peer_id: putter_id,

            kind: PortKind::Getter,

            peer_connector: self.id,

        });

        self.ports.push(Port{

            self_id: putter_id,

            peer_id: getter_id,

            kind: PortKind::Putter,

            peer_connector: self.id,

        });

        return Channel{ getter_id, putter_id };

    }

    pub(crate) fn add_port(&mut self, port: Port) {

        debug_assert!(!self.ports.iter().any(|v| v.self_id == port.self_id));

        self.ports.push(port);

    }

    pub(crate) fn remove_port(&mut self, id: PortIdLocal) -> Port {