API to deduplicate task queue

ChangeLog.md

@@ -1,3 +1,7 @@
+# v0.8.0 - 2/12/2025
+
+- New parameter `~undup` to `Sel.now` to avoid duplicate events
+
 # v0.7.0 - 13/06/2025
 
 - New API Sel.On.promise (fullfillment or rejection)

lib/events.ml

@@ -128,7 +128,7 @@ let make_event ?(priority=0) ?name it =
 type 'a event_ =
   | SystemEvent of 'a system_event
   | QueueEvent of 'a queue_event
-  | Task of 'a task_event
+  | Task of ('a -> 'a -> bool) option * 'a task_event [@print fun f fmt (_,x) -> pp_task_event f fmt x]
   [@@deriving show]
 
 let map_system_event f = function
@@ -143,7 +143,7 @@ let map_queue_event f = function
 let map_task_event f x = f x
 
 let map f = function
-  | Task e -> Task (map_task_event f e)
+  | Task(_,e) -> Task (None, map_task_event f e)
   | SystemEvent e -> SystemEvent (map_system_event f e)
   | QueueEvent e -> QueueEvent(map_queue_event f e)  
 
@@ -248,8 +248,8 @@ let promise ?priority ?name (_,r as p : 'a Promise.t) (k : 'a Promise.state -> '
 
 end
 
-let now ?priority ?name task : 'a Event.t =
-  make_event ?priority ?name @@ Task task
+let now ?priority ?name ?undup task : 'a Event.t =
+  make_event ?priority ?name @@ Task (undup,task)
 
 let advance_queue _ _ = function
   | WaitQueue1(q1,_) as x when Queue.is_empty q1 -> (), No x

lib/sel.ml

@@ -26,8 +26,8 @@ let partition_events l =
         partition_events ((drop_event_type x e,e.priority) :: sys) queue task rest
     | { WithAttributes.it = QueueEvent x; _ } as e :: rest ->
         partition_events sys ((drop_event_type x e,e.priority) :: queue) task rest
-    | { WithAttributes.it = Task x; _ } as e :: rest ->
-        partition_events sys queue ((drop_event_type x e,e.priority) :: task) rest
+    | { WithAttributes.it = Task (u,x); _ } as e :: rest ->
+        partition_events sys queue ((Option.map (fun f x y -> f x.WithAttributes.it y.WithAttributes.it) u, drop_event_type x e, e.priority) :: task) rest
   in
     partition_events [] [] [] l
 
@@ -80,11 +80,11 @@ let add { system; queue; tasks; ready } l =
     { e with priority } in
   let l = List.map tick l in
   let new_sys, new_queue, new_tasks = partition_events l in
-  {
+  { 
     ready;
     system = Sorted.append system (Sorted.of_list new_sys);
-    queue = Sorted.append queue (Sorted.of_list new_queue);
-    tasks = Sorted.append tasks (Sorted.of_list new_tasks);
+    queue  = Sorted.append queue (Sorted.of_list new_queue);
+    tasks  = Sorted.append_uniq tasks new_tasks;
   }
 
 end

lib/sel.mli

@@ -174,10 +174,13 @@ end
             loop (Todo.add todo es)
       in
         loop (Todo.add Todo.empty [now (Fold((+),0,[1;2;3])); ...])
+
+      The [~undup] comparison test, when provided, is used to filter out
+      duplicates from the list of tasks.
     ]}
 
 *)
-val now : ?priority:int -> ?name:string -> 'a -> 'a Event.t
+val now : ?priority:int -> ?name:string -> ?undup:('a -> 'a -> bool) -> 'a -> 'a Event.t
 
 (** Set of events being waited for *)
 module Todo : sig

lib/sorted.ml

@@ -21,8 +21,6 @@ let cmp_priority { user = u1; insertion = i1} {user=u2; insertion=i2} =
 
 let max_priority = { user = max_int; insertion = max_int }
 
-let min_priority p1 p2 = if cmp_priority p1 p2 < 0 then p1 else p2
-
 let default_priority = { user = 0; insertion = 0 }
 
 let eq_user { user = u1; _} {user = u2; _} = u1 = u2
@@ -83,15 +81,18 @@ let min l =
   | [] -> max_priority, l
   | (_,p) :: _ -> p, l
 
+
 let cons x p l =
   match l.data with
   | (_,q) :: _ when l.sorted && lt_priority p q -> { sorted = true; data = (x,p) :: l.data }
   | _ -> { sorted =false; data = (x,p) :: l.data }
-let cons_opt = function
-  | Some(x,p) -> cons x p
-  | None -> fun x -> x
-
 let append l1 l2 = { sorted = false; data = l1.data @ l2.data }
+
+let rec append_uniq l1 = function
+  | [] -> l1
+  | (None,x,p) :: xs -> append_uniq (append l1 { sorted = false; data = [x,p] }) xs
+  | (Some f,x,p) :: xs -> append_uniq (append (filter (fun a -> not(f x a)) l1) { sorted = false; data = [x,p] }) xs
+
 let concat l = { sorted = false; data = List.concat (List.map (fun x -> x.data) l) }
 
 let of_list l = { sorted = false; data = l }

