[lib]

crate-type = [

	"rlib", # for use as a Rust dependency. 

	"cdylib" # for FFI use, typically C.

]

[features]

default = ["ffi"]

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

# Reowolf 1.0 Implementation

This library provides connectors as a generalization of sockets for use in communication over the internet. This repository is one of the deliverables of the [Reowolf project](https://nlnet.nl/project/Reowolf/) funded by the NLNet foundation.

## Compilation instructions

1. Install the latest stable Rust toolchain (`rustup install stable`) using the [rustup](https://rustup.rs/) CLI tool, available for most platforms.

1. Have `cargo` (the Rust language package manager) download source dependencies, and compile the library with release-level optimizations with `cargo build --release`: 

	- The resulting dylib can be found in `./target/release/`, to be used with the header file: `./reowolf.h`.

	- *Note*: A list of immediate ancestor dependencies is visible in `./Cargo.toml`.

	- *Note*: Run `cargo test --release` to run unit tests with release-level optimizations.

## Using the library

- The library may be used as a Rust dependency by adding it as a git dependency, i.e., by adding line `reowolf_rs = { git = "https://scm.cwi.nl/FM/reowolf" }` to downstream crate's manifest file, `Cargo.toml`.

- The library may be used as a dynamically-linked library using its C ABI as the cdylib written to `./target/release` when compiled with release optimizations, in combination to the header file `./reowolf.h`.

- When compiled on Linux, the compiled library will include definitions of pseudo-socket procedures declared in `./pseudo-socket.h` when compiled with `cargo build --release --features ffi_pseudo_socket_api`. The added functionality is only needed when requiring that connectors expose a socket-like API.

## Examples

The `examples` directory contains example usages of connectors for message passing over the internet. The programs within require that the library is compiled as a dylib (see above).

## Notes

3. Running `cbindgen > reowolf.h` from the root will overwrite the header file. This is only necessary to update it.  

/*

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

*/

macro_rules! log {

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

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

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

        // }

    }};

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

    }};

}

        let was = *self;

        *self = true;

        !was

    }

}

// CuUndecided provides a mostly immutable view into the ConnectorUnphased structure,

// making it harder to accidentally mutate its contents in a way that cannot be rolled back.

impl CuUndecided for ConnectorUnphased {

    fn logger_and_protocol_description(&mut self) -> (&mut dyn Logger, &ProtocolDescription) {

        (&mut *self.logger, &self.proto_description)

    }

    fn logger(&mut self) -> &mut dyn Logger {

        &mut *self.logger

    }

    fn proto_description(&self) -> &ProtocolDescription {

        &self.proto_description

    }

    fn native_component_id(&self) -> ComponentId {

        self.native_component_id

    }

}

impl<'a, K, V> MapTempsGuard<'a, K, V> {

    fn reborrow(&mut self) -> MapTempsGuard<'_, K, V> {

                proto_component_id,

                unrun_components.len()

            );

            let (logger, proto_description) = cu.logger_and_protocol_description();

            let mut ctx = NonsyncProtoContext {

                ips: &mut ips,

                logger,

                proto_component_id,

                unrun_components: &mut unrun_components,

            };

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

            log!(

                "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 => assert!(branching_proto_components

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

                    .is_none()), // Some(_) returned IFF some component identifier key is overwritten (BAD!)

            }

        for &child in comm.neighborhood.children.iter() {

            comm.endpoint_manager.send_to_comms(child, &msg)?;

        }

        let ret = match decision {

            Decision::Failure => {

                // untouched port/component fields of `cu` are NOT overwritten.

                // the result is a rollback.

                Err(Se::RoundFailure)

            }

            Decision::Success(predicate) => {

                // commit changes to component states

                cu.proto_components.clear();

                    // "flatten" branching components, committing the speculation

                    // consistent with the predicate decided upon.

                    branching_proto_components

                        .into_iter()

                );

                // commit changes to ports and id_manager

                log!(

                    "End round with (updated) component states {:?}",

                );

                // consume native

                Ok(Some(round_ok))

            }

        };

        log!(cu.logger(), "Sync round ending! Cleaning up");

        ret

    }

    // Once the synchronous round has been started, this procedure

    // routs and handles payloads, receives control messages from neighboring connectors,

    // checks for timeout, and aggregates solutions until a distributed decision is reached.

    // The decision is either a solution (success case), or a distributed timeout rollback (failure case)

    // The final possible outcome is an unrecoverable error, which results from some fundamental misbehavior,

                logger,

                "Considering native branch {:?} with to_get {:?} gotten {:?}",

                &branch_predicate,

                &branch.to_get,

                &branch.gotten

            );

            if branch.is_ended() && branch_predicate.assigns_subset(solution_predicate) {

                let NativeBranch { index, gotten, .. } = branch;

                log!(logger, "Collapsed native has gotten {:?}", &gotten);

                return RoundEndedNative { batch_index: index, gotten };

            }

        }

        panic!("Native had no branches matching pred {:?}", solution_predicate);

    }

}

