Skip to main content

Local Scheduler Guide

Overview​

The CLIO Runtime uses a pluggable scheduler architecture to control how tasks are mapped to workers and how workers are organized. Scheduling happens at two levels: container-level scheduling resolves what to execute (via Container::ScheduleTask), and the Scheduler decides where to execute it (which worker thread). This document explains both levels and how to build custom schedulers.

Table of Contents​

  1. Architecture Overview
  2. Two-Level Scheduling
  3. Scheduler Interface
  4. Worker Lifecycle
  5. Implementing a Custom Scheduler
  6. DefaultScheduler Example
  7. Best Practices
  8. Integration Points

Architecture Overview​

Component Responsibilities​

The CLIO Runtime separates concerns across three main components:

  • ConfigManager: Manages configuration (number of threads, queue depth, etc.)
  • WorkOrchestrator: Creates workers, spawns threads, assigns lanes to workers (1:1 mapping for all workers)
  • Scheduler: Decides worker partitioning, task-to-worker mapping, and load balancing
  • IpcManager: Manages shared memory, queues, and provides task routing infrastructure (RouteTask, RouteLocal, RouteGlobal)
  • Container: Provides per-pool dynamic scheduling via ScheduleTask

Data Flow​

┌─────────────────┐
│  ConfigManager  │──→ num_threads, queue_depth
└─────────────────┘
         │
         ↓
┌─────────────────┐
│ WorkOrchestrator│──→ Creates num_threads + 1 workers
└─────────────────┘
         │
         ↓
┌─────────────────┐
│   Scheduler     │──→ Tracks worker groups for routing decisions
└─────────────────┘    Updates IpcManager with scheduler queue count
         │
         ↓
┌─────────────────┐
│ WorkOrchestrator│──→ Maps ALL workers to lanes (1:1 mapping)
└─────────────────┘    Spawns OS threads for each worker
         │
         ↓
┌─────────────────┐
│   IpcManager    │──→ num_sched_queues used for client task mapping
└─────────────────┘

Task Routing Flow​

When a task is submitted, IpcManager::RouteTask orchestrates the full routing pipeline:

RouteTask(future, force_enqueue)
  │
  ├─ 1. Container::ScheduleTask()    → resolve Dynamic pool query
  │                                     (e.g., Dynamic → DirectHash)
  │
  ├─ 2. ResolvePoolQuery()           → resolve to physical node(s)
  │
  ├─ 3. IsTaskLocal()                → local or remote?
  │     │
  │     ├─ local:  RouteLocal()      → resolve container, pick worker
  │     │           │
  │     │           ├─ RuntimeMapTask()  → scheduler picks dest worker
  │     │           │
  │     │           ├─ dest == current && !force_enqueue → execute directly
  │     │           └─ otherwise → enqueue to dest worker's lane
  │     │
  │     └─ remote: RouteGlobal()     → enqueue to net_queue_ for SendIn
  │
  └─ return true if task can be executed by current worker

Two-Level Scheduling​

Level 1: Container Scheduling (ScheduleTask)​

Before the Scheduler decides which worker runs a task, the container decides which container instance handles it. This is the Container::ScheduleTask virtual method.

#include "clio_runtime/container.h"

using namespace clio::run;

// Containers override the virtual ScheduleTask hook to transform a task's
// PoolQuery before routing. The base-class default simply returns the task's
// existing pool query unchanged. (Leaving the other pure-virtual Container
// methods unimplemented keeps this class abstract — it is only for
// illustration and is never instantiated.)
class PassthroughContainer : public Container {
 public:
  PoolQuery ScheduleTask(const clio::run::shared_ptr<Task> &task) override {
    return task->pool_query_;  // Default: no transformation
  }
};

Purpose: Transform a high-level PoolQuery (often Dynamic) into a concrete routing mode (DirectHash, Local, Broadcast, etc.) based on the task's semantics.

Example: A distributed filesystem container might schedule read tasks to the node holding the target block:

#include "clio_runtime/container.h"

using namespace clio::run;

class MyFsContainer : public Container {
 public:
  static constexpr u32 kRead = 10;
  static constexpr u32 kWrite = 11;

  PoolQuery ScheduleTask(const clio::run::shared_ptr<Task> &task) override {
    if (task->method_ == kRead || task->method_ == kWrite) {
      // Route to the container holding the target block
      u32 block_id = GetBlockId(task);
      return PoolQuery::DirectHash(block_id);
    }
    // Metadata ops stay local
    return PoolQuery::Local();
  }

