&self.0

    }

    pub fn as_mut_slice(&mut self) -> &mut [u8] {

        Arc::make_mut(&mut self.0) as _

    }

    pub fn concat_with(&mut self, other: &Self) {

        let bytes = other.as_slice().iter().copied();

        let me = Arc::make_mut(&mut self.0);

        me.extend(bytes);

    }

}

impl serde::Serialize for Payload {

    where

        S: serde::Serializer,

    {

        let inner: &Vec<u8> = &self.0;

        inner.serialize(serializer)

    }

}

impl<'de> serde::Deserialize<'de> for Payload {

    where

        D: serde::Deserializer<'de>,

    {

        let inner: Vec<u8> = Vec::deserialize(deserializer)?;

        Ok(Self(Arc::new(inner)))

    }

}

impl std::iter::FromIterator<u8> for Payload {

    fn from_iter<I: IntoIterator<Item = u8>>(it: I) -> Self {

        Self(Arc::new(it.into_iter().collect()))

    }

}

    }

}

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

pub struct ImportedDeclarationId(DeclarationId);

impl ImportedDeclarationId {

    pub fn upcast(self) -> DeclarationId {

        self.0

    }

}

pub struct Heap {

    // Phase 0: allocation

    protocol_descriptions: Arena<Root>,

    pragmas: Arena<Pragma>,

    imports: Arena<Import>,

    identifiers: Arena<Identifier>,

    type_annotations: Arena<TypeAnnotation>,

    variables: Arena<Variable>,

    definitions: Arena<Definition>,

    statements: Arena<Statement>,

    expressions: Arena<Expression>,

    declarations: Arena<Declaration>,

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 MAX_RECURSION: usize = 1024;

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;

                let raw = String::from_utf8_lossy(data);

                let val = raw.parse::<i64>().unwrap();

                if val >= BYTE_MIN && val <= BYTE_MAX {

                    Value::Byte(ByteValue(val as i8))

                } else if val >= SHORT_MIN && val <= SHORT_MAX {

                    Value::Short(ShortValue(val as i16))

                } else if val >= INT_MIN && val <= INT_MAX {

                    Value::Int(IntValue(val as i32))

                } else {

                    Value::Long(LongValue(val))

                }

            }

        }

    }

    fn set(&mut self, index: &Value, value: &Value) -> Option<Value> {

        // The index must be of integer type, and non-negative

        let the_index: usize;

        match index {

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

                let index = i64::from(index);

                if index < 0 || index >= MESSAGE_MAX_LENGTH {

                    // It is inconsistent to update out of bounds

                    return None;

                }

                // Signal the put upwards

                Err(EvalContinuation::Put(port, message))

            }

            Statement::Expression(stmt) => {

                // Evaluate expression

                let value = self.store.eval(h, ctx, stmt.expression)?;

                // Continue to next statement

                self.position = stmt.next;

                Err(EvalContinuation::Stepping)

            }

        }

    }

    }

}

// #[cfg(test)]

// mod tests {

//     extern crate test_generator;

//     use std::fs::File;

//     use std::io::Read;

//     use std::path::Path;

//     use test_generator::test_resources;

use std::fmt;

use std::fs::File;

use std::io;

use std::path::Path;

use backtrace::Backtrace;

pub struct InputSource {

    filename: String,

    input: Vec<u8>,

    line: usize,

    column: usize,

    offset: usize,

}

static STD_LIB_PDL: &'static [u8] = b"

primitive forward(in i, out o) {

    while(true) synchronous() put(o, get(i));

}

pub struct ProtocolDescription {

    heap: Heap,

    source: InputSource,

    root: RootId,

}

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

pub struct ComponentState {

    prompt: Prompt,

}

pub enum EvalContext<'a> {

    Nonsync(&'a mut NonsyncProtoContext<'a>),

    Sync(&'a mut SyncProtoContext<'a>),

}

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

impl std::fmt::Debug for ProtocolDescription {

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

    }

}

impl ProtocolDescription {

    pub fn parse(buffer: &[u8]) -> Result<Self, String> {

        let mut heap = Heap::new();

        let mut source = InputSource::from_buffer(buffer).unwrap();

        let mut parser = Parser::new(&mut source);

        match parser.parse(&mut heap) {

            Ok(root) => {

                return Ok(ProtocolDescription { heap, source, root });

            }

            Err(err) => {

                            _ => unreachable!(),

                        }

                        return SyncBlocker::PutMsg(value, payload);

                    }

                },

            }

        }

    }

}

