more refs become atomics

raphael-proust · raphael-proust · commit ba710d1db20c · 2025-06-24T14:27:18.000+02:00
diff --git a/CHANGES b/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 ======
diff --git a/src/core/lwt.ml b/src/core/lwt.ml
@@ -770,11 +770,9 @@ module Exception_filter = struct
     | Out_of_memory -> false
     | Stack_overflow -> false
     | _ -> true
-  let v =
-    (* Default value: the legacy behaviour to avoid breaking programs *)
-    ref handle_all
-  let set f = v := f
-  let run e = !v e
+  let v = Atomic.make handle_all_except_runtime
+  let set f = Atomic.set v f
+  let run e = (Atomic.get v) e
 end
 
 module Sequence_associated_storage :
@@ -820,11 +818,10 @@ struct
     mutable value : 'v option;
   }
 
-  let next_key_id = ref 0
+  let next_key_id = Atomic.make 0
 
   let new_key () =
-    let id = !next_key_id in
-    next_key_id := id + 1;
+    let id = Atomic.fetch_and_add next_key_id 1 in
     {id = id; value = None}
 
   let current_storage = Domain.DLS.new_key (fun () -> Storage_map.empty)
diff --git a/src/unix/lwt_preemptive.ml b/src/unix/lwt_preemptive.ml
@@ -17,23 +17,23 @@ open Lwt.Infix
    +-----------------------------------------------------------------+ *)
 
 (* Minimum number of preemptive threads: *)
-let min_threads : int ref = ref 0
+let min_threads : int Atomic.t = Atomic.make 0
 
 (* Maximum number of preemptive threads: *)
-let max_threads : int ref = ref 0
+let max_threads : int Atomic.t = Atomic.make 0
 
 (* Size of the waiting queue: *)
-let max_thread_queued = ref 1000
+let max_thread_queued = Atomic.make 1000
 
 let get_max_number_of_threads_queued _ =
-  !max_thread_queued
+  Atomic.get max_thread_queued
 
 let set_max_number_of_threads_queued n =
   if n < 0 then invalid_arg "Lwt_preemptive.set_max_number_of_threads_queued";
-  max_thread_queued := n
+  Atomic.set max_thread_queued n
 
 (* The total number of preemptive threads currently running: *)
