Using Let Syntax

As was discussed in Chapter 7, Error Handling, there is a special syntax, which we call let syntax, designed for working with monads, which we can enable by enabling ppx_let.    

#require "ppx_let";;

Here’s what the bind-using version of count_lines looks like using that syntax.  

let count_lines filename =
  let%bind text = Reader.file_contents filename in
  return (List.length (String.split text ~on:'\n'));;
>val count_lines : string -> int Deferred.t = <fun>

And here’s the map-based version of count_lines.  

let count_lines filename =
  let%map text = Reader.file_contents filename in
  List.length (String.split text ~on:'\n');;
>val count_lines : string -> int Deferred.t = <fun>

The difference here is just syntactic, with these examples compiling down to the same thing as the corresponding examples written using infix operators. What’s nice about let syntax is that it highlights the analogy between monadic bind and OCaml’s built-in let-binding, thereby making your code more uniform and more readable.

Let syntax works for any monad, and you decide which monad is in use by opening the appropriate Let_syntax module. Opening Async also implicitly opens Deferred.Let_syntax, but in some contexts you may want to do that explicitly.

For the most part, let syntax is easier to read and work with, and you should default to it when using Async, which is what we’ll do for the remainder of the chapter.

Ivars and Upon

Deferreds are usually built using combinations of bind, map and return, but sometimes you want to construct a deferred where you can programmatically decide when it gets filled in. This can be done using an ivar. (The term ivar dates back to a language called Concurrent ML that was developed by John Reppy in the early ’90s. The “i” in ivar stands for incremental.)   

There are three fundamental operations for working with an ivar: you can create one, using Ivar.create; you can read off the deferred that corresponds to the ivar in question, using Ivar.read; and you can fill an ivar, thus causing the corresponding deferred to become determined, using Ivar.fill. These operations are illustrated below:

let ivar = Ivar.create ();;
>val ivar : '_weak1 Ivar.t =
>  {Async_kernel__.Types.Ivar.cell = Async_kernel__Types.Cell.Empty}
let def = Ivar.read ivar;;
>val def : '_weak1 Deferred.t = <abstr>
Deferred.peek def;;
>- : '_weak2 option = None
Ivar.fill ivar "Hello";;
>- : unit = ()
Deferred.peek def;;
>- : string option = Some "Hello"

Ivars are something of a low-level feature; operators like map, bind and return are typically easier to use and think about. But ivars can be useful when you want to build a synchronization pattern that isn’t already well supported.

As an example, imagine we wanted a way of scheduling a sequence of actions that would run after a fixed delay. In addition, we’d like to guarantee that these delayed actions are executed in the same order they were scheduled in. Here’s a signature that captures this idea:

module type Delayer_intf = sig
  type t
  val create : Time.Span.t -> t
  val schedule : t -> (unit -> 'a Deferred.t) -> 'a Deferred.t
end;;
>module type Delayer_intf =
>  sig
>    type t
>    val create : Time.Span.t -> t
>    val schedule : t -> (unit -> 'a Deferred.t) -> 'a Deferred.t
>  end

An action is handed to schedule in the form of a deferred-returning thunk (a thunk is a function whose argument is of type unit). A deferred is handed back to the caller of schedule that will eventually be filled with the contents of the deferred value returned by the thunk. To implement this, we’ll use an operator called upon, which has the following signature:  

#show upon;;
>val upon : 'a Deferred.t -> ('a -> unit) -> unit

Like bind and return, upon schedules a callback to be executed when the deferred it is passed is determined; but unlike those calls, it doesn’t create a new deferred for this callback to fill.

Our delayer implementation is organized around a queue of thunks, where every call to schedule adds a thunk to the queue and also schedules a job in the future to grab a thunk off the queue and run it. The waiting will be done using the function after, which takes a time span and returns a deferred which becomes determined after that time span elapses:

module Delayer : Delayer_intf = struct
  type t = { delay: Time.Span.t;
             jobs: (unit -> unit) Queue.t;
           }

  let create delay =
    { delay; jobs = Queue.create () }

  let schedule t thunk =
    let ivar = Ivar.create () in
    Queue.enqueue t.jobs (fun () ->
      upon (thunk ()) (fun x -> Ivar.fill ivar x));
    upon (after t.delay) (fun () ->
      let job = Queue.dequeue_exn t.jobs in
      job ());
    Ivar.read ivar
end;;
>module Delayer : Delayer_intf

This code isn’t particularly long, but it is subtle. In particular, note how the queue of thunks is used to ensure that the enqueued actions are run in the order they were scheduled, even if the thunks scheduled by upon are run out of order. This kind of subtlety is typical of code that involves ivars and upon, and because of this, you should stick to the simpler map/bind/return style of working with deferreds when you can.

let my_bind d ~f =
  let i = Ivar.create () in
  upon d (fun x -> upon (f x) (fun y -> Ivar.fill i y));
  Ivar.read i;;
>val my_bind : 'a Deferred.t -> f:('a -> 'b Deferred.t) -> 'b Deferred.t =
>  <fun>

Tail-Calls and Chains of Deferreds

There’s another memory problem you might be concerned about, which is the allocation of deferreds. If you think about the execution of copy_blocks, you’ll see it’s creating a chain of deferreds, two per time through the loop. The length of this chain is unbounded, and so, naively, you’d think this would take up an unbounded amount of memory as the echo process continues.

Happily, this is a case that Async knows how to optimize. In particular, the whole chain of deferreds should become determined precisely when the final deferred in the chain is determined, in this case, when the Eof condition is hit. Because of this, we could safely replace all of these deferreds with a single deferred. Async does just this, and so there’s no memory leak after all.

This is essentially a form of tail-call optimization, lifted to the Deferred monad. Indeed, you can tell that the bind in question doesn’t lead to a memory leak in more or less the same way you can tell that the tail recursion optimization should apply, which is that the bind that creates the deferred is in tail-position. In other words, nothing is done to that deferred once it’s created; it’s simply returned as is.  

Functions that Never Return

The call to never_returns around the call to Scheduler.go is a little bit surprising, but it has a purpose: to make it clear to whoever invokes Scheduler.go that the function never returns.    

By default, a function that doesn’t return will have an inferred return type of 'a:

let rec loop_forever () = loop_forever ();;
>val loop_forever : unit -> 'a = <fun>
let always_fail () = assert false;;
>val always_fail : unit -> 'a = <fun>

This is a little odd, but it does make sense. After all, if a function never returns, we’re free to impute any type at all to its non-existent return value. As a result, from a typing perspective, a function that never returns can fit into any context within your program.

But that itself can be problematic, especially with a function like Scheduler.go, where the fact that it never returns is perhaps not entirely obvious. The point of never_returns is to create an explicit marker so the user knows that the function in question doesn’t return.

To do this, Scheduler.go is defined to have a return value of Nothing.t.  

#show Scheduler.go;;
>val go : ?raise_unhandled_exn:bool -> unit -> never_returns

never_returns is just an alias of Nothing.t.

Nothing.t is uninhabited, which means there are no values of that type. As such, a function can’t actually return a value of type Nothing.t, so only a function that never returns can have Nothing.t as its return type! And we can cause a function that never returns to have a return value of Nothing.t by just adding a type annotation.   

let rec loop_forever () : Nothing.t = loop_forever ();;
>val loop_forever : unit -> never_returns = <fun>

The function never_returns consumes a value of type Nothing.t and returns an unconstrained type 'a.

#show_val never_returns;;
>val never_returns : never_returns -> 'a

If you try to write a function that uses Scheduler.go, and just assumes that it returns unit, you’ll get a helpful type error.

let do_stuff n =
  let x = 3 in
  if n > 0 then Scheduler.go ();
  x + n;;
>Line 3, characters 19-34:
>Error: This expression has type never_returns
>       but an expression was expected of type unit
>       because it is in the result of a conditional with no else branch

We can fix this by inserting a call to never_returns, thus making the fact that Scheduler.go doesn’t return apparent to the reader.

let do_stuff n =
  let x = 3 in
  if n > 0 then never_returns (Scheduler.go ());
  x + n;;
>val do_stuff : int -> int = <fun>

Improving the Echo Server

Let’s try to go a little bit farther with our echo server by walking through a few improvements. In particular, we will:

• Add a proper command-line interface with Command

• Add a flag to specify the port to listen on and a flag to make the server echo back the capitalized version of whatever was sent to it

• Simplify the code using Async’s Pipe interface

The following code does all of this:

open Core
open Async

let run ~uppercase ~port =
  let host_and_port =
    Tcp.Server.create
      ~on_handler_error:`Raise
      (Tcp.Where_to_listen.of_port port)
      (fun _addr r w ->
        Pipe.transfer
          (Reader.pipe r)
          (Writer.pipe w)
          ~f:(if uppercase then String.uppercase else Fn.id))
  in
  ignore
    (host_and_port
      : (Socket.Address.Inet.t, int) Tcp.Server.t Deferred.t);
  Deferred.never ()

let () =
  Command.async
    ~summary:"Start an echo server"
    (let%map_open.Command uppercase =
       flag
         "-uppercase"
         no_arg
         ~doc:" Convert to uppercase before echoing back"
     and port =
       flag
         "-port"
         (optional_with_default 8765 int)
         ~doc:" Port to listen on (default 8765)"
     in
     fun () -> run ~uppercase ~port)
  |> Command_unix.run

Note the use of Deferred.never in the run function. As you might guess from the name, Deferred.never returns a deferred that is never determined. In this case, that indicates that the echo server doesn’t ever shut down.  

The biggest change in the preceding code is the use of Async’s Pipe. A Pipe is an asynchronous communication channel that’s used for connecting different parts of your program. You can think of it as a consumer/producer queue that uses deferreds for communicating when the pipe is ready to be read from or written to. Our use of pipes is fairly minimal here, but they are an important part of Async, so it’s worth discussing them in some detail.  

Pipes are created in connected read/write pairs:

let (r,w) = Pipe.create ();;
>val r : '_weak3 Pipe.Reader.t = <abstr>
>val w : '_weak3 Pipe.Writer.t = <abstr>

r and w are really just read and write handles to the same underlying object. Note that r and w have weakly polymorphic types, as discussed in Chapter 8, Imperative Programming, and so can only contain values of a single, yet-to-be-determined type.

If we just try and write to the writer, we’ll see that we block indefinitely in utop. You can break out of the wait by hitting Control-C:

Pipe.write w "Hello World!";;
>Interrupted.

That’s because a pipe has a certain amount of internal slack, a number of slots in the pipe to which something can be written before the write will block. By default, a pipe has zero slack, which means that the deferred returned by a write is determined only when the value is read out of the pipe.

let (r,w) = Pipe.create ();;
>val r : '_weak4 Pipe.Reader.t = <abstr>
>val w : '_weak4 Pipe.Writer.t = <abstr>
let write_complete = Pipe.write w "Hello World!";;
>val write_complete : unit Deferred.t = <abstr>
Pipe.read r;;
>- : [ `Eof | `Ok of string ] = `Ok "Hello World!"
write_complete;;
>- : unit = ()

In the function run, we’re taking advantage of one of the many utility functions provided for pipes in the Pipe module. In particular, we’re using Pipe.transfer to set up a process that takes data from a reader-pipe and moves it to a writer-pipe. Here’s the type of Pipe.transfer:

Pipe.transfer;;
>- : 'a Pipe.Reader.t -> 'b Pipe.Writer.t -> f:('a -> 'b) -> unit Deferred.t =
><fun>

The two pipes being connected are generated by the Reader.pipe and Writer.pipe call respectively. Note that pushback is preserved throughout the process, so that if the writer gets blocked, the writer’s pipe will stop pulling data from the reader’s pipe, which will prevent the reader from reading in more data.

Importantly, the deferred returned by Pipe.transfer becomes determined once the reader has been closed and the last element is transferred from the reader to the writer. Once that deferred becomes determined, the server will shut down that client connection. So, when a client disconnects, the rest of the shutdown happens transparently.

The command-line parsing for this program is based on the Command library that we introduced in Chapter 15, Command Line Parsing. Opening Async, shadows the Command module with an extended version that contains the async call:

#show Command.async_spec;;
>val async_spec :
>  ('a, unit Deferred.t) Async.Command.basic_spec_command Command.with_options

This differs from the ordinary Command.basic call in that the main function must return a Deferred.t, and that the running of the command (using Command_unix.run) automatically starts the Async scheduler, without requiring an explicit call to Scheduler.go. 

URI Handling

HTTP URLs, which identify endpoints across the Web, are actually part of a more general family known as Uniform Resource Identifiers (URIs). The full URI specification is defined in RFC3986 and is rather complicated. Luckily, the uri library provides a strongly typed interface that takes care of much of the hassle.    

We’ll need a function for generating the URIs that we’re going to use to query the DuckDuckGo servers:

open Core
open Async

(* Generate a DuckDuckGo search URI from a query string *)
let query_uri query =
  let base_uri =
    Uri.of_string "http://api.duckduckgo.com/?format=json"
  in
  Uri.add_query_param base_uri ("q", [ query ])

A Uri.t is constructed from the Uri.of_string function, and a query parameter q is added with the desired search query. The library takes care of encoding the URI correctly when outputting it in the network protocol.

Parsing JSON Strings

The HTTP response from DuckDuckGo is in JSON, a common (and thankfully simple) format that is specified in RFC4627. We’ll parse the JSON data using the Yojson library, which was introduced in Chapter 18, Handling Json Data.    

We expect the response from DuckDuckGo to come across as a JSON record, which is represented by the Assoc tag in Yojson’s JSON variant. We expect the definition itself to come across under either the key “Abstract” or “Definition,” and so the following code looks under both keys, returning the first one for which a nonempty value is defined:

(* Extract the "Definition" or "Abstract" field from the DuckDuckGo
   results *)
let get_definition_from_json json =
  match Yojson.Safe.from_string json with
  | `Assoc kv_list ->
    let find key =
      match List.Assoc.find ~equal:String.equal kv_list key with
      | None | Some (`String "") -> None
      | Some s -> Some (Yojson.Safe.to_string s)
    in
    (match find "Abstract" with
    | Some _ as x -> x
    | None -> find "Definition")
  | _ -> None

Executing an HTTP Client Query

Now let’s look at the code for dispatching the search queries over HTTP, using the Cohttp library:     

(* Execute the DuckDuckGo search *)
let get_definition word =
  let%bind _, body = Cohttp_async.Client.get (query_uri word) in
  let%map string = Cohttp_async.Body.to_string body in
  word, get_definition_from_json string

To better understand what’s going on, it’s useful to look at the type for Cohttp_async.Client.get, which we can do in utop:

#require "cohttp-async";;
#show Cohttp_async.Client.get;;
>val get :
>  ?interrupt:unit Deferred.t ->
>  ?ssl_config:Conduit_async.V2.Ssl.Config.t ->
>  ?headers:Cohttp.Header.t ->
>  Uri.t -> (Cohttp.Response.t * Cohttp_async.Body.t) Deferred.t

The get call takes as a required argument a URI and returns a deferred value containing a Cohttp.Response.t (which we ignore) and a pipe reader to which the body of the request will be streamed.

In this case, the HTTP body probably isn’t very large, so we call Cohttp_async.Body.to_string to collect the data from the connection as a single deferred string, rather than consuming the data incrementally.

Running a single search isn’t that interesting from a concurrency perspective, so let’s write code for dispatching multiple searches in parallel. First, we need code for formatting and printing out the search result:

(* Print out a word/definition pair *)
let print_result (word, definition) =
  printf
    "%s\n%s\n\n%s\n\n"
    word
    (String.init (String.length word) ~f:(fun _ -> '-'))
    (match definition with
    | None -> "No definition found"
    | Some def ->
      String.concat ~sep:"\n" (Wrapper.wrap (Wrapper.make 70) def))

We use the Wrapper module from the textwrap package to do the line wrapping. It may not be obvious that this routine is using Async, but it does: the version of printf that’s called here is actually Async’s specialized printf that goes through the Async scheduler rather than printing directly. The original definition of printf is shadowed by this new one when you open Async. An important side effect of this is that if you write an Async program and forget to start the scheduler, calls like printf won’t actually generate any output!

The next function dispatches the searches in parallel, waits for the results, and then prints:

(* Run many searches in parallel, printing out the results after
   they're all done. *)
let search_and_print words =
  let%map results = Deferred.all (List.map words ~f:get_definition) in
  List.iter results ~f:print_result

We used List.map to call get_definition on each word, and Deferred.all to wait for all the results. Here’s the type of Deferred.all:

Deferred.all;;
>- : 'a Deferred.t list -> 'a list Deferred.t = <fun>

The list returned by Deferred.all reflects the order of the deferreds passed to it. As such, the definitions will be printed out in the same order that the search words are passed in, no matter what order the queries return in. It also means that no printing occurs until all results arrive.

We could rewrite this code to print out the results as they’re received (and thus potentially out of order) as follows:

(* Run many searches in parallel, printing out the results as you
   go *)
let search_and_print words =
  Deferred.all_unit
    (List.map words ~f:(fun word ->
         get_definition word >>| print_result))

The difference is that we both dispatch the query and print out the result in the closure passed to map, rather than wait for all of the results to get back and then print them out together. We use Deferred.all_unit, which takes a list of unit deferreds and returns a single unit deferred that becomes determined when every deferred on the input list is determined. We can see the type of this function in utop:

Deferred.all_unit;;
>- : unit Deferred.t list -> unit Deferred.t = <fun>

Finally, we create a command-line interface using Command.async:

let () =
  Command.async
    ~summary:"Retrieve definitions from duckduckgo search engine"
    (let%map_open.Command words =
       anon (sequence ("word" %: string))
     in
     fun () -> search_and_print words)
  |> Command_unix.run

And that’s all we need for a simple but usable definition searcher:

dune exec -- ./search.exe "Concurrent Programming" "OCaml"
>Concurrent Programming
>----------------------
>
>"Concurrent computing is a form of computing in which several
>computations are executed during overlapping time
>periods—concurrently—instead of sequentially. This is a property
>of a system—this may be an individual program, a computer, or a
>network—and there is a separate execution point or \"thread of
>control\" for each computation. A concurrent system is one where a
>computation can advance without waiting for all other computations to
>complete."
>
>OCaml
>-----
>
>"OCaml, originally named Objective Caml, is the main implementation of
>the programming language Caml, created by Xavier Leroy, Jérôme
>Vouillon, Damien Doligez, Didier Rémy, Ascánder Suárez and others
>in 1996. A member of the ML language family, OCaml extends the core
>Caml language with object-oriented programming constructs."

Monitors

try_with is a useful tool for handling exceptions in Async, but it’s not the whole story. All of Async’s exception-handling mechanisms, try_with included, are built on top of Async’s system of monitors, which are inspired by the error-handling mechanism in Erlang of the same name. Monitors are fairly low-level and are only occasionally used directly, but it’s nonetheless worth understanding how they work.  

In Async, a monitor is a context that determines what to do when there is an unhandled exception. Every Async job runs within the context of some monitor, which, when the job is running, is referred to as the current monitor. When a new Async job is scheduled, say, using bind or map, it inherits the current monitor of the job that spawned it.

Monitors are arranged in a tree—when a new monitor is created (say, using Monitor.create), it is a child of the current monitor. You can explicitly run jobs within a monitor using within, which takes a thunk that returns a nondeferred value, or within', which takes a thunk that returns a deferred. Here’s an example:

let blow_up () =
  let monitor = Monitor.create ~name:"blow up monitor" () in
  within' ~monitor maybe_raise;;
>val blow_up : unit -> unit Deferred.t = <fun>
blow_up ();;
>- : unit = ()
blow_up ();;
>Exception: (monitor.ml.Error Exit ("Caught by monitor blow up monitor"))

In addition to the ordinary stack-trace, the exception displays the trace of monitors through which the exception traveled, starting at the one we created, called “blow up monitor.” The other monitors you see come from utop’s special handling of deferreds.

Monitors can do more than just augment the error-trace of an exception. You can also use a monitor to explicitly handle errors delivered to that monitor. The Monitor.detach_and_get_error_stream call is a particularly important one. It detaches the monitor from its parent, handing back the stream of errors that would otherwise have been delivered to the parent monitor. This allows one to do custom handling of errors, which may include reraising errors to the parent. Here is a very simple example of a function that captures and ignores errors in the processes it spawns.

let swallow_error () =
  let monitor = Monitor.create () in
  Stream.iter (Monitor.detach_and_get_error_stream monitor)
    ~f:(fun _exn -> printf "an error happened\n");
  within' ~monitor (fun () ->
    let%bind () = after (Time.Span.of_sec 0.25) in
    failwith "Kaboom!");;
>val swallow_error : unit -> 'a Deferred.t = <fun>

The deferred returned by this function is never determined, since the computation ends with an exception rather than a return value. That means that if we run this function in utop, we’ll never get our prompt back.

We can fix this by using Deferred.any along with a timeout to get a deferred we know will become determined eventually. Deferred.any takes a list of deferreds, and returns a deferred which will become determined assuming any of its arguments becomes determined.

Deferred.any [ after (Time.Span.of_sec 0.5)
             ; swallow_error () ];;
>an error happened
>- : unit = ()

As you can see, the message “an error happened” is printed out before the timeout expires.

Here’s an example of a monitor that passes some exceptions through to the parent and handles others. Exceptions are sent to the parent using Monitor.send_exn, with Monitor.current being called to find the current monitor, which is the parent of the newly created monitor.

exception Ignore_me;;
>exception Ignore_me
let swallow_some_errors exn_to_raise =
  let child_monitor  = Monitor.create  () in
  let parent_monitor = Monitor.current () in
  Stream.iter
    (Monitor.detach_and_get_error_stream child_monitor)
    ~f:(fun error ->
      match Monitor.extract_exn error with
      | Ignore_me -> printf "ignoring exn\n"
      | _ -> Monitor.send_exn parent_monitor error);
  within' ~monitor:child_monitor (fun () ->
    let%bind () = after (Time.Span.of_sec 0.25) in
    raise exn_to_raise);;
