)]

pub struct ComponentId(Id); // PUB because it can be returned by errors

/// Identifier of a port in a session

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(transparent)]

pub struct PortId(Id);

/// A safely aliasable heap-allocated payload of message bytes

#[derive(Default, Eq, PartialEq, Clone, Ord, PartialOrd)]

pub struct Payload(Arc<Vec<u8>>);

#[derive(

    Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,

)]

/// "Orientation" of a port, determining whether they can send or receive messages with `put` and `get` respectively.

#[repr(C)]

pub enum Polarity {

    Putter, // output port (from the perspective of the component)

    Getter, // input port (from the perspective of the component)

}

#[derive(

    Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,

)]

/// "Orientation" of a transport-layer network endpoint, dictating how it's connection procedure should

/// be conducted. Corresponds with connect() / accept() familiar to TCP socket programming.

#[repr(C)]

pub enum EndpointPolarity {

    Active,  // calls connect()

    Passive, // calls bind() listen() accept()

}

#[derive(Debug, Clone)]

pub(crate) enum NonsyncBlocker {

    Inconsistent,

    ComponentExit,

    SyncBlockStart,

}

#[derive(Debug, Clone)]

pub(crate) enum SyncBlocker {

    Inconsistent,

    SyncBlockEnd,

    CouldntReadMsg(PortId),

    CouldntCheckFiring(PortId),

    PutMsg(PortId, Payload),

}

pub(crate) struct DenseDebugHex<'a>(pub &'a [u8]);

pub(crate) struct DebuggableIter<I: Iterator<Item = T> + Clone, T: Debug>(pub(crate) I);

///////////////////// IMPL /////////////////////

impl IdParts for Id {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

        (self.connector_id, self.u32_suffix)

    }

}

impl IdParts for PortId {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

        self.0.id_parts()

    }

}

impl IdParts for ComponentId {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

        self.0.id_parts()

    }

}

impl U32Stream {

    pub(crate) fn next(&mut self) -> u32 {

        if self.next == u32::MAX {

            panic!("NO NEXT!")

        }

        self.next += 1;

        self.next - 1

    }

    pub(crate) fn n_skipped(mut self, n: u32) -> Self {

        self.next = self.next.saturating_add(n);

        self

    }

}

impl From<Id> for PortId {

    fn from(id: Id) -> PortId {

        Self(id)

    }

}

impl From<Id> for ComponentId {

    fn from(id: Id) -> Self {

        Self(id)

    }

}

impl From<&[u8]> for Payload {

    fn from(s: &[u8]) -> Payload {

        Payload(Arc::new(s.to_vec()))

    }

}

impl Payload {

use crate::common::*;

use core::hash::Hash;

use core::marker::PhantomData;

#[derive(serde::Serialize, serde::Deserialize)]

pub struct Id<T> {

    index: u32,

    _phantom: PhantomData<T>,

}

#[derive(Debug, serde::Serialize, serde::Deserialize)]

    store: Vec<T>,

}

//////////////////////////////////

impl<T> Debug for Id<T> {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        f.debug_struct("Id").field("index", &self.index).finish()

    }

}

impl<T> Clone for Id<T> {

    fn clone(&self) -> Self {

        *self

    }

}

impl<T> Copy for Id<T> {}

impl<T> PartialEq for Id<T> {

    fn eq(&self, other: &Self) -> bool {

        self.index.eq(&other.index)

    }

}

impl<T> Eq for Id<T> {}

impl<T> Hash for Id<T> {

    fn hash<H: std::hash::Hasher>(&self, h: &mut H) {

        self.index.hash(h);

    }

}

impl<T> Arena<T> {

    pub fn new() -> Self {

        Self { store: vec![] }

    }

    pub fn alloc_with_id(&mut self, f: impl FnOnce(Id<T>) -> T) -> Id<T> {

        use std::convert::TryFrom;

        let id = Id {

            index: u32::try_from(self.store.len()).expect("Out of capacity!"),

            _phantom: Default::default(),

        };

        self.store.push(f(id));

        id

    }

    pub fn iter(&self) -> impl Iterator<Item = (Id<T>, &T)> {

        (0..).map(|index| Id { index, _phantom: Default::default() }).zip(self.store.iter())

    }

}

impl<T> core::ops::Index<Id<T>> for Arena<T> {

