Reserve connector capacity for every caller
This commit is contained in:
parent
be1776ab45
commit
2f29b90eeb
2 changed files with 80 additions and 14 deletions
|
|
@ -110,6 +110,10 @@ impl CodexDaemonConfig {
|
|||
{
|
||||
return Err(CodexDaemonError::InvalidConfig("runtime bounds"));
|
||||
}
|
||||
let required_connections = required_connection_capacity(&self.callers)?;
|
||||
if self.max_connections < required_connections {
|
||||
return Err(CodexDaemonError::InvalidConfig("connection capacity"));
|
||||
}
|
||||
let mut caller_ids = HashSet::new();
|
||||
for caller in &self.callers {
|
||||
if caller.caller_runtime_id.trim().is_empty()
|
||||
|
|
@ -132,6 +136,37 @@ impl CodexDaemonConfig {
|
|||
}
|
||||
Ok(bind)
|
||||
}
|
||||
|
||||
pub fn admit_caller(&mut self, caller: CodexCallerConfig) -> Result<(), CodexDaemonError> {
|
||||
let mut candidate = self.clone();
|
||||
if candidate
|
||||
.callers
|
||||
.iter()
|
||||
.any(|existing| existing.caller_runtime_id == caller.caller_runtime_id)
|
||||
{
|
||||
return Err(CodexDaemonError::InvalidConfig("duplicate caller"));
|
||||
}
|
||||
candidate.max_frame_bytes = candidate.max_frame_bytes.max(
|
||||
caller
|
||||
.max_payload_bytes
|
||||
.saturating_add(FRAME_OVERHEAD_BUDGET),
|
||||
);
|
||||
candidate.callers.push(caller);
|
||||
candidate.max_connections = required_connection_capacity(&candidate.callers)?;
|
||||
candidate.validate()?;
|
||||
*self = candidate;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn required_connection_capacity(callers: &[CodexCallerConfig]) -> Result<usize, CodexDaemonError> {
|
||||
callers
|
||||
.iter()
|
||||
.try_fold(0_usize, |total, caller| {
|
||||
total.checked_add(caller.max_concurrent_requests)
|
||||
})
|
||||
.map(|total| total.max(8))
|
||||
.ok_or(CodexDaemonError::InvalidConfig("connection capacity"))
|
||||
}
|
||||
|
||||
pub fn load_daemon_config(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonError> {
|
||||
|
|
@ -494,6 +529,48 @@ mod tests {
|
|||
config.validate().unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn caller_admission_reserves_every_independently_admitted_request_slot() {
|
||||
let root = Path::new("/srv/codex-connector");
|
||||
let first = CodexCallerConfig {
|
||||
caller_runtime_id: "ghostlight-dungeon-yggdrasil".to_string(),
|
||||
connection_key_file: root.join("ghostlight.key"),
|
||||
connection_key_epoch: 1,
|
||||
allowed_models: vec!["gpt-5.6-luna".to_string()],
|
||||
max_concurrent_requests: 8,
|
||||
max_payload_bytes: 1_048_576,
|
||||
max_output_tokens: 16_384,
|
||||
};
|
||||
let mut config = CodexDaemonConfig::single_caller(
|
||||
"127.0.0.1:4103",
|
||||
root.join("codex"),
|
||||
[7; 32],
|
||||
root.join("codex-home"),
|
||||
root.join("replay.cc"),
|
||||
first,
|
||||
);
|
||||
config
|
||||
.admit_caller(CodexCallerConfig {
|
||||
caller_runtime_id: "epiphany-model-runtime".to_string(),
|
||||
connection_key_file: root.join("epiphany.key"),
|
||||
connection_key_epoch: 1,
|
||||
allowed_models: vec!["gpt-5.6-luna".to_string()],
|
||||
max_concurrent_requests: 8,
|
||||
max_payload_bytes: 1_048_576,
|
||||
max_output_tokens: 16_384,
|
||||
})
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(config.max_connections, 16);
|
||||
config.validate().unwrap();
|
||||
|
||||
config.max_connections = 15;
|
||||
assert!(matches!(
|
||||
config.validate(),
|
||||
Err(CodexDaemonError::InvalidConfig("connection capacity"))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() {
|
||||
let mut encoded = Vec::new();
|
||||
|
|
|
|||
17
src/main.rs
17
src/main.rs
|
|
@ -82,13 +82,6 @@ fn admit_caller_config(args: &[OsString]) -> Result<(), String> {
|
|||
let mut config = codex_connector::load_daemon_config(&PathBuf::from(source))
|
||||
.map_err(|error| error.to_string())?;
|
||||
let caller_runtime_id = os_text(caller_runtime_id, "caller runtime")?;
|
||||
if config
|
||||
.callers
|
||||
.iter()
|
||||
.any(|caller| caller.caller_runtime_id == caller_runtime_id)
|
||||
{
|
||||
return Err(format!("caller {caller_runtime_id} is already admitted"));
|
||||
}
|
||||
let caller = CodexCallerConfig {
|
||||
caller_runtime_id,
|
||||
connection_key_file: PathBuf::from(connection_key_file),
|
||||
|
|
@ -98,13 +91,9 @@ fn admit_caller_config(args: &[OsString]) -> Result<(), String> {
|
|||
max_payload_bytes: parse_number(max_payload_bytes, "max payload bytes")?,
|
||||
max_output_tokens: parse_number(max_output_tokens, "max output tokens")?,
|
||||
};
|
||||
config.max_frame_bytes = config
|
||||
.max_frame_bytes
|
||||
.max(caller.max_payload_bytes.saturating_add(4096));
|
||||
config.max_connections = config
|
||||
.max_connections
|
||||
.max(caller.max_concurrent_requests.max(8));
|
||||
config.callers.push(caller);
|
||||
config
|
||||
.admit_caller(caller)
|
||||
.map_err(|error| error.to_string())?;
|
||||
codex_connector::write_daemon_config(&PathBuf::from(destination), &config)
|
||||
.map_err(|error| error.to_string())
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue