| 1 | use std::{ |
| 2 | collections::BTreeMap, |
| 3 | fs, io, |
| 4 | os::unix::net::{UnixListener, UnixStream}, |
| 5 | path::PathBuf, |
| 6 | process, |
| 7 | sync::{Arc, Mutex, MutexGuard}, |
| 8 | thread, |
| 9 | time::{Duration, Instant}, |
| 10 | }; |
| 11 | |
| 12 | use crate::{ |
| 13 | client, |
| 14 | config::Config, |
| 15 | paths, |
| 16 | proto::{self, Acquire, Event, LeaseId, LeaseInfo, Reader, Request, Status}, |
| 17 | sched::{self, Capacity, Disk, FsId, Need, Queued, Reason, Running, Verdict}, |
| 18 | sys, |
| 19 | units::Bytes, |
| 20 | }; |
| 21 | |
| 22 | struct Run { |
| 23 | started: Instant, |
| 24 | rss: Bytes, |
| 25 | reminded: Option<Instant>, |
| 26 | } |
| 27 | |
| 28 | struct Lease { |
| 29 | req: Acquire, |
| 30 | heavy: bool, |
| 31 | fs: Vec<FsId>, |
| 32 | queued_at: Instant, |
| 33 | run: Option<Run>, |
| 34 | out: UnixStream, |
| 35 | waiting: Option<Reason>, |
| 36 | last_pos: usize, |
| 37 | watchers: Vec<UnixStream>, |
| 38 | } |
| 39 | |
| 40 | impl Lease { |
| 41 | fn elapsed(&self) -> Duration { |
| 42 | self.run |
| 43 | .as_ref() |
| 44 | .map_or(self.queued_at, |r| r.started) |
| 45 | .elapsed() |
| 46 | } |
| 47 | |
| 48 | fn overdue(&self, grace: f64) -> bool { |
| 49 | self.run.is_some() && self.elapsed() > self.req.need.time.mul_f64(grace) |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | struct State { |
| 54 | cfg: Config, |
| 55 | next: LeaseId, |
| 56 | leases: BTreeMap<LeaseId, Lease>, |
| 57 | cap: Capacity, |
| 58 | baselines: BTreeMap<FsId, Bytes>, |
| 59 | conns: usize, |
| 60 | idle_since: Instant, |
| 61 | } |
| 62 | |
| 63 | type Shared = Arc<Mutex<State>>; |
| 64 | |
| 65 | fn lock(shared: &Shared) -> MutexGuard<'_, State> { |
| 66 | shared |
| 67 | .lock() |
| 68 | .unwrap_or_else(std::sync::PoisonError::into_inner) |
| 69 | } |
| 70 | |
| 71 | impl State { |
| 72 | fn new(cfg: Config) -> Self { |
| 73 | let mut st = Self { |
| 74 | cfg, |
| 75 | next: 1, |
| 76 | leases: BTreeMap::new(), |
| 77 | cap: Capacity { |
| 78 | ram_budget: Bytes::ZERO, |
| 79 | ram_live: Bytes::ZERO, |
| 80 | ram_reserve: Bytes::ZERO, |
| 81 | cores: 1, |
| 82 | disks: BTreeMap::new(), |
| 83 | pressure: false, |
| 84 | }, |
| 85 | baselines: BTreeMap::new(), |
| 86 | conns: 0, |
| 87 | idle_since: Instant::now(), |
| 88 | }; |
| 89 | st.apply_budget(); |
| 90 | st |
| 91 | } |
| 92 | |
| 93 | fn apply_budget(&mut self) { |
| 94 | let (total, live) = sys::memory(); |
| 95 | self.cap.ram_budget = self.cfg.budget.ram.resolve(total); |
| 96 | self.cap.ram_live = live; |
| 97 | self.cap.ram_reserve = self.cfg.budget.ram_reserve; |
| 98 | self.cap.cores = self.cfg.budget.cores.resolve(sys::cores()); |
| 99 | } |
| 100 | |
| 101 | /// running leases keep the heaviness they started with, so a reload can't |
| 102 | /// retroactively put two heavies side by side |
| 103 | fn reload(&mut self, cfg: Config) { |
| 104 | self.cfg = cfg; |
| 105 | self.apply_budget(); |
| 106 | let queued: Vec<(LeaseId, Need)> = self |
| 107 | .leases |
| 108 | .iter() |
| 109 | .filter(|(_, l)| l.run.is_none()) |
| 110 | .map(|(id, l)| (*id, l.req.need)) |
| 111 | .collect(); |
| 112 | for (id, need) in queued { |
| 113 | let heavy = self.heavy(&need); |
| 114 | if let Some(l) = self.leases.get_mut(&id) { |
| 115 | l.heavy = heavy; |
| 116 | } |
| 117 | } |
| 118 | self.refresh(); |
| 119 | self.reschedule(); |
| 120 | } |
| 121 | |
| 122 | fn heavy(&self, need: &Need) -> bool { |
| 123 | let h = &self.cfg.heavy; |
| 124 | need.ram >= h.ram || need.disk >= h.disk || need.cores >= h.cores.resolve(sys::cores()) |
| 125 | } |
| 126 | |
| 127 | /// capped at a tenth of the fs, or a 16G tmpfs /tmp with a 20G reserve |
| 128 | /// admits nothing ever |
| 129 | fn reserve(&self, size: Bytes) -> Bytes { |
| 130 | self.cfg.budget.disk_reserve.min(Bytes(size.0 / 10)) |
| 131 | } |
| 132 | |
| 133 | fn fs_for(&mut self, cwd: &PathBuf) -> Vec<FsId> { |
| 134 | let mut ids = Vec::new(); |
| 135 | for path in std::iter::once(cwd).chain(&self.cfg.budget.disk_paths) { |
| 136 | if let Some(fs) = sys::fs(path) { |
| 137 | if !ids.contains(&fs.id) { |
| 138 | ids.push(fs.id); |
| 139 | } |
| 140 | let reserve = self.reserve(fs.size); |
| 141 | self.cap.disks.entry(fs.id).or_insert(Disk { |
| 142 | free: fs.free, |
| 143 | reserve, |
| 144 | consumed: Bytes::ZERO, |
| 145 | }); |
| 146 | } |
| 147 | } |
| 148 | ids |
| 149 | } |
| 150 | |
| 151 | fn refresh(&mut self) { |
| 152 | let (_, live) = sys::memory(); |
| 153 | self.cap.ram_live = live; |
| 154 | self.cap.pressure = self |
| 155 | .cfg |
| 156 | .budget |
| 157 | .psi_max |
| 158 | .is_some_and(|max| sys::pressure().is_some_and(|p| p > max)); |
| 159 | |
| 160 | let mut paths: Vec<PathBuf> = self.leases.values().map(|l| l.req.cwd.clone()).collect(); |
| 161 | paths.extend(self.cfg.budget.disk_paths.iter().cloned()); |
| 162 | let mut disks = BTreeMap::new(); |
| 163 | for path in paths { |
| 164 | if let Some(fs) = sys::fs(&path) { |
| 165 | disks.insert(fs.id, (fs.free, self.reserve(fs.size))); |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | let busy: Vec<FsId> = self |
| 170 | .leases |
| 171 | .values() |
| 172 | .filter(|l| l.run.is_some()) |
| 173 | .flat_map(|l| l.fs.iter().copied()) |
| 174 | .collect(); |
| 175 | self.baselines.retain(|id, _| busy.contains(id)); |
| 176 | self.cap.disks = disks |
| 177 | .into_iter() |
| 178 | .map(|(id, (free, reserve))| { |
| 179 | let consumed = if busy.contains(&id) { |
| 180 | self.baselines |
| 181 | .entry(id) |
| 182 | .or_insert(free) |
| 183 | .saturating_sub(free) |
| 184 | } else { |
| 185 | Bytes::ZERO |
| 186 | }; |
| 187 | ( |
| 188 | id, |
| 189 | Disk { |
| 190 | free, |
| 191 | reserve, |
| 192 | consumed, |
| 193 | }, |
| 194 | ) |
| 195 | }) |
| 196 | .collect(); |
| 197 | } |
| 198 | |
| 199 | fn running(&self) -> Vec<Running> { |
| 200 | self.leases |
| 201 | .values() |
| 202 | .filter_map(|l| { |
| 203 | let run = l.run.as_ref()?; |
| 204 | Some(Running { |
| 205 | need: l.req.need, |
| 206 | heavy: l.heavy, |
| 207 | fs: l.fs.clone(), |
| 208 | rss: run.rss, |
| 209 | remaining: l.req.need.time.saturating_sub(run.started.elapsed()), |
| 210 | }) |
| 211 | }) |
| 212 | .collect() |
| 213 | } |
| 214 | |
| 215 | fn reschedule(&mut self) { |
| 216 | let queue: Vec<Queued> = self |
| 217 | .leases |
| 218 | .iter() |
| 219 | .filter(|(_, l)| l.run.is_none()) |
| 220 | .map(|(id, l)| Queued { |
| 221 | id: *id, |
| 222 | need: l.req.need, |
| 223 | heavy: l.heavy, |
| 224 | fs: l.fs.clone(), |
| 225 | }) |
| 226 | .collect(); |
| 227 | if queue.is_empty() { |
| 228 | return; |
| 229 | } |
| 230 | let verdicts = sched::plan( |
| 231 | &queue, |
| 232 | &self.running(), |
| 233 | &self.cap, |
| 234 | self.cfg.queue.backfill_max_time.0, |
| 235 | ); |
| 236 | let cores = self.cap.cores; |
| 237 | let mut position = 0; |
| 238 | for (id, verdict) in verdicts { |
| 239 | let Some(lease) = self.leases.get_mut(&id) else { |
| 240 | continue; |
| 241 | }; |
| 242 | match verdict { |
| 243 | Verdict::Admit => { |
| 244 | lease.run = Some(Run { |
| 245 | started: Instant::now(), |
| 246 | rss: Bytes::ZERO, |
| 247 | reminded: None, |
| 248 | }); |
| 249 | let _ = proto::send( |
| 250 | &mut lease.out, |
| 251 | &Event::Granted { |
| 252 | lease: id, |
| 253 | cores: lease.req.need.cores.clamp(1, cores.max(1)), |
| 254 | heavy: lease.heavy, |
| 255 | }, |
| 256 | ); |
| 257 | } |
| 258 | Verdict::Wait(reason) => { |
| 259 | position += 1; |
| 260 | lease.waiting = Some(reason); |
| 261 | if lease.last_pos != position { |
| 262 | lease.last_pos = position; |
| 263 | let _ = proto::send( |
| 264 | &mut lease.out, |
| 265 | &Event::Queued { |
| 266 | lease: id, |
| 267 | position, |
| 268 | reason: Some(reason), |
| 269 | }, |
| 270 | ); |
| 271 | } |
| 272 | } |
| 273 | } |
| 274 | } |
| 275 | } |
| 276 | |
| 277 | fn remove(&mut self, id: LeaseId) { |
| 278 | if let Some(lease) = self.leases.remove(&id) { |
| 279 | for mut w in lease.watchers { |
| 280 | let _ = proto::send(&mut w, &Event::Fired { overdue: false }); |
| 281 | } |
| 282 | } |
| 283 | if self.leases.is_empty() { |
| 284 | self.idle_since = Instant::now(); |
| 285 | } |
| 286 | self.refresh(); |
| 287 | self.reschedule(); |
| 288 | } |
| 289 | |
| 290 | fn fire_overdue(&mut self) { |
| 291 | let grace = self.cfg.overrun.time_grace; |
| 292 | for lease in self.leases.values_mut() { |
| 293 | if !lease.watchers.is_empty() && lease.overdue(grace) { |
| 294 | for mut w in lease.watchers.drain(..) { |
| 295 | let _ = proto::send(&mut w, &Event::Fired { overdue: true }); |
| 296 | } |
| 297 | } |
| 298 | } |
| 299 | } |
| 300 | |
| 301 | fn info(&self) -> Vec<LeaseInfo> { |
| 302 | let grace = self.cfg.overrun.time_grace; |
| 303 | let mut position = 0; |
| 304 | self.leases |
| 305 | .iter() |
| 306 | .map(|(id, l)| { |
| 307 | if l.run.is_none() { |
| 308 | position += 1; |
| 309 | } |
| 310 | LeaseInfo { |
| 311 | id: *id, |
| 312 | cmd: l.req.cmd.clone(), |
| 313 | cwd: l.req.cwd.clone(), |
| 314 | session: l.req.session.clone(), |
| 315 | need: l.req.need, |
| 316 | heavy: l.heavy, |
| 317 | running: l.run.is_some(), |
| 318 | elapsed: l.elapsed().as_secs(), |
| 319 | rss: l.run.as_ref().map_or(Bytes::ZERO, |r| r.rss), |
| 320 | overdue: l.overdue(grace), |
| 321 | waiting: l.waiting.filter(|_| l.run.is_none()), |
| 322 | position: if l.run.is_none() { position } else { 0 }, |
| 323 | } |
| 324 | }) |
| 325 | .collect() |
| 326 | } |
| 327 | |
| 328 | fn status(&self) -> Status { |
| 329 | let (ram_free, cores_free) = sched::headroom(&self.running(), &self.cap); |
| 330 | Status { |
| 331 | leases: self.info(), |
| 332 | ram_free, |
| 333 | ram_budget: self.cap.ram_budget, |
| 334 | cores_free, |
| 335 | cores: self.cap.cores, |
| 336 | } |
| 337 | } |
| 338 | } |
| 339 | |
| 340 | fn serve(shared: &Shared, mut stream: UnixStream) -> io::Result<()> { |
| 341 | let mut reader = Reader::new(&stream)?; |
| 342 | let Some(req) = reader.recv::<Request>()? else { |
| 343 | return Ok(()); |
| 344 | }; |
| 345 | match req { |
| 346 | Request::Acquire(req) => { |
| 347 | let id = { |
| 348 | let mut st = lock(shared); |
| 349 | let id = st.next; |
| 350 | st.next += 1; |
| 351 | let fs = st.fs_for(&req.cwd); |
| 352 | let heavy = st.heavy(&req.need); |
| 353 | st.leases.insert( |
| 354 | id, |
| 355 | Lease { |
| 356 | req, |
| 357 | heavy, |
| 358 | fs, |
| 359 | queued_at: Instant::now(), |
| 360 | run: None, |
| 361 | out: stream.try_clone()?, |
| 362 | waiting: None, |
| 363 | last_pos: 0, |
| 364 | watchers: Vec::new(), |
| 365 | }, |
| 366 | ); |
| 367 | st.reschedule(); |
| 368 | id |
| 369 | }; |
| 370 | hold(shared, id, &mut reader); |
| 371 | lock(shared).remove(id); |
| 372 | Ok(()) |
| 373 | } |
| 374 | Request::Status => { |
| 375 | let status = lock(shared).status(); |
| 376 | proto::send(&mut stream, &Event::Status(status)) |
| 377 | } |
| 378 | Request::Kill { lease } => { |
| 379 | let mut st = lock(shared); |
| 380 | let reply = match st.leases.get_mut(&lease) { |
| 381 | Some(l) => { |
| 382 | let _ = proto::send(&mut l.out, &Event::Kill); |
| 383 | Event::Ok |
| 384 | } |
| 385 | None => Event::Err { |
| 386 | msg: format!("no lease {lease}"), |
| 387 | }, |
| 388 | }; |
| 389 | drop(st); |
| 390 | proto::send(&mut stream, &reply) |
| 391 | } |
| 392 | Request::Extend { lease, by } => { |
| 393 | let mut st = lock(shared); |
| 394 | let reply = match st.leases.get_mut(&lease) { |
| 395 | Some(l) => { |
| 396 | l.req.need.time += by.0; |
| 397 | if let Some(run) = &mut l.run { |
| 398 | run.reminded = None; |
| 399 | } |
| 400 | Event::Ok |
| 401 | } |
| 402 | None => Event::Err { |
| 403 | msg: format!("no lease {lease}"), |
| 404 | }, |
| 405 | }; |
| 406 | drop(st); |
| 407 | proto::send(&mut stream, &reply) |
| 408 | } |
| 409 | Request::Watch { lease } => { |
| 410 | let mut st = lock(shared); |
| 411 | let grace = st.cfg.overrun.time_grace; |
| 412 | match st.leases.get_mut(&lease) { |
| 413 | Some(l) if l.overdue(grace) => { |
| 414 | proto::send(&mut stream, &Event::Fired { overdue: true }) |
| 415 | } |
| 416 | Some(l) => { |
| 417 | l.watchers.push(stream); |
| 418 | Ok(()) |
| 419 | } |
| 420 | None => proto::send(&mut stream, &Event::Fired { overdue: false }), |
| 421 | } |
| 422 | } |
| 423 | Request::Overdue { session } => { |
| 424 | let mut st = lock(shared); |
| 425 | let grace = st.cfg.overrun.time_grace; |
| 426 | let every = st.cfg.overrun.remind_every.0; |
| 427 | let mut due = Vec::new(); |
| 428 | for (id, l) in &mut st.leases { |
| 429 | if l.req.session.as_deref() != Some(&session) || !l.overdue(grace) { |
| 430 | continue; |
| 431 | } |
| 432 | let Some(run) = &mut l.run else { continue }; |
| 433 | if run.reminded.is_some_and(|t| t.elapsed() < every) { |
| 434 | continue; |
| 435 | } |
| 436 | run.reminded = Some(Instant::now()); |
| 437 | due.push(*id); |
| 438 | } |
| 439 | let leases = st |
| 440 | .info() |
| 441 | .into_iter() |
| 442 | .filter(|i| due.contains(&i.id)) |
| 443 | .collect(); |
| 444 | drop(st); |
| 445 | proto::send(&mut stream, &Event::Overdue { leases }) |
| 446 | } |
| 447 | Request::Started { .. } | Request::Usage { .. } | Request::Done => proto::send( |
| 448 | &mut stream, |
| 449 | &Event::Err { |
| 450 | msg: "no lease on this connection".into(), |
| 451 | }, |
| 452 | ), |
| 453 | } |
| 454 | } |
| 455 | |
| 456 | /// a client that vanishes without saying done got sigkilled or similar, and |
| 457 | /// pdeathsig only reaches its direct child. take the rest of the group down |
| 458 | /// so a headless build doesn't keep eating ram off the books |
| 459 | fn hold(shared: &Shared, id: LeaseId, reader: &mut Reader) { |
| 460 | let mut pgid = None; |
| 461 | let clean = loop { |
| 462 | match reader.recv::<Request>() { |
| 463 | Ok(Some(Request::Usage { rss })) => { |
| 464 | if let Some(run) = lock(shared) |
| 465 | .leases |
| 466 | .get_mut(&id) |
| 467 | .and_then(|l| l.run.as_mut()) |
| 468 | { |
| 469 | run.rss = rss; |
| 470 | } |
| 471 | } |
| 472 | Ok(Some(Request::Started { pid })) => pgid = Some(pid), |
| 473 | Ok(Some(Request::Done)) => break true, |
| 474 | Ok(Some(_)) => {} |
| 475 | Ok(None) | Err(_) => break false, |
| 476 | } |
| 477 | }; |
| 478 | if !clean && let Some(pgid) = pgid.filter(|p| *p > 1) { |
| 479 | unsafe { libc::kill(-pgid, libc::SIGTERM) }; |
| 480 | } |
| 481 | } |
| 482 | |
| 483 | fn tick(shared: &Shared) { |
| 484 | loop { |
| 485 | thread::sleep(Duration::from_secs(1)); |
| 486 | let mut st = lock(shared); |
| 487 | st.refresh(); |
| 488 | st.reschedule(); |
| 489 | st.fire_overdue(); |
| 490 | if st.conns == 0 && st.leases.is_empty() && st.idle_since.elapsed() >= st.cfg.queue.linger.0 |
| 491 | { |
| 492 | // a spawner holds this while it clears a stale socket, so taking it |
| 493 | // here stops us unlinking our successor's socket on the way out |
| 494 | if let Some(_guard) = try_spawn_lock() { |
| 495 | let _ = fs::remove_file(paths::socket()); |
| 496 | process::exit(0); |
| 497 | } |
| 498 | } |
| 499 | } |
| 500 | } |
| 501 | |
| 502 | /// watches the directory, not the file. editors save by renaming over it, |
| 503 | /// which would orphan a watch on the old inode after the first edit |
| 504 | #[cfg(target_os = "linux")] |
| 505 | fn watch_config(shared: &Shared) { |
| 506 | use std::os::unix::ffi::OsStrExt; |
| 507 | |
| 508 | let path = paths::config_file(); |
| 509 | let (Some(dir), Some(name)) = (path.parent(), path.file_name()) else { |
| 510 | return; |
| 511 | }; |
| 512 | let _ = fs::create_dir_all(dir); |
| 513 | let Ok(cdir) = std::ffi::CString::new(dir.as_os_str().as_bytes()) else { |
| 514 | return; |
| 515 | }; |
| 516 | let fd = unsafe { libc::inotify_init1(libc::IN_CLOEXEC) }; |
| 517 | if fd < 0 { |
| 518 | return; |
| 519 | } |
| 520 | let mask = libc::IN_CLOSE_WRITE |
| 521 | | libc::IN_MOVED_TO |
| 522 | | libc::IN_MOVED_FROM |
| 523 | | libc::IN_CREATE |
| 524 | | libc::IN_DELETE; |
| 525 | if unsafe { libc::inotify_add_watch(fd, cdir.as_ptr(), mask) } < 0 { |
| 526 | return; |
| 527 | } |
| 528 | |
| 529 | let header = std::mem::size_of::<libc::inotify_event>(); |
| 530 | let mut buf = [0u8; 4096]; |
| 531 | loop { |
| 532 | let n = unsafe { libc::read(fd, buf.as_mut_ptr().cast(), buf.len()) }; |
| 533 | let Ok(n) = usize::try_from(n) else { |
| 534 | return; |
| 535 | }; |
| 536 | let mut hit = false; |
| 537 | let mut off = 0; |
| 538 | while off + header <= n { |
| 539 | let ev: libc::inotify_event = |
| 540 | unsafe { std::ptr::read_unaligned(buf.as_ptr().add(off).cast()) }; |
| 541 | let len = ev.len as usize; |
| 542 | let raw = &buf[off + header..(off + header + len).min(n)]; |
| 543 | let ev_name = raw.split(|b| *b == 0).next().unwrap_or_default(); |
| 544 | hit |= ev_name == name.as_bytes(); |
| 545 | off += header + len; |
| 546 | } |
| 547 | if !hit { |
| 548 | continue; |
| 549 | } |
| 550 | thread::sleep(Duration::from_millis(100)); |
| 551 | match Config::load() { |
| 552 | Ok(cfg) => { |
| 553 | lock(shared).reload(cfg); |
| 554 | eprintln!("roundabout: config reloaded"); |
| 555 | } |
| 556 | Err(e) => eprintln!("roundabout: kept the old config, {e}"), |
| 557 | } |
| 558 | } |
| 559 | } |
| 560 | |
| 561 | fn try_spawn_lock() -> Option<client::Flock> { |
| 562 | let file = fs::OpenOptions::new() |
| 563 | .write(true) |
| 564 | .open(paths::spawn_lock()) |
| 565 | .ok()?; |
| 566 | client::Flock::try_exclusive(file) |
| 567 | } |
| 568 | |
| 569 | pub fn main(cfg: Config) -> ! { |
| 570 | // fork once more so the client that spawned us can reap its child now |
| 571 | // instead of carrying a zombie for the length of its build |
| 572 | match unsafe { libc::fork() } { |
| 573 | -1 => { |
| 574 | eprintln!("roundabout: fork failed: {}", io::Error::last_os_error()); |
| 575 | process::exit(1); |
| 576 | } |
| 577 | 0 => {} |
| 578 | _ => process::exit(0), |
| 579 | } |
| 580 | |
| 581 | let listener = match UnixListener::bind(paths::socket()) { |
| 582 | Ok(l) => l, |
| 583 | Err(e) if e.kind() == io::ErrorKind::AddrInUse && client::connect().is_some() => { |
| 584 | process::exit(0) |
| 585 | } |
| 586 | Err(e) => { |
| 587 | eprintln!("roundabout: bind: {e}"); |
| 588 | process::exit(1); |
| 589 | } |
| 590 | }; |
| 591 | |
| 592 | let shared: Shared = Arc::new(Mutex::new(State::new(cfg))); |
| 593 | { |
| 594 | let shared = shared.clone(); |
| 595 | thread::spawn(move || tick(&shared)); |
| 596 | } |
| 597 | #[cfg(target_os = "linux")] |
| 598 | { |
| 599 | let shared = shared.clone(); |
| 600 | thread::spawn(move || watch_config(&shared)); |
| 601 | } |
| 602 | |
| 603 | for stream in listener.incoming() { |
| 604 | let Ok(stream) = stream else { continue }; |
| 605 | lock(&shared).conns += 1; |
| 606 | let shared = shared.clone(); |
| 607 | thread::spawn(move || { |
| 608 | if let Err(e) = serve(&shared, stream) { |
| 609 | eprintln!("roundabout: {e}"); |
| 610 | } |
| 611 | let mut st = lock(&shared); |
| 612 | st.conns -= 1; |
| 613 | if st.leases.is_empty() { |
| 614 | st.idle_since = Instant::now(); |
| 615 | } |
| 616 | }); |
| 617 | } |
| 618 | process::exit(0) |
| 619 | } |