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

    "-L",           # lib path flag

    "./",           # where to look for libs

    "-lreowolf_rs", # add lib called "reowolf_rs"

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

    c_file,         # input source file

    "-o",           # output flag

    c_file[:-2]     # output filename

  ];

    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 {

                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

            }

        };

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

        }

        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

        let mut draining = HashMap::default();

        let finished = &mut self.branches;

        std::mem::swap(&mut draining, finished);

        for (predicate, mut branch) in draining.drain() {

            log!(cu.logger, "visiting native branch {:?} with {:?}", &branch, &predicate);

            // check if this branch expects to receive it

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

            let mut feed_branch = |branch: &mut NativeBranch, predicate: &Predicate| {

                let was = branch.gotten.insert(getter, send_payload_msg.payload.clone());

                assert!(was.is_none());

                branch.to_get.remove(&getter);

                if branch.to_get.is_empty() {

                    let route = Route::LocalComponent(ComponentId::Native);

                    solution_storage.submit_and_digest_subtree_solution(

                        &mut *cu.logger,

                        route,

                        predicate.clone(),

                    );

                }

            };

            if predicate.query(var) != Some(true) {

                // optimization. Don't bother trying this branch

                log!(

                    cu.logger,

                    feed_branch(&mut branch2, &predicate2);

                    log!(

                        cu.logger,

                        "new subsuming pred created {:?}. forking and feeding",

                        &predicate2

                    );

                    finished.insert(predicate, branch);

                    finished.insert(predicate2, branch2);

                }

            }

        }

    }

        for (branch_predicate, branch) in self.branches {

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

                return RoundOk { batch_index: index, gotten };

            }

        }

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

    }

}

// |putter, m| {

//     let getter = *cu.port_info.peers.get(&putter).unwrap();

//     payloads_to_get.push((getter, m));

// },

impl BranchingProtoComponent {

    fn drain_branches_to_blocked(

                B::SyncBlockEnd => {

                    // make concrete all variables

                    for &port in ports.iter() {

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

                        predicate.assigned.entry(var).or_insert(false);

                    }

                    // submit solution for this component

                    solution_storage.submit_and_digest_subtree_solution(

                        &mut *cu.logger,

                        Route::LocalComponent(ComponentId::Proto(proto_component_id)),

                        predicate.clone(),

                    );

                    // move to "blocked"

                    drainer.add_output(predicate, branch);

                }

                B::CouldntReadMsg(port) => {

                    // move to "blocked"

                    assert!(!branch.inbox.contains_key(&port));

                    drainer.add_output(predicate, branch);

                }

                B::CouldntCheckFiring(port) => {

                    // sanity check

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

                    assert!(predicate.query(var).is_none());

            logger,

            "feeding proto component {:?} getter {:?} {:?}",

            proto_component_id,

            getter,

            &send_payload_msg

        );

        let BranchingProtoComponent { branches, ports } = self;

        let mut unblocked = HashMap::default();

        let mut blocked = HashMap::default();

        // partition drain from branches -> {unblocked, blocked}

        log!(logger, "visiting {} blocked branches...", branches.len());

        for (predicate, mut branch) in branches.drain() {

            use CommonSatResult as Csr;

            log!(logger, "visiting branch with pred {:?}", &predicate);

            match predicate.common_satisfier(&send_payload_msg.predicate) {

                Csr::Nonexistant => {

                    // this branch does not receive the message

                    log!(logger, "skipping branch");

                    blocked.insert(predicate, branch);

                }

                Csr::Equivalent | Csr::FormerNotLatter => {

                    // retain the existing predicate, but add this payload

                    log!(logger, "feeding this branch without altering its predicate");

                    branch.feed_msg(getter, send_payload_msg.payload.clone());

            payload_msg_sender,

            proto_component_id,

            ports,

        )?;

        // swap the blocked branches back

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

        log!(cu.logger, "component settles down with branches: {:?}", branches.keys());

        Ok(())

    }

    fn collapse_with(self, solution_predicate: &Predicate) -> ProtoComponent {

        let BranchingProtoComponent { ports, branches } = self;

        for (branch_predicate, branch) in branches {

                let ProtoComponentBranch { state, .. } = branch;

                return ProtoComponent { state, ports };

            }

        }

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

    }

    fn initial(ProtoComponent { state, ports }: ProtoComponent) -> Self {

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

    }

}

impl SolutionStorage {

