Lwt manual


When writing a program, a common developer's task is to handle IO operations. Indeed most software interact with several different resources, such as:

  • the kernel, by doing system calls
  • the user, by reading the keyboard, the mouse, or any input device
  • a graphical server, to build graphical user interface
  • other computers, by using the network
  • ...

When this list contains only one item, it is pretty easy to handle. However as this list grows it becomes harder and harder to make everything works together. Several choices have been proposed to solve this problem:

  • using a main loop, and integrate all components we are interacting with into this main loop.
  • using preemptive system threads

Both solutions have their advantages and their drawbacks. For the first one, it may work, but it becomes very complicated to write a piece of asynchronous sequential code. The typical example is graphical user interfaces freezing and not redrawing themselves because they are waiting for some blocking part of the code to complete.

If you already wrote code using preemptive threads, you should know that doing it right with threads is a hard job. Moreover system threads consume non negligible resources, and so you can only launch a limited number of threads at the same time. Thus this is not a real solution.

Lwt offers a new alternative. It provides very light-weight cooperative threads; ``launching'' a thread is a very fast operation, it does not require a new stack, a new process, or anything else. Moreover context switches are very fast. In fact, it is so easy that we will launch a thread for every system call. And composing cooperative threads will allow us to write highly asynchronous programs.

In a first part, we will explain the concepts of Lwt, then we will describe the many sub-libraries of Lwt.

The Lwt core library

In this section we describe the basics of Lwt. It is advised to start an ocaml toplevel and try the given code examples. To start, launch ocaml in a terminal or in emacs with the tuareg mode, and type:

# #use "topfind";;
# #require "lwt.simple-top";;

lwt.simple-top makes sure Lwt threads can run while using the toplevel. You do not need it if you are using utop.

Lwt concepts

Let's take a classical function of the Pervasives module:

# Pervasives.input_char;;
- : in_channel -> char = <fun>

This function will wait for a character to come on the given input channel, and then return it. The problem with this function is that it is blocking: while it is being executed, the whole program will be blocked, and other events will not be handled until it returns.

Now let's look at the lwt equivalent:

# Lwt_io.read_char;;
- : Lwt_io.input_channel -> char Lwt.t = <fun>

As you can see, it does not return a character but something of type char Lwt.t. The type 'a Lwt.t is the type of threads returning a value of type 'a. Actually the Lwt_io.read_char will try to read a character from the given input channel and immediately returns a light-weight thread.

Now, let's see what we can do with a Lwt thread. The following code creates a pipe, and launches a thread reading on the input side:

# let ic, oc = Lwt_io.pipe ();;
val ic : Lwt_io.input_channel = <abstr>
val oc : Lwt_io.output_channel = <abstr>
# let t = Lwt_io.read_char ic;;
val t : char Lwt.t = <abstr>

We can now look at the state of our newly created thread:

# Lwt.state t;;
- : char Lwt.state = Lwt.Sleep

A thread may be in one of the following states:

  • Return x, which means that the thread has terminated successfully and returned the value x
  • Fail exn, which means that the thread has terminated, but instead of returning a value, it failed with the exception exn
  • Sleep, which means that the thread is currently sleeping and has not yet returned a value or an exception

The thread t is sleeping because there is currently nothing to read from the pipe. Let's write something:

# Lwt_io.write_char oc 'a';;
- : unit Lwt.t = <abstr>
# Lwt.state t;;
- : char Lwt.state = Lwt.Return 'a'

So, after we write something, the reading thread has been awoken and has returned the value 'a'.

Primitives for thread creation

There are several primitives for creating Lwt threads. These functions are located in the module Lwt.

