[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

#include "../../reowolf.h"

#include "../utility.c"

int main(int argc, char** argv) {

	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);

	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);

	rw_err_peek(c);

	PortId putter, getter;

	FfiSocketAddr local_addr = {{0, 0, 0, 0}, 8000};

	FfiSocketAddr peer_addr =  {{8, 8, 8, 1}, 8001};

	rw_err_peek(c);

	connector_add_udp_mediator_component(c, &putter, &getter, local_addr, peer_addr);

	connector_connect(c, -1);

	rw_err_peek(c);

	// Prepare a message to send

	size_t msg_len = 16;

	char * msg_ptr = malloc(msg_len);

	memset(msg_ptr, 42, msg_len);

	int i;

	for(i=0; i<10; i++) {

		connector_put_bytes(c, putter, msg_ptr, msg_len);

		rw_err_peek(c);

		connector_sync(c, -1);

		rw_err_peek(c);

	}

	printf("Exiting\n");

	protocol_description_destroy(pd);

	connector_destroy(c);

	free(msg_ptr);

	sleep(1.0);

	return 0;

}

script_path = os.path.dirname(os.path.realpath(__file__));

for c_file in glob.glob(script_path + "/*/*.c", recursive=False):

  print("compiling", c_file)

  args = [

    "gcc",          # compiler

    "-std=c11",     # C11 mode

    "-Wl,-R./",     # pass -R flag to linker: produce relocatable object

    c_file,         # input source file

    "-o",           # output flag

    c_file[:-2],    # output filename

    "-L",           # lib path flag

    "./",           # where to look for libs

    "-lreowolf_rs"  # add lib called "reowolf_rs"

  ];

  subprocess.run(args)

input("Blocking until newline...");

#ifndef REOWOLF_HEADER_DEFINED

#define REOWOLF_HEADER_DEFINED

#include <stdarg.h>

#include <stdbool.h>

#include <stdint.h>

#include <stdlib.h>

#define RW_BAD_FD -5

#define RW_BAD_SOCKADDR -8

#define RW_CLOSE_FAIL -4

#define RW_CONNECT_FAILED -6

#define RW_LOCK_POISONED -3

#define RW_OK 0

#define RW_TL_ERR -1

#define RW_WOULD_BLOCK -7

#define RW_WRONG_STATE -2

typedef enum {

  EndpointPolarity_Active,

  EndpointPolarity_Passive,

} EndpointPolarity;

typedef enum {

  Polarity_Putter,

  Polarity_Getter,

} Polarity;

typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;

typedef struct Connector Connector;

typedef uint32_t ConnectorId;

typedef uint32_t U32Suffix;

typedef struct {

  ConnectorId connector_id;

  U32Suffix u32_suffix;

typedef struct {

  uint8_t ipv4[4];

  uint16_t port;

} FfiSocketAddr;

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a string slice for the component's identifier in the connector's configured protocol description,

 * - a set of ports (represented as a slice; duplicates are ignored) in the native component's interface,

 * the connector creates a new (internal) protocol component C, such that the set of native ports are moved to C.

 * Usable in {setup, communication} states.

 */

int connector_add_component(Connector *connector,

                            const uint8_t *ident_ptr,

                            uintptr_t ident_len,

                            const PortId *ports_ptr,

                            uintptr_t ports_len);

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a utf-8 encoded socket address,

 * - the logical polarity of P,

 * - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,

 * returns P, a port newly added to the native interface.

 */

int connector_add_net_port(Connector *connector,

                           PortId *port,

                           FfiSocketAddr addr,

                           Polarity port_polarity,

                           EndpointPolarity endpoint_polarity);

/**

 * Given an initialized connector in setup or connecting state,

 * - Creates a new directed port pair with logical channel putter->getter,

 * - adds the ports to the native component's interface,

 * - and returns them using the given out pointers.

 * Usable in {setup, communication} states.

 */

void connector_add_port_pair(Connector *connector, PortId *out_putter, PortId *out_getter);

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a utf-8 encoded BIND socket addresses (i.e., "local"),

 * - a utf-8 encoded CONNECT socket addresses (i.e., "peer"),

 * returns [P, G] via out pointers [putter, getter],

 * - where P is a Putter port that sends messages into the socket

 * - where G is a Getter port that recvs messages from the socket

 */

int connector_add_udp_mediator_component(Connector *connector,

                                         PortId *putter,

                                         PortId *getter,

                                         FfiSocketAddr local_addr,

                                         FfiSocketAddr peer_addr);

/**

 * Connects this connector to the distributed system of connectors reachable through endpoints,

 * Usable in setup state, and changes the state to communication.

 */

int connector_connect(Connector *connector, int64_t timeout_millis);

/**

 * Destroys the given a pointer to the connector on the heap, freeing its resources.

 * Usable in {setup, communication} states.

 */

void connector_destroy(Connector *connector);

int connector_get(Connector *connector, PortId port);

const uint8_t *connector_gotten_bytes(Connector *connector, PortId port, uintptr_t *out_len);

/**

 * Initializes `out` with a new connector using the given protocol description as its configuration.

 * The connector uses the given (internal) connector ID.

 */

Connector *connector_new(const Arc_ProtocolDescription *pd);

Connector *connector_new_logging(const Arc_ProtocolDescription *pd,

                                 const uint8_t *path_ptr,

                                 uintptr_t path_len);

intptr_t connector_next_batch(Connector *connector);

void connector_print_debug(Connector *connector);

/**

 * Convenience function combining the functionalities of

 * "payload_new" with "connector_put_payload".

 */

int connector_put_bytes(Connector *connector,

                        PortId port,

                        const uint8_t *bytes_ptr,

                        uintptr_t bytes_len);

intptr_t connector_sync(Connector *connector, int64_t timeout_millis);

/**

 * Given an initialized protocol description, initializes `out` with a clone which can be independently created or destroyed.

 */

Arc_ProtocolDescription *protocol_description_clone(const Arc_ProtocolDescription *pd);

/**

 * Destroys the given initialized protocol description and frees its resources.

 */

void protocol_description_destroy(Arc_ProtocolDescription *pd);

/**

 * Parses the utf8-encoded string slice to initialize a new protocol description object.

 * - On success, initializes `out` and returns 0

 * - On failure, stores an error string (see `reowolf_error_peek`) and returns -1

 */

Arc_ProtocolDescription *protocol_description_parse(const uint8_t *pdl, uintptr_t pdl_len);

/**

 * Returns length (via out pointer) and pointer (via return value) of the last Reowolf error.

 * - pointer is NULL iff there was no last error

 * - data at pointer is null-delimited

 * - len does NOT include the length of the null-delimiter

 * If len is NULL, it will not written to.

 */

const uint8_t *reowolf_error_peek(uintptr_t *len);

#endif /* REOWOLF_HEADER_DEFINED */

    fmt::{Debug, Formatter},

    hash::Hash,

    ops::Range,

    time::Duration,

};

