Streams

New implementation of the module Lwt_stream. The main differences are:

• it does not uses weak pointers anymore but you can still clone streams without leaking memory,
• addition of bounded "push" streams,
• allow to keep a reference for "push" streams.

Bounded push streams are created using the function Lwt_stream.​create_bounded which returns an object to control the push end-point of the stream.

It is also possible to add a reference to a value in a stream, this is especially useful when it is just used as a proxy for another data structure. For example to create a stream from a react event one can write:

let stream, push, set_ref = Lwt_stream.create_with_reference () in
set_ref (map (fun x -> push (Some x)) event);
stream

This way the event pushing occurences of event to the stream won't be garbage collected before stream.

Change of behavior

The behavior when accessing a stream from multiple threads at the same time changed. When several threads are waiting on an empty stream, and new elements become available:

• before: threads are restarted one by one with new elements,
• now: threads are all wake up with the first new element.

For example with the following code:

let st, push = Lwt_stream.create () in
let t1 = Lwt_stream.get st in
let t2 = Lwt_stream.get st in
push (Some 1);
push (Some 2);
Lwt.bind t1 (fun x1 -> Lwt.bind t2 (fun x2 -> Lwt.return (x1, x2)))

the result with Lwt < 2.4.0 is (1, 2), and with Lwt >= 2.4.0 it is (1, 1).

The behavior in this case was not specified.

Lwt.async

A new primitive Lwt.​async has been added to the module Lwt. It is a better replacement of the old Lwt.​ignore_result function (kept for compatibility):

val async : (unit -> 'a t) -> unit

If the function given as argument raises an exception or the resulting thread fails, the exception is given to Lwt.async_exception_hook, which defaults to printing the exception and exiting the program.

The goal of async is just to ensure that unexpected errors are reported; threads whose result is ignored should raise no exception in a robust program (i.e. all errors should be handled).

Cancelable threads

Now functions registered with Lwt.​on_cancel are guaranteed to be called first. For example, in the following code:

let waiter, wakener = Lwt.task () in
Lwt.on_cancel waiter f;
let t = Lwt.bind waiter g in
...

it is guaranteed that if waiter is canceled then f will be called before g.

Lwt_preemptive.run_in_main

If you want to execute some Lwt code from another preemptive thread than the main one you can now use the function Lwt_preemptive.​run_in_main. It will send the function to the main thread which will execute it and send back the result.

val run_in_main : (unit -> 'a Lwt.t) -> 'a

Windows

The module Lwt_process can now be used on Windows.

Unix jobs

API for unix jobs have been simplified, now only one C external is needed. The job structure contains one more field result containing the C callback responsible for freeing allocated data and returning the result as an OCaml value.

From the OCaml side, you must call the function Lwt_unix.​run_job instead of Lwt_unix.​execute_job.