    type Output = T;

    fn index(&self, id: Id<T>) -> &Self::Output {

        self.store.index(id.index as usize)

    }

use std::collections::HashMap;

use std::fmt;

use std::fmt::{Debug, Display, Formatter};

use std::{i16, i32, i64, i8};

use crate::common::*;

use crate::protocol::ast::*;

use crate::protocol::EvalContext;

const BYTE_MIN: i64 = i8::MIN as i64;

const BYTE_MAX: i64 = i8::MAX as i64;

const SHORT_MIN: i64 = i16::MIN as i64;

const SHORT_MAX: i64 = i16::MAX as i64;

const INT_MIN: i64 = i32::MIN as i64;

const INT_MAX: i64 = i32::MAX as i64;

const MESSAGE_MAX_LENGTH: i64 = SHORT_MAX;

const ONE: Value = Value::Byte(ByteValue(1));

trait ValueImpl {

    fn exact_type(&self) -> Type;

    fn is_type_compatible(&self, t: &Type) -> bool;

}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

pub enum Value {

    Input(InputValue),

    Output(OutputValue),

    Message(MessageValue),

    Boolean(BooleanValue),

    Byte(ByteValue),

    Short(ShortValue),

    Int(IntValue),

    Long(LongValue),

    InputArray(InputArrayValue),

    OutputArray(OutputArrayValue),

    MessageArray(MessageArrayValue),

    BooleanArray(BooleanArrayValue),

    ByteArray(ByteArrayValue),

    ShortArray(ShortArrayValue),

    IntArray(IntArrayValue),

    LongArray(LongArrayValue),

}

impl Value {

    pub fn receive_message(buffer: &Payload) -> Value {

        Value::Message(MessageValue(Some(buffer.clone())))

    }

    fn create_message(length: Value) -> Value {

        match length {

            Value::Byte(_) | Value::Short(_) | Value::Int(_) | Value::Long(_) => {

                let length: i64 = i64::from(length);

                if length < 0 || length > MESSAGE_MAX_LENGTH {

                    // Only messages within the expected length are allowed

                    Value::Message(MessageValue(None))

                } else {

                        let var = rctx.ips.port_info.spec_var_for(putter);

                        let was = predicate.assigned.insert(var, SpecVal::FIRING);

                        if was == Some(SpecVal::SILENT) {

                            // Discard the branch, as it clearly has contradictory requirements for this value.

                            log!(cu.logger(), "Proto component {:?} tried to PUT on port {:?} when pred said var {:?}==Some(false). inconsistent!",

                            proto_component_id, putter, var);

                            drop((predicate, branch));

                        } else {

                            // Note that this port has put this round,

                            // and assert that this isn't its 2nd time putting this round (otheriwse PDL programming error)

                            assert!(branch.inner.did_put_or_get.insert(putter));

                            log!(cu.logger(), "Proto component {:?} with pred {:?} putting payload {:?} on port {:?} (using var {:?})",

                            proto_component_id, &predicate, &payload, putter, var);

                            // Send the given payload (by buffering it).

                            let msg = SendPayloadMsg { predicate: predicate.clone(), payload };

                            rctx.putter_push(cu, putter, msg);

                            // Branch can still make progress. Schedule to be rerun

                            Self::insert_branch_merging(cd.swap, predicate, branch);

                        }

                    }

                    B::SyncBlockEnd => {

                        // This branch reached the end of it's synchronous block

                        // assign all variables of owned ports that DIDN'T fire to SILENT

                        for port in rctx.ips.port_info.ports_owned_by(proto_component_id) {

                            let var = rctx.ips.port_info.spec_var_for(*port);

                            let actually_exchanged = branch.inner.did_put_or_get.contains(port);

                            let val = *predicate.assigned.entry(var).or_insert(SpecVal::SILENT);

                            let speculated_to_fire = val == SpecVal::FIRING;

                            if actually_exchanged != speculated_to_fire {

                                log!(cu.logger(), "Inconsistent wrt. port {:?} var {:?} val {:?} actually_exchanged={}, speculated_to_fire={}",

                                port, var, val, actually_exchanged, speculated_to_fire);

                                // IMPLICIT inconsistency

                                drop((predicate, branch));

                                continue 'branch_iter;

                            }

                        }

                        // submit solution for this component

                        let subtree_id = SubtreeId::LocalComponent(proto_component_id);

                        rctx.solution_storage.submit_and_digest_subtree_solution(

                            cu,

                            subtree_id,

                            predicate.clone(),

                        );

                        branch.ended = true;

                        // This branch exits the cyclic drain

                        Self::insert_branch_merging(cd.output, predicate, branch);

                    }

                }

            }

            std::mem::swap(cd.input, cd.swap);

        }

