[dev-dependencies]

# test-generator = "0.3.0"

crossbeam-utils = "0.7.2"

lazy_static = "1.4.0"

[lib]

crate-type = [

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

	"cdylib" # for FFI use, typically C.

]

[features]

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

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

#include "../utility.c"

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

	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);

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

	rw_err_peek(c);

	PortId putter, getter;

	FfiSocketAddr local_addr = {{0, 0, 0, 0}, 8000};

	FfiSocketAddr peer_addr =  {{8, 8, 8, 1}, 8001};

	rw_err_peek(c);

	connector_add_udp_mediator_component(c, &putter, &getter, local_addr, peer_addr);

	connector_connect(c, -1);

	rw_err_peek(c);

	// Prepare a message to send

	size_t msg_len = 16;

	char * msg_ptr = malloc(msg_len);

	memset(msg_ptr, 42, msg_len);

	int i;

	for(i=0; i<10; i++) {

		connector_put_bytes(c, putter, msg_ptr, msg_len);

		rw_err_peek(c);

		connector_sync(c, -1);

		rw_err_peek(c);

	}

	printf("Exiting\n");

	protocol_description_destroy(pd);

	connector_destroy(c);

	free(msg_ptr);

	sleep(1.0);

	return 0;

}

script_path = os.path.dirname(os.path.realpath(__file__));

for c_file in glob.glob(script_path + "/*/*.c", recursive=False):

  print("compiling", c_file)

  args = [

    "gcc",          # compiler

    "-std=c11",     # C11 mode

    "-Wl,-R./",     # pass -R flag to linker: produce relocatable object

    c_file,         # input source file

    "-o",           # output flag

    c_file[:-2],    # output filename

    "-L",           # lib path flag

    "./",           # where to look for libs

    "-lreowolf_rs"  # add lib called "reowolf_rs"

  ];

typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;

typedef struct Connector Connector;

typedef uint32_t ConnectorId;

typedef uint32_t U32Suffix;

typedef struct {

  ConnectorId connector_id;

  U32Suffix u32_suffix;

typedef struct {

  uint8_t ipv4[4];

  uint16_t port;

} FfiSocketAddr;

/**

 * 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.

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

 */

int connector_put_bytes(Connector *connector,

                        PortId port,

                        const uint8_t *bytes_ptr,

                        uintptr_t bytes_len);

    pub(crate) connector_id: ConnectorId,

    pub(crate) u32_suffix: U32Suffix,

}

#[derive(Clone, Debug, Default)]

pub struct U32Stream {

    next: u32,

}

#[derive(

    Copy, Clone, Eq, PartialEq, Ord, Hash, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(transparent)]

pub struct PortId(Id);

pub struct Payload(Arc<Vec<u8>>);

#[derive(

    Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[repr(C)]

pub enum Polarity {

    Putter, // output port (from the perspective of the component)

    Getter, // input port (from the perspective of the component)

}

#[derive(

    Debug, Eq, PartialEq, Clone, Hash, Copy, Ord, PartialOrd, serde::Serialize, serde::Deserialize,

)]

#[derive(Debug, Clone)]

pub(crate) enum SyncBlocker {

    Inconsistent,

    SyncBlockEnd,

    CouldntReadMsg(PortId),

    CouldntCheckFiring(PortId),

    PutMsg(PortId, Payload),

    NondetChoice { n: u16 },

}

pub(crate) struct DenseDebugHex<'a>(pub &'a [u8]);

///////////////////// IMPL /////////////////////

impl IdParts for Id {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

        (self.connector_id, self.u32_suffix)

    }

}

impl IdParts for PortId {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

        self.0.id_parts()

    }

}

impl IdParts for ProtoComponentId {

