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/dir_test.mbt

@@ -23,7 +23,7 @@ async test "read_all" {
   @json.inspect(list, content=[
     "fs.mbt", "stub.c", "dir.mbt", "README.md", "utils.mbt", "dir_test.mbt", "eof_test.mbt",
     "README.mbt.md", "constants.mbt", "moon.pkg.json", "stat_test.mbt", "walk_test.mbt",
-    "mkdir_test.mbt", "access_test.mbt", "create_test.mbt", "seek_wbtest.mbt", "read_all_test.mbt",
+    "mkdir_test.mbt", "access_test.mbt", "create_test.mbt", "read_all_test.mbt",
     "realpath_test.mbt", "unimplemented.mbt", "pkg.generated.mbti", "text_file_test.mbt",
     "timestamp_test.mbt", "random_access_test.mbt",
   ])
@@ -55,8 +55,8 @@ async test "as_dir" {
     "utils.mbt: Regular", "dir_test.mbt: Regular", "eof_test.mbt: Regular", "README.mbt.md: Regular",
     "constants.mbt: Regular", "moon.pkg.json: Regular", "stat_test.mbt: Regular",
     "walk_test.mbt: Regular", "mkdir_test.mbt: Regular", "access_test.mbt: Regular",
-    "create_test.mbt: Regular", "seek_wbtest.mbt: Regular", "read_all_test.mbt: Regular",
-    "realpath_test.mbt: Regular", "unimplemented.mbt: Regular", "pkg.generated.mbti: Regular",
-    "text_file_test.mbt: Regular", "timestamp_test.mbt: Regular", "random_access_test.mbt: Regular",
+    "create_test.mbt: Regular", "read_all_test.mbt: Regular", "realpath_test.mbt: Regular",
+    "unimplemented.mbt: Regular", "pkg.generated.mbti: Regular", "text_file_test.mbt: Regular",
+    "timestamp_test.mbt: Regular", "random_access_test.mbt: Regular",
   ])
 }

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.
@@ -281,9 +249,6 @@ pub async fn File::size(self : File) -> Int64 {
 ///|
 extern "C" fn as_dir_ffi(fd : Int) -> Directory = "fdopendir"
 
-///|
-extern "C" fn Directory::is_null(dir : Directory) -> Bool = "moonbitlang_async_dir_is_null"
-
 ///|
 /// Convert a file to directory.
 /// If the file is not a directory, an error will be raised.
@@ -295,7 +260,7 @@ extern "C" fn Directory::is_null(dir : Directory) -> Bool = "moonbitlang_async_d
 pub fn File::as_dir(self : File) -> Directory raise {
   let fd = self.io.detach_from_event_loop()
   let dir = as_dir_ffi(fd)
-  if dir.is_null() {
+  if @c_buffer.ptr_is_null(dir) {
     let context = "@fs.File::as_dir()"
     @fd_util.close(fd, context~)
     @os_error.check_errno(context)
@@ -406,6 +371,10 @@ pub async fn symlink(target~ : StringView, path : StringView) -> Unit {
   @event_loop.symlink(target, path, context="@fs.link()")
 }
 
+///|
+#borrow(path)
+extern "C" fn chmod_ffi(path : Bytes, mode : Int) -> Int = "moonbitlang_async_chmod"
+
 ///|
 /// Change the permission of a file.
 /// Permission is represented as an integer in UNIX permission style.
@@ -414,5 +383,18 @@ pub async fn symlink(target~ : StringView, path : StringView) -> Unit {
 /// - users in the owner group of the file can read the file (`4`)
 /// - other users can do nothing to the file
 pub async fn chmod(path : StringView, mode : Int) -> Unit {
-  @event_loop.chmod(path, mode, context="@fs.chmod()")
+  struct Job {
+    path : Bytes
+    mode : Int
+    mut err : Int
+  }
+  let path_bytes = @encoding/utf8.encode(path)
+  let job : Job = { path: path_bytes, mode, err: 0 }
+  @event_loop.perform_job_in_worker(job, job => if chmod_ffi(job.path, job.mode) <
+    0 {
+    job.err = @os_error.get_errno()
+  })
+  if job.err != 0 {
+    raise @os_error.OSError(job.err, context="@fs.chmod(\{repr(path)})")
+  }
 }

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

@@ -23,21 +23,13 @@
 #include <sys/stat.h>
 #include <moonbit.h>
 
-int moonbitlang_async_dir_is_null(DIR *dir) {
-  return dir == 0;
-}
-
 moonbit_bytes_t moonbitlang_async_dirent_name(struct dirent *dirent) {
   int len = strlen(dirent->d_name);
   moonbit_bytes_t result = moonbit_make_bytes(len, 0);
   memcpy(result, dirent->d_name, len);
   return result;
 }
 
-int moonbitlang_async_dirent_is_null(struct dirent *dirent) {
-  return dirent == 0;
-}
-
 int moonbitlang_async_get_R_OK() {
   return R_OK;
 }
@@ -85,3 +77,7 @@ int moonbitlang_async_get_O_TRUNC() {
 int moonbitlang_async_get_O_CREAT() {
   return O_CREAT;
 }
+
+int moonbitlang_async_chmod(char *path, int32_t mode) {
+  return chmod(path, mode);
+}

src/fs/stub.c

@@ -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;
+}