 private:
  u32 GetBlockId(const clio::run::shared_ptr<Task> &task) { return 0; }
};

ScheduleTask is called by RouteTask before pool query resolution. The returned PoolQuery then goes through ResolvePoolQuery to determine concrete physical nodes and containers.

Level 2: Worker Scheduling (RuntimeMapTask)​

After container and node resolution, RouteLocal calls Scheduler::RuntimeMapTask to pick the specific worker thread. This is where I/O-size routing, task group affinity, and network worker pinning happen.

Scheduler Interface​

All schedulers must inherit from the Scheduler base class and implement the following methods:

Required Methods​

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/ipc_manager.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

// Every scheduler derives from clio::run::Scheduler and implements these
// methods. The signatures mirror clio_runtime/scheduler/scheduler.h exactly —
// the `override` keyword makes the compiler verify them. (Leaving the pure
// virtuals only declared keeps this class abstract; it is a skeleton.)
class ExampleScheduler : public Scheduler {
 public:
  // Partition workers into groups after WorkOrchestrator creates them
  void DivideWorkers(WorkOrchestrator *work_orch) override;

  // Map tasks from clients to worker lanes
  u32 ClientMapTask(IpcManager *ipc_manager, const Future<Task> &task) override;

  // Map tasks from runtime workers to other workers. `container` is the
  // resolved execution container (may be null) used for task-group affinity.
  u32 RuntimeMapTask(Worker *worker, const Future<Task> &task,
                     ContainerHold container) override;

  // Rebalance load across workers (called periodically by workers)
  void RebalanceWorker(Worker *worker) override;

  // Adjust the polling interval for a periodic task after it runs
  void AdjustPolling(const clio::run::shared_ptr<Task> &task) override;

  // Get designated GPU / network workers (optional; default nullptr)
  Worker *GetGpuWorker() const override { return nullptr; }
  Worker *GetNetWorker() const override { return nullptr; }
};

Method Details​

DivideWorkers(WorkOrchestrator *work_orch)​

Purpose: Partition workers into functional groups after they've been created.

Called: Once during initialization, after WorkOrchestrator creates all workers but before threads are spawned.

Responsibilities:

  • Access workers via work_orch->GetWorker(worker_id)
  • Organize workers into scheduler-specific groups (e.g., scheduler worker, I/O workers, network worker, GPU worker)
  • Update IpcManager with the scheduling queue count via IpcManager::SetNumSchedQueues()
  • Set the network lane via IpcManager::SetNetLane() so the runtime knows where to enqueue network tasks

Important: All workers are assigned lanes by WorkOrchestrator::SpawnWorkerThreads() using 1:1 mapping. The scheduler does NOT control lane assignment — it only tracks worker groups for routing decisions.

Example:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/ipc_manager.h"
#include "clio_runtime/worker.h"
#include <vector>

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void DivideWorkers(WorkOrchestrator *work_orch) override {
    u32 total_workers = work_orch->GetTotalWorkerCount();

    // Worker 0: scheduler worker (metadata + small I/O)
    scheduler_worker_ = work_orch->GetWorker(0);

    // Workers 1..N-2: I/O workers (large I/O round-robin)
    for (u32 i = 1; i < total_workers - 1; ++i) {
      io_workers_.push_back(work_orch->GetWorker(i));
    }

    // Worker N-1: network worker
    net_worker_ = work_orch->GetWorker(total_workers - 1);

    // IMPORTANT: Update IpcManager. SetNetLane takes the send and recv lanes;
    // a single-net-worker scheduler passes the same lane for both.
    IpcManager *ipc = CLIO_IPC;
    if (ipc) {
      ipc->SetNumSchedQueues(1);  // Client tasks go to scheduler worker
      if (net_worker_) {
        ipc->SetNetLane(net_worker_->GetLane(), net_worker_->GetLane());
      }
    }
  }

  // Other Scheduler overrides omitted for brevity.

 private:
  Worker *scheduler_worker_ = nullptr;
  std::vector<Worker *> io_workers_;
  Worker *net_worker_ = nullptr;
};

ClientMapTask(IpcManager *ipc_manager, const Future<Task> &task)​

Purpose: Determine which worker lane a task from a client should be assigned to.

Called: When clients submit tasks to the runtime via SendRuntimeClient.

Responsibilities:

  • Return a lane ID in range [0, num_sched_queues)
  • Use ipc_manager->GetNumSchedQueues() to get valid lane count
  • Route special tasks (e.g., network Send/Recv) to the appropriate worker
  • Common strategies: PID+TID hash, round-robin, locality-aware