        Ok(())

    }

    // Feed this branching protocol component the given message, and

    // then run all branches until they are once again blocked.

    fn feed_msg(

        &mut self,

        cu: &mut impl CuUndecided,

        rctx: &mut RoundCtx,

        proto_component_id: ComponentId,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

        pcb_temps: MapTempsGuard<'_, Predicate, ProtoComponentBranch>,

    ) -> Result<(), UnrecoverableSyncError> {

        log!(

            cu.logger(),

            "feeding proto component {:?} getter {:?} {:?}",

            proto_component_id,

            getter,

            &send_payload_msg

        );

        let (mut unblocked, pcb_temps) = pcb_temps.split_first_mut();

        let (mut blocked, pcb_temps) = pcb_temps.split_first_mut();

        // partition drain from self.branches -> {unblocked, blocked} (not cyclic)

        log!(cu.logger(), "visiting {} blocked branches...", self.branches.len());

        for (predicate, mut branch) in self.branches.drain() {

            if branch.ended {

                log!(cu.logger(), "Skipping ended branch with {:?}", &predicate);

                Self::insert_branch_merging(&mut blocked, predicate, branch);

                continue;

            }

            use AssignmentUnionResult as Aur;

            log!(cu.logger(), "visiting branch with pred {:?}", &predicate);

            // We give each branch a chance to receive this message,

            // those that do are maybe UNBLOCKED, and all others remain BLOCKED.

            match predicate.assignment_union(&send_payload_msg.predicate) {

                Aur::Nonexistant => {

                    // this branch does not receive the message. categorize into blocked.

                    log!(cu.logger(), "skipping branch");

                    Self::insert_branch_merging(&mut blocked, predicate, branch);

                }

                Aur::Equivalent | Aur::FormerNotLatter => {

                    // retain the existing predicate, but add this payload

                    log!(cu.logger(), "feeding this branch without altering its predicate");

                peer: Some(i),

                polarity: Putter,

                owner: self.proto_component_id,

            },

        );

        self.ips.port_info.map.insert(

            i,

            PortInfo {

                route: Route::LocalComponent,

                peer: Some(o),

                polarity: Getter,

                owner: self.proto_component_id,

            },

        );

        self.ips

            .port_info

            .owned

            .entry(self.proto_component_id)

            .or_default()

            .extend([o, i].iter().copied());

        log!(

            self.logger,

            "Component {:?} port pair (out->in) {:?} -> {:?}",

            self.proto_component_id,

            o,

            i

        );

        [o, i]

    }

}

impl SyncProtoContext<'_> {

    // The component calls the runtime back, inspecting whether it's associated

    // preidcate has already determined a (speculative) value for the given port's firing variable.

    pub(crate) fn is_firing(&mut self, port: PortId) -> Option<bool> {

        let var = self.rctx.ips.port_info.spec_var_for(port);

        self.predicate.query(var).map(SpecVal::is_firing)

    }

    // The component calls the runtime back, trying to inspect a port's message

    pub(crate) fn read_msg(&mut self, port: PortId) -> Option<&Payload> {

        let maybe_msg = self.branch_inner.inbox.get(&port);

        if maybe_msg.is_some() {

            // Make a note that this component has received

            // this port's message 1+ times this round

            self.branch_inner.did_put_or_get.insert(port);

        }

        maybe_msg

    }

}

#[derive(Debug)]

pub struct Connector {

    unphased: ConnectorUnphased,

    phased: ConnectorPhased,

}

/// Characterizes a type which can write lines of logging text.

/// The implementations provided in the `logging` module are likely to be sufficient,

/// but for added flexibility, users are able to implement their own loggers for use

/// by connectors.

pub trait Logger: Debug + Send + Sync {

    fn line_writer(&mut self) -> Option<&mut dyn std::io::Write>;

}

