split into moonpool.private and moonpool; format; remove DLA

benchs/dune

@@ -1,6 +1,6 @@
-
 (executables
  (names fib_rec pi)
- (preprocess (action
+ (preprocess
+  (action
    (run %{project_root}/src/cpp/cpp.exe %{input-file})))
  (libraries moonpool unix trace trace-tef domainslib))

dune

@@ -1,6 +1,8 @@
-
 (env
-  (_ (flags :standard -strict-sequence -warn-error -a+8 -w +a-4-40-42-70)))
+ (_
+  (flags :standard -strict-sequence -warn-error -a+8 -w +a-4-40-42-70)))
 
-(mdx (libraries moonpool threads)
-     (enabled_if (>= %{ocaml_version} 5.0)))
+(mdx
+ (libraries moonpool threads)
+ (enabled_if
+  (>= %{ocaml_version} 5.0)))

dune-project

@@ -29,8 +29,7 @@
     :with-test)))
  (depopts
    (trace (>= 0.6))
-   thread-local-storage
-   (domain-local-await (>= 0.2)))
+   thread-local-storage)
  (tags
   (thread pool domain futures fork-join)))
 

moonpool.opam

@@ -21,7 +21,6 @@ depends: [
 depopts: [
   "trace" {>= "0.6"}
   "thread-local-storage"
-  "domain-local-await" {>= "0.2"}
 ]
 build: [
   ["dune" "subst"] {dev}

src/core/d_pool_.ml

@@ -33,8 +33,6 @@ let domains_ : (worker_state option * Domain_.t option) Lock.t array =
      in a tight loop), and if nothing happens it tries to stop to free resources.
 *)
 let work_ idx (st : worker_state) : unit =
-  Dla_.setup_domain ();
-
   let main_loop () =
     let continue = ref true in
     while !continue do

src/core/dune

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

src/core/fifo_pool.ml

@@ -60,11 +60,7 @@ let worker_thread_ (self : state) (runner : t) ~on_exn ~around_task : unit =
     done
   in
 
-  try
-    (* handle domain-local await *)
-    Dla_.using ~prepare_for_await:Suspend_.prepare_for_await
-      ~while_running:main_loop
-  with Bb_queue.Closed -> ()
+  try main_loop () with Bb_queue.Closed -> ()
 
 let default_thread_init_exit_ ~dom_id:_ ~t_id:_ () = ()
 

src/core/suspend_.ml

@@ -9,8 +9,6 @@ type suspension_handler = {
 [@@@ifge 5.0]
 [@@@ocaml.alert "-unstable"]
 
-module A = Atomic_
-
 type _ Effect.t += Suspend : suspension_handler -> unit Effect.t
 
 let[@inline] suspend h = Effect.perform (Suspend h)
@@ -35,24 +33,6 @@ let with_suspend ~name ~on_suspend ~(run : name:string -> task -> unit)
 
   E.try_with f () { E.effc }
 
-(* DLA interop *)
-let prepare_for_await () : Dla_.t =
-  (* current state *)
-  let st : (string * (name:string -> task -> unit) * suspension) option A.t =
-    A.make None
-  in
-
-  let release () : unit =
-    match A.exchange st None with
-    | None -> ()
-    | Some (name, run, k) -> run ~name (fun () -> k (Ok ()))
-  and await () : unit =
-    suspend { handle = (fun ~name ~run k -> A.set st (Some (name, run, k))) }
-  in
-
-  let t = { Dla_.release; await } in
-  t
-
 [@@@ocaml.alert "+unstable"]
 [@@@else_]
 

src/core/suspend_.mli

@@ -51,9 +51,6 @@ val suspend : suspension_handler -> unit
 
 [@@@endif]
 
-val prepare_for_await : unit -> Dla_.t
-(** Our stub for DLA. Unstable. *)
-
 val with_suspend :
   name:string ->
   on_suspend:(unit -> unit) ->

src/core/ws_pool.ml

@@ -227,9 +227,7 @@ let worker_thread_ (self : state) ~(runner : t) (w : worker_state) : unit =
            tasks *)
         Mutex.unlock self.mutex
   in