Example:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/ipc_manager.h"

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  u32 ClientMapTask(IpcManager *ipc_manager,
                    const Future<Task> &task) override {
    u32 num_lanes = ipc_manager->GetNumSchedQueues();
    if (num_lanes == 0) return 0;

    // Network tasks (Send/Recv from admin pool) → last (network) lane
    Task *task_ptr = task.get();
    if (task_ptr != nullptr && task_ptr->pool_id_ == kAdminPoolId) {
      u32 method_id = task_ptr->method_;
      if (method_id == 14 || method_id == 15 ||
          method_id == 20 || method_id == 21) {
        return num_lanes - 1;
      }
    }

    // Default: scheduler worker (lane 0)
    return 0;
  }

  // Other Scheduler overrides omitted for brevity.
};

RuntimeMapTask(Worker *worker, const Future<Task> &task, Container *container)​

Purpose: Determine which worker should execute a task when routing from within the runtime.

Called: By IpcManager::RouteLocal() after the execution container has been resolved. If the returned worker ID differs from the current worker (or if force_enqueue is set), the task is enqueued to the destination worker's lane.

Parameters:

  • worker: The current worker (may be nullptr if called from a non-worker thread)
  • task: The task being routed
  • container: The resolved execution container. Used for task-group affinity lookups. May be nullptr when called without a resolved container.

Responsibilities:

  • Return a worker ID for task execution
  • Route periodic network tasks (Send/Recv) to the dedicated network worker
  • Implement I/O-size-based routing (large I/O → dedicated workers)
  • Implement task group affinity (pin related tasks to the same worker)

Example:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/container.h"
#include "clio_runtime/worker.h"
#include <atomic>
#include <vector>

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  u32 RuntimeMapTask(Worker *worker, const Future<Task> &task,
                     ContainerHold container) override {
    Task *task_ptr = task.get();

    // Task group affinity: if this task's group is already pinned, use that
    // worker. `container` is a ContainerHold (reached via operator->).
    if (container != nullptr && task_ptr != nullptr &&
        !task_ptr->task_group_.IsNull()) {
      int64_t group_id = task_ptr->task_group_.id_;
      ScopedCoRwReadLock read_lock(container->task_group_lock_);
      auto it = container->task_group_map_.find(group_id);
      if (it != container->task_group_map_.end() && it->second != nullptr) {
        return it->second->GetId();
      }
    }

    // Periodic network tasks → network worker
    if (task_ptr != nullptr && task_ptr->IsPeriodic()) {
      if (task_ptr->pool_id_ == kAdminPoolId) {
        u32 method_id = task_ptr->method_;
        if (method_id == 14 || method_id == 15) {
          if (net_worker_ != nullptr) {
            return net_worker_->GetId();
          }
        }
      }
    }

    // Large I/O → round-robin across I/O workers. PredictedStat() is populated
    // by IpcManager before this scheduler hook runs.
    if (task_ptr != nullptr && !io_workers_.empty()) {
      if (task_ptr->PredictedStat().io_size_ >= kLargeIOThreshold) {
        u32 idx = next_io_idx_.fetch_add(1, std::memory_order_relaxed)
                  % static_cast<u32>(io_workers_.size());
        return io_workers_[idx]->GetId();
      }
    }

    // Small I/O / metadata → scheduler worker
    if (scheduler_worker_ != nullptr) {
      return scheduler_worker_->GetId();
    }

    return worker ? worker->GetId() : 0;
  }

  // Other Scheduler overrides omitted for brevity.

 private:
  static constexpr size_t kLargeIOThreshold = 4096;
  Worker *scheduler_worker_ = nullptr;
  std::vector<Worker *> io_workers_;
  Worker *net_worker_ = nullptr;
  std::atomic<u32> next_io_idx_{0};
};

RebalanceWorker(Worker *worker)​

Purpose: Balance load across workers by stealing or delegating tasks.

Called: Periodically by workers after processing tasks.

Responsibilities:

  • Implement work stealing algorithms
  • Migrate tasks between workers
  • Optional — can be a no-op for simple schedulers

Example:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void RebalanceWorker(Worker *worker) override {
    // Simple schedulers can leave this empty
    (void)worker;
  }

  // Other Scheduler overrides omitted for brevity.
};

AdjustPolling(RunContext *run_ctx)​

Purpose: Adjust polling intervals for periodic tasks based on work done.

Called: After each execution of a periodic task.

