notifications now work with an abstract id

raphael-proust · raphael-proust · commit 9a538c518127 · 2025-09-11T11:07:43.000+02:00
abstract id includes domain id for location of the domain
this breaks backwards compat with 5.* less
diff --git a/src/unix/lwt_gc.ml b/src/unix/lwt_gc.ml
@@ -20,18 +20,16 @@ let ensure_termination t =
         (fun () -> Lwt_main.Exit_hooks.remove hook; Lwt.return_unit))
   end
 
-let finaliser f =
-  (* In order for the domain id to be consistent, wherever the real finaliser is
-     called, we pass it in the continuation. *)
-  let domain_id = Domain.self () in
+let finaliser ?domain f =
+  let domain = match domain with None -> Domain.self () | Some domain -> domain in
   (* In order not to create a reference to the value in the
      notification callback, we use an initially unset option cell
      which will be filled when the finaliser is called. *)
   let opt = ref None in
   let id =
     Lwt_unix.make_notification
       ~once:true
-      domain_id
+      domain
       (fun () ->
          match !opt with
          | None ->
@@ -43,10 +41,10 @@ let finaliser f =
   (* The real finaliser: fill the cell and send a notification. *)
   (fun x ->
      opt := Some x;
-     Lwt_unix.send_notification domain_id id)
+     Lwt_unix.send_notification id)
 
-let finalise f x =
-  Gc.finalise (finaliser f) x
+let finalise ?domain f x =
+  Gc.finalise (finaliser ?domain f) x
 
 (* Exit hook for a finalise_or_exit *)
 let foe_exit f called weak () =
diff --git a/src/unix/lwt_gc.mli b/src/unix/lwt_gc.mli
@@ -9,14 +9,25 @@
     thread to a value, without having to use [Lwt_unix.run] in the
     finaliser. *)
 
-val finalise : ('a -> unit Lwt.t) -> 'a -> unit
+val finalise : ?domain:Domain.id -> ('a -> unit Lwt.t) -> 'a -> unit
   (** [finalise f x] ensures [f x] is evaluated after [x] has been
       garbage collected. If [f x] yields, then Lwt will wait for its
       termination at the end of the program.
 
       Note that [f x] is not called at garbage collection time, but
-      later in the main loop. *)
+      later in the main loop.
+
+      If [domain] is provided, then [f x] is evaluated in the corresponding
+      domain. Otherwise it is evaluated in the domain calling [finalise]. If
+      Lwt is not running in the domain set to run the finaliser, an
+      unspecified error occurs at an unspecified time or the finaliser doesn't
+      run or some other bad thing happens. *)
 
 val finalise_or_exit : ('a -> unit Lwt.t) -> 'a -> unit
   (** [finalise_or_exit f x] call [f x] when [x] is garbage collected
-      or (exclusively) when the program exits. *)
+      or (exclusively) when the program exits.
+
+      The finaliser [f] is called in the same domain that called
+      [finalise_or_exit]. If there is no Lwt scheduler running in this domain an
+      unspecified error occurs. You can use [Lwt_preemptive.run_in_domain] to
+      bypass the same-domain limitation. *)
diff --git a/src/unix/lwt_main.ml b/src/unix/lwt_main.ml
@@ -29,7 +29,7 @@ let run p =
       let cbs = (Lwt.Private.Multidomain_sync.get_sent_callbacks[@alert "-trespassing"]) domain_id in
       Lwt_sequence.iter_l (fun f -> f ()) cbs
     ) in
-    (Lwt.Private.Multidomain_sync.register_notification[@alert "-trespassing"]) domain_id (fun () -> Lwt_unix.send_notification domain_id n)
+    (Lwt.Private.Multidomain_sync.register_notification[@alert "-trespassing"]) domain_id (fun () -> Lwt_unix.send_notification n)
   end
   in
   let rec run_loop () =
diff --git a/src/unix/lwt_main.mli b/src/unix/lwt_main.mli
@@ -77,8 +77,14 @@ val abandon_yielded_and_paused : unit -> unit [@@deprecated "Use Lwt.abandon_pau
 
 
 (** Hook sequences. Each module of this type is a set of hooks, to be run by Lwt
-    at certain points during execution. See modules {!Enter_iter_hooks},
-    {!Leave_iter_hooks}, and {!Exit_hooks}. *)
+    at certain points during execution.
+
+    Hooks are added for the current domain. If you are calling the Hook
+    functions from a domain where Lwt is not running a scheduler then some
+    unspecified error may occur. If you need to set some Hooks to/from a
+    different domain, you can use [Lwt_preemptive.run_in_domain].
+
+    See modules {!Enter_iter_hooks}, {!Leave_iter_hooks}, and {!Exit_hooks}. *)
 module type Hooks =
 sig
   type 'return_value kind
