async -- 퓨처, 실행기, 비동기 I/O #

#include <std/async/async.h>   // everything that runs on any target
#include <std/async/host.h>    // block_on, HostPark            (hosted)
#include <std/async/pool.h>    // ThreadPool                    (hosted)
#include <std/async/net.h>     // Runtime, async TCP            (hosted)

#include <std/async/async.h>
#include <std/async/host.h>
extern int printf(const char* fmt, ...);

struct Countdown {
    int          st;      // resume point
    int          i;       // lives across awaits: a field, not a local
    struct Sleep d;       // the child future being awaited
    struct Poll<int> poll(&Context cx);
    void drop();
};

struct Poll<int> Countdown::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 1);                       // one await point, numbered 1
    for (self.i = 3; self.i > 0; self.i = self.i - 1) {
        self.d = Sleep::new(100LL);                // ms
        AWAIT(1, self.d);                          // Pending until the timer fires
        printf("%d\n", self.i);
    }
    ASYNC_RETURN(42);
    ASYNC_END;
}

void Countdown::drop() {
    if (self.st == 1) self.d.drop();               // abandoned while sleeping
    self.st = ASYNC_DONE;
}

int main() {
    struct Countdown c;
    c.st = ASYNC_INIT;
    int v = block_on(&c);                          // prints 3 2 1, returns 42
    c.drop();                                      // a Drop type: finished, so this does nothing
    return v == 42 ? 0 : 1;
}

generic<T> struct Poll {
    int ready_;                     // 1 = Ready, 0 = Pending
    T   value;                      // set when Ready
    static struct Poll<T> ready(T v);
    static struct Poll<T> pending();
    int is_ready() const;
    int is_pending() const;
    ?T  take() consume;             // the value, or null if Pending
};

trait Future {
    void poll(&Context cx);         // checked by name: returns struct Poll<T> for some T
    void drop();
}

typedef int Unit;                   // the output of a future that produces nothing (Rust's ())
#define UNIT 0

struct Waker {
    generic<T> static struct Waker new(fn void(?&T, unsigned long) wake, ?&T data, unsigned long tag);
    generic<T> static struct Waker counted(fn void(?&T, unsigned long) wake,
                                           fn void(?&T, unsigned long) clone,
                                           fn void(?&T, unsigned long) drop,
                                           ?&T data, unsigned long tag);
    static struct Waker noop();

    void wake() consume;                 // wake_by_ref(), then drop()
    void wake_by_ref() const;
    struct Waker clone() const;          // runs the clone hook
    void drop();                         // runs the drop hook; a dropped waker wakes nothing
    int  will_wake(const &Waker other) const;   // same task? (skip re-registering)
    int  is_noop() const;
};

struct WakerSlot {                     // one waiting future (Rust's AtomicWaker)
    static struct WakerSlot new();
    void register_(const &Waker w);    // keep a clone (replacing a different one)
    int  wake();                       // wake and release; 1 if there was one
    void clear();
};

#define ASYNC_WAITERS 32
struct WaitList {                      // several waiting futures, no heap
    static struct WaitList new();
    int  add(int id, const &Waker w);  // register or refresh; returns the id, 0 if full
    void remove(int id);
    int  contains(int id) const;
    int  wake_one();
    int  wake_all();
    int  len() const;
};

struct Context {
    struct Waker  waker;     // the task being polled
    ?&TimerQueue  timers;    // the executor's timers: what Sleep/Interval/Timeout use
    ?&struct IoDriver io_;   // the executor's IoDriver (std/async/net.h), via io_driver(cx)
    ?&Erased      exec_;     // the executor, and its type-erased spawn (async_spawn; see std/erased.h)
    ?fn int(?&Erased, ?&struct TaskHeader) spawn_;

    static struct Context from_waker(struct Waker w);   // no timers, no I/O, no spawning
    static struct Context noop();                        // Waker::noop(): poll by hand
    void wake_by_ref() const;
};

#include <std/async/coro.h>      // (async.h includes it)

struct Poll<int> Fetch::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 2);
    self.total = 0;
    for (self.i = 0; self.i < 3; self.i = self.i + 1) {
        self.delay = Sleep::new(10LL);
        AWAIT(1, self.delay);
        self.total = self.total + self.i;
    }
    ASYNC_YIELD(2);
    ASYNC_RETURN(self.total);
    ASYNC_END;
}

