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

pub enum SyncRoundDecision {

    None,

    Solution,

    Failure,

}

#[derive(Debug)]

pub struct SyncPartialSolution {

    pub channel_mapping: Vec<SyncSolutionChannel>,

    pub decision: SyncRoundDecision,

}

impl Default for SyncPartialSolution {

    fn default() -> Self {

        return Self{

            channel_mapping: Vec::new(),

            decision: SyncRoundDecision::None,

        }

    }

}

#[derive(Debug)]

pub enum SyncMessageContent {

    NotificationOfLeader,

    LocalSolution(CompId, SyncLocalSolution), // local solution of the specified component

    PartialSolution(SyncPartialSolution), // partial solution of multiple components

    GlobalSolution,

    GlobalFailure,

}

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

// Control messages

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

#[derive(Debug)]

pub struct ControlMessage {

    pub(crate) id: ControlId,

    pub sender_comp_id: CompId,

    pub target_port_id: Option<PortId>,

    pub content: ControlMessageContent,

}

#[derive(Copy, Clone, Debug)]

pub enum ControlMessageContent {

    Ack,

    BlockPort(PortId),

    UnblockPort(PortId),

    PortPeerChangedBlock(PortId),

    PortPeerChangedUnblock(PortId, CompId),

}

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

// Messages (generic)

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

#[derive(Debug)]

pub struct MessageSyncHeader {

    pub sync_round: u32,

    pub sending_id: CompId,

    pub highest_id: CompId,

}

#[derive(Debug)]

pub enum Message {

    Data(DataMessage),

    Sync(SyncMessage),

    Control(ControlMessage),

    Poll,

}

impl Message {

    pub(crate) fn target_port(&self) -> Option<PortId> {

        match self {

            Message::Data(v) =>

                return Some(v.data_header.target_port),

            Message::Control(v) =>

                return v.target_port_id,

            Message::Sync(_) =>

                return None,

            Message::Poll =>

                return None,

        }

    }

    pub(crate) fn modify_target_port(&mut self, port_id: PortId) {

        match self {

            Message::Data(v) =>

                v.data_header.target_port = port_id,

            Message::Control(v) =>

                v.target_port_id = Some(port_id),

            Message::Sync(_) => unreachable!(), // should never be called for this message type

            Message::Poll => unreachable!(),

        }

    }

}

    }

}

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

/// user-defined component then we set up the PDL code state. Otherwise we

/// construct a custom component. This does NOT take care of port and message

/// management.

pub(crate) fn create_component(

    protocol: &ProtocolDescription,

    definition_id: ProcedureDefinitionId, type_id: TypeId,

    arguments: ValueGroup, num_ports: usize

) -> Box<dyn Component> {

    let definition = &protocol.heap[definition_id];

    debug_assert!(definition.kind == ProcedureKind::Primitive || definition.kind == ProcedureKind::Composite);

    if definition.source.is_builtin() {

        // Builtin component

        let component: Box<dyn Component> = match definition.source {

            ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),

            ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),

            _ => unreachable!(),

        };

        return component;

    } else {

        // User-defined component

        let prompt = Prompt::new(

            &protocol.types, &protocol.heap,

            definition_id, type_id, arguments

        );

        let component = CompPDL::new(prompt, num_ports);

        return Box::new(component);

    }

}

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

// Generic component messaging utilities (for sending and receiving)

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

/// Default handling of sending a data message. In case the port is blocked then

/// the `ExecState` will become blocked as well. Note that

/// `default_handle_control_message` will ensure that the port becomes

/// unblocked if so instructed by the receiving component. The returned

/// scheduling value must be used.

#[must_use]

pub(crate) fn default_send_data_message(

    exec_state: &mut CompExecState, transmitting_port_id: PortId, value: ValueGroup,

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

    debug_assert_eq!(exec_state.mode, CompMode::Sync);

    let port_handle = comp_ctx.get_port_handle(transmitting_port_id);

    let port_info = comp_ctx.get_port(port_handle);

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

        // Port is blocked, so we cannot send

        exec_state.set_as_blocked_put(transmitting_port_id, value);

    } else {

        // Port is not blocked, so send to the peer

        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(&sched_ctx.runtime, Message::Data(annotated_message), true);

    }

}

pub(crate) enum IncomingData {

    PlacedInSlot,

    SlotFull(DataMessage),

}

/// Default handling of receiving a data message. In case there is no room for

/// the message it is returned from this function. Note that this function is

/// different from PDL code performing a `get` on a port; this is the case where

/// the message first arrives at the component.

// NOTE: This is supposed to be a somewhat temporary implementation. It would be

//  nicest if the sending component can figure out it cannot send any more data.

#[must_use]