Responsibilities:

  • Modify run_ctx->yield_time_us_ based on run_ctx->did_work_
  • Implement adaptive polling (exponential backoff when idle)
  • Reduce CPU usage for idle periodic tasks
  • Important: co_await on Futures sets yield_time_us_ to 0, so this method must restore it to prevent busy-looping

Example:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/task.h"

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void AdjustPolling(const clio::run::shared_ptr<Task> &task) override {
    if (task.IsNull()) return;

    // co_await on a Future sets the task's yield_time to 0, so restore its
    // true period (period_ns_, in microseconds) to keep periodic tasks from
    // busy-looping. Adaptive back-off would replace this single line.
    task->SetYieldTimeUs(task->TruePeriodNs() / 1000.0);
  }

  // Other Scheduler overrides omitted for brevity.
};

Worker Lifecycle​

Understanding the worker lifecycle is crucial for scheduler implementation:

1. ConfigManager loads configuration (num_threads, queue_depth)
   ↓
2. WorkOrchestrator::Init()
   - Creates num_threads + 1 workers
   - Calls Scheduler::DivideWorkers()
   ↓
3. Scheduler::DivideWorkers()
   - Tracks workers into functional groups (scheduler, I/O, network, GPU)
   - Updates IpcManager::SetNumSchedQueues()
   - Sets network lane via IpcManager::SetNetLane()
   ↓
4. WorkOrchestrator::StartWorkers()
   - Calls SpawnWorkerThreads()
   - Maps ALL workers to lanes (1:1 mapping: worker i → lane i)
   - Spawns actual OS threads
   ↓
5. Workers run task processing loops
   - ProcessNewTask: pop futures from lane, route via RouteTask()
     - RouteTask calls Container::ScheduleTask() for dynamic resolution
     - RouteLocal calls Scheduler::RuntimeMapTask() for worker selection
     - If dest != current worker (or force_enqueue), re-enqueue to dest lane
   - Call Scheduler::RebalanceWorker() periodically
   - Call Scheduler::AdjustPolling() after periodic task execution

Implementing a Custom Scheduler​

Step 1: Create Header File​

Create context-runtime/include/clio_runtime/scheduler/my_scheduler.h. The real schedulers declare their subclass inside namespace clio::run (see default_sched.h); the using namespace clio::run; here just keeps the doctest self-contained.

#ifndef CLIO_RUNTIME_INCLUDE_SCHEDULER_MY_SCHEDULER_H_
#define CLIO_RUNTIME_INCLUDE_SCHEDULER_MY_SCHEDULER_H_

#include <atomic>
#include <vector>
#include "clio_runtime/scheduler/scheduler.h"

using namespace clio::run;

class MyScheduler : public Scheduler {
 public:
  MyScheduler() : scheduler_worker_(nullptr), net_worker_(nullptr),
                  gpu_worker_(nullptr), next_io_idx_{0} {}
  ~MyScheduler() override = default;

  void DivideWorkers(WorkOrchestrator *work_orch) override;
  u32 ClientMapTask(IpcManager *ipc_manager, const Future<Task> &task) override;
  u32 RuntimeMapTask(Worker *worker, const Future<Task> &task,
                     ContainerHold container) override;
  void RebalanceWorker(Worker *worker) override;
  void AdjustPolling(const clio::run::shared_ptr<Task> &task) override;
  Worker *GetGpuWorker() const override { return gpu_worker_; }
  Worker *GetNetWorker() const override { return net_worker_; }

 private:
  Worker *scheduler_worker_;
  std::vector<Worker *> io_workers_;
  Worker *net_worker_;
  Worker *gpu_worker_;
  std::atomic<u32> next_io_idx_{0};
};

#endif  // CLIO_RUNTIME_INCLUDE_SCHEDULER_MY_SCHEDULER_H_

Step 2: Implement Methods​

