CP-49158: Use exponential backoff for delay between recursive calls

This delay was right after we waited for a new event, delaying all event
responses by 50ms (including task completions).
Eliminate the first delay, so that if we find the event we're looking after the
DB update, then we can return immediately.

On spurious wakeups (e.g. not the event we subscribed for) the delay is still
useful, so keep it for recursive calls after the first one, and exponentially
increase it up to the configured maximum.

No feature flag, this is a relatively small change, and we use exponential backoffs
elsewhere in XAPI already.

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

ocaml/xapi-aux/throttle.ml

@@ -43,6 +43,8 @@ end
 module Batching = struct
   type t = {delay_before: float; delay_between: float}
 
+  type arg = float
+
   let make ~delay_before ~delay_between = {delay_before; delay_between}
 
   (** [perform_delay delay] calls {!val:Thread.delay} when [delay] is non-zero.
@@ -56,11 +58,12 @@ module Batching = struct
     if delay > Float.epsilon then
       Thread.delay delay
 
-  let with_recursive config f arg =
+  let with_recursive config f () =
     let rec self arg =
-      perform_delay config.delay_between ;
-      (f [@tailcall]) self arg
+      let arg = Float.min config.delay_between (arg *. 2.) in
+      perform_delay arg ; (f [@tailcall]) self arg
     in
+    let self0 arg = (f [@tailcall]) self arg in
     perform_delay config.delay_before ;
-    f self arg
+    f self0 (config.delay_between /. 16.)
 end

ocaml/xapi-aux/throttle.mli

@@ -27,28 +27,33 @@ module Batching : sig
   (** batching delay configuration *)
   type t
 
+  (** internal argument for recursive calls *)
+  type arg = private float
+
   val make : delay_before:float -> delay_between:float -> t
   (** [make ~delay_before ~delay_between] creates a configuration,
     where we delay the API call by [delay_before] once,
     and then with [delay_between] between each recursive call.
    *)
 
-  val with_recursive : t -> (('a -> 'b) -> 'a -> 'b) -> 'a -> 'b
+  val with_recursive : t -> ((arg -> 'b) -> arg -> 'b) -> unit -> 'b
   (** [with_recursive config f arg] calls [f self arg], where [self] can be used
     for recursive calls.
 
-    A [delay_before] amount of seconds is inserted once, and [delay_between] is inserted between recursive calls:
+    [arg] needs to be passed through unchanged.
+
+    A [delay_before] amount of seconds is inserted once, and [delay_between] is inserted between recursive calls,
+    except the first one:
     {v
       delay_before
       f ...
         self ...
-         delay_between
          f ...
           self ...
           delay_between
           f ...
      v}
 
-    The delays are determined by [config]
+    The delays are determined by [config], and [delay_between] uses an exponential backoff, up to [config.delay_between] delay.
    *)
 end

ocaml/xapi/xapi_event.ml

@@ -492,11 +492,11 @@ let rec next ~__context =
   in
   (* Like grab_range () only guarantees to return a non-empty range by blocking if necessary *)
   let grab_nonempty_range =
-    Throttle.Batching.with_recursive batching @@ fun self () ->
+    Throttle.Batching.with_recursive batching @@ fun self arg ->
     let last_id, end_id = grab_range () in
     if last_id = end_id then
       let (_ : int64) = wait subscription end_id in
-      (self [@tailcall]) ()
+      (self [@tailcall]) arg
     else
       (last_id, end_id)
   in
@@ -603,7 +603,7 @@ let from_inner __context session subs from from_t deadline batching =
   let msg_gen, messages, tableset, (creates, mods, deletes, last) =
     with_call session subs (fun sub ->
         let grab_nonempty_range =
-          Throttle.Batching.with_recursive batching @@ fun self () ->
+          Throttle.Batching.with_recursive batching @@ fun self arg ->
           let ( (msg_gen, messages, _tableset, (creates, mods, deletes, last))
                 as result
               ) =
@@ -622,7 +622,7 @@ let from_inner __context session subs from from_t deadline batching =
             (* last id the client got is equivalent to the current one *)
             last_msg_gen := msg_gen ;
             wait2 sub last deadline ;
-            (self [@tailcall]) ()
+            (self [@tailcall]) arg
           ) else
             result
         in