/// A logger that appends the logged strings to a growing byte buffer

#[derive(Debug)]

pub struct VecLogger(ConnectorId, Vec<u8>);

/// A trivial logger that always returns None, such that no logging information is ever written.

#[derive(Debug)]

pub struct DummyLogger;

/// A logger that writes the logged lines to a given file.

#[derive(Debug)]

pub struct FileLogger(ConnectorId, std::fs::File);

// Interface between protocol state and the connector runtime BEFORE all components

// ave begun their branching speculation. See ComponentState::nonsync_run.

pub(crate) struct NonsyncProtoContext<'a> {

    ips: &'a mut IdAndPortState,

    logger: &'a mut dyn Logger,

    unrun_components: &'a mut Vec<(ComponentId, ComponentState)>, // lives for Nonsync phase

    proto_component_id: ComponentId,                              // KEY in id->component map

}

// Interface between protocol state and the connector runtime AFTER all components

// have begun their branching speculation. See ComponentState::sync_run.

pub(crate) struct SyncProtoContext<'a> {

    rctx: &'a RoundCtx,

    branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch

    predicate: &'a Predicate,                        // KEY in pred->branch map

}

// The data coupled with a particular protocol component branch, but crucially omitting

// the `ComponentState` such that this may be passed by reference to the state with separate

// access control.

#[derive(Default, Debug, Clone)]

struct ProtoComponentBranchInner {

    did_put_or_get: HashSet<PortId>,

    inbox: HashMap<PortId, Payload>,

}

// A speculative variable that lives for the duration of the synchronous round.

// Each is assigned a value in domain `SpecVal`.

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

struct SpecVar(PortId);

// The codomain of SpecVal. Has two associated constants for values FIRING and SILENT,

// but may also enumerate many more values to facilitate finer-grained nondeterministic branching.

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

struct SpecVal(u16);

// Data associated with a successful synchronous round, retained afterwards such that the

// native component can freely reflect on how it went, reading the messages received at their

// inputs, and reflecting on which of their connector's synchronous batches succeeded.

#[derive(Debug)]

struct RoundEndedNative {

    batch_index: usize,

    gotten: HashMap<PortId, Payload>,

}

// Implementation of a set in terms of a vector (optimized for reading, not writing)

#[derive(Default)]

struct VecSet<T: std::cmp::Ord> {

    // invariant: ordered, deduplicated

    vec: Vec<T>,

}

// Allows a connector to remember how to forward payloads towards the component that

// owns their destination port. `LocalComponent` corresponds with messages for components

// managed by the connector itself (hinting for it to look it up in a local structure),

// whereas the other variants direct the connector to forward the messages over the network.

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

enum Route {

    LocalComponent,

    NetEndpoint { index: usize },

    UdpEndpoint { index: usize },

}

// The outcome of a synchronous round, representing the distributed consensus.

// In the success case, the attached predicate encodes a row in the session's trace table.

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

use crate::common::*;

use crate::runtime::*;

impl TokenTarget {

    // subdivides the domain of usize into

    // [NET_ENDPOINT][UDP_ENDPOINT  ]

    // ^0            ^usize::MAX/2   ^usize::MAX

    const HALFWAY_INDEX: usize = usize::MAX / 2;

}

impl From<Token> for TokenTarget {

    fn from(Token(index): Token) -> Self {

        if let Some(shifted) = index.checked_sub(Self::HALFWAY_INDEX) {

            TokenTarget::UdpEndpoint { index: shifted }

        } else {

            TokenTarget::NetEndpoint { index }

        }

    }

}

impl Into<Token> for TokenTarget {

    fn into(self) -> Token {

        match self {

            TokenTarget::UdpEndpoint { index } => Token(index + Self::HALFWAY_INDEX),

            TokenTarget::NetEndpoint { index } => Token(index),

        }

    }

}

impl Connector {

    /// Create a new connector structure with the given protocol description (via Arc to facilitate sharing).

    /// The resulting connector will start in the setup phase, and cannot be used for communication until the

    /// `connect` procedure completes.

    /// # Safety

    /// The correctness of the system's underlying distributed algorithms requires that no two

    /// connectors have the same ID. If the user does not know the identifiers of other connectors in the

