Changeset - 1f3007fd929e
[Not reviewed]
0 5 0
Christopher Esterhuyse - 5 years ago 2020-07-23 15:15:40
christopher.esterhuyse@gmail.com
restricted pseudo-socket API to connection-oriented API
5 files changed with 78 insertions and 167 deletions:
0 comments (0 inline, 0 general)
Cargo.toml
Show inline comments
 
@@ -32,17 +32,17 @@ 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]
 
# compile target: dynamically linked library using C ABI
 
crate-type = ["cdylib"]
 

	
 
[features]
 
default = ["ffi", "session_optimization", "ffi_pseudo_socket_api"]
 
default = ["ffi", "session_optimization"]
 
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
 
\ No newline at end of file
src/ffi/mod.rs
Show inline comments
 
@@ -70,25 +70,25 @@ impl StoredError {
 
                }
 
            }
 
        })
 
    }
 
}
 
thread_local! {
 
    static STORED_ERROR: RefCell<StoredError> = RefCell::new(StoredError::default());
 
}
 

	
 
pub const ERR_OK: c_int = 0;
 
pub const ERR_REOWOLF: c_int = -1;
 
pub const WRONG_STATE: c_int = -2;
 
pub const CC_MAP_LOCK_POISONED: c_int = -3;
 
pub const LOCK_POISONED: c_int = -3;
 
pub const CLOSE_FAIL: c_int = -4;
 
pub const BAD_FD: c_int = -5;
 
pub const CONNECT_FAILED: c_int = -6;
 
pub const WOULD_BLOCK: c_int = -7;
 
pub const BAD_SOCKADDR: c_int = -8;
 
pub const SEND_BEFORE_CONNECT: c_int = -9;
 

	
 
///////////////////// 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
src/ffi/pseudo_socket_api.rs
Show inline comments
 
@@ -5,36 +5,34 @@ use std::{
 
    collections::HashMap,
 
    ffi::c_void,
 
    net::{Ipv4Addr, SocketAddr, SocketAddrV4},
 
    os::raw::c_int,
 
    sync::RwLock,
 
};
 
///////////////////////////////////////////////////////////////////
 

	
 
struct FdAllocator {
 
    next: Option<c_int>,
 
    freed: Vec<c_int>,
 
}
 
struct ConnectorBound {
 
    connector: Connector,
 
    putter: PortId,
 
    getter: PortId,
 
}
 
struct ConnectorComplex {
 
    // invariants:
 
    // 1. connector is a upd-socket singleton
 
    // 2. putter and getter are ports in the native interface with the appropriate polarities
 
    // 3. connected_to always mirrors connector's single udp socket's connect addr. both are overwritten together.
 
    connected_to: Option<SocketAddr>,
 
    connector_bound: Option<ConnectorBound>,
 
enum ConnectorComplex {
 
    Setup {
 
        local: Option<SocketAddr>,
 
        peer: Option<SocketAddr>,
 
    },
 
    Communication {
 
        connector: Connector,
 
        putter: PortId,
 
        getter: PortId,
 
    },
 
}
 
#[derive(Default)]
 
struct CcMap {
 
    fd_to_cc: HashMap<c_int, RwLock<ConnectorComplex>>,
 
    fd_allocator: FdAllocator,
 
}
 
///////////////////////////////////////////////////////////////////
 
unsafe fn payload_from_raw(bytes_ptr: *const c_void, bytes_len: usize) -> Payload {
 
    let bytes_ptr = std::mem::transmute(bytes_ptr);
 
    let bytes = &*slice_from_raw_parts(bytes_ptr, bytes_len);
 
    Payload::from(bytes)
 
}
 
@@ -60,230 +58,147 @@ impl FdAllocator {
 
        }
 
        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: c_int) {
 
        self.freed.push(fd);
 
    }
 
}
 
lazy_static::lazy_static! {
 
    static ref CC_MAP: RwLock<CcMap> = Default::default();
 
    static ref LOCK_POISONED: RwLock<CcMap> = Default::default();
 
}
 
impl ConnectorComplex {
 