Create context-runtime/src/scheduler/my_scheduler.cc. (The class is re-declared inline here so the doctest is self-contained; in real code it would come from the header above via #include.)

#include <atomic>
#include <vector>

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/config_manager.h"
#include "clio_runtime/container.h"
#include "clio_runtime/ipc_manager.h"
#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

class MyScheduler : public Scheduler {
 public:
  MyScheduler() : scheduler_worker_(nullptr), net_worker_(nullptr),
                  gpu_worker_(nullptr), next_io_idx_{0} {}

  void DivideWorkers(WorkOrchestrator *work_orch) override;
  u32 ClientMapTask(IpcManager *ipc_manager, const Future<Task> &task) override;
  u32 RuntimeMapTask(Worker *worker, const Future<Task> &task,
                     ContainerHold container) override;
  void RebalanceWorker(Worker *worker) override;
  void AdjustPolling(const clio::run::shared_ptr<Task> &task) override;
  Worker *GetGpuWorker() const override { return gpu_worker_; }
  Worker *GetNetWorker() const override { return net_worker_; }

 private:
  Worker *scheduler_worker_;
  std::vector<Worker *> io_workers_;
  Worker *net_worker_;
  Worker *gpu_worker_;
  std::atomic<u32> next_io_idx_{0};
};

void MyScheduler::DivideWorkers(WorkOrchestrator *work_orch) {
  if (!work_orch) return;

  u32 total_workers = work_orch->GetTotalWorkerCount();

  scheduler_worker_ = work_orch->GetWorker(0);
  net_worker_ = work_orch->GetWorker(total_workers - 1);

  if (total_workers > 2) {
    gpu_worker_ = work_orch->GetWorker(total_workers - 2);
    for (u32 i = 1; i < total_workers - 1; ++i) {
      io_workers_.push_back(work_orch->GetWorker(i));
    }
  }

  IpcManager *ipc = CLIO_IPC;
  if (ipc) {
    ipc->SetNumSchedQueues(1);
    if (net_worker_) {
      // Single net worker: same lane for both send and recv.
      ipc->SetNetLane(net_worker_->GetLane(), net_worker_->GetLane());
    }
  }
}

u32 MyScheduler::ClientMapTask(IpcManager *ipc_manager,
                               const Future<Task> &task) {
  u32 num_lanes = ipc_manager->GetNumSchedQueues();
  if (num_lanes == 0) return 0;
  (void)task;
  return 0;  // All client tasks → scheduler worker
}

u32 MyScheduler::RuntimeMapTask(Worker *worker, const Future<Task> &task,
                                ContainerHold container) {
  Task *task_ptr = task.get();

  // Task group affinity
  if (container != nullptr && task_ptr != nullptr &&
      !task_ptr->task_group_.IsNull()) {
    int64_t group_id = task_ptr->task_group_.id_;
    ScopedCoRwReadLock read_lock(container->task_group_lock_);
    auto it = container->task_group_map_.find(group_id);
    if (it != container->task_group_map_.end() && it->second != nullptr) {
      return it->second->GetId();
    }
  }

  // Periodic network tasks → network worker
  if (task_ptr != nullptr && task_ptr->IsPeriodic() &&
      task_ptr->pool_id_ == kAdminPoolId) {
    u32 m = task_ptr->method_;
    if ((m == 14 || m == 15 || m == 20 || m == 21) && net_worker_) {
      return net_worker_->GetId();
    }
  }

  // Large I/O → round-robin across I/O workers
  if (task_ptr != nullptr && !io_workers_.empty() &&
      task_ptr->PredictedStat().io_size_ >= 4096) {
    u32 idx = next_io_idx_.fetch_add(1, std::memory_order_relaxed)
              % static_cast<u32>(io_workers_.size());
    return io_workers_[idx]->GetId();
  }

  // Default → scheduler worker
  if (scheduler_worker_) return scheduler_worker_->GetId();
  return worker ? worker->GetId() : 0;
}

void MyScheduler::RebalanceWorker(Worker *worker) { (void)worker; }

void MyScheduler::AdjustPolling(const clio::run::shared_ptr<Task> &task) {
  if (task.IsNull()) return;
  task->SetYieldTimeUs(task->TruePeriodNs() / 1000.0);
}

Step 3: Register Scheduler​

Schedulers are constructed by name in SchedulerFactory::Get (context-runtime/src/scheduler/scheduler_factory.cc). Add a branch for your scheduler there:

#include <memory>
#include <string>

#include "clio_runtime/scheduler/scheduler_factory.h"
#include "clio_runtime/scheduler/default_sched.h"
#include "clio_runtime/scheduler/local_sched.h"

using namespace clio::run;

// Body of SchedulerFactory::Get: dispatch on the scheduler name and construct
// the matching scheduler. Add a branch for your own scheduler.
std::unique_ptr<Scheduler> MakeScheduler(const std::string &sched_name) {
  if (sched_name == "default") {
    return std::make_unique<DefaultScheduler>();
  }
  if (sched_name == "local") {
    return std::make_unique<LocalScheduler>();
  }
  // Add your scheduler here:
  //   if (sched_name == "my_scheduler") {
  //     return std::make_unique<MyScheduler>();
  //   }

  // Unknown name → fall back to the default scheduler.
  HLOG(kWarning, "Unknown scheduler name '{}', using default scheduler",
       sched_name);
  return std::make_unique<DefaultScheduler>();
}

Step 4: Configure​

Update your configuration file to use the new scheduler:

runtime:
  local_sched: "my_scheduler"
  num_threads: 4
  queue_depth: 1024

DefaultScheduler Example​

The DefaultScheduler provides a reference implementation with I/O-size-based routing and task group affinity.

Worker Partitioning​

  • Worker 0: Scheduler worker — handles metadata and small I/O (< 4KB)
  • Workers 1..N-2: I/O workers — handle large I/O via round-robin
  • Worker N-2: Also serves as the GPU worker (polls GPU queues)
  • Worker N-1: Network worker — handles all Send/Recv/ClientSend/ClientRecv tasks
  • SetNumSchedQueues(1) — client tasks all go to the scheduler worker initially

Dynamic Scheduling via ScheduleTask​

The DefaultScheduler works hand-in-hand with Container::ScheduleTask. When a task arrives with a Dynamic pool query, the container's ScheduleTask resolves it before RuntimeMapTask picks the worker:

  1. Task submitted with PoolQuery::Dynamic()
  2. RouteTask calls container->ScheduleTask(task) → returns e.g. DirectHash(block_id)
  3. ResolvePoolQuery resolves DirectHash to a physical node + container
  4. IsTaskLocal checks if the target is this node
  5. RouteLocal calls RuntimeMapTask to pick the worker

This two-level design means containers control what gets executed where, while the scheduler controls how workers are utilized.

Task Group Affinity​

The DefaultScheduler implements task group affinity to pin related tasks to the same worker. Each Container maintains a task_group_map_ mapping group IDs to workers:

  1. When RuntimeMapTask sees a task with a non-null task_group_, it checks the container's task_group_map_
  2. If the group is already mapped to a worker, the task goes to that worker
  3. If not, normal routing selects a worker and records the mapping

This ensures tasks in the same group (e.g., operations on the same file handle) execute on the same worker, avoiding lock contention and improving cache locality.

I/O-Size Routing​

  • Tasks with stat_.io_size_ >= 4096 → round-robin across I/O workers
  • Tasks with stat_.io_size_ < 4096 → scheduler worker (worker 0)
  • Network tasks (admin pool methods 14, 15, 20, 21) → network worker

Force Enqueue​

RouteTask accepts a force_enqueue parameter (default false). When true, RouteLocal always enqueues the task to the destination worker's lane, even if the destination is the current worker. This is used by SendRuntime (non-worker thread path) which cannot execute tasks directly and must always enqueue.

Code Reference​

See implementation in:

  • Header: context-runtime/include/clio_runtime/scheduler/default_sched.h
  • Implementation: context-runtime/src/scheduler/default_sched.cc

Best Practices​

1. Always Update IpcManager in DivideWorkers​

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/ipc_manager.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void DivideWorkers(WorkOrchestrator *work_orch) override {
    // ... partition workers ...
    u32 num_client_lanes = 1;

    IpcManager *ipc = CLIO_IPC;
    if (ipc) {
      ipc->SetNumSchedQueues(num_client_lanes);
      if (net_worker_) {
        ipc->SetNetLane(net_worker_->GetLane(), net_worker_->GetLane());
      }
    }
  }

  // Other Scheduler overrides omitted for brevity.

 private:
  Worker *net_worker_ = nullptr;
};

