[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"] }

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

lazy_static = "1.4.0"

[lib]

# compile target: dynamically linked library using C ABI

crate-type = ["cdylib"]

[features]

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

#include <stdio.h>

#include <stdlib.h>

#include <errno.h>

// allocates a buffer!

char * buffer_pdl(char * filename) {

	FILE *f = fopen(filename, "rb");

	if (f == NULL) {

		printf("Opening pdl file returned errno %d!\n", errno);

		exit(1);

	}

	fseek(f, 0, SEEK_END);

	long fsize = ftell(f);

	fseek(f, 0, SEEK_SET);

	char *pdl = malloc(fsize + 1);

	fread(pdl, 1, fsize, f);

	fclose(f);

	pdl[fsize] = 0;

	return pdl;

}

use super::*;

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]

    // assuming _domain is AF_INET and _type is SOCK_DGRAM

    let fd = w.fd_allocator.alloc();

    w.fd_to_connector.insert(fd, AtomicRefCell::new(MaybeConnector::New));

    fd

}

#[no_mangle]

    // ignoring HOW

    w.fd_allocator.free(fd);

    if w.fd_to_connector.remove(&fd).is_some() {

        ERR_OK

    } else {

        CLOSE_FAIL

    }

}

#[no_mangle]

    use MaybeConnector as Mc;

    // assuming _domain is AF_INET and _type is SOCK_DGRAM

    let mc = if let Some(mc) = r.fd_to_connector.get(&fd) { mc } else { return BAD_FD };

    let _ = if let Mc::New = mc { () } else { return WRONG_STATE };

    *mc = Mc::Bound(address.read());

    ERR_OK

}

#[no_mangle]

    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 mc = if let Some(mc) = r.fd_to_connector.get(&fd) { mc } else { return BAD_FD };

    *mc = Mc::Connected(Connector {});

    ERR_OK

}

#[no_mangle]

    use MaybeConnector as Mc;

    // assuming _domain is AF_INET and _type is SOCK_DGRAM

    // TODO

}

use super::*;

use crate::common::*;

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

#[derive(Default)]

struct GetterBuffer {

    getters_and_sends: Vec<(PortId, SendPayloadMsg)>,

}

struct RoundCtx {

    solution_storage: SolutionStorage,

    spec_var_stream: SpecVarStream,

    getter_buffer: GetterBuffer,

    deadline: Option<Instant>,

}

struct BranchingNative {

    branches: HashMap<Predicate, NativeBranch>,

}

#[derive(Clone, Debug)]

struct NativeBranch {

    index: usize,

    gotten: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>, // native branch is ended iff to_get.is_empty()

}

#[derive(Debug)]

struct SolutionStorage {

    old_local: HashSet<Predicate>,

    new_local: HashSet<Predicate>,

    // this pair acts as SubtreeId -> HashSet<Predicate> which is friendlier to iteration

    subtree_solutions: Vec<HashSet<Predicate>>,

    subtree_id_to_index: HashMap<SubtreeId, usize>,

}

#[derive(Debug)]

struct BranchingProtoComponent {

    ports: HashSet<PortId>,

    branches: HashMap<Predicate, ProtoComponentBranch>,

}

#[derive(Debug, Clone)]

struct ProtoComponentBranch {

    inbox: HashMap<PortId, Payload>,

    state: ComponentState,

    untaken_choice: Option<u16>,

    ended: bool,

}

struct CyclicDrainer<'a, K: Eq + Hash, V> {

    input: &'a mut HashMap<K, V>,

    inner: CyclicDrainInner<'a, K, V>,

}

struct CyclicDrainInner<'a, K: Eq + Hash, V> {

    swap: &'a mut HashMap<K, V>,

    output: &'a mut HashMap<K, V>,

}

trait ReplaceBoolTrue {

    fn replace_with_true(&mut self) -> bool;

}

impl ReplaceBoolTrue for bool {

    fn replace_with_true(&mut self) -> bool {

        let was = *self;

        *self = true;

        !was

    }

}

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

impl RoundCtxTrait for RoundCtx {

    fn get_deadline(&self) -> &Option<Instant> {

        &self.deadline

    }

    fn getter_add(&mut self, getter: PortId, msg: SendPayloadMsg) {

        self.getter_buffer.getter_add(getter, msg)

    }

}

impl Connector {

