Changeset - bdcf814bde8c
[Not reviewed]
1 5 2
Christopher Esterhuyse - 5 years ago 2020-07-10 12:27:39
christopher.esterhuyse@gmail.com
first step toward socket-mimicking FFI
7 files changed with 642 insertions and 39 deletions:
0 comments (0 inline, 0 general)
Cargo.toml
Show inline comments
 
@@ -18,13 +18,15 @@ serde = { version = "1.0.114", features = ["derive"] }
 
getrandom = "0.1.14" # tiny crate. used to guess controller-id
 

	
 
# network
 
# integer-encoding = "1.1.5"
 
# byteorder = "1.3.4"
 
mio = { version = "0.7.0", package = "mio", features = ["udp", "tcp", "os-poll"] }
 

	
 
# protocol
 
backtrace = "0.3"
 

	
 
# socket ffi
 
lazy_static = { version = "1.4.0", optional = true}
 
atomic_refcell = { version = "0.1.6", optional = true }
 

	
 
[dev-dependencies]
 
# test-generator = "0.3.0"
 
crossbeam-utils = "0.7.2"
 
@@ -36,6 +38,7 @@ crate-type = ["cdylib"]
 

	
 
[features]
 
default = ["ffi", "session_optimization"]
 
ffi = [] # no feature dependencies
 
ffi = [] # see src/ffi.rs
 
ffi_socket_api = ["ffi", "lazy_static", "atomic_refcell"]
 
endpoint_logging = [] # see src/macros.rs
 
session_optimization = [] # see src/runtime/setup.rs
 
\ No newline at end of file
reowolf.h
Show inline comments
 