impl EvalContext<'_> {

    // fn random(&mut self) -> LongValue {

    //     match self {

    //         EvalContext::Nonsync(_context) => todo!(),

    //         EvalContext::Sync(_) => unreachable!(),

    //     }

    // }

    fn new_component(&mut self, args: &[Value], init_state: ComponentState) -> () {

        match self {

            EvalContext::Nonsync(context) => {

                let mut moved_ports = HashSet::new();

                for arg in args.iter() {

                    match arg {

                        Value::Output(OutputValue(port)) => {

                            moved_ports.insert(*port);

                        }

                        Value::Input(InputValue(port)) => {

                            moved_ports.insert(*port);

                        }

                        _ => {}

                    }

                }

                context.new_component(moved_ports, init_state)

            }

            EvalContext::Sync(_) => unreachable!(),

        }

    }

    fn new_channel(&mut self) -> [Value; 2] {

        match self {

            EvalContext::Nonsync(context) => {

                let [from, to] = context.new_port_pair();

                let from = Value::Output(OutputValue(from));

                let to = Value::Input(InputValue(to));

                return [from, to];

            }

            EvalContext::Sync(_) => unreachable!(),

        }

    }

    fn fires(&mut self, port: Value) -> Option<Value> {

        match self {

            EvalContext::Nonsync(_) => unreachable!(),

            EvalContext::Sync(context) => match port {

                Value::Output(OutputValue(port)) => context.is_firing(port).map(Value::from),

                Value::Input(InputValue(port)) => context.is_firing(port).map(Value::from),

                _ => unreachable!(),

            },

        }

    }

    fn get(&mut self, port: Value) -> Option<Value> {

        match self {

            EvalContext::Nonsync(_) => unreachable!(),

            EvalContext::Sync(context) => match port {

                Value::Output(OutputValue(port)) => {

                    context.read_msg(port).map(Value::receive_message)

                }

                Value::Input(InputValue(port)) => {

                    context.read_msg(port).map(Value::receive_message)

                }

                _ => unreachable!(),

            },

        }

    }

        }

        log!(

            cu.logger,

            "All {} proto components are now done with Nonsync phase",

            branching_proto_components.len(),

        );

        // NOTE: all msgs in outbox are of form (Getter, Payload)

        let mut payloads_to_get: Vec<(PortId, SendPayloadMsg)> = vec![];

        // create the solution storage

        let mut solution_storage = {

            let e = comm.neighborhood.children.iter().map(|&index| Route::Endpoint { index });

            SolutionStorage::new(n.chain(c).chain(e))

        };

        log!(cu.logger, "Solution storage initialized");

        // 2. kick off the native

        log!(

            cu.logger,

            "Translating {} native batches into branches...",

            comm.native_batches.len()

        );

        let mut branching_native = BranchingNative { branches: Default::default() };

                let getter = *cu.port_info.peers.get(&putter).unwrap();

                payloads_to_get.push((getter, msg));

            }

            if to_get.is_empty() {

                log!(

                    cu.logger,

                    "Native submitting solution for batch {} with {:?}",

                    index,

                    &predicate

                );

                solution_storage.submit_and_digest_subtree_solution(

                    &mut *cu.logger,

                    predicate.clone(),

                );

            }

            let branch = NativeBranch { index, gotten: Default::default(), to_get };

            if let Some(existing) = branching_native.branches.insert(predicate, branch) {

                // TODO

                return Err(Se::IndistinguishableBatches([index, existing.index]));

            }

        }

        log!(cu.logger, "Done translating native batches into branches");

        comm.native_batches.push(Default::default());

        let decision = 'undecided: loop {

            // drain payloads_to_get, sending them through endpoints / feeding them to components

            while let Some((getter, send_payload_msg)) = payloads_to_get.pop() {

                assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));

                match cu.port_info.routes.get(&getter).unwrap() {

                    Route::Endpoint { index } => {

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

                        comm.endpoint_manager.send_to(*index, &msg).unwrap();

                    }

                        cu,

                        &mut solution_storage,

                        // &mut Pay

                        getter,

                        send_payload_msg,

                    ),

                        if let Some(branching_component) =

                            branching_proto_components.get_mut(proto_component_id)

                        {

                            let proto_component_id = *proto_component_id;

                            // let ConnectorUnphased { port_info, proto_description, .. } = cu;

                            branching_component.feed_msg(

                                cu,

                                &mut solution_storage,

                                proto_component_id,

                                &mut payloads_to_get,

                                getter,

                                send_payload_msg,

        let mut draining = HashMap::default();

        let finished = &mut self.branches;

        std::mem::swap(&mut draining, finished);

        for (predicate, mut branch) in draining.drain() {

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

            // check if this branch expects to receive it

            let var = cu.port_info.firing_var_for(getter);

            let mut feed_branch = |branch: &mut NativeBranch, predicate: &Predicate| {

                let was = branch.gotten.insert(getter, send_payload_msg.payload.clone());

                assert!(was.is_none());

                branch.to_get.remove(&getter);

                if branch.to_get.is_empty() {

                    solution_storage.submit_and_digest_subtree_solution(

                        &mut *cu.logger,

                        route,

                        predicate.clone(),

                    );

                }

            };

            if predicate.query(var) != Some(true) {

                // optimization. Don't bother trying this branch

                log!(

                    cu.logger,

                    "skipping branch with {:?} that doesn't want the message (fastpath)",

                        // branch is inconsistent. throw it away

                        drop((predicate, branch));

                    }

                    B::SyncBlockEnd => {

                        // make concrete all variables

                        for &port in ports.iter() {

                            let var = cu.port_info.firing_var_for(port);

                            predicate.assigned.entry(var).or_insert(false);

                        }

                        // submit solution for this component

                        solution_storage.submit_and_digest_subtree_solution(

                            &mut *cu.logger,

                            predicate.clone(),

                        );

                        // move to "blocked"

                        drainer.add_output(predicate, branch);

                    }

                    B::CouldntReadMsg(port) => {

                        // move to "blocked"

                        assert!(!branch.inbox.contains_key(&port));

                        drainer.add_output(predicate, branch);

                    }

                    B::CouldntCheckFiring(port) => {

                        // sanity check

        log!(

            self.logger,

            "Component {:?} added new component with state {:?}, moving ports {:?}",

            self.proto_component_id,

            &state,

            &moved_ports

        );

        assert!(self.proto_component_ports.is_subset(&moved_ports));

        // 2. remove ports from old component & update port->route

        let new_id = self.id_manager.new_proto_component_id();

        for port in moved_ports.iter() {

            self.proto_component_ports.remove(port);

        }

        // 3. create a new component

        self.unrun_components.push((new_id, ProtoComponent { state, ports: moved_ports }));

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        // adds two new associated ports, related to each other, and exposed to the proto component

        let [o, i] = [self.id_manager.new_port_id(), self.id_manager.new_port_id()];

        self.proto_component_ports.insert(o);

        self.proto_component_ports.insert(i);

        // {polarity, peer, route} known. {} unknown.

        self.port_info.polarities.insert(o, Putter);

        self.port_info.polarities.insert(i, Getter);

        self.port_info.peers.insert(o, i);

        self.port_info.peers.insert(i, o);

        self.port_info.routes.insert(o, route);

        self.port_info.routes.insert(i, route);

        log!(

            self.logger,

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

            self.proto_component_id,

            o,

            i

        );

        [o, i]

    }

}

use error::*;

#[derive(Debug)]

pub struct RoundOk {

    batch_index: usize,

    gotten: HashMap<PortId, Payload>,

}

#[derive(Debug)]

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

    // invariant: ordered, deduplicated

    vec: Vec<T>,

}

    Native,

    Proto(ProtoComponentId),

}

pub enum Route {

    Endpoint { index: usize },

}

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

pub struct MyPortInfo {

    polarity: Polarity,

    port: PortId,

}

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

pub enum Decision {

    Failure,

    Success(Predicate),

}

    SetupMsg(SetupMsg),

    CommMsg(CommMsg),

}

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

