Add an option to control sleep interval between calls to read_poll

[tomjridge]

The default is as before: 0.1 s sleep between calls to read_poll. A value of 0.0 skips sleeping altogether. This is possibly useful if events are being lost - a lower sleep may result in fewer events being lost.

[sadiqj]

I think being able to control the sleep is a good idea but could we go further and have a simple control loop that tunes it adaptively?

[tomjridge]

I'm not familiar with adaptive tuning, but I assume it is something like:

1. Check the number of events read from the most recent poll (n, say)
2. With two bounds (hi and lo): If n>= hi, assume the rate of production is high, and don't sleep (ie sleep for 0s); if n<= lo, sleep for 0.1s, say; if lo < n < hi, sleep proportionately between 0 and 0.1s

Something like this?

[sadiqj]

Yes. Another way to think about it is that we're aiming to sleep for some amount of time that means we expect to read N events when we do poll.

If we poll and we read fewer than N events then increase our interval (up to hi), if we read fewer then decrease (down to lo).

Events are a little bursty because they tend to all happen during GCs, so N probably wants to be reasonably big (hundreds, maybe thousand?).

[tomjridge]

Some sample code. There's lots of possible choices here. I've guessed at what might be good choices. But I'm worried that the approach feels very ad hoc. What do you think?

let clamp lo hi (x:float) =
  match () with
  | _ when x < lo -> lo
  | _ when x > hi -> hi
  | _ -> x

let next_sleep_dur ~n_target ~lo_dur ~hi_dur () = 
  assert(lo_dur < hi_dur);
  (* delta is the amount we increment or decrement the duration; currently 10% of the
     target range *)
  let delta = (hi_dur -. lo_dur) *. 0.1 in 
  fun current_sleep_dur (n_read:int) ->
    if n_read > n_target then
      (* we want to decrease sleep duration if possible; we are worried about losing events
         so we immediately drop to the lowest possible duration *)
      lo_dur
    else if n_read < n_target / 2 then
      (* if we are reading substantially less events than our target, gradually increase
         sleep, clamped between lo_dur and hi_dur *)
      clamp lo_dur hi_dur (current_sleep_dur +. delta)
    else current_sleep_dur

[sadiqj]

That looks like it could work. Have you tested it on a couple of examples?