feat(platform): execution modules and co-located coordinator/replica by lterrac · Pull Request #76 · hw-native-sys/pypto-serving

lterrac · 2026-07-13T09:46:21Z

Draft — depends on #53. Do not merge until PR #53 (feat/platform-config) is merged into main.

What this adds

Building on the configuration and channel-controller foundation from #53, this PR introduces the execution layer:

Coordinator module (include/modules/roles/coordinator/) — receives jobs, dispatches them to replicas, collects results via completion callback, exposes getRuntimePlan() for replica lifecycle snapshots
Replica module (include/modules/roles/replica/) — receives a job from the coordinator, runs the user-supplied processFc, sends the result back
Supporting roles — JobFactory, Job, TaskContext, TaskScheduler, RequestManager, Subscription
Router module (include/modules/router/) — routes incoming requests to the coordinator and forwards results back to the requester
ModuleDeploymentRunner (examples/include/runtime/) — helper that wires coordinator, replica, router, and request-manager into a multi-rank deployment from a single Deployment config
Python bridge (python/bridge.cpp) — pybind11 Bridge class: single-rank coordinator+replica, exposes submit(bytes), getResult(timeoutMs), shutdown() to Python
singlePartition benchmark (examples/modules/singlePartition/) — measures coordinator→replica overhead vs direct processFc call
coordinatorReplica example (examples/modules/coordinatorReplica/) — co-located coordinator and replica on one MPI rank
Model-support quickstart doc (docs/model-support-quickstart.md)

Dependencies

PR	Status	Required
#53 — configuration types, channelController, service	open	✅ must merge first

Test plan

PR feat(platform): add configuration types, module layer, and host control plane #53 merged into main
ninja -C build compiles clean
ninja test --suite examples passes (singlePartition, coordinatorReplica)
ninja test --suite platform passes

🤖 Generated with Claude Code

…Deployment, and service modules Ports the configuration layer (Deployment, Partition, Replica, Task, Edge, RequestManager) and three modules from the upstream reference implementation, renaming the namespace to serving throughout. Wires the new module examples into the meson build and verifies the full build is clean. - configuration/: JSON-serializable deployment graph types; Deployment.verify() checks edge connectivity, producer/consumer symmetry, and request-manager boundary constraints - channelController: reconciliation-loop module that diffs desired vs actual Input/Output channels and drives exchange/fence/initialize in one pass - broadcastDeployment: RPC-based module that distributes the serialized Deployment from the deployer instance to all workers at startup - service: thin Module wrapper around taskr::Runtime for cooperative-polling services; addService() registers external taskr::Service pointers - examples/include/: deployment helpers (readAndParseConfiguration, buildLocalChannelsFromDeployment, assignEdgeManagers) and channel helpers (createDesiredSingleLocalChannels, waitUntilReady, edge name builders) - examples/modules/: broadcastDeployment, channelController, and service example programs with policy.json fixtures and meson test targets Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- Add mutex locks to hasProducer/hasConsumer to prevent data race with the reconcile service thread modifying _actual* maps - Fix format string typo '%''' → '%s' in deployment verify() Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

The coordinator and its first replica share the same instance, so additional replicas (i > 0) each need their own distinct instance ID. The required instance count is adjusted accordingly. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- telephoneGame: call popMessage() after non-root branch consumes message - channelController: return -1 after abort() so argv[1] is never read on bad argc - service: guard memSpaces and computeResources iterators before dereferencing - partition: default _coordinatorInstanceId to 0 (safe for single-node deployments) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…ort interface docs - architecture.md: instance roles, module lifecycle, channel model, deferred scope - simpler-integration.md: replica device assignment, RuntimeBinaries, processFc contract, hot-path staging and future in-device tensor channels - model-support-interface.md: what Model Support provides vs owns, RuntimePlan future contract - README.md: updated index linking all new and existing docs Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

…ica example Ports channelDispatcher, taskScheduler, roles (coordinator, replica, dependency, job, jobFactory, taskContext), and requestManager from upstream. Adds a coordinatorReplica example that demonstrates the co-located topology where a single MPI rank runs both coordinator and replica roles simultaneously. Role detection is driven by the deployment config rather than instance ID ranges, which is what makes co-location possible: the same instance can appear as coordinator AND as its first replica in the partition list. Internal coordinator-replica channels work for co-located ranks because HiCR's SPSC channels handle same-process source/target transparently. The 2-partition ring policy (2 MPI ranks) exercises the full coordinator → replica → coordinator pipeline end-to-end. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

- replica: copy message data before acquiring _activeJobsMutex to avoid serializing blocking memcpy/fence under lock contention - jobFactory: rename instanceId parameter to partitionIndex and use .at() for bounds-checked access; instance IDs don't map 1:1 to partition indices - job: pass dependency_t by const reference to avoid string copies on lookup Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