Why: Clients use GetNumSchedQueues() to map tasks to lanes. If this doesn't match the actual number of workers/lanes, tasks will be mapped to non-existent or wrong workers.

2. Route Network Tasks to the Network Worker​

Both ClientMapTask and RuntimeMapTask should route Send/Recv tasks (methods 14/15/20/21 from admin pool) to the dedicated network worker (last worker). This prevents network I/O from blocking task processing workers.

3. Implement Task Group Affinity​

Use container->task_group_map_ to pin related tasks to the same worker. Protect map access with container->task_group_lock_ (read lock for lookups, write lock for updates).

4. Validate Lane IDs​

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/ipc_manager.h"

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  u32 ClientMapTask(IpcManager *ipc_manager,
                    const Future<Task> &task) override {
    u32 num_lanes = ipc_manager->GetNumSchedQueues();
    if (num_lanes == 0) return 0;

    u32 lane = ComputeLane(task);
    return lane % num_lanes;  // Ensure lane is in valid range
  }

  // Other Scheduler overrides omitted for brevity.

 private:
  u32 ComputeLane(const Future<Task> &task) { return 0; }
};

5. Handle Null Pointers​

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void DivideWorkers(WorkOrchestrator *work_orch) override {
    if (!work_orch) return;
    // ... proceed ...
  }

  u32 RuntimeMapTask(Worker *worker, const Future<Task> &task,
                     ContainerHold container) override {
    (void)task;
    (void)container;
    return worker ? worker->GetId() : 0;
  }

  // Other Scheduler overrides omitted for brevity.
};