diff --git a/src/unix/lwt_preemptive.ml b/src/unix/lwt_preemptive.ml
@@ -78,7 +78,7 @@ struct
 end
 
 type thread = {
-  task_cell: (int * (unit -> unit)) CELL.t;
+  task_cell: (Lwt_unix.notification_id * (unit -> unit)) CELL.t;
   (* Channel used to communicate notification id and tasks to the
      worker thread. *)
 
@@ -104,7 +104,7 @@ let rec worker_loop worker =
      decreased the maximum: *)
   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;
+  Lwt_unix.send_notification id;
   if worker.reuse then worker_loop worker
 
 (* create a new worker: *)
@@ -186,6 +186,7 @@ let detach f args =
   get_worker () >>= fun worker ->
   let waiter, wakener = Lwt.wait () in
   let id =
+    (* call back the domain that called the [detach] function: self *)
     Lwt_unix.make_notification ~once:true (Domain.self ())
       (fun () -> Lwt.wakeup_result wakener !result)
   in
@@ -216,23 +217,27 @@ let jobs = Queue.create ()
 (* Mutex to protect access to [jobs]. *)
 let jobs_mutex = Mutex.create ()
 
-let job_notification =
-  Lwt_unix.make_notification (Domain.self ())
+let job_notification = Domain_map.create_protected_map ()
+let get_job_notification d =
+  Domain_map.init job_notification d
     (fun () ->
-       (* Take the first job. The queue is never empty at this
-          point. *)
-       Mutex.lock jobs_mutex;
-       let thunk = Queue.take jobs in
-       Mutex.unlock jobs_mutex;
-       ignore (thunk ()))
+      Lwt_unix.make_notification (Domain.self ())
+        (fun () ->
+           (* Take the first job. The queue is never empty at this
+              point. *)
+           Mutex.lock jobs_mutex;
+           let thunk = Queue.take jobs in
+           Mutex.unlock jobs_mutex;
+           ignore (thunk ()))
+  )
 
 let run_in_domain_dont_wait d f =
   (* Add the job to the queue. *)
   Mutex.lock jobs_mutex;
   Queue.add f jobs;
   Mutex.unlock jobs_mutex;
   (* Notify the main thread. *)
-  Lwt_unix.send_notification d job_notification
+  Lwt_unix.send_notification (get_job_notification d)
 
 (* There is a potential performance issue from creating a cell every time this
    function is called. See:
diff --git a/src/unix/lwt_unix.cppo.ml b/src/unix/lwt_unix.cppo.ml
@@ -84,6 +84,8 @@ let notifiers = Domain_map.create_protected_map ()
    https://github.com/ocsigen/lwt/pull/278. *)
 let current_notification_id = Atomic.make (0x7FFFFFFF - 1000)
 
+type notification_id = Domain.id * int
+
 let make_notification ?(once=false) domain_id f =
   let id = Atomic.fetch_and_add current_notification_id 1 in
   Domain_map.update notifiers domain_id
@@ -95,17 +97,17 @@ let make_notification ?(once=false) domain_id f =
       | Some notifiers ->
           Notifiers.add notifiers id { notify_once = once; notify_handler = f };
           Some notifiers);
-  id
+  (domain_id, id)
 
-let stop_notification domain_id id =
+let stop_notification (domain_id, id) =
   Domain_map.update notifiers domain_id
     (function
       | None -> None
       | Some notifiers ->
           Notifiers.remove notifiers id;
           Some notifiers)
 
-let set_notification domain_id id f =
+let set_notification (domain_id, id) f =
   Domain_map.update notifiers domain_id
     (function
       | None -> raise Not_found
@@ -114,7 +116,7 @@ let set_notification domain_id id f =
           Notifiers.replace notifiers id { notifier with notify_handler = f };
           Some notifiers)
 
-let call_notification domain_id id =
+let call_notification (domain_id, id) =
   match Domain_map.find notifiers domain_id with
   | None -> ()
   | Some notifiers ->
@@ -209,7 +211,7 @@ let run_job_aux async_method job result =
       jobs in
     ignore begin
       (* Create the notification for asynchronous wakeup. *)
