}

}

pub(crate) struct CompPDL {

    pub mode: Mode,

    pub mode_port: PortId, // when blocked on a port

    pub mode_value: ValueGroup, // when blocked on a put

    pub prompt: Prompt,

    pub control: ControlLayer,

    pub consensus: Consensus,

    pub sync_counter: u32,

    pub exec_ctx: ExecCtx,

    // TODO: Temporary field, simulates future plans of having one storage place

    //  reserved per port.

    // Should be same length as the number of ports. Corresponding indices imply

    // message is intended for that port.

    pub inbox_main: Vec<Option<DataMessage>>,

    pub inbox_backup: Vec<DataMessage>,

}

impl CompPDL {

    pub(crate) fn new(initial_state: Prompt, num_ports: usize) -> Self {

        let mut inbox_main = Vec::new();

        inbox_main.reserve(num_ports);

        for _ in 0..num_ports {

            inbox_main.push(None);

        }

        return Self{

            mode: Mode::NonSync,

            mode_port: PortId::new_invalid(),

            mode_value: ValueGroup::default(),

            prompt: initial_state,

            control: ControlLayer::default(),

            consensus: Consensus::new(),

            sync_counter: 0,

            exec_ctx: ExecCtx{

                stmt: ExecStmt::None,

            },

            inbox_main,

            inbox_backup: Vec::new(),

        }

    }

