simon/moonpool
1
0
Fork 0
mirror of https://github.com/c-cube/moonpool.git synced 2025-12-06 03:05:30 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

lwt: handle fibers in moonpool_lwt

Browse source
This commit is contained in:
Simon Cruanes 2025-09-05 10:23:06 -04:00
parent 00078d8b43
commit 9e814ecb48
No known key found for this signature in database
GPG key ID: EBFFF6F283F3A2B4
2 changed files with 6 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/lwt/dune
View file

@ -5,6 +5,7 @@
(>= %{ocaml_version} 5.0)) (>= %{ocaml_version} 5.0))
(libraries (libraries
(re_export moonpool) (re_export moonpool)
moonpool.fib
picos picos
(re_export lwt) (re_export lwt)
lwt.unix)) lwt.unix))

9
src/lwt/moonpool_lwt.ml
View file

@ -49,9 +49,9 @@ module Scheduler_state = struct
let[@inline never] add_action_from_another_thread_ (self : st) f : unit = let[@inline never] add_action_from_another_thread_ (self : st) f : unit =
Mutex.lock self.mutex; Mutex.lock self.mutex;
Queue.push f self.actions_from_other_threads; Queue.push f self.actions_from_other_threads;
Mutex.unlock self.mutex;
if not (Atomic.exchange self.has_notified true) then if not (Atomic.exchange self.has_notified true) then
Lwt_unix.send_notification self.notification; Lwt_unix.send_notification self.notification
Mutex.unlock self.mutex
let[@inline] on_lwt_thread_ (self : st) : bool = let[@inline] on_lwt_thread_ (self : st) : bool =
Thread.id (Thread.self ()) = self.thread Thread.id (Thread.self ()) = self.thread
@ -73,9 +73,10 @@ module Scheduler_state = struct
failwith "moonpool-lwt: cleanup from the wrong thread"; failwith "moonpool-lwt: cleanup from the wrong thread";
Option.iter Lwt_main.Enter_iter_hooks.remove st.enter_hook; Option.iter Lwt_main.Enter_iter_hooks.remove st.enter_hook;
Option.iter Lwt_main.Leave_iter_hooks.remove st.leave_hook; Option.iter Lwt_main.Leave_iter_hooks.remove st.leave_hook;
Lwt_unix.stop_notification st.notification;
Atomic.set cur_st None Atomic.set cur_st None
| _ -> () | None -> failwith "moonpool-lwt: cleanup failed (no current active state)"
end end
module Ops = struct module Ops = struct
@ -290,7 +291,7 @@ let[@inline] is_setup () = Option.is_some @@ Atomic.get Scheduler_state.cur_st
let spawn_lwt f : _ Lwt.t = let spawn_lwt f : _ Lwt.t =
let st = Main_state.get_st () in let st = Main_state.get_st () in
let lwt_fut, lwt_prom = Lwt.wait () in let lwt_fut, lwt_prom = Lwt.wait () in
M.Runner.run_async st.as_runner (fun () -> Moonpool_fib.spawn_top_ignore ~on:st.as_runner (fun () ->
try try
let x = f () in let x = f () in
Lwt.wakeup lwt_prom x Lwt.wakeup lwt_prom x