channelDispatcher/module.hpp: - Add RAII EdgeUnlockGuard so edge->unlock() fires even if handler throws - Pop ignored messages before continuing to prevent infinite retry on same head requestManager/module.hpp: - Guard Prompt constructor against nullptr+zero-size pointer arithmetic roles/replica/module.hpp: - Add CleanupGuard RAII so context.freeOwnedOutputs() and freeInputSlots() always run on exception from _processFc or output validation; guard is dismissed only after all validation passes, before the output push loop roles/taskContext.hpp: - Return a static zero-length empty slot instead of throwing for null inputs, preserving the present-but-empty dependency contract Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

coderabbitai · 2026-07-13T09:46:28Z

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 401dbb84-3718-47f9-945a-90feef9df18f

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

🔍 Trigger review

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

_{Comment @coderabbitai help to get the list of available commands.}

gemini-code-assist

Code Review

This pull request implements the host-side control plane for distributed PyPTO Serving, introducing modules for deployment configuration, RPC-based configuration broadcasting, channel reconciliation, and TaskR service management, along with comprehensive documentation and examples. The review feedback highlights several critical areas for improvement, including a data race on replica tracking, performance bottlenecks from head-of-line blocking in job dispatching and holding channel locks during handler execution, potential memory slot leaks under exception paths across multiple modules, and a potential null pointer dereference in deployment verification.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

gemini-code-assist · 2026-07-13T09:53:21Z

+  __INLINE__ void addReplica(const instanceId_t replicaId, const inputChannelMap_t &inputChannels, const outputChannelMap_t &outputChannels)
+  {
+    if (_replicas.contains(replicaId)) HICR_THROW_LOGIC("Replica %lu already registered.", replicaId);
+    _replicas[replicaId] = ReplicaChannels{
+      .inputChannels  = inputChannels,
+      .outputChannels = outputChannels,
+    };
+    _readyReplicas.push(replicaId);
+  }


In addReplica, _readyReplicas.push(replicaId) is called without acquiring _readyReplicasMutex. Since _readyReplicas is accessed and modified concurrently by the service thread in dispatchReadyJobs and replicaResponseHandler under _readyReplicasMutex, this causes a data race and undefined behavior.

References

To prevent race conditions on shared data structures, ensure that any modifications or accesses are properly protected by their corresponding mutexes.

gemini-code-assist · 2026-07-13T09:53:21Z

+  __INLINE__ void dispatchReadyJobs()
+  {
+    std::shared_ptr<Job> job = nullptr;
+    {
+      std::lock_guard lock(_pendingJobsMutex);
+      if (_pendingJobs.empty()) return;
+      job = _pendingJobs.front();
+      _pendingJobs.pop();
+    }
+
+    if (job->isReady() == false)
+    {
+      std::lock_guard lock(_pendingJobsMutex);
+      _pendingJobs.push(job);
+      return;
+    }
+
+    instanceId_t replicaId;
+    {
+      std::lock_guard lock(_readyReplicasMutex);
+      if (_readyReplicas.empty())
+      {
+        std::lock_guard pendingLock(_pendingJobsMutex);
+        _pendingJobs.push(job);
+        return;
+      }
+      replicaId = _readyReplicas.front();
+      _readyReplicas.pop();
+    }
+
+    {
+      std::lock_guard lock(_replicaJobsMutex);
+      _replicaJobs[replicaId] = job->getMetadata().getId();
+    }
+    const auto &replica = _replicas.at(replicaId);
+    for (auto &[inputName, dependency] : job->getInputDependencies())
+    {
+      if (replica.inputChannels.contains(inputName) == false) HICR_THROW_RUNTIME("Replica %lu has no input channel for dependency '%s'.", replicaId, inputName.c_str());
+      const auto message = message_t(dependency.getDataPointer(), dependency.getDataSize(), job->getMetadata());
+      replica.inputChannels.at(inputName)->pushMessageLocking(message);
+      dependency.freeDataSlot();
+    }
+  }


The current implementation of dispatchReadyJobs only pops and processes at most one job per periodic tick. If the popped job is not ready, it is pushed back and the function returns immediately, causing head-of-line blocking where a non-ready job blocks all subsequent ready jobs in the queue. Additionally, it limits dispatch throughput to one job per tick even when multiple replicas and ready jobs are available. We should loop through the queue up to its initial size to find and dispatch all ready jobs to available replicas in a single call, while avoiding nested locks to prevent deadlocks.

gemini-code-assist · 2026-07-13T09:53:21Z

