[package]

name = "reowolf_rs"

version = "0.1.4"

authors = [

	"Christopher Esterhuyse <esterhuy@cwi.nl, christopher.esterhuyse@gmail.com>",

	"Hans-Dieter Hiep <hdh@cwi.nl>"

]

edition = "2018"

[dependencies]

# convenience macros

maplit = "1.0.2"

derive_more = "0.99.2"

# runtime

bincode = "1.3.1"

serde = { version = "1.0.114", features = ["derive"] }

getrandom = "0.1.14" # tiny crate. used to guess controller-id

# network

mio = { version = "0.7.0", package = "mio", features = ["udp", "tcp", "os-poll"] }

socket2 = { version = "0.3.12", optional = true }

# protocol

backtrace = "0.3"

lazy_static = "1.4.0"

# ffi

# socket ffi

libc = { version = "^0.2", optional = true }

os_socketaddr = { version = "0.1.0", optional = true }

[dev-dependencies]

# test-generator = "0.3.0"

crossbeam-utils = "0.7.2"

lazy_static = "1.4.0"

[lib]

crate-type = [

	"rlib", # for use as a Rust dependency. 

	"cdylib" # for FFI use, typically C.

]

[features]

ffi = [] # see src/ffi/mod.rs

ffi_pseudo_socket_api = ["ffi", "libc", "os_socketaddr"]# see src/ffi/pseudo_socket_api.rs.

endpoint_logging = [] # see src/macros.rs

session_optimization = [] # see src/runtime/setup.rs

use std::fmt;

use std::fs::File;

use std::io;

use std::path::Path;

use backtrace::Backtrace;

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

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

}

primitive sync(in i, out o) {

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

}

primitive alternator(in i, out l, out r) {

    while(true) {

        synchronous() if(fires(i)) put(l, get(i));

        synchronous() if(fires(i)) put(r, get(i));

    }

}

primitive replicator(in i, out l, out r) {

    }

}

primitive merger(in l, in r, out o) {

    while(true) synchronous {

        if(fires(l))      put(o, get(l));

        else if(fires(r)) put(o, get(r));

    }

}";

impl InputSource {

    // Constructors

    pub fn new<R: io::Read, S: ToString>(filename: S, reader: &mut R) -> io::Result<InputSource> {

        let mut vec = STD_LIB_PDL.to_vec();

        reader.read_to_end(&mut vec)?;

        Ok(InputSource {

            filename: filename.to_string(),

            input: vec,

            line: 1,

            column: 1,

            offset: 0,

        })

    }

    // Constructor helpers

    pub fn from_file(path: &Path) -> io::Result<InputSource> {

        let filename = path.file_name();

        match filename {

            Some(filename) => {

                let mut f = File::open(path)?;

                InputSource::new(filename.to_string_lossy(), &mut f)

            }

            None => Err(io::Error::new(io::ErrorKind::NotFound, "Invalid path")),

        }

    }

    pub fn from_string(string: &str) -> io::Result<InputSource> {

        let buffer = Box::new(string);

        let mut bytes = buffer.as_bytes();

        InputSource::new(String::new(), &mut bytes)

    }

    pub fn from_buffer(buffer: &[u8]) -> io::Result<InputSource> {

        InputSource::new(String::new(), &mut Box::new(buffer))

    }

    // Internal methods

    pub fn pos(&self) -> InputPosition {

        InputPosition { line: self.line, column: self.column, offset: self.offset }

    }

    pub fn error<S: ToString>(&self, message: S) -> ParseError {

        self.pos().parse_error(message)

    }

    pub fn is_eof(&self) -> bool {

        self.next() == None

    }

    pub fn next(&self) -> Option<u8> {

        if self.offset < self.input.len() {

            Some((*self.input)[self.offset])

        } else {

            None

        }

    }

    pub fn lookahead(&self, pos: usize) -> Option<u8> {

        if let Some(x) = usize::checked_add(self.offset, pos) {

            if x < self.input.len() {

                return Some((*self.input)[x]);

            }

        }

        None

    }

    pub fn consume(&mut self) {

        match self.next() {

            Some(x) if x == b'\r' && self.lookahead(1) != Some(b'\n') || x == b'\n' => {

                self.line += 1;

                self.offset += 1;

                self.column = 1;

            }

            Some(_) => {

                self.offset += 1;

                self.column += 1;

            }

            None => {}

        }

    }

}

impl fmt::Display for InputSource {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        self.pos().fmt(f)

    }

}

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