lib/sorted.mli

@@ -37,13 +37,13 @@ type 'a view =
 
 val look : 'a t -> 'a view
 val cons : 'a -> priority -> 'a t -> 'a t
-val cons_opt : ('a * priority) option -> 'a t -> 'a t
 val nil : 'a t
 val is_nil : 'a t -> bool
 val length : 'a t -> int
 val filter : ('a -> bool) -> 'a t -> 'a t
 val for_all : ('a -> bool) -> 'a t -> bool
 val append : 'a t -> 'a t -> 'a t
+val append_uniq : 'a t -> (('a -> 'a -> bool) option * 'a * priority) list -> 'a t
 val concat : 'a t list -> 'a t
 val map_filter : ('a -> 'b option) -> 'a t -> 'b list
 val min : 'a t -> priority * 'a t

sel.opam

@@ -33,3 +33,4 @@ build: [
   ]
 ]
 dev-repo: "git+https://github.com/gares/sel.git"
+x-maintenance-intent: ["(latest)"]

test/perf.ml

@@ -32,7 +32,7 @@ let pipe () =
 (*****************************************************************************)
 
 let %test_unit "sel.event.http_cle.perf" =
-  Caml.Gc.compact ();
+  Stdlib.Gc.compact ();
   let read, write = pipe () in
   let e = On.httpcle read b2s in
   let t0 = Unix.gettimeofday () in
@@ -48,7 +48,7 @@ let %test_unit "sel.event.http_cle.perf" =
   Stdlib.Printf.eprintf "time to pop %d httpcle events: %f\n" n (t1 -. t0)
 ;;
 let%test_unit "sel.event.promise.perf" =
-  Caml.Gc.compact ();
+  Stdlib.Gc.compact ();
   let many = 30_000 in (* Loc of Pfff.v *)
   let x0 = Unix.gettimeofday () in
   for i = 1 to many do
@@ -68,7 +68,7 @@ let%test_unit "sel.event.promise.perf" =
   Stdlib.Printf.eprintf "resolving %d promises: %4.3f\n" many (x1 -. x0);
 ;;
 let%test_unit "sel.event.promise.thread.spawn.perf" =
-  Caml.Gc.compact ();
+  Stdlib.Gc.compact ();
   let many = 30_000 in (* Loc of Pfff.v *)
   let x0 = Unix.gettimeofday () in
   for i = 1 to many do
@@ -89,31 +89,31 @@ let%test_unit "sel.event.promise.thread.spawn.perf" =
   Stdlib.Printf.eprintf "resolving %d promises via thread spawning: %4.3f\n" many (x1 -. x0);
 ;;
 let%test_unit "sel.event.promise.thread.synchronization.perf" =
-  Caml.Gc.compact ();
+  Stdlib.Gc.compact ();
   let many = 30_000 in (* Loc of Pfff.v *)
 
-  let c = Condition.create () in
-  let m = Mutex.create () in
+  let c = Stdlib.Condition.create () in
+  let m = Stdlib.Mutex.create () in
   let x = ref None in
   let t = Thread.create (fun () ->
     let stop = ref false in
     while not !stop do
-      Mutex.lock m;
-      while match !x with None -> true | _ -> false do Condition.wait c m; done;
+      Stdlib.Mutex.lock m;
+      while match !x with None -> true | _ -> false do Stdlib.Condition.wait c m; done;
       match !x with
       | None -> assert false
-      | Some (p,i) -> Promise.fulfill p i; x := None; Condition.signal c; Mutex.unlock m; stop := i = many
+      | Some (p,i) -> Promise.fulfill p i; x := None; Stdlib.Condition.signal c; Stdlib.Mutex.unlock m; stop := i = many
     done) () in
 
   let x0 = Unix.gettimeofday () in
   for i = 1 to many do
     let p, r = Promise.make () in
     let todo = Todo.add Todo.empty [On.promise p (fun e -> e)] in
 
-    Mutex.lock m;
-    while match !x with None -> false | _ -> true do Condition.wait c m; done;
+    Stdlib.Mutex.lock m;
+    while match !x with None -> false | _ -> true do Stdlib.Condition.wait c m; done;
     x := Some (r,i);
-    Condition.signal c; Mutex.unlock m;
+    Stdlib.Condition.signal c; Stdlib.Mutex.unlock m;
 
     let ready, todo = pop todo in
     let n =

test/test1.ml

@@ -271,6 +271,14 @@ let %test_unit "sel.event.http_cle.recurring.err" =
   [%test_eq: string option] ready (Some "4\n6.");
 ;;
 
+let %test_unit "sel.event.now.nodup" =
+  let e1 = now ~priority:1 ~undup:(=) 1 in
+  let e2 = now ~priority:1 1 in
+  let todo = Todo.add Todo.empty [e1;e2] in
+  let ready, todo = wait todo in
+  [%test_eq: bool] (Todo.is_empty todo) true;
+  [%test_eq: int list] ready [1];
+;;  
 
 let %test_unit "sel.event.now.order0" =
   let e1 = now ~priority:1 1 in