// pub(crate) use indexmap::IndexSet;

pub(crate) use maplit::hashmap;

pub(crate) use mio::{

    net::{TcpListener, TcpStream},

    Events, Interest, Poll, Token,

};

pub(crate) use std::{

    collections::{BTreeMap, HashMap, HashSet},

    convert::TryInto,

    io::{Read, Write},

    net::SocketAddr,

    sync::Arc,

    time::Instant,

};

pub(crate) use Polarity::*;

pub(crate) trait IdParts {

    fn id_parts(self) -> (ConnectorId, U32Suffix);

}

pub type ConnectorId = u32;

pub type U32Suffix = u32;

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

// acquired via error in the Rust API

pub struct ProtoComponentId(Id);

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(C)]

pub struct Id {

    pub(crate) connector_id: ConnectorId,

    pub(crate) u32_suffix: U32Suffix,

}

#[derive(Clone, Debug, Default)]

pub struct U32Stream {

    next: u32,

}

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(transparent)]

pub struct PortId(Id);

pub struct Payload(Arc<Vec<u8>>);

#[derive(

    Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(C)]

pub enum Polarity {

    Putter, // output port (from the perspective of the component)

    Getter, // input port (from the perspective of the component)

}

#[derive(

    Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,

)]

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

    NondetChoice { n: u16 },

}

pub(crate) struct DenseDebugHex<'a>(pub &'a [u8]);

///////////////////// 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 ProtoComponentId {

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

        }

        self.next += 1;

        self.next - 1

    }

    pub(crate) fn n_skipped(mut self, n: u32) -> Self {

        self.next = self.next.saturating_add(n);

        self

    }

}

impl From<Id> for PortId {

    fn from(id: Id) -> PortId {

        Self(id)

    }

}

impl From<Id> for ProtoComponentId {

    fn from(id: Id) -> ProtoComponentId {

        Self(id)

    }

}

impl From<&[u8]> for Payload {

    fn from(s: &[u8]) -> Payload {

        Payload(Arc::new(s.to_vec()))

    }

}

impl Payload {

