allow multiple schdulers to run in parallel

.github/workflows/workflow.yml

@@ -15,16 +15,6 @@ jobs:
         os:
           - ubuntu-latest
         ocaml-compiler:
-          - "4.08"
-          - "4.09"
-          - "4.10"
-          - "4.11"
-          - "4.12"
-          - "4.13"
-          - "4.14"
-          - "5.0"
-          - "5.1"
-          - "5.2"
           - "5.3"
         libev:
           - true

CHANGES

@@ -1,3 +1,9 @@
+===== 6.0.0 =====
+
+  * Support multiple scheduler running in parallel in separate domains.
+
+  * Exception filter defaults to letting systems exceptions through.
+
 ===== 5.9.0 =====
 
 ====== Additions ======

dune-project

@@ -57,7 +57,8 @@ a single thread by default. This reduces the need for locks or other
 synchronization primitives. Code can be run in parallel on an opt-in basis.
 ")
  (depends
-  (ocaml (>= 4.08))
+  (ocaml (>= 4.14))
+  domain_shims
   (cppo (and :build (>= 1.1.0)))
   (ocamlfind (and :dev (>= 1.7.3-1)))
   (odoc (and :with-doc (>= 2.3.0)))

lwt.opam

@@ -21,7 +21,8 @@ doc: "https://ocsigen.org/lwt"
 bug-reports: "https://github.com/ocsigen/lwt/issues"
 depends: [
   "dune" {>= "2.7"}
-  "ocaml" {>= "4.08"}
+  "ocaml" {>= "4.14"}
+  "domain_shims"
   "cppo" {build & >= "1.1.0"}
   "ocamlfind" {dev & >= "1.7.3-1"}
   "odoc" {with-doc & >= "2.3.0"}

src/core/domain_map.ml

@@ -0,0 +1,57 @@
+module Domain_map : Map.S with type key = Domain.id = Map.Make(struct
+  type t = Domain.id
+  let compare d1 d2 = Int.compare (d1 : Domain.id :> int) (d2 : Domain.id :> int)
+end)
+
+(* Protected domain map reference with per-reference mutex *)
+type 'a protected_map = {
+  mutex : Mutex.t;
+  mutable map : 'a Domain_map.t;
+}
+
+let create_protected_map () = {
+  mutex = Mutex.create ();
+  map = Domain_map.empty;
+}
+
+let with_lock protected_map f =
+  Mutex.lock protected_map.mutex;
+  Fun.protect f ~finally:(fun () -> Mutex.unlock protected_map.mutex)
+
+let update_map protected_map f =
+  with_lock protected_map (fun () ->
+    let old_map = protected_map.map in
+    let new_map = f old_map in
+    protected_map.map <- new_map)
+
+let add protected_map key value =
+  update_map protected_map (Domain_map.add key value)
+
+let remove protected_map key =
+  update_map protected_map (Domain_map.remove key)
+
+let update protected_map key f =
+  update_map protected_map (Domain_map.update key f)
+
+let find protected_map key =
+  with_lock protected_map (fun () -> Domain_map.find_opt key protected_map.map)
+
+let extract protected_map key =
+  with_lock protected_map (fun () ->
+    match Domain_map.find_opt key protected_map.map with
+    | None -> None
+    | Some v ->
+        protected_map.map <- Domain_map.remove key protected_map.map;
+        Some v)
+
+let size protected_map =
+  with_lock protected_map (fun () -> Domain_map.cardinal protected_map.map)
+
+let init protected_map key init_value =
+  with_lock protected_map (fun () ->
+    match Domain_map.find_opt key protected_map.map with
+    | Some existing -> existing
+    | None ->
+        let new_value = init_value () in
+        protected_map.map <- Domain_map.add key new_value protected_map.map;
+        new_value)

src/core/domain_map.mli

@@ -0,0 +1,38 @@
+(** Domain-indexed maps with thread-safe operations
+
+    Only intended to use internally, not for general release.
+
+    Note that these function use a lock. A single lock.
+    - Probably not optimal
+    - Deadlock if you call one of those functions inside another (e.g., use
+    `init` rather than `find`+`update`
+ *)
+
+(** Thread-safe wrapper for domain maps *)
+type 'a protected_map
+
+(** Create a new protected map with an empty map inside and a dedicated mutex,
+    the map is keyed on domain ids, and operations are synchronised via a mutex.
+    *)
+val create_protected_map : unit -> 'a protected_map
+
+(** Add a key-value binding to the map *)
+val add : 'a protected_map -> Domain.id -> 'a -> unit
+
+(** Remove a key from the map *)
+val remove : 'a protected_map -> Domain.id -> unit
+
+(** Update a binding using the underlying map's update function *)
+val update : 'a protected_map -> Domain.id -> ('a option -> 'a option) -> unit
+
+(** Find a value by key, returning None if not found *)
+val find : 'a protected_map -> Domain.id -> 'a option
+
+(** Find + remove but hit the mutex only once *)
+val extract : 'a protected_map -> Domain.id -> 'a option
+
+(** Get the number of bindings in the map *)
+val size : 'a protected_map -> int
+
+(** Initialize a key with a value if it doesn't exist, return existing or new value *)
+val init : 'a protected_map -> Domain.id -> (unit -> 'a) -> 'a

src/core/dune

@@ -2,6 +2,7 @@
  (public_name lwt)
  (synopsis "Monadic promises and concurrent I/O")
  (wrapped false)
+ (libraries domain_shims)
  (instrumentation
   (backend bisect_ppx)))