Asynchronous fanin/fanout.

[brettviren]

I believe Phlex lacks a DFP graph pattern is included in WCT. This may be okay for Phlex but it is critical in WCT to supply the so called "sparse processing" mode, aka "asynchronous fanin/fanout" aka "hydra" (the WCT node category that implements it). Likely this particular use case need not ever "leak" out of the WCT DFP graph and into the Phlex graph but I'll describe it in case the general pattern would be needed in some Phlex context.

The basic idea is to NOT require all input to be populated prior to executing the node and/or allowing only a subset of output ports to be populated.

Specifically for WCT, we narrow this allowance to exactly one input port or exactly one output port being populated.

This is used to implement "sparse processing" mode in the following way:

depos -> fanout -> simsp -> fanin -> downstream
            |                 ^
            |                 |
            +----> dummy -----+

Where this pipeline would be drawn for each, say, APA in a graph that spans the entire DUNE FDHD. Not drawn is an upstream (normal) fanout that sends depos into each pipeline.

Here, each time a "fanout" is called it evaluates its input depos in the context of its configured APA and decides if they are "interesting" or not. If interesting the depos are output to port 0 going to "simsp" else port 1 going to "dummy".

The "simsp" is the relatively expensive full simulation + signal processing, and the "dummy" is trivial and fast and simply outputs an empty frame matching the data type output by "simsp" but with close to zero content (and specifically no signal waveforms).

The final fanin gets input on port 0 or port 1 but not both and simply outputs what it gets to downstream consumers.

---

I tried to figure out if/how Phlex supports this by pointing an LLM at phlex code and the design doc and it concludes that the input side (fanin) is not supported but the output can be. To save the tokens, I'll paste that LLM verbiage but I do not claim to fully understand it.

What capability is there for a node to execute based on incomplete
input across multiple ports or to produce output on a subset of ports?
    
Based on the research already gathered:
    
Input side — firing on incomplete/partial ports:

The core TBB join_node with tag_matching requires all input ports to
receive a matching message before firing. There is no native "fire on
any N of M ports" in the Phlex graph as described. The
multilayer_join_node always waits for all inputs (it uses tag_matching
join, and the repeater nodes exist precisely to guarantee all ports
get a message even when data lives at different hierarchy levels). So
incomplete input across multiple ports is not a supported execution
mode — the design ensures completeness via repeaters rather than
tolerating absence.
    
Output side — producing to a subset of ports:

Two mechanisms exist: 

1. Predicates + filter nodes: A predicate node returns a bool; the
filter composite node accumulates predicate decisions via decision_map
and only forwards the message downstream if the decision is "pass."
This is the primary mechanism for conditionally suppressing output on
a path.

2. Selective try_put() on multi-port output tuples: Unfold and
provider nodes hold tuple output ports and call try_put() per-port
independently. A port that never receives a try_put() simply produces
no message — downstream nodes on that edge get nothing. This is how
an unfold node can produce zero or more child cell messages.
  
So the asymmetry is: inputs must be complete (the framework enforces
this), but outputs can be selectively produced either by predicate
gating or by the algorithm simply not calling try_put() on a given
output port.
    

[brettviren]

Here is perhaps a more applicable example:

We will have several "sources" of initial kinematics. These might be from file or from upstream Phlex subgraphs:

1. Ar39 decays.
2. cosmic muons.
3. atmospheric neutrinos.
4. beam neutrinos.

We want to merge these into a single stream of kinematics in time order for input to Geant4. (Note, we can argue that we should instead run Geant4 separately on each source but that simply moves the problem up to how to handle different sources of depos).

That "merge node" can not be a simple "join" type node taking a 4-tuple because each source has very different mean interaction times and those times are randomly distributed according to a distribution.

The merge algorithm MUST (I believe) be able to selectively decide when to accept fresh input from any given port based on what input it already has from all ports so that it can minimally buffer until it is safe to perform a partial merge. To reduce buffer bloat, best if the merge node also may perform "eager output" of any time period for which all input ports have "spoken".

I think of two alternatives, both with downsides.

• Force all sources to produce output collections that span common, lock-stepped / synced windows. This will either lead to emitting many empty collections or defining very large windows or both. This is because of the very different natural window size across the sources (microsecond for Ar39 and what, days for atmospheric).

• Take an additive approach. Pick the "interesting" kinematics (eg beam) an have special MC/sim that given a window around that, fill in more kinematics that might be causally connected. This may require inclusion of kinematics far in the future (eg, for neutrons, muon-decay electrons). It also would not be a source of studies for the "uninteresting" kinematics providing pure background samples.

[brettviren]

A maybe need it example: If we do not require empty place holder records then we can expect async merging pattern to apply when we bring together per-APA/CRU streams across their full detector module. The async merge is needed due to each stream being sparse without the placeholders.