>val swallow_some_errors : exn -> 'a Deferred.t = <fun>

Note that we use Monitor.extract_exn to grab the underlying exception that was thrown. Async wraps exceptions it catches with extra information, including the monitor trace, so you need to grab the underlying exception if you want to depend on the details of the original exception thrown.

If we pass in an exception other than Ignore_me, like, say, the built-in exception Not_found, then the exception will be passed to the parent monitor and delivered as usual:

exception Another_exception;;
>exception Another_exception
Deferred.any [ after (Time.Span.of_sec 0.5)
             ; swallow_some_errors Another_exception ];;
>Exception:
>(monitor.ml.Error (Another_exception) ("Caught by monitor (id 69)")).

If instead we use Ignore_me, the exception will be ignored, and the computation will finish when the timeout expires.

Deferred.any [ after (Time.Span.of_sec 0.5)
             ; swallow_some_errors Ignore_me ];;
>ignoring exn
>- : unit = ()

In practice, you should rarely use monitors directly, and instead use functions like try_with and Monitor.protect that are built on top of monitors. One example of a library that uses monitors directly is Tcp.Server.create, which tracks both exceptions thrown by the logic that handles the network connection and by the callback for responding to an individual request, in either case responding to an exception by closing the connection. It is for building this kind of custom error handling that monitors can be helpful.

