working mana_replay but doesn't do any analysis yet

reaches 210 MiB/s with is690-9Li_3H_run094

external/mesytec-mvlc

@@ -1 +1 @@
-Subproject commit 7dec563d2b5abbfb83b843457f6c7c446b987f50
+Subproject commit 1c1f32dbc587747219a942aabe2a2668fab1ae82

src/tools/mana_auto_replay.cc

@@ -55,6 +55,12 @@ inline int64_t round_up(int64_t n, int64_t p)
 
 struct Arena
 {
+    Arena() = default;
+    ~Arena() = default;
+    Arena(const Arena &) = delete;
+    Arena &operator=(const Arena &) = delete;
+    Arena(Arena &&) = delete;
+    Arena &operator=(Arena &&) = delete;
     struct Segment
     {
         std::vector<u8> data;
@@ -116,19 +122,26 @@ struct Arena
 
 struct ListfileContext
 {
-    mvlc::listfile::ZipReader zipReader;
+    std::unique_ptr<mvlc::listfile::ZipReader> zipReader;
     mvlc::listfile::ReadHandle *readHandle;
     mvlc::listfile::ListfileReaderHelper readerHelper;
     mvlc::CrateConfig crateConfig;
+
+    ListfileContext() = default;
+    ListfileContext(const ListfileContext &) = delete;
+    ListfileContext &operator=(const ListfileContext &) = delete;
+    ListfileContext(ListfileContext &&) = default;
+    ListfileContext &operator=(ListfileContext &&) = default;
 };
 
 std::optional<ListfileContext> make_listfile_context(const std::string &filename)
 {
     try
     {
-        ListfileContext ctx{};
-        ctx.zipReader.openArchive(filename);
-        ctx.readHandle = ctx.zipReader.openEntry(ctx.zipReader.firstListfileEntryName());
+        ListfileContext ctx;
+        ctx.zipReader = std::make_unique<mvlc::listfile::ZipReader>();
+        ctx.zipReader->openArchive(filename);
+        ctx.readHandle = ctx.zipReader->openEntry(ctx.zipReader->firstListfileEntryName());
         ctx.readerHelper = mvlc::listfile::make_listfile_reader_helper(ctx.readHandle);
         auto configData = ctx.readerHelper.preamble.findCrateConfig();
         if (!configData)
@@ -146,10 +159,44 @@ std::optional<ListfileContext> make_listfile_context(const std::string &filename
             for (const auto &module_: ctx.crateConfig.stacks[i].getGroups())
             {
                 auto meta = fmt::format("{}", fmt::join(module_.meta, ", "));
-                std::cout << fmt::format("  module[{}]: {}, {}\n", mi++, module_.name, meta);
+                std::cout << fmt::format("  module[{}]: name={}, meta={{{}}}", mi++, module_.name,
+                                         meta);
+                if (!mvlc::produces_output(module_))
+                    std::cout << ", (no output)";
+                std::cout << "\n";
             }
         }
 
+        return std::move(ctx);
+    }
+    catch (const std::exception &e)
+    {
+        std::cerr << fmt::format("Error: {}\n", e.what());
+        return {};
+    }
+}
+
+struct ParserContext
+{
+    using Parser = mvlc::readout_parser::ReadoutParserState;
+    using Counters = mvlc::readout_parser::ReadoutParserCounters;
+    using Callbacks = mvlc::readout_parser::ReadoutParserCallbacks;
+
+    Parser parser;
+    Counters counters;
+    Callbacks callbacks;
+    mvlc::CrateConfig crateConfig;
+};
+
+std::optional<ParserContext> make_parser_context(const mvlc::CrateConfig &crateConfig,
+                                                 ParserContext::Callbacks callbacks)
+{
+    try
+    {
+        ParserContext ctx{};
+        ctx.parser = mvlc::readout_parser::make_readout_parser(crateConfig.stacks);
+        ctx.crateConfig = crateConfig;
+        ctx.callbacks = callbacks;
         return ctx;
     }
     catch (const std::exception &e)
@@ -159,6 +206,46 @@ std::optional<ListfileContext> make_listfile_context(const std::string &filename
     }
 }
 
+struct AnalysisContext
+{
+};
+
+void event_data_callback(AnalysisContext *ctx, int crateIndex, int eventIndex,
+                         const mvlc::readout_parser::ModuleData *moduleDataList,
+                         unsigned moduleCount)
+{
+    assert(moduleDataList);
+    // std::cout << fmt::format("ReadoutEvent: ctx={}, crateId={}, eventIndex={}, moduleCount={}\n",
+    //                          fmt::ptr(ctx), crateIndex, eventIndex, moduleCount);
+}
+
+void system_event_callback(AnalysisContext *ctx, int crateIndex, const u32 *header, u32 size)
+{
+    assert(header);
+    // std::cout << fmt::format("SystemEvent: ctx={}, crateId={}, size={}\n", fmt::ptr(ctx),
+    // crateIndex,
+    //                          size);
+}
+
+size_t process_one_buffer(size_t bufferNumber, ListfileContext &listfileContext,
+                          ParserContext &parserContext, AnalysisContext &analysisContext)
+{
+    listfileContext.readerHelper.destBuf().clear();
+    auto buffer = mvlc::listfile::read_next_buffer(listfileContext.readerHelper);
+
+    if (!buffer->used())
+        return 0;
+
+    auto bufferView = buffer->viewU32();
+    parserContext.parser.userContext = &analysisContext;
+
+    mvlc::readout_parser::parse_readout_buffer(
+        listfileContext.readerHelper.bufferFormat, parserContext.parser, parserContext.callbacks,
+        parserContext.counters, bufferNumber, bufferView.data(), bufferView.size());
+
+    return buffer->used();
+}
+
 int main(int argc, char *argv[])
 {
     auto f = cmrc::mnode::resources::get_filesystem().open("data/vme_module_data_sources.json");
@@ -176,5 +263,60 @@ int main(int argc, char *argv[])
     if (!listfileContext)
         return 1;
 
+    std::optional<AnalysisContext> analysisContext = AnalysisContext();
+
+    if (!analysisContext)
+        return 1;
+
+    auto event_data = [](void *ctx, int crateIndex, int eventIndex,
+                         const mvlc::readout_parser::ModuleData *moduleDataList,
+                         unsigned moduleCount)
+    {
+        event_data_callback(static_cast<AnalysisContext *>(ctx), crateIndex, eventIndex,
+                            moduleDataList, moduleCount);
+    };
+
+    auto system_event = [](void *ctx, int crateIndex, const u32 *header, u32 size)
+    { system_event_callback(static_cast<AnalysisContext *>(ctx), crateIndex, header, size); };
+
+    auto parserContext =
+        make_parser_context(listfileContext->crateConfig, {event_data, system_event});
+
+    if (!parserContext)
+        return 1;
+
+    size_t bufferNumber = 0;
+    size_t totalBytesProcessed = 0;
+    size_t bytesProcessed = 0;
+    const std::chrono::milliseconds ReportInterval(500);
+    mvlc::util::Stopwatch sw;
+
+    auto report =
+        [](const mvlc::util::Stopwatch sw, size_t bufferNumber, size_t totalBytesProcessed)
+    {
+        auto s = sw.get_elapsed().count() / (1000.0 * 1000.0);
+        auto bytesPerSecond = totalBytesProcessed / s;
+        auto MiBPerSecond = bytesPerSecond / (1u << 20);
+        std::cout << fmt::format("Processed {} buffers, {} bytes. t={} s, rate={} MiB/s\n",
+                                 bufferNumber, totalBytesProcessed, s, MiBPerSecond);
+    };
+
+    do
+    {
+        bytesProcessed =
+            process_one_buffer(bufferNumber, *listfileContext, *parserContext, *analysisContext);
+        totalBytesProcessed += bytesProcessed;
+        ++bufferNumber;
+
+        if (auto elapsed = sw.get_interval(); elapsed >= ReportInterval)
+        {
+            report(sw, bufferNumber, totalBytesProcessed);
+            sw.interval();
+        }
+    }
+    while (bytesProcessed > 0);
+
+    report(sw, bufferNumber, totalBytesProcessed);
+
     return 0;
 }

src/tools/mana_replay_api.h

@@ -9,6 +9,15 @@ extern "C"
 #endif
 {
 
+// descriptors - daq structure
+// daq   -> events   -> modules -> data_sources
+// alternative concepts:
+// crate -> triggers -> modules -> data_sources
+// example data sources: "amplitude", "channel_time", "timestamp"
+// data is float, data sources have a size of at least 1
+// todo: add number of valid bits in the data source
+// todo: rng handling?
+
 typedef struct
 {
     const char *name;