    /// system, it is advised to guess it using Connector::random_id (relying on the exceptionally low probability of an error).

    /// Sessions with duplicate connector identifiers will not result in any memory unsafety, but cannot be guaranteed

    /// to preserve their configured protocols.

    /// Fortunately, in most realistic cases, the presence of duplicate connector identifiers will result in an

    /// error during `connect`, observed as a peer misbehaving.

    pub fn new(

        mut logger: Box<dyn Logger>,

        proto_description: Arc<ProtocolDescription>,

        connector_id: ConnectorId,

    ) -> Self {

        log!(&mut *logger, "Created with connector_id {:?}", connector_id);

        let mut id_manager = IdManager::new(connector_id);

        let native_component_id = id_manager.new_component_id();

        Self {

            unphased: ConnectorUnphased {

                proto_description,

                proto_components: Default::default(),

                logger,

                native_component_id,

                ips: IdAndPortState { id_manager, port_info: Default::default() },

            },

            phased: ConnectorPhased::Setup(Box::new(ConnectorSetup {

                net_endpoint_setups: Default::default(),

                        log!(cu.logger, "I expected the scatter from my parent only!");

                        return Err(Ce::SetupAlgMisbehavior);

                    }

                    break 'scatter_loop optimized_map;

                }

                msg @ Msg::CommMsg { .. } => {

                    log!(cu.logger, "delaying msg {:?} during scatter recv", msg);

                    comm.endpoint_manager.delayed_messages.push((recv_index, msg));

                }

                msg @ S(Sm::SessionGather { .. })

                | msg @ S(Sm::YouAreMyParent)

                | msg @ S(Sm::MyPortInfo(..))

                | msg @ S(Sm::LeaderAnnounce { .. })

                | msg @ S(Sm::LeaderWave { .. }) => {

                    log!(cu.logger, "discarding old message {:?} during election", msg);

                }

            }

        }

    } else {

        // by computing it myself

        log!(cu.logger, "I am the leader! I will optimize this session");

        leader_session_map_optimize(&mut *cu.logger, unoptimized_map)?

    };

    log!(

        cu.logger,

        "Optimized info map is {:?}. Sending to children {:?}",

        &optimized_map,

        comm.neighborhood.children.iter()

    );

    log!(cu.logger, "All session info dumped!: {:#?}", &optimized_map);

    // extract my own ConnectorId's entry

    let optimized_info =

        optimized_map.get(&cu.ips.id_manager.connector_id).expect("HEY NO INFO FOR ME?").clone();

    // broadcast the optimized session info to my children

    let msg = S(Sm::SessionScatter { optimized_map });

    for &child in comm.neighborhood.children.iter() {

        comm.endpoint_manager.send_to_setup(child, &msg)?;

    }

    // apply local optimizations

    apply_my_optimizations(cu, comm, optimized_info)?;

    log!(cu.logger, "Session optimizations applied");

    Ok(())

}

// Defines the optimization function, consuming an optimized map,

// and returning an optimized map.

fn leader_session_map_optimize(

    logger: &mut dyn Logger,

) -> Result<HashMap<ConnectorId, SessionInfo>, ConnectError> {

    log!(logger, "Session map optimize START");

    // currently, it's the identity function

    log!(logger, "Session map optimize END");

    Ok(m)

}

// Modify the given connector's internals to reflect

// the given session info

fn apply_my_optimizations(

    cu: &mut ConnectorUnphased,

    comm: &mut ConnectorCommunication,

    session_info: SessionInfo,

) -> Result<(), ConnectError> {

    let SessionInfo {

        proto_components,

        port_info,

        serde_proto_description,

        endpoint_incoming_to_getter,

    } = session_info;

    // simply overwrite the contents

    println!("BEFORE: {:#?}\n{:#?}", cu, comm);

    cu.ips.port_info = port_info;

    assert!(cu.ips.port_info.invariant_preserved());

    cu.proto_components = proto_components;

    cu.proto_description = serde_proto_description.0;

    for (ee, getter) in comm

        .endpoint_manager

        .net_endpoint_store

        .endpoint_exts

        .iter_mut()

        .zip(endpoint_incoming_to_getter)

    {

        ee.getter_for_incoming = getter;

    }

    // println!("AFTER: {:#?}\n{:#?}", cu, comm);

    Ok(())

}