pub content: SyncMessageContent,

}

pub struct SyncLocalSolutionEntry {

    pub self_port_id: PortId,

    pub peer_comp_id: CompId,

pub type SyncLocalSolution = Vec<SyncLocalSolutionEntry>;

pub struct SyncSolutionPort {

    pub self_comp_id: CompId,

    pub self_port_id: PortId,

    pub port_kind: PortKind,

}

pub struct SyncSolutionChannel {

    pub putter: Option<SyncSolutionPort>,

    pub getter: Option<SyncSolutionPort>,

}

    None,

    Solution,

    Failure,

}

pub struct SyncPartialSolution {

    pub submissions_by: Vec<(CompId, bool)>,

    pub channel_mapping: Vec<SyncSolutionChannel>,

    pub decision: SyncRoundDecision,

}

pub enum SyncMessageContent {

    NotificationOfLeader,

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

    SyncEnd, // awaiting a solution, i.e. encountered the end of the sync block

    BlockedGet,

    BlockedPut,

}

impl Mode {

        match self {

            Mode::NonSync | Mode::Sync =>

                return true,

                return false,

        }

    }

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

            return;

        }

        }

    }

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

        use EvalContinuation as EC;

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

                if let Some(message) = &self.inbox_main[port_index] {

                    // Check if we can actually receive the message

                        // Message was received. Make sure any blocked peers and

                        // pending messages are handled.

                        let message = self.inbox_main[port_index].take().unwrap();

            },

            // Results that can be returned outside of sync mode

            EC::ComponentTerminated => {

                return Ok(CompScheduling::Exit);

            },

            EC::SyncBlockStart => {

                    Value::Output(port_id_to_eval(channel.putter_id)),

                    Value::Input(port_id_to_eval(channel.getter_id))

                ));

                return Ok(CompScheduling::Immediate);

            }

        }

    }

    fn handle_sync_end(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx) {

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

        self.mode = Mode::SyncEnd;

    }

        let port_info = comp_ctx.get_port(source_port_id);

        let peer_info = comp_ctx.get_peer(port_info.peer_comp_id);

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

    }

    /// Handles a message that came in through the public inbox. This function

            debug_assert_eq!(_peer_comp_id, target_comp_id);

            let peer = comp_ctx.get_peer(target_comp_id);

        }

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

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

        }

    }

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

                            // FIX @NoDirectHandle

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

                            to_ack = new_to_ack;

                        },

                        AckAction::ScheduleComponent(to_schedule) => {

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

        }

    }

        }

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

                let peer_port_id = port_info.peer_id;

                let port_info = creator_ctx.get_port_mut(peer_port_id);

                port_info.peer_comp_id = created_ctx.id;

            } else {

                has_reroute_entry = true;

                let message = self.control.add_reroute_entry(

                    creator_ctx.id, port_info.peer_id, port_info.peer_comp_id,

                    port_info.self_id, created_ctx.id, schedule_entry_id

                );

                let peer_info = creator_ctx.get_peer(port_info.peer_comp_id);

            }

            // Transfer port and create temporary reroute entry

            let (port_info, peer_info) = Self::remove_port_from_component(creator_ctx, port_id);

            if port_info.state == PortState::Blocked {

            }

            Self::add_port_to_component(sched_ctx, created_ctx, port_info);

            // Maybe remove peer from the creator

            if let Some(mut peer_info) = peer_info {

                let remove_from_runtime = peer_info.handle.decrement_users();

        if !has_reroute_entry {

            // We can schedule the component immediately

            self.control.remove_schedule_entry(schedule_entry_id);

            sched_ctx.runtime.enqueue_work(comp_key);

        } // else: wait for the `Ack`s, they will trigger the scheduling of the component

    }

        return (port_info, Some(peer_info));

    }

    fn add_port_to_component(sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_info: Port) {

        // Add the port info

        let peer_comp_id = port_info.peer_comp_id;

        debug_assert!(!comp_ctx.ports.iter().any(|v| v.self_id == port_info.self_id));

        comp_ctx.ports.push(port_info);

            Some(peer_index) => {

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

                peer_info.num_associated_ports += 1;

            },

            None => {

                comp_ctx.peers.push(Peer{

                    num_associated_ports: 1,

                    handle,

                });

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

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

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

use super::component_pdl::*;

    }

}