    pub(crate) fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {

        sched_ctx.log(&format!("handling message: {:?}", message));

        if let Some(new_target) = self.control.should_reroute(&message) {

            let mut target = sched_ctx.runtime.get_component_public(new_target);

            let _should_remove = target.decrement_users();

            debug_assert!(!_should_remove);

            return;

        }

        match message {

            Message::Data(message) => {

                self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

            },

            Message::Control(message) => {

                self.handle_incoming_control_message(sched_ctx, comp_ctx, message);

            },

            Message::Sync(message) => {

                self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);

            }

        }

    }

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

    // Running component and handling changes in global component state

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

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

        use EvalContinuation as EC;

        let can_run = self.mode.can_run();

        sched_ctx.log(&format!("Running component (mode: {:?}, can run: {})", self.mode, can_run));

        if !can_run {

            return Ok(CompScheduling::Sleep);

        }

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

        match run_result {

            EC::Stepping => unreachable!(), // execute_prompt runs until this is no longer returned

            EC::BranchInconsistent | EC::NewFork | EC::BlockFires(_) => todo!("remove these"),

            // Results that can be returned in sync mode

            EC::SyncBlockEnd => {

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

                self.handle_sync_end(sched_ctx, comp_ctx);

                return Ok(CompScheduling::Immediate);

            },

            EC::BlockGet(port_id) => {

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

                debug_assert!(self.exec_ctx.stmt.is_none());

                let port_id = port_id_from_eval(port_id);

                let port_index = comp_ctx.get_port_index(port_id).unwrap();

        let port_info = &mut comp_ctx.ports[port_index];

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

        let _peer_comp_id = port_info.peer_comp_id;

        if port_info.state == PortState::Open {

            let (target_comp_id, block_message) =

                self.control.set_port_and_peer_blocked(target_port_id, comp_ctx);

            debug_assert_eq!(_peer_comp_id, target_comp_id);

            let peer = comp_ctx.get_peer(target_comp_id);

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

        }

        // But we still need to remember the message, so:

        self.inbox_backup.push(message);

    }

    /// Handles when a message has been handed off from the inbox to the PDL

    /// code. We check to see if there are more messages waiting and, if not,

    /// then we handle the case where the port might have been blocked

    /// previously.

    fn handle_received_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_id: PortId) {

        let port_index = comp_ctx.get_port_index(port_id).unwrap();

        debug_assert!(self.inbox_main[port_index].is_none()); // because we just received it

        // Check for any more messages

        for message_index in 0..self.inbox_backup.len() {

            let message = &self.inbox_backup[message_index];

            if message.data_header.target_port == port_id {

                // One more message for this port

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

                debug_assert_eq!(comp_ctx.get_port(port_id).state, PortState::Blocked); // since we had >1 message on the port

                self.inbox_main[port_index] = Some(message);

                return;

            }

        }

        // Did not have any more messages. So if we were blocked, then we need

        // to send the "unblock" message.

        let port_info = &comp_ctx.ports[port_index];

        if port_info.state == PortState::Blocked {

            let (peer_comp_id, message) = self.control.set_port_and_peer_unblocked(port_id, comp_ctx);

            let peer_info = comp_ctx.get_peer(peer_comp_id);

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

        match message.content {

            ControlMessageContent::Ack => {

                let mut to_ack = message.id;

                loop {

                    let action = self.control.handle_ack(to_ack, sched_ctx, comp_ctx);

                    match action {

                        AckAction::SendMessageAndAck(target_comp, message, new_to_ack) => {

                            // FIX @NoDirectHandle

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

                            to_ack = new_to_ack;

                        },

                        AckAction::ScheduleComponent(to_schedule) => {

                            // FIX @NoDirectHandle

                            wake_up_if_sleeping(sched_ctx, to_schedule, &handle);

                            break;

                        },

                        AckAction::None => {

                            break;

                        }

                    }

                }

            },

            ControlMessageContent::BlockPort(port_id) => {

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

                // blocked.

                let port_info = comp_ctx.get_port_mut(port_id);

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

                if port_info.state != PortState::Closed {

                    debug_assert_ne!(port_info.state, PortState::Blocked); // implies unnecessary messages

                    port_info.state = PortState::Blocked;

                }

            },

            ControlMessageContent::ClosePort(port_id) => {

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

                // the component handle as well

                let port_index = comp_ctx.get_port_index(port_id).unwrap();

                let port_info = &mut comp_ctx.ports[port_index];

                let peer_comp_id = port_info.peer_comp_id;

                port_info.state = PortState::Closed;

                let peer_index = comp_ctx.get_peer_index(peer_comp_id).unwrap();

                let peer_info = &mut comp_ctx.peers[peer_index];

                peer_info.num_associated_ports -= 1;

                if peer_info.num_associated_ports == 0 {

                    // TODO: @Refactor clean up all these uses of "num_associated_ports"

                    let should_remove = peer_info.handle.decrement_users();

                    if should_remove {

                        let comp_key = unsafe{ peer_info.id.upgrade() };

                        sched_ctx.runtime.destroy_component(comp_key);

                    }

                    comp_ctx.peers.remove(peer_index);

                }

            }

            ControlMessageContent::UnblockPort(port_id) => {

                // We were previously blocked (or already closed)

                let port_info = comp_ctx.get_port(port_id);

                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.unblock_local_port(sched_ctx, comp_ctx, port_id);

                }

            },

            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_info = comp_ctx.get_port_mut(port_id);

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

                if port_info.state == PortState::Open {

                    port_info.state = PortState::Blocked;

                }

            },

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

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

                let port_info = comp_ctx.get_port_mut(port_id);

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

                port_info.peer_comp_id = new_comp_id;

                self.unblock_local_port(sched_ctx, comp_ctx, port_id);

            }

        }

    }

    fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> Option<CompScheduling> {

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

        let is_success = match decision {

            SyncRoundDecision::None => {

                // No decision yet

                return None;

            },

            SyncRoundDecision::Solution => true,

            SyncRoundDecision::Failure => false,

        };

        // If here then we've reached a conclusion

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

        self.mode = Mode::NonSync;

        if is_success {

            // We can simply continue executing. So we do nothing extra!

        } else {

            todo!("handle this better, show some kind of error");

            self.handle_component_exit(sched_ctx, comp_ctx);

            return Some(CompScheduling::Exit);

        }

        return None;

    }

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

    // Handling ports

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

    /// Marks the local port as being unblocked. If the execution was blocked on

    /// sending a message over this port, then execution will continue and the

    /// message will be sent.

    fn unblock_local_port(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_id: PortId) {

        let port_info = comp_ctx.get_port_mut(port_id);

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

        port_info.state = PortState::Open;

        // Safety note: rerouting should occur during the time when we're

        // notifying a peer of a new component. During this period that

        // component hasn't been executed yet, so cannot have died yet.

        // FIX @NoDirectHandle

        let target_port = message.target_port();

        if target_port.is_none() {

            return None;

        }

        let target_port = target_port.unwrap();

        for entry in &self.entries {

            if let ControlContent::PeerChange(entry) = &entry.content {

                if entry.target_port == target_port {

                    return Some(entry.new_target_comp);

                }

            }

        }

        return None;

    }

    pub(crate) fn handle_ack(&mut self, entry_id: ControlId, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) -> AckAction {

        let entry_index = self.get_entry_index(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 {

            return AckAction::None;

        }

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

        match &entry.content {

            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

                // be unblocked again.

                let target_comp_id = content.source_comp;

                let message_to_send = ControlMessage{

                    id: ControlId::new_invalid(),

                    sender_comp_id: comp_ctx.id,

                    target_port_id: Some(content.source_port),

                    content: ControlMessageContent::PortPeerChangedUnblock(

                        content.source_port,

                        content.new_target_comp

                    )

                };

                let to_ack = content.schedule_entry_id;

                self.entries.remove(entry_index);

                return AckAction::SendMessageAndAck(target_comp_id, message_to_send, 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);

            },

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

            ControlContent::ClosedPort(port_id) => {

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

                // that component.

                let port_index = comp_ctx.get_port_index(*port_id).unwrap();

                debug_assert_eq!(comp_ctx.ports[port_index].state, PortState::Blocked);

                let peer_id = comp_ctx.ports[port_index].peer_comp_id;

                let peer_index = comp_ctx.get_peer_index(peer_id).unwrap();

                let peer_info = &mut comp_ctx.peers[peer_index];

                peer_info.num_associated_ports -= 1;

                if peer_info.num_associated_ports == 0 {

                    let should_remove = peer_info.handle.decrement_users();

                    if should_remove {

                        let comp_key = unsafe{ peer_info.id.upgrade() };

                        sched_ctx.runtime.destroy_component(comp_key);

                    }

                    comp_ctx.peers.remove(peer_index);

                }

                return AckAction::None;

            }

        }

    }

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

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