Lwt_react.limit: fix redundant event occurrences according to documented semantics

[meadofpoetry]

possible fix for #565

[meadofpoetry]

And tests fail since it's suggested that there should be two events per limiting thread, not one. So it should either be documented or be fixed, I think.

[aantron]

The fix does look correct. Thanks!

I suggest fixing the test. I believe it involves just signaling the condition one more time in the right place(as you are probably aware).

I think there is still a race condition remaining, which may or may not be worth fixing at this point. limit uses an on_success callback. If another on_success callback is attached to the "limiter" promise, and is called first, it could push an event into the underlying event stream. That event would be emitted by limit immediately. The on_success callback in limit would then run, emitting the previously-delayed event right after that. I suppose with two interfering on_success callbacks instead of one, it's possible to even lose the queued event and have it be replaced by a later one, all while waiting to emit the original one. These conditions are probably very, very rare, as one would have to write some pretty nasty code on the user side to trigger them.

[meadofpoetry]

Do you mean that in case of simultaneous events there could be more than one callbacks attached? I'm not sure if it is possible with react since mapped function will be called sequentially, am I wrong?

I'll fix the tests.

[aantron]

Thanks!

I'll try to make a small program to demonstrate what I mean with the race condition, perhaps later today or tomorrow.

[aantron]

Ok, here it is:

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 ();
  *)

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

  push ();
  push ();

  Lwt_main.run (Lwt_unix.sleep 2.)

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

That on_success callback inside limit could be replaced with bind, finalize, etc., which makes this scenario much more realistic.

What happens here is that the on_success callback that is visible above is attached to the limiter promise first, and the on_success callback inside Lwt_react.limit is attached second, after the limiter promise is returned to that code. So, the callback seen above runs first, and the callback inside lwt_react.ml runs second.

On the first push (), everything works as expected, and it results in printing 0.0 (seconds elapsed).

The second push () is "trapped" by limit for now, also as expected.

One second later, the limiter promise resolves. The first on_success callback, which is visible above in the test program, runs, and does a third push. This third push is "gated" by the same limiter promise that just resolved, because it hasn't been replaced yet. So, this third push is passed through immediately, beating out the delayed second push, and printing 1.0 (seconds elapsed).

After that, the second on_success callback, attached inside lwt_react.ml runs (the one you just fixed, thank you :)). This now unconditionally emits the delayed second push, also printing 1.0 (seconds elapsed).

This is clearly a separate issue, so perhaps I'll open one (or you are welcome to).