/// then it will be used to add the peer. Otherwise it will be retrieved

    /// from the runtime using its ID.

    pub(crate) fn add_peer(&mut self, port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, peer_comp_id: CompId, handle: Option<&CompHandle>) {

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_comp_id);

        debug_assert!(!port.associated_with_peer);

            return;

        }

    /// Removes a peer associated with a port.

    pub(crate) fn remove_peer(&mut self, sched_ctx: &SchedulerCtx, port_handle: LocalPortHandle, peer_id: CompId) {

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_id);

            return;

        }

        let peer_index = self.get_peer_index_by_id(peer_id).unwrap();

        let peer = &mut self.peers[peer_index];

        peer.num_associated_ports -= 1;

        if peer.num_associated_ports == 0 {

            let mut peer = self.peers.remove(peer_index);

            if let Some(key) = peer.handle.decrement_users() {

        return &mut self.ports[index];

    }

    pub(crate) fn get_peer(&self, peer_handle: LocalPeerHandle) -> &Peer {

        let index = self.must_get_peer_index(peer_handle);

        return &self.peers[index];

        return self.ports.iter();

    }

    #[inline]

    pub(crate) fn iter_peers(&self) -> impl Iterator<Item=&Peer> {

        return self.peers.iter();

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

    #[inline]

    }

    fn must_get_port_index(&self, handle: LocalPortHandle) -> usize {

    EvalContinuation, EvalResult, EvalError

};

use crate::runtime2::scheduler::SchedulerCtx;

use crate::runtime2::communication::*;

use super::component_context::*;

use super::control_layer::*;

use super::consensus::Consensus;

pub(crate) enum Mode {

    NonSync, // not in sync mode

    Sync, // in sync mode, can interact with other components

    SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block

    BlockedGet,

    BlockedPut,

}

pub(crate) struct CompPDL {

    pub(crate) fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {

        use EvalContinuation as EC;

        }

        let run_result = self.execute_prompt(&sched_ctx)?;

            // Results that can be returned in sync mode

            EC::SyncBlockEnd => {

                debug_assert_eq!(self.mode, Mode::Sync);

            },

            EC::BlockGet(port_id) => {

                debug_assert_eq!(self.mode, Mode::Sync);

                        self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);

                        return Ok(CompScheduling::Immediate);

                    } else {

                        return Ok(CompScheduling::Sleep);

                    }

                } else {

            },

            // Results that can be returned outside of sync mode

            EC::ComponentTerminated => {

            },

            EC::SyncBlockStart => {

                debug_assert_eq!(self.mode, Mode::NonSync);

        self.mode = Mode::Sync;

    }

        sched_ctx.log("Component ending sync mode (now waiting for solution)");

        let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);

        self.mode = Mode::SyncEnd;

    }

        debug_assert_eq!(self.mode, Mode::SyncEnd);

        sched_ctx.log(&format!("Handling sync decision: {:?}", decision));

        let is_success = match decision {

            SyncRoundDecision::None => {

                // No decision yet

            },

            SyncRoundDecision::Solution => true,

            SyncRoundDecision::Failure => false,

        };

        // If here then we've reached a decision

        if is_success {

            self.consensus.notify_sync_decision(decision);

        } else {

        }

    }

    fn handle_component_exit(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {

        sched_ctx.log("Component exiting");

    }

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

            ControlMessageContent::Ack => {

                let mut to_ack = message.id;

                loop {

                    match action {

                            // FIX @NoDirectHandle

                            let mut handle = sched_ctx.runtime.get_component_public(target_comp);

                            handle.send_message(sched_ctx, Message::Control(message), true);

                            let _should_remove = handle.decrement_users();

                            debug_assert!(_should_remove.is_none());

                        },

                        AckAction::ScheduleComponent(to_schedule) => {

                            // FIX @NoDirectHandle

                            sched_ctx.runtime.enqueue_work(key);

                            let _should_remove = handle.decrement_users();

                            debug_assert!(_should_remove.is_none());

                        },

                    }

                }

            },

                let peer_comp_id = comp_ctx.get_port(port_handle).peer_comp_id;

                let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

                send_control_ack_message(sched_ctx, comp_ctx, message.id, peer_handle);

                comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id);

            },

            ControlMessageContent::UnblockPort(port_id) => {

                // We were previously blocked (or already closed)

                    },

                    None => {

                        // Peer port remains with creator component.

                        created_port_info.peer_comp_id = creator_ctx.id;

                        created_ctx.add_peer(pair.created_handle, sched_ctx, creator_ctx.id, None);

                    }

            // Remove peer if appropriate

            let creator_port_info = creator_ctx.get_port(pair.creator_handle);

            let creator_port_index = creator_ctx.get_port_index(pair.creator_handle);

            creator_ctx.remove_port(pair.creator_handle);

            // Transfer any messages

                let peer_port_handle = creator_ctx.get_port_handle(created_port_info.peer_port_id);

                let peer_port_info = creator_ctx.get_port_mut(peer_port_handle);

                peer_port_info.peer_comp_id = created_ctx.id;

            }

        }