struct Poll<int> Fetch::poll(&Context cx) {
    &AsyncResume __async_st = &(self.st);            // alias of the state field
    AsyncResume __async_point_1 = 1;                  // one marker per declared point
    AsyncResume __async_point_2 = 2;
    switch (*__async_st) {                            // resume where we stopped
    case 0: break;                                    // first poll: start at the top
    case 1: goto __async_resume_1;
    case 2: goto __async_resume_2;
    default:
        std::panic_at("std/async: coroutine polled after it completed", ...);
        return Poll::pending();
    }
    self.total = 0;
    for (self.i = 0; self.i < 3; self.i = self.i + 1) {
        self.delay = Sleep::new(10LL);
        *__async_st = __async_point_1;                // AWAIT(1, self.delay)
    __async_resume_1:                                 //   <- a goto target inside the loop
        if ((self.delay).poll(cx).is_pending() != 0) return Poll::pending();
        self.total = self.total + self.i;
    }
    *__async_st = __async_point_2;                    // ASYNC_YIELD(2)
    cx.wake_by_ref();
    return Poll::pending();
__async_resume_2:
    { }
    return std::async_finish_(__async_st, Poll::ready(self.total));   // ASYNC_RETURN
    std::panic_at("... reached ASYNC_END without ASYNC_RETURN", ...);  // ASYNC_END
    return Poll::pending();
}

// wrong: 'n' would be garbage after resuming (compile error)
struct Poll<int> Bad::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 1);
    int n = compute();
    AWAIT(1, self.delay);
    ASYNC_RETURN(n);
    ASYNC_END;
}

// right: the state struct holds it
struct Poll<int> Good::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 1);
    self.n = compute();
    AWAIT(1, self.delay);
    ASYNC_RETURN(self.n);
    ASYNC_END;
}

void Fetch::drop() {
    if (self.st == 1) self.delay.drop();
    if (self.st == 2) self.rx.drop();
    self.st = ASYNC_DONE;
}

struct Twice {
    int          st;
    struct Sleep d;
    int          round;
    struct Poll<int> poll(&Context cx);
    void drop();
};

struct Poll<int> Twice::poll(&Context cx) {
    switch (self.st) {
    case 0: break;
    case 1: goto resume1;
    default: return Poll::pending();
    }
    for (self.round = 0; self.round < 2; self.round = self.round + 1) {
        self.d = Sleep::new(50LL);
        self.st = 1;
    resume1:
        if (self.d.poll(cx).is_pending() != 0) return Poll::pending();
    }
    self.st = -1;
    return Poll::ready(self.round);
}

void Twice::drop() { if (self.st == 1) self.d.drop(); self.st = -1; }

struct Flag {                      // Ready once someone calls set()
    int              is_set;
    struct WakerSlot waiter;
    struct Poll<Unit> poll(&Context cx);
    void drop();
    void set();
};

struct Poll<Unit> Flag::poll(&Context cx) {
    if (self.is_set != 0) return Poll::ready(UNIT);
    self.waiter.register_(&cx.waker);         // first register ...
    return Poll::pending();                    // ... then report Pending
}
void Flag::set()  { self.is_set = 1; self.waiter.wake(); }
void Flag::drop() { self.waiter.drop(); }   // 갖고 있던 waker를 놓습니다

generic<F: Future, T> struct AsyncTask {
    struct TaskHeader hdr;             // what executors and wakers see
    F                 fut;
    T                 out[1];          // the output once finished
    static struct AsyncTask<F, T> new(F fut);
    int is_finished() const;           // done, either way
    int is_aborted() const;
    ?T  take();                        // the output if it finished (not aborted)
};

generic<T> struct JoinHandle {         // returned by spawn
    struct Poll<?T> poll(&Context cx); // await it: Ready(some(v)), or Ready(null) if aborted
    void abort();                      // cancel: the executor drops the future
    void drop();                       // stop waiting (the task keeps running: detach)
    int  is_finished() const;
};

static struct AsyncTask<struct Fetch, int> t1;      // static storage
t1 = AsyncTask::new(Fetch::new());
struct JoinHandle<int> h = ex.spawn(&t1);

struct LocalExecutor {
    struct TimerQueue timers;           // handed to every poll as cx.timers
    static struct LocalExecutor new();
    generic<F, T> struct JoinHandle<T> spawn(&AsyncTask<F, T> task);
    int  run_ready();                   // one round over the woken tasks
    int  turn(long long now_ms);        // advance the timers to now, then one round
    generic<P: Park> void run(&P park); // until every task has finished
    generic<P: Park, F: Future, T> T block_on(&P park, &F fut);
    int  live_count() const;
    int  has_ready() const;
};

trait Park {
    long long now_ms();               // monotonic milliseconds
    void park(long long timeout_ms);  // wait up to timeout (-1: until unpark), or less
    void unpark();                    // end the current (or the next) park early
}

struct LocalExecutor ex = LocalExecutor::new();
struct VirtualClock clk = VirtualClock::new(0LL);
ex.spawn(&t1);
ex.spawn(&t2);
ex.run(&clk);
// clk.now is now exactly the virtual time the tasks needed