-let threads_count = ref 0
+let threads_count = Atomic.make 0
 
 (* +-----------------------------------------------------------------+
    | Preemptive threads management                                   |
@@ -102,14 +102,14 @@ let rec worker_loop worker =
   task ();
   (* If there is too much threads, exit. This can happen if the user
      decreased the maximum: *)
-  if !threads_count > !max_threads then worker.reuse <- false;
+  if Atomic.get threads_count > Atomic.get max_threads then worker.reuse <- false;
   (* Tell the main thread that work is done: *)
   Lwt_unix.send_notification (Domain.self ()) id;
   if worker.reuse then worker_loop worker
 
 (* create a new worker: *)
 let make_worker () =
-  incr threads_count;
+  Atomic.incr threads_count;
   let worker = {
     task_cell = CELL.make ();
     thread = Thread.self ();
@@ -130,7 +130,7 @@ let add_worker worker =
 let get_worker () =
   if not (Queue.is_empty workers) then
     Lwt.return (Queue.take workers)
-  else if !threads_count < !max_threads then
+  else if Atomic.get threads_count < Atomic.get max_threads then
     Lwt.return (make_worker ())
   else
     (Lwt.add_task_r [@ocaml.warning "-3"]) waiters
@@ -139,33 +139,33 @@ let get_worker () =
    | Initialisation, and dynamic parameters reset                    |
    +-----------------------------------------------------------------+ *)
 
-let get_bounds () = (!min_threads, !max_threads)
+let get_bounds () = (Atomic.get min_threads, Atomic.get max_threads)
 
 let set_bounds (min, max) =
   if min < 0 || max < min then invalid_arg "Lwt_preemptive.set_bounds";
-  let diff = min - !threads_count in
-  min_threads := min;
-  max_threads := max;
+  let diff = min - Atomic.get threads_count in
+  Atomic.set min_threads min;
+  Atomic.set max_threads max;
   (* Launch new workers: *)
   for _i = 1 to diff do
     add_worker (make_worker ())
   done
 
-let initialized = ref false
+let initialized = Atomic.make false
 
 let init min max _errlog =
-  initialized := true;
+  Atomic.set initialized true;
   set_bounds (min, max)
 
 let simple_init () =
-  if not !initialized then begin
-    initialized := true;
+  if not (Atomic.get initialized) then begin
+    Atomic.set initialized true;
     set_bounds (0, 4)
   end
 
-let nbthreads () = !threads_count
+let nbthreads () = Atomic.get threads_count
 let nbthreadsqueued () = Lwt_sequence.fold_l (fun _ x -> x + 1) waiters 0
-let nbthreadsbusy () = !threads_count - Queue.length workers
+let nbthreadsbusy () = Atomic.get threads_count - Queue.length workers
 
 (* +-----------------------------------------------------------------+
    | Detaching                                                       |
@@ -199,7 +199,7 @@ let detach f args =
          (* Put back the worker to the pool: *)
          add_worker worker
        else begin
-         decr threads_count;
+         Atomic.decr threads_count;
          (* Or wait for the thread to terminates, to free its associated
             resources: *)
          Thread.join worker.thread
diff --git a/src/unix/lwt_unix.cppo.ml b/src/unix/lwt_unix.cppo.ml
@@ -21,33 +21,33 @@ type async_method =
   | Async_detach
   | Async_switch
 
-let default_async_method_var = ref Async_detach
+let default_async_method_var = Atomic.make Async_detach
 
 let () =
   try
     match Sys.getenv "LWT_ASYNC_METHOD" with
     | "none" ->
-      default_async_method_var := Async_none
+      Atomic.set default_async_method_var Async_none
     | "detach" ->
-      default_async_method_var := Async_detach
+      Atomic.set default_async_method_var Async_detach
     | "switch" ->
-      default_async_method_var := Async_switch
+      Atomic.set default_async_method_var Async_switch
     | str ->
       Printf.eprintf
         "%s: invalid lwt async method: '%s', must be 'none', 'detach' or 'switch'\n%!"
         (Filename.basename Sys.executable_name) str
   with Not_found ->
     ()
 
-let default_async_method () = !default_async_method_var
-let set_default_async_method am = default_async_method_var := am
+let default_async_method () = Atomic.get default_async_method_var
+let set_default_async_method am = Atomic.set default_async_method_var am
 
 let async_method_key = Lwt.new_key ()
 
 let async_method () =
   match Lwt.get async_method_key with
   | Some am -> am
-  | None -> !default_async_method_var
+  | None -> Atomic.get default_async_method_var
 
 let with_async_none f =
   Lwt.with_value async_method_key (Some Async_none) f
@@ -232,7 +232,7 @@ let choose_async_method = function
   | None ->
     match Lwt.get async_method_key with
     | Some am -> am
-    | None -> !default_async_method_var
+    | None -> Atomic.get default_async_method_var
 
 external self_result : 'a job -> 'a = "lwt_unix_self_result"
 (* returns the result of a job using the [result] field of the C
@@ -2260,6 +2260,7 @@ type signal_handler = {
 
 and signal_handler_id = signal_handler option ref
 
+(* TODO: what to do about signals? *)
 let signals = ref Signal_map.empty
 let signal_count () =
   Signal_map.fold
@@ -2375,6 +2376,7 @@ let do_wait4 flags pid =
 let wait_children = Lwt_sequence.create ()
 let wait_count () = Lwt_sequence.length wait_children
 
+(* TODO: what to do about signals? especially sigchld signal? *)
 let sigchld_handler_installed = ref false
 
 let install_sigchld_handler () =
