Lwt — OCaml promises and concurrent I/O

Краткая справка

Lwt — самая популярная и распространенная монадическая библиотека для асинхронного программирования посредством механизма обещаний (с англ. promises).

В сущности состоит из независимой реализации промисов (библиотека lwt) и реализации взаимодействия с системой и исполнения (resolving) промисов (библиотека lwt.unix).

Основное руководство находится на сайте Ocsigen. Помимо этого библиотека хорошо документирована в коде, как публичный интерфейс так и внутренние вещи.

Помимо прочего активно используется в среде MirageOS.

Анонс 6.x

В новой major-версии библиотеки вводится поддержка direct-style написания асинхронного кода взамен громоздкой монадики и поддержка multicore за счёт использования современных возможностей OCaml 5.

run (fun () ->
  let continue = ref true in
  while !continue do
    match await @@ Lwt_io.read_line ic with
    | line -> await @@ Lwt_io.write_line oc line
    | exception End_of_file -> continue := false
  done)

Discuss OCaml: [ANN] Lwt.6.0.0~alpha (direct-style)

Интересные ссылки

• Заметка о Lwt, Василий Ёркин

---

• CS3110, 8.7. Promises — детальной рассмотрение дизайна и устройства промисов;
• Скринкаст Промисы под капотом
• tiny-async-lib — игрушечная библиотека для понимания устройства Lwt, можете также посмотреть видео-разбор его исходного кода и пост с форума;
• Gist dead_simple_read_files_in_lwt.ml;

---

• Lwt: a Cooperative Thread Library — whitepaper про устройство Lwt;
• Детали реализации core'а смотрите в файле lwt.ml и т.д.;

Пример

Пример Lwt-программы, которая запрашивает первую страницу Google и терпит неудачу, если запрос не завершен в течение пяти секунд:

open Lwt.Syntax

let () =
  let request =
    let* addresses = Lwt_unix.getaddrinfo "google.com" "80" [] in
    let google = Lwt_unix.((List.hd addresses).ai_addr) in

    Lwt_io.(with_connection google (fun (incoming, outgoing) ->
      let* () = write outgoing "GET / HTTP/1.1\r\n" in
      let* () = write outgoing "Connection: close\r\n\r\n" in
      let* response = read incoming in
      Lwt.return (Some response)))
  in

  let timeout =
    let* () = Lwt_unix.sleep 5. in
    Lwt.return None
  in

  match Lwt_main.run (Lwt.pick [request; timeout]) with
  | Some response -> print_string response
  | None -> prerr_endline "Request timed out"; exit 1

Монадический синтаксис

---

PPX

ppx_lwt — препроцессинг для удобной монадики с Lwt-промисами.

let%lwt user = get_user_from_api "dad" in
(* ... *)
send_message "some text";%lwt

Модуль Lwt.Syntax

let open Lwt.Syntax in
let* user =  get_user_from_api "dad" in
(* ... *)
send_message "some text"

Использование в Dune-проекте

1. Произведите установку библиотеки согласно инструкции
2. Добавьте его в ваш dune-файл вашего компонента

   (executable 
     ...
     (libraries lwt.unix))

Использование в Utop

Utop умееь автоматически резолвить промис.

utop # #require "lwt.unix";;
utop # Lwt.return ();;
- : unit = ()

Switches — управление освобождением ресурсов

См. также

• Рецепт Освобождения ресурсов

Библиотека Lwt предоставляет удобную абстракцию под названием switches (свитчи), эдакая область видимости, к которой мы привязываем ресурсы, и по выходу из которой они будут освобождены, даже в случае исключения.

Интерфейс модуля построен так, что он должен быть использован внутри функций с опциональными параметрами.

let connect ?switch uri = 
  let%lwt conn = open_connection uri in 
  (* ... *)
  Lwt_switch.add_hock switch (fun () -> close_connection conn);
  (* ... *)

let _ = 
  Lwt_switch.with_switch @@ fun switch ->
  let%lwt conn_a = connect ~switch uri in 
  let%lwt conn_b = connect ~switch uri in 
  (* ... *)

Примеры использования

• Применение в библиотеке nats.ocaml

Работа с TCP/IP (в примере)

To-Do

Добавить больше слов что-ли...

TCP клинт

В реальных проектах

• Nats_lwt.Connection — реализация подключения к NATS серверу;

open Lwt.Infix

let host = "127.0.0.1"
and port = 8080

let () =
  Lwt_main.run
  @@ Lwt_io.with_connection
       Unix.(ADDR_INET (inet_addr_of_string host, port))
       (fun (ic, oc) ->
         Lwt_io.write oc "GET / HTTP/1.1\r\n\r\n";%lwt
         Lwt_io.read_line ic >>= Lwt_io.printl)

module Tcp_connection = struct
  type t = { ic : Lwt_io.input_channel; oc : Lwt_io.output_channel }

  let create ~host ~port =
    let open Unix in

    (* Создание "сырого" Unix-сокет. *)
    let socket_fd = Lwt_unix.socket PF_INET SOCK_STREAM 0 in

    (* Создание TCP-соединения по указному адресу. *)
    let address = ADDR_INET (inet_addr_of_string host, port) in
    Lwt_unix.connect socket_fd address;%lwt

    (* Оборачивание сокета в удобную абстракцию каналов. *)
    let ic = Lwt_io.of_fd ~mode:Lwt_io.Input socket_fd in
    let oc = Lwt_io.of_fd ~mode:Lwt_io.Output socket_fd in

    Lwt.return { ic; oc }

  let read { ic; _ } = Lwt_io.read ic
  let write { oc; _ } s = Lwt_io.write oc s

  (* Для закрытия сокета достаточно закрыть один из каналов. *)
  let close { ic; _ } = Lwt_io.close ic
end

open Lwt.Infix

let () =
  Lwt_main.run @@
  let%lwt connection = Tcp_connection.create ~host:"127.0.0.1" ~port:8080 in

  (* Чтение и вывод считанной строки в stdout. *)
  Tcp_connection.read connection >>= Lwt_io.printl;%lwt

  Tcp_connection.close connection

TCP сервер

let serve ~host ~port =
  let%lwt server =
    Lwt_io.establish_server_with_client_address
      (ADDR_INET (Unix.inet_addr_of_string host, port))
    @@ fun _ (_, oc) ->
    Lwt_io.write_line oc "Hello from server!";%lwt
    Lwt_io.flush oc
  in

  let%lwt _ = fst @@ Lwt.wait () in
  Lwt_io.shutdown_server server

let () = Lwt_main.run @@ serve ~host:"127.0.0.1" ~port:8080

let serve ~host ~port =
  let socket = Lwt_unix.socket PF_INET SOCK_STREAM 0 in

  Lwt_unix.bind socket (ADDR_INET (Unix.inet_addr_of_string host, port));%lwt
  Lwt_unix.listen socket 10;

  while%lwt true do
    let%lwt client_socket, _ = Lwt_unix.accept socket in
    let oc = Lwt_io.of_fd ~mode:Output client_socket in

    Lwt_io.write_line oc "Hello from server!";%lwt
    Lwt_io.flush oc;%lwt
    
    Lwt_unix.close client_socket
  done;%lwt

  Lwt_unix.close socket

let () = Lwt_main.run @@ serve ~host:"127.0.0.1" ~port:8080

Фичи

Never-промис

Тут мы создаем промис, который никогда не будет зарезолвен, а значит последовательность не продолжится.

let never = fst @@ Lwt.wait ()

let%lwt _ = never in (* ... *)