pub enum SetupMsg {

    MyPortInfo(MyPortInfo),

    LeaderWave { wave_leader: ConnectorId },

    LeaderAnnounce { tree_leader: ConnectorId },

    YouAreMyParent,

    SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },

    SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },

}

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

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

pub struct CommMsg {

    pub round_index: usize,

    pub contents: CommMsgContents,

}

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

pub enum CommMsgContents {

    SendPayload(SendPayloadMsg),

    Suggest { suggestion: Decision }, // SINKWARD

    Announce { decision: Decision },  // SINKAWAYS

}

#[derive(Debug, PartialEq)]

pub enum CommonSatResult {

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

pub struct Endpoint {

    inbox: Vec<u8>,

    stream: TcpStream,

}

pub struct ProtoComponent {

    state: ComponentState,

    ports: HashSet<PortId>,

}

pub trait Logger: Debug {

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

}

#[derive(Debug)]

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

#[derive(Debug)]

pub struct DummyLogger;

#[derive(Debug)]

#[derive(Debug)]

pub struct EndpointManager {

    // invariants:

    // 1. endpoint N is registered READ | WRITE with poller

    // 2. Events is empty

    poll: Poll,

    events: Events,

    polled_undrained: IndexSet<usize>,

    delayed_messages: Vec<(usize, Msg)>,

    undelayed_messages: Vec<(usize, Msg)>,

    endpoint_exts: Vec<EndpointExt>,

}

pub struct PortInfo {

    polarities: HashMap<PortId, Polarity>,