@@ -139,3 +139,481 @@ Arc_ProtocolDescription *protocol_description_parse(const uint8_t *pdl, uintptr_
 
const uint8_t *reowolf_error_peek(uintptr_t *len);
 

	
 
#endif /* REOWOLF_HEADER_DEFINED */
 
/* CBindgen generated */
 

	
 
#ifndef REOWOLF_HEADER_DEFINED
 
#define REOWOLF_HEADER_DEFINED
 

	
 
#include <stdarg.h>
 
#include <stdbool.h>
 
#include <stdint.h>
 
#include <stdlib.h>
 

	
 
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 int ErrorCode;
 

	
 
typedef uint32_t ConnectorId;
 

	
 
typedef uint32_t PortSuffix;
 

	
 
typedef struct {
 
  ConnectorId connector_id;
 
  PortSuffix u32_suffix;
 
} Id;
 

	
 
typedef Id PortId;
 

	
 
/**
 
 * 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.
 
 */
 
ErrorCode 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.
 
 */
 
ErrorCode connector_add_net_port(Connector *connector,
 
                                 PortId *port,
 
                                 const uint8_t *addr_str_ptr,
 
                                 uintptr_t addr_str_len,
 
                                 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
 
 */
 
ErrorCode connector_add_udp_port(Connector *connector,
 
                                 PortId *putter[2],
 
                                 const uint8_t *local_addr_str_ptr,
 
                                 uintptr_t local_addr_str_len,
 
                                 const uint8_t *peer_addr_str_ptr,
 
                                 uintptr_t peer_addr_str_len);
 

	
 
/**
 
 * Connects this connector to the distributed system of connectors reachable through endpoints,
 
 * Usable in setup state, and changes the state to communication.
 
 */
 
ErrorCode 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);
 

	
 
ErrorCode 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".
 
 */
 
ErrorCode 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 */
 
/* CBindgen generated */
 

	
 
#ifndef REOWOLF_HEADER_DEFINED
 
#define REOWOLF_HEADER_DEFINED
 

	
 
#include <stdarg.h>
 
#include <stdbool.h>
 
#include <stdint.h>
 
#include <stdlib.h>
 

	
 
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 int32_t ErrorCode;
 

	
 
typedef uint32_t ConnectorId;
 

	
 
typedef uint32_t PortSuffix;
 

	
 
typedef struct {
 
  ConnectorId connector_id;
 
  PortSuffix u32_suffix;
 
} Id;
 

	
 
typedef Id PortId;
 

	
 
/**
 
 * 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.
 
 */
 
ErrorCode 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.
 
 */
 
ErrorCode connector_add_net_port(Connector *connector,
 
                                 PortId *port,
 
                                 const uint8_t *addr_str_ptr,
 
                                 uintptr_t addr_str_len,
 
                                 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
 
 */
 
ErrorCode connector_add_udp_port(Connector *connector,
 
                                 PortId (*putter)[2],
 
                                 const uint8_t *local_addr_str_ptr,
 
                                 uintptr_t local_addr_str_len,
 
                                 const uint8_t *peer_addr_str_ptr,
 
                                 uintptr_t peer_addr_str_len);
 

	
 
/**
 
 * Connects this connector to the distributed system of connectors reachable through endpoints,
 
 * Usable in setup state, and changes the state to communication.
 
 */
 
ErrorCode 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);
 

	
 
ErrorCode 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".
 
 */
 
ErrorCode 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 */
 
/* CBindgen generated */
 

	
 
#ifndef REOWOLF_HEADER_DEFINED
 
#define REOWOLF_HEADER_DEFINED
 

	
 
#include <stdarg.h>
 
#include <stdbool.h>
 
#include <stdint.h>
 
#include <stdlib.h>
 

	
 
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 int ErrorCode;
 

	
 
typedef uint32_t ConnectorId;
 

	
 
typedef uint32_t PortSuffix;
 

	
 
typedef struct {
 
  ConnectorId connector_id;
 
  PortSuffix u32_suffix;
 
} Id;
 

	
 
typedef Id PortId;
 

	
 
/**
 
 * 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.
 
 */
 
ErrorCode 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.
 
 */
 
ErrorCode connector_add_net_port(Connector *connector,
 
                                 PortId *port,
 
                                 const uint8_t *addr_str_ptr,
 
                                 uintptr_t addr_str_len,
 
                                 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
 
 */
 
ErrorCode connector_add_udp_port(Connector *connector,
 
                                 PortId *putter,
 
                                 PortId *getter,
 
                                 const uint8_t *local_addr_str_ptr,
 
                                 uintptr_t local_addr_str_len,
 
                                 const uint8_t *peer_addr_str_ptr,
 
                                 uintptr_t peer_addr_str_len);
 

	
 
/**
 
 * Connects this connector to the distributed system of connectors reachable through endpoints,
 
 * Usable in setup state, and changes the state to communication.
 
 */
 
ErrorCode 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);
 

	
 
ErrorCode 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".
 
 */
 
ErrorCode 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 */
src/ffi/mod.rs
Show inline comments
 
file renamed from src/runtime/ffi.rs to src/ffi/mod.rs
 
use super::*;
 

	
 
use core::cell::RefCell;
 
use core::convert::TryFrom;
 
use std::slice::from_raw_parts as slice_from_parts;
 
// use std::os::raw::{c_char, c_int, c_uchar, c_uint};
 

	
 
use crate::{common::*, runtime::*};
 
use core::{cell::RefCell, convert::TryFrom};
 
use std::slice::from_raw_parts as slice_from_raw_parts;
 

	
 
#[cfg(feature = "ffi_socket_api")]
 
pub mod socket_api;
 
///////////////////////////////////////////////
 
type ErrorCode = std::os::raw::c_int;
 
#[derive(Default)]
 
struct StoredError {
 
    // invariant: len is zero IFF its occupied
 
@@ -61,14 +62,11 @@ impl StoredError {
 
thread_local! {
 
    static STORED_ERROR: RefCell<StoredError> = RefCell::new(StoredError::default());
 
}
 

	
 
type ErrorCode = i32;
 

	
 
unsafe fn tl_socketaddr_from_raw(
 
    bytes_ptr: *const u8,
 
    bytes_len: usize,
 
) -> Result<SocketAddr, i32> {
 
    std::str::from_utf8(&*slice_from_parts(bytes_ptr, bytes_len))
 
) -> Result<SocketAddr, ErrorCode> {
 
    std::str::from_utf8(&*slice_from_raw_parts(bytes_ptr, bytes_len))
 
        .map_err(|err| {
 
            StoredError::tl_debug_store(&err);
 
            -1
 
@@ -107,7 +105,7 @@ pub unsafe extern "C" fn protocol_description_parse(
 
    pdl_len: usize,
 
) -> *mut Arc<ProtocolDescription> {
 
    StoredError::tl_clear();
 
    match ProtocolDescription::parse(&*slice_from_parts(pdl, pdl_len)) {
 
    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());
 
@@ -139,7 +137,7 @@ pub unsafe extern "C" fn connector_new_logging(
 
    path_len: usize,
 
) -> *mut Connector {
 
    StoredError::tl_clear();
 
    let path_bytes = &*slice_from_parts(path_ptr, path_len);
 
    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) => {
 
@@ -213,8 +211,8 @@ pub unsafe extern "C" fn connector_add_component(
 
) -> ErrorCode {
 
    StoredError::tl_clear();
 
    match connector.add_component(
 
        &*slice_from_parts(ident_ptr, ident_len),
 
        &*slice_from_parts(ports_ptr, ports_len),
 
        &*slice_from_raw_parts(ident_ptr, ident_len),
 
        &*slice_from_raw_parts(ports_ptr, ports_len),
 
    ) {
 
        Ok(()) => 0,
 
        Err(err) => {
 
@@ -360,7 +358,7 @@ pub unsafe extern "C" fn connector_put_bytes(
 
    bytes_len: usize,
 
) -> ErrorCode {
 
    StoredError::tl_clear();
 
    let bytes = &*slice_from_parts(bytes_ptr, bytes_len);
 
    let bytes = &*slice_from_raw_parts(bytes_ptr, bytes_len);
 
    match connector.put(port, Payload::from(bytes)) {
 
        Ok(()) => 0,
 
        Err(err) => {
 
@@ -450,7 +448,7 @@ pub unsafe extern "C" fn connector_gotten_bytes(
 
//     bytes_len: usize,
 
//     out_payload: *mut Payload,
 
// ) {
 
//     let bytes: &[u8] = &*slice_from_parts(bytes_ptr, bytes_len);
 
//     let bytes: &[u8] = &*slice_from_raw_parts(bytes_ptr, bytes_len);
 
//     out_payload.write(Payload::from(bytes));
 
// }
 

	
src/ffi/socket_api.rs
Show inline comments
 
new file 100644
 
use atomic_refcell::AtomicRefCell;
 
use std::{collections::HashMap, ffi::c_void, net::SocketAddr, os::raw::c_int, sync::RwLock};
 
///////////////////////////////////////////////////////////////////
 

	
 
type ConnectorFd = c_int;
 
struct Connector {}
 
struct FdAllocator {
 
    next: Option<ConnectorFd>,
 
    freed: Vec<ConnectorFd>,
 
}
 
enum MaybeConnector {
 
    New,
 
    Bound(SocketAddr),
 
    Connected(Connector),
 
}
 
#[derive(Default)]
 
struct ConnectorStorage {
 
    fd_to_connector: HashMap<ConnectorFd, AtomicRefCell<MaybeConnector>>,
 
    fd_allocator: FdAllocator,
 
}
 
///////////////////////////////////////////////////////////////////
 
impl Default for FdAllocator {
 
    fn default() -> Self {
 
        Self {
 
            next: Some(0), // positive values used only
 
            freed: vec![],
 
        }
 
    }
 
}
 
impl FdAllocator {
 
    fn alloc(&mut self) -> ConnectorFd {
 
        if let Some(fd) = self.freed.pop() {
 
            return fd;
 
        }
 
        if let Some(fd) = self.next {
 
            self.next = fd.checked_add(1);
 
            return fd;
 
        }
 
        panic!("No more Connector FDs to allocate!")
 
    }
 
    fn free(&mut self, fd: ConnectorFd) {
 
        self.freed.push(fd);
 
    }
 
}
 
lazy_static::lazy_static! {
 
    static ref CONNECTOR_STORAGE: RwLock<ConnectorStorage> = Default::default();
 
}
 
///////////////////////////////////////////////////////////////////
 

	
 
#[no_mangle]
 
pub extern "C" fn socket(_domain: c_int, _type: c_int) -> c_int {
 
    // assuming _domain is AF_INET and _type is SOCK_DGRAM
 
    let mut w = CONNECTOR_STORAGE.write().expect("Fd lock poisoned!");
 
    let fd = w.fd_allocator.alloc();
 
    w.fd_to_connector.insert(fd, AtomicRefCell::new(MaybeConnector::New));
 
    fd
 
}
 

	
 
#[no_mangle]
 
pub extern "C" fn close(fd: ConnectorFd, _how: c_int) -> c_int {
 
    // ignoring HOW
 
    let mut w = CONNECTOR_STORAGE.write().expect("Fd lock poisoned!");
 
    w.fd_allocator.free(fd);
 
    if w.fd_to_connector.remove(&fd).is_some() {
 
        0
 
    } else {
 
        -1
 
    }
 
}
 

	
 
#[no_mangle]
 
pub extern "C" fn bind(fd: ConnectorFd, address: *const SocketAddr, _address_len: usize) -> c_int {
 
    use MaybeConnector as Mc;
 
    // assuming _domain is AF_INET and _type is SOCK_DGRAM
 
    let r = CONNECTOR_STORAGE.read().expect("Fd lock poisoned!");
 
    if let Some(maybe_conn) = r.fd_to_connector.get(&fd) {
 
        let mc: &mut Mc = &mut maybe_conn.borrow_mut();
 
        match mc {
 
            Mc::New => {
 
                *mc = Mc::Bound(address.read());
 
                0
 
            }
 
            _ => -1, // connector in wrong state
 
        }
 
    } else {
 
        // no connector for this fd
 
        return -2;
 
    }
 
}
 

	
 
#[no_mangle]
 
pub extern "C" fn connect(
 
    fd: ConnectorFd,
 
    _address: *const SocketAddr,
 
    _address_len: usize,
 
) -> c_int {
 
    use MaybeConnector as Mc;
 
    // assuming _domain is AF_INET and _type is SOCK_DGRAM
 
    let r = CONNECTOR_STORAGE.read().expect("Fd lock poisoned!");
 
    if let Some(maybe_conn) = r.fd_to_connector.get(&fd) {
 
        let mc: &mut Mc = &mut maybe_conn.borrow_mut();
 
        match mc {
 
            Mc::Bound(_local) => {
 
                *mc = Mc::Connected(Connector {});
 
                0
 
            }
 
            _ => -1, // connector in wrong state
 
        }
 
    } else {
 
        // no connector for this fd
 
        return -2;
 
    }
 
}
 
#[no_mangle]
 
pub extern "C" fn send(fd: ConnectorFd, msg: *const c_void, len: usize, flags: c_int) -> isize {
 
    use MaybeConnector as Mc;
 
    // assuming _domain is AF_INET and _type is SOCK_DGRAM
 
    let r = CONNECTOR_STORAGE.read().expect("Fd lock poisoned!");
 
    if let Some(maybe_conn) = r.fd_to_connector.get(&fd) {
 
        let mc: &mut Mc = &mut maybe_conn.borrow_mut();
 
        match mc {
 
            Mc::Bound(_local) => {
 
                *mc = Mc::Connected(Connector {});
 
                0
 
            }
 
            _ => -1, // connector in wrong state
 
        }
 
    } else {
 
        // no connector for this fd
 
        return -2;
 
    }
 
}
src/lib.rs
Show inline comments
 
@@ -10,4 +10,4 @@ pub use protocol::ProtocolDescription;
 
pub use runtime::{error, Connector, DummyLogger, FileLogger, VecLogger};
 

	
 
#[cfg(feature = "ffi")]
 
pub use runtime::ffi;
 
pub mod ffi;
src/runtime/mod.rs
Show inline comments
 
@@ -8,9 +8,6 @@ mod logging;
 
/// cbindgen:ignore
 
mod setup;
 

	
 
#[cfg(feature = "ffi")]
 
pub mod ffi;
 

	
 
#[cfg(test)]
 
mod tests;
 

	
 
@@ -376,7 +373,7 @@ impl Drop for Connector {
 
    }
 
}
 
impl Connector {
 
    fn random_id() -> ConnectorId {
 
    pub(crate) fn random_id() -> ConnectorId {
 
        type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];
 
        unsafe {
 
            let mut bytes = std::mem::MaybeUninit::<Bytes8>::uninit();
src/runtime/tests.rs
Show inline comments
 
@@ -50,7 +50,12 @@ lazy_static::lazy_static! {
 
        Payload::from(TEST_MSG_BYTES)
 
    };
 
}
 

	
 
fn new_u8_buffer(cap: usize) -> Vec<u8> {
 
    let mut v = Vec::with_capacity(cap);
 
    // Safe! len will cover owned bytes in valid state
 
    unsafe { v.set_len(cap) }
 
    v
 
}
 
//////////////////////////////////////////
 

	
 
#[test]
 
@@ -713,12 +718,7 @@ fn reowolf_to_udp() {
 
            // udp thread
 
            let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
 
            udp.connect(sock_addrs[0]).unwrap();
 
            let mut buf = unsafe {
 
                // canonical way to create uninitalized byte buffer
 
                let mut v = Vec::with_capacity(256);
 
                v.set_len(256);
 
                v
 
            };
 
            let mut buf = new_u8_buffer(256);
 
            let len = udp.recv(&mut buf).unwrap();
 
            assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
 
            barrier.wait();
 
@@ -782,12 +782,7 @@ fn udp_reowolf_swap() {
 
            // udp thread
 
            let udp = std::net::UdpSocket::bind(sock_addrs[1]).unwrap();
 
            udp.connect(sock_addrs[0]).unwrap();
 
            let mut buf = unsafe {
 
                // canonical way to create uninitalized byte buffer
 
                let mut v = Vec::with_capacity(256);
 
                v.set_len(256);
 
                v
 
            };
 
            let mut buf = new_u8_buffer(256);
 
            udp.send(TEST_MSG_BYTES).unwrap();
 
            let len = udp.recv(&mut buf).unwrap();
 
            assert_eq!(TEST_MSG_BYTES, &buf[0..len]);
0 comments (0 inline, 0 general)