Lwt: [easy-ish] Fix race condition in Lwt_react.E.limit

[aantron]

Lwt_react.E.limit takes a rate-limiting function and a React event stream, and rate-limits the event stream:

let rate_limited_event_stream =
  Lwt_react.E.limit (fun () -> Lwt_unix.sleep 1.0) event_stream

rate_limited_event_stream is the same as event_stream, but if the event occurs sooner than one second since the last time, the occurrence is buffered, and emitted later, after one second has actually passed.

Only one occurrence is buffered. So, if the event occurs at times 0.00, 0.25, and 0.50, the first occurrence at time 0.00 is passed straight through, the second occurrence at time 0.25 is buffered to be emitted at time 1.00, but is then overwritten at time 0.50 by the third occurrence. Only the third occurrence is finally emitted at time 1.00.

The current implementation has a race condition, however. If event_stream is triggered from inside the rate-limiting function, that occurrence bypasses the buffering logic, and is emitted immediately.

The rest of this post gives code to reproduce the problem, a link to the offending implementation, and a description of what is happening.

Observing the problem

Install packages lwt and lwt_react:

opam install lwt lwt_react

Here is the repro case. It includes a small "control," which is some commented-out code that works correctly, and then code that triggers the bug in this issue. The difference between the two is only what is inside the rate-limiting function.

let () =
  let start_time = Unix.gettimeofday () in
  let print_elapsed_time () =
    Printf.printf "%.1f\n%!" (Unix.gettimeofday () -. start_time)
  in

  let underlying_event, push = Lwt_react.E.create () in



  (* "Control": this works as expected if uncommented, printing 0.0, 1.0: *)
  (*
  underlying_event
  |> Lwt_react.E.limit (fun () ->
    Lwt_unix.sleep 1.)
  |> Lwt_react.E.map print_elapsed_time
  |> ignore;

  push ();
  push (); (* Gets buffered. *)
  *)



  (* ...but we are going to do this instead, which prints 0.0, 1.0, 1.0: *)
  underlying_event
  |> Lwt_react.E.limit (fun () ->
    let limiter = Lwt_unix.sleep 1. in
    Lwt.on_success limiter push;            (* <------- !!!!! *)
    limiter)
  |> Lwt_react.E.map print_elapsed_time
  |> ignore;

  push ();
  push (); (* Gets buffered. *)
  (* If this was working correctly, we would expect to see either

       0.0, 1.0

     or

       0.0, 1.0, 2.0

     depending on whether the buffered occurrence is overwritten. *)



  Lwt_main.run (Lwt_unix.sleep 3.)

You can paste this code into a file in a new directory, then compile and run it with

ocamlfind opt -linkpkg -package lwt_react,lwt.unix race.ml && ./a.out

The offending code

...can be seen here:

https://github.com/ocsigen/lwt/blob/e24f2d7bc697a4f2496926ac7ce263102ea3402e/src/react/lwt_react.ml#L43-L63

The problem is that Lwt_react.E.limit attaches an on_success callback to the rate-limiting promise limiter. However, the repro case above also attaches an on_success callback to limiter, and it does so first, before limiter is even returned to Lwt_react.E.limit. As a result, the on_success callback in the repro case runs first when limiter becomes resolved.

More realistically, the repro case could have had a bind (>>=), or any other Lwt operation, applied to limiter, and it would be processed first.

In this case, that first on_success does a push into the underlying event stream, which breaks the internal logic of Lwt_react.E.limit. Effectively, Lwt_react.E.limit is assuming that its internal on_success callback will run first when limiter resolves, but we are violating that assumption.

In more detail, at the time the problem occurs, the two push calls at the end of the program have run. The first one caused an event to be emitted, and output to be printed to the screen. The second one is buffered inside cell. limiter is a promise that is pending for one second, and then resolves.

At that point, one second later when limiter resolves, the first callback of limiter to run is the on_success callback from our repro case, above. Then, this condition, which is a check for whether limiter is still pending, is already false. So, Lwt_react.E.limit thinks, correctly, that the rate-limiting delay has expired, but incorrectly emits the extra push that the repro case callback is doing. This is because it ignores the fact that there is a buffered push waiting inside cell.

Then, Lwt_react.E.limit's own on_success callback runs, second. This one, instead, emits the buffered push waiting inside cell, but ignores the status of limiter, which is now a fresh, new, one-second promise.

So, the two on_success callbacks trigger behaviors that are wrong in a perfectly complementary way, resulting in too many events emitted.

As a side note, this was originally found while fixing #566.

Working

See CONTRIBUTING.md for instructions on how to set up for Lwt development. You will probably want to reinstall your pinned Lwt from time to time, to compile the repro case against it and see the effect:

opam pin add lwt .
opam reinstall -y lwt

If you have any concerns, please don't hesitate to reach out :)

Thank you! :tada:

[AvniFatehpuria]

I'd like to try fixing this issue!

[aantron]

Great, thanks for taking a look! Feel free to ping if needed :)

[AvniFatehpuria]

I tried compiliing and running the example broken code a few times and it always printed:
0.0
1.0
2.0
which is one of the correct options.
What could be going on?

[aantron]

@AvniFatehpuria Thanks again for looking. Indeed, the posted program doesn't fail anymore. I modified it into this one, which does:

open Lwt.Infix

let () =
  let start_time = Unix.gettimeofday () in
  let print_elapsed_time n =
    Printf.printf "%i %.1f\n%!" n (Unix.gettimeofday () -. start_time)
  in

  let underlying_event, push = Lwt_react.E.create () in

  underlying_event
  |> Lwt_react.E.limit (fun () ->
    let p = Lwt_unix.sleep 1. in
    Lwt.async (fun () ->
      Lwt_unix.sleep 0.1 >|= fun () ->
      Lwt.on_success p (fun () -> push 2));
    p)
  |> Lwt_react.E.map print_elapsed_time
  |> ignore;

  push 0;
  push 1;

  Lwt_main.run (Lwt_unix.sleep 2.5)

