target

/.idea

**/*.rs.bk

Cargo.lock

main

examples/*/*.exe

examples/*.dll

examples/reowolf*

logs

        serde_proto_description: SerdeProtocolDescription(

            (A big honkin' protocol description),

        ),

        port_info: PortInfo {

            polarities: {

            },

            peers: {

            },

            routes: {

                    Native,

                ),

                    Native,

                ),

            },

        },

        proto_components: {},

    },

}

#include <stdio.h>

#include <string.h>

#include "../../reowolf.h"

#include "../utility.c"

int main(int argc, char** argv) {

	char * pdl_ptr = buffer_pdl("eg_protocols.pdl");

	size_t pdl_len = strlen(pdl_ptr);

	Arc_ProtocolDescription * pd = protocol_description_parse(pdl_ptr, pdl_len);

	char logpath[] = "./6_amy_log.txt";

	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);

	printf("Err %s\n", reowolf_error_peek(NULL));

	PortId putter, getter;

	connector_add_port_pair(c, &putter, &getter);

	connector_connect(c, -1);

	connector_print_debug(c);

	printf("Let's try to put without get\n");

	connector_put_bytes(c, putter, "hello", 5);

	// connector_get(c, getter);

	int err = connector_sync(c, 5000);

	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));

	printf("Let's try again, doing both\n");

	connector_put_bytes(c, putter, "hello", 5);

	connector_get(c, getter);

	err = connector_sync(c, 5000);

	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));

	size_t msg_len;

	const char * msg_ptr = connector_gotten_bytes(c, getter, &msg_len);

	printf("Got msg `%.*s`\n", msg_len, msg_ptr);

	protocol_description_destroy(pd);

	connector_destroy(c);

	free(pdl_ptr);

	return 0;

}

#[macro_use]

mod macros;

mod common;

mod protocol;

mod runtime;

pub use protocol::ProtocolDescription;

pub use runtime::{error, Connector, DummyLogger, FileLogger, VecLogger};

                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(),

        );

        // NOTE: all msgs in outbox are of form (Getter, Payload)

        let mut payloads_to_get: Vec<(PortId, SendPayloadMsg)> = vec![];

        // create the solution storage

        let mut solution_storage = {

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

            let c =

                cu.proto_components.keys().map(|&id| Route::LocalComponent(ComponentId::Proto(id)));

            let e = comm.neighborhood.children.iter().map(|&index| Route::Endpoint { index });

            SolutionStorage::new(n.chain(c).chain(e))

        };

        log!(cu.logger, "Solution storage initialized");

        // 2. kick off the native

        log!(

            cu.logger,

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

            comm.native_batches.len()

        );

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

        'native_branches: for (index, NativeBatch { to_get, to_put }) in

            comm.native_batches.drain(..).enumerate()

        {

            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.firing_var_for(port);

                    predicate.assigned.insert(var, true);

                }

                // assign falses for silent ports

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

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

                    if let Some(true) = predicate.assigned.insert(var, false) {

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

                        continue 'native_branches;

                    }

                }

                predicate

            };

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

            // put all messages

            for (putter, payload) in to_put {

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

                log!(cu.logger, "Native branch {} sending msg {:?}", index, &msg);

                payloads_to_get.putter_send(cu, putter, msg)?;

            }

            if to_get.is_empty() {

                log!(

                    cu.logger,

                    "Native submitting solution for batch {} with {:?}",

                    index,

                    &predicate

                );

                solution_storage.submit_and_digest_subtree_solution(

                    &mut *cu.logger,

                    Route::LocalComponent(ComponentId::Native),

                    predicate.clone(),

                );

            }

            let branch = NativeBranch { index, gotten: Default::default(), to_get };

            if let Some(existing) = branching_native.branches.insert(predicate, branch) {

                return Err(Se::IndistinguishableBatches([index, existing.index]));

            }

        }

        let decision = Self::sync_reach_decision(

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

            solution_storage,

            payloads_to_get,

            deadline,

        )?;

        log!(cu.logger, "Committing to decision {:?}!", &decision);

        // propagate the decision to children

        let msg = Msg::CommMsg(CommMsg {

            round_index: comm.round_index,

            contents: CommMsgContents::Announce { decision: decision.clone() },

        });

        log!(

            cu.logger,

            "Announcing decision {:?} through child endpoints {:?}",

            &msg,

            &comm.neighborhood.children

        );

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

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

        }

        let ret = match decision {

            Decision::Failure => {

                // dropping {branching_proto_components, branching_native}

                Err(Se::RoundFailure)

            }

            Decision::Success(predicate) => {

                // commit changes to component states

                cu.proto_components.clear();

                cu.proto_components.extend(

                    // consume branching proto components

                    branching_proto_components

                        .into_iter()

                        .map(|(id, bpc)| (id, bpc.collapse_with(&predicate))),

                );

                log!(

                    cu.logger,

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

                    cu.proto_components.keys()

                );

                // consume native

                Ok(Some(branching_native.collapse_with(&predicate)))

            }

        };

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

        // dropping {solution_storage, payloads_to_get}

        ret

    }

    fn sync_reach_decision(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        branching_native: &mut BranchingNative,

        branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,

        mut solution_storage: SolutionStorage,

        mut payloads_to_get: Vec<(PortId, SendPayloadMsg)>,

        mut deadline: Option<Instant>,

    ) -> Result<Decision, SyncError> {

        let mut already_requested_failure = false;

        if branching_native.branches.is_empty() {

            log!(cu.logger, "Native starts with no branches! Failure!");

            match comm.neighborhood.parent {

                Some(parent) => {

                    if already_requested_failure.replace_with_true() {

                        Self::request_failure(cu, comm, parent)?

                    } else {

                        log!(cu.logger, "Already requested failure");

                    }

                }

                None => {

                    log!(cu.logger, "No parent. Deciding on failure");

                    return Ok(Decision::Failure);

                }

            }

        }

        log!(cu.logger, "Done translating native batches into branches");

        // run all proto components to their sync blocker

        log!(

            cu.logger,

            "Running all {} proto components to their sync blocker...",

            branching_proto_components.len()

        );

        for (&proto_component_id, proto_component) in branching_proto_components.iter_mut() {

            let BranchingProtoComponent { ports, branches } = proto_component;

            let mut swap = HashMap::default();

            let mut blocked = HashMap::default();

            // drain from branches --> blocked

            let cd = CyclicDrainer::new(branches, &mut swap, &mut blocked);

            BranchingProtoComponent::drain_branches_to_blocked(

                cd,

                cu,

                &mut solution_storage,

                &mut payloads_to_get,

                proto_component_id,

                ports,

            )?;

            // swap the blocked branches back

            std::mem::swap(&mut blocked, branches);

            if branches.is_empty() {

                log!(cu.logger, "{:?} has become inconsistent!", proto_component_id);

                if let Some(parent) = comm.neighborhood.parent {

                    if already_requested_failure.replace_with_true() {

                        Self::request_failure(cu, comm, parent)?

                    } else {

                        log!(cu.logger, "Already requested failure");

                    }

                } else {

                    log!(cu.logger, "As the leader, deciding on timeout");

                    return Ok(Decision::Failure);

                }

            }

        }

        log!(cu.logger, "All proto components are blocked");

        log!(cu.logger, "Entering decision loop...");

        comm.endpoint_manager.undelay_all();

        'undecided: loop {

            // drain payloads_to_get, sending them through endpoints / feeding them to components

            while let Some((getter, send_payload_msg)) = payloads_to_get.pop() {

                assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));

                match cu.port_info.routes.get(&getter) {

                    None => {

                        log!(

                            cu.logger,

                            "Delivery to getter {:?} msg {:?} failed. Physical route unmapped!",

                            getter,

                            &send_payload_msg

                        );

                    }

                    Some(Route::Endpoint { index }) => {

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::SendPayload(send_payload_msg),

                        });

                        comm.endpoint_manager.send_to_comms(*index, &msg)?;

                    }

                    Some(Route::LocalComponent(ComponentId::Native)) => branching_native.feed_msg(

                        cu,

                        &mut solution_storage,

                        // &mut Pay

                        getter,

                        &send_payload_msg,

                    ),

                    Some(Route::LocalComponent(ComponentId::Proto(proto_component_id))) => {

                        if let Some(branching_component) =

                            branching_proto_components.get_mut(proto_component_id)

                        {

                            let proto_component_id = *proto_component_id;

                            // let ConnectorUnphased { port_info, proto_description, .. } = cu;

                            branching_component.feed_msg(

                                cu,

                                &mut solution_storage,

                                proto_component_id,

                                &mut payloads_to_get,

                                getter,

                                &send_payload_msg,

                            )?;

                            if branching_component.branches.is_empty() {

                                log!(

                                    cu.logger,

                                    "{:?} has become inconsistent!",

                                    proto_component_id

                                );

                                if let Some(parent) = comm.neighborhood.parent {

                                    if already_requested_failure.replace_with_true() {

                                        Self::request_failure(cu, comm, parent)?

                                    } else {

                                        log!(cu.logger, "Already requested failure");

                                    }

                                } else {

                                    log!(cu.logger, "As the leader, deciding on timeout");

mod communication;

mod endpoints;

pub mod error;

mod logging;

mod setup;

#[cfg(test)]

mod tests;

use crate::common::*;

use error::*;

#[derive(Debug)]

pub struct RoundOk {

    batch_index: usize,

    gotten: HashMap<PortId, Payload>,

}

pub struct VecSet<T: std::cmp::Ord> {

    // invariant: ordered, deduplicated

    vec: Vec<T>,

}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

pub enum ComponentId {

    Native,

    Proto(ProtoComponentId),

}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash, serde::Serialize, serde::Deserialize)]

pub enum Route {

    LocalComponent(ComponentId),

    Endpoint { index: usize },

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub struct MyPortInfo {

    polarity: Polarity,

    port: PortId,

}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

pub enum Decision {

    Failure,

    Success(Predicate),

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub enum Msg {

    SetupMsg(SetupMsg),

    CommMsg(CommMsg),

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub enum SetupMsg {

    MyPortInfo(MyPortInfo),

    LeaderWave { wave_leader: ConnectorId },

    LeaderAnnounce { tree_leader: ConnectorId },

    YouAreMyParent,

    SessionGather { unoptimized_map: HashMap<ConnectorId, SessionInfo> },

    SessionScatter { optimized_map: HashMap<ConnectorId, SessionInfo> },

}

#[derive(Debug, Clone)]

pub(crate) struct SerdeProtocolDescription(Arc<ProtocolDescription>);

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub struct SessionInfo {

    serde_proto_description: SerdeProtocolDescription,

    port_info: PortInfo,

    proto_components: HashMap<ProtoComponentId, ProtoComponent>,

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub struct CommMsg {

    pub round_index: usize,

    pub contents: CommMsgContents,

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub enum CommMsgContents {

    SendPayload(SendPayloadMsg),

    Suggest { suggestion: Decision }, // SINKWARD

    Announce { decision: Decision },  // SINKAWAYS

}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]

pub struct SendPayloadMsg {

    predicate: Predicate,

    payload: Payload,

}

#[derive(Debug, PartialEq)]

pub enum CommonSatResult {

    FormerNotLatter,

    LatterNotFormer,

    Equivalent,

    New(Predicate),

    Nonexistant,

}

pub struct Endpoint {

    inbox: Vec<u8>,

    stream: TcpStream,

}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

pub struct ProtoComponent {

    state: ComponentState,

    ports: HashSet<PortId>,

}

pub trait Logger: Debug {

    fn line_writer(&mut self) -> &mut dyn std::io::Write;

}

#[derive(Debug)]

pub struct VecLogger(ConnectorId, Vec<u8>);

use crate as reowolf;

use crossbeam_utils::thread::scope;

use reowolf::{

    error::*,

    EndpointPolarity::{Active, Passive},

    Polarity::{Getter, Putter},

    *,

};

use std::{fs::File, net::SocketAddr, path::Path, sync::Arc, time::Duration};

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

fn next_test_addr() -> SocketAddr {

    use std::{

        net::{Ipv4Addr, SocketAddrV4},

        sync::atomic::{AtomicU16, Ordering::SeqCst},

    };

    static TEST_PORT: AtomicU16 = AtomicU16::new(5_000);

    let port = TEST_PORT.fetch_add(1, SeqCst);

    SocketAddrV4::new(Ipv4Addr::LOCALHOST, port).into()

}

fn file_logged_connector(connector_id: ConnectorId, dir_path: &Path) -> Connector {

    let _ = std::fs::create_dir(dir_path); // we will check failure soon

    let path = dir_path.join(format!("cid_{:?}.txt", connector_id));

    let file = File::create(path).unwrap();

    let file_logger = Box::new(FileLogger::new(connector_id, file));

    Connector::new(file_logger, MINIMAL_PROTO.clone(), connector_id, 8)

}

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

#[test]

fn basic_connector() {

    Connector::new(Box::new(DummyLogger), MINIMAL_PROTO.clone(), 0, 0);

}

#[test]

fn basic_logged_connector() {

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

    file_logged_connector(0, test_log_path);

}

#[test]

fn new_port_pair() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, _] = c.new_port_pair();

    let [_, _] = c.new_port_pair();

}

#[test]

fn new_sync() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [o, i] = c.new_port_pair();

    c.add_component(b"sync", &[i, o]).unwrap();

}

#[test]

fn new_net_port() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let sock_addr = next_test_addr();

    let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();

    let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();

}

#[test]

fn trivial_connect() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn single_node_connect() {

    let sock_addr = next_test_addr();

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

    let mut c = file_logged_connector(0, test_log_path);

    let _ = c.new_net_port(Getter, sock_addr, Passive).unwrap();

    let _ = c.new_net_port(Putter, sock_addr, Active).unwrap();

    c.connect(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn minimal_net_connect() {

    let sock_addr = next_test_addr();

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

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            let _ = c.new_net_port(Getter, sock_addr, Active).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            let _ = c.new_net_port(Putter, sock_addr, Passive).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

        });

    })

    .unwrap();

}

#[test]

fn put_no_sync() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [o, _] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn wrong_polarity_bad() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn dup_put_bad() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [o, _] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn trivial_sync() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.sync(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn unconnected_gotten_err() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());

}

#[test]

fn connected_gotten_err_no_round() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    assert_eq!(reowolf::error::GottenError::NoPreviousRound, c.gotten(i).unwrap_err());

}

#[test]

fn connected_gotten_err_ungotten() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.sync(Some(Duration::from_secs(1))).unwrap();

    assert_eq!(reowolf::error::GottenError::PortDidntGet, c.gotten(i).unwrap_err());

}

#[test]

fn native_polarity_checks() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [o, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    // fail...

    c.get(o).unwrap_err();

    // succeed..

    c.get(i).unwrap();

}

#[test]

fn native_multiple_gets() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.get(i).unwrap();

    c.get(i).unwrap_err();

}

#[test]

fn next_batch() {

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

    let mut c = file_logged_connector(0, test_log_path);

    c.next_batch().unwrap_err();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.next_batch().unwrap();

    c.next_batch().unwrap();

    c.next_batch().unwrap();

}

#[test]

fn native_self_msg() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [o, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.get(i).unwrap();

    c.sync(Some(Duration::from_secs(1))).unwrap();

}

#[test]

fn two_natives_msg() {

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

    let sock_addr = next_test_addr();

    scope(|s| {

        s.spawn(|_| {

            let mut c = file_logged_connector(0, test_log_path);

            let g = c.new_net_port(Getter, sock_addr, Active).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

            c.get(g).unwrap();

            c.sync(Some(Duration::from_secs(1))).unwrap();

            c.gotten(g).unwrap();

        });

        s.spawn(|_| {

            let mut c = file_logged_connector(1, test_log_path);

            let p = c.new_net_port(Putter, sock_addr, Passive).unwrap();

            c.connect(Some(Duration::from_secs(1))).unwrap();

            c.sync(Some(Duration::from_secs(1))).unwrap();

        });

    })

    .unwrap();

}

#[test]

fn trivial_nondet() {

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

    let mut c = file_logged_connector(0, test_log_path);

    let [_, i] = c.new_port_pair();

    c.connect(Some(Duration::from_secs(1))).unwrap();

    c.get(i).unwrap();

    // getting 0 batch

    c.next_batch().unwrap();

    // silent 1 batch

    assert_eq!(1, c.sync(Some(Duration::from_secs(1))).unwrap());

    c.gotten(i).unwrap_err();

}

#[test]