[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...");

/* CBindgen generated */

#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 */

///////////////////// PRELUDE /////////////////////

pub(crate) use crate::protocol::{ComponentState, ProtocolDescription};

pub(crate) use crate::runtime::{error::AddComponentError, NonsyncProtoContext, SyncProtoContext};

pub(crate) use core::{

    cmp::Ordering,

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

        }

        Payload(Arc::new(v))

    }

    pub fn len(&self) -> usize {

        self.0.len()

    }

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

        &self.0

    }

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

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

    }

    pub fn concatenate_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 {

    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>

    where

        S: serde::Serializer,

    {

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

        inner.serialize(serializer)

    }

}

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

    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>

    where

        D: serde::Deserializer<'de>,

    {

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

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

    }

}

impl From<Vec<u8>> for Payload {

    fn from(s: Vec<u8>) -> Self {

        Self(s.into())

    }

}

impl Debug for PortId {

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

        let (a, b) = self.id_parts();

        write!(f, "pid{}_{}", a, b)

    }

}

impl Debug for ProtoComponentId {

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

        let (a, b) = self.id_parts();

        write!(f, "cid{}_{}", a, b)

    }

}

impl Debug for Payload {

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

        write!(f, "Payload[{:?}]", DenseDebugHex(self.as_slice()))

    }

}

impl std::ops::Not for Polarity {

    type Output = Self;

    fn not(self) -> Self::Output {

        use Polarity::*;

        match self {

            Putter => Getter,

            Getter => Putter,

        }

    }

}

impl Debug for DenseDebugHex<'_> {

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

        for b in self.0 {

            write!(f, "{:02X?}", b)?;

        }

        Ok(())

    }

}

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

use core::{cell::RefCell, convert::TryFrom};

use std::os::raw::c_int;

use std::slice::from_raw_parts as slice_from_raw_parts;

#[cfg(all(target_os = "linux", feature = "ffi_pseudo_socket_api"))]

/// cbindgen:ignore

pub mod pseudo_socket_api;

// Temporary simplfication: ignore ipv6. To revert, just refactor this structure and its usages

#[repr(C)]

pub struct FfiSocketAddr {

    pub ipv4: [u8; 4],

    pub port: u16,

}

impl Into<SocketAddr> for FfiSocketAddr {

    fn into(self) -> SocketAddr {

        (self.ipv4, self.port).into()

    }

}

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

#[derive(Default)]

struct StoredError {

    // invariant: len is zero IFF its occupied

    // contents are 1+ bytes because we also store the NULL TERMINATOR

    buf: Vec<u8>,

}

impl StoredError {

    const NULL_TERMINATOR: u8 = 0;

    fn clear(&mut self) {

        // no null terminator either!

        self.buf.clear();

    }

    fn debug_store<E: Debug>(&mut self, error: &E) {

        let _ = write!(&mut self.buf, "{:?}", error);

        self.buf.push(Self::NULL_TERMINATOR);

    }

    fn tl_debug_store<E: Debug>(error: &E) {

        STORED_ERROR.with(|stored_error| {

            let mut stored_error = stored_error.borrow_mut();

            stored_error.clear();

            stored_error.debug_store(error);

        })

    }

    fn bytes_store(&mut self, bytes: &[u8]) {

        let _ = self.buf.write_all(bytes);

        self.buf.push(Self::NULL_TERMINATOR);

    }

    fn tl_bytes_store(bytes: &[u8]) {

        STORED_ERROR.with(|stored_error| {

            let mut stored_error = stored_error.borrow_mut();

            stored_error.clear();

            stored_error.bytes_store(bytes);

        })

    }

    fn tl_clear() {

        STORED_ERROR.with(|stored_error| {

            let mut stored_error = stored_error.borrow_mut();

            stored_error.clear();

        })

    }

    fn tl_bytes_peek() -> (*const u8, usize) {

        STORED_ERROR.with(|stored_error| {

            let stored_error = stored_error.borrow();

            match stored_error.buf.len() {

                0 => (core::ptr::null(), 0), // no error!

                n => {

                    // stores an error of length n-1 AND a NULL TERMINATOR

                    (stored_error.buf.as_ptr(), n - 1)

                }

            }

        })

    }

}

thread_local! {

    static STORED_ERROR: RefCell<StoredError> = RefCell::new(StoredError::default());

}

pub const RW_OK: c_int = 0;

pub const RW_TL_ERR: c_int = -1;

pub const RW_WRONG_STATE: c_int = -2;

pub const RW_LOCK_POISONED: c_int = -3;

pub const RW_CLOSE_FAIL: c_int = -4;

pub const RW_BAD_FD: c_int = -5;

pub const RW_CONNECT_FAILED: c_int = -6;

pub const RW_WOULD_BLOCK: c_int = -7;

pub const RW_BAD_SOCKADDR: c_int = -8;

///////////////////// REOWOLF //////////////////////////

/// 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.

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