mod ast;

mod eval;

pub mod inputsource;

mod lexer;

mod library;

mod parser;

use crate::common::*;

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

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

use crate::protocol::inputsource::*;

use crate::protocol::parser::*;

use std::hint::unreachable_unchecked;

pub struct ProtocolDescriptionImpl {

    heap: Heap,

    source: InputSource,

}

impl std::fmt::Debug for ProtocolDescriptionImpl {

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

        write!(f, "Protocol")

    }

}

impl ProtocolDescription for ProtocolDescriptionImpl {

    type S = ComponentStateImpl;

    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(ProtocolDescriptionImpl { heap, source, root });

            }

            Err(err) => {

                let mut vec: Vec<u8> = Vec::new();

                err.write(&source, &mut vec).unwrap();

                Err(String::from_utf8_lossy(&vec).to_string())

            }

        }

    }

    fn component_polarities(&self, identifier: &[u8]) -> Result<Vec<Polarity>, MainComponentErr> {

        let h = &self.heap;

        let root = &h[self.root];

        let def = root.get_definition_ident(h, identifier);

        if def.is_none() {

            return Err(MainComponentErr::NoSuchComponent);

        }

        let def = &h[def.unwrap()];

        if !def.is_component() {

            return Err(MainComponentErr::NoSuchComponent);

        }

        for &param in def.parameters().iter() {

            let param = &h[param];

            let type_annot = &h[param.type_annotation];

            if type_annot.the_type.array {

                return Err(MainComponentErr::NonPortTypeParameters);

            }

            match type_annot.the_type.primitive {

                PrimitiveType::Input | PrimitiveType::Output => continue,

                _ => {

                    return Err(MainComponentErr::NonPortTypeParameters);

                }

            }

        }

        let mut result = Vec::new();

        for &param in def.parameters().iter() {

            let param = &h[param];

            let type_annot = &h[param.type_annotation];

            let ptype = &type_annot.the_type.primitive;

            if ptype == &PrimitiveType::Input {

                result.push(Polarity::Getter)

            } else if ptype == &PrimitiveType::Output {

                result.push(Polarity::Putter)

            } else {

                unreachable!()

            }

        }

        Ok(result)

    }

    fn new_main_component(&self, identifier: &[u8], ports: &[Key]) -> ComponentStateImpl {

        let mut args = Vec::new();

        for (&x, y) in ports.iter().zip(self.component_polarities(identifier).unwrap()) {

            match y {

                Polarity::Getter => args.push(Value::Input(InputValue(x))),

            }

        }

        let h = &self.heap;

        let root = &h[self.root];

        let def = root.get_definition_ident(h, identifier).unwrap();

    }

}

#[derive(Debug, Clone)]

pub struct ComponentStateImpl {

    prompt: Prompt,

}

impl ComponentState for ComponentStateImpl {

    type D = ProtocolDescriptionImpl;

    fn pre_sync_run<C: MonoContext<D = ProtocolDescriptionImpl, S = Self>>(

        &mut self,

        context: &mut C,

        pd: &ProtocolDescriptionImpl,

    ) -> MonoBlocker {

        let mut context = EvalContext::Mono(context);

        loop {

            let result = self.prompt.step(&pd.heap, &mut context);

            match result {

                // In component definitions, there are no return statements

                Ok(_) => unreachable!(),

                Err(cont) => match cont {

                    EvalContinuation::Stepping => continue,

                    EvalContinuation::Inconsistent => return MonoBlocker::Inconsistent,

                    EvalContinuation::Terminal => return MonoBlocker::ComponentExit,

                    EvalContinuation::SyncBlockStart => return MonoBlocker::SyncBlockStart,

                    // Not possible to end sync block if never entered one

                    EvalContinuation::SyncBlockEnd => unreachable!(),

                    EvalContinuation::NewComponent(args) => {

                        todo!();

                        continue;

                    }

                    // Outside synchronous blocks, no fires/get/put happens

                    EvalContinuation::BlockFires(val) => unreachable!(),

                    EvalContinuation::BlockGet(val) => unreachable!(),

                    EvalContinuation::Put(port, msg) => unreachable!(),

                },

            }

        }

    }

