// Check for ports that are being sent

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

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

            self.workspace_ports.clear();

        }

        // Construct data header

        // TODO: Handle multiple firings. Right now we just assign the current

        //  branch to the `None` value because we know we can only send once.

        let data_header = DataHeader{

            sending_port: port_info.self_id,

            target_port: port_info.peer_id,

            new_mapping: branch.cur_marker,

        };

        // Update port mapping

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == port_info.channel_id {

                mapping.expected_firing = Some(true);

                mapping.registered_id = Some(branch.cur_marker);

            }

        }

        } else {

            // Free spot available

            index = self.free.pop().unwrap();

            key = ConnectorKey{ index: index as u32 };

            connector.ctx.id = key.downcast();

            unsafe {

                let target = self.connectors.get_mut(index);

                std::ptr::write(*target, connector);

            }

        }

        return key;

    }

    /// Destroys a connector. Caller should make sure it is not scheduled for

    /// execution. Otherwise one experiences "bad stuff" (tm).

    fn destroy(&mut self, key: ConnectorKey) {

        unsafe {

            let target = self.connectors.get_mut(key.index as usize);

            std::ptr::drop_in_place(*target);

            // Note: but not deallocating!

        }

        self.free.push(key.index as usize);

    }

}

impl Drop for ConnectorStore {

    fn drop(&mut self) {

        // Everything in the freelist already had its destructor called, so only

        // has to be deallocated

        for free_idx in self.free.iter().copied() {

            unsafe {

                let memory = self.connectors.get_mut(free_idx);

                let layout = std::alloc::Layout::for_value(&**memory);

        }

    }

    /// Handles inbox messages while shutting down. This intends to handle the

    /// case where a component cleanly exited outside of a sync region, but a

    /// peer, before receiving the `CloseChannel` message, sent a message inside

    /// a sync region. This peer should be notified that its message is not

    /// received by a component in a sync region.

    fn handle_inbox_while_shutting_down(&mut self, scheduled: &mut ScheduledConnector) {

        // Note: we're not handling the public inbox, we're dealing with the

        // private one!

        debug_assert!(scheduled.shutting_down);

            let target_port_and_round_number = match &message {

                Message::Data(msg) => Some((msg.data_header.target_port, msg.sync_header.sync_round)),

                Message::SyncComp(_) => None,

                Message::SyncPort(msg) => Some((msg.target_port, msg.sync_header.sync_round)),

                Message::SyncControl(_) => None,

                Message::Control(_) => None,

            };

            if let Some((target_port, sync_round)) = target_port_and_round_number {

                // This message is aimed at a port, but we're shutting down, so

                // notify the peer that its was not received properly.

                // (also: since we're shutting down, we're not in sync mode and

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

}

            messages: &self.inbox_messages,

            next_index: 0,

            max_index: self.inbox_len_read,

            match_port_id

        };

    }

    /// Retrieves the next unread message from the inbox `None` if there are no

    /// (new) messages to read.

    // TODO: Fix the clone of the data message, entirely unnecessary

    pub(crate) fn read_next_message(&mut self) -> Option<Message> {

        if !self.is_in_sync { return None; }

        if self.inbox_len_read == self.inbox_messages.len() { return None; }

        // We want to keep data messages in the inbox, because we need to check

        // them in the future. We don't want to keep sync messages around, we

        // should only handle them once. Control messages should never be in

        // here.

        let message = &self.inbox_messages[self.inbox_len_read];

        match &message {

            Message::Data(content) => {

                // Keep message in inbox for later reading

                self.inbox_len_read += 1;

                return Some(Message::Data(content.clone()));

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

// pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime

// pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed

// pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)

pub(crate) const NUM_THREADS: u32 = 1;

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

    run_test_in_runtime(CODE, |api| {

        api.create_connector("", "immediate_failure_outside_sync", ValueGroup::new_stack(Vec::new()))

            .expect("create component");

        api.create_connector("", "immediate_failure_inside_sync", ValueGroup::new_stack(Vec::new()))

            .expect("create component");

    })

}

#[test]

fn test_shared_sync_failure() {

    // Same as above. One of the components should fail, the other should follow

    const CODE: &'static str = "

    enum Location { BeforeSync, AfterPut, AfterGet, AfterSync, Never }

    primitive failing_at_location(in<bool> input, out<bool> output, Location loc) {

        u32[] failure_array = {};

        while (true) {

            if (loc == Location::BeforeSync) failure_array[0];

            sync {

                put(output, true);

                if (loc == Location::AfterPut) failure_array[0];

                auto received = get(input);

                assert(received);

                if (loc == Location::AfterGet) failure_array[0];

        }

    }

    composite constructor(Location loc) {

        channel output_a -> input_a;

        channel output_b -> input_b;

        new failing_at_location(input_a, output_b, Location::Never);

        new failing_at_location(input_b, output_a, loc);

    }

    ";

    run_test_in_runtime(CODE, |api| {

            // Create the channels

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

                Value::Enum(variant)

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

        }

    })

}