From 350b43391c6cfabc22a261240be386ce7da5c77b Mon Sep 17 00:00:00 2001 From: Luca Date: Tue, 7 Jul 2026 15:13:51 +0200 Subject: [PATCH 1/3] feat(platform): add execution modules and co-located coordinator/replica example MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../examples/include/channels/helpers.hpp | 17 ++ platform/examples/meson.build | 1 + .../coordinatorReplica/coordinator.hpp | 159 +++++++++++ .../coordinatorReplica/coordinatorReplica.cpp | 152 ++++++++++ .../modules/coordinatorReplica/meson.build | 16 ++ .../modules/coordinatorReplica/policy.json | 53 ++++ .../modules/coordinatorReplica/replica.hpp | 86 ++++++ .../modules/channelDispatcher/module.hpp | 128 +++++++++ .../include/modules/requestManager/module.hpp | 170 ++++++++++++ .../modules/roles/coordinator/module.hpp | 259 ++++++++++++++++++ platform/include/modules/roles/dependency.hpp | 69 +++++ platform/include/modules/roles/job.hpp | 104 +++++++ platform/include/modules/roles/jobFactory.hpp | 83 ++++++ .../include/modules/roles/replica/module.hpp | 209 ++++++++++++++ .../include/modules/roles/taskContext.hpp | 105 +++++++ .../include/modules/taskScheduler/module.hpp | 67 +++++ platform/include/system/channels/base.hpp | 3 +- 17 files changed, 1680 insertions(+), 1 deletion(-) create mode 100644 platform/examples/modules/coordinatorReplica/coordinator.hpp create mode 100644 platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp create mode 100644 platform/examples/modules/coordinatorReplica/meson.build create mode 100644 platform/examples/modules/coordinatorReplica/policy.json create mode 100644 platform/examples/modules/coordinatorReplica/replica.hpp create mode 100644 platform/include/modules/channelDispatcher/module.hpp create mode 100644 platform/include/modules/requestManager/module.hpp create mode 100644 platform/include/modules/roles/coordinator/module.hpp create mode 100644 platform/include/modules/roles/dependency.hpp create mode 100644 platform/include/modules/roles/job.hpp create mode 100644 platform/include/modules/roles/jobFactory.hpp create mode 100644 platform/include/modules/roles/replica/module.hpp create mode 100644 platform/include/modules/roles/taskContext.hpp create mode 100644 platform/include/modules/taskScheduler/module.hpp diff --git a/platform/examples/include/channels/helpers.hpp b/platform/examples/include/channels/helpers.hpp index 08fd3e3..735f943 100644 --- a/platform/examples/include/channels/helpers.hpp +++ b/platform/examples/include/channels/helpers.hpp @@ -118,6 +118,23 @@ __INLINE__ void waitUntilReady(const std::shared_ptrisReady()) { std::this_thread::sleep_for(std::chrono::milliseconds(intervalMs)); } } +__INLINE__ std::shared_ptr getEdgeByName(serving::configuration::Deployment &deployment, const std::string &edgeName) +{ + for (const auto &edge : deployment.getEdges()) + if (edge->getName() == edgeName) return edge; + HICR_THROW_LOGIC("Deployment has no edge named '%s'.", edgeName.c_str()); +} + +__INLINE__ std::string makeCoordinatorToReplicaEdgeName(const std::string &edgeName, const HiCR::Instance::instanceId_t coordinatorId, const HiCR::Instance::instanceId_t replicaId) +{ + return edgeName + "-coordinator" + std::to_string(coordinatorId) + "-replica" + std::to_string(replicaId); +} + +__INLINE__ std::string makeReplicaToCoordinatorEdgeName(const std::string &edgeName, const HiCR::Instance::instanceId_t replicaId, const HiCR::Instance::instanceId_t coordinatorId) +{ + return edgeName + "-replica" + std::to_string(replicaId) + "-coordinator" + std::to_string(coordinatorId); +} + __INLINE__ serving::configuration::Edge makeInternalEdgeFromTemplate(const std::string &name, const serving::configuration::Edge &edgeTemplate) { serving::configuration::Edge edge(name, edgeTemplate.getBufferCapacity(), edgeTemplate.getBufferSize()); diff --git a/platform/examples/meson.build b/platform/examples/meson.build index 9c95df8..5943ef2 100644 --- a/platform/examples/meson.build +++ b/platform/examples/meson.build @@ -2,6 +2,7 @@ modulesDependency = declare_dependency(include_directories: include_directories( subdir('modules/broadcastDeployment') subdir('modules/channelController') +subdir('modules/coordinatorReplica') subdir('modules/service') subdir('system/engine') diff --git a/platform/examples/modules/coordinatorReplica/coordinator.hpp b/platform/examples/modules/coordinatorReplica/coordinator.hpp new file mode 100644 index 0000000..3e80368 --- /dev/null +++ b/platform/examples/modules/coordinatorReplica/coordinator.hpp @@ -0,0 +1,159 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +using instanceId_t = HiCR::Instance::instanceId_t; + +struct CoordinatorRuntime +{ + using messageType_t = serving::system::channels::Message::messageType_t; + using input_t = std::shared_ptr; + + std::shared_ptr coordinatorModule; + std::shared_ptr jobFactory; + desiredSingleLocalChannels_t graphChannels; + std::pair graphUnsubscription; + std::shared_ptr> completed = std::make_shared>(false); +}; + +__INLINE__ void submitGraphJob(const std::shared_ptr &jobFactory, + const std::shared_ptr &coordinatorModule, + const std::string &jobName, + const std::string &inputEdgeName, + const serving::system::channels::Message &message) +{ + auto job = std::make_shared(jobFactory->createJob(jobName, message.getMetadata())); + auto &dependency = job->getInputDependency(inputEdgeName); + dependency.storeData(message.getData(), message.getSize()); + dependency.setSatisfied(true); + coordinatorModule->submitJob(job); +} + +__INLINE__ void handleGraphMessage(const bool isRoot, + const std::shared_ptr> &completed, + const std::shared_ptr &jobFactory, + const std::shared_ptr &coordinatorModule, + const std::string &jobName, + const std::string &inputEdgeName, + const serving::system::channels::Message &message) +{ + if (isRoot) + { + completed->store(true); + return; + } + submitGraphJob(jobFactory, coordinatorModule, jobName, inputEdgeName, message); +} + +__INLINE__ void routeOutput(const std::shared_ptr &outputChannel, + const serving::system::channels::Message::metadata_t &metadata, + const std::vector &outputs) +{ + for (const auto &output : outputs) + { + waitUntilReady(outputChannel); + const auto message = serving::system::channels::Message( + output.data == nullptr ? nullptr : static_cast(output.data->getPointer()), output.data == nullptr ? 0 : output.data->getSize(), metadata); + outputChannel->pushMessageLocking(message); + } +} + +// partitionIndex is the index of the partition this instance coordinates. +// replicaIds lists all replica instance IDs for this partition (first may be co-located). +__INLINE__ CoordinatorRuntime coordinator(const HiCR::Instance &instance, + const size_t partitionIndex, + const std::vector &replicaIds, + serving::configuration::Deployment &deployment, + const serving::system::channels::keyBuilderFc_t &keyBuilder, + const std::shared_ptr &channelControllerModule, + const std::shared_ptr &channelDispatcherModule, + const std::shared_ptr &serviceModule, + serving::system::Engine &serving, + const serving::system::channels::Message::messageType_t messageType) +{ + const auto instanceId = instance.getId(); + const auto isRoot = instance.isRootInstance(); + + CoordinatorRuntime coordinatorRuntime; + + coordinatorRuntime.jobFactory = std::make_shared(partitionIndex, deployment); + coordinatorRuntime.coordinatorModule = std::make_shared(100); + coordinatorRuntime.coordinatorModule->addMessageType(messageType); + + serving.addModule("Coordinator", coordinatorRuntime.coordinatorModule); + serviceModule->addService("Coordinator", coordinatorRuntime.coordinatorModule->getService()); + + coordinatorRuntime.graphChannels = createDesiredSingleLocalChannels(deployment, instanceId, keyBuilder, channelControllerModule); + + const auto &task = deployment.getPartitions().at(partitionIndex)->getTasks().front(); + + // Wire internal channels to every replica in this partition. + for (const auto replicaId : replicaIds) + { + serving::modules::roles::coordinator::Module::inputChannelMap_t replicaInputChannels; + serving::modules::roles::coordinator::Module::outputChannelMap_t replicaOutputChannels; + size_t channelOffset = 0; + + for (const auto &inputName : task->getInputs()) + { + const auto edge = getEdgeByName(deployment, inputName); + const auto internalEdgeName = makeCoordinatorToReplicaEdgeName(inputName, instanceId, replicaId); + const auto internalEdge = makeInternalEdgeFromTemplate(internalEdgeName, *edge); + const auto channelId = static_cast(1000 + instanceId * 100 + channelOffset++); + replicaInputChannels[inputName] = channelControllerModule->addDesiredProducer(replicaId, channelId, internalEdge, keyBuilder).lock(); + } + channelOffset = 0; + for (const auto &outputName : task->getOutputs()) + { + const auto edge = getEdgeByName(deployment, outputName); + const auto internalEdgeName = makeReplicaToCoordinatorEdgeName(outputName, replicaId, instanceId); + const auto internalEdge = makeInternalEdgeFromTemplate(internalEdgeName, *edge); + const auto channelId = static_cast(2000 + instanceId * 100 + channelOffset++); + replicaOutputChannels[outputName] = channelControllerModule->addDesiredConsumer(replicaId, channelId, internalEdge, keyBuilder).lock(); + } + coordinatorRuntime.coordinatorModule->addReplica(replicaId, replicaInputChannels, replicaOutputChannels); + } + + const auto jobName = task->getFunctionName(); + const auto inputEdgeName = coordinatorRuntime.graphChannels.inputInfo.edge->getName(); + + serving::modules::Subscription graphSubscription( + messageType, + coordinatorRuntime.graphChannels.input, + [isRoot, completed = coordinatorRuntime.completed, jobFactory = coordinatorRuntime.jobFactory, coordinatorModule = coordinatorRuntime.coordinatorModule, jobName, inputEdgeName]( + const std::shared_ptr, const serving::system::channels::Message &message) { + handleGraphMessage(isRoot, completed, jobFactory, coordinatorModule, jobName, inputEdgeName, message); + }); + + channelDispatcherModule->subscribe(graphSubscription); + coordinatorRuntime.graphUnsubscription = {messageType, coordinatorRuntime.graphChannels.input}; + + coordinatorRuntime.coordinatorModule->setCompletionCallback( + [outputChannel = coordinatorRuntime.graphChannels.output](const serving::system::channels::Message::metadata_t &metadata, + const std::vector &outputs) { + routeOutput(outputChannel, metadata, outputs); + }); + + for (auto &subscription : coordinatorRuntime.coordinatorModule->buildSubscriptions()) { channelDispatcherModule->subscribe(subscription); } + return coordinatorRuntime; +} diff --git a/platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp b/platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp new file mode 100644 index 0000000..347dc63 --- /dev/null +++ b/platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp @@ -0,0 +1,152 @@ +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "coordinator.hpp" +#include "replica.hpp" + +constexpr size_t replicasPerPartition = 1; + +constexpr serving::system::channels::Message::messageType_t kMessageType = 100; + +int main(int argc, char *argv[]) +{ + auto runtime = makeRuntime(&argc, &argv); + const auto &instance = runtime.instanceManager->getCurrentInstance(); + const auto instanceId = instance->getId(); + const auto isRoot = instance->isRootInstance(); + auto &serving = *runtime.serving; + + if (argc != 2) + { + fprintf(stderr, "Error: Must provide config file path.\n"); + runtime.instanceManager->abort(-1); + return -1; + } + + serving::configuration::Deployment deployment; + readAndParseConfiguration(argv, deployment, runtime.instanceManager, replicasPerPartition); + assignEdgeManagers(deployment, runtime.communicationManager.get(), runtime.memoryManager.get(), runtime.bufferMemorySpace); + + // Determine roles from the deployment config. + // In the co-located topology, the same instance can be BOTH coordinator and replica. + bool isCoordinator = false; + bool isReplica = false; + size_t coordinatorPartitionIndex = 0; + size_t replicaPartitionIndex = 0; + instanceId_t myCoordinatorId = 0; + std::vector myReplicaIds; + + for (size_t i = 0; i < deployment.getPartitions().size(); i++) + { + const auto &partition = deployment.getPartitions()[i]; + if (partition->getCoordinatorInstanceId() == instanceId) + { + isCoordinator = true; + coordinatorPartitionIndex = i; + for (const auto &r : partition->getReplicas()) myReplicaIds.push_back(r->getInstanceId()); + } + for (const auto &replica : partition->getReplicas()) + { + if (replica->getInstanceId() == instanceId) + { + isReplica = true; + replicaPartitionIndex = i; + myCoordinatorId = partition->getCoordinatorInstanceId(); + } + } + } + + if (isCoordinator) printf("[Instance %lu] Coordinator (partition %zu)\n", instanceId, coordinatorPartitionIndex); + if (isReplica) printf("[Instance %lu] Replica (coordinator: %lu)\n", instanceId, myCoordinatorId); + + serving::system::channels::keyBuilderFc_t keyBuilder = defaultChannelKeyBuilder; + std::vector managerOrder = {runtime.communicationManager.get()}; + auto channelControllerModule = std::make_shared(instanceId, managerOrder); + auto channelDispatcherModule = std::make_shared(20); + auto serviceModule = std::make_shared(runtime.taskr); + + serviceModule->addService("ChannelController", channelControllerModule->getService()); + serviceModule->addService("ChannelDispatcher", channelDispatcherModule->getService()); + serving.addModule("ChannelController", channelControllerModule); + serving.addModule("ChannelDispatcher", channelDispatcherModule); + serving.addModule("Service", serviceModule); + + std::optional coordinatorRt; + std::optional replicaRt; + + if (isCoordinator) + { + coordinatorRt = coordinator( + *instance, coordinatorPartitionIndex, myReplicaIds, deployment, keyBuilder, channelControllerModule, channelDispatcherModule, serviceModule, serving, kMessageType); + } + if (isReplica) + { + replicaRt = + replica(instanceId, myCoordinatorId, replicaPartitionIndex, deployment, keyBuilder, channelControllerModule, channelDispatcherModule, serviceModule, serving, kMessageType); + } + + serving.initialize(); + serving.run(); + + if (isRoot && isCoordinator && coordinatorRt.has_value()) + { + waitUntilReady(coordinatorRt->graphChannels.output); + + const std::string payload = "telephone-start"; + serving::system::channels::Message::metadata_t md; + md.type = kMessageType; + md.groupId = 1; + md.sequenceId = 1; + serving::system::channels::Message msg(reinterpret_cast(payload.data()), payload.size(), md); + coordinatorRt->graphChannels.output->pushMessageLocking(msg); + printf("[Instance %lu] Sent: %s\n", instanceId, payload.c_str()); + + while (!coordinatorRt->completed->load()) { std::this_thread::sleep_for(std::chrono::milliseconds(50)); } + + printf("[Instance %lu] Message returned to root. Terminating.\n", instanceId); + serving.terminate(); + } + + serving.await(); + + // Cleanup + if (coordinatorRt.has_value()) + { + channelDispatcherModule->unsubscribe(coordinatorRt->graphUnsubscription.first, coordinatorRt->graphUnsubscription.second); + for (const auto &[type, input] : coordinatorRt->coordinatorModule->buildUnsubscriptions()) { channelDispatcherModule->unsubscribe(type, input); } + removeDesiredSingleLocalChannels(channelControllerModule, coordinatorRt->graphChannels); + const auto &task = deployment.getPartitions().at(coordinatorPartitionIndex)->getTasks().front(); + for (const auto replicaId : myReplicaIds) + { + for (const auto &inputName : task->getInputs()) channelControllerModule->removeDesiredProducer(makeCoordinatorToReplicaEdgeName(inputName, instanceId, replicaId)); + for (const auto &outputName : task->getOutputs()) channelControllerModule->removeDesiredConsumer(makeReplicaToCoordinatorEdgeName(outputName, replicaId, instanceId)); + } + } + + if (replicaRt.has_value()) + { + for (const auto &[type, input] : replicaRt->replicaModule->buildUnsubscriptions()) { channelDispatcherModule->unsubscribe(type, input); } + const auto &task = deployment.getPartitions().at(replicaPartitionIndex)->getTasks().front(); + for (const auto &inputName : task->getInputs()) channelControllerModule->removeDesiredConsumer(makeCoordinatorToReplicaEdgeName(inputName, myCoordinatorId, instanceId)); + for (const auto &outputName : task->getOutputs()) channelControllerModule->removeDesiredProducer(makeReplicaToCoordinatorEdgeName(outputName, instanceId, myCoordinatorId)); + } + + runtime.instanceManager->finalize(); + + return 0; +} diff --git a/platform/examples/modules/coordinatorReplica/meson.build b/platform/examples/modules/coordinatorReplica/meson.build new file mode 100644 index 0000000..66e388b --- /dev/null +++ b/platform/examples/modules/coordinatorReplica/meson.build @@ -0,0 +1,16 @@ +testSuite = ['examples', 'modules'] + +e = executable('coordinatorReplica', ['coordinatorReplica.cpp'], dependencies: [servingBuildDep, modulesDependency]) +if get_option('buildTests') + test( + 'coordinatorReplica', + mpirunExecutable, + args: [ + '-np', '2', + '--oversubscribe', e.full_path(), + meson.current_source_dir() + '/policy.json', + ], + timeout: 60, + suite: testSuite, + ) +endif diff --git a/platform/examples/modules/coordinatorReplica/policy.json b/platform/examples/modules/coordinatorReplica/policy.json new file mode 100644 index 0000000..cfe98f5 --- /dev/null +++ b/platform/examples/modules/coordinatorReplica/policy.json @@ -0,0 +1,53 @@ +{ + "Name": "Coordinator-Replica Ring", + "Settings": { + "Heartbeat": { + "Enabled": false, + "Visible": false, + "Interval": 500, + "Tolerance": 1000 + }, + "Control Buffer": { + "Capacity": 32, + "Size": 16384 + } + }, + "Request Manager": { + "Input": "", + "Output": "" + }, + "Partitions": [ + { + "Name": "P0", + "Tasks": [ + { + "Function Name": "T0", + "Inputs": ["Channel1"], + "Outputs": ["Channel0"], + "Dependencies": [] + } + ] + }, + { + "Name": "P1", + "Tasks": [ + { + "Function Name": "T1", + "Inputs": ["Channel0"], + "Outputs": ["Channel1"], + "Dependencies": [] + } + ] + } + ], + "Edges": [ + { + "Name": "Channel0", + "Buffer Size": 4096 + }, + { + "Name": "Channel1", + "Buffer Size": 4096 + } + ] +} diff --git a/platform/examples/modules/coordinatorReplica/replica.hpp b/platform/examples/modules/coordinatorReplica/replica.hpp new file mode 100644 index 0000000..f8d9d9b --- /dev/null +++ b/platform/examples/modules/coordinatorReplica/replica.hpp @@ -0,0 +1,86 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +using instanceId_t = HiCR::Instance::instanceId_t; + +struct ReplicaRuntime +{ + std::shared_ptr replicaModule; + instanceId_t coordinatorId; +}; + +__INLINE__ ReplicaRuntime replica(const instanceId_t instanceId, + const instanceId_t coordinatorId, + const size_t partitionIndex, + serving::configuration::Deployment &deployment, + const serving::system::channels::keyBuilderFc_t &keyBuilder, + const std::shared_ptr &channelControllerModule, + const std::shared_ptr &channelDispatcherModule, + const std::shared_ptr &serviceModule, + serving::system::Engine &serving, + const serving::system::channels::Message::messageType_t messageType) +{ + ReplicaRuntime replicaRuntime; + replicaRuntime.coordinatorId = coordinatorId; + + const auto &task = deployment.getPartitions().at(partitionIndex)->getTasks().front(); + + auto processFc = [instanceId, inputName = task->getInputs().front(), outputName = task->getOutputs().front()](serving::modules::roles::TaskContext &context) { + const auto input = context.getInput(inputName); + std::string text(reinterpret_cast(input->getPointer()), input->getSize()); + text += " -> R"; + text += std::to_string(instanceId); + printf("[Instance %lu] Replica processed: %s\n", instanceId, text.c_str()); + context.setOutput(outputName, text.data(), text.size()); + }; + + replicaRuntime.replicaModule = std::make_shared(task->getFunctionName(), task->getInputs(), task->getOutputs(), processFc); + replicaRuntime.replicaModule->addMessageType(messageType); + + serving.addModule("Replica", replicaRuntime.replicaModule); + + serving::modules::roles::replica::Module::inputMap_t inputChannels; + serving::modules::roles::replica::Module::outputMap_t outputChannels; + size_t channelOffset = 0; + + for (const auto &inputName : task->getInputs()) + { + const auto edge = getEdgeByName(deployment, inputName); + const auto internalEdgeName = makeCoordinatorToReplicaEdgeName(inputName, coordinatorId, instanceId); + const auto internalEdge = makeInternalEdgeFromTemplate(internalEdgeName, *edge); + const auto channelId = static_cast(1000 + coordinatorId * 100 + channelOffset++); + inputChannels[inputName] = channelControllerModule->addDesiredConsumer(coordinatorId, channelId, internalEdge, keyBuilder).lock(); + } + channelOffset = 0; + for (const auto &outputName : task->getOutputs()) + { + const auto edge = getEdgeByName(deployment, outputName); + const auto internalEdgeName = makeReplicaToCoordinatorEdgeName(outputName, instanceId, coordinatorId); + const auto internalEdge = makeInternalEdgeFromTemplate(internalEdgeName, *edge); + const auto channelId = static_cast(2000 + coordinatorId * 100 + channelOffset++); + outputChannels[outputName] = channelControllerModule->addDesiredProducer(coordinatorId, channelId, internalEdge, keyBuilder).lock(); + } + replicaRuntime.replicaModule->setCoordinator(coordinatorId, outputChannels, inputChannels); + + for (auto &subscription : replicaRuntime.replicaModule->buildSubscriptions()) { channelDispatcherModule->subscribe(subscription); } + + return replicaRuntime; +} diff --git a/platform/include/modules/channelDispatcher/module.hpp b/platform/include/modules/channelDispatcher/module.hpp new file mode 100644 index 0000000..c31aec3 --- /dev/null +++ b/platform/include/modules/channelDispatcher/module.hpp @@ -0,0 +1,128 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include +#include + +#include + +#include +#include + +#include + +namespace serving::modules::channelDispatcher +{ + +class Module final : public modules::Module +{ + public: + + Module(const size_t intervalMs) + : modules::Module(intervalMs) + {} + + ~Module() override = default; + + __INLINE__ void subscribe(const Subscription &subscription) + { + const auto &edge = subscription.getEdge(); + + std::lock_guard guard(_subscriptionMutex); + + _subscribedEdges.insert(edge); + + const auto key = std::make_pair(edge.get(), subscription.getType()); + if (_subscriptionToHandlerMap.contains(key)) HICR_THROW_LOGIC("A handler is already subscribed for this input edge and message type."); + _subscriptionToHandlerMap.insert({key, subscription.getHandler()}); + } + + __INLINE__ void unsubscribe(const channels::Message::messageType_t type, const std::shared_ptr edge) + { + std::lock_guard guard(_subscriptionMutex); + + const auto key = std::make_pair(edge.get(), type); + if (_subscriptionToHandlerMap.contains(key) == false) return; + _subscriptionToHandlerMap.erase(key); + + bool hasRemainingHandlers = false; + for (const auto &[channelMessagePair, _] : _subscriptionToHandlerMap) + if (channelMessagePair.first == edge.get()) + { + hasRemainingHandlers = true; + break; + } + if (hasRemainingHandlers == false) _subscribedEdges.erase(edge); + } + + __INLINE__ void poll() + { + // reduce contention + _pollEdges.clear(); + { + std::lock_guard guard(_subscriptionMutex); + _pollEdges.reserve(_subscribedEdges.size()); + _pollEdges.assign(_subscribedEdges.begin(), _subscribedEdges.end()); + } + + for (const auto &edge : _pollEdges) + { + 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 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(); + } + } + + void initialize() override {} + void run() override {} + void terminate() override {} + void await() override {} + void finalize() override + { + _subscribedEdges.clear(); + _subscriptionToHandlerMap.clear(); + } + + protected: + + void service() override { poll(); } + + private: + + std::mutex _subscriptionMutex; + std::set> _subscribedEdges; + std::map, messageHandler_t> _subscriptionToHandlerMap; + + // Used to reduce contention in poll() + std::vector> _pollEdges; +}; +} // namespace serving::modules::channelDispatcher diff --git a/platform/include/modules/requestManager/module.hpp b/platform/include/modules/requestManager/module.hpp new file mode 100644 index 0000000..75b753b --- /dev/null +++ b/platform/include/modules/requestManager/module.hpp @@ -0,0 +1,170 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include +#include +#include + +namespace serving::modules::roles::requestManager +{ + +class Module final : public serving::modules::Module +{ + public: + + using input_t = std::shared_ptr; + using output_t = std::shared_ptr; + using message_t = serving::system::channels::Message; + using metadata_t = serving::system::channels::Message::metadata_t; + using messageId_t = serving::system::channels::Message::messageId_t; + using messageType_t = serving::system::channels::Message::messageType_t; + using groupId_t = serving::system::channels::Message::groupId_t; + using sequenceId_t = serving::system::channels::Message::sequenceId_t; + + class Prompt final + { + public: + + Prompt(const metadata_t metadata, const uint8_t *data, const size_t size) + : _metadata(metadata), + _input(data, data + size) + {} + + [[nodiscard]] const metadata_t &getMetadata() const { return _metadata; } + [[nodiscard]] messageId_t getId() const { return _metadata.getId(); } + [[nodiscard]] const std::vector &getInput() const { return _input; } + [[nodiscard]] const std::vector &getResponse() const { return _response; } + [[nodiscard]] bool hasResponse() const { return _hasResponse.load(); } + [[nodiscard]] message_t getInputMessage() const { return message_t(_input.data(), _input.size(), _metadata); } + + private: + + friend class Module; + + void setResponse(const message_t &message) + { + if (message.getMetadata().getId() != getId()) { HICR_THROW_LOGIC("Response metadata does not match prompt metadata."); } + _response.assign(message.getData(), message.getData() + message.getSize()); + _hasResponse.store(true); + } + metadata_t _metadata; + std::vector _input; + std::vector _response; + std::atomic _hasResponse = false; + }; + + Module() + : serving::modules::Module() + {} + + __INLINE__ void addMessageType(const messageType_t messageType) + { + if (_hasMessageType) [[unlikely]] + HICR_THROW_LOGIC("Request manager message type is already configured."); + _messageType = messageType; + _hasMessageType = true; + } + + __INLINE__ void setPromptOutput(const output_t &output) + { + if (output == nullptr) [[unlikely]] + HICR_THROW_LOGIC("Request manager prompt output cannot be null."); + _promptOutput = output; + } + + __INLINE__ void setResultInput(const input_t &input) + { + if (input == nullptr) [[unlikely]] + HICR_THROW_LOGIC("Request manager result input cannot be null."); + _resultInput = input; + } + + [[nodiscard]] __INLINE__ std::shared_ptr 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(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; + } + + [[nodiscard]] __INLINE__ std::vector buildSubscriptions() + { + if (_hasMessageType == false) HICR_THROW_LOGIC("Request manager has no message type configured."); + if (_resultInput == nullptr) HICR_THROW_LOGIC("Request manager result input has not been set."); + std::vector subscriptions; + subscriptions.emplace_back(_messageType, _resultInput, [this](const input_t input, const message_t &message) { this->resultMessageHandler(input, message); }); + return subscriptions; + } + + [[nodiscard]] __INLINE__ std::vector> buildUnsubscriptions() const + { + std::vector> unsubscriptions; + if (_hasMessageType && _resultInput != nullptr) { unsubscriptions.push_back({_messageType, _resultInput}); } + return unsubscriptions; + } + + void initialize() override {} + void run() override {} + void terminate() override {} + void await() override {} + void finalize() override {} + + protected: + + void service() override {} + + private: + + __INLINE__ void resultMessageHandler(const input_t, const message_t &message) + { + if (message.getMetadata().type != _messageType) { HICR_THROW_RUNTIME("Request manager received unexpected message type %u.", message.getMetadata().type); } + + std::shared_ptr prompt; + { + std::lock_guard lock(_activePromptsMutex); + const auto promptKey = message.getMetadata().getId(); + if (_activePrompts.contains(promptKey) == false) { HICR_THROW_RUNTIME("Request manager received result for unknown prompt %lu.", promptKey); } + prompt = _activePrompts.at(promptKey); + _activePrompts.erase(promptKey); + } + prompt->setResponse(message); + } + + static constexpr groupId_t _defaultSessionId = 0; + output_t _promptOutput = nullptr; + input_t _resultInput = nullptr; + messageType_t _messageType = 0; + bool _hasMessageType = false; + std::atomic _nextPromptId = 1; + std::mutex _activePromptsMutex; + std::unordered_map> _activePrompts; +}; +} // namespace serving::modules::roles::requestManager diff --git a/platform/include/modules/roles/coordinator/module.hpp b/platform/include/modules/roles/coordinator/module.hpp new file mode 100644 index 0000000..1f2d41b --- /dev/null +++ b/platform/include/modules/roles/coordinator/module.hpp @@ -0,0 +1,259 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "../job.hpp" +#include "../jobFactory.hpp" + +namespace serving::modules::roles::coordinator +{ + +class Module final : public modules::Module +{ + public: + + using input_t = std::shared_ptr; + using output_t = std::shared_ptr; + using message_t = serving::system::channels::Message; + using metadata_t = serving::system::channels::Message::metadata_t; + using messageId_t = serving::system::channels::Message::messageId_t; + using messageType_t = serving::system::channels::Message::messageType_t; + using instanceId_t = HiCR::Instance::instanceId_t; + using jobName_t = serving::modules::roles::JobFactory::jobName_t; + using edgeName_t = std::string; + + struct JobOutput + { + edgeName_t name; + std::shared_ptr data; + }; + + using completionCallback_t = std::function &outputs)>; + + Module(const size_t intervalMs) + : modules::Module(intervalMs) + {} + + using inputChannelMap_t = std::unordered_map; + using outputChannelMap_t = std::unordered_map; + + __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); + } + + __INLINE__ void addMessageType(const messageType_t messageType) { _messageTypes.insert(messageType); } + + __INLINE__ void setCompletionCallback(completionCallback_t callback) { _completionCallback = callback; } + + __INLINE__ void submitJob(const std::shared_ptr &job) + { + const auto jobId = job->getMetadata().getId(); + { + std::lock_guard lock(_jobsMutex); + if (_jobs.contains(jobId)) { return; } + _jobs[jobId] = job; + } + { + std::lock_guard lock(_pendingJobsMutex); + _pendingJobs.push(job); + } + } + + [[nodiscard]] __INLINE__ std::vector buildSubscriptions() + { + std::vector subscriptions; + size_t subscriptionCount = 0; + for (const auto &[_, replica] : _replicas) subscriptionCount += replica.outputChannels.size() * _messageTypes.size(); + subscriptions.reserve(subscriptionCount); + for (const auto &[replicaId, replica] : _replicas) + { + for (const auto &[outputName, readChannel] : replica.outputChannels) + { + for (const auto messageType : _messageTypes) + subscriptions.emplace_back( + messageType, readChannel, [this, replicaId, outputName](const input_t, const message_t &message) { this->replicaResponseHandler(replicaId, outputName, message); }); + } + } + return subscriptions; + } + + [[nodiscard]] __INLINE__ std::vector> buildUnsubscriptions() const + { + std::vector> unsubscriptions; + size_t unsubscriptionCount = 0; + for (const auto &[_, replica] : _replicas) unsubscriptionCount += replica.outputChannels.size() * _messageTypes.size(); + unsubscriptions.reserve(unsubscriptionCount); + for (const auto &[_, replica] : _replicas) + { + for (const auto &[_, readChannel] : replica.outputChannels) + { + for (const auto &messageType : _messageTypes) { unsubscriptions.push_back({messageType, readChannel}); } + } + } + return unsubscriptions; + } + + void initialize() override + { + if (_messageTypes.empty()) HICR_THROW_LOGIC("Coordinator has no message types configured."); + if (_replicas.empty()) HICR_THROW_LOGIC("Coordinator has no replicas."); + } + + void run() override {} + void terminate() override {} + void await() override {} + void finalize() override {} + + protected: + + void service() override { dispatchReadyJobs(); } + + private: + + struct ReplicaChannels + { + inputChannelMap_t inputChannels; + outputChannelMap_t outputChannels; + }; + + __INLINE__ void dispatchReadyJobs() + { + std::shared_ptr 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(); + } + } + + __INLINE__ void replicaResponseHandler(const instanceId_t replicaId, const edgeName_t &outputName, const message_t &message) + { + const auto jobId = message.getMetadata().getId(); + + { + std::lock_guard lock(_replicaJobsMutex); + if (!_replicaJobs.contains(replicaId)) { HICR_THROW_RUNTIME("Received response from replica %lu with no assigned job.", replicaId); } + if (_replicaJobs.at(replicaId) != jobId) + { + HICR_THROW_RUNTIME("Received response from replica %lu for job %lu, but expected job %lu.", replicaId, jobId, _replicaJobs.at(replicaId)); + } + } + + std::shared_ptr job = nullptr; + { + std::lock_guard lock(_jobsMutex); + job = _jobs.at(jobId); + } + + auto &dependency = job->getOutputDependency(outputName); + if (dependency.isSatisfied()) HICR_THROW_RUNTIME("Output dependency '%s' for job %lu is already satisfied.", outputName.c_str(), jobId); + dependency.storeData(message.getData(), message.getSize()); + dependency.setSatisfied(true); + + if (job->isFinished() == false) return; + + std::vector outputs; + outputs.reserve(job->getOutputDependencies().size()); + for (auto &[_, outputDependency] : job->getOutputDependencies()) + { + outputs.push_back(JobOutput{ + .name = outputDependency.getName(), + .data = outputDependency.getData(), + }); + } + + { + std::lock_guard lock(_jobsMutex); + _jobs.erase(jobId); + } + + { + std::lock_guard lock(_replicaJobsMutex); + _replicaJobs.erase(replicaId); + } + + { + std::lock_guard lock(_readyReplicasMutex); + _readyReplicas.push(replicaId); + } + if (_completionCallback != nullptr) { _completionCallback(job->getMetadata(), outputs); } + + for (auto &[_, outputDependency] : job->getOutputDependencies()) { outputDependency.freeDataSlot(); } + } + + std::unordered_map _replicas; + + std::unordered_set _messageTypes; + + std::unordered_map> _jobs; + std::mutex _jobsMutex; + + std::queue> _pendingJobs; + std::mutex _pendingJobsMutex; + + std::queue _readyReplicas; + std::mutex _readyReplicasMutex; + + std::unordered_map _replicaJobs; + std::mutex _replicaJobsMutex; + + completionCallback_t _completionCallback; +}; +} // namespace serving::modules::roles::coordinator diff --git a/platform/include/modules/roles/dependency.hpp b/platform/include/modules/roles/dependency.hpp new file mode 100644 index 0000000..d633d8d --- /dev/null +++ b/platform/include/modules/roles/dependency.hpp @@ -0,0 +1,69 @@ +#pragma once + +#include + +#include +#include + +#include + +namespace serving::modules::roles +{ + +class Dependency +{ + public: + + Dependency() = delete; + ~Dependency() = default; + + Dependency(const std::string &name, const configuration::Edge &edgeInfo) + : _name(name), + _edgeInfo(edgeInfo) + {} + + [[nodiscard]] __INLINE__ bool isSatisfied() const { return _isSatisfied; } + __INLINE__ void setSatisfied(const bool satisfied = true) { _isSatisfied = satisfied; } + + [[nodiscard]] __INLINE__ const std::shared_ptr getData() const { return _data; } + [[nodiscard]] __INLINE__ bool hasData() const { return _data != nullptr; } + [[nodiscard]] __INLINE__ size_t getDataSize() const { return hasData() ? _data->getSize() : 0; } + [[nodiscard]] __INLINE__ const uint8_t *getDataPointer() const { return hasData() ? static_cast(_data->getPointer()) : nullptr; } + + __INLINE__ void freeDataSlot() + { + if (_data == nullptr) return; + _edgeInfo.getPayloadMemoryManager()->freeLocalMemorySlot(_data); + _data = nullptr; + } + + [[nodiscard]] __INLINE__ const std::string &getName() const { return _name; } + + __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; + } + + private: + + const std::string _name; + const configuration::Edge _edgeInfo; + std::shared_ptr _data = nullptr; + bool _isSatisfied = false; +}; +} // namespace serving::modules::roles diff --git a/platform/include/modules/roles/job.hpp b/platform/include/modules/roles/job.hpp new file mode 100644 index 0000000..736a192 --- /dev/null +++ b/platform/include/modules/roles/job.hpp @@ -0,0 +1,104 @@ +#pragma once + +#include +#include +#include + +#include + +#include +#include + +#include "./dependency.hpp" + +namespace serving::modules::roles +{ + +using dependency_t = std::string; + +class Job +{ + public: + + Job(const std::string &name, + const system::channels::Message::metadata_t &messageMetadata, + std::unordered_map &inputDependencies, + std::unordered_map &outputDependencies) + : _name(name), + _messageMetadata(messageMetadata), + _inputDependencies(inputDependencies), + _outputDependencies(outputDependencies) + {} + + ~Job() = default; + + [[nodiscard]] __INLINE__ Dependency &getInputDependency(const dependency_t dependencyName) { return _inputDependencies.at(dependencyName); } + [[nodiscard]] __INLINE__ Dependency &getOutputDependency(const dependency_t dependencyName) { return _outputDependencies.at(dependencyName); } + [[nodiscard]] __INLINE__ std::unordered_map &getInputDependencies() { return _inputDependencies; } + [[nodiscard]] __INLINE__ std::unordered_map &getOutputDependencies() { return _outputDependencies; } + [[nodiscard]] __INLINE__ const std::unordered_map &getInputDependencies() const { return _inputDependencies; } + [[nodiscard]] __INLINE__ const std::unordered_map &getOutputDependencies() const { return _outputDependencies; } + [[nodiscard]] __INLINE__ const system::channels::Message::metadata_t &getMetadata() const { return _messageMetadata; } + + [[nodiscard]] __INLINE__ bool isReady() const + { + for (const auto &[_, dependency] : _inputDependencies) + { + if (dependency.isSatisfied() == false) return false; + } + return true; + } + + [[nodiscard]] __INLINE__ bool isFinished() const + { + for (const auto &[_, dependency] : _outputDependencies) + { + if (dependency.isSatisfied() == false) return false; + } + return true; + } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + js["Name"] = _name; + js["Ready"] = isReady(); + js["Finished"] = isFinished(); + + js["Message metadata"] = nlohmann::json::object(); + js["Message metadata"]["Type"] = _messageMetadata.type; + js["Message metadata"]["Group"] = _messageMetadata.groupId; + js["Message metadata"]["Sequence"] = _messageMetadata.sequenceId; + js["Message metadata"]["ID"] = _messageMetadata.getId(); + + js["Input Dependencies"] = nlohmann::json::array(); + for (const auto &[dependencyName, dependency] : _inputDependencies) + { + nlohmann::json depJs; + depJs["Name"] = dependency.getName(); + depJs["Dependency Name"] = dependencyName; + depJs["Satisfied"] = dependency.isSatisfied(); + depJs["Data Size"] = (dependency.getData() == nullptr) ? 0 : dependency.getData()->getSize(); + js["Input Dependencies"].push_back(depJs); + } + js["Output Dependencies"] = nlohmann::json::array(); + for (const auto &[dependencyName, dependency] : _outputDependencies) + { + nlohmann::json depJs; + depJs["Name"] = dependency.getName(); + depJs["Dependency Name"] = dependencyName; + depJs["Satisfied"] = dependency.isSatisfied(); + depJs["Data Size"] = (dependency.getData() == nullptr) ? 0 : dependency.getData()->getSize(); + js["Output Dependencies"].push_back(depJs); + } + return js; + } + + protected: + + const std::string _name; + const system::channels::Message::metadata_t _messageMetadata; + std::unordered_map _inputDependencies; + std::unordered_map _outputDependencies; +}; +} // namespace serving::modules::roles diff --git a/platform/include/modules/roles/jobFactory.hpp b/platform/include/modules/roles/jobFactory.hpp new file mode 100644 index 0000000..7192f33 --- /dev/null +++ b/platform/include/modules/roles/jobFactory.hpp @@ -0,0 +1,83 @@ +#pragma once + +#include +#include + +#include +#include + +#include "./dependency.hpp" +#include "./job.hpp" + +namespace serving::modules::roles +{ + +class JobFactory +{ + public: + + using instanceId_t = HiCR::Instance::instanceId_t; + using jobName_t = std::string; + using dependency_t = std::string; + + JobFactory(const instanceId_t instanceId, const configuration::Deployment &deployment) + : _jobInputDependencies(buildTaskInputDependencies(*deployment.getPartitions()[instanceId])), + _jobOutputDependencies(buildTaskOutputDependencies(*deployment.getPartitions()[instanceId])), + _edgeInfos(buildEdgeInfos(deployment)) + {} + + [[nodiscard]] __INLINE__ Job createJob(const std::string &jobName, const system::channels::Message::metadata_t &metadata) + { + auto inputDependencies = std::unordered_map(); + 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); + } + } + + auto outputDependencies = std::unordered_map(); + 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); + } + } + + return Job(jobName, metadata, inputDependencies, outputDependencies); + } + + private: + + std::unordered_map> buildEdgeInfos(const configuration::Deployment &deployment) + { + auto edgeInfos = std::unordered_map>(); + for (const auto &edge : deployment.getEdges()) { edgeInfos[edge->getName()] = edge; } + return edgeInfos; + } + + std::unordered_map> buildTaskInputDependencies(const configuration::Partition &partition) + { + auto taskInfos = std::unordered_map>(); + for (const auto &task : partition.getTasks()) { taskInfos[task->getFunctionName()] = task->getInputs(); } + return taskInfos; + } + + std::unordered_map> buildTaskOutputDependencies(const configuration::Partition &partition) + { + auto taskInfos = std::unordered_map>(); + for (const auto &task : partition.getTasks()) { taskInfos[task->getFunctionName()] = task->getOutputs(); } + return taskInfos; + } + + const std::unordered_map> _jobInputDependencies; + const std::unordered_map> _jobOutputDependencies; + const std::unordered_map> _edgeInfos; +}; +} // namespace serving::modules::roles diff --git a/platform/include/modules/roles/replica/module.hpp b/platform/include/modules/roles/replica/module.hpp new file mode 100644 index 0000000..8576412 --- /dev/null +++ b/platform/include/modules/roles/replica/module.hpp @@ -0,0 +1,209 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include +#include + +#include + +namespace serving::modules::roles::replica +{ + +class Module final : public serving::modules::Module +{ + public: + + using instanceId_t = HiCR::Instance::instanceId_t; + using input_t = std::shared_ptr; + using output_t = std::shared_ptr; + using message_t = serving::system::channels::Message; + using metadata_t = serving::system::channels::Message::metadata_t; + using messageId_t = serving::system::channels::Message::messageId_t; + using messageType_t = serving::system::channels::Message::messageType_t; + using edgeName_t = std::string; + using inputMap_t = std::unordered_map; + using outputMap_t = std::unordered_map; + using processFc_t = std::function; + using memorySlot_t = std::shared_ptr; + + Module(const std::string &taskName, const std::vector &inputNames, const std::vector &outputNames, const processFc_t &processFc) + : serving::modules::Module(), + _taskName(taskName), + _inputNames(inputNames.begin(), inputNames.end()), + _outputNames(outputNames.begin(), outputNames.end()), + _isCoordinatorSet(false), + _processFc(processFc) + {} + + ~Module() override = default; + + __INLINE__ void setCoordinator(const instanceId_t coordinatorId, const outputMap_t &outputChannels, const inputMap_t &inputChannels) + { + if (_isCoordinatorSet) { HICR_THROW_LOGIC("Coordinator %lu already registered.", _coordinatorId); } + _coordinatorId = coordinatorId; + _outputChannels = outputChannels; + _inputChannels = inputChannels; + _isCoordinatorSet = true; + } + + __INLINE__ void addMessageType(const messageType_t messageType) { _messageTypes.insert(messageType); } + + [[nodiscard]] __INLINE__ std::vector buildSubscriptions() + { + std::vector out; + out.reserve(_messageTypes.size() * _inputChannels.size()); + for (const auto &[inputName, inputChannel] : _inputChannels) + { + for (const auto messageType : _messageTypes) + out.emplace_back(messageType, inputChannel, [this, inputName](const input_t, const message_t &message) { this->coordinatorMessageHandler(inputName, message); }); + } + return out; + } + + [[nodiscard]] __INLINE__ std::vector> buildUnsubscriptions() const + { + std::vector> out; + out.reserve(_messageTypes.size() * _inputChannels.size()); + for (const auto &[_, inputChannel] : _inputChannels) + { + for (const auto messageType : _messageTypes) out.push_back({messageType, inputChannel}); + } + return out; + } + + void initialize() override + { + if (_isCoordinatorSet == false) HICR_THROW_LOGIC("Coordinator has not been set"); + if (_messageTypes.empty()) HICR_THROW_LOGIC("Replica has no message types configured."); + for (const auto &inputName : _inputNames) + if (_inputChannels.contains(inputName) == false) HICR_THROW_LOGIC("Replica has no input channel for dependency '%s'.", inputName.c_str()); + for (const auto &outputName : _outputNames) + if (_outputChannels.contains(outputName) == false) HICR_THROW_LOGIC("Replica has no output channel for dependency '%s'.", outputName.c_str()); + } + + void run() override {} + void terminate() override {} + void await() override {} + void finalize() override {} + + protected: + + void service() override {} + + private: + + struct ActiveJob + { + metadata_t metadata; + std::unordered_map inputs; + }; + + __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 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(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); + } + + [[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(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; + } + + [[nodiscard]] __INLINE__ serving::modules::roles::TaskContext::outputConfigMap_t buildOutputConfigs() const + { + serving::modules::roles::TaskContext::outputConfigMap_t configs; + for (const auto &[name, output] : _outputChannels) configs.emplace(name, output->getConfig()); + return configs; + } + + __INLINE__ void freeInputSlots(ActiveJob &job) const + { + for (auto &[name, slot] : job.inputs) + { + if (slot == nullptr) continue; + _inputChannels.at(name)->getConfig().payloadMemoryManager->freeLocalMemorySlot(slot); + slot = nullptr; + } + } + + [[nodiscard]] __INLINE__ bool isReady(const ActiveJob &job) const + { + for (const auto &inputName : _inputNames) + if (job.inputs.contains(inputName) == false) return false; + return true; + } + + const std::string _taskName; + const std::unordered_set _inputNames; + const std::unordered_set _outputNames; + + bool _isCoordinatorSet; + processFc_t _processFc; + + HiCR::Instance::instanceId_t _coordinatorId; + inputMap_t _inputChannels; + outputMap_t _outputChannels; + std::unordered_set _messageTypes; + std::unordered_map _activeJobs; + std::mutex _activeJobsMutex; +}; +} // namespace serving::modules::roles::replica diff --git a/platform/include/modules/roles/taskContext.hpp b/platform/include/modules/roles/taskContext.hpp new file mode 100644 index 0000000..d973cb5 --- /dev/null +++ b/platform/include/modules/roles/taskContext.hpp @@ -0,0 +1,105 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include + +namespace serving::modules::roles +{ + +class TaskContext final +{ + public: + + using metadata_t = serving::system::channels::Message::metadata_t; + using memorySlot_t = std::shared_ptr; + using outputConfigMap_t = std::unordered_map; + + struct Output + { + std::string name; + memorySlot_t data; + bool owned = false; + }; + + TaskContext(const std::string &name, const metadata_t &metadata, const std::unordered_map &inputs, const outputConfigMap_t &outputConfigs) + : _name(name), + _metadata(metadata), + _inputs(inputs), + _outputConfigs(outputConfigs) + {} + + [[nodiscard]] __INLINE__ const std::string &getName() const { return _name; } + [[nodiscard]] __INLINE__ const metadata_t &getMetadata() const { return _metadata; } + + [[nodiscard]] __INLINE__ memorySlot_t getInput(const std::string &name) const + { + if (_inputs.contains(name) == false) HICR_THROW_RUNTIME("Task '%s' has no input '%s'.", _name.c_str(), name.c_str()); + const auto &input = _inputs.at(name); + if (input == nullptr) HICR_THROW_RUNTIME("Task '%s' input '%s' has no memory slot.", _name.c_str(), name.c_str()); + return input; + } + + __INLINE__ void setOutput(const std::string &name, const memorySlot_t &data) + { + if (_outputIndex.contains(name)) HICR_THROW_RUNTIME("Task '%s' set output '%s' more than once.", _name.c_str(), name.c_str()); + if (_outputConfigs.contains(name) == false) HICR_THROW_RUNTIME("Task '%s' set undeclared output '%s'.", _name.c_str(), name.c_str()); + if (data == nullptr) HICR_THROW_RUNTIME("Task '%s' cannot set output '%s' from a null memory slot.", _name.c_str(), name.c_str()); + + _outputIndex[name] = _outputs.size(); + _outputs.push_back(Output{.name = name, .data = data, .owned = false}); + } + + __INLINE__ void setOutput(const std::string &name, const void *data, const size_t size) + { + if (_outputIndex.contains(name)) HICR_THROW_RUNTIME("Task '%s' set output '%s' more than once.", _name.c_str(), name.c_str()); + if (data == nullptr && size > 0) HICR_THROW_LOGIC("Task '%s' cannot set output '%s' from null data with non-zero size.", _name.c_str(), name.c_str()); + if (_outputConfigs.contains(name) == false) HICR_THROW_RUNTIME("Task '%s' set undeclared output '%s'.", _name.c_str(), name.c_str()); + + memorySlot_t outputSlot = nullptr; + if (size > 0) + { + const auto &cfg = _outputConfigs.at(name); + const auto srcSlot = cfg.payloadMemoryManager->registerLocalMemorySlot(cfg.payloadMemorySpace, const_cast(data), size); + outputSlot = cfg.payloadMemoryManager->allocateLocalMemorySlot(cfg.payloadMemorySpace, size); + cfg.payloadCommunicationManager->memcpy(outputSlot, 0, srcSlot, 0, size); + cfg.payloadCommunicationManager->fence(outputSlot, 0, 1); + cfg.payloadMemoryManager->deregisterLocalMemorySlot(srcSlot); + } + + _outputIndex[name] = _outputs.size(); + _outputs.push_back(Output{.name = name, .data = outputSlot, .owned = true}); + } + + [[nodiscard]] __INLINE__ const std::vector &getOutputs() const { return _outputs; } + + __INLINE__ void freeOwnedOutputs() + { + for (auto &output : _outputs) + { + if (output.owned == false || output.data == nullptr) continue; + _outputConfigs.at(output.name).payloadMemoryManager->freeLocalMemorySlot(output.data); + output.data = nullptr; + } + } + + private: + + const std::string _name; + const metadata_t _metadata; + const std::unordered_map &_inputs; + const outputConfigMap_t &_outputConfigs; + std::vector _outputs; + std::unordered_map _outputIndex; +}; + +} // namespace serving::modules::roles diff --git a/platform/include/modules/taskScheduler/module.hpp b/platform/include/modules/taskScheduler/module.hpp new file mode 100644 index 0000000..5cb8f8a --- /dev/null +++ b/platform/include/modules/taskScheduler/module.hpp @@ -0,0 +1,67 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include + +#include + +namespace serving::modules::taskScheduler +{ + +class Module final : public serving::modules::Module +{ + public: + + using taskFunction_t = taskr::function_t; + + Module(std::shared_ptr computeManager, std::shared_ptr taskr) + : serving::modules::Module(), + _computeManager(computeManager), + _taskr(taskr) + {} + + ~Module() override = default; + + __INLINE__ void addTask(const std::string &name, const taskFunction_t &function) + { + if (_taskNameToIndex.contains(name)) HICR_THROW_LOGIC("Task '%s' is already registered in taskScheduler module.", name.c_str()); + _functions.push_back(std::make_unique(_computeManager.get(), function)); + _tasks.push_back(std::make_unique(_functions.back().get())); + _taskNameToIndex[name] = _tasks.size() - 1; + } + + void initialize() override + { + _taskr->setTaskCallbackHandler(HiCR::tasking::Task::callback_t::onTaskSuspend, [&](taskr::Task *task) { _taskr->resumeTask(task); }); + _taskr->setFinishOnLastTask(false); + for (auto &task : _tasks) _taskr->addTask(task.get()); + _taskr->initialize(); + } + + void run() override { _taskr->run(); } + + void terminate() override { _taskr->setFinishOnLastTask(true); } + + void await() override { _taskr->await(); } + + void finalize() override { _taskr->finalize(); } + + protected: + + void service() override {} + + private: + + std::shared_ptr _computeManager; + std::shared_ptr _taskr; + + std::vector> _functions; + std::vector> _tasks; + std::unordered_map _taskNameToIndex; +}; +} // namespace serving::modules::taskScheduler diff --git a/platform/include/system/channels/base.hpp b/platform/include/system/channels/base.hpp index 82ac843..aab9c92 100644 --- a/platform/include/system/channels/base.hpp +++ b/platform/include/system/channels/base.hpp @@ -123,7 +123,8 @@ class Base __INLINE__ void lock() { _lock.lock(); } __INLINE__ void unlock() { _lock.unlock(); } - __INLINE__ bool isReady() const { return _isReady; } + __INLINE__ bool isReady() const { return _isReady; } + __INLINE__ const channelConfig_t &getConfig() const { return _config; } protected: From 9362af582c3265fde209f034b494131d3604d0aa Mon Sep 17 00:00:00 2001 From: Luca Date: Tue, 7 Jul 2026 15:22:33 +0200 Subject: [PATCH 2/3] fix(pr): address Gemini review comments on PR #60 - 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 --- platform/include/modules/roles/job.hpp | 4 ++-- platform/include/modules/roles/jobFactory.hpp | 7 +++---- platform/include/modules/roles/replica/module.hpp | 12 +++++++++--- 3 files changed, 14 insertions(+), 9 deletions(-) diff --git a/platform/include/modules/roles/job.hpp b/platform/include/modules/roles/job.hpp index 736a192..6faaf7c 100644 --- a/platform/include/modules/roles/job.hpp +++ b/platform/include/modules/roles/job.hpp @@ -32,8 +32,8 @@ class Job ~Job() = default; - [[nodiscard]] __INLINE__ Dependency &getInputDependency(const dependency_t dependencyName) { return _inputDependencies.at(dependencyName); } - [[nodiscard]] __INLINE__ Dependency &getOutputDependency(const dependency_t dependencyName) { return _outputDependencies.at(dependencyName); } + [[nodiscard]] __INLINE__ Dependency &getInputDependency(const dependency_t &dependencyName) { return _inputDependencies.at(dependencyName); } + [[nodiscard]] __INLINE__ Dependency &getOutputDependency(const dependency_t &dependencyName) { return _outputDependencies.at(dependencyName); } [[nodiscard]] __INLINE__ std::unordered_map &getInputDependencies() { return _inputDependencies; } [[nodiscard]] __INLINE__ std::unordered_map &getOutputDependencies() { return _outputDependencies; } [[nodiscard]] __INLINE__ const std::unordered_map &getInputDependencies() const { return _inputDependencies; } diff --git a/platform/include/modules/roles/jobFactory.hpp b/platform/include/modules/roles/jobFactory.hpp index 7192f33..48b6e1b 100644 --- a/platform/include/modules/roles/jobFactory.hpp +++ b/platform/include/modules/roles/jobFactory.hpp @@ -16,13 +16,12 @@ class JobFactory { public: - using instanceId_t = HiCR::Instance::instanceId_t; using jobName_t = std::string; using dependency_t = std::string; - JobFactory(const instanceId_t instanceId, const configuration::Deployment &deployment) - : _jobInputDependencies(buildTaskInputDependencies(*deployment.getPartitions()[instanceId])), - _jobOutputDependencies(buildTaskOutputDependencies(*deployment.getPartitions()[instanceId])), + JobFactory(const size_t partitionIndex, const configuration::Deployment &deployment) + : _jobInputDependencies(buildTaskInputDependencies(*deployment.getPartitions().at(partitionIndex))), + _jobOutputDependencies(buildTaskOutputDependencies(*deployment.getPartitions().at(partitionIndex))), _edgeInfos(buildEdgeInfos(deployment)) {} diff --git a/platform/include/modules/roles/replica/module.hpp b/platform/include/modules/roles/replica/module.hpp index 8576412..a39ea28 100644 --- a/platform/include/modules/roles/replica/module.hpp +++ b/platform/include/modules/roles/replica/module.hpp @@ -118,14 +118,20 @@ class Module final : public serving::modules::Module 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(); + // 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("Replica received duplicate input dependency '%s' for job %lu.", inputName.c_str(), jobId); - job.inputs[inputName] = copyMessageData(inputName, message); + 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); From dcb539c2f78a3e1d80ea0acae1b250865b60865e Mon Sep 17 00:00:00 2001 From: Luca Date: Wed, 8 Jul 2026 14:04:27 +0200 Subject: [PATCH 3/3] fix(pr): address CodeRabbit review comments on PR #60 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 --- .../modules/channelDispatcher/module.hpp | 15 ++++---- .../include/modules/requestManager/module.hpp | 2 +- .../include/modules/roles/replica/module.hpp | 34 ++++++++++++++++--- .../include/modules/roles/taskContext.hpp | 3 +- 4 files changed, 40 insertions(+), 14 deletions(-) diff --git a/platform/include/modules/channelDispatcher/module.hpp b/platform/include/modules/channelDispatcher/module.hpp index c31aec3..25d73f7 100644 --- a/platform/include/modules/channelDispatcher/module.hpp +++ b/platform/include/modules/channelDispatcher/module.hpp @@ -74,12 +74,13 @@ class Module final : public modules::Module for (const auto &edge : _pollEdges) { edge->lock(); - - if (edge->hasMessage() == false) + struct EdgeUnlockGuard { - edge->unlock(); - continue; - } + const std::shared_ptr &edge; + ~EdgeUnlockGuard() { edge->unlock(); } + } unlockGuard{edge}; + + if (edge->hasMessage() == false) { continue; } const auto message = edge->getMessage(); const auto messageType = message.getMetadata().type; @@ -90,15 +91,13 @@ class Module final : public modules::Module 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(); } } diff --git a/platform/include/modules/requestManager/module.hpp b/platform/include/modules/requestManager/module.hpp index 75b753b..f7c824b 100644 --- a/platform/include/modules/requestManager/module.hpp +++ b/platform/include/modules/requestManager/module.hpp @@ -39,7 +39,7 @@ class Module final : public serving::modules::Module Prompt(const metadata_t metadata, const uint8_t *data, const size_t size) : _metadata(metadata), - _input(data, data + size) + _input((size > 0 && data != nullptr) ? std::vector(data, data + size) : std::vector{}) {} [[nodiscard]] const metadata_t &getMetadata() const { return _metadata; } diff --git a/platform/include/modules/roles/replica/module.hpp b/platform/include/modules/roles/replica/module.hpp index a39ea28..ce00c90 100644 --- a/platform/include/modules/roles/replica/module.hpp +++ b/platform/include/modules/roles/replica/module.hpp @@ -140,6 +140,25 @@ class Module final : public serving::modules::Module auto outputConfigs = buildOutputConfigs(); serving::modules::roles::TaskContext context(_taskName, activeJob.metadata, activeJob.inputs, outputConfigs); + + // Scope guard: if _processFc or any output validation throws, free owned + // outputs and input slots before unwinding to prevent slot leaks. + struct CleanupGuard + { + serving::modules::roles::TaskContext &context; + ActiveJob &activeJob; + std::function freeInputSlotsFn; + bool dismissed = false; + ~CleanupGuard() + { + if (!dismissed) + { + context.freeOwnedOutputs(); + freeInputSlotsFn(activeJob); + } + } + } guard{context, activeJob, [this](ActiveJob &j) { freeInputSlots(j); }}; + _processFc(context); std::unordered_set sentOutputs; @@ -148,16 +167,23 @@ class Module final : public serving::modules::Module 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); + } + 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()); + + // All validation passed. Dismiss the guard so outputs stay alive for the + // coordinator completion callback, which is responsible for freeing them. + guard.dismissed = true; + + for (const auto &output : context.getOutputs()) + { const auto response = message_t( output.data == nullptr ? nullptr : static_cast(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(); + // Input slots are no longer needed once outputs have been dispatched. freeInputSlots(activeJob); } diff --git a/platform/include/modules/roles/taskContext.hpp b/platform/include/modules/roles/taskContext.hpp index d973cb5..6617369 100644 --- a/platform/include/modules/roles/taskContext.hpp +++ b/platform/include/modules/roles/taskContext.hpp @@ -45,7 +45,7 @@ class TaskContext final { if (_inputs.contains(name) == false) HICR_THROW_RUNTIME("Task '%s' has no input '%s'.", _name.c_str(), name.c_str()); const auto &input = _inputs.at(name); - if (input == nullptr) HICR_THROW_RUNTIME("Task '%s' input '%s' has no memory slot.", _name.c_str(), name.c_str()); + if (input == nullptr) return _emptySlot; return input; } @@ -94,6 +94,7 @@ class TaskContext final private: + inline static const memorySlot_t _emptySlot = std::make_shared(nullptr, 0); const std::string _name; const metadata_t _metadata; const std::unordered_map &_inputs;