};

use super::scheduler::{ComponentCtx, ComponentPortChange, MessageTicket};

struct BranchAnnotation {

    channel_mapping: Vec<ChannelAnnotation>,

    cur_marker: BranchMarker,

}

#[derive(Debug)]

pub(crate) struct LocalSolution {

    component: ConnectorId,

    final_branch_id: BranchId,

    port_mapping: Vec<(ChannelId, BranchMarker)>,

}

#[derive(Debug, Clone)]

pub(crate) struct GlobalSolution {

    channel_mapping: Vec<(ChannelId, BranchMarker)>, // TODO: This can go, is debugging info

}

#[derive(Debug, PartialEq, Eq)]

pub enum RoundConclusion {

    Failure,

    Success(BranchId),

}

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

// Consensus

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

struct Peer {

    id: ConnectorId,

    encountered_this_round: bool,

    expected_sync_round: u32,

}

/// The consensus algorithm. Currently only implemented to find the component

/// with the highest ID within the sync region and letting it handle all the

/// local solutions.

///

/// The type itself serves as an experiment to see how code should be organized.

// TODO: Flatten all datastructures

                    }

                }

            }

            target_mapping.push((

                channel_id,

                port.registered_id.unwrap_or(BranchMarker::new_invalid())

            ));

        }

        let local_solution = LocalSolution{

            component: ctx.id,

            final_branch_id: branch_id,

            port_mapping: target_mapping,

        };

        let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalSolution(local_solution), ctx);

        return maybe_conclusion;

    }

    /// Notifies the consensus algorithm about the chosen branch to commit to

    /// memory (may be the invalid "start" branch)

    pub fn end_sync(&mut self, branch_id: BranchId, final_ports: &mut Vec<ComponentPortChange>) {

        debug_assert!(self.is_in_sync());

        self.solution_combiner.clear();

        self.handled_wave = false;

        self.conclusion = None;

        self.ack_remaining = 0;

        // And modify persistent storage

        self.sync_round += 1;

        for peer in self.peers.iter_mut() {

            peer.encountered_this_round = false;

            peer.expected_sync_round += 1;

        }

    }

    // --- Handling messages

    /// Prepares a message for sending. Caller should have made sure that

    /// sending the message is consistent with the speculative state.

    pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtx) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

        let branch = &mut self.branch_annotations[branch_id.index as usize];

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        if cfg!(debug_assertions) {

        match &message.content {

            SyncCompContent::LocalFailure |

            SyncCompContent::LocalSolution(_) |

            SyncCompContent::PartialSolution(_) |

            SyncCompContent::AckFailure |

            SyncCompContent::Presence(_) => {

                // Needs to be handled by the leader

                return self.send_to_leader_or_handle_as_leader(message.content, ctx);

            },

            SyncCompContent::GlobalSolution(solution) => {

                // Found a global solution

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                    .unwrap();

                return Some(RoundConclusion::Success(*branch_id));

            },

            SyncCompContent::GlobalFailure => {

                // Global failure of round, send Ack to leader

                println!("DEBUGERINO: Got GlobalFailure, sending Ack in response");

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                let _result = self.send_to_leader_or_handle_as_leader(SyncCompContent::AckFailure, ctx);

                debug_assert!(_result.is_none());

                return Some(RoundConclusion::Failure);

            },

            SyncCompContent::Notification => {

        }

        return None;

    }

    fn handle_global_solution_as_leader(&mut self, global_solution: GlobalSolution, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        if self.conclusion.is_some() {

            return None;

        }

        // Handle the global solution

        let mut my_final_branch_id = BranchId::new_invalid();

            if connector_id == ctx.id {

                // This is our solution branch

                my_final_branch_id = branch_id;

                continue;

            }

            let message = SyncCompMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: connector_id,

                content: SyncCompContent::GlobalSolution(global_solution.clone()),

            };

            ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel

        }

        debug_assert!(my_final_branch_id.is_valid());

        self.conclusion = Some(RoundConclusion::Success(my_final_branch_id));

        return Some(RoundConclusion::Success(my_final_branch_id));

    }

    fn handle_global_failure_as_leader(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        debug_assert!(self.solution_combiner.failure_reported && self.solution_combiner.check_for_global_failure());

        if self.conclusion.is_some() {

            // Already sent out a failure

            return None;

        for presence in &self.solution_combiner.presence {

            if presence.owner_a != ctx.id {

                // Did not add it ourselves

                if encountered.push(presence.owner_a) {

                    // Not yet sent a message

                    let message = SyncCompMessage{

                        sync_header: self.create_sync_header(ctx),

                        target_component_id: presence.owner_a,

                        content: SyncCompContent::GlobalFailure,

                    };

                    ctx.submit_message(Message::SyncComp(message)).unwrap(); // unwrap: sending to component instead of through channel

                }

                if owner_b != ctx.id {

                    if encountered.push(owner_b) {

                        let message = SyncCompMessage{

                            sync_header: self.create_sync_header(ctx),

                            target_component_id: owner_b,

                            content: SyncCompContent::GlobalFailure,

                        };

                        ctx.submit_message(Message::SyncComp(message)).unwrap();

                    }

                }

            }

        }

}

#[derive(Debug, Clone)]

struct ComponentPeer {

    target_id: ConnectorId,

    target_index: usize, // in array of global solution components

    involved_channels: Vec<ChannelId>,

}

#[derive(Debug, Clone)]

struct ComponentLocalSolutions {

    component: ConnectorId,

    peers: Vec<ComponentPeer>,

    solutions: Vec<MatchedLocalSolution>,

    all_peers_present: bool,

}

#[derive(Debug, Clone)]

pub(crate) struct ComponentPresence {

    component_id: ConnectorId,

    channels: Vec<LocalChannelPresence>,

}

#[derive(Debug, Clone)]

        return Self{

            local: Vec::new(),

            presence: Vec::new(),

            failure_reported: false,

        };

    }

    /// Adds a new local solution to the global solution storage. Will check the

    /// new local solutions for matching against already stored local solutions

    /// of peer connectors.

    fn add_solution_and_check_for_global_solution(&mut self, solution: LocalSolution) -> Option<GlobalSolution> {

        let component_id = solution.component;

        let solution = MatchedLocalSolution{

            final_branch_id: solution.final_branch_id,

            channel_mapping: solution.port_mapping,

            matches: Vec::new(),

        };

        // Create an entry for the solution for the particular component

        let component_exists = self.local.iter_mut()

            .enumerate()

            .find(|(_, v)| v.component == component_id);

        let (component_index, solution_index, new_component) = match component_exists {

            Some((component_index, storage)) => {

                // Entry for component exists, so add to solutions

                let solution_index = storage.solutions.len();

                storage.solutions.push(solution);

                (component_index, solution_index, false)

            }

            None => {

                // Entry for component does not exist yet

                let component_index = self.local.len();

                self.local.push(ComponentLocalSolutions{

                    component: component_id,

                    peers: Vec::new(),

                    solutions: vec![solution],

                    all_peers_present: false,

                });

                (component_index, 0, true)

            }

        };

        // If this is a solution of a component that is new to us, then we check

        // in the stored solutions which other components are peers of the new

        // one.

        if new_component {

            // Stack length is 1, hence we're back at our initial solution.

            // Since that doesn't yield a global solution, we simply:

            return None;

        }

        // Constructing the representation of the global solution

        debug_assert_eq!(stack.len(), self.local.len());

        let mut final_branches = Vec::with_capacity(stack.len());

        for entry in &stack {

            let component = &self.local[entry.component_index];

            let solution = &component.solutions[entry.solution_index];

        }

        // Just debugging here, TODO: @remove

        let mut total_num_channels = 0;

        for entry in &stack {

            let component = &self.local[entry.component_index];

            total_num_channels += component.solutions[0].channel_mapping.len();

        }

        total_num_channels /= 2;

        let mut final_mapping = Vec::with_capacity(total_num_channels);

        let mut total_num_checked = 0;

    fn combine(&mut self, combiner: SolutionCombiner) -> Option<LeaderConclusion> {

        self.failure_reported = self.failure_reported || combiner.failure_reported;

        // Handle local solutions

        if self.local.is_empty() {

            // Trivial case

            self.local = combiner.local;

        } else {

            for local in combiner.local {

                for matched in local.solutions {

                    let local_solution = LocalSolution{

                        component: local.component,

                        final_branch_id: matched.final_branch_id,

                        port_mapping: matched.channel_mapping,

                    };

                    let maybe_solution = self.add_solution_and_check_for_global_solution(local_solution);

                    if let Some(global_solution) = maybe_solution {

                        return Some(LeaderConclusion::Solution(global_solution));

                    }

                }

            }

        }

        // Handle channel presence

    /// Receiving messages from the public inbox and handling them or storing

    /// them in the component's private inbox

    fn handle_inbox_messages(&mut self, scheduled: &mut ScheduledConnector) {

        let connector_id = scheduled.ctx.id;

        while let Some(message) = scheduled.public.inbox.take_message() {

            // Check if the message has to be rerouted because we have moved the

            // target port to another component.

            self.debug_conn(connector_id, &format!("Handling message\n --- {:#?}", message));

            if let Some(target_port) = message.target_port() {

                if let Some(other_component_id) = scheduled.router.should_reroute(target_port) {

                    self.debug_conn(connector_id, " ... Rerouting the message");

                    continue;

                }

                match scheduled.ctx.get_port_by_id(target_port) {

                    Some(port_info) => {

                        if port_info.state == PortState::Closed {

                            // We're no longer supposed to receive messages

                            // (rerouted message arrived much later!)

                            continue

                        }

                    },

                    None => {

mod api_component;

mod speculation;

mod data_transmission;

mod sync_failure;

use super::*;

use crate::{PortId, ProtocolDescription};

use crate::common::Id;

use crate::protocol::eval::*;

use crate::runtime2::native::{ApplicationSyncAction};

// Generic testing constants, use when appropriate to simplify stress-testing

fn create_runtime(pdl: &str) -> Runtime {

    let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");

    let runtime = Runtime::new(NUM_THREADS, protocol);

    return runtime;

}

fn run_test_in_runtime<F: Fn(&mut ApplicationInterface)>(pdl: &str, constructor: F) {

    let protocol = ProtocolDescription::parse(pdl.as_bytes())

        .expect("parse PDL");

        if (loc == Location::BeforeSync) failure_array[0];

        sync {

            put(output, true);

            if (loc == Location::AfterPut) failure_array[0];

            auto received = get(input);

            assert(received);

            if (loc == Location::AfterGet) failure_array[0];

        }

        if (loc == Location::AfterSync) failure_array[0];

    }

}

    channel output_a -> input_a;

    channel output_b -> input_b;

    new failing_at_location(input_b, output_a, loc);

    new failing_at_location(input_a, output_b, Location::Never);

}

    channel output_a -> input_a;

    channel output_b -> input_b;

    new failing_at_location(input_b, output_a, Location::Never);

    new failing_at_location(input_a, output_b, loc);

#[test]

    // One fails, the other one should somehow detect it and fail as well. This

    // variant constructs the failing component first.

    run_test_in_runtime(SHARED_SYNC_CODE, |api| {

        for variant in 0..4 { // all `Location` enum variants, except `Never`.

            // Create the channels

                Value::Enum(variant)

            ])).expect("create connector");

        }

    })

}

#[test]

    // One fails, the other one should somehow detect it and fail as well. This

    // variant constructs the successful component first.

    run_test_in_runtime(SHARED_SYNC_CODE, |api| {

        for variant in 0..4 {

                Value::Enum(variant)

            ])).expect("create connector");

        }

    })

}