use crate::runtime2::runtime::*;

use crate::runtime2::communication::*;

use super::component_context::*;

pub struct PortAnnotation {

            });

            handle.send_message(sched_ctx, message, true);

            let _should_remove = handle.decrement_users();

        }

    }

        let mut leader_info = sched_ctx.runtime.get_component_public(self.highest_id);

        leader_info.send_message(sched_ctx, message, true);

        let should_remove = leader_info.decrement_users();

            sched_ctx.runtime.destroy_component(key);

        }

    }

    schedule_entry_id: ControlId,

}

pub(crate) enum AckAction {

    None,

    ScheduleComponent(CompId),

}

        return None;

    }

        let entry_index = self.get_entry_index_by_id(entry_id).unwrap();

        let entry = &mut self.entries[entry_index];

        debug_assert!(entry.ack_countdown > 0);

        entry.ack_countdown -= 1;

        if entry.ack_countdown != 0 {

        }

        // All `Ack`s received, take action based on the kind of entry

            ControlContent::PeerChange(content) => {

                // If change of peer is ack'd. Then we are certain we have

                // rerouted all of the messages, and the sender's port can now

                    )

                };

                let to_ack = content.schedule_entry_id;

            },

            ControlContent::ScheduleComponent(to_schedule) => {

                // If all change-of-peers are `Ack`d, then we're ready to

                // schedule the component!

            },

            ControlContent::BlockedPort(_) => unreachable!(),

                // If a closed port is Ack'd, then we remove the reference to

                // that component.

                let port_info = comp_ctx.get_port(port_handle);

                debug_assert_eq!(port_info.state, PortState::Closed);

            }

        }

    }

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

    // Port transfer (due to component creation)

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

        });

        // increment counter on schedule entry

        return Message::Control(ControlMessage{

            id: entry_id,

    // Blocking, unblocking, and closing ports

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

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 1,

        });

            ControlMessage{

                id: entry_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(peer_port_id),

                content: ControlMessageContent::ClosePort(peer_port_id),

            }

    }

    /// Adds a control entry to track that a port is blocked. Expects the caller

    pub public: CompPublic,

    pub code: CompPDL,

    pub ctx: CompCtx,

}

/// Should contain everything that is accessible in a thread-safe manner

    /// Returns the `CompKey` to the component if it should be destroyed

    pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {

        debug_assert!(!self.decremented, "illegal to 'decrement_users' twice");

        let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);

        let new_count = old_count - 1;

        if new_count == 0 {

            return Some(unsafe{ self.id.upgrade() });

        }

            code: component,

            ctx: context,

            inbox: inbox_queue,

        };

        let index = self.components.submit(reserved.reservation, component);

            code: comp,

            ctx,

            inbox: inbox_queue,

        };

        let index = self.components.create(comp);

    }

    pub(crate) fn destroy_component(&self, key: CompKey) {

        self.decrement_active_components();

        self.components.destroy(key.0);

    }

        if new_val == 0 {

            // Just to be sure, in case the last thing that gets destroyed is an

            // API instead of a thread.

            self.work_condvar.notify_all();

        }

    }

use super::component::*;

use super::runtime::*;

/// Data associated with a scheduler thread

pub(crate) struct Scheduler {

    // local utilities

    fn mark_component_as_sleeping(&self, key: CompKey, component: &mut RuntimeComp) {

        debug_assert_eq!(key.downgrade(), component.ctx.id); // make sure component matches key

        debug_assert_eq!(component.public.sleeping.load(Ordering::Acquire), false); // we're executing it, so it cannot be sleeping

        }

    }

    fn mark_component_as_exiting(&self, sched_ctx: &SchedulerCtx, component: &mut RuntimeComp) {

            }

        }

    }

}

    pub fn reserve(&self) -> ComponentReservation {

        let lock = self.inner.lock_shared();

        return ComponentReservation::new(index);

    }