    fn try_become_connected(&mut self) {
 
        match self {
 
            ConnectorComplex::Setup { Some(local), Some(peer) } => {
 
                // setup complete
 
                let connector = Connector::new(crate::DummyLogger, TRIVIAL_PD.clone(), Connector::random_id());
 
                let [putter, getter] = connector.new_udp_mediator_component(local, peer).unwrap();
 
                *self = ConnectorComplex::Communication { connector, putter, getter }
 
            }
 
            _ => {} // setup incomplete
 
        }
 
    }
 
}
 
///////////////////////////////////////////////////////////////////
 

	
 
fn connected_complex(local: SocketAddr, peer: SocketAddr) -> ConnectorComplex {
 
}
 
/////////////////////////////////
 
#[no_mangle]
 
pub extern "C" fn rw_socket(_domain: c_int, _type: c_int) -> c_int {
 
    // ignoring domain and type
 
    // get writer lock
 
    let mut w = if let Ok(w) = CC_MAP.write() { w } else { return CC_MAP_LOCK_POISONED };
 
    let mut w = if let Ok(w) = LOCK_POISONED.write() { w } else { return CCMLP };
 
    let fd = w.fd_allocator.alloc();
 
    let cc = ConnectorComplex { connected_to: None, connector_bound: None };
 
    let cc = ConnectorComplex::Setup { local: None, peer: None };
 
    w.fd_to_cc.insert(fd, RwLock::new(cc));
 
    fd
 
}
 

	
 
#[no_mangle]
 
pub extern "C" fn rw_close(fd: c_int, _how: c_int) -> c_int {
 
    // ignoring HOW
 
    // get writer lock
 
    let mut w = if let Ok(w) = CC_MAP.write() { w } else { return CC_MAP_LOCK_POISONED };
 
    let mut w = if let Ok(w) = LOCK_POISONED.write() { w } else { return CCMLP };
 
    if w.fd_to_cc.remove(&fd).is_some() {
 
        w.fd_allocator.free(fd);
 
        ERR_OK
 
    } else {
 
        CLOSE_FAIL
 
    }
 
}
 

	
 
#[no_mangle]
 
pub unsafe extern "C" fn rw_bind(fd: c_int, addr: *const sockaddr, addr_len: socklen_t) -> c_int {
 
    // assuming _domain is AF_INET and _type is SOCK_DGRAM
 
    let addr = match addr_from_raw(addr, addr_len) {
 
        Some(addr) => addr,
 
        _ => return BAD_SOCKADDR,
 
    };
 
    // get outer reader, inner writer locks
 
    let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED };
 
    let r = if let Ok(r) = LOCK_POISONED.read() { r } else { return CCMLP };
 
    let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CCMLP };
 
    let cc: &mut ConnectorComplex = &mut cc;
 
    if cc.connector_bound.is_some() {
 
        // already bound!
 
        return WRONG_STATE;
 
    match cc {
 
        ConnectorComplex::Communication { ..} => WRONG_STATE,
 
        ConnectorComplex::Setup { local, peer } => {
 
            ConnectorComplex::Setup { local, .. } => {
 
            *local = Some(addr);
 
            cc.try_become_connected();
 
            ERR_OK
 
        }
 
    }
 
    cc.connector_bound = {
 
        let mut connector = Connector::new(
 
            Box::new(crate::DummyLogger),
 
            crate::TRIVIAL_PD.clone(),
 
            Connector::random_id(),
 
        );
 
        // maintain invariant: if cc.connected_to.is_some():
 
        //   cc.connected_to matches the connected address of the socket
 
        let peer_addr = cc.connected_to.unwrap_or_else(dummy_peer_addr);
 
        let [putter, getter] = connector.new_udp_mediator_component(addr, peer_addr).unwrap();
 
        Some(ConnectorBound { connector, putter, getter })
 
    };
 
    ERR_OK
 
}
 

	
 
#[no_mangle]
 
pub unsafe extern "C" fn rw_connect(
 
    fd: c_int,
 
    addr: *const sockaddr,
 
    addr_len: socklen_t,
 
) -> c_int {
 
    let addr = match addr_from_raw(addr, addr_len) {
 
        Some(addr) => addr,
 
        _ => return BAD_SOCKADDR,
 
    };
 
    // assuming _domain is AF_INET and _type is SOCK_DGRAM
 
    // get outer reader, inner writer locks
 
    let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED };
 
    let r = if let Ok(r) = LOCK_POISONED.read() { r } else { return CCMLP };
 
    let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CCMLP };
 
    let cc: &mut ConnectorComplex = &mut cc;
 
    if let Some(ConnectorBound { connector, .. }) = &mut cc.connector_bound {
 
        // already bound. maintain invariant by overwriting the socket's connection (DUMMY or otherwise)
 
        if connector.get_mut_udp_ee(0).unwrap().sock.connect(addr).is_err() {
 
            return CONNECT_FAILED;
 
    match cc {
 
        ConnectorComplex::Communication { .. } => WRONG_STATE,
 
        ConnectorComplex::Setup { peer, .. } => {
 
            *peer = Some(addr);
 
            cc.try_become_connected();
 
            ERR_OK
 
        }
 
    }
 
    cc.connected_to = Some(addr);
 
    ERR_OK
 
}
 

	
 
#[no_mangle]
 
pub unsafe extern "C" fn rw_send(
 
    fd: c_int,
 
    bytes_ptr: *const c_void,
 
    bytes_len: usize,
 
    _flags: c_int,
 
) -> isize {
 
    // ignoring flags
 
    // get outer reader, inner writer locks
 
    let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
 
    let r = if let Ok(r) = LOCK_POISONED.read() { r } else { return CCMLP as isize };
 
    let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CCMLP as isize };
 
    let cc: &mut ConnectorComplex = &mut cc;
 
    if cc.connected_to.is_none() {
 
        return SEND_BEFORE_CONNECT as isize;
 
    }
 
    if let Some(ConnectorBound { connector, putter, .. }) = &mut cc.connector_bound {
 
        // is bound
 
        connector.put(*putter, payload_from_raw(bytes_ptr, bytes_len)).unwrap();
 
        connector.sync(None).unwrap();
 
        bytes_len as isize
 
    } else {
 
        // is not bound
 
        WRONG_STATE as isize
 
    }
 
}
 