pub struct InputPosition {

    line: usize,

    column: usize,

    offset: usize,

}

impl InputPosition {

    fn context<'a>(&self, source: &'a InputSource) -> &'a [u8] {

        let start = self.offset - (self.column - 1);

        let mut end = self.offset;

        while end < source.input.len() {

            let cur = (*source.input)[end];

            if cur == b'\n' || cur == b'\r' {

                break;

            }

            end += 1;

        }

        &source.input[start..end]

    }

    fn parse_error<S: ToString>(&self, message: S) -> ParseError {

        ParseError { position: *self, message: message.to_string(), backtrace: Backtrace::new() }

    }

    fn eval_error<S: ToString>(&self, message: S) -> EvalError {

        EvalError { position: *self, message: message.to_string(), backtrace: Backtrace::new() }

    }

}

impl fmt::Display for InputPosition {

    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

        write!(f, "{}:{}", self.line, self.column)

    }

}

pub trait SyntaxElement {

    fn position(&self) -> InputPosition;

    fn error<S: ToString>(&self, message: S) -> EvalError {

        self.position().eval_error(message)

    }

}

#[derive(Debug, Clone)]

pub struct ParseError {

    position: InputPosition,

    message: String,

    backtrace: Backtrace,

}

impl ParseError {

    pub fn new<S: ToString>(position: InputPosition, message: S) -> ParseError {

        ParseError { position, message: message.to_string(), backtrace: Backtrace::new() }

    }

    // Diagnostic methods

    pub fn write<A: io::Write>(&self, source: &InputSource, writer: &mut A) -> io::Result<()> {

        if !source.filename.is_empty() {

            writeln!(

                writer,

                "Parse error at {}:{}: {}",

                source.filename, self.position, self.message

            )?;

        } else {

            writeln!(writer, "Parse error at {}: {}", self.position, self.message)?;

        }

        let line = self.position.context(source);

        writeln!(writer, "{}", String::from_utf8_lossy(line))?;

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

        for pos in 1..self.position.column {

            let c = line[pos - 1];

            if c == b'\t' {

                arrow.push(b'\t')

            } else {

                arrow.push(b' ')

            }

        }

        arrow.push(b'^');

        writeln!(writer, "{}", String::from_utf8_lossy(&arrow))

    }

    pub fn print(&self, source: &InputSource) {

        self.write(source, &mut std::io::stdout()).unwrap()

    }

    pub fn display<'a>(&'a self, source: &'a InputSource) -> DisplayParseError<'a> {

        DisplayParseError::new(self, source)

    }

}

impl From<ParseError> for io::Error {

    fn from(_: ParseError) -> io::Error {

        io::Error::new(io::ErrorKind::InvalidInput, "parse error")

    }

}

#[derive(Clone, Copy)]

pub struct DisplayParseError<'a> {

    error: &'a ParseError,

    source: &'a InputSource,

}

impl DisplayParseError<'_> {

    fn new<'a>(error: &'a ParseError, source: &'a InputSource) -> DisplayParseError<'a> {

        DisplayParseError { error, source }

    }

}