+    for (const auto &edge : _pollEdges)
+    {
+      edge->lock();
+      struct EdgeUnlockGuard
+      {
+        const std::shared_ptr<channels::Input> &edge;
+        ~EdgeUnlockGuard() { edge->unlock(); }
+      } unlockGuard{edge};
+
+      if (edge->hasMessage() == false) { continue; }
+
+      const auto       message     = edge->getMessage();
+      const auto       messageType = message.getMetadata().type;
+      const auto       key         = std::make_pair(edge.get(), messageType);
+      messageHandler_t handler;
+      {
+        std::lock_guard<std::mutex> guard(_subscriptionMutex);
+        if (_subscriptionToHandlerMap.contains(key) == false)
+        {
+          printf("[ChannelDispatcher] No handler found for message type %u. Message will be ignored.\n", messageType);
+          edge->popMessage();
+          continue;
+        }
+        handler = _subscriptionToHandlerMap.at(key);
+      }
+      handler(edge, message);
+      edge->popMessage();
+    }


The input channel edge is locked during the entire execution of handler(edge, message). Since the handler can invoke slow user-defined processing functions (such as _processFc which runs heavy LLM kernels on device), holding the channel lock for the entire duration blocks other threads or producers trying to push messages to this channel, causing a severe performance bottleneck. We should pop the message and unlock the edge before invoking the handler to allow concurrent channel operations.

gemini-code-assist · 2026-07-13T09:53:21Z

+    const auto copiedData = copyMessageData(inputName, message);
+    const auto jobId      = message.getMetadata().getId();
+    ActiveJob  activeJob;
+    {
+      std::lock_guard lock(_activeJobsMutex);
+      if (_activeJobs.contains(jobId) == false) _activeJobs[jobId] = ActiveJob{.metadata = message.getMetadata()};
+      auto &job = _activeJobs.at(jobId);
+      if (job.inputs.contains(inputName))
+      {
+        if (copiedData != nullptr) _inputChannels.at(inputName)->getConfig().payloadMemoryManager->freeLocalMemorySlot(copiedData);
+        HICR_THROW_RUNTIME("Replica received duplicate input dependency '%s' for job %lu.", inputName.c_str(), jobId);
+      }
+      job.inputs[inputName] = copiedData;
+      if (isReady(job) == false) return;
+
+      activeJob = std::move(job);
+      _activeJobs.erase(jobId);
+    }


If _activeJobs[jobId] = ... or job.inputs[inputName] = copiedData throws a std::bad_alloc exception, the allocated copiedData memory slot is leaked because it is not yet tracked by _activeJobs or any cleanup guard. We should use a local scope guard to manage copiedData and free it if an exception is thrown before it is safely stored in _activeJobs.

gemini-code-assist · 2026-07-13T09:53:21Z

+  [[nodiscard]] __INLINE__ memorySlot_t copyMessageData(const edgeName_t &inputName, const message_t &message) const
+  {
+    if (message.getSize() == 0) return nullptr;
+
+    const auto &cfg     = _inputChannels.at(inputName)->getConfig();
+    const auto  srcSlot = cfg.payloadMemoryManager->registerLocalMemorySlot(cfg.payloadMemorySpace, const_cast<uint8_t *>(message.getData()), message.getSize());
+    auto        dstSlot = cfg.payloadMemoryManager->allocateLocalMemorySlot(cfg.payloadMemorySpace, message.getSize());
+    cfg.payloadCommunicationManager->memcpy(dstSlot, 0, srcSlot, 0, message.getSize());
+    cfg.payloadCommunicationManager->fence(dstSlot, 0, 1);
+    cfg.payloadMemoryManager->deregisterLocalMemorySlot(srcSlot);
+    return dstSlot;
+  }


If memcpy, fence, or deregisterLocalMemorySlot throws an exception, the allocated dstSlot is leaked because it is not yet returned or tracked by any cleanup mechanism. We should use a local scope guard to automatically free dstSlot if an exception is thrown before the function returns.

gemini-code-assist · 2026-07-13T09:53:21Z

+  __INLINE__ void storeData(const uint8_t *srcPtr, const size_t size)
+  {
+    if (srcPtr == nullptr && size > 0) HICR_THROW_LOGIC("Dependency '%s' cannot store null data with non-zero size.", _name.c_str());
+
+    freeDataSlot();
+    if (size == 0) return;
+
+    auto edgeMemoryManager        = _edgeInfo.getPayloadMemoryManager();
+    auto edgeMemorySpace          = _edgeInfo.getPayloadMemorySpace();
+    auto edgeCommunicationManager = _edgeInfo.getPayloadCommunicationManager();
+
+    const auto srcSlot = edgeMemoryManager->registerLocalMemorySlot(edgeMemorySpace, (void *)srcPtr, size);
+    auto       dstSlot = edgeMemoryManager->allocateLocalMemorySlot(edgeMemorySpace, size);
+    edgeCommunicationManager->memcpy(dstSlot, 0, srcSlot, 0, size);
+    edgeCommunicationManager->fence(dstSlot, 0, 1);
+    edgeMemoryManager->deregisterLocalMemorySlot(srcSlot);
+
+    _data = dstSlot;
+  }


