diff --git a/.agents/skills/README.md b/.agents/skills/README.md index 7c8801a..6f5ced9 100644 --- a/.agents/skills/README.md +++ b/.agents/skills/README.md @@ -8,3 +8,13 @@ tooling without maintaining duplicate copies. Add or update skill directories here. Each skill directory must contain its own `SKILL.md`. + +## Skills + +| Skill | Description | +|---|---| +| `create-feature-request` | File a GitHub feature request following the repo template | +| `create-issue` | File a GitHub bug or docs issue following the repo template | +| `rebase-on-main` | Rebase the current feature branch on latest main and force-push, resolving conflicts by preserving the intent of both sides | +| `retrigger-ci` | Retry a failing CI run until it passes when the failure is flaky and unrelated to our changes | +| `platform-style-fix` | Check and fix clang-format style violations in `platform/include` and `platform/examples` | diff --git a/platform/docs/README.md b/platform/docs/README.md index 639ee50..2d601d5 100644 --- a/platform/docs/README.md +++ b/platform/docs/README.md @@ -22,21 +22,27 @@ platform/ build/ generated build output; not documented here ``` -## Module Areas +## Documentation +- [Architecture](architecture.md): instance roles, module system, channel model, and scope. +- [Simpler Integration](simpler-integration.md): how replica ranks launch Simpler, the `processFc` interface, and the path to in-device tensor channels. +- [Model Support Interface](model-support-interface.md): what Model Support owns, provides, and will consume from the platform. - [System](system/README.md): engine lifecycle and cross-instance start/stop control. - [Channels](system/channels.md): payload, coordination, metadata, input, output, and message primitives. - -Configuration types, the service module, channel controller, and broadcast deployment are tracked in issue #32 and are not part of this initial PR. +- [Modules](modules/README.md): configuration, service, channel controller, and broadcast deployment. ## Runtime Shape The current platform runtime is built around `serving::system::Engine`. The engine owns a set of `serving::modules::Module` instances, initializes them, starts them across instances through RPC, waits for termination, and finalizes them. -This initial PR covers the following building blocks: +This PR covers the following building blocks: - Engine lifecycle: cross-instance start/stop over RPC (`serving::system::Engine`). - Module base interface: initialize/run/terminate/await/finalize lifecycle with optional `taskr::Service` (`serving::modules::Module`). - Channel primitives: `Input`, `Output`, `Message`, and `MessageTypeRegistry` for host-side control traffic. +- Configuration types: `Deployment`, `Partition`, `Task`, `Edge`, `Replica`, `RequestManager` with JSON serialization and verification. +- `broadcastDeployment`: deployer-to-worker deployment config distribution over RPC. +- `channelController`: desired-vs-actual reconciliation loop for host-side SPSC channels. +- `service`: wraps `taskr::Runtime` for cooperative background services. -Deployment graph representation, deployment broadcast, desired-state channel creation, dynamic scaling, topology-aware replacement, fault recovery, and Python bindings are not implemented in this PR. +Dynamic scaling, fault recovery, `channelDispatcher`, `taskScheduler`, executor roles, heartbeat, `RuntimePlan` update protocol, in-device tensor channels, and Python bindings are deferred to follow-up PRs. diff --git a/platform/docs/architecture.md b/platform/docs/architecture.md new file mode 100644 index 0000000..c50c6d5 --- /dev/null +++ b/platform/docs/architecture.md @@ -0,0 +1,69 @@ +# Platform Architecture + +## What the platform layer is + +The platform layer is the **host-side control plane** for a distributed PyPTO Serving deployment. It manages: + +- Instance lifecycle (start, stop, health) +- Deployment configuration distribution +- Host-side channel creation and teardown +- Module initialization and service scheduling + +It is not a model execution layer. Model kernels, tensor movement, KV cache, batching, and token scheduling belong to Model Support and are executed through Simpler on each device. The platform exists so that Model Support does not have to manage MPI ranks, HiCR channels, or distributed lifecycle directly. + +## Instance roles + +Every MPI rank in a deployment plays one of four roles, determined by its instance ID and the deployment configuration: + +| Role | Count | Responsibility | +|---|---|---| +| **Deployer** | 1 (root) | Reads deployment config, broadcasts it to all other ranks, then transitions to a coordinator or replica role | +| **Coordinator** | one per partition | Routes jobs to replicas, accumulates outputs, fires completion callbacks | +| **Replica** | one or more per partition | Executes model computation via Simpler on a dedicated NPU device | +| **Request Manager** | 1 | Entry point for client requests; maps to the partition owning the user-interface edge | + +Device assignment for replicas: `deviceId = instanceId - numPartitions`. Each replica rank owns exactly one NPU. + +## Module system + +All platform behaviour is expressed as `serving::modules::Module` instances owned by the `serving::system::Engine`. The engine drives a fixed lifecycle: + +``` +initialize() → run() → [service loop] → terminate() → await() → finalize() +``` + +Modules can register a periodic background service (via `taskr::Service`) that runs inside the service loop between the `run()` and `terminate()` phases. The engine coordinates the lifecycle of all instances over RPC, so every rank progresses through the same phases in lockstep. + +### Modules in this PR + +| Module | Purpose | +|---|---| +| `broadcastDeployment` | Deployer sends deployment JSON to all workers over RPC at initialize time | +| `channelController` | Desired-vs-actual reconciliation loop; creates and tears down HiCR SPSC channels | +| `service` | Wraps `taskr::Runtime`; owns and drives background `taskr::Service` instances | + +### Deferred modules (next PRs) + +| Module | Purpose | +|---|---| +| `channelDispatcher` | Polls subscribed input channels; dispatches messages to registered handlers | +| `taskScheduler` | Registers named `taskr::Task` instances; drives taskr through the module lifecycle | +| `roles::coordinator` | Job queue and replica dispatch; completion callback when all outputs are gathered | +| `roles::replica` | Receives coordinator input, invokes `processFc`, returns outputs | +| `heartbeat` | Periodic health check between coordinators and replicas | + +## Channel model + +All channels in this PR are **host-side HiCR SPSC channels** carrying variable-size payloads and fixed-size metadata. They are used for control traffic and small tensors. + +In-device tensor channels — where the hot path (prefill/decode token data) moves directly between NPU devices without staging through host memory — are **not implemented here**. They are the next major milestone and unblock the TP/PP tensor data path. + +## What is deliberately out of scope + +- Per-token scheduling (hot path) +- KV cache management +- Batching and sampling policy +- Python bindings +- `RuntimePlan` update protocol and watch/subscribe API +- Dynamic scaling, drain, and fault recovery +- Topology-aware placement diff --git a/platform/docs/model-support-interface.md b/platform/docs/model-support-interface.md new file mode 100644 index 0000000..1dd7cd6 --- /dev/null +++ b/platform/docs/model-support-interface.md @@ -0,0 +1,74 @@ +# Model Support Interface + +This document describes what the Model Support layer needs to know and control when integrating with the platform. + +## What the platform tells you + +At startup, after `broadcastDeployment` completes, every rank knows: + +- **Its role**: coordinator, replica, or request manager — determined by instance ID and the deployment config +- **Its device index** (replicas only): `deviceId = instanceId - numPartitions` +- **The full deployment graph**: partitions, tasks, edges, and which edges carry user-interface traffic + +Model Support should use this information to initialize Simpler on replica ranks before the engine starts. + +## What you provide + +### 1. Compiled Simpler artifacts + +Replica ranks need the five runtime library paths in `RuntimeBinaries`. The platform does not locate or validate them. Pass them via command-line arguments or configuration alongside the deployment JSON. + +### 2. A `processFc` per task function name + +For each task declared in the deployment configuration, Model Support registers a function: + +```cpp +std::function +``` + +This is the only point where Model Support code runs during execution. Everything else — job routing, input aggregation, output forwarding, channel lifecycle — is handled by the platform. + +### 3. Request/response edge names + +The deployment config declares a `RequestManager` with an input edge name and an output edge name. Model Support must ensure: + +- The input edge name matches the edge from which user requests arrive +- The output edge name matches the edge to which results are written +- The task that reads the input edge does not have any other inter-partition inputs +- The task that writes the output edge does not have any other inter-partition outputs + +These constraints are verified by `Deployment::verify()` at startup. + +## What you do not own + +| Concern | Owner | +|---|---| +| MPI rank management | Platform (via HiCR `InstanceManager`) | +| Channel creation and teardown | Platform (`channelController` module) | +| Deployment broadcast | Platform (`broadcastDeployment` module) | +| Job routing from coordinator to replica | Platform (`coordinator::Module`) | +| Input aggregation and output forwarding | Platform (`replica::Module`) | +| Engine lifecycle (init/run/terminate) | Platform (`Engine`) | +| Replica health monitoring | Platform (heartbeat, future) | + +## RuntimePlan — future interface + +The current PR does not yet expose a `RuntimePlan`. A future milestone will provide a versioned, observable snapshot of what the platform has actually instantiated: + +- Which replicas are live, draining, or unavailable +- Which channels exist and what their endpoints are +- Safe `active → draining → removed` state transitions before any channel is deleted + +Model Support will consume this object to make routing and scheduling decisions (KV-locality-aware replica selection, drain-aware request assignment, channel handle lookup) without duplicating platform resource state. The same object will be queryable by operators and higher-level control loops for observability. + +Until `RuntimePlan` is implemented, Model Support should treat the deployment configuration as the ground truth of what is running. + +## Tensor channel hot path — future + +All channels are currently host-side. For the prefill/decode token data path, in-device tensor channels will carry tensor payloads directly between NPU devices without staging through host memory. When implemented: + +- Replica `processFc` will receive device-memory `LocalMemorySlot` handles rather than host buffers +- `toDevice` / `toHost` staging in the process function becomes unnecessary +- The `processFc` signature does not change + +Model Support should design the process function to be forward-compatible: check whether the incoming `LocalMemorySlot` is a host or device buffer and branch accordingly. diff --git a/platform/docs/modules/README.md b/platform/docs/modules/README.md new file mode 100644 index 0000000..3e5be68 --- /dev/null +++ b/platform/docs/modules/README.md @@ -0,0 +1,52 @@ +# Modules + +Source paths: + +- `include/modules/module.hpp` +- `include/modules/subscription.hpp` +- `include/modules/service/module.hpp` +- `include/modules/channelController/module.hpp` +- `include/modules/broadcastDeployment/module.hpp` + +The modules layer defines reusable platform components that can be installed into `serving::system::Engine`. Modules use a common lifecycle and may optionally expose a periodic TaskR service. + +## Base Module + +`serving::modules::Module` is the abstract base class for all platform modules. + +Main responsibilities: + +- Provide a uniform lifecycle: `initialize()`, `run()`, `terminate()`, `await()`, and `finalize()`. +- Optionally create a `taskr::Service` when constructed with a positive interval. +- Let the engine discover and register periodic services through `hasService()` and `getService()`. + +The base module does not define deployment semantics itself. It only defines how a platform component participates in the runtime lifecycle. + +## Lifecycle + +The engine calls module methods in this order: + +1. `initialize()` before the instance is marked running. +2. `run()` when the deployer starts itself or sends start RPCs to workers. +3. `terminate()` after stop is requested. +4. `await()` after termination starts. +5. `finalize()` after module work has stopped. + +Periodic modules implement their work in the protected `service()` method. The base constructor wraps that method in a `taskr::Service` using the requested interval. + +## Subscription + +`serving::modules::Subscription` binds a message type, an input channel, and a callback: + +- Message type: `channels::Message::messageType_t`. +- Edge: `std::shared_ptr`. +- Handler: `std::function, const channels::Message &)>`. + +This type is a small ownership wrapper for message-driven modules. It does not poll or dispatch by itself. + +## Relation To Platform Design + +The module interface supports the issue #32 goal that platform management should be a side service, not a model-execution abstraction. Modules can bootstrap deployment state, reconcile channels, or run control services without placing platform calls in the token-level model path. + +## Future work +- More modules are coming \ No newline at end of file diff --git a/platform/docs/modules/broadcast-deployment.md b/platform/docs/modules/broadcast-deployment.md new file mode 100644 index 0000000..569bf56 --- /dev/null +++ b/platform/docs/modules/broadcast-deployment.md @@ -0,0 +1,51 @@ +# Broadcast Deployment Module + +Source path: `include/modules/broadcastDeployment/module.hpp` + +The broadcast deployment module distributes a serialized deployment configuration from a deployer instance to other instances through RPC. + +## Main Type + +`serving::modules::broadcastDeployment::Module` derives from `serving::modules::Module`. + +Constructor inputs: + +- `HiCR::InstanceManager` +- `HiCR::ComputeManager` +- `HiCR::frontend::RPCEngine` +- Deployer instance id. +- Current instance id. +- Optional `serving::configuration::Deployment` for the deployer. + +The deployer constructor receives the deployment object. Worker instances use the constructor without a deployment and fetch it during initialization. + +## RPC Flow + +All instances register an RPC target named `__SERVING_REQUEST_DEPLOYMENT_CONFIGURATION_RPC_NAME`. + +When a worker initializes: + +1. It requests the deployment RPC from the deployer instance. +2. It waits for the return value. +3. It parses the returned JSON string. +4. It constructs a local `configuration::Deployment` from that JSON. +5. It frees the RPC return-value memory slot. + +When the deployer initializes: + +- It listens for deployment requests from non-deployer instances. +- On request, it serializes `_deployment` to JSON and submits the serialized string as the RPC return value. + +## Relation To Platform Design + +This module implements the static bootstrap direction from issue #32. It ensures all instances can obtain the same desired deployment state before channels, services, coordinators, and replicas are initialized. + +## Example + +`examples/modules/broadcastDeployment/broadcastDeployment.cpp` parses a deployment only on the root instance, creates the broadcast module on every instance, initializes the engine, and prints the received deployment on each instance. + +## Current Limitations + +- Deployment distribution is init-only and does not publish incremental updates. +- There is no versioning or consistency protocol for dynamic reconfiguration. +- The deployer listens in a loop based on the current instance list and assumes startup-time distribution. diff --git a/platform/docs/modules/channel-controller.md b/platform/docs/modules/channel-controller.md new file mode 100644 index 0000000..9116022 --- /dev/null +++ b/platform/docs/modules/channel-controller.md @@ -0,0 +1,63 @@ +# Channel Controller Module + +Source path: `include/modules/channelController/module.hpp` + +The channel controller manages channel creation through a desired-state reconciliation loop. It turns desired producer and consumer registrations into initialized HiCR-backed `Output` and `Input` channels. + +## Main Type + +`serving::modules::channelController::Module` derives from `serving::modules::Module` and runs as a periodic service by default. + +Constructor inputs: + +- Current instance id. +- Ordered list of `HiCR::CommunicationManager *` objects. +- Global memory-slot exchange tag. +- Reconciliation interval in milliseconds. + +At least one communication manager is required. + +## Desired State + +The controller keeps separate desired and actual maps for producers and consumers. + +`addDesiredProducer(targetId, channelId, edge, keyBuilder)` creates a desired `Output` channel for an edge. + +`addDesiredConsumer(sourceId, channelId, edge, keyBuilder)` creates a desired `Input` channel for an edge. + +`removeDesiredProducer(edgeName)` and `removeDesiredConsumer(edgeName)` remove desired entries. + +`hasProducer(edgeName)` and `hasConsumer(edgeName)` report whether actual channels exist. + +`getProducer(edgeName)` and `getConsumer(edgeName)` return initialized actual channels. + +## Reconciliation Flow + +The private `reconcile()` method: + +1. Diffs desired producers/consumers against actual producers/consumers. +2. Collects memory slots required by missing channels. +3. Groups slots by communication manager. +4. Exchanges global memory slots through the configured managers. +5. Fences each manager on the exchange tag. +6. Initializes missing `Output` and `Input` channels. +7. Moves initialized channels into actual maps. +8. Removes stale actual channels when desired entries disappear. + +The module calls `reconcile()` during `initialize()` and from its periodic `service()` method. + +## Relation To Platform Design + +This module implements the channel-management direction from issue #32: model support can request channels by desired intent, while platform code handles lifecycle and resource exchange. + +The current implementation works at the HiCR channel level. Future integration can add a higher-level distinction between host control channels and device tensor payload channels while preserving the same desired-state reconciliation model. + +## Example + +`examples/modules/channelController/channelController.cpp` parses a deployment, assigns edge managers, registers desired local producer/consumer channels, waits until both are ready, runs a telephone-game message exchange, removes desired channels, and terminates. + +## Current Limitations + +- The TODO in the source notes that every instance currently needs to run this service for MPI-style backends. +- The controller does not yet expose a frontend that hides backend-specific details from callers. +- Placement policy is provided indirectly through edge HiCR manager assignments. diff --git a/platform/docs/modules/configuration.md b/platform/docs/modules/configuration.md new file mode 100644 index 0000000..c89a192 --- /dev/null +++ b/platform/docs/modules/configuration.md @@ -0,0 +1,92 @@ +# Configuration Module + +Source paths: + +- `include/modules/configuration/deployment.hpp` +- `include/modules/configuration/partition.hpp` +- `include/modules/configuration/task.hpp` +- `include/modules/configuration/replica.hpp` +- `include/modules/configuration/edge.hpp` +- `include/modules/configuration/requestManager.hpp` + +The configuration module represents the desired deployment graph for the platform layer. It describes partitions, host-side tasks, replicas, request-manager endpoints, and communication edges. + +## Main Types + +`serving::configuration::Deployment` is the top-level object. It owns: + +- A deployment name. +- A list of `Partition` objects. +- A list of `Edge` objects. +- A `RequestManager` object. +- Heartbeat settings. +- Control-buffer settings. + +`Partition` describes a logical stage of the application. It stores: + +- A partition name. +- The coordinator instance id. +- A list of `Task` objects. +- A list of `Replica` objects. + +`Task` describes a named function within a partition. It stores: + +- Function name. +- Input edge names. +- Output edge names. +- Intra-partition dependencies by function name. + +`Replica` identifies a concrete instance assigned to execute a partition. + +`Edge` describes a communication link. It stores: + +- Edge name. +- Producer partition. +- Consumer partition. +- Buffer capacity and size. +- Payload HiCR manager objects. +- Coordination HiCR manager objects. +- Prompt/result flags for request-manager boundary edges. + +`RequestManager` defines the external entry and result edges for the deployment. + +## Serialization + +All configuration objects serialize to and deserialize from `nlohmann::json`. Runtime-only HiCR manager pointers and memory spaces are not serialized; they are assigned after parsing by runtime setup code. + +The serialized deployment shape includes: + +- `Name` +- `Partitions` +- `Edges` +- `Request Manager` +- `Settings` + +## Validation + +`Deployment::verify()` performs graph sanity checks and fills edge producer/consumer metadata. + +Important checks include: + +- Task dependencies must refer to tasks in the same partition. +- Tasks cannot depend on themselves. +- Edge names must not be duplicated. +- Inputs and outputs must refer to defined edges. +- Each non-request-manager edge must have exactly one producer and one consumer. +- An edge cannot connect a partition to itself, except request-manager prompt/result edges. +- The request-manager input and output edges must be used. +- The partition consuming the external input must not also consume cross-partition inputs. +- The partition producing the external output must not also produce cross-partition outputs. + +After validation, request-manager boundary edges are marked as prompt or result edges. + +## Relation To Platform Design + +The deployment graph is the current implementation of the static deployment API described in issue #32. It gives the platform a desired-state view of partitions, tasks, replicas, and edges before dynamic scaling or fault recovery are added. + +## Current Limitations + +- Placement concepts such as host/device placement are not explicit JSON fields yet. +- Replicas and coordinator instance ids are optional because they may be assigned at runtime, but no placement optimizer is implemented here. +- Heartbeat settings are represented, but recovery policy is not implemented in this module. +- Edge payload/control placement is represented through runtime HiCR manager assignments rather than a high-level placement enum. diff --git a/platform/docs/modules/service.md b/platform/docs/modules/service.md new file mode 100644 index 0000000..8c3ca83 --- /dev/null +++ b/platform/docs/modules/service.md @@ -0,0 +1,41 @@ +# Service Module + +Source path: `include/modules/service/module.hpp` + +The service module adapts TaskR services into the platform module lifecycle. It lets the platform run one or more periodic control services even when there are no normal TaskR tasks. + +## Main Type + +`serving::modules::service::Module` derives from `serving::modules::Module` and owns a shared `taskr::Runtime`. + +It maintains a map from service name to non-owning `taskr::Service *`. + +## Behavior + +`addService(name, service)` registers a service with the module. Names must be unique. + +During `initialize()`: + +- The TaskR runtime is configured with `setFinishOnLastTask(false)` so service-only workloads do not finish immediately. +- Registered services are added to the TaskR runtime. +- The TaskR runtime is initialized. + +During `run()`: + +- The TaskR runtime starts running. + +During `terminate()`: + +- The TaskR runtime is switched back to `setFinishOnLastTask(true)` so it can exit. + +During `await()` and `finalize()`: + +- The module waits for TaskR completion, finalizes the runtime, and clears the service registry. + +## Example + +`examples/modules/service/service.cpp` creates a simple periodic `helloWorld` service, registers it with `service::Module`, installs the module into `serving::system::Engine`, and terminates from the root instance after a short delay. + +## Relation To Platform Design + +This module supports the passive platform-management role from issues #32 and #13. After deployment bootstrap, periodic services such as heartbeat, monitoring, channel reconciliation, or future scaling decisions can run through TaskR without becoming part of model execution. \ No newline at end of file diff --git a/platform/docs/simpler-integration.md b/platform/docs/simpler-integration.md new file mode 100644 index 0000000..16b90ef --- /dev/null +++ b/platform/docs/simpler-integration.md @@ -0,0 +1,72 @@ +# Simpler Integration + +## Which ranks launch Simpler + +Only **replica** ranks initialize a Simpler runtime instance. Coordinator and request manager ranks are pure host-side and never touch device memory or Simpler APIs. + +Device assignment: `deviceId = instanceId - numPartitions` + +With two partitions and one replica each, ranks 0 and 1 are coordinators, ranks 2 and 3 are replicas owning devices 0 and 1 respectively. + +## Initialization + +Each replica rank initializes Simpler in `main()` before the serving engine starts: + +```cpp +hllm::simpler::RuntimeBinaries bins{hostLib, aicpuLib, aicoreKernel, dispatcherLib, simplerLogLib}; +auto rt = std::make_unique(); +rt->init(bins, deviceId); +mnist::loadKernels(*rt, artifactDir); +``` + +`RuntimeBinaries` holds five paths to compiled runtime artifacts: + +| Field | File | Purpose | +|---|---|---| +| `host` | `libhost_runtime.so` | Core device runtime | +| `aicpu` | `libaicpu_kernel.so` | CPU-side kernel support | +| `aicore` | `aicore_kernel.o` | NPU core kernel binary | +| `dispatcher` | `libsimpler_aicpu_dispatcher.so` | On-device dispatcher | +| `simplerLog` | `libsimpler_log.so` | Logging (preloaded RTLD_GLOBAL) | + +Model Support is responsible for providing and locating these artifacts. The platform does not interpret or validate them. + +## `processFc` — the task execution callback + +The replica module accepts a user-provided function with signature: + +```cpp +std::function +``` + +The platform calls this function once per job, after all input dependencies have arrived. Inside the function, Model Support: + +1. Reads inputs from `context.getInput(edgeName)` — returns a `LocalMemorySlot` containing the host buffer +2. Stages inputs to device with `rt->toDevice(ptr, bytes)` +3. Dispatches a Simpler kernel with `rt->run(callableId, args, config)` +4. Stages outputs back with `rt->toHost(hostPtr, devPtr, bytes)` +5. Registers outputs with `context.setOutput(edgeName, ptr, size)` + +The platform guarantees that: +- All declared inputs are present and ready before the call +- Outputs registered via `setOutput` are forwarded to the coordinator after the call returns +- The function is called at most once per job; re-entrancy is not required + +## `SimplerRuntime` API surface + +```cpp +void init(const RuntimeBinaries &bins, int deviceId); +void loadCallable(const void *blob, size_t size, uint32_t id); +void run(uint32_t callableId, ChipStorageTaskArgs &args, ChipCallConfig &config); +void *toDevice(const void *hostPtr, size_t bytes); +void toHost(void *hostPtr, const void *devPtr, size_t bytes); +void *alloc(size_t bytes); +void free(void *devPtr); +void finalize(); +``` + +## Channel model and hot path + +All channels in the current platform are host-side. Tensor payloads passed through `processFc` are staged through host memory (`toDevice` / `toHost`). This is correct for control traffic and small tensors but adds latency on the hot path for large prefill/decode tensors. + +**In-device tensor channels** (direct NPU-to-NPU transfer without host staging) are the next milestone. When implemented, the `processFc` interface will remain the same — Model Support will simply receive device-memory `LocalMemorySlot` handles instead of host buffers, and the staging calls become unnecessary. diff --git a/platform/examples/include/channels/helpers.hpp b/platform/examples/include/channels/helpers.hpp new file mode 100644 index 0000000..08fd3e3 --- /dev/null +++ b/platform/examples/include/channels/helpers.hpp @@ -0,0 +1,131 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include +#include + +#include + +struct localChannels_t +{ + std::vector inputs; + std::vector outputs; +}; + +struct singleLocalChannels_t +{ + localInput_t input; + localOutput_t output; +}; + +struct desiredSingleLocalChannels_t +{ + std::shared_ptr input; + std::shared_ptr output; + std::string inputName; + std::string outputName; + localInput_t inputInfo; + localOutput_t outputInfo; +}; + +__INLINE__ localChannels_t buildLocalChannelInfos(const serving::configuration::Deployment &deployment, + const HiCR::Instance::instanceId_t instanceId, + const serving::system::channels::keyBuilderFc_t &keyBuilder) +{ + localChannels_t channels; + buildLocalChannelsFromDeploymentWithIds(deployment, instanceId, keyBuilder, channels.inputs, channels.outputs); + return channels; +} + +__INLINE__ singleLocalChannels_t buildSingleLocalChannelInfos(const serving::configuration::Deployment &deployment, + const HiCR::Instance::instanceId_t instanceId, + const serving::system::channels::keyBuilderFc_t &keyBuilder) +{ + auto channels = buildLocalChannelInfos(deployment, instanceId, keyBuilder); + if (channels.inputs.size() != 1) HICR_THROW_LOGIC("Expected exactly one local input channel, got %lu.", channels.inputs.size()); + if (channels.outputs.size() != 1) HICR_THROW_LOGIC("Expected exactly one local output channel, got %lu.", channels.outputs.size()); + return singleLocalChannels_t{ + .input = channels.inputs.front(), + .output = channels.outputs.front(), + }; +} + +__INLINE__ std::shared_ptr createDesiredInput(const std::shared_ptr &channelController, + const localInput_t &inputInfo, + const serving::system::channels::keyBuilderFc_t &keyBuilder) +{ + auto input = channelController->addDesiredConsumer(inputInfo.sourceInstanceId, inputInfo.channelId, *inputInfo.edge, keyBuilder).lock(); + return input; +} + +__INLINE__ std::shared_ptr createDesiredOutput(const std::shared_ptr &channelController, + const localOutput_t &outputInfo, + const serving::system::channels::keyBuilderFc_t &keyBuilder) +{ + auto output = channelController->addDesiredProducer(outputInfo.targetInstanceId, outputInfo.channelId, *outputInfo.edge, keyBuilder).lock(); + return output; +} + +__INLINE__ desiredSingleLocalChannels_t createDesiredSingleLocalChannels(const serving::configuration::Deployment &deployment, + const HiCR::Instance::instanceId_t instanceId, + const serving::system::channels::keyBuilderFc_t &keyBuilder, + const std::shared_ptr &channelController) +{ + auto infos = buildSingleLocalChannelInfos(deployment, instanceId, keyBuilder); + auto input = createDesiredInput(channelController, infos.input, keyBuilder); + auto output = createDesiredOutput(channelController, infos.output, keyBuilder); + return desiredSingleLocalChannels_t{ + .input = input, + .output = output, + .inputName = infos.input.edge->getName(), + .outputName = infos.output.edge->getName(), + .inputInfo = infos.input, + .outputInfo = infos.output, + }; +} + +__INLINE__ void removeDesiredInputs(const std::shared_ptr &channelController, const std::vector &inputs) +{ + for (const auto &input : inputs) { channelController->removeDesiredConsumer(input.edge->getName()); } +} + +__INLINE__ void removeDesiredOutputs(const std::shared_ptr &channelController, const std::vector &outputs) +{ + for (const auto &output : outputs) { channelController->removeDesiredProducer(output.edge->getName()); } +} + +__INLINE__ void removeDesiredSingleLocalChannels(const std::shared_ptr &channelController, desiredSingleLocalChannels_t &channels) +{ + channelController->removeDesiredConsumer(channels.inputName); + channelController->removeDesiredProducer(channels.outputName); + channels.input = {}; + channels.output = {}; +} + +__INLINE__ void waitUntilReady(const std::shared_ptr &channel, const size_t intervalMs = 500) +{ + while (!channel->isReady()) { std::this_thread::sleep_for(std::chrono::milliseconds(intervalMs)); } +} + +__INLINE__ serving::configuration::Edge makeInternalEdgeFromTemplate(const std::string &name, const serving::configuration::Edge &edgeTemplate) +{ + serving::configuration::Edge edge(name, edgeTemplate.getBufferCapacity(), edgeTemplate.getBufferSize()); + edge.setPayloadCommunicationManager(edgeTemplate.getPayloadCommunicationManager()); + edge.setPayloadMemoryManager(edgeTemplate.getPayloadMemoryManager()); + edge.setPayloadMemorySpace(edgeTemplate.getPayloadMemorySpace()); + edge.setCoordinationCommunicationManager(edgeTemplate.getCoordinationCommunicationManager()); + edge.setCoordinationMemoryManager(edgeTemplate.getCoordinationMemoryManager()); + edge.setCoordinationMemorySpace(edgeTemplate.getCoordinationMemorySpace()); + return edge; +} \ No newline at end of file diff --git a/platform/examples/include/deployment/helpers.hpp b/platform/examples/include/deployment/helpers.hpp new file mode 100644 index 0000000..5fe05b8 --- /dev/null +++ b/platform/examples/include/deployment/helpers.hpp @@ -0,0 +1,236 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include + +struct localInput_t +{ + HiCR::Instance::instanceId_t sourceInstanceId = -1; + HiCR::Instance::instanceId_t targetInstanceId = -1; + serving::system::channels::channelId_t channelId = -1; + std::shared_ptr edge = {}; + std::shared_ptr channel = nullptr; // optional: used by legacy examples +}; + +struct localOutput_t +{ + HiCR::Instance::instanceId_t sourceInstanceId = -1; + HiCR::Instance::instanceId_t targetInstanceId = -1; + serving::system::channels::channelId_t channelId = -1; + std::shared_ptr edge = {}; + std::shared_ptr channel; // optional: used by legacy examples +}; + +__INLINE__ serving::system::channels::slotKeys_t defaultChannelKeyBuilder(const HiCR::Instance::instanceId_t sourceInstanceId, + const HiCR::Instance::instanceId_t targetInstanceId, + const serving::system::channels::channelId_t channelId) +{ + using key_t = HiCR::GlobalMemorySlot::globalKey_t; + const key_t src = (key_t(sourceInstanceId) & ((1ull << 20) - 1)) << 44; + const key_t dst = (key_t(targetInstanceId) & ((1ull << 20) - 1)) << 24; + const key_t ch = (key_t(channelId) & ((1ull << 20) - 1)) << 4; + auto makeKey = [&](const key_t slot) -> key_t { return src | dst | ch | (slot & 0xFull); }; + return { + .dataConsumerSizesBufferKey = makeKey(0), + .dataConsumerPayloadBufferKey = makeKey(1), + .dataConsumerCoordinationBufferForSizesKey = makeKey(2), + .dataConsumerCoordinationBufferForPayloadKey = makeKey(3), + .dataProducerCoordinationBufferForSizesKey = makeKey(4), + .dataProducerCoordinationBufferForPayloadKey = makeKey(5), + .metadataConsumerPayloadBufferKey = makeKey(6), + .metadataConsumerCoordinationBufferKey = makeKey(7), + .metadataProducerCoordinationBufferKey = makeKey(8), + }; +} + +__INLINE__ void assignEdgeManagers(serving::configuration::Deployment &deployment, + HiCR::CommunicationManager *communicationManager, + HiCR::MemoryManager *memoryManager, + const std::shared_ptr &memorySpace) +{ + for (const auto &edge : deployment.getEdges()) + { + edge->setPayloadCommunicationManager(communicationManager); + edge->setPayloadMemoryManager(memoryManager); + edge->setPayloadMemorySpace(memorySpace); + edge->setCoordinationCommunicationManager(communicationManager); + edge->setCoordinationMemoryManager(memoryManager); + edge->setCoordinationMemorySpace(memorySpace); + } +} + +__INLINE__ void inferEdgeEndpointsFromTasks(serving::configuration::Deployment &deployment) +{ + std::set edgeNameSet; + + for (const auto &edge : deployment.getEdges()) + { + const auto &edgeName = edge->getName(); + if (edgeNameSet.contains(edgeName)) HICR_THROW_LOGIC("Repeated edge name '%s' in deployment.", edgeName.c_str()); + edgeNameSet.insert(edgeName); + } + + std::map producerPartitionMap; + std::map consumerPartitionMap; + + for (const auto &partition : deployment.getPartitions()) + { + const auto &partitionName = partition->getName(); + + for (const auto &task : partition->getTasks()) + { + for (const auto &output : task->getOutputs()) + { + if (edgeNameSet.contains(output) == false) HICR_THROW_LOGIC("Task '%s' references undefined output edge '%s'.", task->getFunctionName().c_str(), output.c_str()); + if (producerPartitionMap.contains(output)) + HICR_THROW_LOGIC("Edge '%s' has multiple producer partitions ('%s' and '%s').", output.c_str(), producerPartitionMap.at(output).c_str(), partitionName.c_str()); + producerPartitionMap[output] = partitionName; + } + for (const auto &input : task->getInputs()) + { + if (edgeNameSet.contains(input) == false) HICR_THROW_LOGIC("Task '%s' references undefined input edge '%s'.", task->getFunctionName().c_str(), input.c_str()); + if (consumerPartitionMap.contains(input)) + HICR_THROW_LOGIC("Edge '%s' has multiple consumer partitions ('%s' and '%s').", input.c_str(), consumerPartitionMap.at(input).c_str(), partitionName.c_str()); + consumerPartitionMap[input] = partitionName; + } + } + } + + for (const auto &edge : deployment.getEdges()) + { + const auto &edgeName = edge->getName(); + + if (producerPartitionMap.contains(edgeName) == false) HICR_THROW_LOGIC("Edge '%s' is never produced by any task.", edgeName.c_str()); + if (consumerPartitionMap.contains(edgeName) == false) HICR_THROW_LOGIC("Edge '%s' is never consumed by any task.", edgeName.c_str()); + + const auto &producer = producerPartitionMap.at(edgeName); + const auto &consumer = consumerPartitionMap.at(edgeName); + + if (producer == consumer) HICR_THROW_LOGIC("Edge '%s' is both produced and consumed by partition '%s'.", edgeName.c_str(), producer.c_str()); + + edge->setProducer(producer); + edge->setConsumer(consumer); + } +} + +__INLINE__ void buildLocalChannelsFromDeploymentWithIds(const serving::configuration::Deployment &deployment, + const HiCR::Instance::instanceId_t myInstanceId, + const serving::system::channels::keyBuilderFc_t &keyBuilder, + std::vector &inputs, + std::vector &outputs) +{ + std::map partitionToInstance; + + for (const auto &partition : deployment.getPartitions()) partitionToInstance[partition->getName()] = partition->getCoordinatorInstanceId(); + + for (serving::configuration::Edge::edgeIndex_t edgeIdx = 0; edgeIdx < deployment.getEdges().size(); edgeIdx++) + { + const auto &edge = deployment.getEdges()[edgeIdx]; + const auto &producerPartition = edge->getProducer(); + const auto &consumerPartition = edge->getConsumer(); + + if (partitionToInstance.contains(producerPartition) == false) + HICR_THROW_LOGIC("Edge '%s' producer partition '%s' is not present in deployment partition map.", edge->getName().c_str(), producerPartition.c_str()); + if (partitionToInstance.contains(consumerPartition) == false) + HICR_THROW_LOGIC("Edge '%s' consumer partition '%s' is not present in deployment partition map.", edge->getName().c_str(), consumerPartition.c_str()); + + const auto sourceInstanceId = partitionToInstance.at(producerPartition); + const auto targetInstanceId = partitionToInstance.at(consumerPartition); + const auto channelId = static_cast(edgeIdx); + + if (myInstanceId == sourceInstanceId) + { + outputs.push_back(localOutput_t{ + .sourceInstanceId = sourceInstanceId, + .targetInstanceId = targetInstanceId, + .channelId = channelId, + .edge = edge, + .channel = std::make_shared(*edge, channelId, sourceInstanceId, targetInstanceId, keyBuilder), + }); + } + + if (myInstanceId == targetInstanceId) + { + inputs.push_back(localInput_t{ + .sourceInstanceId = sourceInstanceId, + .targetInstanceId = targetInstanceId, + .channelId = channelId, + .edge = edge, + .channel = std::make_shared(*edge, channelId, sourceInstanceId, targetInstanceId, keyBuilder), + }); + } + } +} + +// Compatibility helper for legacy examples that still consume raw channel vectors. +__INLINE__ void buildLocalChannelsFromDeployment(const serving::configuration::Deployment &deployment, + const HiCR::Instance::instanceId_t myInstanceId, + const serving::system::channels::keyBuilderFc_t &keyBuilder, + std::vector> &inputs, + std::vector> &outputs) +{ + std::vector localInputs; + std::vector localOutputs; + + buildLocalChannelsFromDeploymentWithIds(deployment, myInstanceId, keyBuilder, localInputs, localOutputs); + + inputs.reserve(inputs.size() + localInputs.size()); + outputs.reserve(outputs.size() + localOutputs.size()); + + for (const auto &in : localInputs) inputs.push_back(in.channel); + for (const auto &out : localOutputs) outputs.push_back(out.channel); +} + +__INLINE__ void readAndParseConfiguration(char *argv[], + serving::configuration::Deployment &deployment, + std::shared_ptr &instanceManager, + const size_t replicasPerPartition = 0) +{ + const std::string servingConfigFilePath = std::string(argv[1]); + std::ifstream servingConfigFs(servingConfigFilePath); + auto servingConfigJs = nlohmann::json::parse(servingConfigFs); + + deployment.deserialize(servingConfigJs); + + inferEdgeEndpointsFromTasks(deployment); + + // The coordinator is co-located with the first replica, so each partition needs + // 1 instance for the coordinator + (replicasPerPartition - 1) extra replica instances. + const auto extraReplicasPerPartition = replicasPerPartition > 0 ? replicasPerPartition - 1 : 0; + const auto instancesRequired = deployment.getPartitions().size() + (deployment.getPartitions().size() * extraReplicasPerPartition); + + if (instanceManager->getInstances().size() < instancesRequired) + { + fprintf(stderr, "Error: %zu instances provided, but %lu are required\n", instanceManager->getInstances().size(), instancesRequired); + instanceManager->abort(-1); + } + + auto instance = instanceManager->getInstances().begin(); + + for (auto &partition : deployment.getPartitions()) + { + const auto partitionInstanceId = (*instance)->getId(); + ++instance; + partition->setCoordinatorInstanceId(partitionInstanceId); + for (size_t i = 0; i < replicasPerPartition; i++) + { + // First replica is co-located with the coordinator; subsequent ones get their own instance. + const auto replicaInstanceId = (i == 0) ? partitionInstanceId : (*instance)->getId(); + if (i > 0) ++instance; + partition->addReplica(std::make_shared(replicaInstanceId)); + } + } +} \ No newline at end of file diff --git a/platform/examples/include/runtime/helpers.hpp b/platform/examples/include/runtime/helpers.hpp new file mode 100644 index 0000000..3548dce --- /dev/null +++ b/platform/examples/include/runtime/helpers.hpp @@ -0,0 +1,82 @@ +#pragma once +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +struct Runtime +{ + hwloc_topology_t hwlocTopologyObject; + std::shared_ptr instanceManager; + std::shared_ptr communicationManager; + std::shared_ptr memoryManager; + std::shared_ptr workerComputeManager; + std::shared_ptr taskComputeManager; + std::shared_ptr rpcEngine; + std::shared_ptr taskr; + std::shared_ptr bufferMemorySpace; + std::vector> computeResources; + std::unique_ptr serving; +}; + +__INLINE__ nlohmann::json makeDefaultTaskrConfig() +{ + nlohmann::json taskrConfig; + taskrConfig["Task Worker Inactivity Time (Ms)"] = 500; + taskrConfig["Task Suspend Interval Time (Ms)"] = 500; + taskrConfig["Minimum Active Task Workers"] = 1; + taskrConfig["Service Worker Count"] = 1; + taskrConfig["Make Task Workers Run Services"] = true; + return taskrConfig; +} + +__INLINE__ Runtime makeRuntime(int *argc, char ***argv, const size_t computeResourceCount = 2) +{ + Runtime runtime; + + hwloc_topology_init(&runtime.hwlocTopologyObject); + HiCR::backend::hwloc::TopologyManager hwlocTopologyManager(&runtime.hwlocTopologyObject); + + const auto topology = hwlocTopologyManager.queryTopology(); + auto device = *topology.getDevices().begin(); + auto memorySpaces = device->getMemorySpaceList(); + runtime.bufferMemorySpace = *memorySpaces.begin(); + auto computeResourcesIt = device->getComputeResourceList().begin(); + for (size_t i = 0; i < computeResourceCount; i++) + { + runtime.computeResources.push_back(*computeResourcesIt); + ++computeResourcesIt; + } + auto rpcComputeResource = *runtime.computeResources.begin(); + + runtime.instanceManager = std::shared_ptr(HiCR::backend::mpi::InstanceManager::createDefault(argc, argv)); + runtime.communicationManager = std::make_shared(); + runtime.memoryManager = std::make_shared(); + runtime.workerComputeManager = std::make_shared(); + runtime.taskComputeManager = std::make_shared(); + + runtime.rpcEngine = std::make_shared( + *runtime.communicationManager, *runtime.instanceManager, *runtime.memoryManager, *runtime.workerComputeManager, runtime.bufferMemorySpace, rpcComputeResource); + runtime.rpcEngine->initialize(); + + if (runtime.computeResources.size() != 2) { HICR_THROW_RUNTIME("Too many CR"); } + runtime.taskr = std::make_shared(runtime.taskComputeManager.get(), runtime.workerComputeManager.get(), runtime.computeResources, makeDefaultTaskrConfig()); + + runtime.serving = std::make_unique(runtime.instanceManager, runtime.taskComputeManager, runtime.rpcEngine, runtime.instanceManager->getRootInstanceId()); + return runtime; +} \ No newline at end of file diff --git a/platform/examples/meson.build b/platform/examples/meson.build index 860f6d7..9c95df8 100644 --- a/platform/examples/meson.build +++ b/platform/examples/meson.build @@ -1 +1,7 @@ +modulesDependency = declare_dependency(include_directories: include_directories('include')) + +subdir('modules/broadcastDeployment') +subdir('modules/channelController') +subdir('modules/service') + subdir('system/engine') diff --git a/platform/examples/modules/broadcastDeployment/broadcastDeployment.cpp b/platform/examples/modules/broadcastDeployment/broadcastDeployment.cpp new file mode 100644 index 0000000..96454fa --- /dev/null +++ b/platform/examples/modules/broadcastDeployment/broadcastDeployment.cpp @@ -0,0 +1,84 @@ +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include + +int main(int argc, char *argv[]) +{ + auto runtime = makeRuntime(&argc, &argv); + const auto &instance = runtime.instanceManager->getCurrentInstance(); + auto &serving = *runtime.serving; + + // Check whether the instance is root + const auto isRoot = runtime.instanceManager->getCurrentInstance()->isRootInstance(); + const auto instanceId = runtime.instanceManager->getCurrentInstance()->getId(); + const auto deployerInstanceId = runtime.instanceManager->getRootInstanceId(); + + ///// Configuration parsing + serving::configuration::Deployment deployment; + + // If I am root, checking arguments. + // Do not assume other instances will have the correct arguments set (e.g., file that exists only on root instance) + if (isRoot == true) + { + if (argc != 2) + { + fprintf(stderr, "Error: Must provide the config file path.\n"); + runtime.instanceManager->abort(-1); + } + // Read and parse config file + readAndParseConfiguration(argv, deployment, runtime.instanceManager); + } + + std::shared_ptr broadcastDeploymentModule; + if (instanceId == deployerInstanceId) + { + broadcastDeploymentModule = std::make_shared( + runtime.instanceManager, runtime.taskComputeManager, runtime.rpcEngine, deployerInstanceId, instanceId, deployment); + } + else + { + broadcastDeploymentModule = + std::make_shared(runtime.instanceManager, runtime.taskComputeManager, runtime.rpcEngine, deployerInstanceId, instanceId); + } + + const auto &receivedDeployment = broadcastDeploymentModule->getDeployment(); + // Adding broadcast deployment module to serving + serving.addModule("BroadcastDeployment", broadcastDeploymentModule); + + // Initializing serving + serving.initialize(); + + // Running serving + serving.run(); + + // Finalizing serving + serving.terminate(); + + // Awaiting serving termination + serving.await(); + + // Printing deployment information to verify it was correctly received + std::this_thread::sleep_for(std::chrono::seconds(instanceId)); // Sleep a bit to ensure the output is not mixed + printf("[Instance %lu] Received deployment configuration:\n%s\n", instanceId, receivedDeployment.serialize().dump(2).c_str()); + + // Finalize Instance Manager + runtime.instanceManager->finalize(); +} diff --git a/platform/examples/modules/broadcastDeployment/meson.build b/platform/examples/modules/broadcastDeployment/meson.build new file mode 100644 index 0000000..cdab03f --- /dev/null +++ b/platform/examples/modules/broadcastDeployment/meson.build @@ -0,0 +1,16 @@ +testSuite = ['examples', 'modules'] + +e = executable('broadcastDeployment', ['broadcastDeployment.cpp'], dependencies: [servingBuildDep, modulesDependency]) +if get_option('buildTests') + test( + 'broadcastDeployment', + mpirunExecutable, + args: [ + '-np', '5', + '--oversubscribe', e.full_path(), + meson.current_source_dir() + '/policy.json', + ], + timeout: 60, + suite: testSuite, + ) +endif \ No newline at end of file diff --git a/platform/examples/modules/broadcastDeployment/policy.json b/platform/examples/modules/broadcastDeployment/policy.json new file mode 100644 index 0000000..177bebe --- /dev/null +++ b/platform/examples/modules/broadcastDeployment/policy.json @@ -0,0 +1,42 @@ +{ + "Name": "Basic Deployment", + "Settings": { + "Heartbeat": { + "Enabled": true, + "Visible": false, + "Interval": 500, + "Tolerance": 1000 + }, + "Control Buffer": { + "Capacity": 32, + "Size": 16384 + } + }, + "Request Manager": { + "Input": "input", + "Output": "output" + }, + "Partitions": [ + { + "Name": "P0", + "Tasks": [] + }, + { + "Name": "P1", + "Tasks": [] + }, + { + "Name": "P2", + "Tasks": [] + }, + { + "Name": "P3", + "Tasks": [] + }, + { + "Name": "P4", + "Tasks": [] + } + ], + "Edges": [] +} \ No newline at end of file diff --git a/platform/examples/modules/channelController/channelController.cpp b/platform/examples/modules/channelController/channelController.cpp new file mode 100644 index 0000000..0e96473 --- /dev/null +++ b/platform/examples/modules/channelController/channelController.cpp @@ -0,0 +1,75 @@ +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include +#include +#include "telephoneGame.hpp" + +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; + + serving::configuration::Deployment deployment; + + if (argc != 2) + { + fprintf(stderr, "Error: Must provide the config file path.\n"); + runtime.instanceManager->abort(-1); + return -1; + } + + readAndParseConfiguration(argv, deployment, runtime.instanceManager); + assignEdgeManagers(deployment, runtime.communicationManager.get(), runtime.memoryManager.get(), runtime.bufferMemorySpace); + + std::vector managerOrder = {runtime.communicationManager.get()}; + auto channelControllerModule = std::make_shared(instanceId, managerOrder); + + auto channels = createDesiredSingleLocalChannels(deployment, instanceId, defaultChannelKeyBuilder, channelControllerModule); + + auto serviceModule = std::make_shared(runtime.taskr); + serviceModule->addService("ChannelController", channelControllerModule->getService()); + + serving.addModule("ChannelController", channelControllerModule); + serving.addModule("Service", serviceModule); + + serving.initialize(); + + serving.run(); + + waitUntilReady(channels.input); + waitUntilReady(channels.output); + + telephoneGame(*channels.input, *channels.output, instanceId, isRoot); + + removeDesiredSingleLocalChannels(channelControllerModule, channels); + + if (isRoot) serving.terminate(); + + serving.await(); + + runtime.instanceManager->finalize(); + + return 0; +} diff --git a/platform/examples/modules/channelController/meson.build b/platform/examples/modules/channelController/meson.build new file mode 100644 index 0000000..76b99ff --- /dev/null +++ b/platform/examples/modules/channelController/meson.build @@ -0,0 +1,16 @@ +testSuite = ['examples', 'modules'] + +e = executable('channelController', ['channelController.cpp'], dependencies: [servingBuildDep, modulesDependency]) +if get_option('buildTests') + test( + 'channelController', + mpirunExecutable, + args: [ + '-np', '5', + '--oversubscribe', e.full_path(), + meson.current_source_dir() + '/policy.json', + ], + timeout: 60, + suite: testSuite, + ) +endif \ No newline at end of file diff --git a/platform/examples/modules/channelController/policy.json b/platform/examples/modules/channelController/policy.json new file mode 100644 index 0000000..7e63bcc --- /dev/null +++ b/platform/examples/modules/channelController/policy.json @@ -0,0 +1,118 @@ +{ + "Name": "Basic Deployment", + "Settings": { + "Heartbeat": { + "Enabled": true, + "Visible": false, + "Interval": 500, + "Tolerance": 1000 + }, + "Control Buffer": { + "Capacity": 32, + "Size": 16384 + } + }, + "Request Manager": { + "Input": "", + "Output": "" + }, + "Partitions": [ + { + "Name": "P0", + "Tasks": [ + { + "Function Name": "T0", + "Inputs": [ + "Channel 0" + ], + "Outputs": [ + "Channel 1" + ], + "Dependencies": [] + } + ] + }, + { + "Name": "P1", + "Tasks": [ + { + "Function Name": "T1", + "Inputs": [ + "Channel 1" + ], + "Outputs": [ + "Channel 2" + ], + "Dependencies": [] + } + ] + }, + { + "Name": "P2", + "Tasks": [ + { + "Function Name": "T2", + "Inputs": [ + "Channel 2" + ], + "Outputs": [ + "Channel 3" + ], + "Dependencies": [] + } + ] + }, + { + "Name": "P3", + "Tasks": [ + { + "Function Name": "T3", + "Inputs": [ + "Channel 3" + ], + "Outputs": [ + "Channel 4" + ], + "Dependencies": [] + } + ] + }, + { + "Name": "P4", + "Tasks": [ + { + "Function Name": "T4", + "Inputs": [ + "Channel 4" + ], + "Outputs": [ + "Channel 0" + ], + "Dependencies": [] + } + ] + } + ], + "Edges": [ + { + "Name": "Channel 0", + "Buffer Size": 4096 + }, + { + "Name": "Channel 1", + "Buffer Size": 4096 + }, + { + "Name": "Channel 2", + "Buffer Size": 4096 + }, + { + "Name": "Channel 3", + "Buffer Size": 4096 + }, + { + "Name": "Channel 4", + "Buffer Size": 4096 + } + ] +} \ No newline at end of file diff --git a/platform/examples/modules/channelController/telephoneGame.hpp b/platform/examples/modules/channelController/telephoneGame.hpp new file mode 100644 index 0000000..e0d3cee --- /dev/null +++ b/platform/examples/modules/channelController/telephoneGame.hpp @@ -0,0 +1,48 @@ +#pragma once + +#include +#include +#include + +#include + +#include +#include +#include + +#define _REPLICAS_PER_PARTITION 1 + +__INLINE__ void telephoneGame(serving::system::channels::Input &inputChannel, + serving::system::channels::Output &outputChannel, + const HiCR::Instance::instanceId_t instanceId, + const bool isRoot) +{ + if (isRoot) + { + // Root instance starts the game by sending a message to the next instance + const std::string text = "Hello from root instance!"; + printf("[Instance %lu][TelephoneGame] Sending message: %s\n", instanceId, text.c_str()); + auto input = serving::system::channels::Message(reinterpret_cast(text.data()), text.size(), serving::system::channels::Message::metadata_t{}); + outputChannel.pushMessageLocking(input); + + // wait for the return message + while (inputChannel.hasMessage() == false) { std::this_thread::sleep_for(std::chrono::milliseconds(500)); } + auto output = inputChannel.getMessage(); + + printf("[Instance %lu][TelephoneGame] Received message: %s\n", instanceId, std::string(reinterpret_cast(output.getData()), output.getSize()).c_str()); + inputChannel.popMessage(); + } + else + { + while (inputChannel.hasMessage() == false) { std::this_thread::sleep_for(std::chrono::milliseconds(500)); } + auto input = inputChannel.getMessage(); + + printf("[Instance %lu][TelephoneGame] Received message: %s\n", instanceId, std::string(reinterpret_cast(input.getData()), input.getSize()).c_str()); + + auto text = std::string(reinterpret_cast(input.getData()), input.getSize()); + printf("[Instance %lu][TelephoneGame] Sending message: %s\n", instanceId, text.c_str()); + auto output = serving::system::channels::Message(reinterpret_cast(text.data()), text.size(), serving::system::channels::Message::metadata_t{}); + outputChannel.pushMessageLocking(output); + inputChannel.popMessage(); + } +} \ No newline at end of file diff --git a/platform/examples/modules/service/meson.build b/platform/examples/modules/service/meson.build new file mode 100644 index 0000000..6f9f85c --- /dev/null +++ b/platform/examples/modules/service/meson.build @@ -0,0 +1,15 @@ +testSuite = ['examples', 'modules'] + +e = executable('service', ['service.cpp'], dependencies: [servingBuildDep, modulesDependency]) +if get_option('buildTests') + test( + 'service', + mpirunExecutable, + args: [ + '-np', '5', + '--oversubscribe', e.full_path() + ], + timeout: 60, + suite: testSuite, + ) +endif \ No newline at end of file diff --git a/platform/examples/modules/service/service.cpp b/platform/examples/modules/service/service.cpp new file mode 100644 index 0000000..b6df7aa --- /dev/null +++ b/platform/examples/modules/service/service.cpp @@ -0,0 +1,119 @@ +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +int main(int argc, char *argv[]) +{ + // Creating HWloc topology object + hwloc_topology_t hwlocTopologyObject; + + // Reserving memory for hwloc + hwloc_topology_init(&hwlocTopologyObject); + + // Initializing host (CPU) topology manager + HiCR::backend::hwloc::TopologyManager hwlocTopologyManager(&hwlocTopologyObject); + + // Gathering topology from the topology manager + const auto topology = hwlocTopologyManager.queryTopology(); + + auto d = *topology.getDevices().begin(); + auto memSpaces = d->getMemorySpaceList(); + if (memSpaces.empty()) HICR_THROW_RUNTIME("No memory spaces found on the queried device"); + auto bufferMemorySpace = *memSpaces.begin(); + + const auto &availableComputeResources = d->getComputeResourceList(); + if (availableComputeResources.size() < 2) HICR_THROW_RUNTIME("Fewer than 2 compute resources available"); + auto computeResourcesIt = availableComputeResources.begin(); + + // Use only 2 cores + std::vector> computeResources; + for (int i = 0; i < 2; i++) + { + computeResources.push_back(*computeResourcesIt); + computeResourcesIt++; + } + auto computeResource = *computeResources.begin(); + + // Getting managers + auto instanceManager = std::shared_ptr(HiCR::backend::mpi::InstanceManager::createDefault(&argc, &argv)); + auto communicationManager = std::make_shared(); + auto memoryManager = std::make_shared(); + auto workerComputeManager = std::make_shared(); + auto taskComputeManager = std::make_shared(); + + // Instantiate RPC Engine + auto rpcEngine = std::make_shared(*communicationManager, *instanceManager, *memoryManager, *workerComputeManager, bufferMemorySpace, computeResource); + + // Initialize RPC Engine + rpcEngine->initialize(); + + // Creating taskr object + nlohmann::json taskrConfig; + taskrConfig["Task Worker Inactivity Time (Ms)"] = 100; // Suspend workers if a certain time of inactivity elapses + taskrConfig["Task Suspend Interval Time (Ms)"] = 100; // Workers suspend for this time before checking back + taskrConfig["Minimum Active Task Workers"] = 1; // Have at least one worker active at all times + taskrConfig["Service Worker Count"] = 1; // Have one dedicated service workers at all times to listen for incoming messages + taskrConfig["Make Task Workers Run Services"] = true; // Workers will check for meta messages in between executions + auto taskr = std::make_shared(taskComputeManager.get(), workerComputeManager.get(), computeResources, taskrConfig); + + // Creating serving Engine object + serving::system::Engine serving(instanceManager, taskComputeManager, rpcEngine, instanceManager->getRootInstanceId()); + + // Check whether the instance is root + const auto isRoot = instanceManager->getCurrentInstance()->isRootInstance(); + const auto instanceId = instanceManager->getCurrentInstance()->getId(); + + auto serviceModule = std::make_unique(taskr); + + // Adding a simple service that prints "Hello world" one time. Here we call terminate from within the task + // to keep the application simple + int reps = 5; + int count = 0; + auto helloWorldFc = [&]() { + if (count > reps) { return; } + + printf("[Instance %lu] Hello World %d!\n", instanceId, count); + count++; + }; + auto helloWorldService = taskr::Service(helloWorldFc, 10); + serviceModule->addService("helloWorld", &helloWorldService); + + // Adding service module to serving + serving.addModule("service", std::move(serviceModule)); + + // Initializing serving + serving.initialize(); + + // Running serving + serving.run(); + + if (isRoot) + { + printf("[Instance %lu] issuing termination\n", instanceId); + + std::this_thread::sleep_for(std::chrono::seconds(1)); + // Finalizing serving + serving.terminate(); + } + + // Awaiting serving termination + serving.await(); + + // Finalize Instance Manager + instanceManager->finalize(); +} diff --git a/platform/include/modules/broadcastDeployment/module.hpp b/platform/include/modules/broadcastDeployment/module.hpp new file mode 100644 index 0000000..6088fae --- /dev/null +++ b/platform/include/modules/broadcastDeployment/module.hpp @@ -0,0 +1,132 @@ +#pragma once + +#include +#include +#include +#include + +#include +#include +#include + +#include +#include + +namespace serving::modules::broadcastDeployment +{ + +#define __SERVING_REQUEST_DEPLOYMENT_CONFIGURATION_RPC_NAME "[serving/modules/broadcastDeployment] Request Deployment Configuration RPC" + +class Module final : public modules::Module +{ + public: + + Module(std::shared_ptr instanceManager, + std::shared_ptr computeManager, + std::shared_ptr rpcEngine, + const HiCR::Instance::instanceId_t deployerInstanceId, + const HiCR::Instance::instanceId_t instanceId, + const configuration::Deployment &deployment) + : modules::Module(), + _instanceManager(instanceManager), + _computeManager(computeManager), + _rpcEngine(rpcEngine), + _deployerInstanceId(deployerInstanceId), + _instanceId(instanceId), + _deployment(deployment) + { + _rpcEngine->addRPCTarget(__SERVING_REQUEST_DEPLOYMENT_CONFIGURATION_RPC_NAME, _computeManager->createExecutionUnit([this](void *) { sendDeploymentConfiguration(); })); + } + + Module(std::shared_ptr instanceManager, + std::shared_ptr computeManager, + std::shared_ptr rpcEngine, + const HiCR::Instance::instanceId_t deployerInstanceId, + const HiCR::Instance::instanceId_t instanceId) + : modules::Module(), + _instanceManager(instanceManager), + _computeManager(computeManager), + _rpcEngine(rpcEngine), + _deployerInstanceId(deployerInstanceId), + _instanceId(instanceId) + { + _rpcEngine->addRPCTarget(__SERVING_REQUEST_DEPLOYMENT_CONFIGURATION_RPC_NAME, _computeManager->createExecutionUnit([this](void *) { sendDeploymentConfiguration(); })); + } + + ~Module() override = default; + + __INLINE__ const configuration::Deployment &getDeployment() const { return _deployment; } + + void initialize() override + { + auto _instances = _instanceManager->getInstances(); + // If I am not the deployer instance, simply request the deployment information from the deployer + if (_instanceId == _deployerInstanceId) + { + for (const auto &instance : _instances) + { + if (instance->getId() == _deployerInstanceId) { continue; } + _rpcEngine->listen(); + } + } + else { retrieveDeployment(); } + } + + // Init-only module (no periodic work) + void run() override {} + void terminate() override {} + void await() override {} + void finalize() override {} + + protected: + + void service() override {} + + private: + + std::shared_ptr _instanceManager; + + std::shared_ptr _computeManager; + + std::shared_ptr _rpcEngine; + + const HiCR::Instance::instanceId_t _deployerInstanceId; + + const HiCR::Instance::instanceId_t _instanceId; + + serving::configuration::Deployment _deployment; + + __INLINE__ void sendDeploymentConfiguration() + { + // Serializing + const auto serializedDeployment = _deployment.serialize().dump(); + + // Returning serialized topology + _rpcEngine->submitReturnValue((void *)serializedDeployment.c_str(), serializedDeployment.size() + 1); + } + + /** + * Retrieves the deployment configuration from the deployer instance. + */ + void retrieveDeployment() + { + // Send request RPC + _rpcEngine->requestRPC(_deployerInstanceId, __SERVING_REQUEST_DEPLOYMENT_CONFIGURATION_RPC_NAME); + + // Wait for serialized information + auto returnValue = _rpcEngine->getReturnValue(); + + // Receiving raw serialized topology information from the worker + std::string deploymentString = (char *)returnValue->getPointer(); + + // Parsing serialized raw topology into a json object + auto deploymentJs = nlohmann::json::parse(deploymentString); + + // Creating the deployment object from the json + _deployment = configuration::Deployment(deploymentJs); + + // Freeing return value + _rpcEngine->getMemoryManager()->freeLocalMemorySlot(returnValue); + } +}; +} // namespace serving::modules::broadcastDeployment \ No newline at end of file diff --git a/platform/include/modules/channelController/module.hpp b/platform/include/modules/channelController/module.hpp new file mode 100644 index 0000000..e18627f --- /dev/null +++ b/platform/include/modules/channelController/module.hpp @@ -0,0 +1,222 @@ +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include +#include +#include + +namespace serving::modules::channelController +{ + +#define __SERVING_DEFAULT_CHANNEL_CONTROLLER_EXCHANGE_TAG 0x0000A100 + +/** + * Creates channels using a reconciliation loop. + * TODO: Right now every instance should be running this module as a service for + * a backend like MPI. use channel creator frontend to mask this. + */ +class Module final : public serving::modules::Module +{ + public: + + using edgeName_t = std::string; + using instanceId_t = HiCR::Instance::instanceId_t; + using input_t = serving::system::channels::Input; + using output_t = serving::system::channels::Output; + using channelId_t = serving::system::channels::channelId_t; + using keyBuilderFc_t = serving::system::channels::keyBuilderFc_t; + + Module(const instanceId_t instance, + const std::vector &communicationManagersInOrder, + const HiCR::GlobalMemorySlot::tag_t exchangeTag = __SERVING_DEFAULT_CHANNEL_CONTROLLER_EXCHANGE_TAG, + const size_t intervalMs = 250) + : serving::modules::Module(intervalMs), + _instanceId(instance), + _communicationManagersInOrder(communicationManagersInOrder), + _exchangeTag(exchangeTag) + { + if (_communicationManagersInOrder.empty()) HICR_THROW_LOGIC("Channel controller requires at least one communication manager."); + } + + ~Module() override = default; + + [[nodiscard]] __INLINE__ std::weak_ptr addDesiredProducer(const instanceId_t targetId, + const channelId_t channelId, + const serving::configuration::Edge &edge, + const keyBuilderFc_t &keyBuilder) + { + std::lock_guard lock(_producerMutex); + const auto &edgeName = edge.getName(); + if (_desiredProducers.contains(edgeName)) HICR_THROW_LOGIC("Desired producer for edge %s to target instance %lu already exists.", edgeName.c_str(), targetId); + _desiredProducers[edgeName] = std::make_shared(edge, channelId, _instanceId, targetId, keyBuilder); + return _desiredProducers[edgeName]; + } + + [[nodiscard]] __INLINE__ std::weak_ptr addDesiredConsumer(const instanceId_t sourceId, + const channelId_t channelId, + const serving::configuration::Edge &edge, + const keyBuilderFc_t &keyBuilder) + { + std::lock_guard lock(_consumerMutex); + const auto &edgeName = edge.getName(); + if (_desiredConsumers.contains(edgeName)) HICR_THROW_LOGIC("Desired consumer for edge %s from source instance %lu already exists.", edgeName.c_str(), sourceId); + _desiredConsumers[edgeName] = std::make_shared(edge, channelId, sourceId, _instanceId, keyBuilder); + return _desiredConsumers[edgeName]; + } + + [[nodiscard]] __INLINE__ bool hasProducer(const edgeName_t edgeName) const + { + std::lock_guard lock(_producerMutex); + return _actualProducers.contains(edgeName); + } + [[nodiscard]] __INLINE__ bool hasConsumer(const edgeName_t edgeName) const + { + std::lock_guard lock(_consumerMutex); + return _actualConsumers.contains(edgeName); + } + + [[nodiscard]] __INLINE__ std::weak_ptr getProducer(const edgeName_t edgeName) const + { + std::lock_guard lock(_producerMutex); + if (_actualProducers.contains(edgeName) == false) HICR_THROW_RUNTIME("No actual producer found for edge %s.", edgeName.c_str()); + return _actualProducers.at(edgeName); + } + + [[nodiscard]] __INLINE__ std::weak_ptr getConsumer(const edgeName_t edgeName) const + { + std::lock_guard lock(_consumerMutex); + if (_actualConsumers.contains(edgeName) == false) HICR_THROW_RUNTIME("No actual consumer found for edge %s.", edgeName.c_str()); + return _actualConsumers.at(edgeName); + } + + __INLINE__ void removeDesiredProducer(const edgeName_t edgeName) + { + std::lock_guard lock(_producerMutex); + _desiredProducers.erase(edgeName); + } + + __INLINE__ void removeDesiredConsumer(const edgeName_t edgeName) + { + std::lock_guard lock(_consumerMutex); + _desiredConsumers.erase(edgeName); + } + + void initialize() override { reconcile(); } + void run() override {} + void terminate() override {} + void await() override {} + void finalize() override + { + std::scoped_lock lock(_producerMutex, _consumerMutex); + _desiredProducers.clear(); + _actualProducers.clear(); + _desiredConsumers.clear(); + _actualConsumers.clear(); + } + + protected: + + void service() override { reconcile(); } + + private: + + __INLINE__ void reconcile() + { + std::unordered_map> producersToCreate; + std::unordered_map> consumersToCreate; + std::vector producersToDelete; + std::vector consumersToDelete; + + // Diff desired vs actual + { + std::scoped_lock lock(_producerMutex, _consumerMutex); + for (const auto &[target, output] : _desiredProducers) + if (_actualProducers.contains(target) == false) producersToCreate[target] = output; + for (const auto &[source, input] : _desiredConsumers) + if (_actualConsumers.contains(source) == false) consumersToCreate[source] = input; + for (const auto &[target, _] : _actualProducers) + if (_desiredProducers.contains(target) == false) producersToDelete.push_back(target); + for (const auto &[source, _] : _actualConsumers) + if (_desiredConsumers.contains(source) == false) consumersToDelete.push_back(source); + } + + // Create missing channels + if (producersToCreate.empty() == false || consumersToCreate.empty() == false) + { + std::vector memorySlotsToExchange; + for (const auto &[_, output] : producersToCreate) output->getMemorySlotsToExchange(memorySlotsToExchange); + for (const auto &[_, input] : consumersToCreate) input->getMemorySlotsToExchange(memorySlotsToExchange); + + std::map> exchangeMap; + for (const auto manager : _communicationManagersInOrder) exchangeMap[manager] = {}; + + for (const auto &entry : memorySlotsToExchange) + { + if (exchangeMap.contains(entry.communicationManager) == false) HICR_THROW_LOGIC("Memory slot exchange manager is not present in configured manager order."); + exchangeMap[entry.communicationManager].push_back({entry.globalKey, entry.memorySlot}); + } + + for (const auto manager : _communicationManagersInOrder) manager->exchangeGlobalMemorySlots(_exchangeTag, exchangeMap[manager]); + for (const auto manager : _communicationManagersInOrder) manager->fence(_exchangeTag); + + for (const auto &[_, output] : producersToCreate) output->initialize(_exchangeTag); + for (const auto &[_, input] : consumersToCreate) input->initialize(_exchangeTag); + + { + std::scoped_lock lock(_producerMutex, _consumerMutex); + for (const auto &[target, output] : producersToCreate) { _actualProducers[target] = output; } + for (const auto &[source, input] : consumersToCreate) { _actualConsumers[source] = input; } + } + } + + // Delete stale channels + std::vector> staleProducers; + std::vector> staleConsumers; + { + std::scoped_lock lock(_producerMutex, _consumerMutex); + staleProducers.reserve(producersToDelete.size()); + for (const auto &target : producersToDelete) + { + auto it = _actualProducers.find(target); + if (it == _actualProducers.end()) continue; + staleProducers.push_back(std::move(it->second)); + _actualProducers.erase(it); + } + staleConsumers.reserve(consumersToDelete.size()); + for (const auto &source : consumersToDelete) + { + auto it = _actualConsumers.find(source); + if (it == _actualConsumers.end()) continue; + staleConsumers.push_back(std::move(it->second)); + _actualConsumers.erase(it); + } + } + } + + const instanceId_t _instanceId; + + mutable std::mutex _producerMutex; + mutable std::mutex _consumerMutex; + const std::vector _communicationManagersInOrder; + const HiCR::GlobalMemorySlot::tag_t _exchangeTag; + + // Desired state + std::unordered_map> _desiredProducers; // key: target instance + std::unordered_map> _desiredConsumers; // key: source instance + + // Actual created state + std::unordered_map> _actualProducers; // key: target instance + std::unordered_map> _actualConsumers; // key: source instance +}; +} // namespace serving::modules::channelController diff --git a/platform/include/modules/configuration/deployment.hpp b/platform/include/modules/configuration/deployment.hpp new file mode 100644 index 0000000..68aba09 --- /dev/null +++ b/platform/include/modules/configuration/deployment.hpp @@ -0,0 +1,353 @@ +#pragma once + +#include +#include +#include +#include + +#include + +#include +#include + +#include "partition.hpp" +#include "edge.hpp" +#include "requestManager.hpp" + +namespace serving::configuration +{ + +class Deployment final +{ + public: + + /** + * Heartbeat is a runtime check mechanism to verify neighbors and replicas are still alive. + * Every 'interval' milliseconds, the coordinator sends a heartbeat message to its own replicas and vice-versa + * If the 'tolerance' interval passes without a heartbeat, then the other side is deemed non-responsive + * Recovery or fail tolerance mechanisms can be activated upon this happening + */ + struct heartbeat_t + { + // Whether to enable the heartbeat service or not + bool enabled; + + // Whether to print the hearbeat arriving on screen + bool visible; + + // Heartbeat interval (every how many ms do we send a heartbeat message to replicas and neighboring coordinators) + size_t interval; + + // Tolerance is how long to wait before alerting that we haven't received a heartbeat + size_t tolerance; + }; + + /** + * The control buffer is the edge between coordinators and replicas to exchange control messages. + * Control messages are not related to the application, but instead to managing deployment aspects + * (e.g., load balancing, fail detection, migration) + */ + static constexpr size_t _defaultControlBufferCapacity = 256; + static constexpr size_t _defaultControlBufferSize = _defaultControlBufferCapacity * 1024; + struct controlBuffer_t + { + // Capacity (the maximum number of pending tokens in the buffer) + size_t capacity = _defaultControlBufferCapacity; + + // Size (the size in bytes that the buffer can hold) + size_t size = _defaultControlBufferSize; + + // HiCR-specific objects to create the control buffers. These are to be set at runtime + HiCR::CommunicationManager *communicationManager = nullptr; + HiCR::MemoryManager *memoryManager = nullptr; + std::shared_ptr memorySpace = nullptr; + }; + + struct settings_t + { + heartbeat_t heartbeat; + controlBuffer_t controlBuffer; + }; + + Deployment(const std::string &name) + : _name(name){}; + Deployment(const nlohmann::json &js) { deserialize(js); }; + Deployment() = default; + ~Deployment() = default; + + __INLINE__ void setName(const std::string &name) { _name = name; } + __INLINE__ void addPartition(const std::shared_ptr partition) { _partitions.push_back(partition); } + __INLINE__ void addChannel(const std::shared_ptr channel) { _edges.push_back(channel); } + + [[nodiscard]] __INLINE__ std::string getName() const { return _name; } + [[nodiscard]] __INLINE__ auto &getPartitions() const { return _partitions; } + [[nodiscard]] __INLINE__ auto &getEdges() const { return _edges; } + [[nodiscard]] __INLINE__ auto &getHeartbeat() const { return _settings.heartbeat; } + [[nodiscard]] __INLINE__ const auto &getControlBuffer() const { return _settings.controlBuffer; } + [[nodiscard]] __INLINE__ auto &getControlBuffer() { return _settings.controlBuffer; } + [[nodiscard]] __INLINE__ auto &getControlBufferConst() const { return _settings.controlBuffer; } + [[nodiscard]] __INLINE__ auto &getRequestManager() const { return _requestManager; } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + + js["Name"] = _name; + + std::vector partitionsJs; + for (const auto &p : _partitions) partitionsJs.push_back(p->serialize()); + js["Partitions"] = partitionsJs; + + std::vector edgesJs; + for (const auto &e : _edges) edgesJs.push_back(e->serialize()); + js["Edges"] = edgesJs; + + js["Request Manager"] = _requestManager->serialize(); + + ////////////////////// Parsing settings + auto settings = std::map(); + + // Heartbeat + auto heartbeat = std::map(); + heartbeat["Enabled"] = _settings.heartbeat.enabled; + heartbeat["Visible"] = _settings.heartbeat.visible; + heartbeat["Interval"] = _settings.heartbeat.interval; + heartbeat["Tolerance"] = _settings.heartbeat.tolerance; + settings["Heartbeat"] = heartbeat; + + // Control Buffer + auto controlBuffer = std::map(); + controlBuffer["Capacity"] = _settings.controlBuffer.capacity; + controlBuffer["Size"] = _settings.controlBuffer.size; + settings["Control Buffer"] = controlBuffer; + + js["Settings"] = settings; + + return js; + } + + __INLINE__ void deserialize(const nlohmann::json &js) + { + // Clearing objects + _partitions.clear(); + _edges.clear(); + + _name = hicr::json::getString(js, "Name"); + + const auto &partitions = hicr::json::getArray(js, "Partitions"); + for (const auto &p : partitions) _partitions.push_back(std::make_shared(p)); + + const auto &edges = hicr::json::getArray(js, "Edges"); + for (const auto &e : edges) _edges.push_back(std::make_shared(e)); + + const auto &requestManagerJs = hicr::json::getObject(js, "Request Manager"); + _requestManager = std::make_shared(requestManagerJs); + + // Getting settings + nlohmann::json settingsJs = hicr::json::getObject(js, "Settings"); + + nlohmann::json heartbeatJs = hicr::json::getObject(settingsJs, "Heartbeat"); + _settings.heartbeat.enabled = hicr::json::getBoolean(heartbeatJs, "Enabled"); + _settings.heartbeat.visible = hicr::json::getBoolean(heartbeatJs, "Visible"); + _settings.heartbeat.interval = hicr::json::getNumber(heartbeatJs, "Interval"); + _settings.heartbeat.tolerance = hicr::json::getNumber(heartbeatJs, "Tolerance"); + + nlohmann::json controlBufferJs = hicr::json::getObject(settingsJs, "Control Buffer"); + _settings.controlBuffer.capacity = hicr::json::getNumber(controlBufferJs, "Capacity"); + _settings.controlBuffer.size = hicr::json::getNumber(controlBufferJs, "Size"); + } + + // Includes all kinds of sanity checks relevant to a deployment + __INLINE__ void verify() const + { + // Getting rqeuest manager input and output edges + const auto &requestManagerInputEdge = _requestManager->getInput(); + const auto &requestManagerOutputEdge = _requestManager->getOutput(); + + // Getting all partition's edges in a set + std::set partitionNameSet; + for (const auto &partition : _partitions) partitionNameSet.insert(partition->getName()); + + // Getting a list of all edges to check the tasks specifying inputs/outputs actually refer to one of these, or user interface + std::set edgeNameSet; + + // Getting a list of all the inputs and output names to verify they correspond to each other at least once + std::set inputSet; + std::set outputSet; + + // Storage for those partitions with user interface input and output + std::shared_ptr userInterfaceInputPartition = nullptr; + std::shared_ptr userInterfaceOutputPartition = nullptr; + + // Now checking all dependencies belong in the same partition + std::set tasksWithDependencies; + for (const auto &partition : _partitions) + { + // Creating set of task names of this partition + std::set _taskNames; + for (const auto &task : partition->getTasks()) _taskNames.insert(task->getFunctionName()); + + for (const auto &task : partition->getTasks()) + for (const auto &dependency : task->getDependencies()) + { + // Check the task doesn't depend on itself + if (dependency == task->getFunctionName()) HICR_THROW_LOGIC("Task %s has a dependency on itself\n", task->getFunctionName().c_str()); + + // Check the dependency has a task that is referenced by this dependency + if (_taskNames.contains(dependency) == false) + HICR_THROW_LOGIC("Task %s has a dependency on '%s' which is not defined in this partition\n", task->getFunctionName().c_str(), dependency.c_str()); + + // Adding dependency to tasks with dependencies + tasksWithDependencies.insert(dependency); + } + } + + // First, the user interface input/outputs are counted as edges for the purposes of this check + for (const auto &edge : _edges) + { + const auto &edgeName = edge->getName(); + if (edgeNameSet.contains(edgeName)) HICR_THROW_LOGIC("Deployment specifies repeated edge or user input '%s'\n", edgeName.c_str()); + edgeNameSet.insert(edgeName); + } + + // For each input / output, remember which partition is its consumer / producer + std::map consumerPartitionMap; + std::map producerPartitionMap; + for (const auto &partition : _partitions) + for (const auto &task : partition->getTasks()) + { + // Make sure all tasks have at least one input + if (task->getInputs().size() == 0 && task->getDependencies().empty() == true) + HICR_THROW_LOGIC("Deployment specifies task in partition '%s' with function name '%s' without any inputs or dependencies\n", + partition->getName().c_str(), + task->getFunctionName().c_str()); + + // Check that the edge is not used as input more than once. All edges must be 1-to-1 + for (const auto &input : task->getInputs()) + { + if (inputSet.contains(input)) HICR_THROW_LOGIC("Deployment specifies input '%s' used more than once\n", input.c_str()); + if (edgeNameSet.contains(input) == false) + HICR_THROW_LOGIC("Deployment specifies task '%s' with an undefined input '%s'\n", task->getFunctionName().c_str(), input.c_str()); + consumerPartitionMap[input] = partition->getName(); + inputSet.insert(input); + + // Check the task that contains the user interface input does not receive any other inputs + if (input == requestManagerInputEdge) userInterfaceInputPartition = partition; + if (input == requestManagerOutputEdge) + HICR_THROW_LOGIC("Deployment specifies task '%s' with user interface output '%s' which is being used as input\n", task->getFunctionName().c_str(), input.c_str()); + if (input == requestManagerInputEdge && task->getInputs().size() > 1) + HICR_THROW_LOGIC( + "Deployment specifies task '%s' with user interface input '%s' which is not the only input of that task\n", task->getFunctionName().c_str(), input.c_str()); + } + + // Make sure all tasks have at least one output + if (task->getOutputs().size() == 0 && tasksWithDependencies.contains(task->getFunctionName()) == false) + HICR_THROW_LOGIC("Deployment specifies task in partition '%s' with function name '%s' without any outputs or dependents\n", + partition->getName().c_str(), + task->getFunctionName().c_str()); + + // Check that the output is not used as input more than once. All edges must be 1-to-1 + for (const auto &output : task->getOutputs()) + { + if (outputSet.contains(output)) HICR_THROW_LOGIC("Deployment specifies output '%s' used more than once\n", output.c_str()); + if (edgeNameSet.contains(output) == false) + HICR_THROW_LOGIC("Deployment specifies task '%s' with an undefined output '%s'\n", task->getFunctionName().c_str(), output.c_str()); + producerPartitionMap[output] = partition->getName(); + outputSet.insert(output); + + // Check the task that contains the user interface input does not receive any other inputs + if (output == requestManagerOutputEdge) userInterfaceOutputPartition = partition; + if (output == requestManagerInputEdge) + HICR_THROW_LOGIC("Deployment specifies task '%s' with user interface input '%s' which is being used as output\n", task->getFunctionName().c_str(), output.c_str()); + if (output == requestManagerOutputEdge && task->getOutputs().size() > 1) + HICR_THROW_LOGIC( + "Deployment specifies task '%s' with user interface output '%s' which is not the only output of task\n", task->getFunctionName().c_str(), output.c_str()); + } + } + + // Check whether all edges have consumer+producer partitions that do exist + for (const auto &edge : _edges) + { + const auto &edgeName = edge->getName(); + if (edgeName != requestManagerInputEdge) + if (edgeName != requestManagerOutputEdge) + if (consumerPartitionMap.contains(edgeName) == false || producerPartitionMap.contains(edgeName) == false) + HICR_THROW_LOGIC("Deployment specifies edge '%s' but it is either not used as input or output (or neither)\n", edge->getName().c_str()); + + // Getting producer and consumer partitions + const auto &producerPartition = producerPartitionMap[edge->getName()]; + const auto &consumerPartition = consumerPartitionMap[edge->getName()]; + + // Checking the edge connects two different partitions + if (producerPartition == consumerPartition) + HICR_THROW_LOGIC("Deployment specifies edge '%s' that is both produced and consumed by partition %s\n", edge->getName().c_str(), producerPartition.c_str()); + + // Setting producer and consumer partitions for the edge + edge->setProducer(producerPartition); + edge->setConsumer(consumerPartition); + } + + // Make sure all inputs are also used as outputs, as long as it is not the user interface input + for (const auto &input : inputSet) + if (input != requestManagerInputEdge) + if (outputSet.contains(input) == false) HICR_THROW_LOGIC("Deployment input '%s' is not associated to any output\n", input.c_str()); + for (const auto &output : outputSet) + if (output != requestManagerOutputEdge) + if (inputSet.contains(output) == false) HICR_THROW_LOGIC("Deployment output '%s' is not associated to any input\n", output.c_str()); + + // Check the user interface input/output are being used + if (userInterfaceInputPartition == nullptr) HICR_THROW_LOGIC("User interface input '%s' is not associated to any partition\n", requestManagerInputEdge.c_str()); + if (userInterfaceOutputPartition == nullptr) HICR_THROW_LOGIC("User interface output '%s' is not associated to any partition\n", requestManagerOutputEdge.c_str()); + + // Make sure the partition which contains the user interface input does not contain any cross-partition inputs + for (const auto &task : userInterfaceInputPartition->getTasks()) + for (const auto &input : task->getInputs()) + if (input != requestManagerInputEdge) + if (producerPartitionMap.at(input) != userInterfaceInputPartition->getName()) + HICR_THROW_LOGIC("Partition %s consumes the user interface input '%s' but also has other inter-partition inputs (e.g.,: '%s')\n", + userInterfaceInputPartition->getName().c_str(), + requestManagerInputEdge.c_str(), + input.c_str()); + + // Make sure the partition which contains the user interface output does not contain any cross-partition outputs + for (const auto &task : userInterfaceOutputPartition->getTasks()) + for (const auto &output : task->getOutputs()) + if (output != requestManagerOutputEdge) + if (consumerPartitionMap.at(output) != userInterfaceOutputPartition->getName()) + HICR_THROW_LOGIC("Partition %s produces the user interface output '%s' but also has other inter-partition inputs (e.g.,: '%s')\n", + userInterfaceOutputPartition->getName().c_str(), + requestManagerOutputEdge.c_str(), + output.c_str()); + + // Setting producer partition for user interface input to be the same as the consumer + for (const auto &edge : _edges) + { + const auto &edgeName = edge->getName(); + + if (edgeName == requestManagerInputEdge) + { + edge->setProducer(userInterfaceInputPartition->getName()); + edge->setConsumer(userInterfaceInputPartition->getName()); + edge->setPromptEdge(true); + } + + if (edgeName == requestManagerOutputEdge) + { + edge->setProducer(userInterfaceOutputPartition->getName()); + edge->setConsumer(userInterfaceOutputPartition->getName()); + edge->setResultEdge(true); + } + } + } + + private: + + std::string _name; + std::vector> _partitions; + std::vector> _edges; + std::shared_ptr _requestManager; + settings_t _settings; + +}; // class Deployment + +} // namespace serving::configuration \ No newline at end of file diff --git a/platform/include/modules/configuration/edge.hpp b/platform/include/modules/configuration/edge.hpp new file mode 100644 index 0000000..60da64d --- /dev/null +++ b/platform/include/modules/configuration/edge.hpp @@ -0,0 +1,108 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include + +namespace serving::configuration +{ + +class Edge +{ + public: + +#define __SERVING_PARTITION_DEFAULT_BUFFER_CAPACITY 1 + + typedef uint64_t edgeIndex_t; + + Edge(const nlohmann::json &js) { deserialize(js); } + Edge(const std::string &name, const size_t bufferCapacity, const size_t bufferSize = 0) + : _name(name), + _bufferCapacity(bufferCapacity), + _bufferSize(bufferSize) + {} + virtual ~Edge() = default; + + __INLINE__ void setName(const std::string &name) { _name = name; } + [[nodiscard]] __INLINE__ std::string getName() const { return _name; } + [[nodiscard]] __INLINE__ std::string getProducer() const { return _producer; } + [[nodiscard]] __INLINE__ std::string getConsumer() const { return _consumer; } + [[nodiscard]] __INLINE__ size_t getBufferCapacity() const { return _bufferCapacity; } + [[nodiscard]] __INLINE__ size_t getBufferSize() const { return _bufferSize; } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + + js["Name"] = _name; + js["Producer"] = _producer; + js["Consumer"] = _consumer; + js["Buffer Capacity"] = _bufferCapacity; + js["Buffer Size"] = _bufferSize; + + return js; + } + + __INLINE__ void deserialize(const nlohmann::json &js) + { + _name = hicr::json::getString(js, "Name"); + if (js.contains("Producer")) _producer = hicr::json::getString(js, "Producer"); + if (js.contains("Consumer")) _consumer = hicr::json::getString(js, "Consumer"); + if (js.contains("Buffer Capacity")) _bufferCapacity = hicr::json::getNumber(js, "Buffer Capacity"); + _bufferSize = hicr::json::getNumber(js, "Buffer Size"); + } + + // Functions to set the HiCR elements required for the creation of edge channels + __INLINE__ void setPayloadCommunicationManager(HiCR::CommunicationManager *const communicationManager) { _payloadCommunicationManager = communicationManager; } + __INLINE__ void setPayloadMemoryManager(HiCR::MemoryManager *const memoryManager) { _payloadMemoryManager = memoryManager; } + __INLINE__ void setPayloadMemorySpace(const std::shared_ptr memorySpace) { _payloadMemorySpace = memorySpace; } + __INLINE__ void setCoordinationCommunicationManager(HiCR::CommunicationManager *const communicationManager) { _coordinationCommunicationManager = communicationManager; } + __INLINE__ void setCoordinationMemoryManager(HiCR::MemoryManager *const memoryManager) { _coordinationMemoryManager = memoryManager; } + __INLINE__ void setCoordinationMemorySpace(const std::shared_ptr memorySpace) { _coordinationMemorySpace = memorySpace; } + + __INLINE__ void setProducer(const std::string &partition) { _producer = partition; } + __INLINE__ void setConsumer(const std::string &partition) { _consumer = partition; } + __INLINE__ void setPromptEdge(const bool isPromptEdge) { _isPromptEdge = isPromptEdge; } + __INLINE__ void setResultEdge(const bool isResultEdge) { _isResultEdge = isResultEdge; } + __INLINE__ void setBufferCapacity(const size_t bufferCapacity) { _bufferCapacity = bufferCapacity; } + __INLINE__ void setBufferSize(const size_t bufferSize) { _bufferSize = bufferSize; } + + // Functions to set the HiCR elements required for the creation of edge channels + [[nodiscard]] __INLINE__ HiCR::CommunicationManager *getPayloadCommunicationManager() const { return _payloadCommunicationManager; } + [[nodiscard]] __INLINE__ HiCR::MemoryManager *getPayloadMemoryManager() const { return _payloadMemoryManager; } + [[nodiscard]] __INLINE__ std::shared_ptr getPayloadMemorySpace() const { return _payloadMemorySpace; } + [[nodiscard]] __INLINE__ HiCR::CommunicationManager *getCoordinationCommunicationManager() const { return _coordinationCommunicationManager; } + [[nodiscard]] __INLINE__ HiCR::MemoryManager *getCoordinationMemoryManager() const { return _coordinationMemoryManager; } + [[nodiscard]] __INLINE__ std::shared_ptr getCoordinationMemorySpace() const { return _coordinationMemorySpace; } + [[nodiscard]] __INLINE__ bool isPromptEdge() const { return _isPromptEdge; } + [[nodiscard]] __INLINE__ bool isResultEdge() const { return _isResultEdge; } + + private: + + std::string _name; + std::string _producer; + std::string _consumer; + size_t _bufferCapacity = __SERVING_PARTITION_DEFAULT_BUFFER_CAPACITY; + size_t _bufferSize; + + // This flag serves to indicate whether these are edges that connect to the request manager + bool _isPromptEdge = false; + bool _isResultEdge = false; + + // HiCR-specific objects to create the payload buffers. These are to be set at runtime + HiCR::CommunicationManager *_payloadCommunicationManager = nullptr; + HiCR::MemoryManager *_payloadMemoryManager = nullptr; + std::shared_ptr _payloadMemorySpace = nullptr; + + // HiCR-specific objects to create the coordination buffers. These are to be set at runtime + HiCR::CommunicationManager *_coordinationCommunicationManager = nullptr; + HiCR::MemoryManager *_coordinationMemoryManager = nullptr; + std::shared_ptr _coordinationMemorySpace = nullptr; + +}; // class Base + +} // namespace serving::configuration \ No newline at end of file diff --git a/platform/include/modules/configuration/partition.hpp b/platform/include/modules/configuration/partition.hpp new file mode 100644 index 0000000..e47f79c --- /dev/null +++ b/platform/include/modules/configuration/partition.hpp @@ -0,0 +1,84 @@ +#pragma once + +#include +#include +#include "task.hpp" +#include "replica.hpp" +#include +#include +#include + +namespace serving::configuration +{ + +class Partition final +{ + public: + + typedef uint64_t partitionIndex_t; + + Partition(const nlohmann::json js) { deserialize(js); }; + Partition(const std::string &name, const HiCR::Instance::instanceId_t instanceId) + : _name(name), + _coordinatorInstanceId(instanceId) + {} + ~Partition() = default; + + __INLINE__ void setName(const std::string &name) { _name = name; } + __INLINE__ void setCoordinatorInstanceId(const HiCR::Instance::instanceId_t instanceId) { _coordinatorInstanceId = instanceId; } + __INLINE__ void addTask(const std::shared_ptr task) { _tasks.push_back(task); } + __INLINE__ void addReplica(const std::shared_ptr replica) { _replicas.push_back(replica); } + + [[nodiscard]] __INLINE__ auto getName() const { return _name; } + [[nodiscard]] __INLINE__ auto getCoordinatorInstanceId() const { return _coordinatorInstanceId; } + [[nodiscard]] __INLINE__ auto &getTasks() const { return _tasks; } + [[nodiscard]] __INLINE__ auto &getReplicas() const { return _replicas; } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + + js["Name"] = _name; + js["Coordinator Instance Id"] = _coordinatorInstanceId; + + std::vector tasksJs; + for (const auto &t : _tasks) tasksJs.push_back(t->serialize()); + js["Tasks"] = tasksJs; + + std::vector replicasJs; + for (const auto &r : _replicas) replicasJs.push_back(r->serialize()); + js["Replicas"] = replicasJs; + + return js; + } + + __INLINE__ void deserialize(const nlohmann::json &js) + { + // Clearing objects + _tasks.clear(); + _replicas.clear(); + + _name = hicr::json::getString(js, "Name"); + if (js.contains("Coordinator Instance Id")) + _coordinatorInstanceId = hicr::json::getNumber(js, "Coordinator Instance Id"); // Optional, as it is determined at runtime + + const auto &tasks = hicr::json::getArray(js, "Tasks"); + for (const auto &t : tasks) _tasks.push_back(std::make_shared(t)); + + // This entry is optional, as it can be decided at runtime + if (js.contains("Replicas")) + { + const auto &replicas = hicr::json::getArray(js, "Replicas"); + for (const auto &r : replicas) _replicas.push_back(std::make_shared(r)); + } + } + + private: + + std::string _name; + HiCR::Instance::instanceId_t _coordinatorInstanceId = 0; + std::vector> _tasks; + std::vector> _replicas; +}; // class Partition + +} // namespace serving::configuration \ No newline at end of file diff --git a/platform/include/modules/configuration/replica.hpp b/platform/include/modules/configuration/replica.hpp new file mode 100644 index 0000000..9d933c5 --- /dev/null +++ b/platform/include/modules/configuration/replica.hpp @@ -0,0 +1,47 @@ +#pragma once + +#include +#include +#include +#include + +namespace serving::configuration +{ + +class Replica final +{ + public: + + typedef uint64_t replicaIndex_t; + + Replica(const nlohmann::json js) { deserialize(js); }; + Replica(const HiCR::Instance::instanceId_t instanceId) + : _instanceId(instanceId) + {} + ~Replica() = default; + + __INLINE__ void setInstanceId(const HiCR::Instance::instanceId_t instanceId) { _instanceId = instanceId; } + + [[nodiscard]] __INLINE__ auto getInstanceId() const { return _instanceId; } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + + js["Instance Id"] = _instanceId; + + return js; + } + + __INLINE__ void deserialize(const nlohmann::json &js) + { + if (js.contains("Instance Id")) _instanceId = hicr::json::getNumber(js, "Instance Id"); // Optional, as it is determined at runtime + } + + private: + + HiCR::Instance::instanceId_t _instanceId; + +}; // class Replica + +} // namespace serving::configuration \ No newline at end of file diff --git a/platform/include/modules/configuration/requestManager.hpp b/platform/include/modules/configuration/requestManager.hpp new file mode 100644 index 0000000..30177d7 --- /dev/null +++ b/platform/include/modules/configuration/requestManager.hpp @@ -0,0 +1,56 @@ +#pragma once + +#include +#include +#include +#include + +namespace serving::configuration +{ + +class RequestManager final +{ + public: + + RequestManager(const nlohmann::json js) { deserialize(js); }; + RequestManager(const std::string &input, const std::string &output) + : _input(input), + _output(output) + {} + ~RequestManager() = default; + + __INLINE__ void setInput(const std::string &input) { _input = input; } + __INLINE__ void setOutput(const std::string &output) { _output = output; } + __INLINE__ void setInstanceId(const HiCR::Instance::instanceId_t instanceId) { _instanceId = instanceId; } + + [[nodiscard]] __INLINE__ auto getInput() const { return _input; } + [[nodiscard]] __INLINE__ auto getOutput() const { return _output; } + [[nodiscard]] __INLINE__ auto getInstanceId() const { return _instanceId; } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + + js["Input"] = _input; + js["Output"] = _output; + js["Instance Id"] = _instanceId; + + return js; + } + + __INLINE__ void deserialize(const nlohmann::json &js) + { + _input = hicr::json::getString(js, "Input"); + _output = hicr::json::getString(js, "Output"); + if (js.contains("Instance Id")) _instanceId = hicr::json::getNumber(js, "Instance Id"); // Optional, as it is determined at runtime + } + + private: + + std::string _input; + std::string _output; + HiCR::Instance::instanceId_t _instanceId; + +}; // class RequestManager + +} // namespace serving::configuration \ No newline at end of file diff --git a/platform/include/modules/configuration/task.hpp b/platform/include/modules/configuration/task.hpp new file mode 100644 index 0000000..c04582b --- /dev/null +++ b/platform/include/modules/configuration/task.hpp @@ -0,0 +1,60 @@ +#pragma once + +#include +#include +#include +#include + +namespace serving::configuration +{ + +class Task final +{ + public: + + Task(const nlohmann::json &js) { deserialize(js); }; + Task(const std::string &functionName) + : _functionName(functionName) + {} + ~Task() = default; + + __INLINE__ void setFunctionName(const std::string &functionName) { _functionName = functionName; } + __INLINE__ void addInput(const std::string &input) { _inputs.push_back(input); } + __INLINE__ void addOutput(const std::string &output) { _outputs.push_back(output); } + __INLINE__ void addDependency(const std::string &functionName) { _dependencies.push_back(functionName); } + + [[nodiscard]] __INLINE__ auto getFunctionName() const { return _functionName; } + [[nodiscard]] __INLINE__ auto &getInputs() const { return _inputs; } + [[nodiscard]] __INLINE__ auto &getOutputs() const { return _outputs; } + [[nodiscard]] __INLINE__ auto &getDependencies() const { return _dependencies; } + + [[nodiscard]] __INLINE__ nlohmann::json serialize() const + { + nlohmann::json js; + + js["Function Name"] = _functionName; + js["Inputs"] = _inputs; + js["Outputs"] = _outputs; + js["Dependencies"] = _dependencies; + + return js; + } + + __INLINE__ void deserialize(const nlohmann::json &js) + { + _functionName = hicr::json::getString(js, "Function Name"); + _inputs = hicr::json::getArray(js, "Inputs"); + _outputs = hicr::json::getArray(js, "Outputs"); + _dependencies = hicr::json::getArray(js, "Dependencies"); + } + + private: + + std::string _functionName; + std::vector _inputs; + std::vector _outputs; + std::vector _dependencies; + +}; // class Task + +} // namespace serving::configuration \ No newline at end of file diff --git a/platform/include/modules/service/module.hpp b/platform/include/modules/service/module.hpp new file mode 100644 index 0000000..a890482 --- /dev/null +++ b/platform/include/modules/service/module.hpp @@ -0,0 +1,65 @@ +#pragma once +#include +#include +#include +#include + +#include +#include + +#include + +namespace serving::modules::service +{ + +class Module final : public serving::modules::Module +{ + public: + + using serviceFunction_t = taskr::Service::serviceFc_t; + + Module(std::shared_ptr taskr) + : serving::modules::Module(), + _taskr(taskr) + {} + + ~Module() override = default; + + // Ingest external service owned by another module (non-owning pointer). + __INLINE__ void addService(const std::string &name, taskr::Service *service) + { + if (_services.contains(name)) HICR_THROW_LOGIC("[Service] Service '%s' is already registered.", name.c_str()); + _services[name] = service; + } + + void initialize() override + { + // Required for services-only mode (no tasks): do not auto-finish immediately. + _taskr->setFinishOnLastTask(false); + for (const auto &[_, service] : _services) _taskr->addService(service); + _taskr->initialize(); + } + + void run() override { _taskr->run(); } + + void terminate() override { _taskr->setFinishOnLastTask(true); } + + void await() override { _taskr->await(); } + + void finalize() override + { + _taskr->finalize(); + _services.clear(); + } + + protected: + + void service() override {} + + private: + + std::shared_ptr _taskr; + + std::unordered_map _services; +}; +} // namespace serving::modules::service