Example: Handling Exceptions with DuckDuckGo

Let’s now go back and improve the exception handling of our DuckDuckGo client. In particular, we’ll change it so that any query that fails is reported without preventing other queries from completing.   

The search code as it is fails rarely, so let’s make a change that allows us to trigger failures more predictably. We’ll do this by making it possible to distribute the requests over multiple servers. Then, we’ll handle the errors that occur when one of those servers is misspecified.

First we’ll need to change query_uri to take an argument specifying the server to connect to:

(* Generate a DuckDuckGo search URI from a query string *)
let query_uri ~server query =
  let base_uri =
    Uri.of_string
      (String.concat [ "http://"; server; "/?format=json" ])
  in
  Uri.add_query_param base_uri ("q", [ query ])

In addition, we’ll make the necessary changes to get the list of servers on the command-line, and to distribute the search queries round-robin across the list of servers.

Now, let’s see what happens when we rebuild the application and run it on two servers, one of which won’t respond to the query.

dune exec -- ./search.exe -servers localhost,api.duckduckgo.com "Concurrent Programming" "OCaml"
>(monitor.ml.Error (Unix.Unix_error "Connection refused" connect 127.0.0.1:80)
> ("Raised by primitive operation at file \"duniverse/async_unix/src/unix_syscalls.ml\", line 1046, characters 17-74"
>  "Called from file \"duniverse/async_kernel/src/deferred1.ml\", line 17, characters 40-45"
>  "Called from file \"duniverse/async_kernel/src/job_queue.ml\", line 170, characters 6-47"
>  "Caught by monitor Tcp.close_sock_on_error"))
[1]