#[no_mangle]
 
pub unsafe extern "C" fn rw_sendto(
 
    fd: c_int,
 
    bytes_ptr: *mut c_void,
 
    bytes_len: usize,
 
    _flags: c_int,
 
    addr: *const sockaddr,
 
    addr_len: socklen_t,
 
) -> isize {
 
    // ignoring flags
 
    let addr = match addr_from_raw(addr, addr_len) {
 
        Some(addr) => addr,
 
        _ => return BAD_SOCKADDR as isize,
 
    };
 
    // get outer reader, inner writer locks
 
    let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
 
    let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
 
    let cc: &mut ConnectorComplex = &mut cc;
 
    if let Some(ConnectorBound { connector, putter, .. }) = &mut cc.connector_bound {
 
        // is bound
 
        // (temporarily) break invariant
 
        if connector.get_mut_udp_ee(0).unwrap().sock.connect(addr).is_err() {
 
            // invariant not broken. nevermind
 
            return CONNECT_FAILED as isize;
 
    match cc {
 
        ConnectorComplex::Setup { .. } => WRONG_STATE as isize,
 
        ConnectorComplex::Communication { connector, putter, .. } => {
 
            let payload = payload_from_raw(bytes_ptr, bytes_len);
 
            connector.put(*putter, payload).unwrap();
 
            connector.sync(None).unwrap();
 
            bytes_len
 
        }
 
        // invariant broken...
 
        connector.put(*putter, payload_from_raw(bytes_ptr, bytes_len)).unwrap();
 
        connector.sync(None).unwrap();
 
        let old_addr = cc.connected_to.unwrap_or_else(dummy_peer_addr);
 
        connector.get_mut_udp_ee(0).unwrap().sock.connect(old_addr).unwrap();
 
        // ...invariant restored
 
        bytes_len as isize
 
    } else {
 
        // is not bound
 
        WRONG_STATE as isize
 
    }
 
}
 

	
 
#[no_mangle]
 
pub unsafe extern "C" fn rw_recv(
 
    fd: c_int,
 
    bytes_ptr: *mut c_void,
 
    bytes_len: usize,
 
    _flags: c_int,
 
) -> isize {
 
    // ignoring flags
 
    // get outer reader, inner writer locks
 
    let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
 
    let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
 
    let cc: &mut ConnectorComplex = &mut cc;
 
    if let Some(ConnectorBound { connector, getter, .. }) = &mut cc.connector_bound {
 
        connector.get(*getter).unwrap();
 
        // this call BLOCKS until it succeeds, and its got no reason to fail
 
        connector.sync(None).unwrap();
 
        // copy from gotten to caller's buffer (truncating if necessary)
 
        let slice = connector.gotten(*getter).unwrap().as_slice();
 
        let cpy_msg_bytes = slice.len().min(bytes_len);
 
        std::ptr::copy_nonoverlapping(slice.as_ptr(), bytes_ptr as *mut u8, cpy_msg_bytes);
 
        // return number of bytes sent
 
        cpy_msg_bytes as isize
 
    } else {
 
        WRONG_STATE as isize // not bound!
 
    }
 
}
 

	
 
#[no_mangle]
 
pub unsafe extern "C" fn rw_recvfrom(
 
    fd: c_int,
 
    bytes_ptr: *mut c_void,
 
    bytes_len: usize,
 
    _flags: c_int,
 
    addr: *mut sockaddr,
 
    addr_len: *mut socklen_t,
 
) -> isize {
 
    // ignoring flags
 
    // get outer reader, inner writer locks
 
    let r = if let Ok(r) = CC_MAP.read() { r } else { return CC_MAP_LOCK_POISONED as isize };
 
    let r = if let Ok(r) = LOCK_POISONED.read() { r } else { return CCMLP as isize };
 
    let cc = if let Some(cc) = r.fd_to_cc.get(&fd) { cc } else { return BAD_FD as isize };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CC_MAP_LOCK_POISONED as isize };
 
    let mut cc = if let Ok(cc) = cc.write() { cc } else { return CCMLP as isize };
 
    let cc: &mut ConnectorComplex = &mut cc;
 
    if let Some(ConnectorBound { connector, getter, .. }) = &mut cc.connector_bound {
 
        connector.get(*getter).unwrap();
 
        // this call BLOCKS until it succeeds, and its got no reason to fail
 
        connector.sync(None).unwrap();
 
        // overwrite addr and addr_len
 
        let recvd_from = connector.get_mut_udp_ee(0).unwrap().received_from_this_round.unwrap();
 
        let os_addr = os_socketaddr::OsSocketAddr::from(recvd_from);
 
        let cpy_addr_bytes = (*addr_len).min(os_addr.capacity());
 
        // ptr-return addr bytes (truncated to addr_len)
 
        let src_u8: *const u8 = std::mem::transmute(os_addr.as_ptr());
 
        let dest_u8: *mut u8 = std::mem::transmute(addr);
 
        std::ptr::copy_nonoverlapping(src_u8, dest_u8, cpy_addr_bytes as usize);
 
        // ptr-return true addr size
 
        *addr_len = os_addr.capacity(); 
 
        // copy from gotten to caller's buffer (truncating if necessary)
 
        let slice = connector.gotten(*getter).unwrap().as_slice();
 
        let cpy_msg_bytes = slice.len().min(bytes_len);
 
        let dest_u8: *mut u8 = std::mem::transmute(bytes_ptr);
 
        std::ptr::copy_nonoverlapping(slice.as_ptr(), dest_u8, cpy_msg_bytes);
 
        // return number of bytes received
 
        cpy_msg_bytes as isize
 
    } else {
 
        WRONG_STATE as isize // not bound!
 
    match cc {
 
        ConnectorComplex::Setup { .. } => WRONG_STATE as isize,
 
        ConnectorComplex::Communication { connector, getter, .. } => {
 
            connector.get(*getter).unwrap();
 
            connector.sync(None).unwrap();
 
            let slice = connector.gotten(*getter).unwrap().as_slice();
 
            let cpy_msg_bytes = slice.len().min(bytes_len);
 
            std::ptr::copy_nonoverlapping(slice.as_ptr(), bytes_ptr as *mut u8, cpy_msg_bytes);
 
            cpy_msg_bytes
 
        }
 
    }
 
}
src/runtime/communication.rs
Show inline comments
 
@@ -104,28 +104,24 @@ impl RoundCtxTrait for RoundCtx {
 
    fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg) {
 
        self.getter_buffer.getter_add(getter, msg)
 
    }
 
}
 
impl Connector {
 
    fn get_comm_mut(&mut self) -> Option<&mut ConnectorCommunication> {
 
        if let ConnectorPhased::Communication(comm) = &mut self.phased {
 
            Some(comm)
 
        } else {
 
            None
 
        }
 
    }
 
    // #[cfg(ffi_socket_api)]
 
    pub(crate) fn get_mut_udp_ee(&mut self, index: usize) -> Option<&mut UdpEndpointExt> {
 
        self.get_comm_mut()?.endpoint_manager.udp_endpoint_store.endpoint_exts.get_mut(index)
 
    }
 
    pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {
 
        use GottenError as Ge;
 
        let comm = self.get_comm_mut().ok_or(Ge::NoPreviousRound)?;
 
        match &comm.round_result {
 
            Err(_) => Err(Ge::PreviousSyncFailed),
 
            Ok(None) => Err(Ge::NoPreviousRound),
 
            Ok(Some(round_ok)) => round_ok.gotten.get(&port).ok_or(Ge::PortDidntGet),
 
        }
 
    }
 
    pub fn next_batch(&mut self) -> Result<usize, WrongStateError> {
 
        // returns index of new batch
 
        let comm = self.get_comm_mut().ok_or(WrongStateError)?;
src/runtime/mod.rs
Show inline comments
 
@@ -31,31 +31,24 @@ pub struct DummyLogger;
 
pub struct FileLogger(ConnectorId, std::fs::File);
 
pub(crate) struct NonsyncProtoContext<'a> {
 
    cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds
 
    proto_component_ports: &'a mut HashSet<PortId>, // sub-structure of component
 
    unrun_components: &'a mut Vec<(ProtoComponentId, ProtoComponent)>, // lives for Nonsync phase
 
    proto_component_id: ProtoComponentId,     // KEY in id->component map
 
}
 
pub(crate) struct SyncProtoContext<'a> {
 
    cu_inner: &'a mut ConnectorUnphasedInner, // persists between rounds
 
    branch_inner: &'a mut ProtoComponentBranchInner, // sub-structure of component branch
 
    predicate: &'a Predicate,                 // KEY in pred->branch map
 
}
 
#[derive(Debug)]
 
pub(crate) struct UdpEndpointExt {
 
    pub(crate) sock: UdpSocket, // already bound and connected
 
    pub(crate) received_from_this_round: Option<SocketAddr>,
 
    outgoing_payloads: HashMap<Predicate, Payload>,
 
    getter_for_incoming: PortId,
 
}
 
#[derive(Default, Debug, Clone)]
 
struct ProtoComponentBranchInner {
 
    untaken_choice: Option<u16>,
 
    did_put_or_get: HashSet<PortId>,
 
    inbox: HashMap<PortId, Payload>,
 
}
 
#[derive(
 
    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 
)]
 