pub(crate) fn default_handle_incoming_data_message(

    exec_state: &mut CompExecState, port_value_slot: &mut Option<DataMessage>,

    comp_ctx: &mut CompCtx, incoming_message: DataMessage,

    sched_ctx: &SchedulerCtx, control: &mut ControlLayer

) -> IncomingData {

    let target_port_id = incoming_message.data_header.target_port;

    if port_value_slot.is_none() {

        // We can put the value in the slot

        *port_value_slot = Some(incoming_message);

        // Check if we're blocked on receiving this message.

        dbg_code!({

            // Our port cannot have been blocked itself, because we're able to

            // directly insert the message into its slot.

            let port_handle = comp_ctx.get_port_handle(target_port_id);

            assert!(!comp_ctx.get_port(port_handle).state.is_blocked());

        });

        if exec_state.is_blocked_on_get(target_port_id) {

            // Return to normal operation

            exec_state.mode = CompMode::Sync;

            exec_state.mode_port = PortId::new_invalid();

            debug_assert!(exec_state.mode_value.values.is_empty());

        }

        return IncomingData::PlacedInSlot

    } else {

        // Slot is already full, so if the port was previously opened, it will

        // now become closed

        let port_handle = comp_ctx.get_port_handle(target_port_id);

        let port_info = comp_ctx.get_port_mut(port_handle);

        debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked()); // i.e. not closed, but will go off if more states are added in the future

        return IncomingData::SlotFull(incoming_message)

    }

}

/// Default handling that has been received through a `get`. Will check if any

/// more messages are waiting, and if the corresponding port was blocked because

/// of full buffers (hence, will use the control layer to make sure the peer

/// will become unblocked).

pub(crate) fn default_handle_received_data_message(

    targeted_port: PortId, slot: &mut Option<DataMessage>, inbox_backup: &mut Vec<DataMessage>,

    comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer

) {

    debug_assert!(slot.is_none()); // because we've just received from it

    // Check if there are any more messages in the backup buffer

    let port_handle = comp_ctx.get_port_handle(targeted_port);

    let port_info = comp_ctx.get_port(port_handle);

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

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == targeted_port {

            // One more message, place it in the slot

            let message = inbox_backup.remove(message_index);

            debug_assert!(port_info.state.is_blocked()); // since we're removing another message from the backup

            *slot = Some(message);

            return;

        }

    }

    // Did not have any more messages, so if we were blocked, then we need to

    // unblock the port now (and inform the peer of this unblocking)

    if port_info.state == PortState::BlockedDueToFullBuffers {

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

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

    }

}

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

    match message.content {

        ControlMessageContent::Ack => {

            default_handle_ack(control, message.id, sched_ctx, comp_ctx);

        },

        ControlMessageContent::BlockPort(port_id) => {

            // One 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::Open {

                // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes

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

            }

        },

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

            // the component handle as well

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

                default_handle_ack(control, control_id, sched_ctx, comp_ctx);

            } else {

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

                comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed

            }

        },

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

            if port_info.state == PortState::BlockedDueToFullBuffers {

                default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

            }

        },

        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::BlockedDueToPeerChange);

            let port_info = comp_ctx.get_port(port_handle);

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

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

        },

        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::BlockedDueToPeerChange);

            let old_peer_id = port_info.peer_comp_id;

            comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);

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

            default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

        }

    }

}

/// Handles a component initiating the exiting procedure, and closing all of its

/// ports. Should only be called once per component (which is ensured by

/// checking and modifying the mode in the execution state).

#[must_use]

pub(crate) fn default_handle_start_exit(

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

    sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx

) -> CompScheduling {

    debug_assert_eq!(exec_state.mode, CompMode::StartExit);

    sched_ctx.log("Component starting exit");

    exec_state.mode = CompMode::BusyExit;

    // Iterating by index to work around borrowing rules

    for port_index in 0..comp_ctx.num_ports() {

        let port = comp_ctx.get_port_by_index_mut(port_index);

        if port.state == PortState::Closed {

            // Already closed, or in the process of being closed

            continue;

        }

        // Mark as closed

        let port_id = port.self_id;

        port.state = PortState::Closed;

        // Notify peer of closing

        let port_handle = comp_ctx.get_port_handle(port_id);

        let (peer, message) = control.initiate_port_closing(port_handle, comp_ctx);

        let peer_info = comp_ctx.get_peer(peer);

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

    }

    return CompScheduling::Immediate; // to check if we can shut down immediately

}

/// Handles a component waiting until all peers are notified that it is quitting

/// (i.e. after calling `default_handle_start_exit`).

#[must_use]