As you can see, we got a “Connection refused” failure, which ends the entire program, even though one of the two queries would have gone through successfully on its own. We can handle the failures of individual connections separately by using the try_with function within each call to get_definition, as follows:

(* Execute the DuckDuckGo search *)
let get_definition ~server word =
  match%map
    try_with (fun () ->
        let%bind _, body =
          Cohttp_async.Client.get (query_uri ~server word)
        in
        let%map string = Cohttp_async.Body.to_string body in
        word, get_definition_from_json string)
  with
  | Ok (word, result) -> word, Ok result
  | Error _ -> word, Error "Unexpected failure"

Here, we first use try_with to capture the exception, and then use match%map (another syntax provided by ppx_let) to convert the error into the form we want: a pair whose first element is the word being searched for, and the second element is the (possibly erroneous) result.  

Now we just need to change the code for print_result so that it can handle the new type:

(* Print out a word/definition pair *)
let print_result (word, definition) =
  printf
    "%s\n%s\n\n%s\n\n"
    word
    (String.init (String.length word) ~f:(fun _ -> '-'))
    (match definition with
    | Error s -> "DuckDuckGo query failed: " ^ s
    | Ok None -> "No definition found"
    | Ok (Some def) ->
      String.concat ~sep:"\n" (Wrapper.wrap (Wrapper.make 70) def))

