Remove possible systematic leak in Imperative priority queue

The queue is implemented with an Array larger than the content.
Instead of copying an initial value taken from first insert or potentially later
when the Array is expanded provide a "default" value to use as a filler.
This allows to provide a value not having references to external object so
not extending their lifetime in an unpredictable way.

Signed-off-by: Frediano Ziglio <[email protected]>

ocaml/libs/xapi-stdext/lib/xapi-stdext-threads/ipq.ml

@@ -15,15 +15,16 @@
 
 type 'a event = {ev: 'a; time: Mtime.t}
 
-type 'a t = {mutable size: int; mutable data: 'a event array}
+type 'a t = {default: 'a event; mutable size: int; mutable data: 'a event array}
 
 exception EmptyHeap
 
-let create n =
+let create n default =
   if n <= 0 then
     invalid_arg "create"
   else
-    {size= -n; data= [||]}
+    let default = {ev= default; time= Mtime_clock.now ()} in
+    {default; size= -n; data= [||]}
 
 let is_empty h = h.size <= 0
 
@@ -32,14 +33,14 @@ let resize h =
   assert (n > 0) ;
   let n' = 2 * n in
   let d = h.data in
-  let d' = Array.make n' d.(0) in
+  let d' = Array.make n' h.default in
   Array.blit d 0 d' 0 n ;
   h.data <- d'
 
 let add h x =
   (* first addition: we allocate the array *)
   if h.size < 0 then (
-    h.data <- Array.make (-h.size) x ;
+    h.data <- Array.make (-h.size) h.default ;
     h.size <- 0
   ) ;
   let n = h.size in
@@ -69,6 +70,7 @@ let remove h s =
   let n = h.size - 1 in
   let d = h.data in
   let x = d.(n) in
+  d.(n) <- h.default ;
   (* moving [x] up in the heap *)
   let rec moveup i =
     let fi = (i - 1) / 2 in
@@ -134,11 +136,13 @@ let iter f h =
     f d.(i)
   done
 
+(*
 let fold f h x0 =
   let n = h.size in
   let d = h.data in
   let rec foldrec x i = if i >= n then x else foldrec (f d.(i) x) (succ i) in
   foldrec x0 0
+*)
 
 (*
 let _ =

ocaml/libs/xapi-stdext/lib/xapi-stdext-threads/ipq.mli

@@ -0,0 +1,55 @@
+(*
+ * Copyright (C) 2024 Cloud Software Group
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published
+ * by the Free Software Foundation; version 2.1 only. with the special
+ * exception on linking described in file LICENSE.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *)
+
+type 'a event = {ev: 'a; time: Mtime.t}
+
+type 'a t
+
+exception EmptyHeap
+
+val create : int -> 'a -> 'a t
+(** [create n default] creates an empty Imperative priority queue.
+   The queue initially is initialized to store [n] elements.
+   The queue will expand beyond [n] automatically if needed.
+   [default] value will the used to fill unused data. *)
+
+val is_empty : 'a t -> bool
+(** Check if the queue is empty *)
+
+val add : 'a t -> 'a event -> unit
+(** Add an event to the queue *)
+
+val remove : 'a t -> int -> unit
+(** Remove an event from the queue passing the index.
+   @raise EmptyHeap if the queue is empty.
+   @raise Invalid_argument if the index is invalid. *)
+
+val find_p : 'a t -> ('a -> bool) -> int
+(** Find the index of an event which matches a given condition
+   or -1 if not found *)
+
+val find : 'a t -> 'a -> int
+(** Find the index of an event which matches a given event
+   or -1 if not found *)
+
+val maximum : 'a t -> 'a event
+(** Return a copy of the event with the next time.
+   @raise EmptyHeap if the queue is empty. *)
+
+val pop_maximum : 'a t -> 'a event
+(** Return and remove the event with the next time.
+   @raise EmptyHeap if the queue is empty. *)
+
+val iter : ('a event -> unit) -> 'a t -> unit
+(** Iterate given function on the list of events in the queue *)

ocaml/libs/xapi-stdext/lib/xapi-stdext-threads/ipq_test.ml

@@ -16,7 +16,7 @@ module Ipq = Xapi_stdext_threads_scheduler.Ipq
 
 (* test we get "out of bound" exception calling Ipq.remove *)
 let test_out_of_index () =
-  let q = Ipq.create 10 in
+  let q = Ipq.create 10 0 in
   Ipq.add q {Ipq.ev= 123; Ipq.time= Mtime_clock.now ()} ;
   let is_oob = function
     | Invalid_argument s when String.ends_with ~suffix:"  out of bounds" s ->
@@ -35,6 +35,25 @@ let test_out_of_index () =
   (* this should succeed *)
   Ipq.remove q 0
 
-let tests = [("test_out_of_index", `Quick, test_out_of_index)]
+(* check queue does not retain some data after being removed *)
+let test_leak () =
+  let default () = () in
+  let q = Ipq.create 10 default in
+  let array = Array.make 1024 'x' in
+  let use_array () = array.(0) <- 'a' in
+  let allocated = Atomic.make true in
+  Gc.finalise (fun _ -> Atomic.set allocated false) array ;
+  Ipq.add q {Ipq.ev= use_array; Ipq.time= Mtime_clock.now ()} ;
+  Ipq.remove q 0 ;
+  Gc.full_major () ;
+  Gc.full_major () ;
+  Alcotest.(check bool) "allocated" false (Atomic.get allocated) ;
+  Ipq.add q {Ipq.ev= default; Ipq.time= Mtime_clock.now ()}
+
+let tests =
+  [
+    ("test_out_of_index", `Quick, test_out_of_index)
+  ; ("test_leak", `Quick, test_leak)
+  ]
 
 let () = Alcotest.run "Ipq" [("generic", tests)]

ocaml/libs/xapi-stdext/lib/xapi-stdext-threads/scheduler.ml

@@ -25,7 +25,9 @@ type t = {func: unit -> unit; ty: func_ty; name: string}
 
 let delay = Delay.make ()
 
-let (queue : t Ipq.t) = Ipq.create 50
+let queue_default = {func= (fun () -> ()); ty= OneShot; name= ""}
+
+let (queue : t Ipq.t) = Ipq.create 50 queue_default
 
 let lock = Mutex.create ()
 

quality-gate.sh

@@ -25,7 +25,7 @@ verify-cert () {
 }
 
 mli-files () {
-  N=498
+  N=497
   # do not count ml files from the tests in ocaml/{tests/perftest/quicktest}
   MLIS=$(git ls-files -- '**/*.mli' | grep -vE "ocaml/tests|ocaml/perftest|ocaml/quicktest|ocaml/message-switch/core_test" | xargs -I {} sh -c "echo {} | cut -f 1 -d '.'" \;)
   MLS=$(git  ls-files -- '**/*.ml'  | grep -vE "ocaml/tests|ocaml/perftest|ocaml/quicktest|ocaml/message-switch/core_test" | xargs -I {} sh -c "echo {} | cut -f 1 -d '.'" \;)