pub(crate) fn default_handle_busy_exit(

    exec_state: &mut CompExecState, control: &ControlLayer,

    sched_ctx: &SchedulerCtx

) -> CompScheduling {

    debug_assert_eq!(exec_state.mode, CompMode::BusyExit);

    if control.has_acks_remaining() {

        sched_ctx.log("Component busy exiting, still has `Ack`s remaining");

        return CompScheduling::Sleep;

    } else {

use crate::runtime2::scheduler::*;

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

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

#[derive(Debug)]

pub struct Port {

    pub self_id: PortId,

    pub peer_comp_id: CompId, // eventually consistent

    pub peer_port_id: PortId, // eventually consistent

    pub kind: PortKind,

    pub state: PortState,

    #[cfg(debug_assertions)] pub(crate) associated_with_peer: bool,

}

pub struct Peer {

    pub id: CompId,

    pub num_associated_ports: u32,

    pub(crate) handle: CompHandle,

}

/// Port and peer management structure. Will keep a local reference counter to

/// the ports associate with peers, additionally manages the atomic reference

/// counter associated with the peers' component handles.

pub struct CompCtx {

    pub id: CompId,

    ports: Vec<Port>,

    peers: Vec<Peer>,

    port_id_counter: u32,

}

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

pub struct LocalPortHandle(PortId);

#[derive(Copy, Clone)]

pub struct LocalPeerHandle(CompId);

impl CompCtx {

    /// Creates a new component context based on a reserved entry in the

    /// component store. This reservation is used such that we already know our

    /// assigned ID.

    pub(crate) fn new(reservation: &CompReserved) -> Self {

        return Self{

            id: reservation.id(),

            ports: Vec::new(),

            peers: Vec::new(),

            port_id_counter: 0,

        }

    }

    /// Creates a new channel that is fully owned by the component associated

    /// with this context.

    pub(crate) fn create_channel(&mut self) -> Channel {

        let putter_id = PortId(self.take_port_id());

        let getter_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id: putter_id,

            peer_port_id: getter_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            peer_comp_id: self.id,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_comp_id: self.id,

            #[cfg(debug_assertions)] associated_with_peer: false,

        });

        return Channel{ putter_id, getter_id };

    }

    /// Adds a new port. Make sure to call `add_peer` afterwards.

    pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {

        let self_id = PortId(self.take_port_id());

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

        dbg_code!(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);

        dbg_code!(assert!(!port.associated_with_peer));

        if !Self::requires_peer_reference(port, self_id, false) {

            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.

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

                self.broadcast_decision(sched_ctx, comp_ctx, round_decision);

            }

            return round_decision;

        } else {

            // Forward the partial solution

            let message = SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: SyncMessageContent::PartialSolution(solution),

            };

            self.send_to_leader(sched_ctx, comp_ctx, Message::Sync(message));

            return SyncRoundDecision::None;

        }

    }

    fn broadcast_decision(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, decision: SyncRoundDecision) {

        debug_assert_eq!(self.highest_id, comp_ctx.id);

        let is_success = match decision {

            SyncRoundDecision::None => unreachable!(),

            SyncRoundDecision::Solution => true,

            SyncRoundDecision::Failure => false,

        };

        let mut peers = Vec::with_capacity(self.solution.solution.channel_mapping.len()); // TODO: @Performance

        for channel in self.solution.solution.channel_mapping.iter() {

            let getter = channel.getter.as_ref().unwrap();

            if getter.self_comp_id != comp_ctx.id && !peers.contains(&getter.self_comp_id) {

                peers.push(getter.self_comp_id);

            }

            if getter.peer_comp_id != comp_ctx.id && !peers.contains(&getter.peer_comp_id) {

                peers.push(getter.peer_comp_id);

            }

        }

        for peer in peers {

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

            let message = Message::Sync(SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: if is_success { SyncMessageContent::GlobalSolution } else { SyncMessageContent::GlobalFailure },

            });

            handle.send_message(&sched_ctx.runtime, message, true);

            let _should_remove = handle.decrement_users();

            debug_assert!(_should_remove.is_none());

        }

    }

    fn send_to_leader(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx, message: Message) {

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

        leader_info.send_message(&sched_ctx.runtime, message, true);

        let should_remove = leader_info.decrement_users();

        if let Some(key) = should_remove {

            sched_ctx.runtime.destroy_component(key);

        }

    }

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

    // Creating message headers

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

    fn create_data_header_and_update_mapping(&mut self, port_info: &Port) -> MessageDataHeader {

        let mut expected_mapping = Vec::with_capacity(self.ports.len());

        let mut port_index = usize::MAX;

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

            if port.self_port_id == port_info.self_id {

                port_index = index; // remember for later updating

            }

            // Add all of the

            let annotation_kind = match port.kind {

                PortKind::Putter => {

                    PortAnnotationKind::Putter(PortAnnotationPutter{

                        self_comp_id: port.self_comp_id,

                        self_port_id: port.self_port_id

                    })

                },

                PortKind::Getter => {

                    if !port.peer_discovered {

                        continue;

                    }

                    PortAnnotationKind::Getter(PortAnnotationGetter{

                        self_comp_id: port.self_comp_id,

                        self_port_id: port.self_port_id,

                        peer_comp_id: port.peer_comp_id,

                        peer_port_id: port.peer_port_id,

                    })

                }

            };

            expected_mapping.push((annotation_kind, port.mapping));

        }

        let new_mapping = self.take_mapping();

        self.ports[port_index].mapping = Some(new_mapping);

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

        return MessageDataHeader{