Now, if we run that same query, we’ll get individualized handling of the connection failures:

dune exec -- ./search.exe -servers localhost,api.duckduckgo.com "Concurrent Programming" OCaml
>Concurrent Programming
>----------------------
>
>DuckDuckGo query failed: Unexpected failure
>
>OCaml
>-----
>
>"OCaml, originally named Objective Caml, is the main implementation of
>the programming language Caml, created by Xavier Leroy, Jérôme
>Vouillon, Damien Doligez, Didier Rémy, Ascánder Suárez and others
>in 1996. A member of the ML language family, OCaml extends the core
>Caml language with object-oriented programming constructs."

Now, only the query that went to localhost failed.

Note that in this code, we’re relying on the fact that Cohttp_async.Client.get will clean up after itself after an exception, in particular by closing its file descriptors. If you need to implement such functionality directly, you may want to use the Monitor.protect call, which is analogous to the protect call described in Chapter 7, Cleaning Up In The Presence Of Exceptions.

Multicore OCaml

OCaml doesn’t support truly parallel threads today, but it will soon. The current development branch of OCaml, which is expected to be released in 2022 as OCaml 5.0, has a long awaited multicore-capable garbage collector, which is the result of years of research and hard implementation work.  

We won’t discuss the multicore gc here in part because it’s not yet released, and in part because there’s a lot of open questions about how OCaml programs should take advantage of multicore in a way that’s safe, convenient, and performant. Given all that, we just don’t know enough to write a chapter about multicore today.