    fn sync_run<C: PolyContext<D = ProtocolDescriptionImpl>>(

        &mut self,

        context: &mut C,

        pd: &ProtocolDescriptionImpl,

    ) -> PolyBlocker {

        let mut context = EvalContext::Poly(context);

        loop {

                &old_predicate,

                &case

            );

            match case {

                Csr::Nonexistant => { /* skip branch */ }

                Csr::FormerNotLatter | Csr::Equivalent => {

                    // Feed the message to this branch in-place. no need to modify pred.

                    if branch.to_get.remove(&ekey) {

                        branch.gotten.insert(ekey, payload.clone());

                        report_if_solution(&branch, &old_predicate, logger);

                    }

                }

                Csr::LatterNotFormer => {

                    // create a new branch with the payload_predicate.

                    let mut forked = branch.clone();

                    if forked.to_get.remove(&ekey) {

                        forked.gotten.insert(ekey, payload.clone());

                        report_if_solution(&forked, &payload_predicate, logger);

                        branches2.insert(payload_predicate.clone(), forked);

                    }

                }

                Csr::New(new) => {

                    // create a new branch with the newly-created predicate

                    let mut forked = branch.clone();

                    if forked.to_get.remove(&ekey) {

                        forked.gotten.insert(ekey, payload.clone());

                        report_if_solution(&forked, &new, logger);

                        branches2.insert(new.clone(), forked);

                    }

                }

            }

            // unlike PolyP machines, Native branches do not become inconsistent

            branches2.insert(old_predicate, branch);

        }

        log!(

            logger,

            "Native now has {} branches with predicates: {:?}",

            branches2.len(),

            branches2.keys().collect::<Vec<_>>()

        );

        std::mem::swap(&mut branches2, &mut self.branches);

    }

    pub fn become_mono(

        mut self,

        decision: &Predicate,

        table_row: &mut HashMap<Key, Payload>,

    ) -> MonoN {

            let BranchN { gotten, sync_batch_index, .. } = branch;

            for (&key, payload) in gotten.iter() {

                assert!(table_row.insert(key, payload.clone()).is_none());

            }

            MonoN { ekeys: self.ekeys, result: Some((sync_batch_index, gotten)) }

        } else {

            panic!("No such solution!")

        }

    }

}

use crate::common::*;

use crate::runtime::{errors::*, *};

pub fn random_controller_id() -> ControllerId {

    type Bytes8 = [u8; std::mem::size_of::<ControllerId>()];

    let mut bytes = Bytes8::default();

    getrandom::getrandom(&mut bytes).unwrap();

    unsafe { std::mem::transmute::<Bytes8, ControllerId>(bytes) }

}

impl Default for Unconfigured {

    fn default() -> Self {

        let controller_id = random_controller_id();

        Self { controller_id }

    }

}

impl Default for Connector {

    fn default() -> Self {

        Self::Unconfigured(Unconfigured::default())

    }

}

impl Connector {

    /// Configure the Connector with the given Pdl description.

    pub fn configure(&mut self, pdl: &[u8], main_component: &[u8]) -> Result<(), ConfigErr> {

        use ConfigErr::*;

        let controller_id = match self {

            Connector::Configured(_) => return Err(AlreadyConfigured),

            Connector::Connected(_) => return Err(AlreadyConnected),

            Connector::Unconfigured(Unconfigured { controller_id }) => *controller_id,

        };

        let protocol_description = Arc::new(ProtocolD::parse(pdl).map_err(ParseErr)?);

        let polarities = protocol_description.component_polarities(main_component)?;

        let configured = Configured {

            controller_id,

            protocol_description,

            bindings: Default::default(),

            polarities,

        };

        *self = Connector::Configured(configured);

        Ok(())

    }

    /// Bind the (configured) connector's port corresponding to the

    pub fn bind_port(

        &mut self,

        proto_port_index: usize,

        binding: PortBinding,

    ) -> Result<(), PortBindErr> {

        use PortBindErr::*;

        match self {

            Connector::Unconfigured { .. } => Err(NotConfigured),

            Connector::Connected(_) => Err(AlreadyConnected),

            Connector::Configured(configured) => {

                if configured.polarities.len() <= proto_port_index {

                    return Err(IndexOutOfBounds);

                }

                configured.bindings.insert(proto_port_index, binding);

                Ok(())

            }

        }

    }

