高性能 C++ 调度器设计与实现

• CPU 和内存的预留/释放机制
• 启动前检查资源是否足够，不足则等待
• 任务结束后自动释放资源

2.1.3 任务生命周期

enum class JobStatus {
    Pending,      // 已提交但尚未调度
    Running,      // 正在运行
    Succeeded,    // 成功结束（exit 0）
    Failed,       // 失败结束（非零退出码）
    Timeout,      // 超时被终止
    Cancelled     // 被取消
};

• 完整状态转换：Pending → Running → Succeeded/Failed/Timeout
• 超时管理：两阶段终止（SIGTERM → 宽限期 → SIGKILL）
• 进程组管理：整组清理避免子进程泄漏

2.1.4 可观测性

• 指标采集：提交数、拒绝数、运行数、成功/失败/超时数、排队延迟
• HTTP 导出：Prometheus 兼容的 /metrics 端点
• 健康检查：/health 端点返回 ok

2.2 非功能性需求

• 线程安全：所有共享状态通过互斥锁保护
• 资源可控：任务结束后正确释放资源，避免泄漏
• 优雅退出：stop() 能够等待任务完成或按策略终止
• 日志完善：关键操作有日志记录，便于排障

2.3 可选特性

• cgroup v2 隔离：CPU 配额和内存限制
• PSI 背压监测：根据系统压力暂停新任务启动
• SQLite 持久化：支持重启后恢复未完成任务
• Cron 调度：支持定时触发任务（简化版 cron 表达式）

架构设计

3.1 总体架构

3.2 线程模型

线程职责：

1. 主线程：处理外部 submit() 调用，加锁操作 pending_ 队列
2. dispatcher 线程：从 pending_ 取任务，检查资源，fork/exec 启动进程
3. reaper 线程：周期性 waitpid 回收子进程，处理超时，释放资源
4. psi 线程（可选）：读取 cgroup pressure 文件，更新背压标志
5. cron 线程（可选）：检查模板到期时间，生成任务实例
6. http 线程（可选）：处理 HTTP 请求，返回指标或健康状态

3.3 数据流

提交路径：

submit() → 校验白/黑名单 → 检查队列上限 → pending_.push_back() → inc_submitted() → cv_.notify_all()

调度路径：

dispatcher_loop() → pick_next_job() → rm_.reserve() → launch_job() → fork/exec → running_[id] = job

回收路径：

reaper_loop() → waitpid(WNOHANG) → 更新 exit_code/status → rm_.release() → cleanup_cgroup() → running_.erase(id)

核心模块设计

4.1 Scheduler（调度器）

职责：编排整个调度流程，管理任务生命周期

关键接口：

class Scheduler {
 public:
    explicit Scheduler(SchedulerOptions opts);
    // 提交任务，返回 job id 或 -1
    int submit(const JobSpec& spec);
    // 启动后台线程
    void start();
    // 停止调度器并等待线程退出
    void stop();
    // 判断是否空闲（无待处理和运行中任务）
    bool idle() const;
    // 获取指标快照
    Metrics::Snapshot metrics() const;
 private:
    void dispatcher_loop();
    void reaper_loop();
    void psi_loop();
    void cron_loop();
    bool launch_job(Job& job);
    bool pick_next_job(Job& out);
    void restore_from_store();
};

关键数据结构：

private:
    SchedulerOptions opts_;           // 配置
    ResourceManager rm_;              // 资源管理器
    std::vector<Job> pending_;        // 待调度队列
    std::unordered_map<int, Job> running_; // 运行中任务表
    mutable std::mutex mu_;           // 互斥锁
    std::condition_variable cv_;      // 条件变量
    std::atomic<bool> shutting_down_{false}; // 关闭标志
    std::atomic<bool> psi_backpressure_{false}; // 背压标志
    Metrics metrics_;                 // 指标收集器
    std::unique_ptr<JobStore> store_; // 持久化存储
    std::unique_ptr<CronScheduler> cron_sched_; // Cron 调度器
    std::unique_ptr<MetricsHttpServer> metrics_server_; // HTTP 服务

