logos-cpp-sdk/doctests/cpp-sdk-concurrent-dispatch.test.yaml

name: "Concurrent Dispatch (C++): a single-threaded module driving a concurrent one"
output: cpp-sdk-concurrent-dispatch.md
project_name: cpp-concurrent-dispatch
release: ""

intro: |
  By default a Logos module handles calls **one at a time** — every method runs
  on the module's event loop, so the author never thinks about thread-safety. A
  handler that blocks (a slow download, a slow RPC) then stalls *every other
  caller* until it returns.

  Setting **`"concurrency": "multi"`** in `metadata.json` opts a module into
  **concurrent dispatch**: each incoming call runs on its own worker thread, so
  one blocking handler no longer holds up the others. The author takes on
  thread-safety in exchange — the impl's methods may run in parallel, so any
  shared state must be synchronized (here, `std::atomic`).

  This doc-test builds two C++ modules and proves it end-to-end through a headless
  `logoscore` daemon:

  | Module | `concurrency` | Role |
  |---|---|---|
  | `slow_worker_module` | **`"multi"`** | `work(ms)` blocks for `ms`, recording the PEAK number of calls running at once |
  | `fanout_driver_module` | `"single"` (default) | an ordinary single-threaded module that fires several `work` calls at the worker without waiting between them (the generated `workAsync` client) |

  The driver is single-threaded, yet it drives the worker concurrently: it fires
  N async calls back-to-back, and because the worker is `multi`, all N run at
  once. We then read the worker's observed peak overlap — it equals N. Flip the
  worker to `"single"` and the same fan-out would serialize (peak 1).

what_you_build: "Two composed C++ modules — a concurrency:\"multi\" worker and an ordinary single-threaded driver — showing a single module driving concurrent work in a multi module, validated through a logoscore daemon."

what_you_learn:
  - "What `concurrency: \"multi\"` does for a C++ module and when to use it"
  - "The multi contract for C++: the impl's methods run concurrently, so shared state must be self-synchronized (std::atomic / std::mutex)"
  - "How a single-threaded caller drives concurrent work: fire `modules().<dep>.<method>Async(...)` without waiting between calls"
  - "That a multi worker overlaps those calls (peak == N) while a single worker would serialize them (peak 1)"