6. Consider Thread Safety​

If your scheduler maintains shared state accessed by multiple workers:

  • Use atomic operations for counters (e.g., next_io_idx_)
  • Use CoRwLock for complex data structures (e.g., task_group_map_)
  • Prefer lock-free designs when possible

7. Test with Different Configurations​

Test your scheduler with various num_threads values:

  • Single thread (num_threads = 1): single worker serves dual role
  • Small (num_threads = 2-4)
  • Large (num_threads = 16+)

Integration Points​

Singletons and Macros​

Access runtime components via global macros:

#include "clio_runtime/config_manager.h"
#include "clio_runtime/ipc_manager.h"
#include "clio_ctp/introspect/system_info.h"
#include "clio_ctp/thread/thread_model_manager.h"

using namespace clio::run;

void example() {
  // Configuration
  ConfigManager *config = CLIO_CONFIG_MANAGER;
  u32 num_threads = config->GetNumThreads();

  // IPC Manager
  IpcManager *ipc = CLIO_IPC;
  u32 num_lanes = ipc->GetNumSchedQueues();

  // System Info
  auto *sys_info = CTP_SYSTEM_INFO;
  auto pid = sys_info->pid_;

  // Thread Model
  auto tid = CTP_THREAD_MODEL->GetTid();

  (void)num_threads;
  (void)num_lanes;
  (void)pid;
  (void)tid;
}

Worker Access​

Access workers through WorkOrchestrator:

#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

void example(WorkOrchestrator *work_orch, u32 worker_id) {
  u32 total_workers = work_orch->GetTotalWorkerCount();
  Worker *worker = work_orch->GetWorker(worker_id);

  // Get worker properties
  u32 id = worker->GetId();
  TaskLane *lane = worker->GetLane();

  (void)total_workers;
  (void)id;
  (void)lane;
}

Container Access​

Containers expose scheduling-related state:

#include "clio_runtime/container.h"
#include "clio_runtime/worker.h"

using namespace clio::run;

class MyContainer : public Container {
 public:
  // Override ScheduleTask for custom dynamic scheduling
  PoolQuery ScheduleTask(const clio::run::shared_ptr<Task> &task) override {
    // Transform Dynamic queries into concrete routing modes
    return PoolQuery::DirectHash(ComputeHash(task));
  }

  void InspectAffinityState() {
    // Task group affinity map (public per-container state):
    //   task_group_map_   -> std::unordered_map<int64_t, Worker*>
    //   task_group_lock_  -> CoRwLock protecting the map
    (void)task_group_map_;
    (void)task_group_lock_;
  }

 private:
  u32 ComputeHash(const clio::run::shared_ptr<Task> &task) { return 0; }
};

Logging​

Use Hermes logging macros:

#include <string>

#include "clio_ctp/util/logging.h"

void example() {
  unsigned num_workers = 4, lane_id = 0, worker_id = 1;
  std::string error_msg = "bad config";

  HLOG(kInfo, "Scheduler initialized with {} workers", num_workers);
  HLOG(kDebug, "Mapping task to lane {}", lane_id);
  HLOG(kWarning, "Worker {} has empty queue", worker_id);
  HLOG(kError, "Invalid configuration: {}", error_msg);
}

Configuration Access​

Read configuration values:

#include <string>

#include "clio_runtime/config_manager.h"

using namespace clio::run;

void example() {
  ConfigManager *config = CLIO_CONFIG_MANAGER;
  u32 num_threads = config->GetNumThreads();
  u32 queue_depth = config->GetQueueDepth();
  std::string sched_name = config->GetLocalSched();

  (void)num_threads;
  (void)queue_depth;
  (void)sched_name;
}

Advanced Topics​

Custom Container Scheduling​

Override ScheduleTask to implement application-specific routing:

#include "clio_runtime/container.h"

using namespace clio::run;

class DistributedKVStore : public Container {
 public:
  static constexpr u32 kGet = 10;
  static constexpr u32 kPut = 11;
  static constexpr u32 kDelete = 12;
  static constexpr u32 kScan = 13;
  static constexpr u32 kCreateIndex = 14;