4.2 ResourceManager（资源管理器）

职责：管理 CPU 和内存配额，提供预留/释放接口

实现要点：

class ResourceManager {
 public:
    explicit ResourceManager(ResourceQuota quota);
    // 尝试预留资源，成功返回 true
    bool reserve(int cpu, size_t mem_mb);
    // 释放资源（必须与 reserve 配对调用）
    void release(int cpu, size_t mem_mb);
    // 查询当前使用情况
    std::pair<int, size_t> used() const;
 private:
    ResourceQuota quota_;
    int used_cpu_{0};
    size_t used_mem_mb_{0};
    mutable std::mutex mu_;
};

核心逻辑（src/resource_manager.cpp）：

bool ResourceManager::reserve(int cpu, size_t mem_mb) {
    std::lock_guard lk(mu_);
    // 不做部分分配：要么全部满足要么拒绝
    if (used_cpu_ + cpu > quota_.total_cpu || used_mem_mb_ + mem_mb > quota_.total_mem_mb) {
        return false;
    }
    used_cpu_ += cpu;
    used_mem_mb_ += mem_mb;
    return true;
}

4.3 Metrics（指标收集器）

职责：提供原子计数器，生成快照和 Prometheus 文本

Prometheus 导出示例：

# TYPE tasks_total counter
tasks_total{status="submitted"} 100
tasks_total{status="rejected"} 5
tasks_total{status="succeeded"} 80
tasks_total{status="failed"} 10
tasks_total{status="timeout"} 5

# TYPE tasks_running_current gauge
tasks_running_current 3

# TYPE tasks_pending_current gauge
tasks_pending_current 5

4.4 CgroupHelper（cgroup 辅助）

职责：创建、绑定和清理任务专属 cgroup

接口：

// 创建任务 cgroup 并设置 CPU/内存限制
std::string create_cgroup_for_job(int job_id, int cpu_cores, size_t mem_mb, const CgroupConfig& cfg);
// 将 pid 加入 cgroup
bool attach_pid_to_cgroup(pid_t pid, const std::string& cg_path);
// 清理 cgroup 目录
void cleanup_cgroup(const std::string& cg_path);

实现细节（src/cgroup_helper.cpp）：

• 在 cfg.base_path 下创建 job_ 子目录
• 写入 cpu.max：<quota_us> <period_us>，例如 100000 100000 表示 1 核
• 写入 memory.max：字节数，例如 268435456 表示 256MB
• 写入 cgroup.procs：将 pid 加入该 cgroup

4.5 JobStore（持久化存储）

职责：通过 SQLite 持久化任务状态，支持重启恢复

数据模型：

CREATE TABLE jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    cmd TEXT NOT NULL,
    cpu_cores INTEGER,
    memory_mb INTEGER,
    timeout_sec INTEGER,
    priority INTEGER,
    status TEXT, -- queued/running/succeeded/failed/timeout/launch_failed
    submit_ms INTEGER,
    start_ms INTEGER,
    end_ms INTEGER,
    exit_code INTEGER
);

关键接口：

class JobStore {
 public:
    bool init(const std::string& path);
    // 插入任务
    int insert_job(const JobSpec& spec, PersistStatus status, int64_t submit_ms, ...);
    // 更新状态
    void update_status(int id, PersistStatus status, ...);
    // 加载未完成任务
    std::vector<PersistedJob> load_unfinished();
};

恢复策略（Scheduler::restore_from_store()）：

void Scheduler::restore_from_store() {
    if (!store_) return;
    auto jobs = store_->load_unfinished();
    for (auto& pj : jobs) {
        Job job;
        job.id = next_id_++;
        job.spec = pj.spec;
        job.status = JobStatus::Pending;
        job.enqueue_time = std::chrono::steady_clock::now();
        pending_.push_back(job);
    }
    cv_.notify_all();
}

4.6 CronScheduler（定时触发）