static volatile long long g_ms = 0;               // incremented by the SysTick handler
static long long board_now() { return g_ms; }
static void board_idle(long long timeout_ms) { unsafe { asm volatile("wfi"); } }   // any interrupt wakes it

static struct LocalExecutor ex;
int main() {
    ex = LocalExecutor::new();
    struct FnPark park = FnPark::new(board_now, board_idle);
    ex.spawn(&blink_task);
    ex.spawn(&uart_task);
    ex.run(&park);
}

#include <std/async/host.h>
generic<F: Future, T> T block_on(&F fut);

#include <std/async/pool.h>
struct ThreadPool {
    struct TimerQueue timers;            // shared by the workers
    static struct ThreadPool new();
    int  start(int n);                   // n worker threads (up to ASYNC_POOL_MAX_THREADS = 64)
    generic<F: Send, T: Send> struct JoinHandle<T> spawn(&AsyncTask<F, T> task);
    generic<F: Future + Send, T: Send> T block_on(F fut);   // run as a task, block this thread for the value
    void wait_idle();                    // every spawned task finished
    void shutdown() consume;             // stop and join the workers
};

static struct ThreadPool pool;
pool = ThreadPool::new();
pool.start(4);
for (int i = 0; i < 32; i = i + 1) { jobs[i] = AsyncTask::new(Job::new(i)); pool.spawn(&jobs[i]); }
pool.wait_idle();
int seven = pool.block_on(Job::new(7));
pool.shutdown();

struct TimerQueue {                     // every executor owns one: cx.timers
    static struct TimerQueue new();
    int       advance(long long now_ms);   // move the clock, wake what is due
    long long now_ms() const;
    long long next_deadline() const;       // -1 if none
    long long park_timeout() const;        // ms until the next deadline (0 = due, -1 = none)
    int       pending() const;
};

struct Sleep {                          // Ready (UNIT) 'ms' after its first poll
    static struct Sleep new(long long ms);
    static struct Sleep until(long long deadline_ms);
    void reset(long long ms);           // re-arm (drops a pending registration)
    int  is_elapsed() const;
};

struct Interval {                       // a stream of ticks: Ready(1), Ready(2), ... one per period
    static struct Interval new(long long period_ms);
};

struct Join2<struct Sleep, struct Fetch, Unit, int> j = join2(Sleep::new(50LL), Fetch::new());
struct Select2<struct AcceptFut, struct Cancelled, struct AsyncTcpStream, Unit> s =
    select2(listener.accept(), token.cancelled_fut());
struct Timeout<struct ReadFut, long long> t = timeout(conn.read(self.buf[0..512]), 1000LL);

generic<T: Send> struct Oneshot {             // one value, sent once
    static struct Oneshot<T> new();
    int  send(T v);                     // 1, or 0 if already sent / closed
    void close();                       // the receiver gets null
    struct OneshotRecv<T> recv();       // future: Ready(some(v)) or Ready(null)
    ?T   try_recv();
};

generic<T: Send> struct AsyncChannel {        // bounded MPMC FIFO over your buffer
    static struct AsyncChannel<T> new([]T storage);
    struct ChanSend<T> send(T v);       // future: Ready(1) sent, Ready(0) closed
    struct ChanRecv<T> recv();          // future: Ready(some(v)), Ready(null) once closed and drained
    int  try_send(T v);                 // 1 sent, 0 full, -1 closed
    ?T   try_recv();
    void close();
    unsigned long len() const;
    unsigned long capacity() const;
};

static int slots[8];
static struct AsyncChannel<int> ch;
ch = AsyncChannel::new(slots[0..8]);
// producer coroutine:  self.tx = ch.send(v);  AWAIT_INTO(1, self.ok, self.tx);
// consumer coroutine:  self.rx = ch.recv();   AWAIT_INTO(1, self.got, self.rx);   // ?int

struct AsyncMutex {
    static struct AsyncMutex new();
    struct MutexLock lock();            // future: Ready once this task holds the lock
    int  try_lock();
    void unlock();
};

struct AsyncSemaphore {
    static struct AsyncSemaphore new(long long permits);
    struct SemAcquire acquire(long long n);
    int  try_acquire(long long n);
    void release(long long n);
    long long available() const;
};

struct AsyncNotify {
    static struct AsyncNotify new();
    struct Notified notified();         // future
    void notify_one();                  // wake one waiter, or store one permit
    void notify_waiters();              // wake everyone waiting now, store nothing
};

struct CancelToken {
    static struct CancelToken new();
    static struct CancelToken child_of(&CancelToken parent);   // cancelled with any ancestor
    void cancel();
    int  is_cancelled() const;
    struct Cancelled cancelled_fut();   // future: Ready once cancelled
};
generic<F, T> struct Cancellable<F, T> with_cancel(F inner, &CancelToken token);