(* ocamlfind opt -linkpkg -package lwt_react,lwt.unix race.ml && ./a.out *)

This prints

0 0.0
2 1.0
1 1.0   # !!!!!!!!!
2 2.0

The difference is that this program's on_success callback is registered on p after Lwt_react.E.limit's on_success callback. It seems #566 fixed the case where the program's callback is registered first as a side-effect, partially hiding the bug.

[AvniFatehpuria]

I just made a PR! The behaviour now is that the limit's on_success calback only pushes an event after ensuring that no other Lwt.is_sleeping is false, i.e, no other event has been pushed in the meanwhile. Please let me know if anything needs to be changed!

[aantron]

Thanks. For future readers, that was #606. I suggest checking whether Lwt_react.S.limit also has this bug.

[AvniFatehpuria]

Checking now! Should I create a separate issue if it does?

[aantron]

A separate PR for merging the code, yes, but not a separate issue :) We can re-use the convo in this one :)

[aantron]

Probably a good starting point is the React (the OCaml one) docs, Ctrl+F for "signal."

Basically, signals are like events, but instead of "occurring" at some point in time, signals continuously maintain a current value (like 3). When that value changes (like 3 to 4), React/Lwt_react calls the functions (like map) that are attached to the signal, to update dependent signals or events.

[AvniFatehpuria]

Thanks, @aantron! I think I get ihow t works now, writing some tests.

[AvniFatehpuria]

what is the expected behavior of this code? For me, it prints 5 3 1:

       let s, push = React.S.create 0 in
       let cond    = Lwt_condition.create () in
       let s'      = Lwt_react.S.limit (fun () ->  Lwt_condition.wait cond) s in
       let l       = ref [] in
       let e       = React.E.map (fun x -> l := x :: !l) (React.S.changes s') in
         ignore e;
         Lwt_condition.signal cond ();
         push 1;
         push 0;
         push 2; (* overwrites previous 0 *)
         Lwt_condition.signal cond ();
         push 3;
         Lwt_condition.signal cond ();
         push 4;
         Lwt_condition.signal cond ();
         print_endline "printing list";
         List.iter (fun a -> print_endline (string_of_int a)) !l;

[aantron]

I would expect 4 2 1 in this case. I don't see where the 5 would come from, any chance you edited the code right before posting it?

[aantron]

My bad, missed one call to signal. Would expect 4 3 2 1.

[AvniFatehpuria]

Whoops, my mistake. I can't find the code I originally meant to post, I guess I accidentally edited it, but here's a similar test:

let s, push = React.S.create 0 in
       let cond    = Lwt_condition.create () in
       let s'      = Lwt_react.S.limit (fun () -> (push 3; Lwt_condition.wait cond)) s in
       let l       = ref [] in
       let e       = React.E.map (fun x-> l := x :: !l) (React.S.changes s') in
       ignore e;
       Lwt_condition.signal cond ();
       push 1;
       push 0;
       push 2;
       Lwt_condition.signal cond ();
       print_endline "printing list";
       List.iter (fun a -> print_endline (string_of_int a)) !l;

What is the expected behavior of this? For me it usually prints only 1, which doesn't seem like the correct behavior given the two calls to signal

[aantron]

@AvniFatehpuria The first call to signal occurs before any push (which should probably be named set for a signal).

So, it doesn't unblock any waiting updates in limit. There haven't been any updates yet.

[SumitPadhiyar]

@aantron The bug relies on the order in which callbacks are called when rate-limiting promise is resolved. Are callbacks attached to promise called in some deterministic manner(i.e FIFO)? Also, Lwt documentation doesn't mention anything regarding the execution order of callbacks attached to promise. Digging further, looking at run_callbacks https://github.com/ocsigen/lwt/blob/46948539754fc7cdecffb19979d77de46dd9b67f/src/core/lwt.ml#L1179 method it looks like it executes callbacks in FIFO order. So is there any execution order of callbacks attached to a promise? If it is, can we rely on it as it is not documented?

[aantron]

@SumitPadhiyar AFAIK the bug is resolved by 4e592eb90b721966f2f0e2f3c4efe8a4536c84fa. Are you still experiencing it?

[SumitPadhiyar]

No, I'm not experiencing the bug. I wanted to know does Lwt give any guarantee regarding execution order of callbacks attached to a promise? As the bug relies on the assumption that callbacks will be executed in FIFO order, so the question.

[aantron]

Lwt doesn't give any guarantee about execution order of callbacks. It does currently execute them in FIFO order.

[SumitPadhiyar]

So precisely the original implementation of this bug, relied on the assumption that callbacks will be processed in FIFO order. Should we add this information of non-deterministic execution order of callbacks to Lwt Resolution loop's documentation?

[aantron]

If speaking precisely, how can a bug have an implementation, unless it is inserted deliberately as some sort of sabotage? An implementation is, by definition, an implementation of some design or intent. The vast majority of bugs are not themselves designed, nor is there intent to introduce them.

I don't understand what exactly you're referring to when you speak about a bug "relying" on an "assumption." No assumption was made, that we relied on to introduce the bug. We made some wrong assumptions while trying to introduce a feature, and those wrong assumptions resulted in the bug. But I don't think those assumptions are that Lwt does or does not call callbacks in a FIFO order. IIRC, the bug could be triggered no matter what the order is, you would just need a different reproducing case, depending on the order, or any case at all, if the order were randomized.