If memcpy, fence, or deregisterLocalMemorySlot throws an exception, the allocated dstSlot is leaked because it is not yet assigned to _data. We should use a local scope guard to automatically free dstSlot if an exception is thrown before it is safely stored in _data.

gemini-code-assist · 2026-07-13T09:53:21Z

+  [[nodiscard]] __INLINE__ std::shared_ptr<Prompt> submit(const uint8_t *data, const size_t size)
+  {
+    if (_hasMessageType == false) HICR_THROW_LOGIC("Request manager has no message type configured.");
+    if (_promptOutput == nullptr) HICR_THROW_LOGIC("Request manager prompt output has not been set.");
+    if (data == nullptr && size > 0) HICR_THROW_LOGIC("Cannot submit a null prompt payload with non-zero size.");
+
+    metadata_t metadata;
+    metadata.type       = _messageType;
+    metadata.groupId    = _defaultSessionId;
+    metadata.sequenceId = _nextPromptId.fetch_add(1);
+
+    auto prompt = std::make_shared<Prompt>(metadata, data, size);
+    {
+      std::lock_guard lock(_activePromptsMutex);
+      const auto      promptKey = metadata.getId();
+      if (_activePrompts.contains(promptKey)) HICR_THROW_LOGIC("Prompt %lu is already active.", promptKey);
+      _activePrompts[promptKey] = prompt;
+    }
+
+    const auto message = message_t(prompt->getInput().data(), prompt->getInput().size(), metadata);
+    _promptOutput->pushMessageLocking(message);
+    return prompt;
+  }


If _promptOutput->pushMessageLocking(message) throws an exception, the prompt remains in _activePrompts forever, leaking the shared pointer and causing potential ID collision issues. We should wrap the push operation in a try-catch block to erase the prompt from _activePrompts if the push fails.

gemini-code-assist · 2026-07-13T09:53:21Z

+  void retrieveDeployment()
+  {
+    // Send request RPC
+    _rpcEngine->requestRPC(_deployerInstanceId, __SERVING_REQUEST_DEPLOYMENT_CONFIGURATION_RPC_NAME);
+
+    // Wait for serialized information
+    auto returnValue = _rpcEngine->getReturnValue();
+
+    // Receiving raw serialized topology information from the worker
+    std::string deploymentString = (char *)returnValue->getPointer();
+
+    // Parsing serialized raw topology into a json object
+    auto deploymentJs = nlohmann::json::parse(deploymentString);
+
+    // Creating the deployment object from the json
+    _deployment = configuration::Deployment(deploymentJs);
+
+    // Freeing return value
+    _rpcEngine->getMemoryManager()->freeLocalMemorySlot(returnValue);
+  }


If nlohmann::json::parse or configuration::Deployment constructor throws an exception due to invalid JSON or corrupted transmission, the allocated returnValue memory slot is leaked because freeLocalMemorySlot is never called. We should use a local scope guard to automatically free returnValue if an exception is thrown.

gemini-code-assist · 2026-07-13T09:53:21Z

+  __INLINE__ void verify() const
+  {
+    // Getting rqeuest manager input and output edges
+    const auto &requestManagerInputEdge  = _requestManager->getInput();
+    const auto &requestManagerOutputEdge = _requestManager->getOutput();
+


If Deployment is default-constructed, _requestManager is initialized to nullptr. Calling verify() on a default-constructed deployment will dereference _requestManager and cause a crash. We should add a defensive null check for _requestManager at the beginning of verify().

lterrac and others added 8 commits July 13, 2026 10:12

gemini-code-assist Bot reviewed Jul 13, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat(platform): execution modules and co-located coordinator/replica#76

feat(platform): execution modules and co-located coordinator/replica#76
lterrac wants to merge 8 commits into
mainfrom
feat/platform-execution

lterrac commented Jul 13, 2026

Uh oh!

coderabbitai Bot commented Jul 13, 2026

Review skipped

Uh oh!

gemini-code-assist Bot left a comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

lterrac commented Jul 13, 2026

What this adds

Dependencies

Test plan

Uh oh!

coderabbitai Bot commented Jul 13, 2026

Review skipped

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 13, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant