#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));

	protocol_description_destroy(pd);

	connector_destroy(c);

	free(pdl_ptr);

	return 0;

}

/* CBindgen generated */

#ifndef REOWOLF_HEADER_DEFINED

#define REOWOLF_HEADER_DEFINED

#include <stdarg.h>

#include <stdbool.h>

#include <stdint.h>

#include <stdlib.h>

typedef enum {

  Active,

  Passive,

} EndpointPolarity;

typedef enum {

  Putter,

  Getter,

} Polarity;

typedef struct Arc_ProtocolDescription Arc_ProtocolDescription;

typedef struct Connector Connector;

typedef int32_t ErrorCode;

typedef uint32_t ConnectorId;

typedef uint32_t PortSuffix;

typedef struct {

  ConnectorId connector_id;

  PortSuffix u32_suffix;

} Id;

typedef Id PortId;

/**

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

 * Usable in {setup, communication} states.

 */

ErrorCode connector_add_component(Connector *connector,

                                  const uint8_t *ident_ptr,

                                  uintptr_t ident_len,

                                  const PortId *ports_ptr,

                                  uintptr_t ports_len);

/**

 * Given

 * - an initialized connector in setup or connecting state,

 * - a utf-8 encoded socket address,

 * - the logical polarity of P,

 * - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,

 * returns P, a port newly added to the native interface.

 */

