From 1067ed3bd4192c989528f7e28640253202952359 Mon Sep 17 00:00:00 2001 From: Julien Sagot Date: Wed, 21 May 2025 16:21:49 +0200 Subject: [PATCH 1/2] Bees: fix async_lwt worker initialization --- src/lib_bees/hive.ml | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/lib_bees/hive.ml b/src/lib_bees/hive.ml index 70ea185b866b..5fb172cafc83 100644 --- a/src/lib_bees/hive.ml +++ b/src/lib_bees/hive.ml @@ -39,14 +39,15 @@ let () = let lwt_scheduler_loop () = let rec loop () : [`Stop_daemon] = let lwt_closure = Eio.Stream.take hive.lwt_tasks_stream in - Lwt_eio.run_lwt_in_main lwt_closure ; + (* The loop will run in the [Event_loop] main domain, so [Eio.run_lwt] is + fine. *) + Lwt_eio.run_lwt lwt_closure ; loop () in loop () in - Tezos_base_unix.Event_loop.on_main_run (fun env switch -> - Eio.Fiber.fork_daemon ~sw:switch (fun () -> - Eio.Domain_manager.run env#domain_mgr lwt_scheduler_loop)) + Tezos_base_unix.Event_loop.on_main_run (fun _env switch -> + Eio.Fiber.fork_daemon ~sw:switch lwt_scheduler_loop) let async_lwt = Eio.Stream.add hive.lwt_tasks_stream -- GitLab From 3993a5556f778d6245717bfd980a29db99aeac1e Mon Sep 17 00:00:00 2001 From: Julien Sagot Date: Thu, 22 May 2025 10:57:21 +0200 Subject: [PATCH 2/2] Bees: enhance tests with Events.emit and Tezos_bees.Hive.async_lwt --- src/lib_bees/test/test_bees_task_worker.ml | 38 ++++++++++++++++++++-- src/lib_bees/test/test_bees_unit.ml | 17 +++++++++- src/lib_bees/test/test_bees_unit_eio.ml | 19 ++++++++++- 3 files changed, 69 insertions(+), 5 deletions(-) diff --git a/src/lib_bees/test/test_bees_task_worker.ml b/src/lib_bees/test/test_bees_task_worker.ml index 58249fa6d366..d077cf985a76 100644 --- a/src/lib_bees/test/test_bees_task_worker.ml +++ b/src/lib_bees/test/test_bees_task_worker.ml @@ -23,10 +23,32 @@ let tztest label fn = | Unix.WEXITED 0 -> return_unit | _ -> Lwt.return_error []) +module Events = struct + let section = ["test_bees_task_worker"] + + include Internal_event.Simple + + let request_received = + declare_0 + ~section + ~name:"request_received" + ~msg:"request received" + ~level:Notice + () + + let emit event param = Tezos_bees.Hive.async_lwt (fun () -> emit event param) +end + +let emit = Events.(emit request_received) + let tests_fibonacci = let test_fibonacci domains = let fib () = let rec fib n = if n <= 1 then n else fib (n - 1) + fib (n - 2) in + let fib n = + emit () ; + fib n + in let input = Stdlib.List.init 5 (fun i -> i + 10) in let expected = List.map fib input in let output = @@ -48,13 +70,23 @@ let tests_reuse = let int_input = 0 in let int_expected = succ int_input in let int_output = - Tezos_bees.Task_worker.launch_task_and_wait "succ(int)" succ int_input + Tezos_bees.Task_worker.launch_task_and_wait + "succ(int)" + (fun i -> + emit () ; + succ i) + int_input in let str_input = "0" in let succ s = int_of_string s |> succ |> string_of_int in let str_expected = succ str_input in let str_output = - Tezos_bees.Task_worker.launch_task_and_wait "succ(str)" succ str_input + Tezos_bees.Task_worker.launch_task_and_wait + "succ(str)" + (fun i -> + emit () ; + succ i) + str_input in let int_output = Eio.Promise.await int_output in let str_output = Eio.Promise.await str_output in @@ -68,7 +100,7 @@ let tests_on_completion_callback = let test = tztest "on_completion_handler" @@ fun () -> let r = ref 0 in - let noop () = () in + let noop () = emit () in let _ = Tezos_bees.Task_worker.launch_task_and_wait "callback" noop () in Assert.equal !r 0 ; let on_completion () = incr r in diff --git a/src/lib_bees/test/test_bees_unit.ml b/src/lib_bees/test/test_bees_unit.ml index 581487d57ea3..49c911028c99 100644 --- a/src/lib_bees/test/test_bees_unit.ml +++ b/src/lib_bees/test/test_bees_unit.ml @@ -33,7 +33,21 @@ module Assert = Assert open Mocked_worker -module Event = Internal_event.Simple + +module Events = struct + let section = ["test_bees_unit"] + + include Internal_event.Simple + + let request_received = + declare_1 + ~section + ~name:"request_received" + ~msg:"request {req} received" + ~level:Notice + ~pp1:Request.pp + ("req", Request.encoding) +end type error += TzCrashError @@ -50,6 +64,7 @@ let create_handlers (type a) ?on_completion ?(slow = false) () = fun _w request -> let open Lwt_result_syntax in let*! () = if slow then Lwt_unix.sleep 0.2 else Lwt.return_unit in + let*! () = Events.(emit request_received) (Request.view request) in match request with | Request.RqA _i -> (return_unit : (r, request_error) result Lwt.t) | Request.RqB -> return_unit diff --git a/src/lib_bees/test/test_bees_unit_eio.ml b/src/lib_bees/test/test_bees_unit_eio.ml index c2fc40b471e0..e9633933e5b4 100644 --- a/src/lib_bees/test/test_bees_unit_eio.ml +++ b/src/lib_bees/test/test_bees_unit_eio.ml @@ -15,7 +15,23 @@ module Assert = Assert open Mocked_worker -module Event = Internal_event.Simple + +module Events = struct + let section = ["test_bees_unit_eio"] + + include Internal_event.Simple + + let request_received = + declare_1 + ~section + ~name:"request_received" + ~msg:"request {req} received" + ~level:Notice + ~pp1:Request.pp + ("req", Request.encoding) + + let emit event param = Tezos_bees.Hive.async_lwt (fun () -> emit event param) +end type error += TzCrashError @@ -34,6 +50,7 @@ let create_handlers (type a) ?on_completion ?(slow = false) () = self -> (r, request_error) Request.t -> (r, request_error) result = fun _w request -> let () = if slow then sleep 0.2 else () in + let () = Events.(emit request_received) (Request.view request) in match request with | Request.RqA _i -> (Ok () : (r, request_error) result) | Request.RqB -> Ok () -- GitLab