/// Representation of the generic operating mode of a component. Although not

/// every state may be used by every kind of (builtin) component, this allows

/// writing standard handlers for particular events in a component's lifetime.

#[derive(Debug, Copy, Clone, PartialEq, Eq)]

pub(crate) enum CompMode {

    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, // blocked because we need to receive a message on a particular port

    BlockedPut, // component is blocked because the port is blocked

    BlockedSelect, // waiting on message to complete the select statement

    NewComponentBlocked, // waiting until ports are in the appropriate state to create a new component

    StartExit, // temporary state: if encountered then we start the shutdown process.

    BusyExit, // temporary state: waiting for Acks for all the closed ports, potentially waiting for sync round to finish

    Exit, // exiting: shutdown process started, now waiting until the reference count drops to 0

}

impl CompMode {

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

        use CompMode::*;

        match self {

            Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect |

            NonSync | NewComponentBlocked | StartExit | BusyExit | Exit => false,

        }

    }

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

        use CompMode::*;

        match self {

            NonSync | Sync | SyncEnd | BlockedGet | BlockedPut | BlockedSelect |

                NewComponentBlocked => false,

            StartExit | BusyExit => true,

            Exit => false,

        }

    }

}

#[derive(Debug)]

pub(crate) enum ExitReason {

    Termination, // regular termination of component

    ErrorInSync,

    ErrorNonSync,

        return

            self.mode == CompMode::BlockedGet &&

            self.mode_port == port;

    }

    pub(crate) fn set_as_blocked_put_without_ports(&mut self, port: PortId, value: ValueGroup) {

        self.mode = CompMode::BlockedPut;

        self.mode_port = port;

        self.mode_value = value;

    }

    pub(crate) fn set_as_blocked_put_with_ports(&mut self, port: PortId, value: ValueGroup) {

        self.mode_port = port;

        self.mode_value = value;

    }

    pub(crate) fn is_blocked_on_put_without_ports(&self, port: PortId) -> bool {

        return

            self.mode == CompMode::BlockedPut &&

            self.mode_port == port;

    }

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

        return self.mode == CompMode::NewComponentBlocked;

    }

}

// TODO: Replace when implementing port sending. Should probably be incorporated

//  into CompCtx (and rename CompCtx into CompComms)

pub(crate) type InboxMain = Vec<Option<DataMessage>>;

pub(crate) type InboxMainRef = [Option<DataMessage>];

pub(crate) type InboxBackup = Vec<DataMessage>;

/// Creates a new component based on its definition. Meaning that if it is a

    let mut ports = Vec::new();

    find_ports_in_value_group(&value, &mut ports);

    if port_info.state.is_closed() {

        // Note: normally peer is eventually consistent, but if it has shut down

        // then we can be sure it is consistent (I think?)

        return Err((

            port_info.last_instruction,

            format!("Cannot send on this port, as the peer (id:{}) has shut down", port_info.peer_comp_id.0)

        ))

    } else if !ports.is_empty() {

            transmitting_port_id, port_instruction, value, exec_state,

            comp_ctx, sched_ctx, control

        )?;

        return Ok(CompScheduling::Sleep);

    } else if port_info.state.is_blocked() {

        // Port is blocked, so we cannot send

        exec_state.set_as_blocked_put_without_ports(transmitting_port_id, value);

        return Ok(CompScheduling::Sleep);

    } else {

        // Port is not blocked and no ports to transfer: send to the peer

/// Handles control messages in the default way. Note that this function may

/// take a lot of actions in the name of the caller: pending messages may be

/// sent, ports may become blocked/unblocked, etc. So the execution

/// (`CompExecState`), control (`ControlLayer`) and consensus (`Consensus`)

/// state may all change.

pub(crate) fn default_handle_control_message(

    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,

    message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) -> Result<(), (PortInstruction, String)> {

    match message.content {

        ControlMessageContent::Ack => {

        },

        ControlMessageContent::BlockPort => {

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

            // blocked.

            let port_to_block = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_block);

            let port_info = comp_ctx.get_port_mut(port_handle);

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

            port_info.state.set(PortStateFlag::BlockedDueToFullBuffers);

        },

        ControlMessageContent::ClosePort(content) => {

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

            port_info.close_at_sync_end = true; // might be redundant (we might set it closed now)

            let peer_comp_id = port_info.peer_comp_id;

            let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

            // One exception to sending an `Ack` is if we just closed the

            // port ourselves, meaning that the `ClosePort` messages got

            // sent to one another.

            if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {

                // The two components (sender and this component) are closing

                // the channel at the same time. So we don't care about the

                // content of the `ClosePort` message.

            } else {

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

                let last_instruction = port_info.last_instruction;

                let port_has_had_message = port_info.received_message_for_sync;

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

                comp_ctx.change_port_peer(sched_ctx, port_handle, None);

                // Handle any possible error conditions (which boil down to: the

                // port has been used, but the peer has died). If not in sync

                // mode then we close the port immediately.

            // We were previously blocked (or already closed)

            let port_to_unblock = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_unblock);

            let port_info = comp_ctx.get_port_mut(port_handle);

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

            debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToFullBuffers));

            port_info.state.clear(PortStateFlag::BlockedDueToFullBuffers);

            default_handle_recently_unblocked_port(

                exec_state, control, consensus, port_handle, sched_ctx,

                comp_ctx, inbox_main, inbox_backup

        },

        ControlMessageContent::PortPeerChangedBlock => {

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

            let port_to_change = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_change);

            let port_info = comp_ctx.get_port_mut(port_handle);

            let peer_comp_id = port_info.peer_comp_id;

            port_info.state.set(PortStateFlag::BlockedDueToPeerChange);

            let port_handle = comp_ctx.get_port_handle(port_to_change);

            let port_info = comp_ctx.get_port(port_handle);

            debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToPeerChange));

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_port_id = new_port_id;

            port_info.state.clear(PortStateFlag::BlockedDueToPeerChange);

            comp_ctx.change_port_peer(sched_ctx, port_handle, Some(new_comp_id));

            default_handle_recently_unblocked_port(

                exec_state, control, consensus, port_handle, sched_ctx,

                comp_ctx, inbox_main, inbox_backup

        }

    }

    return Ok(());

}