enum Mode {

    NonSync,

    SyncBusy,

impl SolutionCombiner {

    fn new() -> Self {

        return Self {

            all_present: false,

        }

    }

    /// Returns a decision for the current round. If there is no decision (yet)

    /// then `RoundDecision::None` is returned.

        if self.all_present {

            return self.solution.decision;

        }

    }

    fn combine_with_partial_solution(&mut self, partial: SyncPartialSolution) {

            self.mark_single_component_submission(comp_id, present);

        }

        // Combine our partial solution with the provided partial solution.

        // This algorithm *could* allow overlap in the partial solutions, but

            // Make sure the new entry is consistent

            let channel = &self.solution.channel_mapping[channel_index];

            if !Self::channel_is_consistent(channel) {

            }

        }

        // Check to see if we have a global solution already

        self.update_all_present();

            dbg_code!(for entry in &self.solution.channel_mapping {

                    debug_assert!(entry.putter.is_some() && entry.getter.is_some());

                });

        }

    }

            self.mark_single_component_submission(entry.peer_comp_id, false);

        }

        // Go through all entries and check if the submitted local solution is

        // consistent with our partial solution

        let mut had_new_entry = false;

        for entry in solution.iter() {

            let preexisting_index = self.find_channel_index_for_local_entry(comp_id, entry);

                self_comp_id: comp_id,

                self_port_id: entry.self_port_id,

                peer_comp_id: entry.peer_comp_id,

                peer_port_id: entry.peer_port_id,

                mapping: entry.mapping,

            };

            match preexisting_index {

                    // the global solution. We'll handle any mismatches along

                    // the way.

                    let channel = &mut self.solution.channel_mapping[entry_index];

                            // Getter should be present in existing entry

                            debug_assert!(channel.getter.is_some() && channel.putter.is_none());

                            channel.putter = Some(new_port);

                            // Putter should be present in existing entry

                            debug_assert!(channel.putter.is_some() && channel.getter.is_none());

                            channel.getter = Some(new_port);

                    if !Self::channel_is_consistent(channel) {

                    }

                },

                None => {

                    // No entry yet. So add it

                    };

                    self.solution.channel_mapping.push(new_solution);

                    had_new_entry = true;

        if !had_new_entry {

            self.update_all_present();

                // No new entries and every component is present. This implies that

                // every component successfully added their local solutions to the

                // global solution. Hence: we have a global solution

                dbg_code!(for entry in &self.solution.channel_mapping {

                    debug_assert!(entry.putter.is_some() && entry.getter.is_some());

                });

        }

    }

    }

    fn mark_single_component_submission(&mut self, comp_id: CompId, will_contribute: bool) {

        debug_assert!(!will_contribute || !self.solution.submissions_by.iter().any(|(id, val)| *id == comp_id && *val)); // if submitting a solution, then we do not expect an existing entry

        for (entry, has_contributed) in self.solution.submissions_by.iter_mut() {

            // port B is part of its channel, but port B may consider port A not

            // to be part of its channel. Before merging the entries (outside of

            // this function) we'll make sure this is not the case.

                    // Expect getter to be present

                    if let Some(cur_entry) = &cur_entry.getter {

                        if might_belong_to_same_channel(cur_entry, comp_id, new_entry) {

                            return Some(entry_index);

                        }

                    }

                    if let Some(cur_entry) = &cur_entry.putter {

                        if might_belong_to_same_channel(cur_entry, comp_id, new_entry) {

                            return Some(entry_index);

                    }

                }

            }

        return None;

    }

    // Managing sync state

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

    pub(crate) fn notify_sync_start(&mut self, comp_ctx: &CompCtx) {

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

        self.highest_id = comp_ctx.id;

        self.make_ports_consistent_with_ctx(comp_ctx);

    }

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

        self.mode = Mode::SyncAwaitingSolution;

        return decision;

    }

    fn make_ports_consistent_with_ctx(&mut self, comp_ctx: &CompCtx) {

        let mut needs_setting_ports = false;

        if comp_ctx.ports.len() != self.ports.len() {

    pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {

        debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end

        let data_header = self.create_data_header_and_update_mapping(port_info);

        let sync_header = self.create_sync_header(comp_ctx);

    /// received.

    pub(crate) fn try_receive_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: &DataMessage) -> bool {

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

        // Make sure the expected mapping matches the currently stored mapping

        for (expected_id, expected_annotation) in &message.data_header.expected_mapping {

            let got_annotation = self.get_annotation(*expected_id);

                return false;

            }

        }

    }

    /// Receives the sync message and updates the consensus state appropriately.

        // Whatever happens: handle the sync header (possibly changing the

        // currently registered leader)

        self.handle_sync_header(sched_ctx, comp_ctx, &message.sync_header);

        match message.content {

            SyncMessageContent::NotificationOfLeader => {

            },

            SyncMessageContent::LocalSolution(solution_generator_id, local_solution) => {

                return self.handle_local_solution(sched_ctx, comp_ctx, solution_generator_id, local_solution);

            },

            SyncMessageContent::PartialSolution(partial_solution) => {

                return self.handle_partial_solution(sched_ctx, comp_ctx, partial_solution);

            SyncMessageContent::GlobalSolution => {

                debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution); // leader can only find global- if we submitted local solution

                todo!("clear port mapping or something");

            },

        }

    }

    fn handle_sync_header(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, header: &MessageSyncHeader) {

            // Sender knows of someone with a higher ID. So store highest ID,

            // notify all peers, and forward local solutions

            for peer in &comp_ctx.peers {

                if peer.id == header.sending_id {

                    continue;

                    sync_header: self.create_sync_header(comp_ctx),

                    content: SyncMessageContent::NotificationOfLeader,

                };

            }

            // Sender has a lower ID, so notify it of our higher one

            let message = SyncMessage{

                sync_header: self.create_sync_header(comp_ctx),

                content: SyncMessageContent::NotificationOfLeader,

            };

            let peer_info = comp_ctx.get_peer(header.sending_id);

        } // else: exactly equal

    }

    // Leader-related methods

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

        }

            sync_header: self.create_sync_header(comp_ctx),

        }));

    }

            return round_decision;

        } else {

            // Forward the solution

                content: SyncMessageContent::LocalSolution(solution_sender_id, solution),

            };

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

        }

    }

        if self.highest_id == comp_ctx.id {

            // We are the leader, combine existing and new solution

            self.solution.combine_with_partial_solution(solution);

            let round_decision = self.solution.get_decision();

        } else {

            // Forward the partial solution

            let message = SyncMessage{

                content: SyncMessageContent::PartialSolution(solution),

            };

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

        }

    }

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

        debug_assert_ne!(self.highest_id, comp_ctx.id); // we're not the leader

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

    }

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

    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;

            if port.id == port_info.self_id {

                port_index = index;

            }

        }

        let new_mapping = self.take_mapping();

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

        return MessageDataHeader{

            expected_mapping,

        };

    }

    fn create_sync_header(&self, comp_ctx: &CompCtx) -> MessageSyncHeader {

        return MessageSyncHeader{

            sending_id: comp_ctx.id,

            highest_id: self.highest_id,

        };

    }

    fn take_mapping(&mut self) -> u32 {

        let mapping = self.mapping_counter;

        self.mapping_counter = self.mapping_counter.wrapping_add(1);

        port.state = PortState::Closed;

        if peer_comp_id == comp_ctx.id {

            return None;

        }