Implement coroutine_wake_up()

Author: rexim

coroutine.c

@@ -37,6 +37,7 @@
         (da)->items[j] = (da)->items[--(da)->count]; \
     } while(0)
 
+#define UNUSED(x) (void)(x)
 #define TODO(message) do { fprintf(stderr, "%s:%d: TODO: %s\n", __FILE__, __LINE__, message); abort(); } while(0)
 #define UNREACHABLE(message) do { fprintf(stderr, "%s:%d: UNREACHABLE: %s\n", __FILE__, __LINE__, message); abort(); } while(0)
 
@@ -285,3 +286,16 @@ size_t coroutine_alive(void)
 {
     return active.count;
 }
+
+void coroutine_wake_up(size_t id)
+{
+    // @speed coroutine_wake_up is linear
+    for (size_t i = 0; i < asleep.count; ++i) {
+        if (asleep.items[i] == id) {
+            da_remove_unordered(&asleep, id);
+            da_remove_unordered(&polls, id);
+            da_append(&active, id);
+            return;
+        }
+    }
+}

coroutine.c3

@@ -4,8 +4,9 @@ import std::net;
 
 def CoroutineFn = fn void(void*);
 
-extern fn int sleep_read(int fd) @extern("coroutine_sleep_read");
-extern fn int sleep_write(int fd) @extern("coroutine_sleep_write");
+extern fn void sleep_read(int fd) @extern("coroutine_sleep_read");
+extern fn void sleep_write(int fd) @extern("coroutine_sleep_write");
+extern fn void wake_up(usz id) @extern("coroutine_wake_up");
 extern fn void init() @extern("coroutine_init");
 extern fn void finish() @extern("coroutine_finish");
 extern fn void yield() @extern("coroutine_yield");

coroutine.h

@@ -34,14 +34,22 @@ size_t coroutine_id(void);
 // to wait until all the "child" coroutines have died.
 size_t coroutine_alive(void);
 
-// Put the current coroutine to sleep until the non-blocking socket `fd` has avaliable data to read.
-// Trying to read from fd after coroutine_sleep_read() should not cause EAGAIN.
+// Put the current coroutine to sleep until the non-blocking socket `fd` has
+// avaliable data to read. Trying to read from fd after coroutine_sleep_read()
+// may still cause EAGAIN, if the coroutine was woken up by coroutine_wake_up
+// before the socket became available for reading.
 void coroutine_sleep_read(int fd);
 
-// Put the current coroutine to sleep until the non-blocking socket `fd` is ready to accept data to write.
-// Trying to write to fd after coroutine_sleep_write() should not cause EAGAIN.
+// Put the current coroutine to sleep until the non-blocking socket `fd` is
+// ready to accept data to write. Trying to write to fd after
+// coroutine_sleep_write() may still cause EAGAIN, if the coroutine was woken up
+// by coroutine_wake_up before the socket became available for writing.
 void coroutine_sleep_write(int fd);
 
+// Wake up coroutine by id if it is currently sleeping due to
+// coroutine_sleep_read() or coroutine_sleep_write() calls.
+void coroutine_wake_up(size_t id);
+
 // TODO: implement sleeping by timeout
 // TODO: add timeouts to coroutine_sleep_read() and coroutine_sleep_write()
 

examples/echo.c3

@@ -4,51 +4,65 @@ import std::io;
 import std::net;
 import coroutine;
 
+bool quit = false;
+usz server_id = 0;
+
 fn void main() {
     coroutine::init();
+    defer coroutine::finish();
+
+    server_id = coroutine::id();
 
     const String HOST = "localhost";
     const uint PORT = 6969;
-    TcpServerSocket server_sock = tcp::listen(HOST, PORT, 69, REUSEADDR)!!;
-    server_sock.sock.set_non_blocking(true)!!;
-
-    io::printfn("[%d] Listening to %s:%d", coroutine::id(), HOST, PORT);
-    while (true) {
-        coroutine::sleep_read(server_sock.sock);
-        TcpSocket client_sock = tcp::accept(&server_sock)!!;
-        client_sock.sock.set_non_blocking(true)!!;
+    TcpServerSocket server = tcp::listen(HOST, PORT, 69, REUSEADDR)!!;
+    server.sock.set_non_blocking(true)!!;
+
+    io::printfn("[%d] Server listening to %s:%d", coroutine::id(), HOST, PORT);
+    while SERVER: (true) {
+        coroutine::sleep_read(server.sock);
+        if (quit) break SERVER;
+        TcpSocket client = tcp::accept(&server)!!;
+        client.sock.set_non_blocking(true)!!;
         coroutine::go(fn void(void *arg) {
             io::printfn("[%d] Client connected!", coroutine::id());
 
-            TcpSocket* client_sock = (TcpSocket*)arg;
+            TcpSocket* client = (TcpSocket*)arg;
             char[] buf = mem::new_array(char, 1024);
             defer {
-                client_sock.close()!!;
-                free(client_sock);
+                client.close()!!;
+                free(client);
                 free(buf.ptr);
             }
 
             while OUTER: (true) {
-                coroutine::sleep_read(client_sock.sock);
-                usz n = client_sock.read(buf)!!;
+                coroutine::sleep_read(client.sock);
+                usz n = client.read(buf)!!;
                 if (n == 0) break OUTER;
                 char[] chunk = buf[0:n];
 
-                if (((String)chunk).trim() == "quit") {
-                    io::printfn("[%d] Client requested to quit", coroutine::id(), chunk.len);
-                    return;
+                switch (((String)chunk).trim()) {
+                    case "quit":
+                        io::printfn("[%d] Client requested to quit", coroutine::id(), chunk.len);
+                        return;
+                    case "shutdown":
+                        io::printfn("[%d] Client requested to shutdown the server", coroutine::id());
+                        quit = true;
+                        coroutine::wake_up(server_id);
+                        return;
                 }
 
                 io::printfn("[%d] Client sent %d bytes", coroutine::id(), chunk.len);
 
                 while (chunk.len > 0) {
-                    coroutine::sleep_write(client_sock.sock);
-                    usz m = client_sock.write(chunk)!!;
+                    coroutine::sleep_write(client.sock);
+                    usz m = client.write(chunk)!!;
                     if (m == 0) break OUTER;
                     chunk = chunk[m..];
                 }
             }
             io::printfn("[%d] Client disconnected", coroutine::id());
-        }, @clone(client_sock));
+        }, @clone(client));
     }
+    io::printfn("[%d] Server has been shutdown", coroutine::id());
 }