/// Handles a component entering the synchronous block. Will ensure that the

/// `Consensus` and the `ComponentCtx` are initialized properly.

pub(crate) fn default_handle_sync_start(

    exec_state: &mut CompExecState, inbox_main: &mut InboxMainRef,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus

) {

    let port_info = comp_ctx.get_port(sending_port_handle);

    debug_assert!(port_info.state.can_send());

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

    let peer_info = comp_ctx.get_peer(peer_handle);

    let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);

    peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);

}

/// Prepares sending a message that contains ports. Only once a particular

/// protocol has completed (where we notify all the peers that the ports will

/// be transferred) will we actually send the message to the recipient.

    sending_port_id: PortId, sending_port_instruction: PortInstruction, value: ValueGroup,

    exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,

    control: &mut ControlLayer

) -> Result<(), (PortInstruction, String)> {

    debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send

    let sending_port_handle = comp_ctx.get_port_handle(sending_port_id);

    let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);

    sending_port_info.last_instruction = sending_port_instruction;

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&value, &mut transmit_ports);

    // Set up the final Ack that triggers us to send our final message

    let unblock_put_control_id = control.add_unblock_put_with_ports_entry();

    for (_, port_id) in &transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(*port_id);

        let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);

        let peer_comp_id = transmit_port_info.peer_comp_id;

        let peer_port_id = transmit_port_info.peer_port_id;

        // Note: we checked earlier that we are currently in sync mode. Now we

        // will check if we've already used the port we're about to transmit.

        if !transmit_port_info.last_instruction.is_none() {

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as it is used in this sync round")

            ));

        }

        if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as that port is already transmitted")

            ));

        }

        // Set the flag for transmission

        transmit_port_info.state.set(PortStateFlag::Transmitted);

        // Block the peer of the port

        let message = control.create_port_transfer_message(unblock_put_control_id, comp_ctx.id, peer_port_id);

        println!("DEBUG: Port transfer message\nControl ID: {:?}\nMessage: {:?}", unblock_put_control_id, message);

        let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

        let peer_info = comp_ctx.get_peer(peer_handle);

        peer_info.handle.send_message_logged(sched_ctx, message, true);

    }

    // We've set up the protocol, once all the PPC's are blocked we are supposed

    // to transfer the message to the recipient. So store it temporarily

    return Ok(());

}

/// Performs the transmission of a data message that contains ports. These were

/// all stored in the component's execution state by the

/// `prepare_send_message_with_ports` function. Port must be ready to send!

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

    // Find all ports again

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&exec_state.mode_value, &mut transmit_ports);

    let port_handle = comp_ctx.get_port_handle(exec_state.mode_port);

    let port_info = comp_ctx.get_port(port_handle);

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

    let message_value = exec_state.mode_value.take();

    let mut annotated_message = consensus.annotate_data_message(comp_ctx, port_info, message_value);

    // And further enhance the message by adding data about the ports that are

    // being transferred

    for (port_locations, transmit_port_id) in transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(transmit_port_id);

        let transmit_port_info = comp_ctx.get_port(transmit_port_handle);

        let transmit_messages = take_port_messages(comp_ctx, transmit_port_id, inbox_main, inbox_backup);

        let mut transmit_port_state = transmit_port_info.state;

        debug_assert!(transmit_port_state.is_set(PortStateFlag::Transmitted));

            peer_comp: transmit_port_info.peer_comp_id,

            peer_port: transmit_port_info.peer_port_id,

            kind: transmit_port_info.kind,

            state: transmit_port_state

        });

        comp_ctx.change_port_peer(sched_ctx, transmit_port_handle, None);

    }

    // And finally, send the message to the peer

    let peer_info = comp_ctx.get_peer(peer_handle);

    peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);

}

/// Handles an `Ack` for the control layer.

