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

platform/examples/include/channels/helpers.hpp

@@ -118,6 +118,23 @@ __INLINE__ void waitUntilReady(const std::shared_ptr<serving::system::channels::
   while (!channel->isReady()) { std::this_thread::sleep_for(std::chrono::milliseconds(intervalMs)); }
 }
 
+__INLINE__ std::shared_ptr<serving::configuration::Edge> 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());

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')

platform/examples/modules/coordinatorReplica/coordinator.hpp

@@ -0,0 +1,159 @@
+#pragma once
+
+#include <atomic>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include <hicr/core/exceptions.hpp>
+#include <hicr/core/instance.hpp>
+
+#include <modules/channelController/module.hpp>
+#include <modules/channelDispatcher/module.hpp>
+#include <modules/configuration/deployment.hpp>
+#include <modules/roles/coordinator/module.hpp>
+#include <modules/roles/job.hpp>
+#include <modules/roles/jobFactory.hpp>
+#include <modules/service/module.hpp>
+#include <system/channels/message.hpp>
+#include <system/engine.hpp>
+
+#include <channels/helpers.hpp>
+#include <deployment/helpers.hpp>
+
+using instanceId_t = HiCR::Instance::instanceId_t;
+
+struct CoordinatorRuntime
+{
+  using messageType_t = serving::system::channels::Message::messageType_t;
+  using input_t       = std::shared_ptr<serving::system::channels::Input>;
+
+  std::shared_ptr<serving::modules::roles::coordinator::Module> coordinatorModule;
+  std::shared_ptr<serving::modules::roles::JobFactory>          jobFactory;
+  desiredSingleLocalChannels_t                                  graphChannels;
+  std::pair<messageType_t, input_t>                             graphUnsubscription;
+  std::shared_ptr<std::atomic<bool>>                            completed = std::make_shared<std::atomic<bool>>(false);
+};
+
+__INLINE__ void submitGraphJob(const std::shared_ptr<serving::modules::roles::JobFactory>          &jobFactory,
+                               const std::shared_ptr<serving::modules::roles::coordinator::Module> &coordinatorModule,
+                               const std::string                                                   &jobName,
+                               const std::string                                                   &inputEdgeName,
+                               const serving::system::channels::Message                            &message)
+{
+  auto  job        = std::make_shared<serving::modules::roles::Job>(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<std::atomic<bool>>                            &completed,
+                                   const std::shared_ptr<serving::modules::roles::JobFactory>          &jobFactory,
+                                   const std::shared_ptr<serving::modules::roles::coordinator::Module> &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<serving::system::channels::Output>                   &outputChannel,
+                            const serving::system::channels::Message::metadata_t                       &metadata,
+                            const std::vector<serving::modules::roles::coordinator::Module::JobOutput> &outputs)
+{
+  for (const auto &output : outputs)
+  {
+    waitUntilReady(outputChannel);
+    const auto message = serving::system::channels::Message(
+      output.data == nullptr ? nullptr : static_cast<const uint8_t *>(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<instanceId_t>                                    &replicaIds,
+                                          serving::configuration::Deployment                                 &deployment,
+                                          const serving::system::channels::keyBuilderFc_t                    &keyBuilder,
+                                          const std::shared_ptr<serving::modules::channelController::Module> &channelControllerModule,
+                                          const std::shared_ptr<serving::modules::channelDispatcher::Module> &channelDispatcherModule,
+                                          const std::shared_ptr<serving::modules::service::Module>           &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<serving::modules::roles::JobFactory>(partitionIndex, deployment);
+  coordinatorRuntime.coordinatorModule = std::make_shared<serving::modules::roles::coordinator::Module>(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<serving::system::channels::channelId_t>(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<serving::system::channels::channelId_t>(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<serving::system::channels::Input>, 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<serving::modules::roles::coordinator::Module::JobOutput> &outputs) {
+      routeOutput(outputChannel, metadata, outputs);
+    });
+
+  for (auto &subscription : coordinatorRuntime.coordinatorModule->buildSubscriptions()) { channelDispatcherModule->subscribe(subscription); }
+  return coordinatorRuntime;
+}

platform/examples/modules/coordinatorReplica/coordinatorReplica.cpp

@@ -0,0 +1,152 @@
+#include <stdio.h>
+#include <thread>
+#include <fstream>
+#include <chrono>
+#include <optional>
+#include <vector>
+
+#include <modules/channelController/module.hpp>
+#include <modules/channelDispatcher/module.hpp>
+#include <modules/configuration/deployment.hpp>
+#include <modules/service/module.hpp>
+#include <system/channels/message.hpp>
+#include <system/engine.hpp>
+
+#include <channels/helpers.hpp>
+#include <deployment/helpers.hpp>
+#include <runtime/helpers.hpp>
+
+#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<instanceId_t> 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<HiCR::CommunicationManager *> managerOrder            = {runtime.communicationManager.get()};
+  auto                                      channelControllerModule = std::make_shared<serving::modules::channelController::Module>(instanceId, managerOrder);
+  auto                                      channelDispatcherModule = std::make_shared<serving::modules::channelDispatcher::Module>(20);
+  auto                                      serviceModule           = std::make_shared<serving::modules::service::Module>(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<CoordinatorRuntime> coordinatorRt;
+  std::optional<ReplicaRuntime>     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<const uint8_t *>(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;
+}

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