Fix multiple issues in periodic scheduler

- Remove signal parameter from add_to_queue. The parameter was false
  only to support an internal usage, external users should always
  alert the thread loop;
- Do not use ">" or other operators to compare Mtime.t, the value
  is intended to be unsigned and should be compared with
  Int64.unsigned_compare as Mtime functions do;
- Limit mutex contention in add_to_queue;
- Protect queue with mutex in remove_from_queue;
- Do not wait huge amount of time if the queue is empty but
  use Delay.wait if possible;
- Fix delete of periodic events. In case the event is processed
  it's removed from the queue. Previously remove_from_queue was
  not able to mark this event as removed;
- Do not race between checking the first event and removing it.
  These 2 actions were done in 2 separate critical section, now
  they are done in a single one.

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

Author: freddy77

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

@@ -27,6 +27,8 @@ let delay = Delay.make ()
 
 let queue_default = {func= (fun () -> ()); ty= OneShot; name= ""}
 
+let (pending_event : t option ref) = ref None
+
 let (queue : t Ipq.t) = Ipq.create 50 queue_default
 
 let lock = Mutex.create ()
@@ -48,48 +50,71 @@ module Clock = struct
         Mtime.min_stamp
 end
 
-let add_to_queue ?(signal = true) name ty start newfunc =
-  with_lock lock (fun () ->
-      let ( ++ ) = Clock.add_span in
-      Ipq.add queue
-        {
-          Ipq.ev= {func= newfunc; ty; name}
-        ; Ipq.time= Mtime_clock.now () ++ start
-        }
-  ) ;
-  if signal then Delay.signal delay
+let add_to_queue name ty start newfunc =
+  let ( ++ ) = Clock.add_span in
+  let item =
+    {Ipq.ev= {func= newfunc; ty; name}; Ipq.time= Mtime_clock.now () ++ start}
+  in
+  with_lock lock (fun () -> Ipq.add queue item) ;
+  Delay.signal delay
 
 let remove_from_queue name =
-  let index = Ipq.find_p queue (fun {name= n; _} -> name = n) in
-  if index > -1 then
-    Ipq.remove queue index
+  with_lock lock @@ fun () ->
+  match !pending_event with
+  | Some ev when ev.name = name ->
+      pending_event := None
+  | Some _ | None ->
+      let index = Ipq.find_p queue (fun {name= n; _} -> name = n) in
+      if index > -1 then
+        Ipq.remove queue index
+
+let add_periodic_pending () =
+  with_lock lock @@ fun () ->
+  match !pending_event with
+  | Some ev ->
+      ( match ev.ty with
+      | Periodic timer ->
+          let ( ++ ) = Clock.add_span in
+          let item = {Ipq.ev; Ipq.time= Mtime_clock.now () ++ timer} in
+          Ipq.add queue item
+      | OneShot ->
+          ()
+      ) ;
+      pending_event := None
+  | None ->
+      ()
 
 let loop () =
   debug "%s started" __MODULE__ ;
   try
     while true do
-      let empty = with_lock lock (fun () -> Ipq.is_empty queue) in
-      if empty then
-        Thread.delay 10.0
-      (* Doesn't happen often - the queue isn't usually empty *)
-      else
-        let next = with_lock lock (fun () -> Ipq.maximum queue) in
-        let now = Mtime_clock.now () in
-        if next.Ipq.time < now then (
-          let todo =
-            (with_lock lock (fun () -> Ipq.pop_maximum queue)).Ipq.ev
-          in
+      let now = Mtime_clock.now () in
+      let deadline, item =
+        with_lock lock @@ fun () ->
+        (* empty: wait till we get something *)
+        if Ipq.is_empty queue then
+          (Clock.add_span now 20.0, None)
+        else
+          let next = Ipq.maximum queue in
+          if Mtime.is_later next.Ipq.time ~than:now then
+            (* not expired: wait till time or interrupted *)
+            (next.Ipq.time, None)
+          else (
+            (* remove expired item *)
+            Ipq.pop_maximum queue |> ignore ;
+            (* save periodic to be scheduled again *)
+            if next.Ipq.ev.ty <> OneShot then pending_event := Some next.Ipq.ev ;
+            (now, Some next.Ipq.ev)
+          )
+      in
+      match item with
+      | Some todo ->
           (try todo.func () with _ -> ()) ;
-          match todo.ty with
-          | OneShot ->
-              ()
-          | Periodic timer ->
-              add_to_queue ~signal:false todo.name todo.ty timer todo.func
-        ) else (* Sleep until next event. *)
+          add_periodic_pending ()
+      | None -> (
+          (* Sleep until next event. *)
           let sleep =
-            Mtime.(span next.Ipq.time now)
-            |> Mtime.Span.(add ms)
-            |> Clock.span_to_s
+            Mtime.(span deadline now) |> Mtime.Span.(add ms) |> Clock.span_to_s
           in
           try ignore (Delay.wait delay sleep)
           with e ->
@@ -105,6 +130,7 @@ let loop () =
                normal delay. New events may be missed."
               detailed_msg ;
             Thread.delay sleep
+        )
     done
   with _ ->
     error

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

@@ -18,8 +18,7 @@ type func_ty =
   | OneShot  (** Fire just once *)
   | Periodic of float  (** Fire periodically with a given period in seconds *)
 
-val add_to_queue :
-  ?signal:bool -> string -> func_ty -> float -> (unit -> unit) -> unit
+val add_to_queue : string -> func_ty -> float -> (unit -> unit) -> unit
 (** Start a new timer. *)
 
 val remove_from_queue : string -> unit