use crate::protocol::eval::ValueGroup;

use super::ConnectorId;

use super::branch::BranchId;

use super::port::{ChannelId, PortIdLocal, PortState};

use super::inbox::{

    Message, DataHeader, SyncHeader, ChannelAnnotation, BranchMarker,

    DataMessage,

    SyncCompMessage, SyncCompContent,

    SyncPortMessage, SyncPortContent,

    SyncControlMessage, SyncControlContent

};

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

// TODO: Have a "branch+port position hint" in case multiple operations are

//  performed on the same port to prevent repeated lookups

// TODO: A lot of stuff should be batched. Like checking all the sync headers

//  and sending "I have a higher ID" messages. Should reduce locking by quite a

//  bit.

// TODO: Needs a refactor. Firstly we have cases where we don't have a branch ID

//  but we do want to enumerate all current ports. So put that somewhere in a

//  central place. Secondly. Error handling and regular message handling is

//  becoming a mess.

pub(crate) struct Consensus {

    // --- State that is cleared after each round

    // Local component's state

                    source_port: self_port_id,

                    target_port: peer_port_id,

                    content: SyncPortContent::SilentPortNotification,

                };

                match ctx.submit_message(Message::SyncPort(message)) {

                    Ok(_) => {

                        self.encountered_ports.push(self_port_id);

                    },

                    Err(_) => {

                        // Seems like we were done with this branch, but one of

                        // the silent ports (in scope) is actually closed

                        return self.notify_of_fatal_branch(branch_id, ctx);

                    }

                }

            }

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

        // TODO: Handle sending and receiving ports

        // Set final ports

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

        // Clear out internal storage to defaults

        println!("DEBUG: ***** Incrementing sync round stuff");

        self.highest_connector_id = ConnectorId::new_invalid();

        self.branch_annotations.clear();

        self.branch_markers.clear();

        self.encountered_ports.clear();

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

            // Check for consistent mapping

            let port = branch.channel_mapping.iter()

                .find(|v| v.channel_id == port_info.channel_id)

                .unwrap();

            debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));

        }

        // Check for ports that are being sent

        debug_assert!(self.workspace_ports.is_empty());

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

        match self.handle_received_sync_header(message.sync_header, ctx) {

            MessageOrigin::Past => return None,

            MessageOrigin::Present => {},

            MessageOrigin::Future => {

                ctx.put_back_message(Message::SyncComp(message));

                return None

            }

        }

        // And handle the contents

        debug_assert_eq!(message.target_component_id, ctx.id);

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

                // We were just interested in the sync header we handled above

                return None;

            }

        }

    }

    pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        match self.handle_received_sync_header(message.sync_header, ctx) {

            MessageOrigin::Past => return None,

            MessageOrigin::Present => {},

            MessageOrigin::Future => {

                ctx.put_back_message(Message::SyncPort(message));

                SyncCompContent::GlobalFailure => {

                    unreachable!("unexpected message content for leader");

                },

            }

        } else {

            // Someone else is the leader

            let message = SyncCompMessage {

                sync_header: self.create_sync_header(ctx),

                target_component_id: self.highest_connector_id,

                content,

            };

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

        }

        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;

        }

        // TODO: Performance

        let mut encountered = VecSet::new();

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

                    }

                }

            }

        }

        println!("DEBUGERINO: Leader entering error state, we need to wait on {:?}", encountered.iter().map(|v| v.index).collect::<Vec<_>>());

        self.conclusion = Some(RoundConclusion::Failure);

        if encountered.is_empty() {

            // We don't have to wait on Acks

            return Some(RoundConclusion::Failure);

        } else {

            self.ack_remaining = encountered.len() as u32;

            return None;

        }

    }

#[derive(Debug, Clone)]

struct MatchedLocalSolution {

    final_branch_id: BranchId,

    channel_mapping: Vec<(ChannelId, BranchMarker)>,

    matches: Vec<ComponentMatches>,

}

#[derive(Debug, Clone)]

struct ComponentMatches {

    target_id: ConnectorId,

