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

use super::*;

use crate::common::*;

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

struct BranchingNative {

    branches: HashMap<Predicate, NativeBranch>,

}

#[derive(Clone, Debug)]

struct NativeBranch {

    index: usize,

    gotten: HashMap<PortId, Payload>,

    to_get: HashSet<PortId>,

}

#[derive(Debug)]

struct SolutionStorage {

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

            }

        }

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

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

                                cu.logger,

                                "We are in round {}, but msg is for round {}. Discard",

                                comm_msg.round_index,

                                comm.round_index,

                            );

                            drop(comm_msg);

                            continue 'undecided;

                        }

                        Ordering::Greater => {

                            log!(

                                cu.logger,

                                "We are in round {}, but msg is for round {}. Buffer",

                                comm_msg.round_index,

                                comm.round_index,

                            );

                            comm.endpoint_manager

                                .delayed_messages

                                .push((endpoint_index, Msg::CommMsg(comm_msg)));

                            continue 'undecided;

                        }

                    },

                };

                match comm_msg_contents {

                    CommMsgContents::SendPayload(send_payload_msg) => {

                        let getter =

                            comm.endpoint_manager.endpoint_exts[endpoint_index].getter_for_incoming;

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

                        log!(

                            cu.logger,

                            "Msg routed to getter port {:?}. Buffer for recv loop",

                            getter,

                        );

                        payloads_to_get.push((getter, send_payload_msg));

                    }

                    CommMsgContents::Suggest { suggestion } => {

                        // only accept this control msg through a child endpoint

                        if comm.neighborhood.children.contains(&endpoint_index) {

                            match suggestion {

                                Decision::Success(predicate) => {

                                    // child solution contributes to local solution

                                    log!(cu.logger, "Child provided solution {:?}", &predicate);

                                    let route = Route::Endpoint { index: endpoint_index };

                                    solution_storage.submit_and_digest_subtree_solution(

                                        &mut *cu.logger,

                                        route,

                                        predicate,

                                    );

                                }

                                Decision::Failure => {

                                    match comm.neighborhood.parent {

                                        None => {

                                            log!(cu.logger, "I decide on my child's failure");

                                            break 'undecided Ok(Decision::Failure);

                                        }

                                        Some(parent) => {

                                            log!(cu.logger, "Forwarding failure through my parent endpoint {:?}", 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,

                                "Discarding suggestion {:?} from non-child endpoint idx {:?}",

                                &suggestion,

                                endpoint_index

                            );

                        }

                    }

                    CommMsgContents::Announce { decision } => {

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

                            // adopt this decision

                            return Ok(decision);

                        } else {

                            log!(

                                cu.logger,

                                "Discarding announcement {:?} from non-parent endpoint idx {:?}",

                                &decision,

                                endpoint_index

                            );

                        }

                    }

                }

            }

            log!(cu.logger, "Endpoint msg recv done");

        }

    }

    fn request_failure(

        cu: &mut ConnectorUnphased,

        comm: &mut ConnectorCommunication,

        parent: usize,

    ) -> Result<(), SyncError> {

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

        let suggestion = Decision::Failure;

        let msg = Msg::CommMsg(CommMsg {

            round_index: comm.round_index,

            contents: CommMsgContents::Suggest { suggestion },

        });

        comm.endpoint_manager.send_to_comms(parent, &msg)

    }

}

impl BranchingNative {

    fn feed_msg(

        &mut self,

        cu: &mut ConnectorUnphased,

        solution_storage: &mut SolutionStorage,

        getter: PortId,

        send_payload_msg: &SendPayloadMsg,

    ) {

        log!(cu.logger, "feeding native getter {:?} {:?}", getter, &send_payload_msg);

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

        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,

                    "skipping branch with {:?} that doesn't want the message (fastpath)",

                    &predicate

                );

                finished.insert(predicate, branch);