tiny-async-lib

A tiny concurrent I/O and promises library that is written to understand Lwt's internals by building a lightweight mental model of it.

openTiny_async_libopenPromise.Syntaxlet main =let*()=Io.(write_all stdout) "Hi! What's your name? "inlet* name =Io.(read_line stdin) inIo.(write_all stdout) ("Hello, "^ name ^"!\n")
let()=Engine.run main

$ dune exec ./examples/hello.exeHi! What's your name? Артём Hello, Артём!

See more examples in the directory.

References

• Chapter 8.7. Promises, where I got my understanding of how to implement promises and how they work
• Beautiful Lwt's source code with detailed implementation comments
• Whitepaper Lwt: a Cooperative Thread Library is a really accessible article to understand the core of Lwt
• Another great resource is the book Unix system programming in OCaml for writing OS-dependent code

Quick start

For build the library, you should have OCaml 4.14 (LTS) and above, and the Dune build system. No out-of-the-box dependencies are required.

To play with the source code you can just do it:

$ git clone https://github.com/dx3mod/tiny-async-lib.git
$ cd ./tiny-async-lib
$ dune build

Via an interactive toplevel environment using the Utop:

$ dune utop

You can also install library using the OPAM package manager:

$ opam tiny-async-lib.dev https://github.com/dx3mod/tiny-async-lib.git

Among other things, it is useful to have API references for easy navigation through the library using odoc.

$ dune build @doc
$ open _build/default/_doc/_html/index.html

In Depth

The Tiny_async_lib consists of three important parts: promises, asynchronous engine (i.e. event loop) and I/O module.

Promises

Promise is the first key abstraction, an abstraction for synchronizing program execution in concurrent (non-sequential) evaluations.

In simple terms, it’s an abstraction over callbacks. Promises allow us to build (monadic) sequential evaluations inside non-sequential evaluations.

Typical example of callbacks for asynchronous (non-sequential) code:

letread_two_files (file1, file2) callback= async_read_file file1 (fun_ -> async_read_file file2 (fun_ -> (* ... *)))
read_two_files ("file-1", "file-2") (fun_ -> (* ... *))

Same thing, but with promises:

letread_two_files (file1, file2) =let* _ = async_read_file file1 inlet* _ = async_read_file file2 in(* ... *)let* _ = read_two_files ("file-1", "file-2") in(* ... *)

A promise is basically an object that acts as a proxy for a result that we don't know yet, usually because we haven't finished computing its value.

It's very much the lazy_t type.

#lazy (1+1);;
- : intlazy_t=<lazy>

A promis can have one of three states: fulfilled (contains a value), rejected (contains an exception), and pending (contains callbacks).

If a promise is fulfilled or rejected, it is called resolved.

Callbacks are functions that are called when a promise is resolved. So when we (monadic) bind, if the promise is in pending state, we add a callback that calls the following monadic sequence when the promise is resolved.

Make a promise

Typical pattern of making raw promises i. e. wrapping callbacks.

letasync_event()=(* The promise is public read-only interface.  The resolver is private interface for resolve the promise. *)let promise, resolver =Promise.make ()in(* Callback wrapping. *)
on_event (funevent -> (* ... *)Promise.fulfill resolver event);
(* Returns the public interface, promise. *)
promise

Now we can write linear code on how to process the promised value.

async_event ()>>= do_something >>= do_something_yet

In details:

#let p = async_event ();;
#Promise.state p;;
- : event Promise.state =Pending[]# p >>=fun_ -> Promise.return ();;
#Promise.state p;;
- : event Promise.state =Pending[<abstr>]

Engine

The second key abstraction and part of the library is an asynchronous I/O engine that polls I/O events and dispatches them to handlers. With this we have multiplexed I/O, event subscription, etc.

letsleepdelay=let promise, resolver =Promise.make ()inEngine.(on_delay instance) delay
@@Event_loop.Handler.make (fun_ -> Promise.fulfill resolver ());
promise

The engine implemented in the library is based on the (unix) select mechanism. Select is very easy to use. It queries read and write ready file descriptors, i.e. those that are ready for processing, and dispatches them to their handlers.

The (typical) asynchronous engine in internals has an event loop. At each iteration of the event loop, the engine polls for new events and calls handlers to handle them.

letperform_iterationengine=(* ... *)let readable_fds, writable_fds, _ =Unix.select
(Io_scheduler.fds ev.wait_readable_descrs)
(Io_scheduler.fds ev.wait_writable_descrs)
[] timeout
inTime_sleepers_scheduler.invoke_handlers ~utime ev.time_sleepers
@@Time_sleepers_scheduler.dequeue_wakened_sleepers ~utime ev.time_sleepers;
Io_scheduler.invoke ev.wait_readable_descrs readable_fds;
Io_scheduler.invoke ev.wait_writable_descrs writable_fds;

With all this in place, it is possible to resolve I/O promises. It's not a big deal. We just have to loop the event loop until the promis is resolved.

letrec runpromise=matchPromise.state promise with|Fulfilledvalue -> value
|Rejectedexc -> raise exc
|Pending_ ->
Event_loop.perform_iteration instance;
run promise

I/O

This part of the library couples promises and engine to do useful programs. There was an example of a sleep function earlier.

Asynchronous engine callback functions (called handlers) are wrapped to create I/O promises. For example, the write_all function:

letwrite_allfdcontents=let promise, resolver =Promise.make ()inlet handler =object (self)
(* ... *)methodon_stop=Promise.fulfill resolver ()methodon_invoke=let bytes_write =Unix.write_substring fd contents all_bytes_write length in
all_bytes_write <- all_bytes_write + bytes_write;
if all_bytes_write = contents_length then self#stop
endinEngine.(on_writeable instance) fd handler;
promise

Afterword

Enjoy it! :<

-->

License

It's not very useful code for real things. Many parts of the implementation are lifted from other solutions. It's a crazy mix. Do whatever you want with that code.