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

内容来自：程序员老廖

1. 项目概述

1.1 背景

TaskScheduler 是一个 C++20 实现的单机任务调度器，用于管理和执行本地任务。它提供了完整的任务生命周期管理能力，包括任务提交、资源管理、进程执行、超时控制和状态跟踪。

1.1.1 单机调度器能用在哪些领域？

1. AI/机器学习训练与推理（最热门场景）

场景：一台服务器上有多个 GPU，需要同时运行多个训练或推理任务。

需求：

• 按 GPU 显存、计算单元分配任务

• 防止任务互相抢占资源（如一个任务占满显存导致其他 OOM）

• 支持优先级（高优模型优先调度）

例子：

• 字节的 ByteScheduler 在单机上调度多卡训练任务

• 华为 MindSpore 的本地执行引擎包含轻量调度模块

• 公司内部的 模型 A/B 测试平台，需并发跑多个小模型

这类系统往往用 C++ 写调度器 + Python 做用户接口。

2. CI/CD 与自动化测试平台

场景：GitLab Runner、Jenkins Agent、内部构建系统。

需求：

• 并发执行多个构建/测试任务（如 make test、npm build）

• 限制每个任务的 CPU/内存，防止一个坏任务拖垮整台机器

• 任务队列管理、失败重试、超时控制

为什么不用 Docker？

• 启动开销大（冷启动 100ms+），而轻量调度器可做到 <10ms

• 某些安全环境禁止容器，只能用进程隔离

腾讯、阿里、美团都有自研的 高性能 CI 执行引擎，底层就是单机调度器。

3. Serverless / FaaS（函数即服务）的本地运行时

场景：用户提交一个函数（如 Python lambda），平台在本地执行。

需求：

• 快速启动、资源隔离、超时 kill

• 高并发（单机每秒处理数百个短生命周期函数）

例子：

• 阿里云 函数计算（FC） 的 Worker 节点

• 华为云 FunctionGraph 的执行代理

技术栈：C++ 调度器 + gVisor/firecracker（可选） + 快照恢复

关键指标：冷启动延迟 < 50ms，这正是 C++ 单机调度器的优势。

4. 边缘计算（Edge Computing）

场景：摄像头、IoT 网关、车载设备等资源受限设备。

需求：

• 低内存占用（<50MB）

• 无依赖（不能跑 K8s）

• 支持定时任务、事件触发任务

例子：

• 自动驾驶车上的感知模块调度

• 工厂 PLC 设备上的数据预处理任务

华为、百度 Apollo、大疆等公司在边缘端大量使用 C++ 调度框架。

5. 游戏服务器（Game Server）

场景：一个物理机部署多个游戏房间（Room）实例。

需求：

• 每个房间独立进程，防止单点崩溃影响全局

• 动态扩缩容（玩家多就多开房间）

• 低延迟通信（调度器需快速响应负载变化）

腾讯天美、米哈游、网易雷火都有类似架构。

1.1.2 互联网大厂真的需要调度器吗？

答案是：非常需要！但通常作为“基础设施组件”而非独立产品

招聘关键词搜索建议：

• “任务调度”

• “资源管理”

• “执行引擎”

• “C++ 后台开发（基础架构）”

• “Serverless Runtime”

1.2 核心目标

• 任务调度：支持任务提交、排队、调度和执行的完整闭环

• 资源管理：基于 CPU 核数和内存上限进行准入控制

• 进程隔离：通过 fork/exec 执行任务，可选 cgroup v2 资源限制

• 可观测性：提供指标采集、HTTP 导出和 Prometheus 兼容格式

• 高可靠性：支持持久化恢复、超时终止、信号管理和 PSI 背压

1.3 技术特点

• 多线程模型：独立的调度、回收、PSI 监测和 Cron 触发线程

• 资源感知：预留/释放机制避免资源超卖

• 灵活配置：支持优先级调度、命令白/黑名单、工作目录限制

• 轻量级实现：核心代码约 600 行，依赖 SQLite 和 Linux 系统调用

视频讲解与源码领取：C++可写简历项目-高性能C++调度器开发

2. 需求分析

2.1 功能性需求

2.1.1 任务提交与管理

struct JobSpec { std::string cmd; // 要执行的命令字符串 int cpu_cores{1}; // 需要的 CPU 核数 size_t memory_mb{256}; // 需要的内存 MB int timeout_sec{0}; // 超时秒数，0 表示不限制 int priority{0}; // 优先级，数值越大优先级越高 };

• 支持通过 JobSpec 提交任务，包含命令、资源需求、超时和优先级

• 队列长度限制：max_queue_size 配置，超出则拒绝提交

• 命令准入：支持白名单/黑名单校验

2.1.2 资源配额管理

struct ResourceQuota { int total_cpu{4}; // 可用 CPU 总核数 size_t total_mem_mb{2048}; // 可用内存总量 MB };

• 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. 架构设计

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. 核心模块设计

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(); // PSI 监测循环 void cron_loop(); // Cron 触发循环 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_<id> 子目录

• 写入 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 <sec>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. 关键流程设计

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. 配置与接口

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 \ # CPU 核数 --mem 256 \ # 内存 MB --timeout 5 \ # 超时秒数 --priority 10 \ # 优先级 --total-cpu 4 \ # 总 CPU --total-mem 2048 \ # 总内存 --cgroup \ # 启用 cgroup --enable-priority \ # 启用优先级 --metrics-port 8080 \ # HTTP 端口 --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 ...

视频讲解与源码领取：C++可写简历项目-高性能C++调度器开发