NonSync,

    SyncBusy,

    SyncAwaitingSolution,

    SelectBusy,

    SelectWait,

}

struct SolutionCombiner {

    solution: SyncPartialSolution,

    matched_channels: usize,

}

impl SolutionCombiner {

    fn new() -> Self {

        return Self {

            solution: SyncPartialSolution::default(),

            matched_channels: 0,

        }

    }

    #[inline]

    fn has_contributions(&self) -> bool {

        return !self.solution.channel_mapping.is_empty();

    }

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

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

    fn get_decision(&self) -> SyncRoundDecision {

        if self.matched_channels == self.solution.channel_mapping.len() {

            debug_assert_ne!(self.solution.decision, SyncRoundDecision::None);

            return self.solution.decision;

        }

        return SyncRoundDecision::None; // even in case of failure: wait for everyone.

    }

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

        debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);

        debug_assert_ne!(partial.decision, SyncRoundDecision::Solution);

        if partial.decision == SyncRoundDecision::Failure {

            self.solution.decision = SyncRoundDecision::Failure;

        }

        for entry in partial.channel_mapping {

            let channel_index = if entry.getter.is_some() && entry.putter.is_some() {

                let channel_index = self.solution.channel_mapping.len();

                self.solution.channel_mapping.push(entry);

                channel_index

            } else if let Some(putter) = entry.putter {

                self.combine_with_putter_port(putter)

            } else if let Some(getter) = entry.getter {

                self.combine_with_getter_port(getter)

            } else {

                unreachable!(); // both putter and getter are None

            };

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

            if let Some(consistent) = Self::channel_is_consistent(channel) {

                if !consistent {

                    self.solution.decision = SyncRoundDecision::Failure;

                }

                self.matched_channels += 1;

            }

        }

        self.update_solution();

    }

    /// Combines the currently stored global solution (if any) with the newly

    /// provided local solution. Make sure to check the `has_decision` return

    /// value afterwards.

    fn combine_with_local_solution(&mut self, _comp_id: CompId, solution: SyncLocalSolution) {

        debug_assert_ne!(self.solution.decision, SyncRoundDecision::Solution);

        // Combine partial solution with the local solution entries

        for entry in solution {

            // Match the current entry up with its peer endpoint, or add a new

            // entry.

            let channel_index = match entry {

                SyncLocalSolutionEntry::Putter(putter) => {

                    self.combine_with_putter_port(putter)

                },

                SyncLocalSolutionEntry::Getter(getter) => {

                    self.combine_with_getter_port(getter)

                }

            };

            // Check if channel is now consistent

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

            if let Some(consistent) = Self::channel_is_consistent(channel) {

                if !consistent {

                    self.solution.decision = SyncRoundDecision::Failure;

                }

                self.matched_channels += 1;

    /// Handles the arrival of a new data message (needs to be called for every

    /// new data message, even though it might not end up being received). This

    /// is used to determine peers of `get`ter ports.

    pub(crate) fn handle_new_data_message(&mut self, comp_ctx: &CompCtx, message: &DataMessage) {

        let target_handle = comp_ctx.get_port_handle(message.data_header.target_port);

        let target_index = comp_ctx.get_port_index(target_handle);

        let annotation = &mut self.ports[target_index];

        debug_assert!(

            !annotation.peer_discovered || (

                annotation.peer_comp_id == message.sync_header.sending_id &&

                annotation.peer_port_id == message.data_header.source_port

            )

        );

        annotation.peer_comp_id = message.sync_header.sending_id;

        annotation.peer_port_id = message.data_header.source_port;

        annotation.peer_discovered = true;

    }

    /// Checks if the data message can be received (due to port annotations), if

    /// it can then `true` is returned and the caller is responsible for handing

    /// the message of to the PDL code. Otherwise the message cannot be

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

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == message.data_header.target_port));

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

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

            // Determine our annotation, in order to do so we need to find the

            // port matching the peer ports

            let mut self_annotation = None;

            let mut self_annotation_found = false;

            match peer_port_kind {

                PortAnnotationKind::Putter(peer_port) => {

                    for self_port in &self.ports {

                        if self_port.kind == PortKind::Getter &&

                            self_port.peer_discovered &&

                            self_port.peer_comp_id == peer_port.self_comp_id &&

                            self_port.peer_port_id == peer_port.self_port_id

                        {

                            self_annotation = self_port.mapping;

                            self_annotation_found = true;

                            break;

                        }

                    }

                },

                PortAnnotationKind::Getter(peer_port) => {

                        // Peer indicates that we talked to it

                        let self_port_handle = comp_ctx.get_port_handle(peer_port.peer_port_id);

                        let self_port_index = comp_ctx.get_port_index(self_port_handle);

                        self_annotation = self.ports[self_port_index].mapping;

                        self_annotation_found = true;

                    }

                }

            }

            if !self_annotation_found {

                continue

            }

            if self_annotation != *expected_annotation {

                return false;

            }

        }

        // Expected mapping matches current mapping, so we will receive the message

        self.set_annotation(message.sync_header.sending_id, &message.data_header);

        // Handle the sync header embedded within the data message

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

        return true;

    }

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

    pub(crate) fn receive_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) -> SyncRoundDecision {

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

                return SyncRoundDecision::None;

            },

            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

                return SyncRoundDecision::Solution;

            },

            SyncMessageContent::GlobalFailure => {

        // multiple rounds, one message per round

        new sender(o_mrom, 5, 1);

        new receiver(i_mrom, 5, 1);

        // one round, multiple messages per round

        new sender(o_ormm, 1, 5);

        new receiver(i_ormm, 1, 5);

        // multiple rounds, multiple messages per round

        new sender(o_mrmm, 5, 5);

        new receiver(i_mrmm, 5, 5);

    }").expect("compilation");

    let rt = Runtime::new(3, true, pd);

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_intermediate_messenger() {

    let pd = ProtocolDescription::parse(b"

    primitive receiver<T>(in<T> rx, u32 num) {

        auto index = 0;

        while (index < num) {

            sync { auto v = get(rx); }

            index += 1;

        }

    }

    primitive middleman<T>(in<T> rx, out<T> tx, u32 num) {

        auto index = 0;

        while (index < num) {

            sync { put(tx, get(rx)); }

            index += 1;

        }

    }

    primitive sender<T>(out<T> tx, u32 num) {

        auto index = 0;

        while (index < num) {

            sync put(tx, 1337);

            index += 1;

        }

    }

    composite constructor_template<T>() {

        auto num = 0;

        channel<T> tx_a -> rx_a;

        channel tx_b -> rx_b;

        new receiver(rx_b, 3);

    }

    composite constructor() {

        new constructor_template<u16>();

    }

    ").expect("compilation");

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_simple_select() {

    let pd = ProtocolDescription::parse(b"

    func infinite_assert<T>(T val, T expected) -> () {

        while (val != expected) { print(\"nope!\"); }

        return ();

    }

    primitive receiver(in<u32> in_a, in<u32> in_b, u32 num_sends) {

        auto num_from_a = 0;

        auto num_from_b = 0;

        while (num_from_a + num_from_b < 2 * num_sends) {

            sync select {

                auto v = get(in_a) -> {

                    print(\"got something from A\");

                    auto _ = infinite_assert(v, num_from_a);

                    num_from_a += 1;

                }

                auto v = get(in_b) -> {

                    print(\"got something from B\");

                    auto _ = infinite_assert(v, num_from_b);

                    num_from_b += 1;

                }

            }

        }

    }

    primitive sender(out<u32> tx, u32 num_sends) {

        auto index = 0;

        while (index < num_sends) {

            sync {

                put(tx, index);

                index += 1;

            }

        }

    }

    composite constructor() {

        auto num_sends = 15;

        channel tx_a -> rx_a;

        channel tx_b -> rx_b;

        new sender(tx_a, num_sends);

        new receiver(rx_a, rx_b, num_sends);

        new sender(tx_b, num_sends);

    }