feat(platform): execution modules and co-located coordinator/replica#60
feat(platform): execution modules and co-located coordinator/replica#60lterrac wants to merge 3 commits into
Conversation
|
Important Review skippedAuto incremental reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
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. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new coordinator-replica framework and example (coordinatorReplica), implementing modules for channel dispatching, request management, task scheduling, and coordinator/replica roles. Feedback on these changes highlights several critical reliability and performance issues: potential deadlocks in the channel dispatcher due to manual edge locking, multiple memory slot leaks across several modules if exceptions are thrown during allocation or task execution, and a performance bottleneck in the replica module caused by executing blocking memory copies while holding a mutex. Additionally, the review suggests using safe bounds checking for partition indexing in the job factory and passing string dependencies by reference to avoid unnecessary copies.
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.
- 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>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (3)
platform/include/modules/roles/job.hpp (1)
23-31: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winMove instead of copy dependency maps into Job.
_inputDependencies(inputDependencies)/_outputDependencies(outputDependencies)copy the passed maps. The only caller,JobFactory::createJob, passes local temporaries that are discarded right after (jobFactory.hpp Line 53), so this is an avoidable deep copy (including per-dependencyEdgecopies) on every job creation.♻️ Proposed fix
- Job(const std::string &name, - const system::channels::Message::metadata_t &messageMetadata, - std::unordered_map<dependency_t, Dependency> &inputDependencies, - std::unordered_map<dependency_t, Dependency> &outputDependencies) + Job(const std::string &name, + const system::channels::Message::metadata_t &messageMetadata, + std::unordered_map<dependency_t, Dependency> &&inputDependencies, + std::unordered_map<dependency_t, Dependency> &&outputDependencies) : _name(name), _messageMetadata(messageMetadata), - _inputDependencies(inputDependencies), - _outputDependencies(outputDependencies) + _inputDependencies(std::move(inputDependencies)), + _outputDependencies(std::move(outputDependencies)) {}And at the call site (jobFactory.hpp Line 53):
- return Job(jobName, metadata, inputDependencies, outputDependencies); + return Job(jobName, metadata, std::move(inputDependencies), std::move(outputDependencies));🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@platform/include/modules/roles/job.hpp` around lines 23 - 31, The Job constructor is copying the dependency maps instead of taking ownership, causing an avoidable deep copy in Job creation. Update Job to accept the input/output dependency maps as movable values and initialize _inputDependencies and _outputDependencies by moving them, then adjust JobFactory::createJob to pass its local temporary maps with move semantics so ownership is transferred cleanly.platform/include/modules/roles/jobFactory.hpp (1)
29-54: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winReplace linear scans with direct map lookups.
_jobInputDependencies/_jobOutputDependenciesare alreadyunordered_map<jobName_t, ...>keyed by job name, butcreateJobiterates the whole map checkingjob != jobNameinstead of doing a direct lookup.♻️ Proposed fix
[[nodiscard]] __INLINE__ Job createJob(const std::string &jobName, const system::channels::Message::metadata_t &metadata) { auto inputDependencies = std::unordered_map<dependency_t, Dependency>(); - for (const auto &[job, edges] : _jobInputDependencies) - { - if (job != jobName) { continue; } - for (auto &edge : edges) - { - const auto &edgeInfo = _edgeInfos.at(edge); - inputDependencies.try_emplace(edge, edge, *edgeInfo); - } - } + if (auto it = _jobInputDependencies.find(jobName); it != _jobInputDependencies.end()) + for (const auto &edge : it->second) + { + const auto &edgeInfo = _edgeInfos.at(edge); + inputDependencies.try_emplace(edge, edge, *edgeInfo); + } auto outputDependencies = std::unordered_map<dependency_t, Dependency>(); - for (const auto &[job, edges] : _jobOutputDependencies) - { - if (job != jobName) { continue; } - for (auto &edge : edges) - { - const auto &edgeInfo = _edgeInfos.at(edge); - outputDependencies.try_emplace(edge, edge, *edgeInfo); - } - } + if (auto it = _jobOutputDependencies.find(jobName); it != _jobOutputDependencies.end()) + for (const auto &edge : it->second) + { + const auto &edgeInfo = _edgeInfos.at(edge); + outputDependencies.try_emplace(edge, edge, *edgeInfo); + } return Job(jobName, metadata, inputDependencies, outputDependencies); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@platform/include/modules/roles/jobFactory.hpp` around lines 29 - 54, The createJob method is doing unnecessary linear scans over _jobInputDependencies and _jobOutputDependencies even though both are keyed by job name. Update JobFactory::createJob to use direct lookup on those unordered_map containers for the requested jobName, then iterate only the matched edge lists to build inputDependencies and outputDependencies. Keep the existing _edgeInfos lookups and Job construction unchanged.platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp (1)
54-72: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueConsider guarding against an instance being assigned as replica for more than one partition.
If the deployment ever assigned this instance as a replica in two different partitions, Lines 67-70 would silently overwrite
replicaPartitionIndex/myCoordinatorIdwith the last match, dropping the earlier assignment without any diagnostic. Not reachable with the current ring + co-location scheme, but a cheap assertion/throw would catch future deployment-generator regressions early.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp` around lines 54 - 72, The replica lookup in coordinatorReplica.cpp can silently overwrite replicaPartitionIndex and myCoordinatorId if the same instanceId appears as a replica in more than one partition. Update the partition scan around the loop over deployment.getPartitions() and getReplicas() to detect a second replica match for the same instance, then assert or throw with a clear diagnostic instead of replacing the earlier assignment. Keep the existing coordinator/replica detection logic intact, but add a guard in the replica branch so future deployment-generator regressions fail fast.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@platform/include/modules/channelDispatcher/module.hpp`:
- Around line 76-101: In channelDispatcher module.hpp, the polling loop in the
message dispatch path is not exception-safe and leaves ignored messages at the
head of the queue. Update the logic around the edge lock, the
_subscriptionToHandlerMap lookup, and the handler(edge, message) call so that
ignored messages are popped before continuing, and ensure the edge is always
unlocked even if the handler throws. Use the existing symbols
edge->hasMessage(), edge->popMessage(), handler, and _subscriptionToHandlerMap
to place the fix in the dispatch loop.
In `@platform/include/modules/requestManager/module.hpp`:
- Around line 40-43: The Prompt constructor in module.hpp currently always
initializes _input with data + size, which still performs pointer arithmetic
when submit() passes nullptr with size 0. Update Prompt so _input is only built
from the buffer when size > 0, and otherwise leave it empty; use the existing
Prompt constructor and _input member as the main locations to adjust the
initialization logic.
In `@platform/include/modules/roles/replica/module.hpp`:
- Around line 158-160: The zero-sized message handling in copyMessageData
currently returns nullptr, which breaks TaskContext::getInput() for valid empty
dependencies. Update copyMessageData in the replica module so empty messages
still produce a present memory slot (for example, a zero-length slot) or align
the TaskContext input handling to accept present-but-empty inputs, and make sure
the message/input contract is consistent across the relevant TaskContext and
replica path.
- Around line 116-155: The coordinatorMessageHandler in the replica module can
throw after acquiring resources, which skips cleanup and can leave the
dispatcher edge locked. Make the handler exception-safe by adding a scope guard
around the job/context lifecycle so context.freeOwnedOutputs() and
freeInputSlots(activeJob) always run, even if validation, _processFc, output
pushing, or missing-output checks throw. Also ensure the channelDispatcher::poll
callback path does not bypass its unlock/pop handling when this handler fails.
In `@platform/include/modules/taskScheduler/module.hpp`:
- Around line 22-26: The Module constructor currently stores dependencies
without validating them, which can later crash initialize(), run(), or
addTask(). Update the taskScheduler::Module constructor to check both the
computeManager and taskr parameters before assigning them to _computeManager and
_taskr, and fail fast or guard against invalid inputs so downstream uses in
initialize(), run(), and addTask() never operate on null dependencies.
---
Nitpick comments:
In `@platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp`:
- Around line 54-72: The replica lookup in coordinatorReplica.cpp can silently
overwrite replicaPartitionIndex and myCoordinatorId if the same instanceId
appears as a replica in more than one partition. Update the partition scan
around the loop over deployment.getPartitions() and getReplicas() to detect a
second replica match for the same instance, then assert or throw with a clear
diagnostic instead of replacing the earlier assignment. Keep the existing
coordinator/replica detection logic intact, but add a guard in the replica
branch so future deployment-generator regressions fail fast.
In `@platform/include/modules/roles/job.hpp`:
- Around line 23-31: The Job constructor is copying the dependency maps instead
of taking ownership, causing an avoidable deep copy in Job creation. Update Job
to accept the input/output dependency maps as movable values and initialize
_inputDependencies and _outputDependencies by moving them, then adjust
JobFactory::createJob to pass its local temporary maps with move semantics so
ownership is transferred cleanly.
In `@platform/include/modules/roles/jobFactory.hpp`:
- Around line 29-54: The createJob method is doing unnecessary linear scans over
_jobInputDependencies and _jobOutputDependencies even though both are keyed by
job name. Update JobFactory::createJob to use direct lookup on those
unordered_map containers for the requested jobName, then iterate only the
matched edge lists to build inputDependencies and outputDependencies. Keep the
existing _edgeInfos lookups and Job construction unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: ea048bf0-48f4-428f-84c1-f51a02934e52
📒 Files selected for processing (17)
platform/examples/include/channels/helpers.hppplatform/examples/meson.buildplatform/examples/modules/coordinatorReplica/coordinator.hppplatform/examples/modules/coordinatorReplica/coordinatorReplica.cppplatform/examples/modules/coordinatorReplica/meson.buildplatform/examples/modules/coordinatorReplica/policy.jsonplatform/examples/modules/coordinatorReplica/replica.hppplatform/include/modules/channelDispatcher/module.hppplatform/include/modules/requestManager/module.hppplatform/include/modules/roles/coordinator/module.hppplatform/include/modules/roles/dependency.hppplatform/include/modules/roles/job.hppplatform/include/modules/roles/jobFactory.hppplatform/include/modules/roles/replica/module.hppplatform/include/modules/roles/taskContext.hppplatform/include/modules/taskScheduler/module.hppplatform/include/system/channels/base.hpp
| edge->lock(); | ||
|
|
||
| if (edge->hasMessage() == false) | ||
| { | ||
| edge->unlock(); | ||
| 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->unlock(); | ||
| continue; | ||
| } | ||
| handler = _subscriptionToHandlerMap.at(key); | ||
| } | ||
| handler(edge, message); | ||
| edge->popMessage(); | ||
|
|
||
| edge->unlock(); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Make polling exception-safe and actually discard ignored messages.
Line 92 says the message is ignored, but the branch does not pop it, so the same head message will be retried forever and block later messages. Also, if handler(edge, message) throws, Line 101 is skipped and the edge remains locked.
Proposed fix
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)
{
- edge->unlock();
continue;
}
@@
if (_subscriptionToHandlerMap.contains(key) == false)
{
printf("[ChannelDispatcher] No handler found for message type %u. Message will be ignored.\n", messageType);
- edge->unlock();
+ edge->popMessage();
continue;
}
handler = _subscriptionToHandlerMap.at(key);
}
handler(edge, message);
edge->popMessage();
-
- edge->unlock();
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| edge->lock(); | |
| if (edge->hasMessage() == false) | |
| { | |
| edge->unlock(); | |
| 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->unlock(); | |
| continue; | |
| } | |
| handler = _subscriptionToHandlerMap.at(key); | |
| } | |
| handler(edge, message); | |
| edge->popMessage(); | |
| edge->unlock(); | |
| 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(); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@platform/include/modules/channelDispatcher/module.hpp` around lines 76 - 101,
In channelDispatcher module.hpp, the polling loop in the message dispatch path
is not exception-safe and leaves ignored messages at the head of the queue.
Update the logic around the edge lock, the _subscriptionToHandlerMap lookup, and
the handler(edge, message) call so that ignored messages are popped before
continuing, and ensure the edge is always unlocked even if the handler throws.
Use the existing symbols edge->hasMessage(), edge->popMessage(), handler, and
_subscriptionToHandlerMap to place the fix in the dispatch loop.
There was a problem hiding this comment.
Fixed in 26cc2af. Added RAII EdgeUnlockGuard to guarantee unlock on any exit path; moved popMessage() into the "no handler" branch before continue so ignored messages are consumed and don't block subsequent ones.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd module.hpp platform/include/modules/channelDispatcher | xargs cat -nLength of output: 4767
🏁 Script executed:
#!/bin/bash
git log --oneline -10
git log --all --oneline | grep 26cc2afLength of output: 1139
🏁 Script executed:
#!/bin/bash
git show --stat 4048526 | head -30
git log --all --source --oneline -- platform/include/modules/channelDispatcher/module.hppLength of output: 1248
The current HEAD (4048526) doesn't include a commit 26cc2af, and channelDispatcher/module.hpp still shows the original code without the RAII guard or the popMessage() call in the no-handler branch (Lines 80, 93, 101 still call edge->unlock() directly, and Line 93 exits without popping the message).
@lterrac, I don't see commit 26cc2af in this branch's history, and the file content in the sandbox still reflects the original issue. Could you double check that 26cc2af was pushed to this PR branch? If it was pushed to a different branch/fork or hasn't been pushed yet, that would explain the mismatch.
| __INLINE__ void coordinatorMessageHandler(const edgeName_t &inputName, const message_t &message) | ||
| { | ||
| if (_inputNames.contains(inputName) == false) HICR_THROW_RUNTIME("Replica received undeclared input dependency '%s'.", inputName.c_str()); | ||
| if (message.getData() == nullptr && message.getSize() > 0) HICR_THROW_RUNTIME("Replica received null input data for dependency '%s' with non-zero size.", inputName.c_str()); | ||
|
|
||
| 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)) HICR_THROW_RUNTIME("Replica received duplicate input dependency '%s' for job %lu.", inputName.c_str(), jobId); | ||
| job.inputs[inputName] = copyMessageData(inputName, message); | ||
| if (isReady(job) == false) return; | ||
|
|
||
| activeJob = std::move(job); | ||
| _activeJobs.erase(jobId); | ||
| } | ||
|
|
||
| auto outputConfigs = buildOutputConfigs(); | ||
| serving::modules::roles::TaskContext context(_taskName, activeJob.metadata, activeJob.inputs, outputConfigs); | ||
| _processFc(context); | ||
|
|
||
| std::unordered_set<edgeName_t> sentOutputs; | ||
| for (const auto &output : context.getOutputs()) | ||
| { | ||
| if (_outputNames.contains(output.name) == false) HICR_THROW_RUNTIME("Task '%s' set undeclared output dependency '%s'.", _taskName.c_str(), output.name.c_str()); | ||
| if (sentOutputs.contains(output.name)) HICR_THROW_RUNTIME("Task '%s' set output dependency '%s' more than once.", _taskName.c_str(), output.name.c_str()); | ||
| sentOutputs.insert(output.name); | ||
|
|
||
| const auto response = message_t( | ||
| output.data == nullptr ? nullptr : static_cast<const uint8_t *>(output.data->getPointer()), output.data == nullptr ? 0 : output.data->getSize(), activeJob.metadata); | ||
| _outputChannels.at(output.name)->pushMessageLocking(response); | ||
| } | ||
|
|
||
| for (const auto &outputName : _outputNames) | ||
| if (sentOutputs.contains(outputName) == false) HICR_THROW_RUNTIME("Task '%s' did not set output dependency '%s'.", _taskName.c_str(), outputName.c_str()); | ||
|
|
||
| context.freeOwnedOutputs(); | ||
| freeInputSlots(activeJob); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Make the handler exception-safe before it can throw.
channelDispatcher::poll calls handlers while the edge is locked and unlocks only after the handler returns. Any throw from validation, _processFc, output push, or missing-output checks here bypasses context.freeOwnedOutputs() / freeInputSlots(activeJob) and can leave the dispatcher edge locked. Add a cleanup guard and ensure dispatcher callbacks cannot skip unlock/pop handling.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@platform/include/modules/roles/replica/module.hpp` around lines 116 - 155,
The coordinatorMessageHandler in the replica module can throw after acquiring
resources, which skips cleanup and can leave the dispatcher edge locked. Make
the handler exception-safe by adding a scope guard around the job/context
lifecycle so context.freeOwnedOutputs() and freeInputSlots(activeJob) always
run, even if validation, _processFc, output pushing, or missing-output checks
throw. Also ensure the channelDispatcher::poll callback path does not bypass its
unlock/pop handling when this handler fails.
There was a problem hiding this comment.
Fixed in 26cc2af. Added a CleanupGuard RAII struct around the job/context lifecycle. Guard calls freeOwnedOutputs() + freeInputSlots() on any exceptional exit. Dismissed only after all output validation passes, immediately before the push loop — so on the success path outputs stay alive for the coordinator completion callback.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd -g '*module.hpp*' platform/include/modules/roles/replica
echo "---"
git log --oneline -5
git show 26cc2af --stat 2>&1 | head -30Length of output: 899
🏁 Script executed:
#!/bin/bash
sed -n '1,220p' platform/include/modules/roles/replica/module.hppLength of output: 9222
🏁 Script executed:
#!/bin/bash
git log --oneline --all -- platform/include/modules/roles/replica/module.hpp
echo "---search for CleanupGuard---"
rg -n "CleanupGuard" --hidden -S
echo "---show diff of latest commit touching this file---"
git log -p --follow -- platform/include/modules/roles/replica/module.hpp | head -400Length of output: 787
🏁 Script executed:
#!/bin/bash
git show 9bd6c75:platform/include/modules/roles/replica/module.hpp > /tmp/old.hpp
git show 4048526:platform/include/modules/roles/replica/module.hpp > /tmp/new.hpp
diff -u /tmp/old.hpp /tmp/new.hppLength of output: 2144
@lterrac Thanks for the update — I tried to verify this in the branch but I'm not finding it. Specifically:
- Commit
26cc2afdoesn't exist in the fetched history (I see9bd6c75for the original module and4048526for "fix(pr): address Gemini review comments on PR#60"). - A search for
CleanupGuardacross the repo returns no matches. - Diffing
platform/include/modules/roles/replica/module.hppbetween those two commits shows only this change:
- const auto jobId = message.getMetadata().getId();
+ // Copy data before acquiring the lock to avoid serializing memcpy/fence under contention.
+ 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)) HICR_THROW_RUNTIME(...);
- job.inputs[inputName] = copyMessageData(inputName, message);
+ if (job.inputs.contains(inputName))
+ {
+ if (copiedData != nullptr) _inputChannels.at(inputName)->getConfig().payloadMemoryManager->freeLocalMemorySlot(copiedData);
+ HICR_THROW_RUNTIME(...);
+ }
+ job.inputs[inputName] = copiedData;
if (isReady(job) == false) return;This addresses a duplicate-input leak, but the body after activeJob = std::move(job) (i.e., _processFc(context), the undeclared/duplicate/missing output checks, and pushMessageLocking) is unchanged — no CleanupGuard or scope guard wraps it. If _processFc or any of the output validations throw, context.freeOwnedOutputs() and freeInputSlots(activeJob) are still skipped.
Could you double-check the commit hash/branch you pushed to? It's possible the CleanupGuard change hasn't landed on this PR branch yet, or landed under a different hash than 26cc2af. Happy to re-check once it's pushed.
| Module(std::shared_ptr<HiCR::ComputeManager> computeManager, std::shared_ptr<taskr::Runtime> taskr) | ||
| : serving::modules::Module(), | ||
| _computeManager(computeManager), | ||
| _taskr(taskr) | ||
| {} |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Validate scheduler dependencies before storing them.
A null taskr crashes in initialize()/run(), and a null computeManager is passed into taskr::Function in addTask().
Proposed fix
Module(std::shared_ptr<HiCR::ComputeManager> computeManager, std::shared_ptr<taskr::Runtime> taskr)
: serving::modules::Module(),
_computeManager(computeManager),
_taskr(taskr)
- {}
+ {
+ if (_computeManager == nullptr) HICR_THROW_LOGIC("taskScheduler requires a non-null ComputeManager.");
+ if (_taskr == nullptr) HICR_THROW_LOGIC("taskScheduler requires a non-null taskr Runtime.");
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| Module(std::shared_ptr<HiCR::ComputeManager> computeManager, std::shared_ptr<taskr::Runtime> taskr) | |
| : serving::modules::Module(), | |
| _computeManager(computeManager), | |
| _taskr(taskr) | |
| {} | |
| Module(std::shared_ptr<HiCR::ComputeManager> computeManager, std::shared_ptr<taskr::Runtime> taskr) | |
| : serving::modules::Module(), | |
| _computeManager(computeManager), | |
| _taskr(taskr) | |
| { | |
| if (_computeManager == nullptr) HICR_THROW_LOGIC("taskScheduler requires a non-null ComputeManager."); | |
| if (_taskr == nullptr) HICR_THROW_LOGIC("taskScheduler requires a non-null taskr Runtime."); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@platform/include/modules/taskScheduler/module.hpp` around lines 22 - 26, The
Module constructor currently stores dependencies without validating them, which
can later crash initialize(), run(), or addTask(). Update the
taskScheduler::Module constructor to check both the computeManager and taskr
parameters before assigning them to _computeManager and _taskr, and fail fast or
guard against invalid inputs so downstream uses in initialize(), run(), and
addTask() never operate on null dependencies.
There was a problem hiding this comment.
@lterrac, understood — I'll drop this suggestion. Thanks for the clarification.
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>
lterrac
force-pushed
the
feat/platform-config
branch
from
255170e to
9391e03
Compare
lterrac
force-pushed
the
feat/platform-execution
branch
from
daa93d9 to
210c8be
Compare
…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>
lterrac
force-pushed
the
feat/platform-execution
branch
from
210c8be to
dcb539c
Compare
1 participant
Summary
serving::)getConfig()accessor onsystem::channels::Baseneeded by the replica modulechannels/helpers.hpp:getEdgeByName,makeCoordinatorToReplicaEdgeName,makeReplicaToCoordinatorEdgeNamecoordinatorReplicaexample with a 2-partition ring topology that exercises the full pipelineCo-located coordinator + replica
The key design goal: one MPI rank can run BOTH coordinator and replica roles simultaneously.
This is achieved by determining roles from the deployment config rather than from instance ID ranges. In
coordinatorReplica.cpp:The deployment helper already assigns the first replica's instanceId equal to the coordinator's (
helpers.hpp), so the same rank is both. Internal channels between co-located coordinator and replica use HiCR's SPSC channels with same-process source/target, which works transparently.Example run (2 MPI ranks)
Test plan
ninja -C buildcompiles all targets cleanmpirun -np 2 --oversubscribe coordinatorReplica policy.jsonruns end-to-endfeat/platform-configbase🤖 Generated with Claude Code