    fn new(routes: impl Iterator<Item = Route>) -> Self {

        let mut subtree_id_to_index: HashMap<Route, usize> = Default::default();

        let mut subtree_solutions = vec![];

        for key in routes {

            subtree_id_to_index.insert(key, subtree_solutions.len());

            subtree_solutions.push(Default::default())

        }

        Self {

}

#[derive(Debug)]

enum TryRecyAnyError {

    Timeout,

    PollFailed,

    EndpointError { error: EndpointError, index: usize },

}

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

impl Endpoint {

    pub(super) fn try_recv<T: serde::de::DeserializeOwned>(

        &mut self,

    ) -> Result<Option<T>, EndpointError> {

        use EndpointError::*;

        // populate inbox as much as possible

        'read_loop: loop {

            let res = self.stream.read_to_end(&mut self.inbox);

            endptlog!(logger, "Stream read to end {:?}", &res);

            match res {

                Err(e) if would_block(&e) => break 'read_loop,

                Ok(0) => break 'read_loop,

                Ok(_) => (),

                Err(_e) => return Err(BrokenEndpoint),

            }

            return -1;

        }

    };

    let sock_address: SocketAddr = match addr_str.parse() {

        Ok(addr) => addr,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            return -2;

        }

    };

    match connector.new_net_port(port_polarity, sock_address, endpoint_polarity) {

        Ok(p) => {

                port.write(p);

            0

        }

        Err(err) => {

            StoredError::tl_debug_store(&err);

            -3

        }

    }

}

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

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

#[no_mangle]

        Ok(n) => n as isize,

        Err(err) => {

            StoredError::tl_debug_store(&err);

            -1

        }

    }

}

#[no_mangle]

pub unsafe extern "C" fn connector_gotten_bytes(

    connector: &mut Connector,

    port: PortId,

) -> *const u8 {

    StoredError::tl_clear();

    match connector.gotten(port) {

        Ok(payload_borrow) => {

            let slice = payload_borrow.as_slice();

            slice.as_ptr()

        }

        Err(err) => {

            StoredError::tl_debug_store(&err);

            std::ptr::null()

        }

    }

}

// #[no_mangle]

// unsafe extern "C" fn connector_gotten_payload(

//     connector: &mut Connector,

    pub fn new(connector_id: ConnectorId) -> Self {

        Self(connector_id, Default::default())

    }

}

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

impl Logger for DummyLogger {

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

        self

    }

}

impl Logger for VecLogger {

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

        self

    }

}

impl Logger for FileLogger {

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

        &mut self.1

    }

}

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

impl Drop for VecLogger {

    fn drop(&mut self) {

        let stdout = std::io::stderr();

        let mut lock = stdout.lock();

        writeln!(lock, "--- DROP LOG DUMP ---").unwrap();

        let _ = std::io::Write::write(&mut lock, self.1.as_slice());

    }

}

pub fn would_block(err: &std::io::Error) -> bool {

    err.kind() == std::io::ErrorKind::WouldBlock

}

impl<T: std::cmp::Ord> VecSet<T> {

    fn new(mut vec: Vec<T>) -> Self {

        vec.sort();

        vec.dedup();

        Self { vec }

    }

    fn contains(&self, element: &T) -> bool {

        self.vec.binary_search(element).is_ok()

    }

    fn iter(&self) -> std::slice::Iter<T> {

        self.vec.iter()

    }

}

impl PortInfo {

    fn firing_var_for(&self, port: PortId) -> FiringVar {

        FiringVar(match self.polarities.get(&port).unwrap() {

            Getter => port,

            Putter => *self.peers.get(&port).unwrap(),

        })

    }

}

        }

        .into()

    }

}

impl Drop for Connector {

    fn drop(&mut self) {

        log!(&mut *self.unphased.logger, "Connector dropping. Goodbye!");

    }

}

impl Connector {

    fn random_id() -> ConnectorId {

        type Bytes8 = [u8; std::mem::size_of::<ConnectorId>()];

    }

    pub fn swap_logger(&mut self, mut new_logger: Box<dyn Logger>) -> Box<dyn Logger> {

        std::mem::swap(&mut self.unphased.logger, &mut new_logger);

        new_logger

    }

    pub fn get_logger(&mut self) -> &mut dyn Logger {

        &mut *self.unphased.logger

    }

