Lwt: Lwt.bind2

[ibukanov]

In my code I need to wait for the result of 2 promises that can run in parallel. If any of them fails, another should be canceled. The code should also properly deal with any failure coming from the promises. To cover this case it is nice to have in Lwt a function like:

val bind_cancel2 : 
  'a1  t -> 'a2 t -> (`a1 -> `a2 -> 'b t) -> 'b t  

where bind_cancel2 promise1 promise2 f waits for the results of promise1 and promise2. If successful, it returns f result1 result2. If any of promise1 or promise2 fails, another is canceled. Then the code waits for the canceled promise to finish to cover the case when it is not cancelable. Then the code returns fail (Lwt.Exception_List exception_list).

Here Lwt.Exception_List is a special exception representing a list of exceptions coming from a parallel execution. For bind_cancel2 it typically contains either [first_fault] when fault happens after another promises succeeded or [first_fault; Lwt.Cancelled] when one promise fails and another is canceled due to that. But it may contain [first_fault; second_fault] if the canceled promise catches Lwt.Cancelled and then calls fail second_fault.

The application can deal with Lwt.Exception_List as it pleased. For example it may log all the exceptions and call exit(1).

[aantron]

Thanks. There is also https://github.com/ocsigen/lwt/issues/325 which overlaps partially if <&> is implemented as pick, but the semantics here are more thorough.

[ibukanov]

After playing with a prototype implementation I realized that a generic bind2 is a rather unsafe interface. The issue is that when one promise succeeds and another fails, a proposed bind2 silently discards the result of the success. In my toy test that resulted in discarding of a just opened file descriptor.

Yet waiting for a sequence of promises to resolve for sure while trying to cancel them on the first failure is a very useful operation. This suggest an interface like

val wait_all : unit t list -> unit t

Compared with ordinary join, it never returns until all promises on the list are resolved. If any of them fails, other are immediately canceled if possible. After that, when all promises finally resolves, failures from all promises on the list are merged together into Lwt.Exception_List as described above and wait_all fails with that.

Compared with my original bind2 idea, all promises on the wait_all list are of the unit type allowing to trivially discard their result. Essentially this takes advantage that there is a natural way to compose unit and error, while there is no why without extra callbacks to merge a generic type and error.

[aantron]

We will have to consider this during the various upcoming large projects in Lwt (https://github.com/ocsigen/lwt/wiki/Roadmap#upcoming-large-projects). Some of these present opportunities to create new alternative APIs.

[raphael-proust]

Here is a stab. The gist is:

• create a thread that chooses between the two; use the thread to send cancel in case of error
• create a thread that joins the two; use the thread as a waiting point
• extract the states of the threads after the joined thread resolves

open Lwt
open Lwt.Infix

let bind2 (p1 : 'a t) (p2 : 'a t) : 'a state list t =
    let por = (p1 >|= ignore) <?> (p2 >|= ignore) in
    let pand = (p1 >|= ignore) <&> (p2 >|= ignore) in

    Lwt.on_failure por (fun _ -> cancel p1; cancel p2);

    (catch (fun () -> pand) (fun _ -> Lwt.return ())) >>= fun () ->
    Lwt.return [ state p1; state p2 ]
;;

And here are a few tests:

let p1 =
    Lwt_unix.sleep 2. >>= fun () ->
    fail Not_found
;;
let p2 =
    Lwt_unix.sleep 1. >>= fun () ->
    return true
;;

let () =
    let t =
        bind2 p1 p2 >>= fun r ->
        assert (r = [ Fail Not_found; Return true ]);
        return ()
    in
    Lwt_main.run t

let p1 =
    Lwt_unix.sleep 1. >>= fun () ->
    fail Not_found
;;
let p2 =
    Lwt_unix.sleep 2. >>= fun () ->
    return true
;;

let () =
    let t =
        bind2 p1 p2 >>= fun r ->
        assert (r = [ Fail Not_found; Fail Canceled ]);
        return ()
    in
    Lwt_main.run t

@ibukanov does that work for your use case? If you produce short programs to test whether it does what you expect, can you post them here? (Even if it does do what you want.)

[ibukanov]

@raphael-proust Semantically it does what I wanted. But it seems it or its bindN version for multiple threads belongs to Lwt_result, as it delegates to the caller the problem of merging the 2 exceptions that may come from 2 threads or deal with the case of [success, failure] when the successful result may not be just thrown away as it needs some cleanup.

[raphael-proust]

I think the simplest is indeed to return a list of success/failure. The stab above does that, but reuses Lwt.state (which is bad because it doesn't statically guarantee that all promises are resolved/failed).

Even reworked, this should be help-up until the join/choose issue #325 is discussed more thoroughly. Essentially, this is similar to the <&&> discussed there with the addition of cancellation on failure. Maybe there should be a

val bind_n: ?cancel_on_fail: bool -> 'a Lwt.t list -> 'a result list Lwt.t
val bind_2: ?cancel_on_fail: bool -> 'a Lwt.t -> 'b Lwt.t -> ('a result * 'b result) Lwt.t

But then someone might also be interested in just getting the first exception as early as possible, not even bothering about the other thread.

[gabelevi]

I found this GitHub issue after finding a couple bugs in my project due to my expectation that Lwt.join would resolve to a failed thread as soon as the first thread fails. I expected this behavior because in the handful of other languages where I've used promises/threads/awaitables, the "run X promises in parallel" APIs would throw/reject/etc as soon as the first thing failed.

I think bind2 as a primitive would be really useful in building the parallelism functions which I need.

For example, I think a bind2 as suggested in this issue is actually a more correct implementation for running multiple threads in parallel. For example, I would expect

let%lwt a = foo ()
and b = bar ()
and c = baz ()
in z

to behave like

let internal_name1 = foo() in
let internal_name2 = bar() in
let internal_name3 = baz() in
Lwt.bind2 ~cancel_on_fail:true internal_name1 internal_name2 (fun a b ->
  Lwt.bind internal_name3 (fun c -> z)
)

where running threads in parallel will fail as soon as anything fails.

You could even use bind2 to simulate the Lwt.join behavior that I expected

(* Same as Lwt.join but fails as soon as the first thread fails *)
let rec join_fail_early ?cancel_on_fail = 
  Lwt_list.fold_left
    (fun acc thread -> Lwt.bind2 acc thread (fun () () -> Lwt.return_unit))
    Lwt.return_unit

If anyone wants the previous behavior where all threads are run to completion, they are free to catch exceptions inside their threads. I suppose they probably should be using ('a, 'b) Pervasives.result instead of interacting with failed threads.

[aantron]

@gabelevi We will probably want to add this kind of join/all in any "new" top-level Lwt API, alternatively I will be happy to merge a PR adding it to the current API :)

[aantron]

Closing this. In light of https://github.com/ocsigen/lwt/issues/283#issuecomment-518014539, I think it's better not to add new helpers that do cancellation to the Lwt API itself (they can remain as helpers in projects that use Lwt, though).

We may still want a join that fails as soon as the first promise fails, but we can open a separate issue for that.