Here are the main primitives:

  • Lwt.return : 'a -> 'a Lwt.t
    creates a thread which has already terminated and returned a value
  • Lwt.fail : exn -> 'a Lwt.t
    creates a thread which has already terminated and failed with an exception
  • Lwt.wait : unit -> 'a Lwt.t * 'a Lwt.u
    creates a sleeping thread and returns this thread plus a wakener (of type 'a Lwt.u) which must be used to wakeup the sleeping thread.

To wake up a sleeping thread, you must use one of the following functions:

  • Lwt.wakeup : 'a Lwt.u -> 'a -> unit
    wakes up the thread with a value.
  • Lwt.wakeup_exn : 'a Lwt.u -> exn -> unit
    wakes up the thread with an exception.

Note that it is an error to wakeup the same thread twice. Lwt will raise Invalid_argument if you try to do so.

With this information, try to guess the result of each of the following expression:

# Lwt.state (Lwt.return 42);;
# Lwt.state (Lwt.fail Exit);;
# let waiter, wakener = Lwt.wait ();;
# Lwt.state waiter;;
# Lwt.wakeup wakener 42;;
# Lwt.state waiter;;
# let waiter, wakener = Lwt.wait ();;
# Lwt.state waiter;;
# Lwt.wakeup_exn wakener Exit;;
# Lwt.state waiter;;

Primitives for thread composition

The most important operation you need to know is bind:

val bind : 'a Lwt.t -> ('a -> 'b Lwt.t) -> 'b Lwt.t

bind t f creates a thread which waits for t to terminate, then passes the result to f. If t is a sleeping thread, then bind t f will be a sleeping thread too, until t terminates. If t fails, then the resulting thread will fail with the same exception. For example, consider the following expression:

  (Lwt_io.read_line Lwt_io.stdin)
  (fun str -> Lwt_io.printlf "You typed %S" str)

This code will first wait for the user to enter a line of text, then print a message on the standard output.

Similarly to bind, there is a function to handle the case when t fails:

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

catch f g will call f (), then waits for its termination, and if it fails with an exception exn, calls g exn to handle it. Note that both exceptions raised with Pervasives.raise and Lwt.fail are caught by catch.

Cancelable threads

In some case, we may want to cancel a thread. For example, because it has not terminated after a timeout. This can be done with cancelable threads. To create a cancelable thread, you must use the Lwt.task function:

val task : unit -> 'a Lwt.t * 'a Lwt.u

It has the same semantics as Lwt.wait except that the sleeping thread can be canceled with Lwt.cancel:

val cancel : 'a Lwt.t -> unit

The thread will then fail with the exception Lwt.Canceled. To execute a function when the thread is canceled, you must use Lwt.on_cancel:

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

Note that it is also possible to cancel a thread which has not been created with Lwt.task. In this case, the deepest cancelable thread connected with the given thread will be cancelled.

For example, consider the following code:

# let waiter, wakener = Lwt.task ();;
val waiter : '_a Lwt.t = <abstr>
val wakener : '_a Lwt.u = <abstr>
# let t = Lwt.bind waiter (fun x -> Lwt.return (x + 1));;
val t : int Lwt.t = <abstr>

Here, cancelling t will in fact cancel waiter. t will then fail with the exception Lwt.Canceled:

# Lwt.cancel t;;
- : unit = ()
# Lwt.state waiter;;
- : int Lwt.state = Lwt.Fail Lwt.Canceled
# Lwt.state t;;
- : int Lwt.state = Lwt.Fail Lwt.Canceled

By the way, it is possible to prevent a thread from being canceled by using the function Lwt.protected:

val protected : 'a Lwt.t -> 'a Lwt.t

Canceling (proctected t) will have no effect on t.

Primitives for multi-thread composition

We now show how to compose several concurrent threads. The main functions for this are in the Lwt module: join, choose and pick.

The first one, join takes a list of threads and waits for all of them to terminate:

val join : unit Lwt.t list -> unit Lwt.t

Moreover, if at least one thread fails, join l will fail with the same exception as the first to fail, after all threads terminate.

Similarly choose waits for at least one thread to terminate, then returns the same value or exception:

val choose : 'a Lwt.t list -> 'a Lwt.t

For example:

# let waiter1, wakener1 = Lwt.wait ();;
val waiter1 : '_a Lwt.t = <abstr>
val wakener1 : '_a Lwt