职责：管理 cron 模板，定时生成任务实例

当前实现：

• 支持简化表达式：@every s，例如 @every 60s 每 60 秒触发一次
• 完整 5 字段 cron（分 时 日 月 周）接口已定义但简化实现

核心逻辑：

void CronScheduler::tick(SubmitCallback submit_cb) {
    auto now = std::chrono::system_clock::now();
    for (auto& tpl : templates_) {
        if (!tpl.enabled) continue;
        if (now >= tpl.next_run) {
            submit_cb(tpl.spec); // 提交任务实例
            tpl.next_run = tpl.cron.next_run(now); // 计算下次触发时间
        }
    }
}

4.7 MetricsHttpServer（HTTP 指标服务）

职责：提供轻量级 HTTP 服务，导出指标和健康检查

路由：

• GET /metrics：返回 Prometheus 文本格式指标
• GET /health：返回 ok

并发模型：

• 监听线程：accept() 接收连接
• 工作线程池：处理请求，生成响应
• 连接队列：有限长度，避免内存膨胀

实现细节（src/metrics_http_server.cpp）：

class MetricsHttpServer {
 public:
    using MetricsHandler = std::function<std::string()>;
    bool start(int port, MetricsHandler handler);
    void stop();
 private:
    void accept_loop();
    void worker_loop();
    // ...
};

关键流程设计

5.1 任务提交流程

5.2 调度启动流程

关键代码（Scheduler::launch_job()）：

bool Scheduler::launch_job(Job& job) {
    // 创建 cgroup
    std::string cg_path;
    if (opts_.cgroup.enabled) {
        cg_path = create_cgroup_for_job(job.id, job.spec.cpu_cores, job.spec.memory_mb, opts_.cgroup);
    }

    pid_t pid = fork();
    if (pid == 0) {
        // 子进程
        setpgid(0, 0); // 建立独立进程组

        if (!cg_path.empty()) {
            attach_pid_to_cgroup(getpid(), cg_path);
        }

        // 设置 rlimit
        if (opts_.rlimit_nofile >= 0) {
            struct rlimit rl;
            rl.rlim_cur = rl.rlim_max = opts_.rlimit_nofile;
            setrlimit(RLIMIT_NOFILE, &rl);
        }

        if (opts_.disable_core_dump) {
            struct rlimit rl;
            rl.rlim_cur = rl.rlim_max = 0;
            setrlimit(RLIMIT_CORE, &rl);
        }

        // 切换工作目录
        if (!opts_.workdir.empty()) {
            chdir(opts_.workdir.c_str());
        }

        // 执行命令
        execl("/bin/sh", "sh", "-c", job.spec.cmd.c_str(), nullptr);
        _exit(127);
    }

    // 父进程
    job.pid = pid;
    job.pgid = pid;
    job.start_time = std::chrono::steady_clock::now();
    job.status = JobStatus::Running;
    return true;
}

5.3 回收超时流程

关键代码（Scheduler::reaper_loop()）：

void Scheduler::reaper_loop() {
    while (!shutting_down_.load()) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));

        std::lock_guard lk(mu_);
        auto now = std::chrono::steady_clock::now();

        for (auto it = running_.begin(); it != running_.end(); ) {
            Job& job = it->second;

            // 检查超时
            if (job.spec.timeout_sec > 0) {
                auto elapsed = std::chrono::duration_cast<std::chrono::seconds>(
                    now - job.start_time).count();

                if (elapsed >= job.spec.timeout_sec) {
                    if (!job.sigterm_sent) {
                        // 发送 SIGTERM
                        kill(-job.pgid, SIGTERM);
                        job.sigterm_sent = true;
                        job.kill_deadline = now + std::chrono::seconds(opts_.kill_grace_sec);
                    } else if (now >= *job.kill_deadline) {
                        // 发送 SIGKILL
                        kill(-job.pgid, SIGKILL);
                    }
                }
            }

            // 尝试回收
            int status;
            pid_t ret = waitpid(job.pid, &status, WNOHANG);
            if (ret > 0) {
                // 进程已退出
                job.exit_code = status;
                job.end_time = now;

                if (job.sigterm_sent) {
                    job.status = JobStatus::Timeout;
                    metrics_.inc_timeout();
                } else if (WIFEXITED(status) && WEXITSTATUS(status) == 0) {
                    job.status = JobStatus::Succeeded;
                    metrics_.inc_succeeded();
                } else {
                    job.status = JobStatus::Failed;
                    metrics_.inc_failed();
                }

                // 释放资源
                rm_.release(job.spec.cpu_cores, job.spec.memory_mb);
                metrics_.dec_running();

                // 清理 cgroup
                if (opts_.cgroup.enabled) {
                    std::string cg_path = opts_.cgroup.base_path + "/job_" + std::to_string(job.id);
                    cleanup_cgroup(cg_path);
                }

                it = running_.erase(it);
            } else {
                ++it;
            }
        }
    }
}

