gstreamer_example.cc

Go to the documentation of this file.

// gstreamer
#include "gstreamer.h"
 
// gemc
#include "glogger.h"
#include "gdynamicdigitization.h"
#include "gutilities.h"
#include "gthreads.h"
 
// c++
#include <atomic>
#include <ranges>
#include <vector>
#include <memory>
#include <unordered_map>
 
const std::string plugin_name = "test_gdynamic_plugin";
 
void run_simulation_in_threads(int                                                              nevents,
                               int                                                              nthreads,
                               const std::shared_ptr<GLogger>&                                  log,
                               const std::shared_ptr<const gdynamicdigitization::dRoutinesMap>& dynamicRoutinesMap,
                               const std::shared_ptr<GOptions>&                                 gopts) {
    // Shared atomic event counter used to distribute work without external locking.
    std::atomic<int> next{1};
 
    // Thread pool. Each worker owns its local streamer map and processes an arbitrary number of events.
    std::vector<jthread_alias> pool;
    pool.reserve(nthreads);
 
    for (int tid = 0; tid < nthreads; ++tid) {
        pool.emplace_back([&, tid]
        {
            log->info(0, "worker ", tid, " started");
 
            int localCount = 0;
 
            // Create all configured streamers for this worker thread.
            auto gstreamer_map = gstreamer::gstreamersMapPtr(gopts, tid);
 
            // Open every backend connection before entering the event loop.
            for (auto& [name, gstreamer] : *gstreamer_map) {
                if (!gstreamer->openConnection()) {
                    log->error(1, "Failed to open connection for GStreamer ", name, " in thread ", tid);
                }
            }
 
            while (true) {
                // Claim the next event number atomically.
                int evn = next.fetch_add(1, std::memory_order_relaxed);
                if (evn > nevents) break;
 
                // Build one event collection with a thread-specific header.
                auto gevent_header = GEventHeader::create(gopts, tid);
                auto eventData     = std::make_shared<GEventDataCollection>(gopts, std::move(gevent_header));
                eventData->setAncestors({
                    {11, 1, 0, 1000.0, 0.0, 0.0, 1000.0, 0.0, 0.0, 0.0},
                    {22, 2, 1, 500.0, 10.0, 0.0, 490.0, 1.0, 2.0, 3.0}
                });
 
                // Populate a small synthetic detector dataset.
                for (unsigned i = 1; i < 11; i++) {
                    auto hit       = GHit::create(gopts);
                    auto true_data = dynamicRoutinesMap->at(plugin_name)->collectTrueInformation(hit, i);
                    auto digi_data = dynamicRoutinesMap->at(plugin_name)->digitizeHit(hit, i);
 
                    eventData->addDetectorDigitizedData("ctof", std::move(digi_data));
                    eventData->addDetectorTrueInfoData("ctof", std::move(true_data));
                }
 
                log->info(0, "worker ", tid, " event ", evn, " has ",
                          eventData->getDataCollectionMap().at("ctof")->getDigitizedData().size(),
                          " digitized hits");
 
                // Publish the event to each configured streamer instance.
                for (const auto& [name, gstreamer] : *gstreamer_map) {
                    gstreamer->publishEventData(eventData);
                }
 
                ++localCount;
            }
 
            // Close all output backends. This also flushes any buffered events.
            for (const auto& [name, gstreamer] : *gstreamer_map) {
                if (!gstreamer->closeConnection()) {
                    log->error(1, "Failed to close connection for GStreamer ", name, " in thread ", tid);
                }
            }
 
            log->info(0, "worker ", tid, " processed ", localCount, " events");
        });
    }
}
 
int main(int argc, char* argv[]) {
    // Build the module option set and parse the command line.
    auto gopts = std::make_shared<GOptions>(argc, argv, gstreamer::defineOptions());
 
    // Create a logger scoped to this example.
    auto log = std::make_shared<GLogger>(gopts, SFUNCTION_NAME, GSTREAMER_LOGGER);
 
    constexpr int nevents  = 200;
    constexpr int nthreads = 4;
 
    // Load the dynamic digitization plugin used to generate synthetic detector content.
    auto dynamicRoutinesMap = gdynamicdigitization::dynamicRoutinesMap({plugin_name}, gopts);
    if (dynamicRoutinesMap->at(plugin_name)->loadConstants(1, "default") == false) {
        log->error(1, "Failed to load constants for dynamic routine", plugin_name,
                   "for run number 1 with variation 'default'.");
    }
 
    // Load streamer plugins before worker threads start. Sanitized Linux builds can fail
    // late dlopen() calls from workers with static TLS exhaustion.
    auto preloaded_gstreamer_map = gstreamer::preloadGStreamerPlugins(gopts);
    (void) preloaded_gstreamer_map;
 
    run_simulation_in_threads(nevents, nthreads, log, dynamicRoutinesMap, gopts);
 
    return EXIT_SUCCESS;
}