    pub fn new(len: usize) -> Payload {

        let mut v = Vec::with_capacity(len);

        unsafe {

            v.set_len(len);

/// - len does NOT include the length of the null-delimiter

/// If len is NULL, it will not written to.

#[no_mangle]

pub unsafe extern "C" fn reowolf_error_peek(len: *mut usize) -> *const u8 {

    let (err_ptr, err_len) = StoredError::tl_bytes_peek();

    if !len.is_null() {

        len.write(err_len);

    }

    err_ptr

}

///////////////////// PROTOCOL DESCRIPTION //////////////////////////

/// Parses the utf8-encoded string slice to initialize a new protocol description object.

/// - On success, initializes `out` and returns 0

/// - On failure, stores an error string (see `reowolf_error_peek`) and returns -1

#[no_mangle]

pub unsafe extern "C" fn protocol_description_parse(

    pdl: *const u8,

    pdl_len: usize,

) -> *mut Arc<ProtocolDescription> {

    StoredError::tl_clear();

    match ProtocolDescription::parse(&*slice_from_raw_parts(pdl, pdl_len)) {

        Ok(new) => Box::into_raw(Box::new(Arc::new(new))),

        Err(err) => {

            StoredError::tl_bytes_store(err.as_bytes());

            std::ptr::null_mut()

        }

    }

}

/// Destroys the given initialized protocol description and frees its resources.

#[no_mangle]

pub unsafe extern "C" fn protocol_description_destroy(pd: *mut Arc<ProtocolDescription>) {

    drop(Box::from_raw(pd))

}

/// Given an initialized protocol description, initializes `out` with a clone which can be independently created or destroyed.

#[no_mangle]

pub unsafe extern "C" fn protocol_description_clone(

    pd: &Arc<ProtocolDescription>,

) -> *mut Arc<ProtocolDescription> {

    Box::into_raw(Box::new(pd.clone()))

}

///////////////////// CONNECTOR //////////////////////////

#[no_mangle]

    pd: &Arc<ProtocolDescription>,

    path_ptr: *const u8,

    path_len: usize,

) -> *mut Connector {

    StoredError::tl_clear();

    let path_bytes = &*slice_from_raw_parts(path_ptr, path_len);

    let path_str = match std::str::from_utf8(path_bytes) {

        Ok(path_str) => path_str,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            return std::ptr::null_mut();

        }

    };

    match std::fs::File::create(path_str) {

        Ok(file) => {

            let file_logger = Box::new(FileLogger::new(connector_id, file));

            let c = Connector::new(file_logger, pd.clone(), connector_id);

            Box::into_raw(Box::new(c))

        }

        Err(err) => {

            StoredError::tl_debug_store(&err);

            std::ptr::null_mut()

        }

    }

}

#[no_mangle]

pub unsafe extern "C" fn connector_print_debug(connector: &mut Connector) {

    println!("Debug print dump {:#?}", connector);

}

/// Initializes `out` with a new connector using the given protocol description as its configuration.

/// The connector uses the given (internal) connector ID.

#[no_mangle]

pub unsafe extern "C" fn connector_new(pd: &Arc<ProtocolDescription>) -> *mut Connector {

    let c = Connector::new(Box::new(DummyLogger), pd.clone(), Connector::random_id());

    Box::into_raw(Box::new(c))

}

/// Destroys the given a pointer to the connector on the heap, freeing its resources.

/// Usable in {setup, communication} states.

#[no_mangle]

pub unsafe extern "C" fn connector_destroy(connector: *mut Connector) {

    drop(Box::from_raw(connector))

}

/// Given an initialized connector in setup or connecting state,

/// - Creates a new directed port pair with logical channel putter->getter,

/// - adds the ports to the native component's interface,

/// - and returns them using the given out pointers.

/// Usable in {setup, communication} states.

#[no_mangle]

pub unsafe extern "C" fn connector_add_port_pair(

    connector: &mut Connector,

    out_putter: *mut PortId,

    out_getter: *mut PortId,

) {

    let [o, i] = connector.new_port_pair();

    out_putter.write(o);

    out_getter.write(i);

}

/// Given

/// - an initialized connector in setup or connecting state,

/// - a string slice for the component's identifier in the connector's configured protocol description,

/// - a set of ports (represented as a slice; duplicates are ignored) in the native component's interface,

/// the connector creates a new (internal) protocol component C, such that the set of native ports are moved to C.

/// Usable in {setup, communication} states.

#[no_mangle]

pub unsafe extern "C" fn connector_add_component(

    connector: &mut Connector,

    ident_ptr: *const u8,

    ident_len: usize,

/*

Change the definition of these macros to control the logging level statically

*/

macro_rules! log {

        if let Some(w) = $logger.line_writer() {

            let _ = writeln!(w, $($arg)*);

        }

    }};

    ($logger:expr, $($arg:tt)*) => {{

    }};

}

        } else {

            batch.to_put.insert(port, payload);

            Ok(())

        }

    }

    pub fn get(&mut self, port: PortId) -> Result<(), PortOpError> {

        use PortOpError as Poe;

        let batch = self.port_op_access(port, Getter)?;

        if batch.to_get.insert(port) {

            Ok(())

        } else {

            Err(Poe::MultipleOpsOnPort)

        }

    }

    // entrypoint for caller. overwrites round result enum, and returns what happened

    pub fn sync(&mut self, timeout: Option<Duration>) -> Result<usize, SyncError> {

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

        match phased {

            ConnectorPhased::Setup { .. } => Err(SyncError::NotConnected),

            ConnectorPhased::Communication(comm) => {

                match &comm.round_result {

                    Err(SyncError::Unrecoverable(e)) => {

                        log!(cu.inner.logger, "Attempted to start sync round, but previous error {:?} was unrecoverable!", e);

                        return Err(SyncError::Unrecoverable(e.clone()));

                    }

                    _ => {}