    pub fn gotten(&mut self, port: PortId) -> Result<&Payload, GottenError> {

        use GottenError as Ge;

        }

    }

    pub fn next_batch(&mut self) -> Result<usize, WrongStateError> {

        // returns index of new batch

    }

    fn port_op_access(

        &mut self,

        port: PortId,

        expect_polarity: Polarity,

    ) -> Result<&mut NativeBatch, PortOpError> {

        use PortOpError as Poe;

        let Self { unphased, phased } = self;

        if !unphased.native_ports.contains(&port) {

            return Err(Poe::PortUnavailable);

        }

        match unphased.port_info.polarities.get(&port) {

            Some(p) if *p == expect_polarity => {}

            Some(_) => return Err(Poe::WrongPolarity),

            None => return Err(Poe::UnknownPolarity),

        }

        match phased {

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

            ConnectorPhased::Communication(comm) => {

                let batch = comm.native_batches.last_mut().unwrap(); // length >= 1 is invariant

                Ok(batch)

            }

        }

    }

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

        use PortOpError as Poe;

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

        if batch.to_put.contains_key(&port) {

            Err(Poe::MultipleOpsOnPort)

        } 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.logger, "Attempted to start sync round, but previous error {:?} was unrecoverable!", e);

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

                    }

                    _ => {}

                }

                comm.round_result = Self::connected_sync(cu, comm, timeout);

                comm.round_index += 1;

                match &comm.round_result {

                    Ok(None) => unreachable!(),

                    Ok(Some(ok_result)) => Ok(ok_result.batch_index),

                    Err(sync_error) => Err(sync_error.clone()),

                }

            }

        }

    }

    // private function. mutates state but returns with round

    // result ASAP (allows for convenient error return with ?)

    fn connected_sync(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        timeout: Option<Duration>,

    ) -> Result<Option<RoundOk>, SyncError> {

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

        use SyncError as Se;

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

        log!(

            cu.logger,

            "~~~ SYNC called with timeout {:?}; starting round {}",

            &timeout,

            comm.round_index

        );

        // 1. run all proto components to Nonsync blockers

        let mut branching_proto_components =

            HashMap::<ProtoComponentId, BranchingProtoComponent>::default();

        let mut unrun_components: Vec<(ProtoComponentId, ProtoComponent)> =

            cu.proto_components.iter().map(|(&k, v)| (k, v.clone())).collect();

        log!(cu.logger, "Nonsync running {} proto components...", unrun_components.len());

        // drains unrun_components, and populates branching_proto_components.

        while let Some((proto_component_id, mut component)) = unrun_components.pop() {

            // TODO coalesce fields

            log!(

                cu.logger,

                "Nonsync running proto component with ID {:?}. {} to go after this",

                proto_component_id,

                unrun_components.len()

            );

            let mut ctx = NonsyncProtoContext {

                logger: &mut *cu.logger,

                port_info: &mut cu.port_info,

                id_manager: &mut cu.id_manager,

                proto_component_id,

                unrun_components: &mut unrun_components,

                proto_component_ports: &mut cu

                    .proto_components

                    .get_mut(&proto_component_id)

                    .unwrap() // unrun_components' keys originate from proto_components

                    .ports,

            };

            let blocker = component.state.nonsync_run(&mut ctx, &cu.proto_description);

            log!(

                cu.logger,

                "proto component {:?} ran to nonsync blocker {:?}",

                proto_component_id,

                &blocker

            );

            use NonsyncBlocker as B;

            match blocker {

                B::ComponentExit => drop(component),

                B::Inconsistent => return Err(Se::InconsistentProtoComponent(proto_component_id)),

                B::SyncBlockStart => {

                    branching_proto_components

                        .insert(proto_component_id, BranchingProtoComponent::initial(component));

                }

            }

        }

        log!(

            cu.logger,

            "All {} proto components are now done with Nonsync phase",

            branching_proto_components.len(),

        );

        // Create temp structures needed for the synchronous phase of the round

        let mut rctx = RoundCtx {

            solution_storage: {

                let n = std::iter::once(SubtreeId::LocalComponent(ComponentId::Native));

                let c = cu

                    .proto_components

                    .keys()

                    .map(|&id| SubtreeId::LocalComponent(ComponentId::Proto(id)));

                let e = comm

                    .neighborhood

                    .children

                    .iter()

                    .map(|&index| SubtreeId::NetEndpoint { index });

                let subtree_id_iter = n.chain(c).chain(e);

                SolutionStorage::new(subtree_id_iter)

            },

            spec_var_stream: cu.id_manager.new_spec_var_stream(),

            getter_buffer: Default::default(),

            deadline: timeout.map(|to| Instant::now() + to),

        };

        log!(cu.logger, "Round context structure initialized");

        // Explore all native branches eagerly. Find solutions, buffer messages, etc.

        log!(

            cu.logger,

            "Translating {} native batches into branches...",

            comm.native_batches.len()

        );

        let native_branch_spec_var = rctx.spec_var_stream.next();

        log!(cu.logger, "Native branch spec var is {:?}", native_branch_spec_var);

        let mut branching_native = BranchingNative { branches: Default::default() };

        'native_branches: for ((native_branch, index), branch_spec_val) in

            comm.native_batches.drain(..).zip(0..).zip(SpecVal::iter_domain())

        {

            let NativeBatch { to_get, to_put } = native_branch;

            let predicate = {

                let mut predicate = Predicate::default();

                // assign trues for ports that fire

                let firing_ports: HashSet<PortId> =

                    to_get.iter().chain(to_put.keys()).copied().collect();

                for &port in to_get.iter().chain(to_put.keys()) {

                    let var = cu.port_info.spec_var_for(port);

                    predicate.assigned.insert(var, SpecVal::FIRING);

                }

                // assign falses for all silent (not firing) ports

                for &port in cu.native_ports.difference(&firing_ports) {

                    let var = cu.port_info.spec_var_for(port);

                    if let Some(SpecVal::FIRING) = predicate.assigned.insert(var, SpecVal::SILENT) {

                        log!(cu.logger, "Native branch index={} contains internal inconsistency wrt. {:?}. Skipping", index, var);

                        continue 'native_branches;

                    }

                }

                // this branch is consistent. distinguish it with a unique var:val mapping and proceed

                predicate.inserted(native_branch_spec_var, branch_spec_val)

            };

            log!(cu.logger, "Native branch index={:?} has consistent {:?}", index, &predicate);

            // send all outgoing messages (by buffering them)

            for (putter, payload) in to_put {

                let msg = SendPayloadMsg { predicate: predicate.clone(), payload };

                proto_description,

                proto_components: Default::default(),

                logger,

                id_manager: IdManager::new(connector_id),

                native_ports: Default::default(),

                port_info: Default::default(),

            },

            phased: ConnectorPhased::Setup(Box::new(ConnectorSetup {

                net_endpoint_setups: Default::default(),

                udp_endpoint_setups: Default::default(),

                surplus_sockets,

            })),

        }

    }

    pub fn new_udp_port(

        &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 mut npid = || cu.id_manager.new_port_id();

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

                cu.native_ports.insert(nin);

                cu.native_ports.insert(nout);

                cu.port_info.polarities.insert(nin, Getter);

                cu.port_info.polarities.insert(nout, Putter);

                cu.port_info.polarities.insert(uin, Getter);

                cu.port_info.polarities.insert(uout, Putter);

                cu.port_info.peers.insert(nin, uout);

                cu.port_info.peers.insert(nout, uin);

                cu.port_info.peers.insert(uin, nout);

                cu.port_info.peers.insert(uout, nin);

                cu.port_info.routes.insert(nin, Route::LocalComponent(ComponentId::Native));

                cu.port_info.routes.insert(nout, Route::LocalComponent(ComponentId::Native));

                cu.port_info.routes.insert(uin, Route::UdpEndpoint { index: udp_index });

                cu.port_info.routes.insert(uout, Route::UdpEndpoint { index: udp_index });

                setup.udp_endpoint_setups.push(UdpEndpointSetup {

                    local_addr,

                    peer_addr,

                    getter_for_incoming: nin,

                });

                Ok([nout, nin])

            }

        }

    }

    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) => {

                let local_port = cu.id_manager.new_port_id();

                cu.native_ports.insert(local_port);

                // {polarity, route} known. {peer} unknown.

                cu.port_info.polarities.insert(local_port, polarity);

                cu.port_info.routes.insert(local_port, Route::LocalComponent(ComponentId::Native));

                log!(

                    cu.logger,

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

                    local_port,

                    polarity,

                    &sock_addr,

                    endpoint_polarity

                );

                setup.net_endpoint_setups.push(NetEndpointSetup {

                    sock_addr,

                    endpoint_polarity,

                    getter_for_incoming: local_port,

                });

                Ok(local_port)

            }

        }

    }

    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) => {

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

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

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

                let mut endpoint_manager = new_endpoint_manager(

                    &mut *cu.logger,

                    &setup.net_endpoint_setups,

                    &setup.udp_endpoint_setups,

                    &mut cu.port_info,

                    &deadline,

                )?;

                log!(

                    cu.logger,

                    "Successfully connected {} endpoints",

                    endpoint_manager.net_endpoint_store.endpoint_exts.len()

                );

                // leader election and tree construction

                let neighborhood = init_neighborhood(

                    cu.id_manager.connector_id,

                    &mut *cu.logger,

                    &mut endpoint_manager,

                    &deadline,

                )?;

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

                let mut comm = ConnectorCommunication {

                    round_index: 0,

                    endpoint_manager,

                    neighborhood,

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

                    round_result: Ok(None),

                };

                if cfg!(feature = "session_optimization") {

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

                }

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

                self.phased = ConnectorPhased::Communication(Box::new(comm));

                Ok(())

            }

        }

    }

}