impl fmt::Display for DisplayParseError<'_> {

    }

    /// Conceptually, this returning [p0, g1] is sugar for:

    /// 1. create port pair [p0, g0]

    /// 2. create port pair [p1, g1]

    /// 3. create udp component with interface of moved ports [p1, g0]

    /// 4. return [p0, g1]

    pub fn new_udp_mediator_component(

        &mut self,

        local_addr: SocketAddr,

        peer_addr: SocketAddr,

    ) -> Result<[PortId; 2], WrongStateError> {

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

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                let udp_index = setup.udp_endpoint_setups.len();

                let udp_cid = cu.ips.id_manager.new_component_id();

                // allocates 4 new port identifiers, two for each logical channel,

                // one channel per direction (into and out of the component)

                let mut npid = || cu.ips.id_manager.new_port_id();

                let [nin, nout, uin, uout] = [npid(), npid(), npid(), npid()];

                // allocate the native->udp_mediator channel's ports

                cu.ips.port_info.map.insert(

                    nout,

                    PortInfo {

                        route: Route::LocalComponent,

                        polarity: Putter,

                        peer: Some(uin),

                        owner: cu.native_component_id,

                    },

                );

                cu.ips.port_info.map.insert(

                    uin,

                    PortInfo {

                        route: Route::UdpEndpoint { index: udp_index },

                        polarity: Getter,

                        peer: Some(uin),

                        owner: udp_cid,

                    },

                );

                // allocate the udp_mediator->native channel's ports

                cu.ips.port_info.map.insert(

                    uout,

                    PortInfo {

                        route: Route::UdpEndpoint { index: udp_index },

                        polarity: Putter,

                        peer: Some(uin),

                        owner: udp_cid,

                    },

                );

                cu.ips.port_info.map.insert(

                    nin,

                    PortInfo {

                        route: Route::LocalComponent,

                        polarity: Getter,

                        peer: Some(uout),

                        owner: cu.native_component_id,

                    },

                );

                // allocate the two ports owned by the UdpMediator component

                // Remember to setup this UdpEndpoint setup during `connect` later.

                setup.udp_endpoint_setups.push(UdpEndpointSetup {

                    local_addr,

                    peer_addr,

                    getter_for_incoming: nin,

                });

                // update owned sets

                cu.ips

                    .port_info

                    .owned

                    .entry(cu.native_component_id)

                    .or_default()

                    .extend([nin, nout].iter().copied());

                cu.ips.port_info.owned.insert(udp_cid, maplit::hashset! {uin, uout});

                // Return the native's output, input port pair

                Ok([nout, nin])

            }

        }

    }

    /// Adds a "dangling" port to the connector in the setup phase,

    /// to be formed into channel during the connect procedure with the given

    /// transport layer information.

    pub fn new_net_port(

        &mut self,

        polarity: Polarity,

        sock_addr: SocketAddr,

        endpoint_polarity: EndpointPolarity,

    ) -> Result<PortId, WrongStateError> {

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

        match phased {

            ConnectorPhased::Communication(..) => Err(WrongStateError),

            ConnectorPhased::Setup(setup) => {

                // allocate a single dangling port with a `None` peer (for now)

                let new_pid = cu.ips.id_manager.new_port_id();

                cu.ips.port_info.map.insert(

                    new_pid,

                    PortInfo {

                        route: Route::LocalComponent,

                        peer: None,

                        owner: cu.native_component_id,

                        polarity,

                    },

                );

                log!(

                    cu.logger,

                    "Added net port {:?} with polarity {:?} addr {:?} endpoint_polarity {:?}",

                    new_pid,

                    polarity,

                    &sock_addr,

                    endpoint_polarity

                );

                // Remember to setup this NetEndpoint setup during `connect` later.

                setup.net_endpoint_setups.push(NetEndpointSetup {

                    sock_addr,

                    endpoint_polarity,

                    getter_for_incoming: new_pid,

                });

                // update owned set

                cu.ips.port_info.owned.entry(cu.native_component_id).or_default().insert(new_pid);

                Ok(new_pid)

            }

        }

    }

    /// Finalizes the connector's setup procedure and forms a distributed system with

    /// all other connectors reachable through network channels. This procedure represents

    /// a synchronization barrier, and upon successful return, the connector can no longer add new network ports,

    /// but is ready to begin the first communication round.

    /// Initially, the connector has a singleton set of _batches_, the only element of which is empty.

    /// This single element starts off selected. The selected batch is modified with `put` and `get`,

    /// and new batches are added and selected with `next_batch`. See `sync` for an explanation of the

    /// purpose of these batches.

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

        use ConnectError as Ce;

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

        match &phased {

            ConnectorPhased::Communication { .. } => {

                log!(cu.logger, "Call to connecting in connected state");

                Err(Ce::AlreadyConnected)

            }

            ConnectorPhased::Setup(setup) => {

                // Idea: Clone `self.unphased`, and then pass the replica to

                // `connect_inner` to do the work, attempting to create a new connector structure

                // in connected state without encountering any errors.

                // If anything goes wrong during `connect_inner`, we simply keep the original `cu`.

                // Ideally, we'd simply clone `cu` in its entirety.

                // However, it isn't clonable, because of the pesky logger.

                // Solution: the original and clone ConnectorUnphased structures

                // 'share' the original logger by using `mem::swap` strategically to pass a dummy back and forth,

                // such that the real logger is wherever we need it to be without violating any invariants.

                let mut cu_clone = ConnectorUnphased {

                    logger: Box::new(DummyLogger),

                    proto_components: cu.proto_components.clone(),

                    native_component_id: cu.native_component_id.clone(),

                    ips: cu.ips.clone(),

                    proto_description: cu.proto_description.clone(),

                };

                // cu has REAL logger...

                std::mem::swap(&mut cu.logger, &mut cu_clone.logger);

                // ... cu_clone has REAL logger.

                match Self::connect_inner(cu_clone, setup, timeout) {

                    Ok(connected_connector) => {

                        *self = connected_connector;

                        Ok(())

                    }

                    Err((err, mut logger)) => {

                        // Put the original logger back in place (in self.unphased, AKA `cu`).

                        // cu_clone has REAL logger...

                        std::mem::swap(&mut cu.logger, &mut logger);

                        // ... cu has REAL logger.

                        Err(err)

                    }

                }

            }

        }

    }

    // Given an immutable setup structure, and my own (cloned) ConnetorUnphased,

    // attempt to complete the setup procedure and return a new connector in Connected state.

    // If anything goes wrong, throw everything in the bin, except for the Logger, which is

    // the only structure that sees lasting effects of the failed attempt.

    fn connect_inner(

        mut cu: ConnectorUnphased,

        setup: &ConnectorSetup,

        timeout: Option<Duration>,

    ) -> Result<Self, (ConnectError, Box<dyn Logger>)> {

        log!(cu.logger, "~~~ CONNECT called timeout {:?}", timeout);

        let deadline = timeout.map(|to| Instant::now() + to);

        let mut try_complete = || {

            // connect all endpoints in parallel; send and receive peer ids through ports

            let mut endpoint_manager = setup_endpoints_and_pair_ports(

                &mut *cu.logger,

                &setup.net_endpoint_setups,

                &setup.udp_endpoint_setups,

                &mut cu.ips.port_info,

                &deadline,

            )?;

            log!(

                cu.logger,

                "Successfully connected {} endpoints. info now {:#?} {:#?}",

                endpoint_manager.net_endpoint_store.endpoint_exts.len(),

                &cu.ips.port_info,

                &endpoint_manager,

            );

            // leader election and tree construction. Learn our role in the consensus tree,

            // from learning who are our children/parents (neighbors) in the consensus tree.

            let neighborhood = init_neighborhood(

                cu.ips.id_manager.connector_id,

                &mut *cu.logger,

                &mut endpoint_manager,

                &deadline,

            )?;

            log!(cu.logger, "Successfully created neighborhood {:?}", &neighborhood);

            // Put it all together with an initial round index of zero.

            let mut comm = ConnectorCommunication {

                round_index: 0,

                endpoint_manager,

                neighborhood,

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

                round_result: Ok(None), // no previous round yet

            };

            if cfg!(feature = "session_optimization") {

                // Perform the session optimization procedure, which may modify the

                // internals of the connector, rerouting ports, moving around connectors etc.

                session_optimize(&mut cu, &mut comm, &deadline)?;

            }

            log!(cu.logger, "connect() finished. setup phase complete");

            Ok(comm)

        };

        match try_complete() {

            Ok(comm) => {

                Ok(Self { unphased: cu, phased: ConnectorPhased::Communication(Box::new(comm)) })

            }

            Err(err) => Err((err, cu.logger)),

        }

    }

}

