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::{Json, Router, routing::get};
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    pub pipe: OwnedMutexGuard<Pipeline>,
31    // Keep the permit after the mutex guard so drop glue unlocks the
32    // pipeline before another waiter can acquire the permit.
33    _permit: OwnedSemaphorePermit,
34}
35
36impl PipelinePool {
37    pub fn new(pipelines: Vec<Pipeline>) -> Self {
38        assert!(
39            !pipelines.is_empty(),
40            "pipeline pool needs at least one slot"
41        );
42        let sem = Arc::new(Semaphore::new(pipelines.len()));
43        Self {
44            slots: pipelines
45                .into_iter()
46                .map(|p| Arc::new(Mutex::new(p)))
47                .collect(),
48            sem,
49        }
50    }
51
52    pub fn n_slots(&self) -> usize {
53        self.slots.len()
54    }
55
56    /// Wait for a free slot and check it out. With `permits == slots`,
57    /// holding a permit guarantees the try_lock scan finds a free slot.
58    pub async fn acquire(&self) -> SlotGuard {
59        let permit = self
60            .sem
61            .clone()
62            .acquire_owned()
63            .await
64            .expect("slot semaphore closed");
65        for s in &self.slots {
66            if let Ok(pipe) = s.clone().try_lock_owned() {
67                return SlotGuard {
68                    pipe,
69                    _permit: permit,
70                };
71            }
72        }
73        unreachable!("semaphore permit held but every slot is locked")
74    }
75}
76
77/// Shared application state: runtime (masks, metrics), a tokenizer
78/// handle that never blocks on generation, and the slot pool.
79pub struct AppState {
80    pub runtime: CortiqRuntime,
81    pub tokenizer: Arc<cortiq_engine::tokenizer::Tokenizer>,
82    pub slots: PipelinePool,
83}
84
85/// Liveness probe — returns 200 as soon as the server is accepting
86/// connections. Used by process managers that embed `cortiq serve` (e.g.
87/// a gateway spawning it as a local model server) to know when it is ready.
88async fn healthz() -> Json<serde_json::Value> {
89    Json(serde_json::json!({ "status": "ok" }))
90}
91
92/// Build the full router with all endpoints.
93pub fn build_router(state: Arc<AppState>) -> Router {
94    Router::new()
95        .route("/healthz", get(healthz))
96        .merge(openai::routes())
97        .merge(api::routes())
98        .merge(dashboard::routes())
99        .layer(CorsLayer::permissive())
100        .with_state(state)
101}
cortiq_server/lib.rs

cortiq_server/
lib.rs