fn new_endpoint_manager(

    logger: &mut dyn Logger,

    net_endpoint_setups: &[NetEndpointSetup],

    udp_endpoint_setups: &[UdpEndpointSetup],

    port_info: &mut PortInfo,

    deadline: &Option<Instant>,

) -> Result<EndpointManager, ConnectError> {

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

    use std::sync::atomic::{AtomicBool, Ordering::SeqCst};

    use ConnectError as Ce;

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

    const WAKER_PERIOD: Duration = Duration::from_millis(300);

    struct WakerState {

        continue_signal: AtomicBool,

        waker: mio::Waker,

    }

    impl WakerState {

        fn waker_loop(&self) {

            while self.continue_signal.load(SeqCst) {

                std::thread::sleep(WAKER_PERIOD);

                let _ = self.waker.wake();

            }

        }

        fn waker_stop(&self) {

            self.continue_signal.store(false, SeqCst);

            // TODO keep waker registered?

        }

    }

    struct Todo {

        // 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: TodoEndpoint,

        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

    }

    struct UdpTodo {

        // becomes completed once we receive our first writable event

        getter_for_incoming: PortId,

        sock: UdpSocket,

    }

    enum TodoEndpoint {

        Accepting(TcpListener),

        NetEndpoint(NetEndpoint),

    }

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

    // 1. Start to construct EndpointManager

    let mut waker_state: Option<Arc<WakerState>> = None;

    let mut poll = Poll::new().map_err(|_| Ce::PollInitFailed)?;

    let mut events =

        Events::with_capacity((net_endpoint_setups.len() + udp_endpoint_setups.len()) * 2 + 4);

    let [mut net_polled_undrained, udp_polled_undrained] = [VecSet::default(), VecSet::default()];

    let mut delayed_messages = vec![];

    // 2. Create net/udp TODOs, each already registered with poll

    let mut net_todos = net_endpoint_setups

        .iter()

        .enumerate()

        .map(|(index, endpoint_setup)| {

            let token = TokenTarget::NetEndpoint { index }.into();

            let todo_endpoint = if let EndpointPolarity::Active = endpoint_setup.endpoint_polarity {

                let mut stream = TcpStream::connect(endpoint_setup.sock_addr)

                    .expect("mio::TcpStream connect should not fail!");

                poll.registry().register(&mut stream, token, BOTH).unwrap();

                TodoEndpoint::NetEndpoint(NetEndpoint { stream, inbox: vec![] })

            } else {

                let mut listener = TcpListener::bind(endpoint_setup.sock_addr)

                    .map_err(|_| Ce::BindFailed(endpoint_setup.sock_addr))?;

                poll.registry().register(&mut listener, token, BOTH).unwrap();

                TodoEndpoint::Accepting(listener)

            };

            Ok(Todo {

                todo_endpoint,

                sent_local_port: false,

                recv_peer_port: None,

                endpoint_setup: endpoint_setup.clone(),

            })

        })

        .collect::<Result<Vec<Todo>, ConnectError>>()?;

    let udp_todos = udp_endpoint_setups

        .iter()

        .enumerate()

        .map(|(index, endpoint_setup)| {

            let mut sock = UdpSocket::bind(endpoint_setup.local_addr)

                .map_err(|_| Ce::BindFailed(endpoint_setup.local_addr))?;

            sock.connect(endpoint_setup.peer_addr)

                .map_err(|_| Ce::UdpConnectFailed(endpoint_setup.peer_addr))?;

            poll.registry()

                .register(&mut sock, TokenTarget::UdpEndpoint { index }.into(), Interest::WRITABLE)

                .unwrap();

            Ok(UdpTodo { sock, getter_for_incoming: endpoint_setup.getter_for_incoming })

        })

        .collect::<Result<Vec<UdpTodo>, ConnectError>>()?;

    // Initially, (1) no net connections have failed, and (2) all udp and net endpoint setups are incomplete

    let mut net_connect_retry_later: HashSet<usize> = Default::default();

    let mut setup_incomplete: HashSet<TokenTarget> = {

        let net_todo_targets_iter =

            (0..net_todos.len()).map(|index| TokenTarget::NetEndpoint { index });

        let udp_todo_targets_iter =

            (0..udp_todos.len()).map(|index| TokenTarget::UdpEndpoint { index });

        net_todo_targets_iter.chain(udp_todo_targets_iter).collect()

    };

    // progress by reacting to poll events. continue until every endpoint is set up

    while !setup_incomplete.is_empty() {

        let remaining = if let Some(deadline) = deadline {

            Some(deadline.checked_duration_since(Instant::now()).ok_or(Ce::Timeout)?)

        } else {

            None

        };

        poll.poll(&mut events, remaining).map_err(|_| Ce::PollFailed)?;

        for event in events.iter() {

            let token = event.token();