// Given a set of net_ and udp_ endpoints to setup,

// port information to flesh out (by discovering peers through channels)

// and a deadline in which to do it,

// try to return:

// - An EndpointManager, containing all the set up endpoints

// - new information about ports acquired through the newly-created channels

fn setup_endpoints_and_pair_ports(

    logger: &mut dyn Logger,

    net_endpoint_setups: &[NetEndpointSetup],

    udp_endpoint_setups: &[UdpEndpointSetup],

    port_info: &mut PortInfoMap,

    deadline: &Option<Instant>,

) -> Result<EndpointManager, ConnectError> {

    use ConnectError as Ce;

    const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE);

    const RETRY_PERIOD: Duration = Duration::from_millis(200);

    // The data for a net endpoint's setup in progress

    struct NetTodo {

        // becomes completed once sent_local_port && recv_peer_port.is_some()

        // we send local port if we haven't already and we receive a writable event

        // we recv peer port if we haven't already and we receive a readbale event

        todo_endpoint: NetTodoEndpoint,

        endpoint_setup: NetEndpointSetup,

        sent_local_port: bool,          // true <-> I've sent my local port

        recv_peer_port: Option<PortId>, // Some(..) <-> I've received my peer's port

    }

    // The data for a udp endpoint's setup in progress

    struct UdpTodo {

        // becomes completed once we receive our first writable event

        getter_for_incoming: PortId,