ErrorCode connector_add_net_port(Connector *connector,

                                 const uint8_t *addr_str_ptr,

                                 uintptr_t addr_str_len,

                                 Polarity port_polarity,

/**

 * Given an initialized connector in setup or connecting state,

 * - Creates a new directed port pair with logical channel putter->getter,

 * - adds the ports to the native component's interface,

 * - and returns them using the given out pointers.

 * Usable in {setup, communication} states.

 */

void connector_add_port_pair(Connector *connector, PortId *out_putter, PortId *out_getter);

/**

 * Connects this connector to the distributed system of connectors reachable through endpoints,

 * Usable in setup state, and changes the state to communication.

 */

ErrorCode connector_connect(Connector *connector, int64_t timeout_millis);

/**

 * Destroys the given a pointer to the connector on the heap, freeing its resources.

 * Usable in {setup, communication} states.

 */

void connector_destroy(Connector *connector);

ErrorCode connector_get(Connector *connector, PortId port);

const uint8_t *connector_gotten_bytes(Connector *connector, PortId port, uintptr_t *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".

 */

ErrorCode connector_put_bytes(Connector *connector,

                              PortId port,

                              const uint8_t *bytes_ptr,

                              uintptr_t bytes_len);

intptr_t connector_sync(Connector *connector, int64_t timeout_millis);

/**

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

 */

Arc_ProtocolDescription *protocol_description_clone(const Arc_ProtocolDescription *pd);

/**

 * Destroys the given initialized protocol description and frees its resources.

 */

void protocol_description_destroy(Arc_ProtocolDescription *pd);

/**

 * Parses the utf8-encoded string slice to initialize a new protocol description object.

 * - On success, initializes `out` and returns 0

 * - On failure, stores an error string (see `reowolf_error_peek`) and returns -1

 */

Arc_ProtocolDescription *protocol_description_parse(const uint8_t *pdl, uintptr_t pdl_len);

/**

 * 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

 * - len does NOT include the length of the null-delimiter

 * If len is NULL, it will not written to.

 */

const uint8_t *reowolf_error_peek(uintptr_t *len);

#endif /* REOWOLF_HEADER_DEFINED */

    old_local: HashSet<Predicate>,

    new_local: HashSet<Predicate>,

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

    subtree_solutions: Vec<HashSet<Predicate>>,

    subtree_id_to_index: HashMap<Route, usize>,

}

#[derive(Debug)]

struct BranchingProtoComponent {

    ports: HashSet<PortId>,

    branches: HashMap<Predicate, ProtoComponentBranch>,

}

#[derive(Debug, Clone)]

struct ProtoComponentBranch {

    inbox: HashMap<PortId, Payload>,

    state: ComponentState,

}

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 PayloadMsgSender {

    fn putter_send(

        &mut self,

        cu: &mut ConnectorUnphased,

        putter: PortId,

        msg: SendPayloadMsg,

    ) -> Result<(), SyncError>;

}

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 Connector {

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

        use GottenError::*;

        let Self { phased, .. } = self;

        match phased {

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

            ConnectorPhased::Communication(comm) => match &comm.round_result {

                Err(_) => Err(PreviousSyncFailed),

                Ok(None) => Err(NoPreviousRound),

                Ok(Some(round_ok)) => round_ok.gotten.get(&port).ok_or(PortDidntGet),

            },

        }

    }

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

        // returns index of new batch

        use NextBatchError::*;

        let Self { phased, .. } = self;

        match phased {

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

            ConnectorPhased::Communication(comm) => {

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

                Ok(comm.native_batches.len() - 1)

            }

        }

    }

    fn port_op_access(

        &mut self,

        port: PortId,

        expect_polarity: Polarity,

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

        use PortOpError::*;

        let Self { unphased, phased } = self;

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

            return Err(PortUnavailable);

        }

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

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

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

            None => return Err(UnknownPolarity),

        }

        match phased {

            ConnectorPhased::Setup { .. } => Err(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::*;

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

        if batch.to_put.contains_key(&port) {

            Err(MultipleOpsOnPort)

        } else {

            batch.to_put.insert(port, payload);

            Ok(())

        }

    }

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

        use PortOpError::*;

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

        if batch.to_get.insert(port) {

            Ok(())

        } else {

            Err(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, phased } = self;

        match phased {

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

            ConnectorPhased::Communication(comm) => {

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

                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;

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

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

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

        );

        // 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]));

            }

        }

        // restore the invariant

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

        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));

                    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");

                                    return Ok(Decision::Failure);

                                }

                            }

                        } else {

                            log!(

                                cu.logger,

                                "Delivery to getter {:?} msg {:?} failed because {:?} isn't here",

                                getter,

                                &send_payload_msg,

                                proto_component_id

                            );

                        }

                    }

                }

            }

            // check if we have a solution yet

            log!(cu.logger, "Check if we have any local decisions...");

            for solution in solution_storage.iter_new_local_make_old() {

                log!(cu.logger, "New local decision with solution {:?}...", &solution);

                match comm.neighborhood.parent {

                    Some(parent) => {

                        log!(cu.logger, "Forwarding to my parent {:?}", parent);

                        let suggestion = Decision::Success(solution);

                        let msg = Msg::CommMsg(CommMsg {

                            round_index: comm.round_index,

                            contents: CommMsgContents::Suggest { suggestion },

                        });

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

                    }

                    None => {

                        log!(cu.logger, "No parent. Deciding on solution {:?}", &solution);

                        return Ok(Decision::Success(solution));

                    }

                }

            }

            // stuck! make progress by receiving a msg

            // try recv messages arriving through endpoints

            log!(cu.logger, "No decision yet. Let's recv an endpoint msg...");

            {

                let (endpoint_index, msg) = loop {

                    match comm.endpoint_manager.try_recv_any_comms(&mut *cu.logger, deadline)? {

                        None => {

                            log!(cu.logger, "Reached user-defined deadling without decision...");

                            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);

                            }

                            deadline = None;

                        }

                        Some((endpoint_index, msg)) => break (endpoint_index, msg),

                    }

                };

                log!(cu.logger, "Received from endpoint {} msg {:?}", endpoint_index, &msg);

                let comm_msg_contents = match msg {

                    Msg::SetupMsg(..) => {

                        log!(cu.logger, "Discarding setup message; that phase is over");

                        continue 'undecided;

                    }

                    Msg::CommMsg(comm_msg) => match comm_msg.round_index.cmp(&comm.round_index) {

                        Ordering::Equal => comm_msg.contents,

                        Ordering::Less => {

                            log!(