Works for one cfg

aux/inc/WireCellAux/FrameSync.h

@@ -21,22 +21,30 @@
 #define WIRECEL_AUX_FRAMESYNC
 
 #include "WireCellIface/IFrameMerge.h"
+#include "WireCellAux/Logger.h"
 
 namespace WireCell::Aux {
 
-    class FrameSync : public IFrameMerge {
+    class FrameSync : Aux::Logger, public IFrameMerge {
         bool m_flushing{false};
       public:
 
         FrameSync(const size_t multiplicity = 2);
         virtual ~FrameSync();
         virtual std::vector<std::string> input_types();
 
-        // hydra
+        // iqs: vector of input queues, mutable
+        // oqs: tuple of output queues
         virtual bool operator()(input_queues& iqs, output_queues& oqs);
 
       private:
-        size_t m_multiplicity;
+        size_t m_multiplicity {2};
+
+        // recursive processing of input queues
+        void flush(input_queues& iqs, output_queues& oqs);
+
+        // input buffer
+        input_queues m_iqs;
     };
 }
 

aux/src/FrameSync.cxx

@@ -9,7 +9,10 @@ WIRECELL_FACTORY(FrameSync,
 
 using namespace WireCell;
 
-Aux::FrameSync::FrameSync(const size_t multiplicity) : m_multiplicity(multiplicity) { }
+Aux::FrameSync::FrameSync(const size_t multiplicity)
+    : Aux::Logger("FrameSync", "glue")
+    , m_multiplicity(multiplicity)
+    , m_iqs(m_multiplicity) { }
 Aux::FrameSync::~FrameSync() { }
 
 std::vector<std::string> Aux::FrameSync::input_types()
@@ -19,51 +22,68 @@ std::vector<std::string> Aux::FrameSync::input_types()
     return ret;
 }
 
-bool Aux::FrameSync::operator()(input_queues& iqs, output_queues& oqs)
+void Aux::FrameSync::flush(input_queues& iqs, output_queues& oqs)
 {
-    size_t neos=0, nmin=0;
-    int min_ident=0;
+    size_t neos=0, nmin=0, nempty=0;
+    int min_ident=std::numeric_limits<int>::max();
     size_t min_index=0;
 
     const size_t nin = iqs.size();
     for (size_t ind=0; ind<nin; ++ind) {
-        auto& iq = iqs[0];
+        auto& iq = iqs[ind];
         if (iq.empty()) {
-            return true;
+            ++nempty;
+            log->debug("port {} empty", ind);
+            continue;
         }
         auto frame = iq.front();
         if (!frame) {
             ++neos;
+            log->debug("port {} eos", ind);
             continue;
         }
+        log->debug("port {} frame {}", ind, frame->ident());
+        /// FIXME: how about equal?
         if (min_ident > frame->ident()) {
             min_ident = frame->ident();
             min_index = ind;
             ++nmin;
         }
     }
+    // log->debug("neos: {} nmin: {} nempty: {}", neos, nmin, nempty);
+
+    // found min (good)
+    if (nmin > 0) {
+        auto frame = iqs[min_index].front();
+        iqs[min_index].pop_front();
+        std::get<0>(oqs).push_back(frame);
+
+        // May have more
+        return flush(iqs, oqs);
+    }
 
+    // eos sync
     if (neos == iqs.size()) {   // we have EOS sync
         for (auto& iq : iqs) {
             iq.pop_front();     // pop all EOS
         }
         std::get<0>(oqs).push_back(nullptr); // forward EOS
 
         // May have more behind the EOS
-        return (*this)(iqs, oqs);
+        return flush(iqs, oqs);
     }
-
-    if (neos+nmin == nin) {     // fully populated
-        auto frame = iqs[min_index].front();
-        iqs[min_index].pop_front();
-        std::get<0>(oqs).push_back(frame);
+}
 
-        // May have more
-        return (*this)(iqs, oqs);
+bool Aux::FrameSync::operator()(input_queues& iqs, output_queues& oqs)
+{
+    // concatenate input queues to buffer
+    for (size_t ind=0; ind<m_iqs.size(); ++ind) {
+        m_iqs[ind].insert(m_iqs[ind].end(), iqs[ind].begin(), iqs[ind].end());
     }
-
-    // If we reach here then we do not have fully populated input
-    // streams and must wait for more.
 
+    // try to flush
+    flush(m_iqs, oqs);
+    log->debug("oqs size: {}", std::get<0>(oqs).size());
+
     return true;
 }

aux/src/FrameTools.cxx

@@ -244,8 +244,13 @@ std::string Aux::taginfo(const WireCell::IFrame::pointer& frame)
     std::stringstream info;
     info << "frame: ident=" << frame->ident()
          << " time=" << frame->time() << " tick=" << frame->tick()
-         << " with "
-         << frame->traces()->size() << " traces.  frame tags:[ ";
+         << " with ";
+    const auto& traces = frame->traces();
+    if(!traces) {
+        info << " null trace ptr!";
+    } else {
+        info << traces->size() << " traces.  frame tags:[ ";
+    }
     for (const auto& tag : frame->frame_tags()) {
         info << "\"" << tag << "\" ";
     }

gen/inc/WireCellGen/DeposOrBust.h

@@ -14,10 +14,11 @@
 
 #include "WireCellIface/IDepos2DeposOrFrame.h"
 #include "WireCellIface/IConfigurable.h"
+#include "WireCellAux/Logger.h"
 
 namespace WireCell::Gen {
 
-    class DeposOrBust : public WireCell::IDepos2DeposOrFrame
+    class DeposOrBust : Aux::Logger, public WireCell::IDepos2DeposOrFrame
     {
       public:
 

gen/src/DeposOrBust.cxx

@@ -10,7 +10,9 @@ WIRECELL_FACTORY(DeposOrBust,
 
 using namespace WireCell;
 
-Gen::DeposOrBust::DeposOrBust() { }
+Gen::DeposOrBust::DeposOrBust()
+    : Aux::Logger("DeposOrBust", "glue")
+    { }
 Gen::DeposOrBust::~DeposOrBust() { }
 
 // When I get an empty deposet, this is what comes out of port 1.
@@ -57,15 +59,18 @@ bool Gen::DeposOrBust::operator()(input_queues& iqs, output_queues& oqs)
         iq.pop_front();
 
         if (!ds) {              // EOS on every ouput
+            log->debug("EOS sync");
             get<0>(oqs).push_back(nullptr);
             get<1>(oqs).push_back(nullptr);
             continue;
         }
 
         if (ds->depos()->empty()) {
+            log->debug("DeposOrBust: got empty deposet");
             get<1>(oqs).push_back(std::make_shared<EmptyFrame>(ds->ident()));
         }
         else {
+            log->debug("DeposOrBust: forward deposet");
             get<0>(oqs).push_back(ds);
         }
     }

gen/src/FrameFanin.cxx

@@ -151,7 +151,9 @@ bool Gen::FrameFanin::operator()(const input_vector& invec, output_pointer& out)
         }
 
         // Insert one intput frames traces.
-        out_traces.insert(out_traces.end(), traces->begin(), traces->end());
+        if (traces) {
+            out_traces.insert(out_traces.end(), traces->begin(), traces->end());
+        }
     }
 
     auto sf = new Aux::SimpleFrame(one->ident(), one->time(), out_traces, one->tick(), out_cmm);

iface/inc/WireCellIface/IHydraNode.h

@@ -282,6 +282,7 @@ namespace WireCell {
         using HydraInput = IHydraInputV<InputType, FaninMultiplicity>;
         using HydraOutput = IHydraOutputT<OutputTuple>;
 
+        using typename HydraInput::input_pointer;
         using typename IHydraNodeBase::any_queue_vector;
         using typename HydraInput::input_queue;
         using typename HydraInput::input_queues;
@@ -292,10 +293,12 @@ namespace WireCell {
         virtual bool operator()(any_queue_vector& anyinqs, any_queue_vector& anyoutqs)
         {
             // vector input
-            const size_t isize = HydraInput::input_types().size();
+            const size_t isize = anyinqs.size();
             input_queues iqs(isize);
             for (size_t ind=0; ind<isize; ++ind) {
-                iqs[ind] = boost::any_cast<input_queue>(anyinqs[ind]);
+                for (auto any : anyinqs[ind]) {
+                    iqs[ind].push_back(boost::any_cast<input_pointer>(any));
+                }
             }
 
             // cross over

tbb/inc/WireCellTbb/HydraCat.h

@@ -123,12 +123,11 @@ namespace WireCellTbb {
                         const any_queue_vector& oqv,
                         std::vector<size_t>& counts)
     {
-        for (const auto& oq : oqv) {
-            for (const auto& o : oq) {
-                bool accepted = std::get<N>(out).try_put(msg_t(counts[N]++, o));
-                if (!accepted) {
-                    std::cerr << "TbbFlow: hydra node input return false when try put on " << N << "\n";
-                }
+        const auto& oq = oqv[N];
+        for (const auto& o : oq) {
+            bool accepted = std::get<N>(out).try_put(msg_t(counts[N]++, o));
+            if (!accepted) {
+                std::cerr << "TbbFlow: hydra node input return false when try put on " << N << "\n";
             }
         }
         if constexpr (N + 1 < Nout) {

tbb/test/test_tbb_basics.cxx

@@ -1,38 +1,58 @@
 #include <tbb/tbb.h>
 #include <iostream>
 
-int main() {
+int main(int argc, char* argv[]) {
+    int numThreads = 1;
+    if (argc > 1) {
+        numThreads = std::stoi(argv[1]);
+    }
+
+    // Set the number of threads using tbb::global_control
+    tbb::global_control control(tbb::global_control::max_allowed_parallelism, numThreads);
+
     tbb::flow::graph graph;
 
     // Define nodes
     tbb::flow::function_node<int, int> node1(graph, tbb::flow::unlimited, [](const int& input) {
         std::cout << "node 1 input: " << input << std::endl;
         // Perform computation for node1
-        return input * 2;
+        return input + 1;
     });
 
     tbb::flow::function_node<int, int> node2(graph, tbb::flow::unlimited, [](const int& input) {
         std::cout << "node 2 input: " << input << std::endl;
         // Perform computation for node2
-        return input + 5;
+        return input + 1;
     });
 
     // Define edges
-
     tbb::flow::make_edge(node1, node2);
 
+    tbb::flow::sequencer_node<int> sequencer(graph, [](const int& input) {
+        std::cout << "sequencer input: " << input << std::endl;
+        // Perform computation for sequencer
+        return input;
+    });
+    tbb::flow::make_edge(node2, sequencer);
+
+    tbb::flow::function_node<int, int> node3(graph, tbb::flow::unlimited, [](const int& input) {
+        std::cout << "node 3 input: " << input << std::endl;
+        // Perform computation for node3
+        return input + 1;
+    });
+    tbb::flow::make_edge(sequencer, node3);
+
     // Add input to the graph
-    const int limit = 1;
+    const int limit = 2;
     int count = 0;
-    tbb::flow::input_node<int> src( graph, [&]( tbb::flow_control &fc ) -> int {
+    tbb::flow::input_node<int> src(graph, [&](tbb::flow_control& fc) -> int {
         std::cout << "src input" << std::endl;
         if (count++ < limit) {
-            return count + 10;
+            return count * 10;
         }
         fc.stop();
         return {};
     });
-
     tbb::flow::make_edge(src, node1);
 
     std::cout << "Running graph" << std::endl;