struct Poll<int> Writer::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 2);
    self.l = g_mutex.lock();
    AWAIT(1, self.l);                   // exclusive from here ...
    self.wr = g_conn.write_all(self.msg[0..self.len]);
    AWAIT_INTO(2, self.ok, self.wr);
    g_mutex.unlock();                   // ... to here, across an await
    ASYNC_RETURN(self.ok);
    ASYNC_END;
}

#include <std/async/net.h>

struct Runtime {                        // tokio's current-thread runtime
    struct LocalExecutor ex;
    struct IoDriver      io;
    static struct Runtime new();
    int  init();                        // open the OS event queue; 0 on success
    void close_();
    generic<F, T> struct JoinHandle<T> spawn(&AsyncTask<F, T> task);
    void run();
    generic<F: Future, T> T block_on(&F fut);
};

struct AsyncTcpListener {
    int fd;                                              // < 0: bind/listen failed
    static struct AsyncTcpListener bind(unsigned short port);   // 0: any free port
    struct AcceptFut accept();                           // -> struct AsyncTcpStream (.fd < 0 on error)
    unsigned short local_port() const;
    int close() consume;
};

struct AsyncTcpStream {
    int fd;
    static struct AsyncTcpStream from_fd(int fd);
    static struct ConnectFut connect(unsigned int addr_network_order, unsigned short port);
    struct ReadFut     read([]unsigned char buf);        // -> long long: bytes, 0 at EOF, -1 on error
    struct WriteFut    write([]const unsigned char buf);       // -> long long: bytes written, -1 on error
    struct WriteAllFut write_all([]const unsigned char buf);   // -> int: 1 all written, 0 on error
    int close() consume;
};

struct IoReady { static struct IoReady new(int fd, int filter); };   // -> UNIT once ready

#include <std/async/async.h>
#include <std/async/net.h>

struct EchoConn {
    int st; struct AsyncTcpStream c; struct ReadFut rd; struct WriteAllFut wr;
    long long n; int ok; unsigned char buf[512];
    struct Poll<int> poll(&Context cx);
    void drop();
};

struct Poll<int> EchoConn::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 2);
    while (1) {
        self.rd = self.c.read(self.buf[0..512]);
        AWAIT_INTO(1, self.n, self.rd);
        if (self.n <= 0LL) break;                       // EOF or error
        self.wr = self.c.write_all(self.buf[0..(unsigned long)self.n]);
        AWAIT_INTO(2, self.ok, self.wr);
        if (self.ok == 0) break;
    }
    unsafe { self.c.close(); }    // early, in place: drop() then closes nothing again
    ASYNC_RETURN(0);
    ASYNC_END;
}

void EchoConn::drop() {
    if (self.st == 1) self.rd.drop();
    if (self.st == 2) self.wr.drop();
    self.c.close();
    self.st = ASYNC_DONE;
}

struct Server {
    int st; int n; struct AsyncTcpListener l; struct AcceptFut acc; struct AsyncTcpStream c;
    struct AsyncTask<struct EchoConn, int> conns[16];
    struct Poll<int> poll(&Context cx);
    void drop();
};

struct Poll<int> Server::poll(&Context cx) {
    ASYNC_BEGIN(self.st, 1);
    for (self.n = 0; self.n < 16; self.n = self.n + 1) {
        self.acc = self.l.accept();
        AWAIT_INTO(1, self.c, self.acc);
        if (self.c.fd < 0) break;
        {
            struct EchoConn e;
            e.st = ASYNC_INIT;
            e.c = std::replace(&self.c, AsyncTcpStream::from_fd(-1));   // move the stream out
            self.conns[self.n] = AsyncTask::new(e);
            drop(async_spawn(cx, &self.conns[self.n]));  // the handler lives in the server's struct
        }
    }
    ASYNC_RETURN(self.n);
    ASYNC_END;
}

void Server::drop() {
    if (self.st == 1) self.acc.drop();
    self.c.drop();
    for (int i = 0; i < self.n; i = i + 1) self.conns[i].drop();
    self.l.close();
    self.st = ASYNC_DONE;
}

static struct Runtime rt;
static struct AsyncTask<struct Server, int> server;

int main() {
    rt = Runtime::new();
    if (rt.init() != 0) return 1;
    struct Server s;
    s.st = ASYNC_INIT;
    s.n = 0;
    s.l = AsyncTcpListener::bind((unsigned short)7000);
    server = AsyncTask::new(s);
    drop(rt.spawn(&server));      // the JoinHandle: not awaited here
    rt.run();
    rt.close_();
    return 0;
}