self.ports.push(Port{

            self_id, peer_comp_id, peer_port_id, kind, state,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Removes a port. Make sure you called `remove_peer` first.

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

        let port_index = self.must_get_port_index(port_handle);

        let port = self.ports.remove(port_index);

        debug_assert!(!port.associated_with_peer);

        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() {

                debug_assert_ne!(key.downgrade(), self.id); // should be upheld by the code that shuts down a component

                sched_ctx.runtime.destroy_component(key);

            }

        }

    }

    pub(crate) fn set_port_state(&mut self, port_handle: LocalPortHandle, new_state: PortState) {

        let port_info = self.get_port_mut(port_handle);

        debug_assert_ne!(port_info.state, PortState::Closed); // because then we do not expect to change the state

        port_info.state = new_state;

    }

    pub(crate) fn iter_ports(&self) -> impl Iterator<Item=&Port> {

        return self.ports.iter();

    }

    #[inline]

    pub(crate) fn iter_ports_mut(&mut self) -> impl Iterator<Item=&mut Port> {

        return self.ports.iter_mut();

    }

    #[inline]

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

        return self.peers.iter();

    }

    #[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!()

    }

    fn must_get_peer_index(&self, handle: LocalPeerHandle) -> usize {

        for (index, peer) in self.peers.iter().enumerate() {

            if peer.id == handle.0 {

                return index;

            }

        }

        unreachable!()

    }

    fn get_peer_index_by_id(&self, comp_id: CompId) -> Option<usize> {

        for (index, peer) in self.peers.iter().enumerate() {

        // to send the "unblock" message.

        if port_info.state == PortState::Blocked {

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

            let (peer_handle, message) = self.control.cancel_port_blocking(comp_ctx, port_handle);

            let peer_info = comp_ctx.get_peer(peer_handle);

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

        }

    }

    fn handle_incoming_control_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: ControlMessage) {

        // Little local utility to send an Ack

        fn send_control_ack_message(sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, causer_id: ControlId, peer_handle: LocalPeerHandle) {

            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) => {

                // The peer of our port has just changed. So we are asked to

                // temporarily block the port (while our original recipient is

                // potentially rerouting some of the in-flight messages) and

                // Ack. Then we wait for the `unblock` call.

                debug_assert_eq!(message.target_port_id, Some(port_id));

                let port_handle = comp_ctx.get_port_handle(port_id);

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

                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;

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

        port_info.state = PortState::Open;

        if self.mode == Mode::BlockedPut && port_id == self.mode_port {

            // We were blocked on the port that just became unblocked, so

            // send the message.

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

            let mut replacement = ValueGroup::default();

            std::mem::swap(&mut replacement, &mut self.mode_value);

            self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, replacement);

            self.mode = Mode::Sync;

        }

        // We'll now actually turn our reservation for a new component into an

        // actual component. Note that we initialize it as "not sleeping" as

        // its initial scheduling might be performed based on `Ack`s in response

        // to message exchanges between remote peers.

        let prompt = Prompt::new(

            &sched_ctx.runtime.protocol.types, &sched_ctx.runtime.protocol.heap,

            definition_id, monomorph_index, arguments,

        );

        let component = CompPDL::new(prompt, port_id_pairs.len());

        let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(

            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;

            while message_index < self.inbox_backup.len() {

                let message = &self.inbox_backup[message_index];

                if message.data_header.target_port == pair.creator_id {

                    // transfer message

                    let mut message = self.inbox_backup.remove(message_index);

                    message.data_header.target_port = pair.created_id;

                    component.code.inbox_backup.push(message);

                } else {

                    message_index += 1;

                }

            }

                        content.new_target_port,

                        content.new_target_comp

                    )

                };

                let to_ack = content.schedule_entry_id;

                return (

                    AckAction::SendMessage(target_comp_id, message_to_send),

                    Some(to_ack)

                );

            },

            ControlContent::ScheduleComponent(to_schedule) => {

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

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

    /// Adds an entry that, when completely ack'd, will schedule a component.

    pub(crate) fn add_schedule_entry(&mut self, to_schedule_id: CompId) -> ControlId {

        let entry_id = self.take_id();

        self.entries.push(ControlEntry{

            id: entry_id,

            ack_countdown: 0, // incremented by calls to `add_reroute_entry`

            content: ControlContent::ScheduleComponent(to_schedule_id),

        });

        return entry_id;

                source_comp: source_comp_id,

                old_target_port: old_target_port_id,

                new_target_port: new_target_port_id,

                new_target_comp: new_comp_id,

                schedule_entry_id,

            }),

        });

        // increment counter on schedule entry

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

        self.entries[entry_index].ack_countdown += 1;

        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,

            ack_countdown: 1,

            content: ControlContent::ClosedPort(port.self_id),

        });

        // Return the messages notifying peer of the closed port

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

        return (

            peer_handle,

            ControlMessage{

                id: entry_id,

                sender_comp_id: comp_ctx.id,

                target_port_id: Some(peer_port_id),