JobScheduler.hpp

Go to the documentation of this file.

// SPDX-FileCopyrightText: 2026 Povl Filip Sonne-Frederiksen
//
// SPDX-License-Identifier: GPL-3.0-or-later
 
#pragma once
 
// A process-wide pipeline job scheduler over many projects (spec 2026-10-08,
// phase S2: ruxd serves several cases from one process).
//
// One JobScheduler owns a bounded pool of worker threads. Each project gets a
// JobQueue from it — a FIFO of that project's jobs, with the same submit /
// status / cancel / listener / writer-lock surface JobRunner always had. The
// rules:
//
//  * **At most one running job per queue.** A project's jobs run strictly in
//    submission order, one at a time, under that project's writer lock — two
//    stages writing the same `.rux` at once is never allowed.
//  * **At most `workers` running jobs in total.** The default is one: the GPU
//    is shared, and the stages are TBB-parallel internally already.
//  * **Round-robin between queues.** A free worker takes the next job from the
//    next queue in rotation that has work and is not already running, so a
//    project that queues ten jobs cannot starve one that queues one.
//
// PROGRESS is per job: the worker installs a core::ScopedProgressObserver for
// the job it runs, so two jobs running at once on two workers each see only
// their own stage's events (core/processing_observer.hpp explains why updates
// made from pool threads still arrive).
//
// RECORDS go to an IJobStore at every state transition (JobStore.hpp): in
// memory for `ruxd --local`, Postgres in server mode. Live progress stays here.
 
#include "reusex/pipeline/JobStore.hpp"
#include "reusex/pipeline/job_types.hpp"
#include "reusex/pipeline/stages.hpp"
 
#include <chrono>
#include <cstddef>
#include <filesystem>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
 
namespace reusex::pipeline {
 
struct JobSchedulerOptions {
  std::size_t workers = 1;
};
 
class JobQueue;

class JobScheduler {
    public:
  explicit JobScheduler(JobSchedulerOptions options = {},
                        std::shared_ptr<IJobStore> store = nullptr);

  ~JobScheduler();
 
  JobScheduler(const JobScheduler &) = delete;
  JobScheduler &operator=(const JobScheduler &) = delete;

  std::unique_ptr<JobQueue>
  make_queue(std::string key, std::filesystem::path project,
             StageExecutor executor = default_stage_executor());
 
  std::size_t workers() const noexcept;

  std::size_t running_count() const;

  std::shared_ptr<IJobStore> store() const;
 
  struct Core; 
 
    private:
  std::shared_ptr<Core> core_;
};

class JobQueue {
    public:
  ~JobQueue();
 
  JobQueue(const JobQueue &) = delete;
  JobQueue &operator=(const JobQueue &) = delete;
 
  const std::string &key() const noexcept;
  const std::filesystem::path &project() const noexcept;

  std::string submit(JobStage stage, std::string parameters = {},
                     std::string submitted_by = {});

  bool cancel(std::string_view id);

  std::optional<JobRecord> job(std::string_view id) const;

  std::vector<JobRecord> jobs() const;

  std::size_t queued_count() const;

  bool is_busy() const;

  bool has_work() const;

  void wait_idle();

  [[nodiscard]] WriterLease
  try_acquire_writer(std::chrono::milliseconds timeout);

  std::size_t add_listener(JobListener listener);
  void remove_listener(std::size_t token);
 
  struct State; 
 
    private:
  friend class JobScheduler;
  JobQueue(std::shared_ptr<JobScheduler::Core> core,
           std::shared_ptr<State> state);
 
  std::shared_ptr<JobScheduler::Core> core_;
  std::shared_ptr<State> state_;
};
 
} // namespace reusex::pipeline