refactor runtime

src/examples/tcp_ping_pong/main.mbt

@@ -35,13 +35,15 @@ async fn server(println : Printer) -> Unit raise {
           defer conn.close()
           let buf = FixedArray::make(1024, b'0')
           while conn.recv(buf) is n && n > 0 {
+            @async.sleep(10)
             let msg = buf.unsafe_reinterpret_as_bytes()[0:n]
             println("server received: \{msg}")
             let reply = b"pong"
             conn.send(reply)
             println("server sent: \{reply}")
             if msg == b"exit" {
               println("server initiate terminate")
+              @async.sleep(20)
               raise ServerTerminate
             }
           } else {
@@ -72,6 +74,7 @@ async fn client(println : Printer, id : Int, msg : Bytes) -> Unit raise {
   println("client \{id} sent: \{msg}")
   let buf = FixedArray::make(1024, b'0')
   if conn.recv(buf) is n && n > 0 {
+    @async.sleep(10)
     let msg = buf.unsafe_reinterpret_as_bytes()[0:n]
     println("client \{id} received: \{msg}")
   } else {

src/examples/tcp_ping_pong/main_test.mbt

@@ -72,6 +72,7 @@ test "main" {
       #|server received: b"\x65\x78\x69\x74"
       #|server sent: b"\x70\x6f\x6e\x67"
       #|server initiate terminate
+      #|client 6 received: b"\x70\x6f\x6e\x67"
       #|server terminate: Cancelled
       #|Err(ServerTerminate)
     ),

src/examples/udp_ping_pong/main.mbt

@@ -30,6 +30,7 @@ async fn server(println : Printer) -> Unit raise {
     let buf = FixedArray::make(1024, b'0')
     while server_sock.recvfrom(buf) is (n, addr) && n > 0 {
       group.spawn_bg(fn() {
+        @async.sleep(10)
         let msg = buf.unsafe_reinterpret_as_bytes()[0:n]
         println("server received: \{msg}")
         let reply = b"pong"
@@ -60,6 +61,7 @@ async fn client(println : Printer, id : Int, msg : Bytes) -> Unit raise {
   println("client \{id} sent: \{msg}")
   let buf = FixedArray::make(1024, b'0')
   if conn.recv(buf) is n && n > 0 {
+    @async.sleep(10)
     let msg = buf.unsafe_reinterpret_as_bytes()[0:n]
     println("client \{id} received: \{msg}")
     conn.close()

src/fs/dir.mbt

@@ -14,8 +14,7 @@
 
 ///|
 /// A directory in file system
-#external
-type Directory
+struct Directory(@event_loop.Directory)
 
 ///|
 extern "C" fn Directory::close_ffi(self : Directory) -> Int = "closedir"
@@ -43,8 +42,7 @@ pub fn opendir(path : Bytes) -> Directory raise {
 }
 
 ///|
-#external
-priv type DirectoryEntry
+typealias @event_loop.DirectoryEntry
 
 ///|
 extern "C" fn get_null_dirent() -> DirectoryEntry = "moonbitlang_async_null_dirent"
@@ -55,17 +53,10 @@ let null_dirent : DirectoryEntry = get_null_dirent()
 ///|
 extern "C" fn DirectoryEntry::name(ent : DirectoryEntry) -> Bytes = "moonbitlang_async_dirent_name"
 
-///|
-extern "C" fn readdir_job(
-  dir : Directory,
-  out : Ref[DirectoryEntry],
-) -> @thread_pool.Job = "moonbitlang_async_make_readdir_job"
-
 ///|
 async fn Directory::read_next(dir : Directory) -> Bytes? raise {
   let out = @ref.new(null_dirent)
-  let job = readdir_job(dir, out)
-  ignore(@event_loop.perform_job(job))
+  ignore(@event_loop.perform_job(Readdir(dir=dir.0, out~)))
   if physical_equal(out.val, null_dirent) {
     None
   } else {

src/fs/fs.mbt

@@ -89,7 +89,7 @@ pub async fn open(
     Some(mode) => (true, mode)
   }
   let flags = make_open_flags(mode~, sync~, append~, create~, truncate~)
-  let job = @thread_pool.open_job(filename, flags, user_mode)
+  let job = @event_loop.Open(path=filename, flags~, mode=user_mode)
   File(@event_loop.perform_job(job))
 }
 
@@ -129,8 +129,7 @@ pub async fn File::read(
   len~ : Int = buf.length() - offset,
 ) -> Int raise {
   let File(fd) = self
-  let job = @thread_pool.read_job(fd, buf, offset, len)
-  @event_loop.perform_job(job)
+  @event_loop.perform_job(Read(fd~, buf~, offset~, len~))
 }
 
 ///|
@@ -147,14 +146,13 @@ pub async fn File::write(
   len~ : Int = buf.length() - offset,
 ) -> Unit raise {
   let File(fd) = self
-  let job = @thread_pool.write_job(fd, buf, offset, len)
-  ignore(@event_loop.perform_job(job))
+  ignore(@event_loop.perform_job(Write(fd~, buf~, offset~, len~)))
 }
 
 ///|
 /// Remove the file located at `path`. `path` must not contain `'\0'`.
 pub async fn remove(path : Bytes) -> Unit raise {
-  let job = @thread_pool.remove_job(path)
+  let job = @event_loop.Remove(path~)
   ignore(@event_loop.perform_job(job))
 }
 

src/fs/moon.pkg.json

@@ -1,8 +1,7 @@
 {
   "import": [
     "moonbitlang/async/os_error",
-    "moonbitlang/async/internal/event_loop",
-    "moonbitlang/async/internal/thread_pool"
+    "moonbitlang/async/internal/event_loop"
   ],
   "test-import": [ "moonbitlang/async" ],
   "native-stub": [ "stub.c" ]