CP-49158: Throttle: add Thread.yield

Give an opportunity for more fields to be filled, e.g. when waiting for a task
to complete, give a chance for the task to actually run.

No feature flag, it only changes timing.

Signed-off-by: Edwin Török <[email protected]>

dune-project

@@ -327,6 +327,7 @@
   (synopsis "The toolstack daemon which implements the XenAPI")
   (description "This daemon exposes the XenAPI and is used by clients such as 'xe' and 'XenCenter' to manage clusters of Xen-enabled hosts.")
   (depends
+    (ocaml (>= 4.09))
     (alcotest :with-test)
     angstrom
     astring

ocaml/xapi-aux/throttle.ml

@@ -56,25 +56,31 @@ module Batching = struct
     in
     {delay_initial; delay_before; delay_between}
 
+  let span_min a b = if Mtime.Span.is_shorter a ~than:b then a else b
+
   (** [perform_delay delay] calls {!val:Thread.delay} when [delay] is non-zero.
 
     Thread.delay 0 provides no fairness guarantees, the current thread may actually be the one that gets the global lock again.
     Instead {!val:Thread.yield} could be used, which does provide fairness guarantees, but it may also introduce large latencies
-    when there are lots of threads waiting for the OCaml runtime lock.
+    when there are lots of threads waiting for the OCaml runtime lock. Only invoke this once, in the [delay_before] section.
    *)
-  let perform_delay delay =
-    if Mtime.Span.is_longer delay ~than:Mtime.Span.zero then
+  let perform_delay ~yield delay =
+     if Mtime.Span.is_longer delay ~than:Mtime.Span.zero then
       Thread.delay (Clock.Timer.span_to_s delay)
-
-  let span_min a b = if Mtime.Span.is_shorter a ~than:b then a else b
+    else if yield then
+      (* this is a low-priority thread, if there are any other threads waiting, then run them now.
+         If there are no threads waiting then this a noop.
+         Requires OCaml >= 4.09 (older versions had fairness issues in Thread.yield)
+      *)
+      Thread.yield ()
 
   let with_recursive_loop config f =
     let rec self arg input =
       let arg = span_min config.delay_between Mtime.Span.(2 * arg) in
-      perform_delay arg ;
+      perform_delay ~yield:false arg ;
       (f [@tailcall]) (self arg) input
     in
-    let self0 arg input = (f [@tailcall]) (self arg) input in
-    perform_delay config.delay_before ;
-    f (self0 config.delay_initial)
+    let self0 input = (f [@tailcall]) (self config.delay_initial) input in
+    perform_delay ~yield:true config.delay_before ;
+    f self0
 end

xapi.opam

@@ -10,6 +10,7 @@ homepage: "https://xapi-project.github.io/"
 bug-reports: "https://github.com/xapi-project/xen-api/issues"
 depends: [
   "dune" {>= "3.15"}
+  "ocaml" {>= "4.09"}
   "alcotest" {with-test}
   "angstrom"
   "astring"