-
-  (* handle domain-local await *)
-  Dla_.using ~prepare_for_await:Suspend_.prepare_for_await ~while_running:main
+  main ()
 
 let default_thread_init_exit_ ~dom_id:_ ~t_id:_ () = ()
 

src/cpp/dune

@@ -2,4 +2,5 @@
 
 (executable
  (name cpp)
- (modes (best exe)))
+ (modes
+  (best exe)))

src/dla_.dummy.ml

@@ -1,13 +0,0 @@
-(** Interface to Domain-local-await.
-
-    This is used to handle the presence or absence of DLA. *)
-
-type t = {
-  release: unit -> unit;
-  await: unit -> unit;
-}
-
-let using : prepare_for_await:(unit -> t) -> while_running:(unit -> 'a) -> 'a =
- fun ~prepare_for_await:_ ~while_running -> while_running ()
-
-let setup_domain () = ()

src/dla_.real.ml

@@ -1,9 +0,0 @@
-type t = Domain_local_await.t = {
-  release: unit -> unit;
-  await: unit -> unit;
-}
-
-let using : prepare_for_await:(unit -> t) -> while_running:(unit -> 'a) -> 'a =
-  Domain_local_await.using
-
-let setup_domain () = Domain_local_await.per_thread (module Thread)

src/dune

@@ -1,17 +0,0 @@
-(library
- (public_name moonpool)
- (name moonpool)
- (private_modules d_pool_ dla_ tracing_)
- (preprocess
-  (action
-   (run %{project_root}/src/cpp/cpp.exe %{input-file})))
- (libraries threads either
-            (select thread_local_storage_.ml from
-             (thread-local-storage -> thread_local_storage_.stub.ml)
-             (-> thread_local_storage_.real.ml))
-            (select tracing_.ml from
-             (trace.core -> tracing_.real.ml)
-             (-> tracing_.dummy.ml))
-            (select dla_.ml from
-             (domain-local-await -> dla_.real.ml)
-             ( -> dla_.dummy.ml))))

src/private/dune

@@ -0,0 +1,20 @@
+(library
+ (name moonpool_private)
+ (public_name moonpool.private)
+ (synopsis "Private internal utils for Moonpool")
+ (preprocess
+  (action
+   (run %{project_root}/src/cpp/cpp.exe %{input-file})))
+ (libraries
+  threads
+  either
+  (select
+   thread_local_storage_.ml
+   from
+   (thread-local-storage -> thread_local_storage_.stub.ml)
+   (-> thread_local_storage_.real.ml))
+  (select
+   tracing_.ml
+   from
+   (trace.core -> tracing_.real.ml)
+   (-> tracing_.dummy.ml))))

src/private/thread_local_storage_.stub.ml

@@ -1,3 +1,2 @@
-
 (* just defer to library *)
 include Thread_local_storage

test/effect-based/dune

@@ -1,11 +1,23 @@
-
 (tests
-  (names t_fib1 t_futs1 t_many t_fib_fork_join
-         t_fib_fork_join_all t_sort t_fork_join t_fork_join_heavy)
-  (preprocess (action
+ (names
+  t_fib1
+  t_futs1
+  t_many
+  t_fib_fork_join
+  t_fib_fork_join_all
+  t_sort
+  t_fork_join
+  t_fork_join_heavy)
+ (preprocess
+  (action
    (run %{project_root}/src/cpp/cpp.exe %{input-file})))
-  (enabled_if (>= %{ocaml_version} 5.0))
-  (libraries moonpool trace trace-tef
-             qcheck-core qcheck-core.runner
-             ;tracy-client.trace
-             ))
+ (enabled_if
+  (>= %{ocaml_version} 5.0))
+ (libraries
+  moonpool
+  trace
+  trace-tef
+  qcheck-core
+  qcheck-core.runner
+  ;tracy-client.trace
+  ))