In any case, while multicore OCaml isn’t here yet, it’s an exciting part of OCaml’s near-term future.

Thread-Safety and Locking

Once you start working with system threads, you’ll need to be careful about mutable data structures. Most mutable OCaml data structures will behave non-deterministically when accessed concurrently by multiple threads. The issues you can run into range from runtime exceptions to corrupted data structures. That means you should almost always use mutexes when sharing mutable data between different systems threads. Even data structures that seem like they should be safe but are mutable under the covers, like lazy values, can behave in surprising ways when accessed from multiple threads.     

There are two commonly available mutex packages for OCaml: the Mutex module that’s part of the standard library, which is just a wrapper over OS-level mutexes and Nano_mutex, a more efficient alternative that takes advantage of some of the locking done by the OCaml runtime to avoid needing to create an OS-level mutex much of the time. As a result, creating a Nano_mutex.t is 20 times faster than creating a Mutex.t, and acquiring the mutex is about 40 percent faster.

Overall, combining Async and threads is quite tricky, but it’s pretty simple if the following two conditions hold:

• There is no shared mutable state between the various threads involved.

• The computations executed by In_thread.run do not make any calls to the Async library.

That said, you can safely use threads in ways that violate these constraints. In particular, foreign threads can acquire the Async lock using calls from the Thread_safe module in Async, and thereby run Async computations safely. This is a very flexible way of connecting threads to the Async world, but it’s a complex use case that is beyond the scope of this chapter.