From d795a45ea67d7ce48455e13f1e3867b5734d9938 Mon Sep 17 00:00:00 2001 From: Enrico Tassi Date: Wed, 12 Feb 2025 15:47:00 +0100 Subject: [PATCH 1/2] API: On.queue_all for batch processing --- lib/events.ml | 14 ++++++++++++++ lib/sel.mli | 5 +++++ test/test1.ml | 17 +++++++++++++++++ 3 files changed, 36 insertions(+) diff --git a/lib/events.ml b/lib/events.ml index 9bd1dc8..ea9f9cc 100644 --- a/lib/events.ml +++ b/lib/events.ml @@ -48,10 +48,12 @@ let pp_system_event _ fmt = function type 'a queue_event = | WaitQueue1 : 'b Queue.t * ('b -> 'a) -> 'a queue_event + | WaitQueueBatch1 : 'b Queue.t * ('b -> 'b list -> 'a) -> 'a queue_event | WaitQueue2 : 'b Queue.t * 'c Queue.t * ('b -> 'c -> 'a) -> 'a queue_event let pp_queue_event _ fmt = function | WaitQueue1 _ -> Stdlib.Format.fprintf fmt "WaitQueue1" + | WaitQueueBatch1 _ -> Stdlib.Format.fprintf fmt "WaitQueueBatch1" | WaitQueue2 _ -> Stdlib.Format.fprintf fmt "WaitQueue2" type 'a task_event = 'a [@@deriving show] @@ -96,6 +98,7 @@ let map_system_event f = function let map_queue_event f = function | WaitQueue1(q1,k) -> WaitQueue1(q1,(fun x -> f (k x))) + | WaitQueueBatch1(q1,k) -> WaitQueueBatch1(q1,(fun x xs -> f (k x xs))) | WaitQueue2(q1,q2,k) -> WaitQueue2(q1,q2,(fun x y -> f (k x y))) let map_task_event f x = f x @@ -180,6 +183,9 @@ let httpcle ?priority ?name fd k : 'a Event.t = let queue ?priority ?name q1 k : 'a Event.t = make_event ?priority ?name @@ QueueEvent (WaitQueue1(q1,k)) +let queue_all ?priority ?name q1 k : 'a Event.t = + make_event ?priority ?name @@ QueueEvent (WaitQueueBatch1(q1,k)) + let queues ?priority ?name q1 q2 k : 'a Event.t = make_event?priority ?name @@ QueueEvent (WaitQueue2(q1,q2,k)) @@ -191,6 +197,14 @@ let now ?priority ?name task : 'a Event.t = let advance_queue _ _ = function | WaitQueue1(q1,_) as x when Queue.is_empty q1 -> (), No x | WaitQueue1(q1,k) -> (), Yes(k (Queue.pop q1)) + | WaitQueueBatch1(q1,_) as x when Queue.is_empty q1 -> (), No x + | WaitQueueBatch1(q1,k) -> + let hd = Queue.pop q1 in + let tl = ref [] in + while not @@ Queue.is_empty q1 do + tl := Queue.pop q1 :: !tl + done; + (), Yes(k hd (List.rev !tl)) | WaitQueue2(q1,q2,_) as x when Queue.is_empty q1 || Queue.is_empty q2 -> (), No x | WaitQueue2(q1,q2,k) -> (), Yes(k (Queue.pop q1) (Queue.pop q2)) diff --git a/lib/sel.mli b/lib/sel.mli index d5010e5..afe0141 100644 --- a/lib/sel.mli +++ b/lib/sel.mli @@ -107,6 +107,11 @@ module On : sig val queue : ?priority:int -> ?name:string -> 'b Queue.t -> ('b -> 'a) -> 'a Event.t + (** Synchronization events between a component and an event. + The queue is emptied, useful to process the contents in batches *) + val queue_all : ?priority:int -> ?name:string -> + 'b Queue.t -> ('b -> 'b list -> 'a) -> 'a Event.t + end (** mix a regular computations with blocking events. E.g. diff --git a/test/test1.ml b/test/test1.ml index 15d69ae..41ce1c7 100644 --- a/test/test1.ml +++ b/test/test1.ml @@ -141,6 +141,23 @@ let%test_unit "sel.event.queue" = [%test_eq: bool] (Todo.is_empty todo) true; ;; +(* queue does not run unless something is pushed *) +let%test_unit "sel.event.queue" = + let q = Stdlib.Queue.create () in + let todo = Todo.add Todo.empty [On.queue_all q (fun () l -> l)] in + (* no progress since the queue is empty *) + let _ready, todo = wait_timeout todo in + (* progress since the queue has a token *) + Stdlib.Queue.push () q; + Stdlib.Queue.push () q; + Stdlib.Queue.push () q; + let ready, todo = pop_opt todo in + [%test_eq: bool] (Option.is_none ready) false; + [%test_eq: unit list option] ready (Some [();()]); + [%test_eq: bool] (Todo.is_empty todo) true; + [%test_eq: bool] (Stdlib.Queue.is_empty q) true; +;; + (* queue2 does not advance unless both queues are pushed *) let%test_unit "sel.event.queue2" = let q1 = Stdlib.Queue.create () in From 662d554da37bb3c03b2511a317a42c6282b0e6b1 Mon Sep 17 00:00:00 2001 From: Enrico Tassi Date: Wed, 12 Feb 2025 15:49:22 +0100 Subject: [PATCH 2/2] fix ci --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 53b99fd..cb54ca3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,14 +10,14 @@ jobs: strategy: matrix: os: [ubuntu-latest, windows-latest, macos-latest] - ocaml-compiler: [4.14.x] + ocaml-compiler: [4.14.x, 5.3.x] runs-on: ${{ matrix.os }} steps: - name: Checkout uses: actions/checkout@v3 - name: Use OCaml ${{ matrix.ocaml-compiler }} - uses: ocaml/setup-ocaml@v2 + uses: ocaml/setup-ocaml@v3 with: ocaml-compiler: ${{ matrix.ocaml-compiler }}