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

            }

                    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,

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

        }

    }

}

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

        };

}

#[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,

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]

                         |  |     |

                         |  |     |

                        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,

        };

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

    }

}