rust-kernel

//! echoserver -- an event-loop echo server over loopback (127.0.0.1).
//!
//! One process serves several clients at once without threads: it never
//! blocks on any single socket. Every socket is nonblocking, and the loop
//! asks the kernel "which of my sockets are ready?" (epoll_wait, or plain
//! poll), then only accepts/reads/writes those:
//!
//!     loop {
//!         ready = epoll_wait(...)        // sleep until something happens
//!         for each ready fd:
//!             listener -> accept new clients, watch them too
//!             client   -> read; echo it back; 0 bytes = client left
//!     }
//!
//! The self-test forks 3 client processes that talk to it over loopback,
//! once with epoll and once with poll, then checks UDP over loopback.
//! With an argument (`echoserver PORT`) it just serves forever.
#![no_std]
#![no_main]

#[macro_use]
#[path = "ulib.rs"]
mod ulib;

extern crate alloc;

use alloc::format;
use alloc::vec::Vec;
use ulib::*;

const LOOPBACK: Ip = Ip([127, 0, 0, 1]);
const CLIENTS: usize = 3;
const ROUNDS: usize = 3;

/// A client process: connect, send ROUNDS messages, check each echo.
fn client(port: u16, id: usize) -> i64 {
    let fd = socket(SOCK_STREAM, 0);
    if connect(fd, LOOPBACK, port) < 0 {
        return 10;
    }
    for round in 0..ROUNDS {
        let msg = format!("hello {} from client {}", round, id);
        write_all(fd, msg.as_bytes());
        let mut buf = [0u8; 64];
        let mut got = 0;
        while got < msg.len() {
            let n = read(fd, &mut buf[got..]);
            if n <= 0 {
                return 11;
            }
            got += n as usize;
        }
        if &buf[..got] != msg.as_bytes() {
            return 12;
        }
    }
    close(fd); // the server sees end-of-file
    0
}

fn listener(port: u16) -> i64 {
    let fd = socket(SOCK_STREAM | O_NONBLOCK, 0); // SOCK_NONBLOCK
    if bind(fd, port) < 0 || listen(fd, 4) < 0 {
        return -1;
    }
    fd
}

/// Accept everyone waiting (nonblocking: stop at EAGAIN).
fn accept_all(lfd: i64, mut add: impl FnMut(i64)) {
    loop {
        let (fd, _) = accept(lfd);
        if fd < 0 {
            break; // -EAGAIN: nobody else waiting
        }
        set_nonblocking(fd, true);
        add(fd);
    }
}

/// Read what's there and echo it back. Returns false when the client left.
fn echo(fd: i64, served: &mut usize) -> bool {
    let mut buf = [0u8; 512];
    loop {
        let n = read(fd, &mut buf);
        if n == -EAGAIN {
            return true; // drained for now
        }
        if n <= 0 {
            return false; // 0 = end of file (or an error)
        }
        write_all(fd, &buf[..n as usize]);
        *served += 1;
    }
}

/// The epoll flavour: register fds once, then wait for just the ready ones.
/// Returns the number of clients served to the end (or -1 on a stall).
fn serve_epoll(lfd: i64, want: usize) -> i64 {
    let ep = epoll_create1(0);
    epoll_ctl(ep, EPOLL_CTL_ADD, lfd, EPOLLIN, lfd as u64);
    let (mut finished, mut served) = (0, 0);
    let mut events = [EpollEvent::default(); 8];
    while finished < want {
        let n = epoll_wait(ep, &mut events, 5000);
        if n <= 0 {
            close(ep);
            return -1; // 5 s with nothing happening: something's wrong
        }
        for ev in &events[..n as usize] {
            let fd = ev.data as i64;
            if fd == lfd {
                accept_all(lfd, |c| {
                    epoll_ctl(ep, EPOLL_CTL_ADD, c, EPOLLIN, c as u64);
                });
            } else if !echo(fd, &mut served) {
                epoll_ctl(ep, EPOLL_CTL_DEL, fd, 0, 0);
                close(fd);
                finished += 1;
            }
        }
    }
    close(ep);
    println!("echoserver: epoll: {} clients, {} echoes", finished, served);
    finished as i64
}

/// The poll flavour: pass the whole fd list on every call.
fn serve_poll(lfd: i64, want: usize) -> i64 {
    let mut fds = alloc::vec![PollFd { fd: lfd as i32, events: POLLIN, revents: 0 }];
    let (mut finished, mut served) = (0, 0);
    while finished < want {
        if poll(&mut fds, 5000) <= 0 {
            return -1;
        }
        let mut new = Vec::new();
        let mut gone = Vec::new();
        for p in fds.iter() {
            if p.revents == 0 {
                continue;
            }
            if p.fd as i64 == lfd {
                accept_all(lfd, |c| new.push(c));
            } else if !echo(p.fd as i64, &mut served) {
                close(p.fd as i64);
                gone.push(p.fd);
                finished += 1;
            }
        }
        fds.retain(|p| !gone.contains(&p.fd));
        fds.extend(new.into_iter().map(|c| PollFd { fd: c as i32, events: POLLIN, revents: 0 }));
    }
    println!("echoserver: poll: {} clients, {} echoes", finished, served);
    finished as i64
}

/// Start the clients, run the server, collect the clients' results.
fn round(c: &mut Checks, port: u16, use_epoll: bool) {
    let lfd = listener(port);
    c.check(lfd >= 0, "listen on a loopback port");
    let mut pids = Vec::new();
    for id in 0..CLIENTS {
        let pid = fork();
        if pid == 0 {
            close(lfd);
            exit(client(port, id));
        }
        pids.push(pid);
    }
    let finished = if use_epoll { serve_epoll(lfd, CLIENTS) } else { serve_poll(lfd, CLIENTS) };
    close(lfd);
    let ok = pids.iter().all(|&pid| wait(pid) == 0);
    let what = if use_epoll { "epoll" } else { "poll" };
    c.check(finished == CLIENTS as i64, &format!("{}: served {} clients to the end", what, CLIENTS));
    c.check(ok, &format!("{}: every client got its echoes back", what));
}

fn main() -> i64 {
    if let Some(port) = arg(1).and_then(|a| a.parse().ok()) {
        let lfd = listener(port);
        if lfd < 0 {
            eprintln!("echoserver: can't listen on port {}", port);
            return 1;
        }
        println!("echoserver: listening on port {} (epoll)", port);
        loop {
            serve_epoll(lfd, usize::MAX);
        }
    }
    let mut c = Checks::new("echoserver");
    round(&mut c, 7070, true);
    round(&mut c, 7071, false);

    // UDP over loopback: a datagram to ourselves.
    let a = socket(SOCK_DGRAM, 0);
    let b = socket(SOCK_DGRAM, 0);
    bind(b, 7072);
    c.check(sendto(a, b"ping", LOOPBACK, 7072) == 4, "UDP sendto 127.0.0.1");
    set_recv_timeout(b, 2000);
    let mut buf = [0u8; 16];
    let (n, from) = recvfrom(b, &mut buf);
    c.check(n == 4 && &buf[..4] == b"ping" && from.ip() == LOOPBACK, "UDP datagram arrives, from 127.0.0.1");
    close(a);
    close(b);
    c.done()
}