  PoolQuery ScheduleTask(const clio::run::shared_ptr<Task> &task) override {
    switch (task->method_) {
      case kGet:
      case kPut:
      case kDelete: {
        // Route key-value ops to the node owning the key's hash partition
        u32 key_hash = ExtractKeyHash(task);
        return PoolQuery::DirectHash(key_hash);
      }
      case kScan:
        // Range scans may span a range of containers (offset, count)
        return PoolQuery::Range(RangeOffset(task), RangeCount(task));
      case kCreateIndex:
        // Metadata ops broadcast to all containers
        return PoolQuery::Broadcast();
      default:
        return PoolQuery::Local();
    }
  }

 private:
  u32 ExtractKeyHash(const clio::run::shared_ptr<Task> &task) { return 0; }
  u32 RangeOffset(const clio::run::shared_ptr<Task> &task) { return 0; }
  u32 RangeCount(const clio::run::shared_ptr<Task> &task) { return 1; }
};

Work Stealing​

Implement work stealing in RebalanceWorker:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/worker.h"
#include <vector>

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void RebalanceWorker(Worker *worker) override {
    if (!worker) return;
    TaskLane *my_lane = worker->GetLane();
    if (my_lane == nullptr || my_lane->Size() != 0) return;

    for (Worker *victim : io_workers_) {
      if (victim == worker) continue;

      TaskLane *victim_lane = victim->GetLane();
      if (victim_lane != nullptr && victim_lane->Size() != 0) {
        Future<Task> stolen_task;
        if (victim_lane->Pop(stolen_task)) {
          my_lane->Push(stolen_task);
          break;
        }
      }
    }
  }

  // Other Scheduler overrides omitted for brevity.

 private:
  std::vector<Worker *> io_workers_;
};

Priority-Based Scheduling​

Use task priorities for scheduling:

#include "clio_runtime/scheduler/scheduler.h"
#include "clio_runtime/work_orchestrator.h"
#include "clio_runtime/worker.h"
#include <vector>

using namespace clio::run;

class ExampleScheduler : public Scheduler {
 public:
  void DivideWorkers(WorkOrchestrator *work_orch) override {
    u32 total = work_orch->GetTotalWorkerCount();
    u32 high_prio_count = total / 2;

    for (u32 i = 0; i < high_prio_count; ++i) {
      high_priority_workers_.push_back(work_orch->GetWorker(i));
    }

    for (u32 i = high_prio_count; i < total - 1; ++i) {
      low_priority_workers_.push_back(work_orch->GetWorker(i));
    }

    // Network worker
    net_worker_ = work_orch->GetWorker(total - 1);
  }

  // Other Scheduler overrides omitted for brevity.

 private:
  std::vector<Worker *> high_priority_workers_;
  std::vector<Worker *> low_priority_workers_;
  Worker *net_worker_ = nullptr;
};

Troubleshooting​

Tasks Not Being Processed​

Symptom: Tasks submitted but never execute

Check:

  1. Did you call IpcManager::SetNumSchedQueues() in DivideWorkers?
  2. Are all workers getting lanes via WorkOrchestrator's 1:1 mapping?
  3. Does ClientMapTask return lane IDs in valid range?
  4. Is IpcManager::SetNetLane() called for the network worker?

Client Mapping Errors​

Symptom: Assertion failures or crashes in ClientMapTask

Check:

  1. Is returned lane ID in range [0, num_sched_queues)?
  2. Did you check for num_lanes == 0?
  3. Are you using modulo to wrap lane IDs?

Tasks Hang After Re-enqueue​

Symptom: Tasks enqueued to a different worker never complete

Check:

  1. When RouteLocal re-enqueues to a different worker, ProcessNewTask on the destination worker updates the RunContext's worker_id_, lane_, and event_queue_ to match the new worker. If RunContext fields are stale, subtask completion events go to the wrong worker.
  2. Check that TASK_STARTED is respected — re-enqueued tasks with live coroutines must be resumed, not restarted.

Worker Crashes​

Symptom: Workers crash during initialization

Check:

  1. Are you checking for null pointers?
  2. Does DivideWorkers handle total_workers < expected?
  3. Is the single-worker case handled (when total_workers == 1)?

References​

  • Scheduler Interface: context-runtime/include/clio_runtime/scheduler/scheduler.h
  • DefaultScheduler: context-runtime/src/scheduler/default_sched.cc
  • Container Base: context-runtime/include/clio_runtime/container.h
  • IpcManager (RouteTask/RouteLocal): context-runtime/src/ipc_manager.cc
  • WorkOrchestrator: context-runtime/src/work_orchestrator.cc
  • Configuration: Configuration Reference