    pub fn new_port_pair(&mut self) -> [PortId; 2] {

        let cu = &mut self.unphased;

        // adds two new associated ports, related to each other, and exposed to the native

        let [o, i] = [cu.id_manager.new_port_id(), cu.id_manager.new_port_id()];

        Ok(Todo {

            todo_endpoint,

            local_port,

            sent_local_port: false,

            recv_peer_port: None,

            endpoint_setup: endpoint_setup.clone(),

        })

    }

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

    // 1. Start to construct EndpointManager

    const WAKER_TOKEN: Token = Token(usize::MAX);

    assert!(endpoint_setups.len() < WAKER_TOKEN.0); // using MAX usize as waker token

    let mut waker_continue_signal: Option<Arc<AtomicBool>> = None;

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

    let mut events = Events::with_capacity(endpoint_setups.len() * 2 + 4);

    let mut polled_undrained = IndexSet::default();

    let mut delayed_messages = vec![];

    // 2. create a registered (TcpListener/Endpoint) for passive / active respectively

    let mut todos = endpoint_setups

        .iter()

        .enumerate()

        .map(|(index, (local_port, endpoint_setup))| {

            init_todo(Token(index), *local_port, endpoint_setup, &mut poll)

        })

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

    // 3. Using poll to drive progress:

    //    - accept an incoming connection for each TcpListener (turning them into endpoints too)

    //    - for each endpoint, send the local PortId

    //    - for each endpoint, recv the peer's PortId, and

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

    let mut setup_incomplete: HashSet<usize> = (0..todos.len()).collect();

    while !setup_incomplete.is_empty() {

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

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

        } else {

            None

        };

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

        for event in events.iter() {

            let token = event.token();

            let Token(index) = token;

            if token == WAKER_TOKEN {

                assert!(waker_continue_signal.is_some());

                for index in connect_failed.drain() {

                    log!(

                        logger,

                        "Restarting connection with endpoint {:?} {:?}",

                        index,

                        todo.endpoint_setup.sock_addr

                    );

                    match &mut todo.todo_endpoint {

                        TodoEndpoint::Endpoint(endpoint) => {

                            let mut new_stream = TcpStream::connect(todo.endpoint_setup.sock_addr)

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

                            std::mem::swap(&mut endpoint.stream, &mut new_stream);

                        }

                        _ => unreachable!(),

                    }

                }

            } else {

                // FIRST try convert this into an endpoint

                if let TodoEndpoint::Accepting(listener) = &mut todo.todo_endpoint {

                    match listener.accept() {

                        Ok((mut stream, peer_addr)) => {

                            poll.registry().deregister(listener).unwrap();

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

                            log!(

                                logger,

                                "Endpoint[{}] accepted a connection from {:?}",

                                index,

                                peer_addr

                            );

                        Err(_) => {

                            log!(logger, "accept() failure on index {}", index);

                            return Err(AcceptFailed(listener.local_addr().unwrap()));

                        }

                    }

                }

                if let TodoEndpoint::Endpoint(endpoint) = &mut todo.todo_endpoint {

                    if event.is_error() {

                        if todo.endpoint_setup.endpoint_polarity == EndpointPolarity::Passive {

                            // right now you cannot retry an acceptor.

                            return Err(AcceptFailed(endpoint.stream.local_addr().unwrap()));

                        }

                        if waker_continue_signal.is_none() {

                            log!(logger, "First connect failure. Starting waker thread");

                            let waker =

                                Arc::new(mio::Waker::new(poll.registry(), WAKER_TOKEN).unwrap());

                            let wcs = Arc::new(AtomicBool::from(true));

                            let wcs2 = wcs.clone();

                            std::thread::spawn(move || {

                                while wcs2.load(std::sync::atomic::Ordering::SeqCst) {

                                    std::thread::sleep(WAKER_PERIOD);

                                    let _ = waker.wake();

                                }

                            });

    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)

}

lazy_static::lazy_static! {

    static ref MINIMAL_PROTO: Arc<ProtocolDescription> = {

    };

}

lazy_static::lazy_static! {

    static ref TEST_MSG: Payload = {

        Payload::from(b"hello" as &[u8])

    };

}

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

#[test]

fn basic_connector() {

#[test]

fn net_self_loop() {

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

    let sock_addr = next_test_addr();

    let mut c = file_logged_connector(0, test_log_path);

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

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

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

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

    c.get(g).unwrap();

    c.sync(Some(Duration::from_millis(500))).unwrap();

}