    target_index: usize,

    match_indices: Vec<usize>, // of local solution in connector

}

#[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)]

pub(crate) struct LocalChannelPresence {

    channel_id: ChannelId,

    is_closed: bool,

}

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

enum PresenceState {

    OnePresent, // one component reported the channel being open

    BothPresent, // two components reported the channel being open

    Closed, // one component reported the channel being closed

}

    parent_entry_index: usize,      // parent that caused the creation of this checking entry

    match_index_in_parent: usize,   // index in the matches array of the parent

    solution_index_in_parent: usize,// index in the solution array of the match entry in the parent

}

enum LeaderConclusion {

    Solution(GlobalSolution),

    Failure,

}

impl SolutionCombiner {

    fn new() -> Self {

        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 {

            let cur_ports = &self.local[component_index].solutions[0].channel_mapping;

            let mut component_peers = Vec::new();

            // Find the matching components

            for (other_index, other_component) in self.local.iter().enumerate() {

                if other_index == component_index {

                    // Don't match against ourselves

                    continue;

                }

                let mut matching_channels = Vec::new();

                for (cur_channel_id, _) in cur_ports {

                    let new_solution_index = match_component.match_indices[new_solution_index_in_parent];

                    let cur_entry = &mut stack[cur_index];

                    cur_entry.solution_index_in_parent = new_solution_index_in_parent;

                    cur_entry.solution_index = new_solution_index;

                    continue 'check_last_stack;

                } else {

                    // We're out of options here. So pop an entry, then in

                    // the next iteration of this backtracking loop we try

                    // to increment that solution

                    stack.pop();

                }

            }

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

        for entry in &stack {

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

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

            for (channel_id, branch_id) in solution.channel_mapping.iter().copied() {

                match final_mapping.iter().find(|(v, _)| *v == channel_id) {

                    Some((_, encountered_branch_id)) => {

                        debug_assert_eq!(*encountered_branch_id, branch_id);

                        total_num_checked += 1;

                    },

                    None => {

            presence: self.presence.clone(),

            failure_reported: self.failure_reported,

        };

        self.local.clear();

        self.presence.clear();

        self.failure_reported = false;

        return Some(result);

    }

    // TODO: Entire routine is quite wasteful. Combine instead of doing all work

    //  again.

    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

        println!("DEBUGERINO: Presence before joining is {:#?}", &self.presence);

        if self.presence.is_empty() {

            // Trivial case

            self.presence = combiner.presence;

            println!("DEBUGERINO: Trivial merging")

        } else {

            for presence in combiner.presence {

                match self.presence.iter_mut().find(|v| v.id == presence.id) {

                    Some(entry) => {

                        // Combine entries. Take first that has Closed, then

                        // check first that has both, then check if they are

                        // combinable

                    if scheduled.router.num_pending_acks() == 0 {

                        // All ports (if any) already closed

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    self.try_go_to_sleep(connector_key, scheduled);

                },

            }

        }

    }

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

                        self.runtime.send_message(other_component_id, 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 => {

                        // Apparently we no longer have a handle to the port

                        continue;

                    }

                }

            }

            // If here, then we should handle the message

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

            if let Message::Control(message) = &message {

                match message.content {

                    ControlContent::PortPeerChanged(port_id, new_target_connector_id) => {

                        // Need to change port target

mod network_shapes;

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

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

    let mut api = runtime.create_interface();

    for _ in 0..NUM_INSTANCES {

        constructor(&mut api);

    }

}

pub(crate) struct TestTimer {

    name: &'static str,

    started: std::time::Instant

}

            .expect("create component");

        api.create_connector("", "immediate_failure_inside_sync", ValueGroup::new_stack(Vec::new()))

            .expect("create component");

    })

}

const SHARED_SYNC_CODE: &'static str = "

enum Location { BeforeSync, AfterPut, AfterGet, AfterSync, Never }

primitive failing_at_location(in<bool> input, out<bool> output, Location loc) {

    u32[] failure_array = {};

    while (true) {

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

        }

    })

}