    pub fn connect(&mut self, timeout: Duration) -> Result<(), ConnectErr> {

        let deadline = Instant::now() + timeout;

        use ConnectErr::*;

        let configured = match self {

            Connector::Unconfigured { .. } => return Err(NotConfigured),

            Connector::Connected(_) => return Err(AlreadyConnected),

            Connector::Configured(configured) => configured,

        };

        // 1. Unwrap bindings or err

        let bound_proto_interface: Vec<(_, _)> = configured

            .iter()

            .copied()

            .enumerate()

            })

            .collect::<Result<Vec<(_, _)>, ConnectErr>>()?;

        let (controller, native_interface) = Controller::connect(

            configured.controller_id,

            configured.protocol_description.clone(),

            &bound_proto_interface[..],

            deadline,

        )?;

        *self = Connector::Connected(Connected {

            native_interface,

            sync_batches: vec![Default::default()],

            controller,

        });

        Ok(())

    }

    pub fn get_mut_logger(&mut self) -> Option<&mut String> {

        match self {

            Connector::Connected(connected) => Some(&mut connected.controller.inner.logger),

            _ => None,

        }

    }

    pub fn put(&mut self, native_port_index: usize, payload: Payload) -> Result<(), PortOpErr> {

        use PortOpErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

        };

        let (ekey, native_polarity) =

            *connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;

        if native_polarity != Putter {

            return Err(WrongPolarity);

        }

        let sync_batch = connected.sync_batches.iter_mut().last().unwrap();

        if sync_batch.puts.contains_key(&ekey) {

            return Err(DuplicateOperation);

        }

        sync_batch.puts.insert(ekey, payload);

        Ok(())

    }

    pub fn get(&mut self, native_port_index: usize) -> Result<(), PortOpErr> {

        use PortOpErr::*;

        let connected = match self {

            Connector::Connected(connected) => connected,

            _ => return Err(NotConnected),

        };

        let (ekey, native_polarity) =

            *connected.native_interface.get(native_port_index).ok_or(IndexOutOfBounds)?;

        if native_polarity != Getter {

            return Err(WrongPolarity);

        }

mod predicate; // TODO later

mod serde;

pub(crate) mod setup;

pub(crate) type ProtocolD = crate::protocol::ProtocolDescriptionImpl;

pub(crate) type ProtocolS = crate::protocol::ComponentStateImpl;

use crate::common::*;

use actors::*;

use endpoint::*;

use errors::*;

#[derive(Debug, PartialEq)]

pub(crate) enum CommonSatResult {

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

#[derive(Clone, Eq, PartialEq, Hash)]

pub(crate) struct Predicate {

    pub assigned: BTreeMap<ChannelId, bool>,

}

#[derive(Debug, Default)]

struct SyncBatch {

    puts: HashMap<Key, Payload>,

    gets: HashSet<Key>,

}

#[derive(Debug)]

pub enum Connector {

    Unconfigured(Unconfigured),

    Configured(Configured),

    Connected(Connected), // TODO consider boxing. currently takes up a lot of stack real estate

}

#[derive(Debug)]

pub struct Unconfigured {

    pub controller_id: ControllerId,

}

#[derive(Debug)]

pub struct Configured {

    controller_id: ControllerId,

    polarities: Vec<Polarity>,

    bindings: HashMap<usize, PortBinding>,

    protocol_description: Arc<ProtocolD>,

}

#[derive(Debug)]

pub struct Connected {

    native_interface: Vec<(Key, Polarity)>,

    sync_batches: Vec<SyncBatch>,

    controller: Controller,

}

#[derive(Debug, Copy, Clone)]

pub enum PortBinding {

    Native,

    Active(SocketAddr),

    Passive(SocketAddr),

}

#[derive(Debug)]

struct Arena<T> {

    storage: Vec<T>,

}

#[derive(Debug)]

struct ReceivedMsg {

    recipient: Key,

    msg: Msg,

}

#[derive(Debug)]

struct MessengerState {

    poll: Poll,

    events: Events,

    delayed: Vec<ReceivedMsg>,

    undelayed: Vec<ReceivedMsg>,

    polled_undrained: IndexSet<Key>,

}

#[derive(Debug)]

struct ChannelIdStream {

    controller_id: ControllerId,

    next_channel_index: ChannelIndex,

}

#[derive(Debug)]

struct Controller {

    protocol_description: Arc<ProtocolD>,

    inner: ControllerInner,

    ephemeral: ControllerEphemeral,

}

#[derive(Debug)]

struct ControllerInner {

use crate::common::*;

use crate::runtime::{

    actors::{MonoN, MonoP},

    endpoint::*,

    errors::*,

    *,

};

#[derive(Debug)]

enum EndpointExtTodo {

    Finished(EndpointExt),

    ActiveConnecting { addr: SocketAddr, polarity: Polarity, stream: TcpStream },

    ActiveRecving { addr: SocketAddr, polarity: Polarity, endpoint: Endpoint },

    PassiveAccepting { addr: SocketAddr, info: EndpointInfo, listener: TcpListener },

    PassiveConnecting { addr: SocketAddr, info: EndpointInfo, stream: TcpStream },

}

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

impl Controller {

    // Given port bindings and a protocol config, create a connector with 1 native node

    pub fn connect(

        major: ControllerId,

        protocol_description: Arc<ProtocolD>,

        bound_proto_interface: &[(PortBinding, Polarity)],

        deadline: Instant,

    ) -> Result<(Self, Vec<(Key, Polarity)>), ConnectErr> {

        use ConnectErr::*;

        let mut logger = String::default();

        log!(&mut logger, "CONNECT PHASE START! MY CID={:?} STARTING LOGGER ~", major);

        let mut channel_id_stream = ChannelIdStream::new(major);

        let mut endpoint_ext_todos = Arena::default();

        let mut ekeys_native = vec![];

        let mut ekeys_proto = vec![];

        let mut ekeys_network = vec![];

        let mut native_interface = vec![];

        /*

        1.  - allocate an EndpointExtTodo for every native and interface port

            - store all the resulting keys in two keylists for the interfaces of the native and proto components

                native: [a, c,    f]

                         |  |     |

                         |  |     |

                proto:  [b, d, e, g]

                               ^todo

                arena: <A,B,C,D,E,F,G>

        */

        for &(binding, polarity) in bound_proto_interface.iter() {

            match binding {

                PortBinding::Native => {

                    let channel_id = channel_id_stream.next();

                    let ([ekey_native, ekey_proto], native_polarity) = {

                        let [p, g] = Endpoint::new_memory_pair();

                        let mut endpoint_to_key = |endpoint, polarity| {

                            endpoint_ext_todos.alloc(EndpointExtTodo::Finished(EndpointExt {

                                endpoint,

                                info: EndpointInfo { polarity, channel_id },

                            }))

                        };

                        let pkey = endpoint_to_key(p, Putter);

                        let gkey = endpoint_to_key(g, Getter);

                        let key_pair = match polarity {

                            Putter => [gkey, pkey],

                            Getter => [pkey, gkey],

                        };

                        (key_pair, !polarity)

                    };

                    native_interface.push((ekey_native, native_polarity));

                    ekeys_native.push(ekey_native);

                    ekeys_proto.push(ekey_proto);

                }

                PortBinding::Passive(addr) => {

                    let channel_id = channel_id_stream.next();

                    let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::PassiveAccepting {

                        addr,

                        info: EndpointInfo { polarity, channel_id },

                        listener: TcpListener::bind(&addr).map_err(|_| BindFailed(addr))?,

                    });

                    ekeys_network.push(ekey_proto);

                    ekeys_proto.push(ekey_proto);

                }

                PortBinding::Active(addr) => {

                    let ekey_proto = endpoint_ext_todos.alloc(EndpointExtTodo::ActiveConnecting {

                        addr,

                        polarity,

                        stream: TcpStream::connect(&addr).unwrap(),

                    });

                    ekeys_network.push(ekey_proto);

                    ekeys_proto.push(ekey_proto);

                }

            }

        }

        log!(&mut logger, "{:03?} setup todos...", major);

        // 2. convert the arena to Arena<EndpointExt>  and return the

        let (mut messenger_state, mut endpoint_exts) =

            Self::finish_endpoint_ext_todos(major, &mut logger, endpoint_ext_todos, deadline)?;

        let n_mono = Some(MonoN { ekeys: ekeys_native.into_iter().collect(), result: None });

        let p_monos = vec![MonoP {

            ekeys: ekeys_proto.into_iter().collect(),

        }];

        // 6. Become a node in a sink tree, computing {PARENT, CHILDREN} from {NEIGHBORS}

        let family = Self::setup_sink_tree_family(

            major,

            &mut logger,

            &mut endpoint_exts,

            &mut messenger_state,

            ekeys_network,

            deadline,

        )?;

        log!(&mut logger, "CONNECT PHASE END! ~");

        let inner = ControllerInner {

            family,

            messenger_state,

            channel_id_stream,

            endpoint_exts,

            mono_ps: p_monos,

            mono_n: n_mono,

            round_index: 0,

            logger,

        };

        let controller = Self { protocol_description, inner, ephemeral: Default::default() };

        Ok((controller, native_interface))

    }

    fn test_stream_connectivity(stream: &mut TcpStream) -> bool {

        use std::io::Write;

        stream.write(&[]).is_ok()

    }

    // inserts

    fn finish_endpoint_ext_todos(

        major: ControllerId,

        logger: &mut String,

        mut endpoint_ext_todos: Arena<EndpointExtTodo>,

        deadline: Instant,

    ) -> Result<(MessengerState, Arena<EndpointExt>), ConnectErr> {

        use {ConnectErr::*, EndpointExtTodo::*};

        // 1. define and setup a poller and event loop

        let edge = PollOpt::edge();

        let [ready_r, ready_w] = [Ready::readable(), Ready::writable()];

        let mut ms = MessengerState {

            poll: Poll::new().map_err(|_| PollInitFailed)?,

            events: Events::with_capacity(endpoint_ext_todos.len()),

use crate::common::*;

use crate::runtime::*;

use PortBinding::*;

use super::*;

#[test]

fn config_ok_0() {

    let pdl = b"primitive main() {}";

    let d = ProtocolD::parse(pdl).unwrap();

    assert_eq!(&pol[..], &[]);

}

#[test]

fn config_ok_2() {

    let pdl = b"primitive main(in x, out y) {}";

    let d = ProtocolD::parse(pdl).unwrap();

}

#[test]

#[should_panic]

fn config_non_port() {

    let pdl = b"primitive main(in q, int q) {}";

    ProtocolD::parse(pdl).unwrap();

}

#[test]

fn config_and_connect_2() {

    let timeout = Duration::from_millis(1_500);

    let addrs = ["127.0.0.1:9000".parse().unwrap(), "127.0.0.1:9001".parse().unwrap()];

    use std::thread;

    let handles = vec![

        //

        thread::spawn(move || {

            let mut x = Connector::Unconfigured(Unconfigured { controller_id: 0 });

            x.bind_port(0, Passive(addrs[0])).unwrap();

            x.bind_port(1, Passive(addrs[1])).unwrap();

            x.connect(timeout).unwrap();

        }),

        thread::spawn(move || {

            let mut x = Connector::Unconfigured(Unconfigured { controller_id: 1 });

            x.bind_port(0, Active(addrs[0])).unwrap();

            x.bind_port(1, Active(addrs[1])).unwrap();

            x.connect(timeout).unwrap();

        }),

    ];

    for h in handles {

        handle(h.join())

    }

}

#[test]

fn bind_too_much() {

    let mut x = Connector::Unconfigured(Unconfigured { controller_id: 0 });

    x.bind_port(0, Native).unwrap();

    assert!(x.bind_port(1, Native).is_err());

}

#[test]

fn config_and_connect_chain() {

    let timeout = Duration::from_millis(1_500);

    let addrs = [

        "127.0.0.1:9002".parse().unwrap(),

        "127.0.0.1:9003".parse().unwrap(),

        "127.0.0.1:9004".parse().unwrap(),

    ];

    use std::thread;

    let handles = vec![

        //

        thread::spawn(move || {

            // PRODUCER A->

            let mut x = Connector::Unconfigured(Unconfigured { controller_id: 0 });

            x.bind_port(0, Active(addrs[0])).unwrap();

            x.connect(timeout).unwrap();

        }),

        thread::spawn(move || {

            // FORWARDER ->B->

            let mut x = Connector::Unconfigured(Unconfigured { controller_id: 1 });

            x.bind_port(0, Passive(addrs[0])).unwrap();

            x.bind_port(1, Active(addrs[1])).unwrap();

            x.connect(timeout).unwrap();

        }),

        thread::spawn(move || {

            // FORWARDER ->C->

            let mut x = Connector::Unconfigured(Unconfigured { controller_id: 2 });

            x.bind_port(0, Passive(addrs[1])).unwrap();

            x.bind_port(1, Active(addrs[2])).unwrap();

            x.connect(timeout).unwrap();

        }),

        thread::spawn(move || {

            // CONSUMER ->D

            let mut x = Connector::Unconfigured(Unconfigured { controller_id: 3 });

            x.bind_port(0, Passive(addrs[2])).unwrap();

            x.connect(timeout).unwrap();

        }),

    ];

    for h in handles {

        handle(h.join())

    }

}