-      let id =
+      let (_, notifid) as id =
         make_notification ~once:true domain_id
           (fun () ->
              Lwt_sequence.remove node;
@@ -220,7 +222,7 @@ let run_job_aux async_method job result =
          notification. *)
       Lwt.pause () >>= fun () ->
       (* The job has terminated, send the result immediately. *)
-      if check_job job id then call_notification domain_id id;
+      if check_job job notifid then call_notification id;
       Lwt.return_unit
     end;
     waiter
@@ -2199,11 +2201,12 @@ let tcflow ch act =
 
 external init_notification : Domain.id -> Unix.file_descr = "lwt_unix_init_notification_stub"
 external send_notification : Domain.id -> int -> unit = "lwt_unix_send_notification_stub"
+let send_notification (d, n) = send_notification d n
 external recv_notifications : Domain.id -> int array = "lwt_unix_recv_notifications_stub"
 
 let handle_notifications (_ : Lwt_engine.event) =
   let domain_id = Domain.self () in
-  Array.iter (call_notification domain_id) (recv_notifications domain_id)
+  Array.iter (fun n -> call_notification (domain_id, n)) (recv_notifications domain_id)
 
 let event_notifications =
   Domain.DLS.new_key (fun () ->
@@ -2247,8 +2250,8 @@ type signal_handler = {
 
 and signal_handler_id = signal_handler option ref
 
-(* TODO: what to do about signals? *)
-let signals = ref Signal_map.empty
+(* TODO: make parallel safe *)
+let signals : ((Domain.id * int) * ((signal_handler_id -> file_perm -> unit) Lwt_sequence.t) ) Signal_map.t ref = ref Signal_map.empty
 let signal_count () =
   Signal_map.fold
     (fun _signum (_id, actions) len -> len + Lwt_sequence.length actions)
@@ -2262,17 +2265,17 @@ let on_signal_full signum handler =
       Signal_map.find signum !signals
     with Not_found ->
       let actions = Lwt_sequence.create () in
-      let notification =
+      let (_, notifid) as notification =
         make_notification (Domain.self ())
           (fun () ->
              Lwt_sequence.iter_l
                (fun f -> f id signum)
                actions)
       in
       (try
-         set_signal signum notification
+         set_signal signum notifid
        with exn when Lwt.Exception_filter.run exn ->
-         stop_notification (Domain.self ()) notification;
+         stop_notification notification;
          raise exn);
       signals := Signal_map.add signum (notification, actions) !signals;
       (notification, actions)
@@ -2294,13 +2297,13 @@ let disable_signal_handler id =
     if Lwt_sequence.is_empty actions then begin
       remove_signal sh.sh_num;
       signals := Signal_map.remove sh.sh_num !signals;
-      stop_notification (Domain.self ()) notification
+      stop_notification notification
     end
 
 let reinstall_signal_handler signum =
   match Signal_map.find signum !signals with
   | exception Not_found -> ()
-  | notification, _ ->
+  | (_, notification), _ ->
     set_signal signum notification
 
 (* +-----------------------------------------------------------------+
diff --git a/src/unix/lwt_unix.cppo.mli b/src/unix/lwt_unix.cppo.mli
@@ -1462,29 +1462,31 @@ val wait_for_jobs : unit -> unit Lwt.t
 (** Lwt internally use a pipe to send notification to the main
     thread. The following functions allow to use this pipe. *)
 
-val make_notification : ?once : bool -> Domain.id -> (unit -> unit) -> int
+type notification_id
+
+val make_notification : ?once : bool -> Domain.id -> (unit -> unit) -> notification_id
   (** [make_notification ?once f] registers a new notifier. It returns the
       id of the notifier. Each time a notification with this id is
       received, [f] is called.
 
       if [once] is specified, then the notification is stopped after
       the first time it is received. It defaults to [false]. *)
 
-val send_notification : Domain.id -> int -> unit
+val send_notification : notification_id -> unit
   (** [send_notification id] sends a notification.
 
       This function is thread-safe. *)
 
-val stop_notification : Domain.id -> int -> unit
+val stop_notification : notification_id -> unit
   (** Stop the given notification. Note that you should not reuse the
       id after the notification has been stopped, the result is
       unspecified if you do so. *)
 
-val call_notification : Domain.id -> int -> unit
+val call_notification : notification_id -> unit
   (** Call the handler associated to the given notification. Note that
       if the notification was defined with [once = true] it is removed. *)
 
-val set_notification : Domain.id -> int -> (unit -> unit) -> unit
+val set_notification : notification_id -> (unit -> unit) -> unit
   (** [set_notification id f] replace the function associated to the
       notification by [f]. It raises [Not_found] if the given
       notification is not found. *)
