coglet 深度解析Cog 机器学习预测服务器的纯 Rust 核心运行时【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cogcoglet 是 Cog 项目Containers for machine learning中负责预测服务执行的 Rust 核心库它以纯 Rust、零 Python 依赖的方式实现了子进程隔离、并发槽位管理与高性能 IPC是 Cog 预测服务器coglet-python 的底层引擎的心脏。阅读本文后你将完整掌握 coglet 的父/子进程双通道架构、PredictionService/Orchestrator/Worker三大核心组件的工作原理、PermitPool槽位并发控制与桥接协议细节以及健康状态机、取消与优雅关闭的完整行为语义。coglet 在 Cog 体系中的定位coglet 的核心定位写在其自身文档首行它是 coglet 预测服务器的核心 Rust 库Core Rust library for the coglet prediction server纯 Rust 实现、不依赖任何 Python 库Python 绑定则独立存在于 coglet-python 中。这种分层意味着预测服务器的全部核心逻辑进程编排、并发控制、IPC 协议、HTTP 服务都可以用 Rust 独立测试与演进Python 侧只保留薄薄的 PyO3 绑定层通过 lib.rs 暴露serve()、active()、_run_worker()等入口整个 crates 工作区见 crates/README.md由coglet核心库与coglet-pythonPyO3 绑定两个 crate 组成统一由 Cargo.toml 工作区清单管理。从整体架构上看coglet 实现了 Cog 的子进程隔离模型HTTP 请求进入父进程父进程把预测任务下发给子进程worker子进程内运行 Python 预测器load()/setup()/predict()。这种模型带来三个核心收益崩溃隔离worker 崩溃可重启而父进程存活、内存隔离GPU 内存泄漏不会累积、以及可按需 SIGKILL 的干净关闭能力。总体架构父进程编排 双通道 IPCcoglet 的核心架构可以浓缩为下面这张取自 crates/coglet/README.md 的模块图coglet ┌─────────────────────────────────────────────────────────────────┐ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ transport/http │ │ │ │ ┌──────────────┐ ┌─────────────────────────────────┐ │ │ │ │ │ server.rs │ │ routes.rs │ │ │ │ │ │ Axum setup │ │ /health, /predictions, /cancel │ │ │ │ │ └──────────────┘ └─────────────────────────────────┘ │ │ │ └───────────────────────────────┬─────────────────────────┘ │ │ │ │ │ ┌───────────────────────────────▼─────────────────────────┐ │ │ │ service.rs │ │ │ │ PredictionService: health, permits, state, webhooks │ │ │ └───────────────────────────────┬─────────────────────────┘ │ │ │ │ │ ┌────────────────────────┼────────────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────┐ ┌────────────────────┐ ┌──────────┐ │ │ │ permit/ │ │ orchestrator.rs │ │webhook.rs│ │ │ │ PermitPool │ │ Parent-side: │ │ Sender │ │ │ │ Slot alloc │ │ spawn, route │ │ Retry │ │ │ └─────────────┘ └─────────┬──────────┘ └──────────┘ │ │ │ │ │ ┌────────────────────────────▼────────────────────────────┐ │ │ │ bridge/ │ │ │ │ ┌──────────────┐ ┌─────────────┐ ┌────────────────┐ │ │ │ │ │ protocol.rs │ │ codec.rs │ │ transport.rs │ │ │ │ │ │ Message types│ │ JSON lines │ │ Unix sockets │ │ │ │ │ └──────────────┘ └─────────────┘ └────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ worker.rs │ │ │ │ Child-side: PredictHandler trait, run_worker loop │ │ │ └─────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘各层职责清晰可辨层次模块职责HTTP 传输层transport/http/server.rs、routes.rsAxum 服务装配与路由处理/health、/predictions、/cancel等服务层service.rsPredictionService健康状态、许可池、预测状态、webhook 的统一所有者编排层orchestrator.rs父进程侧spawn 子进程、事件循环、消息路由并发控制permit/pool.rsPermitPool槽位许可管理与分配Webhookwebhook.rsWebhookSender节流、重试IPC 桥bridge/protocol.rs、codec.rs、transport.rs消息类型、JSON lines 编解码、Unix socket 传输子进程侧worker.rsPredictHandlertrait、run_worker事件循环双通道 IPC控制通道与槽位套接字父进程与 worker 之间并非单一通道而是精心设计的两套通道对应 protocol.rs 的注释说明控制通道stdin/stdoutJSON lines承载生命周期消息——Init、Cancel、Shutdown、Healthcheck父 → 子以及Ready、Idle、Failed、Fatal、ShuttingDown子 → 父。一行一条 JSON 消息。槽位套接字Unix domain socket每个 slot 一条承载预测数据——SlotRequest::Predict下发放LogLine、OutputChunk、Metric、Done、Failed、Cancelled回传。每个槽位独立 socket避免头端阻塞head-of-line blocking。从源码实现看slot socket 在父进程侧通过 NamedSocketTransport::create 创建路径格式为{temp_dir}/coglet-{pid}/slot-{n}.sock在 Linux 上还支持抽象命名空间AbstractSocketTransport无文件系统残留、自动清理。worker 子进程侧则通过connect反连回这些 socket随后父进程调用accept_connections完成握手。目录结构一个 crate一套清晰分层目录树 展示了源码组织的完整脉络以下为节选并标注核心注释coglet/ └── src/ ├── lib.rs # Public API exports │ │ # Core Types ├── health.rs # Health, SetupStatus, SetupResult ├── prediction.rs # Prediction state machine ├── predictor.rs # PredictionResult, PredictionError, PredictionOutput ├── version.rs # VersionInfo │ │ # Service Layer ├── service.rs # PredictionService - lifecycle, state, webhooks ├── webhook.rs # WebhookSender, webhook types │ │ # Orchestrator (Parent Process) ├── orchestrator.rs # spawn_worker, OrchestratorHandle, event loop │ │ # Worker (Child Process) ├── worker.rs # run_worker, PredictHandler trait, SetupError │ │ # Concurrency Control ├── permit/ │ ├── pool.rs # PermitPool - slot permit management │ └── slot.rs # PredictionSlot - permit prediction binding │ │ # IPC Bridge ├── bridge/ │ ├── protocol.rs # ControlRequest, ControlResponse, SlotRequest, SlotResponse │ ├── codec.rs # JsonCodec - newline-delimited JSON │ └── transport.rs # Unix socket transport, ChildTransportInfo │ │ # HTTP Transport └── transport/ └── http/ ├── server.rs # ServerConfig, serve() └── routes.rs # Route handlers, request/response types公共 API 导出集中在 lib.rs包括Health、SetupResult、Prediction、PredictionStatus、PredictionResult、PredictionService、Orchestrator、PermitPool、InputValidator、run_worker等并提供了install_crypto_provider()用于一次性安装 rustls 的 ring TLS 加密提供程序必须在任何reqwest::Client创建前调用可重复调用。三大核心组件深入PredictionService预测状态的唯一所有者service.rs 中的PredictionService是与传输层无关的预测生命周期管理服务它统一管理健康状态Unknown → Starting → Ready/SetupFailedPermitPool Orchestrator 引用两者通过OrchestratorState原子地一起设置保证池与编排器总是同时就绪活动预测DashMapString, PredictionEntry作为预测状态的单一事实来源single source of truth取消CancellationToken 编排器委托Webhook从Prediction的 mutation 方法set_processing、set_succeeded等触发不维护双份状态。典型的组装方式取自 README 示例let service PredictionService::new_no_pool() .with_health(Health::Starting) .with_version(version); // Later, after worker is ready: service.set_orchestrator(pool, handle).await; service.set_health(Health::Ready).await;几个值得注意的源码细节set_health(Health::Ready)在没有编排器时会静默忽略源码中有明确 warn 日志保证 READY 必须先有编排器HealthSnapshot提供is_ready()与is_busy()READY 但可用槽位为 0判定strip_and_validate_input()在单次锁获取内完成剔除未知字段 校验 为省略的可选无默认值字段注入 null三步strip_validate_inject保证 predict 与 train 两条路径的排序不变量完全一致输入超过MAX_INLINE_IPC_SIZE6 MiB见 protocol.rs时build_slot_request会把输入溢出写盘到/tmp/coglet/predictions/{id}/inputs/spill_*.jsonworker 端rehydrate_input读盘、反序列化后立即删除该文件同步预测使用SyncPredictionGuardHTTP 连接断开时 axum 丢弃响应 future从而触发 guard 的 drop进而调用service.cancel(id)同时触发 CancellationToken 与编排器取消disarm()可在正常结束时解除武装。Orchestrator父进程侧的 worker 生命周期管理orchestrator.rs 负责 spawn 子进程并维持其生命周期README 给出了完整的启动流程spawn_worker(config) │ ├─▶ Create Unix socket transport (N slots) ├─▶ Spawn: python -c import coglet; coglet.server._run_worker() ├─▶ Send Init message via stdin ├─▶ Wait for worker to connect sockets ├─▶ Wait for Ready message (with timeout) ├─▶ Populate PermitPool with slot writers ├─▶ Spawn event loop task └─▶ Return OrchestratorReady {pool, schema, handle}事件循环统一处理来自 worker 的各种响应ControlResponse::Idle—— 槽位可接收下一个预测ControlResponse::Failed—— 槽位被毒化poisoned标记不可用SlotResponse::Log/Output/Done/Failed—— 路由到对应预测worker 崩溃—— 失败所有进行中的预测。补充说明事件循环还处理ControlResponse::Fatalworker 不可恢复错误父进程应毒化所有槽位并失败所有在途预测、DroppedLogs背压丢弃日志的系统诊断以及HealthcheckResult用户自定义健康检查结果。此外upload_file实现了与 Python cog 的put_file_to_signed_endpoint一致的签名上传逻辑PUT Content-Type、跟随重定向、从 Location 头取最终 URL 并剥离查询参数。Worker子进程侧的事件循环worker.rs 是子进程侧实现核心是PredictHandlertrait 与run_worker循环run_worker(handler, config) │ ├─▶ Connect to slot sockets (from env) ├─▶ Setup control channel (stdin/stdout) ├─▶ Run handler.setup() with log routing ├─▶ Send Ready {slots, schema} ├─▶ Enter event loop: │ - ControlRequest::Cancel → handler.cancel(slot) │ - ControlRequest::Shutdown → exit │ - SlotRequest::Predict → spawn prediction task └─▶ Exit on shutdown or all slots poisoned两个源码级防护机制值得注意panic hook 即致命错误通道worker 安装全局 panic hookinstall_panic_hook任何 panic 都会尽力发送ControlResponse::Fatal { reason }给父进程然后std::process::abort()。这意味着任意调用点的panic!/.expect()都会自动获得正确的致命行为无需额外辅助代码日志截断worker 日志经truncate_worker_log在 4 MiB 处按字符边界截断并追加[**** LOG LINE TRUNCATED AT 4 MiB ****]标记protocol.rs 中的truncate_worker_log测试覆盖了长/短日志与多字节 UTF-8 场景避免超大日志行引发 panic 或撑爆通道。worker 进程内部的 Python 侧结构见 crates/README.md包含PythonPredictorload()/setup()/predict()、SlotLogWriter基于 ContextVar 的 stdout/stderr 路由与Audit Hook保护运行流、对用户覆写采用 Tee 模式这些由 coglet-python/src 下的log_writer.rs、audit.rs、cancel.rs等实现。PermitPool基于槽位的并发控制permit/pool.rs 实现了槽位 许可的并发控制模型max_concurrency决定槽位数量每个槽位同一时刻最多运行一个预测。README 给出的核心用法let pool PermitPool::new(max_concurrency); // Add slot with its socket writer pool.add_permit(slot_id, writer); // Acquire permit (returns None if at capacity) let permit pool.try_acquire()?; // Send prediction request permit.send(SlotRequest::Predict { id, input }).await?; // Return permit when done drop(permit);源码实现采用了typestate类型状态模式保证编译期状态转换安全许可permit有三种状态类型状态行为PermitInUse正在运行预测into_idle()转为空闲drop 时归还池into_poisoned()转为毒化永久不归还PermitIdle完成后 drop 自动归还池除非池级 poison 标志已置位PermitPoisoned永久失败drop 时仅告警容量缩减关键设计是毒化poisoning是池级属性pool.poison(slot_id)无论槽位是空闲在池中还是在用被预测持有都会置位共享的AtomicBool标志try_acquire会跳过已毒化许可PermitIdle::drop看到标志后也不再归还。配套的SlotIdleToken机制确保只有 worker 确认槽位空闲后许可才归还——若 5 秒内未被消费会打印告警ALERT_THRESHOLD提示槽位可能无法回归池中。相关单元测试pool_add_and_acquire、permit_orphaned_when_poisoned、pool_poison_idle_slot等验证了这些语义。桥接协议一整套 JSON 消息类型bridge/protocol.rs 定义了父子通信的全部消息类型全部 JSON 序列化并使用{type: ...}判别字段serde tagsnake_case。以下是 README 中的完整清单控制通道stdin/stdoutControlRequest父 → 子Init、Cancel、Shutdown源码中还有HealthcheckControlResponse子 → 父Ready、Log、Idle、Failed、Cancelled、ShuttingDown源码中还有WorkerLog、Fatal、DroppedLogs、HealthcheckResult。槽位通道Unix socketSlotRequest父 → 子Predict携带id、input或input_file、output_dir、contextSlotResponse子 → 父Log、Output、Done、Failed、Cancelled源码中为LogLine、OutputChunk、Metric、Done、Failed、Cancelled、ProtocolVersion、FileOutput。README 给出的两个通道消息示例// Control Channel {type: init, predictor_ref: predict.py:Predictor, num_slots: 2, ...} {type: cancel, slot: uuid} {type: shutdown} {type: ready, slots: [uuid1, uuid2], schema: {...}} {type: log, source: stdout, data: Loading model...} {type: idle, slot: uuid} {type: failed, slot: uuid, error: Setup failed: ...} {type: shutting_down} // Slot Sockets { type: predict, id: pred_123, input: { prompt: Hello } } {type: log, source: stdout, data: Processing...} {type: output, output: chunk} {type: done, id: pred_123, output: Hello, world!, predict_time: 0.5} {type: failed, id: pred_123, error: ValueError: ...} {type: cancelled, id: pred_123}几个增强细节SlotId使用UUID v4而非数组索引避免混淆与意外复用SlotId文档注释明确说明这一点Done消息携带predict_time与is_stream信号predictor 返回 list/generator/iterator 时为 true作为 schema 缺失时的流式输出兜底Metric消息支持Replace/Increment/Append三种合并模式SLOT_RESPONSE_PROTOCOL_VERSION 1作为未来协议演进的显式标记。HTTP 传输层与预测流程HTTP 层由 transport/http/routes.rs 实现核心端点包括GET /—— 服务发现根端点返回cog_version、docs_url、openapi_url、predictions_url、predictions_idempotent_url、predictions_cancel_url等支持训练时额外追加/trainings相关端点GET /healthhealth_check—— 返回status含UNHEALTHY响应态、setupSetupResult、versionPOST /predictionscreate_prediction及其幂等变体POST/PUT /predictions/{prediction_id}create_prediction_idempotent/create_prediction_with_idDELETE /predictions/{id}或PUT /predictions/{id}/cancelcancel_prediction。PredictionRequest支持id可选幂等 ID、input、contextdict[str, str]通过current_scope().context提供给预测器、webhookURL 与webhook_events_filter。Webhook 发送器webhook.rs的默认配置为非终止更新节流500ms可用环境变量COG_THROTTLE_RESPONSE_INTERVAL覆盖、终止 webhook 最多重试12 次、退避基数100ms、重试状态码429/500/502/503/504并支持WEBHOOK_AUTH_TOKENBearer 认证与 W3C Trace Context 透传。结合 crates/README.md 的预测流程图一次完整预测的链路为HTTP 请求 → 父进程POST /predictions→ 获取槽位许可并注册预测 → 经槽位 socket 下发SlotRequest::Predict→ worker 设置 ContextVar、调用predict()→ 流式回传Log/Output→ 最终回传Done {id, output, predict_time}→ 父进程更新预测状态、释放许可、发送 webhook → 返回200 OK。启动序列则是 HTTP 服务先行启动健康检查返回 STARTING编排器异步完成建 socket → spawn worker → 发 Init → 等 Ready → 填充 PermitPool → 启动事件循环 → 置 READY。行为语义健康状态、预测状态、取消与关闭健康状态机Unknown ──▶ Starting ──┬──▶ Ready ◀──▶ Busy │ └──▶ SetupFailed ──▶ Defunct各状态语义health.rs 中Health枚举定义Unknown初始状态健康检查返回 body 中的状态Startingsetup()进行中Ready可接受预测BusyREADY 但所有槽位都在使用中新预测返回 HTTP 409SetupFailedsetup()抛异常Defunct不可恢复错误。HealthResponse额外包含瞬态Unhealthy用户自定义健康检查失败不存储为内部状态SetupResult记录started_at、completed_at、statusstarting/succeeded/failed与捕获的logs。对应测试可见 integration-tests/tests/healthcheck*.txtar 系列。预测状态机Starting ──▶ Processing ──┬──▶ Succeeded ├──▶ Failed └──▶ CanceledPredictionStatusprediction.rs提供is_terminal()判定Succeeded/Failed/Canceled 为终态并围绕Prediction实现了流式事件广播start/output/log/metric/completed五类事件、事件回放历史容量默认 1024可用环境变量COG_STREAM_HISTORY_CAPACITY调整设为 0 可禁用回放以及用户指标支持点路径键如timing.preprocess与三种合并模式。预测状态快照build_state_snapshot是 webhook 载荷、GET 响应与终止响应的统一数据源终止状态时自动合并predict_time指标。取消链路README 给出了完整的取消流程调用HTTP DELETE /predictions/{id}或PUT /predictions/{id}/cancel父进程发送ControlRequest::Cancel { slot }worker 调用handler.cancel(slot)同步预测SIGUSR1 在 Python 侧触发KeyboardInterrupt异步预测对 asyncio 任务调用future.cancel()预测以SlotResponse::Cancelled返回。源码侧service.cancel(id)会同时触发CancellationToken供 Rust 侧观察者如上传任务使用并委托编排器发送取消crates/coglet-python/src/cancel.rs实现了同步预测的 SIGUSR1 取消支持。集成测试覆盖见 integration-tests/tests/cancel_async_prediction.txtar、cancel_sync_prediction.txtar 与 cancel_repeated.txtar。关闭路径优雅关闭SIGTERM await_explicit_shutdown停止接受新预测等待在途预测完成发送ControlRequest::Shutdownworker 响应ShuttingDown后退出父进程退出。立即关闭SIGTERM 未携带该标志发送ControlRequest::Shutdown取消在途预测退出。worker 崩溃控制通道关闭事件循环检测到失败所有在途预测健康状态转为 Defunct。槽位毒化若某个槽位 socket 出错写失败等该槽位被标记为毒化不再接收新预测若所有槽位均被毒化worker 退出。源码用SlotOutcome枚举在类型层面保证毒化槽位只能产生 Failed、不能产生 Idleenum SlotOutcome { Idle(SlotId), // Ready for next prediction Poisoned { slot, error }, // Slot is dead }关键设计决策回顾结合 crates/README.md 的总结coglet 的五个关键设计决策构成了其整体风格子进程隔离worker 独立进程运行换取崩溃隔离、内存隔离与干净关闭单 worker 模式始终恰好一个 worker 子进程不做动态扩缩容——父进程轻量重活全在 worker 中槽位并发每个槽位一对 Unix socketmax_concurrency决定槽位数许可制保证单槽单预测ContextVar 日志路由异步预测可能 spawn 子任务ContextVar 沿调用栈传播预测 ID即使从派生任务也能正确路由日志Audit Hook 保护用户代码可能替换sys.stdoutaudit hook 拦截后用TeeWriter包装其流既保留日志路由又让用户代码按预期工作。结语与深入阅读coglet 以纯 Rust 核心 PyO3 薄绑定的分层方式把 Cog 预测服务器的进程编排、并发控制、IPC 协议与 HTTP 服务全部沉淀为可独立测试的 Rust 代码是理解 Cog 子进程隔离模型的必读入口。想进一步深入可以按以下路径继续crates/coglet/README.md 与 crates/README.md架构总览与本组件文档crates/coglet/src/service.rs、orchestrator.rs、worker.rs三大核心组件实现crates/coglet/src/bridge/protocol.rs、permit/pool.rs桥接协议与并发控制细节含丰富的单元测试crates/coglet-python/src/lib.rs 与 crates/coglet-python/README.mdPyO3 绑定与 Python 侧 worker 桥integration-tests/tests以coglet_*.txtar、healthcheck*.txtar、cancel_*.txtar为代表的端到端行为验证用例。【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 coglet 深度解析：Cog 机器学习预测服务器的纯 Rust 核心运行时
coglet 深度解析Cog 机器学习预测服务器的纯 Rust 核心运行时【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cogcoglet 是 Cog 项目Containers for machine learning中负责预测服务执行的 Rust 核心库它以纯 Rust、零 Python 依赖的方式实现了子进程隔离、并发槽位管理与高性能 IPC是 Cog 预测服务器coglet-python 的底层引擎的心脏。阅读本文后你将完整掌握 coglet 的父/子进程双通道架构、PredictionService/Orchestrator/Worker三大核心组件的工作原理、PermitPool槽位并发控制与桥接协议细节以及健康状态机、取消与优雅关闭的完整行为语义。coglet 在 Cog 体系中的定位coglet 的核心定位写在其自身文档首行它是 coglet 预测服务器的核心 Rust 库Core Rust library for the coglet prediction server纯 Rust 实现、不依赖任何 Python 库Python 绑定则独立存在于 coglet-python 中。这种分层意味着预测服务器的全部核心逻辑进程编排、并发控制、IPC 协议、HTTP 服务都可以用 Rust 独立测试与演进Python 侧只保留薄薄的 PyO3 绑定层通过 lib.rs 暴露serve()、active()、_run_worker()等入口整个 crates 工作区见 crates/README.md由coglet核心库与coglet-pythonPyO3 绑定两个 crate 组成统一由 Cargo.toml 工作区清单管理。从整体架构上看coglet 实现了 Cog 的子进程隔离模型HTTP 请求进入父进程父进程把预测任务下发给子进程worker子进程内运行 Python 预测器load()/setup()/predict()。这种模型带来三个核心收益崩溃隔离worker 崩溃可重启而父进程存活、内存隔离GPU 内存泄漏不会累积、以及可按需 SIGKILL 的干净关闭能力。总体架构父进程编排 双通道 IPCcoglet 的核心架构可以浓缩为下面这张取自 crates/coglet/README.md 的模块图coglet ┌─────────────────────────────────────────────────────────────────┐ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ transport/http │ │ │ │ ┌──────────────┐ ┌─────────────────────────────────┐ │ │ │ │ │ server.rs │ │ routes.rs │ │ │ │ │ │ Axum setup │ │ /health, /predictions, /cancel │ │ │ │ │ └──────────────┘ └─────────────────────────────────┘ │ │ │ └───────────────────────────────┬─────────────────────────┘ │ │ │ │ │ ┌───────────────────────────────▼─────────────────────────┐ │ │ │ service.rs │ │ │ │ PredictionService: health, permits, state, webhooks │ │ │ └───────────────────────────────┬─────────────────────────┘ │ │ │ │ │ ┌────────────────────────┼────────────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────┐ ┌────────────────────┐ ┌──────────┐ │ │ │ permit/ │ │ orchestrator.rs │ │webhook.rs│ │ │ │ PermitPool │ │ Parent-side: │ │ Sender │ │ │ │ Slot alloc │ │ spawn, route │ │ Retry │ │ │ └─────────────┘ └─────────┬──────────┘ └──────────┘ │ │ │ │ │ ┌────────────────────────────▼────────────────────────────┐ │ │ │ bridge/ │ │ │ │ ┌──────────────┐ ┌─────────────┐ ┌────────────────┐ │ │ │ │ │ protocol.rs │ │ codec.rs │ │ transport.rs │ │ │ │ │ │ Message types│ │ JSON lines │ │ Unix sockets │ │ │ │ │ └──────────────┘ └─────────────┘ └────────────────┘ │ │ │ └─────────────────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────────────────┐ │ │ │ worker.rs │ │ │ │ Child-side: PredictHandler trait, run_worker loop │ │ │ └─────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘各层职责清晰可辨层次模块职责HTTP 传输层transport/http/server.rs、routes.rsAxum 服务装配与路由处理/health、/predictions、/cancel等服务层service.rsPredictionService健康状态、许可池、预测状态、webhook 的统一所有者编排层orchestrator.rs父进程侧spawn 子进程、事件循环、消息路由并发控制permit/pool.rsPermitPool槽位许可管理与分配Webhookwebhook.rsWebhookSender节流、重试IPC 桥bridge/protocol.rs、codec.rs、transport.rs消息类型、JSON lines 编解码、Unix socket 传输子进程侧worker.rsPredictHandlertrait、run_worker事件循环双通道 IPC控制通道与槽位套接字父进程与 worker 之间并非单一通道而是精心设计的两套通道对应 protocol.rs 的注释说明控制通道stdin/stdoutJSON lines承载生命周期消息——Init、Cancel、Shutdown、Healthcheck父 → 子以及Ready、Idle、Failed、Fatal、ShuttingDown子 → 父。一行一条 JSON 消息。槽位套接字Unix domain socket每个 slot 一条承载预测数据——SlotRequest::Predict下发放LogLine、OutputChunk、Metric、Done、Failed、Cancelled回传。每个槽位独立 socket避免头端阻塞head-of-line blocking。从源码实现看slot socket 在父进程侧通过 NamedSocketTransport::create 创建路径格式为{temp_dir}/coglet-{pid}/slot-{n}.sock在 Linux 上还支持抽象命名空间AbstractSocketTransport无文件系统残留、自动清理。worker 子进程侧则通过connect反连回这些 socket随后父进程调用accept_connections完成握手。目录结构一个 crate一套清晰分层目录树 展示了源码组织的完整脉络以下为节选并标注核心注释coglet/ └── src/ ├── lib.rs # Public API exports │ │ # Core Types ├── health.rs # Health, SetupStatus, SetupResult ├── prediction.rs # Prediction state machine ├── predictor.rs # PredictionResult, PredictionError, PredictionOutput ├── version.rs # VersionInfo │ │ # Service Layer ├── service.rs # PredictionService - lifecycle, state, webhooks ├── webhook.rs # WebhookSender, webhook types │ │ # Orchestrator (Parent Process) ├── orchestrator.rs # spawn_worker, OrchestratorHandle, event loop │ │ # Worker (Child Process) ├── worker.rs # run_worker, PredictHandler trait, SetupError │ │ # Concurrency Control ├── permit/ │ ├── pool.rs # PermitPool - slot permit management │ └── slot.rs # PredictionSlot - permit prediction binding │ │ # IPC Bridge ├── bridge/ │ ├── protocol.rs # ControlRequest, ControlResponse, SlotRequest, SlotResponse │ ├── codec.rs # JsonCodec - newline-delimited JSON │ └── transport.rs # Unix socket transport, ChildTransportInfo │ │ # HTTP Transport └── transport/ └── http/ ├── server.rs # ServerConfig, serve() └── routes.rs # Route handlers, request/response types公共 API 导出集中在 lib.rs包括Health、SetupResult、Prediction、PredictionStatus、PredictionResult、PredictionService、Orchestrator、PermitPool、InputValidator、run_worker等并提供了install_crypto_provider()用于一次性安装 rustls 的 ring TLS 加密提供程序必须在任何reqwest::Client创建前调用可重复调用。三大核心组件深入PredictionService预测状态的唯一所有者service.rs 中的PredictionService是与传输层无关的预测生命周期管理服务它统一管理健康状态Unknown → Starting → Ready/SetupFailedPermitPool Orchestrator 引用两者通过OrchestratorState原子地一起设置保证池与编排器总是同时就绪活动预测DashMapString, PredictionEntry作为预测状态的单一事实来源single source of truth取消CancellationToken 编排器委托Webhook从Prediction的 mutation 方法set_processing、set_succeeded等触发不维护双份状态。典型的组装方式取自 README 示例let service PredictionService::new_no_pool() .with_health(Health::Starting) .with_version(version); // Later, after worker is ready: service.set_orchestrator(pool, handle).await; service.set_health(Health::Ready).await;几个值得注意的源码细节set_health(Health::Ready)在没有编排器时会静默忽略源码中有明确 warn 日志保证 READY 必须先有编排器HealthSnapshot提供is_ready()与is_busy()READY 但可用槽位为 0判定strip_and_validate_input()在单次锁获取内完成剔除未知字段 校验 为省略的可选无默认值字段注入 null三步strip_validate_inject保证 predict 与 train 两条路径的排序不变量完全一致输入超过MAX_INLINE_IPC_SIZE6 MiB见 protocol.rs时build_slot_request会把输入溢出写盘到/tmp/coglet/predictions/{id}/inputs/spill_*.jsonworker 端rehydrate_input读盘、反序列化后立即删除该文件同步预测使用SyncPredictionGuardHTTP 连接断开时 axum 丢弃响应 future从而触发 guard 的 drop进而调用service.cancel(id)同时触发 CancellationToken 与编排器取消disarm()可在正常结束时解除武装。Orchestrator父进程侧的 worker 生命周期管理orchestrator.rs 负责 spawn 子进程并维持其生命周期README 给出了完整的启动流程spawn_worker(config) │ ├─▶ Create Unix socket transport (N slots) ├─▶ Spawn: python -c import coglet; coglet.server._run_worker() ├─▶ Send Init message via stdin ├─▶ Wait for worker to connect sockets ├─▶ Wait for Ready message (with timeout) ├─▶ Populate PermitPool with slot writers ├─▶ Spawn event loop task └─▶ Return OrchestratorReady {pool, schema, handle}事件循环统一处理来自 worker 的各种响应ControlResponse::Idle—— 槽位可接收下一个预测ControlResponse::Failed—— 槽位被毒化poisoned标记不可用SlotResponse::Log/Output/Done/Failed—— 路由到对应预测worker 崩溃—— 失败所有进行中的预测。补充说明事件循环还处理ControlResponse::Fatalworker 不可恢复错误父进程应毒化所有槽位并失败所有在途预测、DroppedLogs背压丢弃日志的系统诊断以及HealthcheckResult用户自定义健康检查结果。此外upload_file实现了与 Python cog 的put_file_to_signed_endpoint一致的签名上传逻辑PUT Content-Type、跟随重定向、从 Location 头取最终 URL 并剥离查询参数。Worker子进程侧的事件循环worker.rs 是子进程侧实现核心是PredictHandlertrait 与run_worker循环run_worker(handler, config) │ ├─▶ Connect to slot sockets (from env) ├─▶ Setup control channel (stdin/stdout) ├─▶ Run handler.setup() with log routing ├─▶ Send Ready {slots, schema} ├─▶ Enter event loop: │ - ControlRequest::Cancel → handler.cancel(slot) │ - ControlRequest::Shutdown → exit │ - SlotRequest::Predict → spawn prediction task └─▶ Exit on shutdown or all slots poisoned两个源码级防护机制值得注意panic hook 即致命错误通道worker 安装全局 panic hookinstall_panic_hook任何 panic 都会尽力发送ControlResponse::Fatal { reason }给父进程然后std::process::abort()。这意味着任意调用点的panic!/.expect()都会自动获得正确的致命行为无需额外辅助代码日志截断worker 日志经truncate_worker_log在 4 MiB 处按字符边界截断并追加[**** LOG LINE TRUNCATED AT 4 MiB ****]标记protocol.rs 中的truncate_worker_log测试覆盖了长/短日志与多字节 UTF-8 场景避免超大日志行引发 panic 或撑爆通道。worker 进程内部的 Python 侧结构见 crates/README.md包含PythonPredictorload()/setup()/predict()、SlotLogWriter基于 ContextVar 的 stdout/stderr 路由与Audit Hook保护运行流、对用户覆写采用 Tee 模式这些由 coglet-python/src 下的log_writer.rs、audit.rs、cancel.rs等实现。PermitPool基于槽位的并发控制permit/pool.rs 实现了槽位 许可的并发控制模型max_concurrency决定槽位数量每个槽位同一时刻最多运行一个预测。README 给出的核心用法let pool PermitPool::new(max_concurrency); // Add slot with its socket writer pool.add_permit(slot_id, writer); // Acquire permit (returns None if at capacity) let permit pool.try_acquire()?; // Send prediction request permit.send(SlotRequest::Predict { id, input }).await?; // Return permit when done drop(permit);源码实现采用了typestate类型状态模式保证编译期状态转换安全许可permit有三种状态类型状态行为PermitInUse正在运行预测into_idle()转为空闲drop 时归还池into_poisoned()转为毒化永久不归还PermitIdle完成后 drop 自动归还池除非池级 poison 标志已置位PermitPoisoned永久失败drop 时仅告警容量缩减关键设计是毒化poisoning是池级属性pool.poison(slot_id)无论槽位是空闲在池中还是在用被预测持有都会置位共享的AtomicBool标志try_acquire会跳过已毒化许可PermitIdle::drop看到标志后也不再归还。配套的SlotIdleToken机制确保只有 worker 确认槽位空闲后许可才归还——若 5 秒内未被消费会打印告警ALERT_THRESHOLD提示槽位可能无法回归池中。相关单元测试pool_add_and_acquire、permit_orphaned_when_poisoned、pool_poison_idle_slot等验证了这些语义。桥接协议一整套 JSON 消息类型bridge/protocol.rs 定义了父子通信的全部消息类型全部 JSON 序列化并使用{type: ...}判别字段serde tagsnake_case。以下是 README 中的完整清单控制通道stdin/stdoutControlRequest父 → 子Init、Cancel、Shutdown源码中还有HealthcheckControlResponse子 → 父Ready、Log、Idle、Failed、Cancelled、ShuttingDown源码中还有WorkerLog、Fatal、DroppedLogs、HealthcheckResult。槽位通道Unix socketSlotRequest父 → 子Predict携带id、input或input_file、output_dir、contextSlotResponse子 → 父Log、Output、Done、Failed、Cancelled源码中为LogLine、OutputChunk、Metric、Done、Failed、Cancelled、ProtocolVersion、FileOutput。README 给出的两个通道消息示例// Control Channel {type: init, predictor_ref: predict.py:Predictor, num_slots: 2, ...} {type: cancel, slot: uuid} {type: shutdown} {type: ready, slots: [uuid1, uuid2], schema: {...}} {type: log, source: stdout, data: Loading model...} {type: idle, slot: uuid} {type: failed, slot: uuid, error: Setup failed: ...} {type: shutting_down} // Slot Sockets { type: predict, id: pred_123, input: { prompt: Hello } } {type: log, source: stdout, data: Processing...} {type: output, output: chunk} {type: done, id: pred_123, output: Hello, world!, predict_time: 0.5} {type: failed, id: pred_123, error: ValueError: ...} {type: cancelled, id: pred_123}几个增强细节SlotId使用UUID v4而非数组索引避免混淆与意外复用SlotId文档注释明确说明这一点Done消息携带predict_time与is_stream信号predictor 返回 list/generator/iterator 时为 true作为 schema 缺失时的流式输出兜底Metric消息支持Replace/Increment/Append三种合并模式SLOT_RESPONSE_PROTOCOL_VERSION 1作为未来协议演进的显式标记。HTTP 传输层与预测流程HTTP 层由 transport/http/routes.rs 实现核心端点包括GET /—— 服务发现根端点返回cog_version、docs_url、openapi_url、predictions_url、predictions_idempotent_url、predictions_cancel_url等支持训练时额外追加/trainings相关端点GET /healthhealth_check—— 返回status含UNHEALTHY响应态、setupSetupResult、versionPOST /predictionscreate_prediction及其幂等变体POST/PUT /predictions/{prediction_id}create_prediction_idempotent/create_prediction_with_idDELETE /predictions/{id}或PUT /predictions/{id}/cancelcancel_prediction。PredictionRequest支持id可选幂等 ID、input、contextdict[str, str]通过current_scope().context提供给预测器、webhookURL 与webhook_events_filter。Webhook 发送器webhook.rs的默认配置为非终止更新节流500ms可用环境变量COG_THROTTLE_RESPONSE_INTERVAL覆盖、终止 webhook 最多重试12 次、退避基数100ms、重试状态码429/500/502/503/504并支持WEBHOOK_AUTH_TOKENBearer 认证与 W3C Trace Context 透传。结合 crates/README.md 的预测流程图一次完整预测的链路为HTTP 请求 → 父进程POST /predictions→ 获取槽位许可并注册预测 → 经槽位 socket 下发SlotRequest::Predict→ worker 设置 ContextVar、调用predict()→ 流式回传Log/Output→ 最终回传Done {id, output, predict_time}→ 父进程更新预测状态、释放许可、发送 webhook → 返回200 OK。启动序列则是 HTTP 服务先行启动健康检查返回 STARTING编排器异步完成建 socket → spawn worker → 发 Init → 等 Ready → 填充 PermitPool → 启动事件循环 → 置 READY。行为语义健康状态、预测状态、取消与关闭健康状态机Unknown ──▶ Starting ──┬──▶ Ready ◀──▶ Busy │ └──▶ SetupFailed ──▶ Defunct各状态语义health.rs 中Health枚举定义Unknown初始状态健康检查返回 body 中的状态Startingsetup()进行中Ready可接受预测BusyREADY 但所有槽位都在使用中新预测返回 HTTP 409SetupFailedsetup()抛异常Defunct不可恢复错误。HealthResponse额外包含瞬态Unhealthy用户自定义健康检查失败不存储为内部状态SetupResult记录started_at、completed_at、statusstarting/succeeded/failed与捕获的logs。对应测试可见 integration-tests/tests/healthcheck*.txtar 系列。预测状态机Starting ──▶ Processing ──┬──▶ Succeeded ├──▶ Failed └──▶ CanceledPredictionStatusprediction.rs提供is_terminal()判定Succeeded/Failed/Canceled 为终态并围绕Prediction实现了流式事件广播start/output/log/metric/completed五类事件、事件回放历史容量默认 1024可用环境变量COG_STREAM_HISTORY_CAPACITY调整设为 0 可禁用回放以及用户指标支持点路径键如timing.preprocess与三种合并模式。预测状态快照build_state_snapshot是 webhook 载荷、GET 响应与终止响应的统一数据源终止状态时自动合并predict_time指标。取消链路README 给出了完整的取消流程调用HTTP DELETE /predictions/{id}或PUT /predictions/{id}/cancel父进程发送ControlRequest::Cancel { slot }worker 调用handler.cancel(slot)同步预测SIGUSR1 在 Python 侧触发KeyboardInterrupt异步预测对 asyncio 任务调用future.cancel()预测以SlotResponse::Cancelled返回。源码侧service.cancel(id)会同时触发CancellationToken供 Rust 侧观察者如上传任务使用并委托编排器发送取消crates/coglet-python/src/cancel.rs实现了同步预测的 SIGUSR1 取消支持。集成测试覆盖见 integration-tests/tests/cancel_async_prediction.txtar、cancel_sync_prediction.txtar 与 cancel_repeated.txtar。关闭路径优雅关闭SIGTERM await_explicit_shutdown停止接受新预测等待在途预测完成发送ControlRequest::Shutdownworker 响应ShuttingDown后退出父进程退出。立即关闭SIGTERM 未携带该标志发送ControlRequest::Shutdown取消在途预测退出。worker 崩溃控制通道关闭事件循环检测到失败所有在途预测健康状态转为 Defunct。槽位毒化若某个槽位 socket 出错写失败等该槽位被标记为毒化不再接收新预测若所有槽位均被毒化worker 退出。源码用SlotOutcome枚举在类型层面保证毒化槽位只能产生 Failed、不能产生 Idleenum SlotOutcome { Idle(SlotId), // Ready for next prediction Poisoned { slot, error }, // Slot is dead }关键设计决策回顾结合 crates/README.md 的总结coglet 的五个关键设计决策构成了其整体风格子进程隔离worker 独立进程运行换取崩溃隔离、内存隔离与干净关闭单 worker 模式始终恰好一个 worker 子进程不做动态扩缩容——父进程轻量重活全在 worker 中槽位并发每个槽位一对 Unix socketmax_concurrency决定槽位数许可制保证单槽单预测ContextVar 日志路由异步预测可能 spawn 子任务ContextVar 沿调用栈传播预测 ID即使从派生任务也能正确路由日志Audit Hook 保护用户代码可能替换sys.stdoutaudit hook 拦截后用TeeWriter包装其流既保留日志路由又让用户代码按预期工作。结语与深入阅读coglet 以纯 Rust 核心 PyO3 薄绑定的分层方式把 Cog 预测服务器的进程编排、并发控制、IPC 协议与 HTTP 服务全部沉淀为可独立测试的 Rust 代码是理解 Cog 子进程隔离模型的必读入口。想进一步深入可以按以下路径继续crates/coglet/README.md 与 crates/README.md架构总览与本组件文档crates/coglet/src/service.rs、orchestrator.rs、worker.rs三大核心组件实现crates/coglet/src/bridge/protocol.rs、permit/pool.rs桥接协议与并发控制细节含丰富的单元测试crates/coglet-python/src/lib.rs 与 crates/coglet-python/README.mdPyO3 绑定与 Python 侧 worker 桥integration-tests/tests以coglet_*.txtar、healthcheck*.txtar、cancel_*.txtar为代表的端到端行为验证用例。【免费下载链接】cogContainers for machine learning项目地址: https://gitcode.com/GitHub_Trending/co/cog创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考