    fn id_parts(self) -> (ConnectorId, U32Suffix) {

/// Given an initialized protocol description, initializes `out` with a clone which can be independently created or destroyed.

#[no_mangle]

pub unsafe extern "C" fn protocol_description_clone(

    pd: &Arc<ProtocolDescription>,

) -> *mut Arc<ProtocolDescription> {

    Box::into_raw(Box::new(pd.clone()))

}

///////////////////// CONNECTOR //////////////////////////

#[no_mangle]

    pd: &Arc<ProtocolDescription>,

    path_ptr: *const u8,

    path_len: usize,

) -> *mut Connector {

    StoredError::tl_clear();

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

            StoredError::tl_debug_store(&err);

            return std::ptr::null_mut();

        }

    };

    match std::fs::File::create(path_str) {

        Ok(file) => {

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

            let c = Connector::new(file_logger, pd.clone(), connector_id);

            Box::into_raw(Box::new(c))

        }

        Err(err) => {

            StoredError::tl_debug_store(&err);

            std::ptr::null_mut()

        }

    }

}

#[no_mangle]

pub unsafe extern "C" fn connector_print_debug(connector: &mut Connector) {

    println!("Debug print dump {:#?}", connector);

}

/// Initializes `out` with a new connector using the given protocol description as its configuration.

/// The connector uses the given (internal) connector ID.

#[no_mangle]

pub unsafe extern "C" fn connector_new(pd: &Arc<ProtocolDescription>) -> *mut Connector {

    let c = Connector::new(Box::new(DummyLogger), pd.clone(), Connector::random_id());

    Box::into_raw(Box::new(c))

/*

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

*/

macro_rules! log {

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

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

        }

    }};

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

    }};

}

    }

    // 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.inner.logger,

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

            &timeout,

            comm.round_index

        );

        // 1. run all proto components to Nonsync blockers

        // NOTE: original components are immutable until Decision::Success

        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.inner.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!(

            );

            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.inner.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

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

                log!(

                    cu.inner.logger,

                    "Children in subtree are: {:?}",

                    subtree_id_iter.clone().collect::<Vec<_>>()

                );

                SolutionStorage::new(subtree_id_iter)

            },

            spec_var_stream: cu.inner.id_manager.new_spec_var_stream(),

            getter_buffer: Default::default(),

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

        };

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

        log!(

            cu.inner.logger,

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

            comm.native_batches.len()

        );

        let native_spec_var = rctx.spec_var_stream.next();

        log!(cu.inner.logger, "Native branch spec var is {:?}", native_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())

                    SubtreeId::LocalComponent(ComponentId::Native),

                    predicate.clone(),

                );

            }

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

                // thanks to the native_spec_var, each batch has a distinct predicate

                unreachable!()

            }

        }

        // restore the invariant: !native_batches.is_empty()

        comm.native_batches.push(Default::default());

        // Call to another big method; keep running this round until a distributed decision is reached

        let decision = Self::sync_reach_decision(

            cu,

            comm,

            &mut branching_native,

            &mut branching_proto_components,

            &mut rctx,

        )?;

        comm.endpoint_manager.udp_endpoints_round_end(&mut *cu.inner.logger, &decision)?;

        // propagate the decision to children

        let msg = Msg::CommMsg(CommMsg {

            round_index: comm.round_index,

            contents: CommMsgContents::CommCtrl(CommCtrlMsg::Announce {

                decision: decision.clone(),

            }),

        });

        log!(

            cu.inner.logger,

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

                        .into_iter()

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

                );

                log!(

                    cu.inner.logger,

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

                    cu.proto_components.keys()

                );

                // consume native

                Ok(Some(branching_native.collapse_with(&mut *cu.inner.logger, &predicate)))

            }

        };

        ret

    }

    fn sync_reach_decision(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        branching_native: &mut BranchingNative,

        branching_proto_components: &mut HashMap<ProtoComponentId, BranchingProtoComponent>,

        rctx: &mut RoundCtx,

    ) -> Result<Decision, UnrecoverableSyncError> {

        let mut already_requested_failure = false;

        if branching_native.branches.is_empty() {