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

        todo!("Remove fires() now");

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

        let annotation = self.consensus.get_annotation(self.branch_id, port_id);

        return annotation.expected_firing.map(|v| Value::Bool(v));

    }

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

        return match self.prepared.take() {

            PreparedStatement::None => None,

            PreparedStatement::CreatedChannel(ports) => Some(ports),

        };

    }

    fn performed_fork(&mut self) -> Option<bool> {

        return match self.prepared.take() {

            PreparedStatement::None => None,

            PreparedStatement::ForkedExecution(path) => Some(path),

            taken => unreachable!("prepared statement is '{:?}' during 'performed_fork()'", taken),

        };

    }

}

                self.tree.push_into_queue(QueueKind::Runnable, left_id);

                self.tree.push_into_queue(QueueKind::Runnable, right_id);

                let left_branch = &mut self.tree[left_id];

                left_branch.prepared = PreparedStatement::ForkedExecution(true);

                let right_branch = &mut self.tree[right_id];

                right_branch.prepared = PreparedStatement::ForkedExecution(false);

            }

            EvalContinuation::Put(port_id, content) => {

                // Branch is attempting to send data

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

                let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                }

                branch.prepared = PreparedStatement::PerformedPut;

                self.tree.push_into_queue(QueueKind::Runnable, branch_id);

                return ConnectorScheduling::Immediate;

            },

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

        }

        // If here then the run result did not require a particular action. We

        // return whether we have more active branches to run or not.

        if self.tree.queue_is_empty(QueueKind::Runnable) {

        // --- initiate discovery wave (to let leader know about all components)

        self.handled_wave = true;

        for mapping in &self.branch_annotations[0].channel_mapping {

            let channel_id = mapping.channel_id;

            let port_info = ctx.get_port_by_channel_id(channel_id).unwrap();

            let message = SyncPortMessage{

                sync_header: self.create_sync_header(ctx),

                source_port: port_info.self_id,

                target_port: port_info.peer_id,

                content: SyncPortContent::NotificationWave,

            };

        }

        return maybe_conclusion;

    }

    /// Notifies the consensus algorithm that a branch has reached the end of

    /// the sync block. A final check for consistency will be performed that the

    /// caller has to handle. Note that

    pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {

        debug_assert!(self.is_in_sync());

        let branch = &self.branch_annotations[branch_id.index as usize];

        for mapping in &branch.channel_mapping {

        let mut target_mapping = Vec::with_capacity(source_mapping.len());

        for port in source_mapping {

            // Note: if the port is silent, and we've never communicated

            // over the port, then we need to do so now, to let the peer

            // component know about our sync leader state.

            let port_desc = ctx.get_port_by_channel_id(port.channel_id).unwrap();

            let self_port_id = port_desc.self_id;

            let peer_port_id = port_desc.peer_id;

            let channel_id = port_desc.channel_id;

            if !self.encountered_ports.contains(&self_port_id) {

                    sync_header: SyncHeader{

                        sending_component_id: ctx.id,

                        highest_component_id: self.highest_connector_id,

                        sync_round: self.sync_round

                    },

                    source_port: self_port_id,

                    target_port: peer_port_id,

                    content: SyncPortContent::SilentPortNotification,

            }

            target_mapping.push((

                channel_id,

                port.registered_id.unwrap_or(BranchMarker::new_invalid())

            ));

        }

        let local_solution = LocalSolution{

            component: ctx.id,

            final_branch_id: branch_id,

            port_mapping: target_mapping,

                    let port_desc = ctx.get_port_by_channel_id(channel_id).unwrap();

                    if port_desc.self_id == message.target_port {

                        // Wave came from this port, no need to send one back

                        continue;

                    }

                    let message = SyncPortMessage{

                        sync_header: self.create_sync_header(ctx),

                        source_port: port_desc.self_id,

                        target_port: port_desc.peer_id,

                        content: SyncPortContent::NotificationWave,

                    };

                }

            }

        }

    }

    pub fn handle_new_sync_control_message(&mut self, message: SyncControlMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        if message.in_response_to_sync_round < self.sync_round {

            // Old message

            return None

        }

        match message.content {

            self.highest_connector_id = sync_header.highest_component_id;

            for peer in self.peers.iter() {

                if peer.id == sync_header.sending_component_id || !peer.encountered_this_round {

                    // Don't need to send it to this one

                    continue

                }

                let message = SyncCompMessage {

                    sync_header: self.create_sync_header(ctx),

                    target_component_id: peer.id,

                    content: SyncCompContent::Notification,

                };

            }

            // But also send our locally combined solution

            self.forward_local_data_to_new_leader(ctx);

        } else if sync_header.highest_component_id < self.highest_connector_id {

            // Sender has lower leader ID, so it should know about our higher

            // one.

            let message = SyncCompMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: sync_header.sending_component_id,

                content: SyncCompContent::Notification

            };

        } // else: exactly equal, so do nothing

        return true;

    }

    /// Handles a (potentially new) peer. Returns `false` if the provided sync

    /// number is different then the expected one.

    fn handle_peer(&mut self, sync_header: &SyncHeader) -> bool {

        let position = self.peers.iter().position(|v| v.id == sync_header.sending_component_id);

        match position {

            Some(index) => {

                let entry = &mut self.peers[index];

                SyncCompContent::Notification | SyncCompContent::GlobalSolution(_) |

                SyncCompContent::GlobalFailure => {

                    unreachable!("unexpected message content for leader");

                },

            }

        } else {

            // Someone else is the leader

            let message = SyncCompMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: self.highest_connector_id,

                content,

            };

        }

        return None;

    }

    fn handle_global_solution_as_leader(&mut self, global_solution: GlobalSolution, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        if self.conclusion.is_some() {

            return None;

        }

        // Handle the global solution

        let mut my_final_branch_id = BranchId::new_invalid();

        for (connector_id, branch_id) in global_solution.component_branches.iter().copied() {

            if connector_id == ctx.id {

                // This is our solution branch

                my_final_branch_id = branch_id;

                continue;

            }

            let message = SyncCompMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: connector_id,

                content: SyncCompContent::GlobalSolution(global_solution.clone()),

            };

        }

        debug_assert!(my_final_branch_id.is_valid());

        self.conclusion = Some(RoundConclusion::Success(my_final_branch_id));

        return Some(RoundConclusion::Success(my_final_branch_id));

    }

    fn handle_global_failure_as_leader(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        debug_assert!(self.solution_combiner.failure_reported && self.solution_combiner.check_for_global_failure());

        if self.conclusion.is_some() {

            return None;

        }

        // TODO: Performance

        let mut encountered = VecSet::new();

        for presence in &self.solution_combiner.presence {

            if presence.added_by != ctx.id {

                // Did not add it ourselves

                if encountered.push(presence.added_by) {

                    // Not yet sent a message

                    let message = SyncCompMessage{

                        sync_header: self.create_sync_header(ctx),

                        target_component_id: presence.added_by,

                        content: SyncCompContent::GlobalFailure,

                    };

                }

            }

        }

        self.conclusion = Some(RoundConclusion::Failure);

        if encountered.is_empty() {

            // We don't have to wait on Acks

            return Some(RoundConclusion::Failure);

        } else {

            return None;

        }

    }

        }

    }

    fn forward_local_data_to_new_leader(&mut self, ctx: &mut ComponentCtx) {

        debug_assert_ne!(self.highest_connector_id, ctx.id);

        if let Some(partial_solution) = self.solution_combiner.drain() {

            let message = SyncCompMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: self.highest_connector_id,

                content: SyncCompContent::PartialSolution(partial_solution),

            };

        }

    }

}

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

// Solution storage and algorithms

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

// TODO: Remove all debug derives

#[derive(Debug, Clone)]

struct MatchedLocalSolution {

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

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

}

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)

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

        }

    }

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

        }

    })

}