← Line-Protocol TCP Server networkinghard ready

Problem

Line-Protocol TCP Server (Rust)

Topics: TCP, line protocol, threads, graceful drain

Problem

Build a small scheme simulator: a TCP server speaking a newline-terminated request/response protocol, with a graceful shutdown that drains in-flight work. (Inspired by the payment-scheme simulators payment platforms build for testing — the protocol here is a generic task runner.)

pub struct Cancel(/* … */);
pub struct Server { /* … */ }
impl Server {
    pub fn new(addr: &str, grace: Duration) -> Server;
    pub fn run(&self, cancel: &Cancel) -> std::io::Result<()>;
    pub fn addr(&self) -> String;
    pub fn wait_started(&self, timeout: Duration) -> bool;
}

Protocol (one request in flight per connection, connections reused):

RUN|<ms>\n → after ~<ms> of simulated work, OK|done\n.
Malformed (not RUN|<int>) → ERR|invalid request\n.
<ms> not a non-negative integer → ERR|invalid amount\n.

Graceful shutdown (when cancel fires):

Stop accepting new connections.
Give in-flight requests up to grace to finish.
A request still running when grace expires → ERR|cancelled\n.
run returns once in-flight work has drained.

Key concepts

Per-connection thread reading newline-framed requests via BufRead::read_line; one request is answered before the next is read.
Drain accounting: a Drain holding an in-flight count (Mutex + Condvar) and a hard AtomicBool. On shutdown the accept loop stops, waits for the count to reach zero up to grace, then sets hard so long-running work bails out with cancelled.
Scope: slow/half-open clients are out of scope — only clean termination by client or server.

Cancel, Server::{new, addr, wait_started}, Drain, and parse are provided; you implement run and the connection handling.

Grading

Your solution.rs is compiled together with a trusted tests.rs (which include!s it) using rustc --test. Only solution.rs is yours to edit.

Sign in to submit your solution.

Leaderboard score: lower is better, 1.0 ≈ reference

#	Name	score	ns/op	allocs

Output

No submission yet.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

// A line-protocol TCP server (a small "scheme simulator"). Each connection
// carries newline-terminated requests, one in flight at a time:
//
//   RUN|<ms>\n   -> OK|done\n               (after ~<ms> of simulated work)
//   <anything>   -> ERR|invalid request\n   (wrong shape)
//   RUN|-5\n     -> ERR|invalid amount\n     (ms not a non-negative integer)
//
// Graceful shutdown: when `cancel` fires the server stops accepting new
// connections and gives in-flight requests up to `grace` to finish; a request
// still running when the grace period ends is answered ERR|cancelled. `run`
// returns once in-flight work has drained.
//
// `Cancel`, `Server::{new, addr, wait_started}`, the `Drain` state, and `parse`
// are provided; implement `run` (accept loop + drain) and the connection handling.

/// Cooperative shutdown signal.
#[derive(Clone, Default)]
pub struct Cancel(Arc<AtomicBool>);

impl Cancel {
    pub fn new() -> Self {
        Cancel(Arc::new(AtomicBool::new(false)))
    }
    pub fn cancel(&self) {
        self.0.store(true, Ordering::SeqCst);
    }
    pub fn is_cancelled(&self) -> bool {
        self.0.load(Ordering::SeqCst)
    }
}

pub struct Server {
    addr: String,
    grace: Duration,
    bound: (Mutex<Option<String>>, Condvar),
}

/// Shared drain state: requests mid-processing, and whether the grace period
/// has expired (the hard deadline to cancel them).
pub struct Drain {
    pub inflight: Mutex<usize>,
    pub idle: Condvar,
    pub hard: AtomicBool,
}

impl Server {
    pub fn new(addr: &str, grace: Duration) -> Server {
        Server {
            addr: addr.to_string(),
            grace,
            bound: (Mutex::new(None), Condvar::new()),
        }
    }

/// Binds, serves until `cancel` fires, then drains in-flight requests within
    /// `grace` and returns.
    pub fn run(&self, cancel: &Cancel) -> std::io::Result<()> {
        let _ = (&self.grace, cancel);
        unimplemented!()
    }

pub fn addr(&self) -> String {
        let (lock, _) = &self.bound;
        lock.lock().unwrap().clone().unwrap_or_else(|| self.addr.clone())
    }

pub fn wait_started(&self, timeout: Duration) -> bool {
        let (lock, cvar) = &self.bound;
        let deadline = Instant::now() + timeout;
        let mut guard = lock.lock().unwrap();
        while guard.is_none() {
            let now = Instant::now();
            if now >= deadline {
                return false;
            }
            let (g, res) = cvar.wait_timeout(guard, deadline - now).unwrap();
            guard = g;
            if res.timed_out() && guard.is_none() {
                return false;
            }
        }
        true
    }

/// Call from `run` once bound to publish the address for `wait_started`/`addr`.
    fn announce(&self, local: String) {
        let (lock, cvar) = &self.bound;
        *lock.lock().unwrap() = Some(local);
        cvar.notify_all();
    }
}

/// Returns Ok(ms) for a well-formed `RUN|<non-negative int>`, else an error reason.
fn parse(line: &str) -> Result<u64, &'static str> {
    let rest = line.strip_prefix("RUN|").ok_or("invalid request")?;
    if rest.is_empty() {
        return Err("invalid request");
    }
    rest.parse::<u64>().map_err(|_| "invalid amount")
}