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

        }

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

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

                        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

            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

    }

}

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,

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

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