return port;

    }

    /// Adds a new peer. This must be called for every port, no matter the

    /// component the channel is connected to. If a `CompHandle` is supplied,

    /// 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 self_id = self.id;

        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;

        }

        dbg_code!(port.associated_with_peer = true);

        match self.get_peer_index_by_id(peer_comp_id) {

            Some(peer_index) => {

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

                peer.num_associated_ports += 1;

            },

            None => {

                let handle = match handle {

                    Some(handle) => handle.clone(),

                    None => sched_ctx.runtime.get_component_public(peer_comp_id)

                };

                self.peers.push(Peer{

                    id: peer_comp_id,

                    num_associated_ports: 1,

                    handle,

                });

            }

        }

    }

    /// Removes a peer associated with a port.

        let self_id = self.id;

        let port = self.get_port_mut(port_handle);

        debug_assert_eq!(port.peer_comp_id, peer_id);

            return;

        }

        debug_assert!(port.associated_with_peer);

        dbg_code!(port.associated_with_peer = false);

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

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

        peer.num_associated_ports -= 1;

        println!(" ****** DEBUG: Removed peer {:?} from {:?}, now at {}", peer.id, self_id, peer.num_associated_ports);

        if peer.num_associated_ports == 0 {

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

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

    }

    #[inline]

    pub(crate) fn num_ports(&self) -> usize {

        return self.ports.len();

    }

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

    // Local utilities

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

    #[inline]

    }

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

        for (index, port) in self.ports.iter().enumerate() {

            if port.self_id == handle.0 {

                return index;

            }

        }

        unreachable!()

    }

            let peer_info = comp_ctx.get_peer(peer_handle);

            peer_info.handle.send_message(sched_ctx, Message::Control(ControlMessage{

                id: causer_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: None,

                content: ControlMessageContent::Ack,

            }), true);

        }

        // Handle the content of the control message, and optionally Ack it

        match message.content {

            ControlMessageContent::Ack => {

            },

            ControlMessageContent::BlockPort(port_id) => {

                // On of our messages was accepted, but the port should be

                // blocked.

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

                debug_assert_eq!(port_info.kind, PortKind::Putter);

                if port_info.state != PortState::Closed {

                    comp_ctx.set_port_state(port_handle, PortState::Blocked);

                }

            },

            ControlMessageContent::ClosePort(port_id) => {

                // Request to close the port. We immediately comply and remove

                // the component handle as well

                let port_handle = comp_ctx.get_port_handle(port_id);

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

                let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

            },

            ControlMessageContent::UnblockPort(port_id) => {

                // We were previously blocked (or already closed)

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_info = comp_ctx.get_port(port_handle);

                debug_assert_eq!(port_info.kind, PortKind::Putter);

                debug_assert!(port_info.state == PortState::Blocked || port_info.state == PortState::Closed);

                if port_info.state == PortState::Blocked {

                    self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

                }

            },

            ControlMessageContent::PortPeerChangedBlock(port_id) => {

                let port_info = comp_ctx.get_port(port_handle);

                let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);

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

            },

            ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {

                let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());

                let port_info = comp_ctx.get_port(port_handle);

                debug_assert!(port_info.state == PortState::Blocked);

                let old_peer_id = port_info.peer_comp_id;

                let port_info = comp_ctx.get_port_mut(port_handle);

                port_info.peer_comp_id = new_comp_id;

                port_info.peer_port_id = new_port_id;

                comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);

                self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

            }

        }

    }

    fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {

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

        debug_assert!(self.mode == Mode::Sync || self.mode == Mode::SyncEnd);

        self.handle_sync_decision(sched_ctx, comp_ctx, decision);

        }

    }

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

    // Handling ports

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

    /// Unblocks a port, potentially continuing execution of the component, in

    /// response to a message that told us to unblock a previously blocked

    fn handle_unblock_port_instruction(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {

        let port_info = comp_ctx.get_port_mut(port_handle);

        let port_id = port_info.self_id;

            reservation, component, created_ctx, false,

        );

        let created_ctx = &component.ctx;

        // Now modify the creator's ports: remove every transferred port and

        // potentially remove the peer component. Here is also where we will

        // transfer messages in the main inbox.

        for pair in port_id_pairs.iter() {

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

            let creator_peer_comp_id = creator_port_info.peer_comp_id;

            creator_ctx.remove_port(pair.creator_handle);

            // Transfer any messages

            let created_port_index = created_ctx.get_port_index(pair.created_handle);

            let created_port_info = created_ctx.get_port(pair.created_handle);

            debug_assert!(component.code.inbox_main[created_port_index].is_none());

            if let Some(mut message) = self.inbox_main.remove(creator_port_index) {

                message.data_header.target_port = pair.created_id;

                component.code.inbox_main[created_port_index] = Some(message);

            }

            let mut message_index = 0;

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

                // schedule the component!

                return (AckAction::ScheduleComponent(to_schedule), None);

            },

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

            ControlContent::ClosedPort(closed_port) => {

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

                // that component.

                let port_handle = comp_ctx.get_port_handle(closed_port);

                let port_info = comp_ctx.get_port(port_handle);

                let port_peer_comp_id = port_info.peer_comp_id;

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

                return (AckAction::None, None);

            }

        }

    }

    pub(crate) fn has_acks_remaining(&self) -> bool {

        return !self.entries.is_empty();

    }

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

    // Port transfer (due to component creation)

        return Message::Control(ControlMessage{

            id: entry_id,

            sender_comp_id: creator_comp_id,

            target_port_id: Some(source_port_id),

            content: ControlMessageContent::PortPeerChangedBlock(source_port_id)

        })

    }

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

    // Blocking, unblocking, and closing ports

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

    /// Initiates the control message procedures for closing a port. Caller must

    /// make sure that the port state has already been set to `Closed`.

    pub(crate) fn initiate_port_closing(&mut self, port_handle: LocalPortHandle, comp_ctx: &CompCtx) -> (LocalPeerHandle, ControlMessage) {

        let port = comp_ctx.get_port(port_handle);

        let peer_port_id = port.peer_port_id;

        debug_assert!(port.state == PortState::Closed);

        // Construct the port-closing entry

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,