    peers: HashMap<PortId, PortId>,

    routes: HashMap<PortId, Route>,

}

#[derive(Debug)]

pub struct Connector {

    unphased: ConnectorUnphased,

    phased: ConnectorPhased,

}

#[derive(Debug)]

pub struct ConnectorCommunication {

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        let cu = &mut self.unphased;

        // adds two new associated ports, related to each other, and exposed to the native

        let [o, i] = [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];

        cu.native_ports.insert(o);

        cu.native_ports.insert(i);

        // {polarity, peer, route} known. {} unknown.

        cu.port_info.polarities.insert(o, Putter);

        cu.port_info.polarities.insert(i, Getter);

        cu.port_info.peers.insert(o, i);

        cu.port_info.peers.insert(i, o);

        cu.port_info.routes.insert(o, route);

        cu.port_info.routes.insert(i, route);

        log!(cu.logger, "Added port pair (out->in) {:?} -> {:?}", o, i);

        [o, i]

    }

    pub fn add_component(

        &mut self,

        identifier: &[u8],

        ports: &[PortId],

    ) -> Result<(), AddComponentError> {

        // called by the USER. moves ports owned by the NATIVE

        use AddComponentError::*;

        }

        for (&expected_polarity, port) in polarities.iter().zip(ports.iter()) {

            if !cu.native_ports.contains(port) {

                return Err(UnknownPort(*port));

            }

            if expected_polarity != *cu.port_info.polarities.get(port).unwrap() {

                return Err(WrongPortPolarity { port: *port, expected_polarity });

            }

        }

        // 3. remove ports from old component & update port->route

        let new_id = cu.id_manager.new_proto_component_id();

        for port in ports.iter() {

        }

        cu.native_ports.retain(|port| !ports.contains(port));

        // 4. add new component

        cu.proto_components.insert(

            new_id,

            ProtoComponent {

                state: cu.proto_description.new_main_component(identifier, ports),

                ports: ports.iter().copied().collect(),

            },

        );

        Ok(())

    }

                        None

                    }

                });

                f.debug_set().entries(iter).finish()

            }

        }

        f.debug_struct("Predicate")

            .field("Trues", &MySet(self, true))

            .field("Falses", &MySet(self, false))

            .finish()

    }

}

        polarity: Polarity,

        sock_addr: SocketAddr,

        endpoint_polarity: EndpointPolarity,

    ) -> Result<PortId, ()> {

        let Self { unphased: up, phased } = self;

        match phased {

            ConnectorPhased::Setup { endpoint_setups, .. } => {

                let endpoint_setup = EndpointSetup { sock_addr, endpoint_polarity };

                let p = up.id_manager.new_port_id();

                up.native_ports.insert(p);

                // {polarity, route} known. {peer} unknown.

                up.port_info.polarities.insert(p, polarity);

                log!(

                    up.logger,

                    "Added net port {:?} with polarity {:?} and endpoint setup {:?} ",

                    p,

                    polarity,

                    &endpoint_setup

                );

                endpoint_setups.push((p, endpoint_setup));

                Ok(p)

            }

            ConnectorPhased::Communication { .. } => Err(()),

        }

            }

            msg @ Msg::CommMsg(..) => {

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

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

            }

        }

    }

    log!(

        cu.logger,

        "Gathered all children's maps. ConnectorId set is... {:?}",

        unoptimized_map.keys()

    );

    unoptimized_map.insert(cu.id_manager.connector_id, my_session_info);

    log!(cu.logger, "Inserting my own info. Unoptimized subtree map is {:?}", &unoptimized_map);

    // acquire the optimized info...

    let optimized_map = if let Some(parent) = comm.neighborhood.parent {

        // ... as a message from my parent

        log!(cu.logger, "Forwarding gathered info to parent {:?}", parent);

        let msg = S(SessionGather { unoptimized_map });

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

        'scatter_loop: loop {

            log!(

                cu.logger,

                msg @ S(SessionGather { .. })

                | msg @ S(YouAreMyParent)

                | msg @ S(MyPortInfo(..))

                | msg @ S(LeaderAnnounce { .. })

                | msg @ S(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");

    };

    log!(

        cu.logger,

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

        &optimized_map,

        comm.neighborhood.children.iter()

    );

    let optimized_info =

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

    let msg = S(SessionScatter { optimized_map });

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

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

    }

    apply_optimizations(cu, comm, optimized_info)?;

    Ok(())

}

fn leader_session_map_optimize(

    unoptimized_map: HashMap<ConnectorId, SessionInfo>,

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

    Ok(unoptimized_map)

}

fn apply_optimizations(

    _comm: &mut ConnectorCommunication,

) -> Result<(), ConnectError> {

    Ok(())

}