struct SpecVar(PortId);
 
#[derive(
 
    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,
 
@@ -167,24 +160,31 @@ struct NetEndpointSetup {
 
#[derive(Debug, Clone)]
 
struct UdpEndpointSetup {
 
    getter_for_incoming: PortId,
 
    local_addr: SocketAddr,
 
    peer_addr: SocketAddr,
 
}
 
#[derive(Debug)]
 
struct NetEndpointExt {
 
    net_endpoint: NetEndpoint,
 
    getter_for_incoming: PortId,
 
}
 
#[derive(Debug)]
 
struct UdpEndpointExt {
 
    sock: UdpSocket, // already bound and connected
 
    received_from_this_round: Option<SocketAddr>,
 
    outgoing_payloads: HashMap<Predicate, Payload>,
 
    getter_for_incoming: PortId,
 
}
 
#[derive(Debug)]
 
struct Neighborhood {
 
    parent: Option<usize>,
 
    children: VecSet<usize>,
 
}
 
#[derive(Debug)]
 
struct IdManager {
 
    connector_id: ConnectorId,
 
    port_suffix_stream: U32Stream,
 
    proto_component_suffix_stream: U32Stream,
 
}
 
#[derive(Debug)]
 
struct UdpInBuffer {
0 comments (0 inline, 0 general)