prerequisites:
  - |
    **Nix** with flakes enabled. Install from [nixos.org](https://nixos.org/download.html), then enable flakes:

    ```bash
    mkdir -p ~/.config/nix
    echo 'experimental-features = nix-command flakes' >> ~/.config/nix/nix.conf
    ```

    Verify: `nix flake --help >/dev/null 2>&1 && echo "Flakes enabled"`
  - "**A Linux or macOS machine.** Nix provides every toolchain (Qt, CMake) during the builds."

sections:
  - title: "Build the tools"
    step: true
    steps:
      - title: "Build logoscore"
        run: "nix build 'github:logos-co/logos-logoscore-cli' --out-link ./logos"
        code_block: |
          nix build 'github:logos-co/logos-logoscore-cli' --out-link ./logos
        check_file: "logos/bin/logoscore"
      - title: "Build lgpm"
        run: "nix build 'github:logos-co/logos-package-manager#cli' -o lgpm"
        check_file: "lgpm/bin/lgpm"

  - title: "Module 1 — the concurrent worker (`slow_worker_module`)"
    step: true
    text: |
      A universal C++ module whose `work(ms)` blocks for `ms`. The single thing
      that makes it concurrent is **`"concurrency": "multi"`** in its metadata.
      Under multi dispatch its methods run on worker threads, so the overlap
      counter is `std::atomic` — the author owns thread-safety.
    steps:
      - title: "The implementation — plain C++, header-only"
        text: |
          Create `slow-worker/src/slow_worker_module_impl.h` — **declarations only**.
          The generator parses this header for the method signatures, so the bodies
          live in a `.cpp` (next step); inline bodies would let the header→`.lidl`
          extraction pick up the calls *inside* them. The counters are atomics
          because `work` may run on several threads at once:
        file:
          path: slow-worker/src/slow_worker_module_impl.h
          language: cpp
          content: |
            #pragma once

            #include <atomic>
            #include <cstdint>

            #include <logos_module_context.h>  // LogosModuleContext base

            // A concurrency:"multi" worker. work(ms) blocks for ms and records the PEAK
            // number of calls running at the same instant. Multi dispatch runs the
            // methods on worker threads, so the counters are atomic.
            class SlowWorkerModuleImpl : public LogosModuleContext {
            public:
                int64_t work(int64_t ms);
                int64_t peak();

            private:
                std::atomic<int64_t> m_inFlight{0};
                std::atomic<int64_t> m_maxSeen{0};
            };
      - title: "The impl .cpp — the method bodies"
        text: |
          The bodies live in the `.cpp` (added to the module's sources via
          `extra_sources`), keeping the parsed header a clean list of signatures:
        file:
          path: slow-worker/src/slow_worker_module_impl.cpp
          language: cpp
          content: |
            #include "slow_worker_module_impl.h"

            #include <chrono>
            #include <thread>

            int64_t SlowWorkerModuleImpl::work(int64_t ms) {
                int64_t now = m_inFlight.fetch_add(1) + 1;
                int64_t prev = m_maxSeen.load();
                while (now > prev && !m_maxSeen.compare_exchange_weak(prev, now)) { /* retry */ }
                std::this_thread::sleep_for(std::chrono::milliseconds(ms > 0 ? ms : 0));
                m_inFlight.fetch_sub(1);
                return ms;
            }

            int64_t SlowWorkerModuleImpl::peak() { return m_maxSeen.load(); }
      - title: "metadata.json — note `concurrency: \"multi\"`"
        text: |
          `interface: "universal"` is the pure-C++ path; **`concurrency: "multi"`**
          turns on concurrent dispatch:
        file:
          path: slow-worker/metadata.json
          language: json
          content: |
            {
              "name": "slow_worker_module",
              "version": "1.0.0",
              "description": "A concurrency:multi worker: work(ms) blocks and records peak overlap",
              "author": "Logos Core Team",
              "type": "core",
              "interface": "universal",
              "concurrency": "multi",
              "category": "general",
              "main": "slow_worker_module_plugin",
              "dependencies": [],
              "codegen": { "impl_class": "SlowWorkerModuleImpl", "impl_header": "slow_worker_module_impl.h" },
              "nix": { "external_libraries": [], "packages": { "build": [], "runtime": [] }, "cmake": { "find_packages": [], "extra_include_dirs": [], "extra_sources": ["src/slow_worker_module_impl.cpp"] } }
            }
      - title: "CMakeLists.txt + flake.nix"
        file:
          path: slow-worker/CMakeLists.txt
          language: cmake
          content: |
            cmake_minimum_required(VERSION 3.14)
            project(SlowWorkerModulePlugin LANGUAGES CXX)
            if(DEFINED ENV{LOGOS_MODULE_BUILDER_ROOT})
                include($ENV{LOGOS_MODULE_BUILDER_ROOT}/cmake/LogosModule.cmake)
            else()
                message(FATAL_ERROR "LOGOS_MODULE_BUILDER_ROOT is not set.")
            endif()
            configure_file(${CMAKE_CURRENT_SOURCE_DIR}/metadata.json
                           ${CMAKE_CURRENT_BINARY_DIR}/metadata.json COPYONLY)
            logos_module(NAME slow_worker_module
                         SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/slow_worker_module_impl.cpp
                         INCLUDE_DIRS ${CMAKE_CURRENT_SOURCE_DIR}/src)
      - title: "flake.nix"
        file:
          path: slow-worker/flake.nix
          language: nix
          content: |
            {
              description = "concurrency:multi C++ worker";
              inputs = {
                logos-module-builder.url = "github:logos-co/logos-module-builder{release}";
              };
              outputs = inputs@{ self, logos-module-builder, ... }:
                let
                  nixpkgs = logos-module-builder.inputs.nixpkgs;
                  systems = [ "aarch64-darwin" "x86_64-darwin" "aarch64-linux" "x86_64-linux" ];
                  forAllSystems = f: nixpkgs.lib.genAttrs systems f;
                in {
                  packages = forAllSystems (system:
                    (logos-module-builder.lib.mkLogosModule {
                      src = ./.;
                      configFile = ./metadata.json;
                      flakeInputs = inputs;
                    }).packages.${system});
                };
            }
      - title: "Build it"
        text: |
          The `{release}` overrides point the builder's C++ SDK, Qt glue, and
          protocol at the commits under test, so the concurrent-dispatch codegen +
          runtime are the ones exercised:
        run: "sh -c 'cd slow-worker && printf \"result\\nresult-*\\n\" > .gitignore && git init -q && git add -A && nix flake update && git add flake.lock && nix build .#lgx -o worker-lgx --override-input logos-module-builder \"github:logos-co/logos-module-builder{release}\" --override-input logos-module-builder/logos-cpp-sdk \"github:logos-co/logos-cpp-sdk{release}\" --override-input logos-module-builder/logos-qt-sdk \"github:logos-co/logos-qt-sdk{release}\" --override-input logos-module-builder/logos-protocol \"github:logos-co/logos-protocol{release}\"'"
        code_block: |
          cd slow-worker
          git init && git add -A && nix flake update && git add flake.lock
          nix build .#lgx -o worker-lgx \
            --override-input logos-module-builder/logos-cpp-sdk 'github:logos-co/logos-cpp-sdk' \
            --override-input logos-module-builder/logos-qt-sdk 'github:logos-co/logos-qt-sdk' \
            --override-input logos-module-builder/logos-protocol 'github:logos-co/logos-protocol'
        check_file: "slow-worker/worker-lgx"

  - title: "Module 2 — the single-threaded driver (`fanout_driver_module`)"
    step: true
    text: |
      An ordinary module (no `concurrency` field → single-threaded) that depends on
      the worker. `fanOut` fires N `workAsync` calls without waiting between them,
      so all N reach the worker before any completes.
    steps:
      - title: "The impl header — declarations only (the generator parses this)"
        file:
          path: fanout-driver/src/fanout_driver_module_impl.h
          language: cpp
          content: |
            #pragma once

            #include <cstdint>
            #include <logos_module_context.h>  // LogosModuleContext base + modules()

            // A single-threaded driver that drives the multi worker concurrently:
            // fanOut fires N workAsync() calls back-to-back.
            class FanoutDriverModuleImpl : public LogosModuleContext {
            public:
                // Fire `n` concurrent work(ms) calls at the worker; returns `n`.
                int64_t fanOut(int64_t n, int64_t ms);
                // Read the worker's observed peak overlap, THROUGH this driver.
                int64_t peak();
            };
      - title: "The impl .cpp — the cross-module fan-out"
        text: |
          The `.cpp` includes the generated `logos_sdk.h` (which defines
          `LogosModules`), so the cross-module calls live here, not in the header
          the generator parses. The async client returns immediately:
        file:
          path: fanout-driver/src/fanout_driver_module_impl.cpp
          language: cpp
          content: |
            #include "fanout_driver_module_impl.h"

            // Generated at build time by logos-cpp-generator. Defines LogosModules with
            // a typed accessor per dependency — here slow_worker_module.
            #include "logos_sdk.h"

            int64_t FanoutDriverModuleImpl::fanOut(int64_t n, int64_t ms)
            {
                for (int64_t i = 0; i < n; ++i) {
                    // Async typed client: fire and return immediately. We don't need the
                    // result here — the point is that all N reach the worker together.
                    modules().slow_worker_module.workAsync(ms, [](int64_t) {});
                }
                return n;
            }

            int64_t FanoutDriverModuleImpl::peak()
            {
                // Read peak via a SYNC call to the worker. The worker is `multi`, so its
                // reply is deferred (a pending marker + completion event); this module's
                // generated client awaits it transparently and returns the real number.
                // (A protocol-0.1 caller — e.g. the logoscore CLI calling the worker
                // directly — would instead see the raw pending marker, so we read peak
                // through the driver.)
                return modules().slow_worker_module.peak();
            }
      - title: "metadata.json — depends on the worker, no concurrency field"
        file:
          path: fanout-driver/metadata.json
          language: json
          content: |
            {
              "name": "fanout_driver_module",
              "version": "1.0.0",
              "description": "Single-threaded driver: fans out concurrent calls to slow_worker_module",
              "author": "Logos Core Team",
              "type": "core",
              "interface": "universal",
              "category": "general",
              "main": "fanout_driver_module_plugin",
              "dependencies": ["slow_worker_module"],
              "codegen": { "impl_class": "FanoutDriverModuleImpl", "impl_header": "fanout_driver_module_impl.h" },
              "nix": { "external_libraries": [], "packages": { "build": [], "runtime": [] }, "cmake": { "find_packages": [], "extra_include_dirs": [], "extra_sources": ["src/fanout_driver_module_impl.cpp"] } }
            }
      - title: "CMakeLists.txt + flake.nix"
        file:
          path: fanout-driver/CMakeLists.txt
          language: cmake
          content: |
            cmake_minimum_required(VERSION 3.14)
            project(FanoutDriverModulePlugin LANGUAGES CXX)
            if(DEFINED ENV{LOGOS_MODULE_BUILDER_ROOT})
                include($ENV{LOGOS_MODULE_BUILDER_ROOT}/cmake/LogosModule.cmake)
            else()
                message(FATAL_ERROR "LOGOS_MODULE_BUILDER_ROOT is not set.")
            endif()
            configure_file(${CMAKE_CURRENT_SOURCE_DIR}/metadata.json
                           ${CMAKE_CURRENT_BINARY_DIR}/metadata.json COPYONLY)
            logos_module(NAME fanout_driver_module
                         SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/src/fanout_driver_module_impl.cpp
                         INCLUDE_DIRS ${CMAKE_CURRENT_SOURCE_DIR}/src)
      - title: "flake.nix — inputs the worker (its published contract drives the typed client)"
        file:
          path: fanout-driver/flake.nix
          language: nix
          content: |
            {
              description = "single-threaded C++ driver over the multi worker";
              inputs = {
                logos-module-builder.url = "github:logos-co/logos-module-builder{release}";
                # The worker's flake — its published .lidl drives modules().slow_worker_module.
                # Placeholder; locked to the local checkout at build time via --override-input.
                slow_worker_module.url = "path:/path/to/slow-worker";
              };
              outputs = inputs@{ self, logos-module-builder, ... }:
                let
                  nixpkgs = logos-module-builder.inputs.nixpkgs;
                  systems = [ "aarch64-darwin" "x86_64-darwin" "aarch64-linux" "x86_64-linux" ];
                  forAllSystems = f: nixpkgs.lib.genAttrs systems f;
                in {
                  packages = forAllSystems (system:
                    (logos-module-builder.lib.mkLogosModule {
                      src = ./.;
                      configFile = ./metadata.json;
                      flakeInputs = inputs;
                    }).packages.${system});
                };
            }
      - title: "Build it"
        run: "sh -c 'cd fanout-driver && printf \"result\\nresult-*\\n\" > .gitignore && git init -q && git add -A && nix flake update --override-input slow_worker_module path:$PWD/../slow-worker && git add flake.lock && nix build .#lgx -o driver-lgx --override-input slow_worker_module path:$PWD/../slow-worker --override-input logos-module-builder \"github:logos-co/logos-module-builder{release}\" --override-input logos-module-builder/logos-cpp-sdk \"github:logos-co/logos-cpp-sdk{release}\" --override-input logos-module-builder/logos-qt-sdk \"github:logos-co/logos-qt-sdk{release}\" --override-input logos-module-builder/logos-protocol \"github:logos-co/logos-protocol{release}\" --override-input slow_worker_module/logos-module-builder/logos-cpp-sdk \"github:logos-co/logos-cpp-sdk{release}\" --override-input slow_worker_module/logos-module-builder/logos-qt-sdk \"github:logos-co/logos-qt-sdk{release}\" --override-input slow_worker_module/logos-module-builder/logos-protocol \"github:logos-co/logos-protocol{release}\"'"
        code_block: |
          cd fanout-driver
          git init && git add -A
          nix flake update --override-input slow_worker_module path:$PWD/../slow-worker
          git add flake.lock
          nix build .#lgx -o driver-lgx \
            --override-input slow_worker_module path:$PWD/../slow-worker \
            --override-input logos-module-builder/logos-cpp-sdk 'github:logos-co/logos-cpp-sdk' \
            --override-input logos-module-builder/logos-qt-sdk 'github:logos-co/logos-qt-sdk' \
            --override-input logos-module-builder/logos-protocol 'github:logos-co/logos-protocol' \
            --override-input slow_worker_module/logos-module-builder/logos-cpp-sdk 'github:logos-co/logos-cpp-sdk'
        check_file: "fanout-driver/driver-lgx"

  - title: "Run it: the single driver overlaps the multi worker"
    step: true
    steps:
      - title: "Install the modules"
        run: |
          mkdir -p modules
          cp -RL ./logos/modules/. ./modules/
          ./lgpm/bin/lgpm --modules-dir ./modules --allow-unsigned install --file slow-worker/worker-lgx/*.lgx
          ./lgpm/bin/lgpm --modules-dir ./modules --allow-unsigned install --file fanout-driver/driver-lgx/*.lgx
        expect_contains:
          - "Installed to:"
      - title: "Start the daemon"
        run: "sh -c './logos/bin/logoscore -D -m ./modules > logs.txt 2>&1 &'"
        code_block: "logoscore -D -m ./modules > logs.txt &"
      - run: "sleep 6"
      - title: "Load the modules"
        run: "./logos/bin/logoscore load-module fanout_driver_module"
        code_block: "logoscore load-module fanout_driver_module"
        expect_contains:
          - "fanout_driver_module"
          - "slow_worker_module"
      - title: "Fan out 4 concurrent calls"
        text: "`fanOut(4, 500)` fires four `work(500)` calls at the worker without waiting, and returns `4` immediately:"
        run: "./logos/bin/logoscore call fanout_driver_module fanOut 4 500"
        code_block: "logoscore call fanout_driver_module fanOut 4 500"
        expect_contains:
          - '"result":4'
      - run: "sleep 2"
      - title: "The worker ran them concurrently (peak == 4)"
        text: |
          `peak` is the most calls the worker saw running at once. Because the
          worker is `concurrency: "multi"`, all four overlapped, so the peak is
          **4**. Had the worker been `"single"`, the peak would be **1**. We read it
          **through the driver** (`fanout_driver_module.peak`), which forwards to the
          worker: the driver is a protocol-0.2 consumer, so it awaits the multi
          worker's deferred reply and returns the real number:
        run: "./logos/bin/logoscore call fanout_driver_module peak"
        code_block: "logoscore call fanout_driver_module peak"
        expect_contains:
          - '"result":4'
      - title: "Stop the daemon"
        run: "./logos/bin/logoscore stop"
        code_block: "logoscore stop"
      - run: "sleep 2"
      - run: "./logos/bin/logoscore status || true"
        code_block: "logoscore status"
        expect_contains:
          - '"status":"not_running"'