refactor thread pool, make it extensible

Author: Guest0x0

src/fs/dir.mbt

@@ -44,7 +44,12 @@ pub async fn rmdir(path : StringView, recursive? : Bool = false) -> Unit {
 
 ///|
 /// A directory in file system
-struct Directory(@event_loop.Directory)
+#external
+type Directory
+
+///|
+#external
+priv type DirectoryEntry
 
 ///|
 extern "C" fn Directory::close_ffi(self : Directory) -> Int = "closedir"
@@ -54,29 +59,68 @@ pub fn Directory::close(self : Directory) -> Unit {
   guard self.close_ffi() == 0
 }
 
+///|
+#borrow(path)
+extern "C" fn opendir_ffi(path : Bytes) -> Directory = "opendir"
+
+///|
+async fn opendir_aux(path : StringView, context~ : String) -> Directory {
+  struct Job {
+    path : Bytes
+    mut result : Directory
+    mut err : Int
+  }
+  let path_bytes = @encoding/utf8.encode(path)
+  let job : Job = { path: path_bytes, result: @c_buffer.null_ptr(), err: 0 }
+  @event_loop.perform_job_in_worker(job, job => {
+    job.result = opendir_ffi(job.path)
+    job.err = @os_error.get_errno()
+  }) catch {
+    err => {
+      if !@c_buffer.ptr_is_null(job.result) {
+        ignore(job.result.close_ffi())
+      }
+      raise err
+    }
+  }
+  if @c_buffer.ptr_is_null(job.result) {
+    raise @os_error.OSError(job.err, context="\{context}: \{repr(path)}")
+  }
+  job.result
+}
+
 ///|
 /// Open the directory at `path`. `path` is encoded UTF8.
 /// If `path` is not a directory, an error will be raised
 pub async fn opendir(path : StringView) -> Directory {
-  @event_loop.opendir(path, context="@fs.opendir()")
+  opendir_aux(path, context="@fs.opendir()")
 }
 
-///|
-using @event_loop {type DirectoryEntry}
-
 ///|
 extern "C" fn DirectoryEntry::name(ent : DirectoryEntry) -> Bytes = "moonbitlang_async_dirent_name"
 
 ///|
-extern "C" fn DirectoryEntry::is_null(ent : DirectoryEntry) -> Bool = "moonbitlang_async_dirent_is_null"
+extern "C" fn readdir_ffi(dir : Directory) -> DirectoryEntry = "readdir"
 
 ///|
 async fn Directory::read_next(dir : Directory) -> String? {
-  let dir_ent = @event_loop.readdir(dir.0, context="@fs.readdir()")
-  if dir_ent.is_null() {
+  struct Job {
+    dir : Directory
+    mut result : DirectoryEntry
+    mut err : Int
+  }
+  let job : Job = { dir, result: @c_buffer.null_ptr(), err: 0 }
+  @event_loop.perform_job_in_worker(job, job => {
+    @os_error.clear_errno()
+    job.result = readdir_ffi(job.dir)
+    job.err = @os_error.get_errno()
+  })
+  if job.err != 0 {
+    raise @os_error.OSError(job.err, context="@fs.readdir()")
+  } else if @c_buffer.ptr_is_null(job.result) {
     None
   } else {
-    Some(@encoding/utf8.decode(dir_ent.name()))
+    Some(@encoding/utf8.decode(job.result.name()))
   }
 }
 
@@ -119,7 +163,7 @@ pub async fn readdir(
   include_special? : Bool = false,
   sort? : Bool = false,
 ) -> Array[String] {
-  let dir : Directory = @event_loop.opendir(path, context="@fs.readdir()")
+  let dir : Directory = opendir_aux(path, context="@fs.readdir()")
   defer dir.close()
   let list = dir.read_all(include_hidden~, include_special~)
   if sort {

src/fs/fs.mbt

@@ -239,38 +239,6 @@ pub async fn remove(path : StringView) -> Unit {
   @event_loop.remove(path, context="@fs.remove()")
 }
 
-///|
-/// Determine how an offset is interpreted when seeking in a file:
-/// - `FromStart`: absolute offset from the start of the file
-/// - `FromEnd`: offset is relative to end of file
-/// - `Relative`: offset is relative to current position in the file
-pub(all) enum SeekMode {
-  FromStart = 0
-  FromEnd
-  Relative
-}
-
-///|
-fn SeekMode::to_int(self : SeekMode) -> Int = "%identity"
-
-///|
-/// Change current position in file for reading and writing.
-/// Can only be applied to a regular file, otherwise `seek` will fail.
-/// The offset is interpreted using `mode`, see `SeekMode` for more detail.
-/// Current position in the file after seeking (relative to start of file)
-/// will be returned.
-#deprecated("use `read_at` or `write_at` instead", skip_current_package=true)
-pub async fn File::seek(self : File, offset : Int64, mode~ : SeekMode) -> Int64 {
-  self.io.seek(offset, mode.to_int(), context="@fs.File::seek()")
-}
-
-///|
-/// Get current position in the file. Can only be applied to a regular file.
-#deprecated("use `read_at` or `write_at` instead", skip_current_package=true)
-pub async fn File::curr_pos(self : File) -> Int64 {
-  self.seek(0, mode=Relative)
-}
-
 ///|
 /// Get the size of the file. This method will not change position in the file.
 /// Can only be applied to a regular file.

src/fs/moon.pkg.json

@@ -4,6 +4,7 @@
     "moonbitlang/async/io",
     "moonbitlang/async/internal/event_loop",
     "moonbitlang/async/internal/fd_util",
+    "moonbitlang/async/internal/c_buffer",
     "moonbitlang/async",
     "moonbitlang/async/semaphore"
   ],

src/fs/pkg.generated.mbti

@@ -66,15 +66,11 @@ pub fn File::as_dir(Self) -> Directory raise
 pub async fn File::atime(Self) -> (Int64, Int)
 pub fn File::close(Self) -> Unit
 pub async fn File::ctime(Self) -> (Int64, Int)
-#deprecated
-pub async fn File::curr_pos(Self) -> Int64
 pub fn File::fd(Self) -> Int
 pub fn File::kind(Self) -> FileKind
 pub async fn File::mtime(Self) -> (Int64, Int)
 pub async fn File::read_at(Self, FixedArray[Byte], position~ : Int64, offset? : Int, len? : Int) -> Int
 pub async fn File::read_exactly_at(Self, Int, position~ : Int64) -> Bytes
-#deprecated
-pub async fn File::seek(Self, Int64, mode~ : SeekMode) -> Int64
 pub async fn File::size(Self) -> Int64
 pub async fn File::sync(Self, only_data? : Bool) -> Unit
 pub async fn File::write_at(Self, BytesView, position~ : Int64) -> Unit
@@ -100,12 +96,6 @@ pub(all) enum Mode {
   ReadWrite
 }
 
-pub(all) enum SeekMode {
-  FromStart
-  FromEnd
-  Relative
-}
-
 pub(all) enum SyncMode {
   NoSync
   Data

src/fs/seek_wbtest.mbt

@@ -48,5 +48,26 @@ extern "C" fn Buffer::null() -> Buffer = "moonbitlang_async_null_pointer"
 ///|
 pub let null : Buffer = Buffer::null()
 
+///|
+#borrow(buf)
+pub extern "C" fn Buffer::is_null(buf : Buffer) -> Bool = "moonbitlang_async_is_null"
+
 ///|
 pub extern "C" fn Buffer::free(buf : Buffer) = "free"
+
+///|
+fn[X, Y] unsafe_cast(x : X) -> Y = "%identity"
+
+///|
+/// Get a null C pointer.
+/// WARNING: the result type MUST be a foreign pointer type declared with `#external`
+pub fn[X] null_ptr() -> X {
+  unsafe_cast(null)
+}
+
+///|
+/// Check if a C pointer is null.
+/// WARNING: the parameter MUST be a foreign pointer type declared with `#external`
+pub fn[X] ptr_is_null(ptr : X) -> Bool {
+  Buffer::is_null(unsafe_cast(ptr))
+}

src/internal/c_buffer/c_buffer.mbt

@@ -4,6 +4,10 @@ package "moonbitlang/async/internal/c_buffer"
 // Values
 pub let null : Buffer
 
+pub fn[X] null_ptr() -> X
+
+pub fn[X] ptr_is_null(X) -> Bool
+
 // Errors
 
 // Types and methods
@@ -14,6 +18,7 @@ pub fn Buffer::blit_to_bytes(Self, dst~ : FixedArray[Byte], offset~ : Int, len~
 pub fn Buffer::free(Self) -> Unit
 #alias("_[_]")
 pub fn Buffer::get(Self, Int) -> Byte
+pub fn Buffer::is_null(Self) -> Bool
 pub fn Buffer::strlen(Self) -> Int
 
 // Type aliases

src/internal/c_buffer/pkg.generated.mbti

@@ -37,3 +37,7 @@ int32_t moonbitlang_async_strlen(char *str) {
 char *moonbitlang_async_null_pointer() {
   return 0;
 }
+
+int32_t moonbitlang_async_is_null(void *ptr) {
+  return ptr == 0;
+}