impl BranchingProtoComponent {

    // Create a singleton-branch branching protocol component as

    // speculation begins, with the given protocol state.

    fn initial(state: ComponentState) -> Self {

        let branch = ProtoComponentBranch { state, inner: Default::default(), ended: false };

        Self { branches: hashmap! { Predicate::default() => branch } }

    }

    // run all the given branches (cd.input) to their SyncBlocker,

                // Rather than choosing either branch, we MERGE them.

                // This means keeping the existing one in-place, and giving it the UNION of the inboxes

                let old = eo.get_mut();

                for (k, v) in branch.inner.inbox.drain() {

                    old.inner.inbox.insert(k, v);

                }

            }

        }

    }

    // Given the predicate for the round's solution, collapse this

    // branching native to an ended branch whose predicate is consistent with it.

        let BranchingProtoComponent { branches } = self;

        for (branch_predicate, branch) in branches {

            if branch.ended && branch_predicate.assigns_subset(solution_predicate) {

                let ProtoComponentBranch { state, .. } = branch;

                return state;

            }

        }

        panic!("ProtoComponent had no branches matching pred {:?}", solution_predicate);

    }

}

impl ProtoComponentBranch {

    // Feed this branch received message.

    // It's safe to receive the same message repeatedly,

    // but if we receive a message with different contents,

    // it's a sign something has gone wrong! keys of type (port, round, predicate)

    // should always map to at most one message value!

    fn feed_msg(&mut self, getter: PortId, payload: Payload) {

        let e = self.inner.inbox.entry(getter);

        use std::collections::hash_map::Entry;

}

// A trait intended to limit the access of the ConnectorUnphased structure

// such that we don't accidentally modify any important component/port data

// while the results of the round are undecided. Why? Any actions during Connector::sync

// are _speculative_ until the round is decided, and we need a safe way of rolling

// back any changes.

trait CuUndecided {

    fn logger(&mut self) -> &mut dyn Logger;

    fn proto_description(&self) -> &ProtocolDescription;

    fn native_component_id(&self) -> ComponentId;

    fn logger_and_protocol_description(&mut self) -> (&mut dyn Logger, &ProtocolDescription);

}

// Represents a set of synchronous port operations that the native component

// has described as an "option" for completing during the synchronous rounds.

// Operations contained here succeed together or not at all.

// A native with N=2+ batches are expressing an N-way nondeterministic choice

#[derive(Debug, Default)]

struct NativeBatch {

    // invariant: putters' and getters' polarities respected

    to_put: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

    let [p0, g0] = c.new_port_pair();

    c.add_component(b"for_msg_byte", &[p0]).unwrap();

    c.connect(None).unwrap();

    for expecting in 0u8..8 {

        c.get(g0).unwrap();

        c.sync(None).unwrap();

        assert_eq!(&[expecting], c.gotten(g0).unwrap().as_slice());

    }

    c.sync(None).unwrap();

}

#[test]

fn eq_no_causality() {

    let test_log_path = Path::new("./logs/eq_no_causality");

    let pdl = b"

    composite eq(in a, in b, out c) {

        channel leftfirsto -> leftfirsti;

        new eqinner(a, b, c, leftfirsto, leftfirsti);

    }

    primitive eqinner(in a, in b, out c, out leftfirsto, in leftfirsti) {

        msg ma = null;

        msg mb = null;

        while(true) synchronous {

                if(fires(leftfirsti)) {

                    // left first! DO USE DUMMY

                    ma = get(a);

                    put(c, ma);

                    mb = get(b);

                    // using dummy!

                    put(leftfirsto, ma);

                    get(leftfirsti);

                } else {

                    // right first! DON'T USE DUMMY

                    mb = get(b);

                    put(c, mb);

                    ma = get(a);

                }

                assert(ma == mb);

            }

        }

    ";

    let pd = reowolf::ProtocolDescription::parse(pdl).unwrap();

    let mut c = file_logged_configured_connector(0, test_log_path, Arc::new(pd));

    /*

    [native]p0-->g0[eq]p1--.

                 g1        |

                 ^---------`

    */

    let [p0, g0] = c.new_port_pair();

    let [p1, g1] = c.new_port_pair();

    c.add_component(b"eq", &[g0, g1, p1]).unwrap();

    /*

                  V--------.

                 g2        |

    [native]p2-->g3[eq]p3--`

    */

    let [p2, g2] = c.new_port_pair();

    let [p3, g3] = c.new_port_pair();

    c.add_component(b"eq", &[g3, g2, p3]).unwrap();

    c.connect(None).unwrap();

    for _ in 0..32 {

        c.put(p0, TEST_MSG.clone()).unwrap();

        c.put(p2, TEST_MSG.clone()).unwrap();

        c.sync(SEC1).unwrap();

    }

}