refactor: split off domain pool to moonpool.dpool

src/core/dune

@@ -1,9 +1,9 @@
 (library
  (public_name moonpool)
  (name moonpool)
- (libraries moonpool.private)
+ (libraries moonpool.private moonpool.dpool)
  (flags :standard -open Moonpool_private)
- (private_modules types_ domain_pool_ util_pool_)
+ (private_modules types_ util_pool_)
  (preprocess
   (action
    (run %{project_root}/src/cpp/cpp.exe %{input-file}))))

src/core/moonpool.ml

@@ -1,3 +1,5 @@
+open Types_
+
 exception Shutdown = Runner.Shutdown
 
 let start_thread_on_some_domain f x =

src/core/types_.ml

@@ -1,4 +1,5 @@
 module TLS = Thread_local_storage_
+module Domain_pool_ = Moonpool_dpool
 
 type ls_value = ..
 

src/core/util_pool_.ml

@@ -1,5 +1,5 @@
 let num_threads ?num_threads () : int =
-  let n_domains = Domain_pool_.n_domains () in
+  let n_domains = Moonpool_dpool.n_domains () in
 
   (* number of threads to run *)
   let num_threads =

src/dpool/dune

@@ -0,0 +1,11 @@
+
+(library
+ (name moonpool_dpool)
+ (public_name moonpool.dpool)
+ (synopsis "Moonpool's domain pool (used to start worker threads)")
+ (preprocess
+  (action
+   (run %{project_root}/src/cpp/cpp.exe %{input-file})))
+ (flags :standard -open Moonpool_private)
+ (libraries
+   moonpool.private))

src/dpool/moonpool_dpool.ml

@@ -1,3 +1,66 @@
+module Bb_queue = struct
+  type 'a t = {
+    mutex: Mutex.t;
+    cond: Condition.t;
+    q: 'a Queue.t;
+  }
+
+  let create () : _ t =
+    { mutex = Mutex.create (); cond = Condition.create (); q = Queue.create () }
+
+  let push (self : _ t) x : unit =
+    Mutex.lock self.mutex;
+    let was_empty = Queue.is_empty self.q in
+    Queue.push x self.q;
+    if was_empty then Condition.broadcast self.cond;
+    Mutex.unlock self.mutex
+
+  let pop (self : 'a t) : 'a =
+    Mutex.lock self.mutex;
+    let rec loop () =
+      if Queue.is_empty self.q then (
+        Condition.wait self.cond self.mutex;
+        (loop [@tailcall]) ()
+      ) else (
+        let x = Queue.pop self.q in
+        Mutex.unlock self.mutex;
+        x
+      )
+    in
+    loop ()
+end
+
+module Lock = struct
+  type 'a t = {
+    mutex: Mutex.t;
+    mutable content: 'a;
+  }
+
+  let create content : _ t = { mutex = Mutex.create (); content }
+
+  let with_ (self : _ t) f =
+    Mutex.lock self.mutex;
+    try
+      let x = f self.content in
+      Mutex.unlock self.mutex;
+      x
+    with e ->
+      Mutex.unlock self.mutex;
+      raise e
+
+  let[@inline] update_map l f =
+    with_ l (fun x ->
+        let x', y = f x in
+        l.content <- x';
+        y)
+
+  let get l =
+    Mutex.lock l.mutex;
+    let x = l.content in
+    Mutex.unlock l.mutex;
+    x
+end
+
 type domain = Domain_.t
 
 type event =

src/private/dune

@@ -1,7 +1,7 @@
 (library
  (name moonpool_private)
  (public_name moonpool.private)
- (synopsis "Private internal utils for Moonpool")
+ (synopsis "Private internal utils for Moonpool (do not rely on)")
  (preprocess
   (action
    (run %{project_root}/src/cpp/cpp.exe %{input-file})))