cortiq_server/
lib.rs

1//! Cortiq Server — OpenAI-compatible API + web management dashboard.
2
3pub mod api;
4pub mod dashboard;
5pub mod openai;
6pub mod streaming;
7
8use axum::{routing::get, Json, Router};
9use cortiq_engine::{CortiqRuntime, Pipeline};
10use std::sync::Arc;
11use tokio::sync::{Mutex, OwnedMutexGuard, OwnedSemaphorePermit, Semaphore};
12use tower_http::cors::CorsLayer;
13
14/// Fixed pool of pipeline slots over ONE shared mmap'd model (roadmap
15/// §3 «serving полностью сериализован», этап 5.1): the weights are
16/// zero-copy shared through `Arc<CmfModel>`, each slot owns its
17/// KV-cache / recurrent state / sampler / workspace. A request checks a
18/// slot out for the duration of one generation, so up to `slots`
19/// requests decode CONCURRENTLY; excess requests queue fairly on the
20/// semaphore. This is bounded-concurrency serving, not yet continuous
21/// batching (этап 5.2+).
22pub struct PipelinePool {
23    slots: Vec<Arc<Mutex<Pipeline>>>,
24    sem: Arc<Semaphore>,
25}
26
27/// A checked-out slot: holds both the concurrency permit and the
28/// pipeline lock until dropped.
29pub struct SlotGuard {
30    _permit: OwnedSemaphorePermit,
31    pub pipe: OwnedMutexGuard<Pipeline>,
32}
33
34impl PipelinePool {
35    pub fn new(pipelines: Vec<Pipeline>) -> Self {
36        assert!(!pipelines.is_empty(), "pipeline pool needs at least one slot");
37        let sem = Arc::new(Semaphore::new(pipelines.len()));
38        Self {
39            slots: pipelines.into_iter().map(|p| Arc::new(Mutex::new(p))).collect(),
40            sem,
41        }
42    }
43
44    pub fn n_slots(&self) -> usize {
45        self.slots.len()
46    }
47
48    /// Wait for a free slot and check it out. With `permits == slots`,
49    /// holding a permit guarantees the try_lock scan finds a free slot.
50    pub async fn acquire(&self) -> SlotGuard {
51        let permit = self
52            .sem
53            .clone()
54            .acquire_owned()
55            .await
56            .expect("slot semaphore closed");
57        for s in &self.slots {
58            if let Ok(pipe) = s.clone().try_lock_owned() {
59                return SlotGuard { _permit: permit, pipe };
60            }
61        }
62        unreachable!("semaphore permit held but every slot is locked")
63    }
64}
65
66/// Shared application state: runtime (masks, metrics), a tokenizer
67/// handle that never blocks on generation, and the slot pool.
68pub struct AppState {
69    pub runtime: CortiqRuntime,
70    pub tokenizer: Arc<cortiq_engine::tokenizer::Tokenizer>,
71    pub slots: PipelinePool,
72}
73
74/// Liveness probe — returns 200 as soon as the server is accepting
75/// connections. Used by process managers that embed `cortiq serve` (e.g.
76/// a gateway spawning it as a local model server) to know when it is ready.
77async fn healthz() -> Json<serde_json::Value> {
78    Json(serde_json::json!({ "status": "ok" }))
79}
80
81/// Build the full router with all endpoints.
82pub fn build_router(state: Arc<AppState>) -> Router {
83    Router::new()
84        .route("/healthz", get(healthz))
85        .merge(openai::routes())
86        .merge(api::routes())
87        .merge(dashboard::routes())
88        .layer(CorsLayer::permissive())
89        .with_state(state)
90}
cortiq_server/lib.rs

cortiq_server/
lib.rs