fn default_handle_ack(

    exec_state: &mut CompExecState, control: &mut ControlLayer, control_id: ControlId,

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, consensus: &mut Consensus,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

    // Since an `Ack` may cause another one, handle them in a loop

    let mut to_ack = control_id;

    loop {

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

        match action {

            AckAction::SendMessage(target_comp, message) => {

                // FIX @NoDirectHandle

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

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

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

                // Note that the component is intentionally not

                // sleeping, so we just wake it up

                debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));

                let key = unsafe { to_schedule.upgrade() };

                sched_ctx.runtime.enqueue_work(key);

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::UnblockPutWithPorts => {

                // Send the message (containing ports) stored in the component

                // execution state to the recipient

                println!("DEBUG: Unblocking put with ports");

                let port_handle = comp_ctx.get_port_handle(exec_state.mode_port);

                default_handle_recently_unblocked_port(

                    exec_state, control, consensus, port_handle, sched_ctx,

                    comp_ctx, inbox_main, inbox_backup

            },

            AckAction::None => {}

        }

        match new_to_ack {

            Some(new_to_ack) => to_ack = new_to_ack,

            None => break,

        }

    }

}

/// Little helper for sending the most common kind of `Ack`

fn default_send_ack(

    causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,

    sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx

) {

    let peer_info = comp_ctx.get_peer(peer_handle);

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

        id: causer_of_ack_id,

        sender_comp_id: comp_ctx.id,

        target_port_id: None,

}

/// Handles the unblocking of a putter port. In case there is a pending message

/// on that port then it will be sent. There are two reasons for calling this

/// function: either a port was blocked (i.e. the Blocked state flag was

/// cleared), or the component is ready to send a message containing ports

/// (stored in the execution state). In this latter case we might still have

/// a blocked port.

fn default_handle_recently_unblocked_port(

    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,

    port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

    let port_info = comp_ctx.get_port_mut(port_handle);

    let port_id = port_info.self_id;

    if port_info.state.is_blocked() {

        // Port is still blocked. We wait until the next control message where

        // we unblock the port.

    }

    if exec_state.is_blocked_on_put_without_ports(port_id) {

        // Annotate the message that we're going to send

        let port_info = comp_ctx.get_port(port_handle); // for immutable access

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

        let to_send = exec_state.mode_value.take();

        let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);

        // Retrieve peer to send the message

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

        let peer_info = comp_ctx.get_peer(peer_handle);

        peer_info.handle.send_message_logged(sched_ctx, Message::Data(to_send), true);

        exec_state.mode = CompMode::Sync;

        exec_state.mode_port = PortId::new_invalid();

    } else if exec_state.is_blocked_on_create_component() {

        let mut ports = Vec::new();

        find_ports_in_value_group(&exec_state.mode_value, &mut ports);

        if ports_not_blocked(comp_ctx, &ports) {

            perform_create_component(

                exec_state, sched_ctx, comp_ctx, control, inbox_main, inbox_backup

            );

        }

    }

}

#[inline]

pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

}

#[inline]

pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {

    return EvalPortId{ id: port_id.0 };

}

            },

            Message::Poll => {

                sched_ctx.info("Received polling event");

            },

        }

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        sched_ctx.info(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));

        match self.exec_state.mode {

            CompMode::BlockedSelect |

            CompMode::NewComponentBlocked => {

                // Not possible: we never enter this state

                unreachable!();

            },

            CompMode::NonSync => {

                // When in non-sync mode

                match &mut self.socket_state {

                    SocketState::Connected(_socket) => {

                        if self.sync_state == SyncState::FinishSyncThenQuit {

                            // Previous request was to let the component shut down

                            self.exec_state.set_as_start_exit(ExitReason::Termination);

                        } else {

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        use EvalContinuation as EC;

        sched_ctx.info(&format!("Running component (mode: {:?})", self.exec_state.mode));

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.exec_state.mode {

            CompMode::NonSync | CompMode::Sync => {

                // continue and run PDL code

            },

            CompMode::SyncEnd | CompMode::BlockedGet |

            CompMode::NewComponentBlocked => {

                return CompScheduling::Sleep;

            }

            CompMode::StartExit => return component::default_handle_start_exit(

                &mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus

            ),

            CompMode::BusyExit => return component::default_handle_busy_exit(

                &mut self.exec_state, &self.control, sched_ctx

            ),

            CompMode::Exit => return component::default_handle_exit(&self.exec_state),

        }

                ) {

                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                }

            },

            Message::Poll => unreachable!(),

        }

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        sched_ctx.info(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));

        match self.exec_state.mode {

            CompMode::NewComponentBlocked => {

                // impossible for this component, no input ports and no select

                // blocks

                unreachable!();

            }

            CompMode::NonSync => {

                // If in non-sync mode then we check if the arguments make sense

                // (at some point in the future, this is just a testing

                // component).

                if self.random_minimum >= self.random_maximum {

                    // Could throw an evaluation error, but lets just panic

                    panic!("going to crash 'n burn your system now, please provide valid arguments");