5.4 PSI 背压流程

目标：根据系统压力暂停新任务启动，避免雪崩

实现原理：

1. psi_thread 周期性读取 /sys/fs/cgroup/scheduler/memory.pressure 和 cpu.pressure
2. 解析 avg10 值（10 秒平均压力）
3. 与阈值比较（如 50.0），超过则设置 psi_backpressure_ = true
4. dispatcher_loop 检查背压标志，若为 true 则跳过本轮调度

关键代码：

void Scheduler::psi_loop() {
    const double threshold = 50.0; // 背压阈值

    while (!shutting_down_.load()) {
        std::this_thread::sleep_for(std::chrono::seconds(1));

        // 读取 memory.pressure
        std::string mem_pressure_file = opts_.cgroup.base_path + "/memory.pressure";
        std::ifstream ifs(mem_pressure_file); // 格式：some avg10=12.34 avg60=... total=...

        double avg10 = parse_psi_avg10(ifs);
        bool pressure = (avg10 > threshold);

        if (pressure != psi_backpressure_.load()) {
            psi_backpressure_.store(pressure);
            metrics_.set_pressure_active(pressure);
            Logger::instance().log(Logger::Level::Info, pressure ? "PSI backpressure activated" : "PSI backpressure cleared");
        }
    }
}

配置与接口

6.1 配置选项

struct SchedulerOptions {
    ResourceQuota quota;              // 资源配额
    CgroupConfig cgroup;              // cgroup 配置
    int max_queue_size{1000};         // 最大队列长度
    int kill_grace_sec{2};            // SIGTERM 宽限期
    bool enable_priority{false};      // 是否启用优先级调度
    bool enable_psi_monitor{false};   // 是否启用 PSI 背压监测
    std::vector<std::string> cmd_whitelist; // 命令白名单
    std::vector<std::string> cmd_blacklist; // 命令黑名单
    std::string workdir;              // 工作目录
    int metrics_http_port{-1};        // HTTP 指标端口
    int rlimit_nofile{-1};            // 文件句柄限制
    bool disable_core_dump{true};     // 禁用 core dump
    bool enable_persistence{false};   // 启用持久化
    std::string db_path{"state/tasks.db"}; // 持久化路径
    bool enable_cron{false};          // 启用 cron
    int cron_tick_ms{1000};           // cron 检查间隔
};

6.2 命令行接口

./scheduler \
    --cmd "echo hello" \
    --cpu 1 \
    --mem 256 \
    --timeout 5 \
    --priority 10 \
    --total-cpu 4 \
    --total-mem 2048 \
    --cgroup \
    --enable-priority \
    --metrics-port 8080 \
    --whitelist ls,echo \
    --blacklist rm,shutdown \
    --workdir /tmp

6.3 HTTP 接口

健康检查：

$ curl http://localhost:8080/health
ok

指标导出：

$ curl http://localhost:8080/metrics
# TYPE tasks_total counter
tasks_total{status="submitted"} 100
tasks_total{status="rejected"} 5
tasks_total{status="succeeded"} 80
...