[OCaml/Morph] Add multi-process and fortunes (#6151)

Refactor and start using opam
Remove static benchmark since there's no real diff
Use other nproc implementation
Prepare for a single-process benchmark
Last fixes for server_io
Fix flambda dockerfile
Use database queries from opium benchmark
Add fortunes route
Add single process test

frameworks/OCaml/morph/.dockerignore

@@ -0,0 +1,6 @@
+_esy
+node_modules
+_opam
+_build
+**/_build
+**/_opam

frameworks/OCaml/morph/benchmark_config.json

@@ -7,6 +7,7 @@
         "plaintext_url": "/plaintext",
         "db_url": "/db",
         "query_url": "/queries/",
+        "fortune_url": "/fortunes",
         "port": 8080,
         "approach": "Realistic",
         "classification": "Micro",
@@ -23,11 +24,12 @@
         "notes": "",
         "versus": "None"
       },
-      "static": {
+      "flambda": {
         "json_url": "/json",
         "plaintext_url": "/plaintext",
         "db_url": "/db",
         "query_url": "/queries/",
+        "fortune_url": "/fortunes",
         "port": 8080,
         "approach": "Realistic",
         "classification": "Micro",
@@ -40,9 +42,31 @@
         "webserver": "None",
         "os": "Linux",
         "database_os": "Linux",
-        "display_name": "Morph",
+        "display_name": "Morph-flambda",
         "notes": "",
         "versus": "None"
+      },
+      "single": {
+        "json_url": "/json",
+        "plaintext_url": "/plaintext",
+        "db_url": "/db",
+        "query_url": "/queries/",
+        "fortune_url": "/fortunes",
+        "port": 8080,
+        "approach": "Realistic",
+        "classification": "Micro",
+        "database": "postgres",
+        "framework": "Morph",
+        "language": "OCaml",
+        "flavor": "None",
+        "orm": "Raw",
+        "platform": "None",
+        "webserver": "None",
+        "os": "Linux",
+        "database_os": "Linux",
+        "display_name": "Morph-single-process",
+        "notes": "This is using a single process since that is more realistic",
+        "versus": "None"
       }
     }
   ]

frameworks/OCaml/morph/morph-flambda.dockerfile

@@ -0,0 +1,25 @@
+FROM ocurrent/opam:alpine-3.12-ocaml-4.11-flambda
+
+# https://caml.inria.fr/pub/docs/manual-ocaml/libref/Gc.html
+# https://linux.die.net/man/1/ocamlrun
+# https://blog.janestreet.com/memory-allocator-showdown/
+ENV OCAMLRUNPARAM a=2,o=240
+
+RUN sudo apk update && sudo apk add openssl-dev && \
+    opam depext dune conf-libev httpaf httpaf-lwt-unix lwt yojson conf-postgresql conf-libffi
+
+COPY src/morph-tfb.opam src/dune-project src/morph-tfb.opam.template ./
+
+RUN opam install --yes --deps-only .
+
+COPY ./src/bin ./bin
+COPY ./src/server_io ./server_io
+COPY ./src/server_io_single ./server_io_single
+COPY ./src/server_io_nproc ./server_io_nproc
+
+ENV SERVER_IO=NPROC
+
+RUN sudo chown -R opam ./bin && sudo chown -R opam ./server_*
+RUN opam exec -- dune build --profile release bin/tfb.exe
+
+ENTRYPOINT _build/default/bin/tfb.exe

frameworks/OCaml/morph/morph-single.dockerfile

@@ -0,0 +1,25 @@
+FROM ocurrent/opam:alpine-3.12-ocaml-4.11
+
+# https://caml.inria.fr/pub/docs/manual-ocaml/libref/Gc.html
+# https://linux.die.net/man/1/ocamlrun
+# https://blog.janestreet.com/memory-allocator-showdown/
+ENV OCAMLRUNPARAM a=2,o=240
+
+RUN sudo apk update && sudo apk add openssl-dev && \
+    opam depext dune conf-libev httpaf httpaf-lwt-unix lwt yojson conf-postgresql conf-libffi
+
+COPY src/morph-tfb.opam src/dune-project src/morph-tfb.opam.template ./
+
+RUN opam install --yes --deps-only .
+
+COPY ./src/bin ./bin
+COPY ./src/server_io ./server_io
+COPY ./src/server_io_single ./server_io_single
+COPY ./src/server_io_nproc ./server_io_nproc
+
+ENV SERVER_IO=SINGLE
+
+RUN sudo chown -R opam ./bin && sudo chown -R opam ./server_*
+RUN opam exec -- dune build --profile release bin/tfb.exe
+
+ENTRYPOINT _build/default/bin/tfb.exe

frameworks/OCaml/morph/morph-static.dockerfile

@@ -1,33 +0,0 @@
-FROM alpine:latest as certs
-RUN apk --update add ca-certificates
-
-FROM reasonnative/web:4.10.1-nightly as builder
-
-RUN mkdir /app
-WORKDIR /app
-
-COPY src/esy.json src/morph-tfb.opam src/dune-project /app/
-
-RUN esy install
-RUN esy build-dependencies --release
-
-COPY ./src/bin /app/bin
-
-RUN esy dune build --profile=docker --release
-
-RUN esy mv "#{self.target_dir / 'default' / 'bin' / 'tfb.exe'}" main.exe
-
-RUN strip main.exe
-
-FROM scratch as runtime
-
-ENV OPENSSL_STATIC=1
-ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt
-ENV SSL_CERT_DIR=/etc/ssl/certs
-COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
-
-WORKDIR /app
-
-COPY --from=builder /app/main.exe main.exe
-
-ENTRYPOINT ["/app/main.exe"]

frameworks/OCaml/morph/morph.dockerfile

@@ -1,15 +1,25 @@
-FROM reasonnative/web:4.10.1-nightly as builder
+FROM ocurrent/opam:alpine-3.12-ocaml-4.11
 
-RUN mkdir /app
-WORKDIR /app
+# https://caml.inria.fr/pub/docs/manual-ocaml/libref/Gc.html
+# https://linux.die.net/man/1/ocamlrun
+# https://blog.janestreet.com/memory-allocator-showdown/
+ENV OCAMLRUNPARAM a=2,o=240
 
-COPY src/esy.json src/morph-tfb.opam src/dune-project /app/
+RUN sudo apk update && sudo apk add openssl-dev && \
+    opam depext dune conf-libev httpaf httpaf-lwt-unix lwt yojson conf-postgresql conf-libffi
 
-RUN esy install
-RUN esy build-dependencies --release
+COPY src/morph-tfb.opam src/dune-project src/morph-tfb.opam.template ./
 
-COPY ./src/bin /app/bin
+RUN opam install --yes --deps-only .
 
-RUN esy build --release
+COPY ./src/bin ./bin
+COPY ./src/server_io ./server_io
+COPY ./src/server_io_single ./server_io_single
+COPY ./src/server_io_nproc ./server_io_nproc
 
-ENTRYPOINT ["esy", "x", "--release", "tfb"]
+ENV SERVER_IO=NPROC
+
+RUN sudo chown -R opam ./bin && sudo chown -R opam ./server_*
+RUN opam exec -- dune build --profile release bin/tfb.exe
+
+ENTRYPOINT _build/default/bin/tfb.exe

frameworks/OCaml/morph/src/bin/Caqti.ml

@@ -0,0 +1,85 @@
+module Archi = struct
+  let start () =
+    let connection_url =
+      "postgresql://benchmarkdbuser:benchmarkdbpass@tfb-database:5432/hello_world?connect_timeout=15"
+    in
+    Caqti_lwt.connect_pool ~max_size:10 (Uri.of_string connection_url)
+    |> Result.map_error Caqti_error.show
+    |> Lwt.return
+
+  let stop
+      (pool : (Caqti_lwt.connection, [> Caqti_error.connect ]) Caqti_lwt.Pool.t)
+      =
+    Logs.info (fun m -> m "Disconnecting from database");
+    Caqti_lwt.Pool.drain pool
+
+  let component :
+      ( unit,
+        (Caqti_lwt.connection, [ | Caqti_error.t ]) Caqti_lwt.Pool.t )
+      Archi_lwt.Component.t =
+    Archi_lwt.Component.make ~start ~stop
+end
+
+module Query = struct
+  type ('res, 'err) query_result =
+    ('res, ([> Caqti_error.call_or_retrieve ] as 'err)) result Lwt.t
+
+  type ('res, 'err) query = Caqti_lwt.connection -> ('res, 'err) query_result
+
+  let get_world : id:int -> (Models.world, 'err) query =
+    let open Models in
+    [%rapper
+      get_one
+        {sql|
+        SELECT @int{id}, @int{randomNumber} FROM World
+        WHERE id = %int{id}
+    |sql}
+        record_out]
+
+  let get_fortunes : unit -> (Models.fortune list, 'err) query =
+    let open Models in
+    [%rapper
+      get_many
+        {sql|
+        SELECT @int{id}, @string{message} FROM Fortune
+    |sql}
+        record_out]
+
+  let update_random_number : random_number:int -> id:int -> (unit, 'err) query =
+    [%rapper
+      execute
+        {sql|
+        UPDATE World
+        SET randomNumber = %int{random_number}
+        WHERE id = %int{id}
+    |sql}]
+end
+
+module Middleware = struct
+  module Env = struct
+    let key = Hmap.Key.create ()
+  end
+
+  let get_db (request : Morph.Request.t) = Hmap.get Env.key request.ctx
+
+  let use request query =
+    let pool = get_db request in
+    Caqti_lwt.Pool.use query pool
+
+  let get_world request (id : int) : (Models.world, string) Lwt_result.t =
+    use request (Query.get_world ~id) |> Lwt_result.map_err Caqti_error.show
+
+  let get_fortunes request : (Models.fortune list, string) Lwt_result.t =
+    use request (Query.get_fortunes ()) |> Lwt_result.map_err Caqti_error.show
+
+  let update_random_number request ~random_number id :
+      (unit, string) Lwt_result.t =
+    use request (Query.update_random_number ~random_number ~id)
+    |> Lwt_result.map_err Caqti_error.show
+
+  let middleware ~db (handler : Morph.Server.handler)
+      (request : Morph.Request.t) =
+    let ctx = Hmap.add Env.key db request.ctx in
+    let next_request = { request with ctx } in
+    handler next_request
+end

frameworks/OCaml/morph/src/bin/Caqti.mli

@@ -0,0 +1,36 @@
+module Archi : sig
+  val start :
+    unit ->
+    ( (Caqti_lwt.connection, [> Caqti_error.connect ]) Caqti_lwt.Pool.t,
+      string )
+    result
+    Lwt.t
+
+  val stop :
+    (Caqti_lwt.connection, [> Caqti_error.connect ]) Caqti_lwt.Pool.t ->
+    unit Lwt.t
+
+  val component :
+    ( unit,
+      (Caqti_lwt.connection, Caqti_error.t) Caqti_lwt.Pool.t )
+    Archi_lwt.Component.t
+end
+
+module Middleware : sig
+  val use :
+    Morph.Request.t ->
+    ((module Caqti_lwt.CONNECTION) -> ('a, Caqti_error.t) Lwt_result.t) ->
+    ('a, Caqti_error.t) Lwt_result.t
+
+  val get_world : Morph.Request.t -> int -> (Models.world, string) Lwt_result.t
+
+  val get_fortunes :
+    Morph.Request.t -> (Models.fortune list, string) Lwt_result.t
+
+  val update_random_number :
+    Morph.Request.t -> random_number:int -> int -> (unit, string) result Lwt.t
+
+  val middleware :
+    db:((module Caqti_lwt.CONNECTION), Caqti_error.t) Caqti_lwt.Pool.t ->
+    Morph.Server.middleware
+end

frameworks/OCaml/morph/src/bin/Db_middleware.re

@@ -1,18 +0,0 @@
-module Env = {
-  let key = Hmap.Key.create();
-};
-
-let get_db = (request: Morph.Request.t) => Hmap.get(Env.key, request.ctx);
-
-let use = (request, query) => {
-  let pool = get_db(request);
-  Caqti_lwt.Pool.use(query, pool);
-};
-
-let middleware:
-  (~db: Caqti_lwt.Pool.t(module Caqti_lwt.CONNECTION, 'e)) =>
-  Morph.Server.middleware =
-  (~db, handler, request) => {
-    let next_request = {...request, ctx: Hmap.add(Env.key, db, request.ctx)};
-    handler(next_request);
-  };

frameworks/OCaml/morph/src/bin/Db_middleware.rei

@@ -1,10 +0,0 @@
-let use:
-  (
-    Morph.Request.t,
-    (module Caqti_lwt.CONNECTION) => Lwt.t(result('a, Caqti_error.t))
-  ) =>
-  Lwt.t(result('a, Caqti_error.t));
-
-let middleware:
-  (~db: Caqti_lwt.Pool.t(module Caqti_lwt.CONNECTION, Caqti_error.t)) =>
-  Morph.Server.middleware;

frameworks/OCaml/morph/src/bin/Handlers.re

@@ -1,27 +1,22 @@
-let select_random =
-  Caqti_request.find(
-    Caqti_type.int,
-    Caqti_type.(tup2(int, int)),
-    "SELECT id, randomNumber FROM World WHERE id = $1",
-  );
+let random_int = () => Random.int(10000) + 1;
 
 let text = _req => {
   Morph.Response.text("Hello, World!") |> Lwt.return;
 };
 
-let json = _req => {
-  let json = `Assoc([("message", `String("Hello, World!"))]);
-  Yojson.Safe.to_string(json) |> Morph.Response.json |> Lwt.return;
-};
+let json: Morph.Server.handler =
+  _req => {
+    let json = `Assoc([("message", `String("Hello, World!"))]);
+    Yojson.Safe.to_string(json) |> Morph.Response.json |> Lwt.return;
+  };
 
 let db = req => {
   open Lwt_result.Infix;
-  let read_db' = (module C: Caqti_lwt.CONNECTION) =>
-    C.find(select_random, Random.int(10000 + 1));
-  Db_middleware.use(req, read_db')
-  |> Lwt_result.map_err(e => `Server(Caqti_error.show(e)))
+  let id = random_int();
+  Caqti.Middleware.get_world(req, id)
+  |> Lwt_result.map_err(e => `Server(e))
   >>= (
-    ((id, randomNumber)) => {
+    (Models.{id, randomNumber}) => {
       `Assoc([("id", `Int(id)), ("randomNumber", `Int(randomNumber))])
       |> Yojson.Safe.to_string
       |> Morph.Response.json
@@ -30,34 +25,46 @@ let db = req => {
   );
 };
 
-let queries = (queries, req: Morph.Request.t) => {
-  open Lwt.Infix;
+let queries = (count, req: Morph.Request.t) => {
+  open Lwt.Syntax;
 
-  let query_ids = List.init(queries, _ => Random.int(10000 + 1));
+  let query_ids = List.init(count, _ => random_int());
 
-  let read_query' = (x, module C: Caqti_lwt.CONNECTION) => {
-    C.find(select_random, x);
-  };
+  let+ worlds_json =
+    List.map(
+      id => {
+        let+ result = Caqti.Middleware.get_world(req, id);
+        switch (result) {
+        | Error(_) => failwith("failed to query")
+        | Ok(Models.{id, randomNumber}) =>
+          `Assoc([("id", `Int(id)), ("randomNumber", `Int(randomNumber))])
+        };
+      },
+      query_ids,
+    )
+    |> Lwt.all;
+
+  let json_string = Yojson.Safe.to_string(`List(worlds_json));
+  Morph.Response.json(json_string);
+};
 
-  let pool = Db_middleware.use(req);
-
-  query_ids
-  |> List.map(id =>
-       pool(read_query'(id))
-       >|= (
-         fun
-         | Ok((id, randomNumber)) =>
-           Ok(
-             `Assoc([
-               ("id", `Int(id)),
-               ("randomNumber", `Int(randomNumber)),
-             ]),
-           )
-         | Error(err) => Error(err)
-       )
-     )
-  |> Lwt.all
-  >|= List.filter_map(Result.to_option)
-  >|= (l => Yojson.Safe.to_string(`List(l)))
-  >|= Morph.Response.json;
+let respond_html = elt => {
+  Morph.Response.html(Format.asprintf("%a", Tyxml.Html.pp(), elt));
+};
+
+let fortunes = req => {
+  open Lwt.Syntax;
+  let+ result = Caqti.Middleware.get_fortunes(req);
+
+  let additional_fortune =
+    Models.{id: 0, message: "Additional fortune added at request time."};
+
+  switch (result) {
+  | Ok(fortunes) =>
+    [additional_fortune, ...fortunes]
+    |> List.sort(Models.compare_fortune)
+    |> View.fortunes_page
+    |> respond_html
+  | Error(str) => Error(`Server(str))
+  };
 };

frameworks/OCaml/morph/src/bin/Handlers.rei

@@ -5,3 +5,5 @@ let json: Morph.Server.handler;
 let db: Morph.Server.handler;
 
 let queries: int => Morph.Server.handler;
+
+let fortunes: Morph.Server.handler;

frameworks/OCaml/morph/src/bin/Headers_middleware.ml

@@ -1,36 +1,38 @@
 let weekday_to_string = function
-| `Mon -> "Mon"
-| `Tue -> "Tue"
-| `Wed -> "Wed"
-| `Thu -> "Thu"
-| `Fri -> "Fri"
-| `Sat -> "Sat"
-| `Sun -> "Sun"
+  | `Mon -> "Mon"
+  | `Tue -> "Tue"
+  | `Wed -> "Wed"
+  | `Thu -> "Thu"
+  | `Fri -> "Fri"
+  | `Sat -> "Sat"
+  | `Sun -> "Sun"
 
 let month_to_string = function
-| 1 -> "Jan"
-| 2 -> "Feb"
-| 3 -> "Mar"
-| 4 -> "Apr"
-| 5 -> "May"
-| 6 -> "Jun"
-| 7 -> "Jul"
-| 8 -> "Aug"
-| 9 -> "Sep"
-| 10 -> "Oct"
-| 11 -> "Nov"
-| 12 -> "Dec"
-| _ -> failwith "month"
+  | 1 -> "Jan"
+  | 2 -> "Feb"
+  | 3 -> "Mar"
+  | 4 -> "Apr"
+  | 5 -> "May"
+  | 6 -> "Jun"
+  | 7 -> "Jul"
+  | 8 -> "Aug"
+  | 9 -> "Sep"
+  | 10 -> "Oct"
+  | 11 -> "Nov"
+  | 12 -> "Dec"
+  | _ -> failwith "month"
 
 let timestamp now =
   let weekday_str = weekday_to_string (Ptime.weekday now) in
-  let ((year, month, day), ((hour, minute, second), _)) = Ptime.to_date_time now in
+  let (year, month, day), ((hour, minute, second), _) =
+    Ptime.to_date_time now
+  in
   let month_str = month_to_string month in
-  Printf.sprintf "%s, %02u %s %04u %02u:%02u:%02u GMT" weekday_str day month_str year hour minute second
+  Printf.sprintf "%s, %02u %s %04u %02u:%02u:%02u GMT" weekday_str day month_str
+    year hour minute second
 
-let make (handler: Morph.Server.handler) request =
+let make (handler : Morph.Server.handler) request =
   let open Lwt.Infix in
-  handler(request) >|= Morph.Response.add_headers [
-    ("Date", (timestamp (Ptime_clock.now ())));
-    ("Server", "Morph");
-  ] 
+  handler request
+  >|= Morph.Response.add_headers
+        [ ("Date", timestamp (Ptime_clock.now ())); ("Server", "Morph") ]

frameworks/OCaml/morph/src/bin/Models.ml

@@ -0,0 +1,5 @@
+type world = { id : int; randomNumber : int }
+
+type fortune = { id : int; message : string }
+
+let compare_fortune a b = String.compare a.message b.message

frameworks/OCaml/morph/src/bin/Morph_wrapper.ml

@@ -0,0 +1,7 @@
+let client_address_key : Unix.sockaddr Hmap.key = Hmap.Key.create ()
+
+let wrap_context (next : Morph.Server.handler)
+    (req : Unix.sockaddr Piaf.Server.ctx) =
+  let open Lwt.Infix in
+  let ctx = Hmap.add client_address_key req.ctx Hmap.empty in
+  next { req with ctx } >|= Morph.Response.response_of_result

frameworks/OCaml/morph/src/bin/Router.ml

@@ -4,33 +4,32 @@ let json_route : (Morph.Server.handler, 'a) Routes.target =
 let plaintext_route : (Morph.Server.handler, 'a) Routes.target =
   Routes.(s "plaintext" /? nil)
 
-let db_route : (Morph.Server.handler, 'a) Routes.target =
-  Routes.(s "db" /? nil)
+let db_route : (Morph.Server.handler, 'a) Routes.target = Routes.(s "db" /? nil)
 
 let query_route_missing : (Morph.Server.handler, 'a) Routes.target =
   Routes.(s "queries" //? nil)
 
-let query_route_int =
-  Routes.(s "queries" / int /? nil)
+let query_route_int = Routes.(s "queries" / int /? nil)
 
+let query_route_str = Routes.(s "queries" / str /? nil)
 
-let query_route_str =
-  Routes.(s "queries" / str /? nil)
+let fortunes_route = Routes.(s "fortunes" /? nil)
 
-let not_found_handler _request =
-  Morph.Response.not_found () |> Lwt.return
+let not_found_handler _request = Morph.Response.not_found () |> Lwt.return
 
-let routes = Routes.[
-  json_route @--> Handlers.json;
-  plaintext_route @--> Handlers.text;
-  db_route @--> Handlers.db;
-  query_route_missing @--> Handlers.queries 1;
-  query_route_int @--> (function
-  | queries when queries > 500 -> Handlers.queries 500
-  | queries when queries < 1 -> Handlers.queries 1
-  | queries -> Handlers.queries queries
-  );
-  query_route_str @--> (fun _ -> Handlers.queries 1);
-]
+let routes =
+  Routes.
+    [
+      json_route @--> Handlers.json;
+      plaintext_route @--> Handlers.text;
+      db_route @--> Handlers.db;
+      query_route_missing @--> Handlers.queries 1;
+      (query_route_int @--> function
+       | queries when queries > 500 -> Handlers.queries 500
+       | queries when queries < 1 -> Handlers.queries 1
+       | queries -> Handlers.queries queries);
+      (query_route_str @--> fun _ -> Handlers.queries 1);
+      fortunes_route @--> Handlers.fortunes;
+    ]
 
-let handler = Morph.Router.make ~get:routes ~print:true ~not_found_handler ();
+let handler = Morph.Router.make ~get:routes ~print:true ~not_found_handler ()

frameworks/OCaml/morph/src/bin/View.re

@@ -0,0 +1,34 @@
+open Tyxml;
+
+module FortuneRow = {
+  let createElement = (~fortune: Models.fortune, ()) => {
+    <tr>
+      <td> {Html.txt(string_of_int(fortune.id))} </td>
+      <td> {Html.txt(fortune.message)} </td>
+    </tr>;
+  };
+};
+
+module FortunesTable = {
+  let table_header = <tr> <th> "id" </th> <th> "message" </th> </tr>;
+
+  let createElement = (~fortunes: list(Models.fortune), ()) => {
+    let table_content:
+      Html.list_wrap(Html.elt([< Html_types.table_content_fun])) = [
+      table_header,
+      ...List.map(fortune => <FortuneRow fortune />, fortunes),
+    ];
+    Html.table(table_content);
+  };
+};
+
+module Layout = {
+  let createElement = (~title, ~children, ()) =>
+    <html>
+      <head> <title> {Html.txt(title)} </title> </head>
+      <body> ...children </body>
+    </html>;
+};
+
+let fortunes_page = fortunes =>
+  <Layout title="Fortunes"> <FortunesTable fortunes /> </Layout>;

frameworks/OCaml/morph/src/bin/dune

@@ -1,13 +1,28 @@
+(* -*- tuareg -*- *)
+
+let server_io =
+  match Sys.getenv_opt "SERVER_IO" with
+  | Some "NPROC" -> "server_io_nproc"
+  | Some "SINGLE" -> "server_io_single"
+  | _ -> "server_io_single"
+
+let () = Jbuild_plugin.V1.send @@ {|
 (executable
  (name tfb)
  (public_name tfb)
- (link_flags -ccopt -lpq -cclib -lpgport -cclib -lpgcommon -ccopt
-   %{env:LDFLAGS=-lpq} -cclib %{env:LIBEV_LIBS=-lev})
+ (ocamlopt_flags
+  (:standard -O3 -unbox-closures))
  (libraries caqti caqti-lwt caqti-driver-postgresql lwt logs logs.fmt fmt.tty
-   morph archi archi-lwt routes yojson ptime ptime.clock ptime.clock.os))
+   morph archi archi-lwt routes yojson ptime ptime.clock ptime.clock.os
+   ppx_rapper.runtime tyxml
+   |} ^ server_io ^ {|)
+  (preprocess
+    (pps ppx_rapper tyxml-jsx)))
 
 (env
  (docker
   (flags
-   (:standard -ccopt -static -ccopt %{env:CFLAGS=/lib} -cclib
-     %{env:LDFLAGS=-lpq} -cclib -L%{env:LD_LIBRARY_PATH=/lib}))))
+   (:standard -ccopt -static -I /usr/include -I /usr/lib -cclib -lldap -cclib
+     -lcrypto -cclib -lssl -cclib -lpq -cclib -lpgport -cclib -lpgcommon
+     -cclib -lev))))
+|}

frameworks/OCaml/morph/src/bin/tfb.re

@@ -1,68 +1,60 @@
-module Database = {
-  let start =
-      ()
-      : Lwt.t(
-          result(
-            Caqti_lwt.Pool.t(module Caqti_lwt.CONNECTION, Caqti_error.t),
-            string,
-          ),
-        ) => {
-    let connection_url = "postgresql://benchmarkdbuser:benchmarkdbpass@tfb-database:5432/hello_world?connect_timeout=15";
-
-    Caqti_lwt.connect_pool(~max_size=10, Uri.of_string(connection_url))
-    |> (
-      fun
-      | Ok(pool) => Ok(pool)
-      | Error(error) => Error(Caqti_error.show(error))
-    )
-    |> Lwt.return;
-  };
-
-  let stop = _pool => {
-    Logs.info(m => m("Disconnected from database")) |> Lwt.return;
-  };
+let error_handler = (_client_addr, ~request as _=?, ~respond, err) => {
+  let error_to_string =
+    fun
+    | `Bad_gateway => "Bad gateway"
+    | `Bad_request => "Bad request"
+    | `Exn(_exn) => "Unhandled server error"
+    | `Internal_server_error => "Internal server error";
+
+  let error_handler =
+    respond(
+      ~headers=Piaf.Headers.of_list([("connection", "close")]),
+      Piaf.Body.of_string(error_to_string(err)),
+    );
 
-  let component = Archi_lwt.Component.make(~start, ~stop);
+  Lwt.return(error_handler);
 };
 
 module WebServer = {
-  let port =
-    try(Sys.getenv("PORT") |> int_of_string) {
-    | _ => 8080
-    };
-
-  let server = Morph.Server.make(~port, ~address=Unix.inet_addr_any, ());
-
   let start = ((), db) => {
-    Logs.app(m => m("Starting server on %n", port));
+    let port =
+      switch (Sys.getenv_opt("PORT")) {
+      | Some(p) => int_of_string(p)
+      | None => 8080
+      };
 
-    let handler =
-      Headers_middleware.make(Db_middleware.middleware(~db, Router.handler));
+    Logs.app(m => m("Starting server on %n", port));
 
-    server.start(handler) |> Lwt_result.ok;
+    let request_handler =
+      Headers_middleware.make(
+        Caqti.Middleware.middleware(~db, Router.handler),
+      )
+      |> Morph_wrapper.wrap_context;
+    Server_io.listen(~request_handler, ~error_handler, port)
+    |> Lwt_result.return;
   };
 
   let stop = _server => {
     Logs.info(m => m("Stopped Server")) |> Lwt.return;
   };
 
-  let component =
+  let component: Archi_lwt.Component.t(unit, unit) =
     Archi_lwt.Component.using(
       ~start,
       ~stop,
-      ~dependencies=[Database.component],
+      ~dependencies=[Caqti.Archi.component],
     );
 };
 
 let system =
   Archi_lwt.System.make([
-    ("database", Database.component),
+    ("database", Caqti.Archi.component),
     ("web_server", WebServer.component),
   ]);
 
 let main = () => {
   open Lwt.Infix;
-  Logger.setup_log(Some(Logs.Info));
+  Logger.setup_log(Some(Logs.Warning));
 
   Archi_lwt.System.start((), system)
   >|= (

frameworks/OCaml/morph/src/dune-project

@@ -1,4 +1,4 @@
-(lang dune 2.5)
+(lang dune 2.6)
 
 (name morph-tfb)
 
@@ -16,8 +16,10 @@
    (>= 4.11.0))
   archi
   archi-lwt
-  pgx
-  pgx_lwt_unix
+  caqti-driver-postgresql
+  caqti
+  caqti-lwt
+  conf-libev
   dune
   fmt
   logs
@@ -26,4 +28,8 @@
   uri
   yojson
   morph
-  piaf))
+  piaf
+  reason
+  tyxml
+  tyxml-jsx
+  ppx_rapper))

frameworks/OCaml/morph/src/esy.json

@@ -41,7 +41,7 @@
   },
   "resolutions": {
     "@reason-native-web/morph": "reason-native-web/morph:morph.json#c7bdd60dbfb99eccd4ce27559117919455061e0e",
-    "@opam/caqti": "ulrikstrid/ocaml-caqti:caqti.opam#c0ead31422c3670f8250a50043a1038cab4f011f",
+    "@opam/caqti": "paurkedal/ocaml-caqti:caqti.opam#4a09a3f5fab4f15c90f1cf1790688f6e6bb94185",
     "@opam/conf-postgresql": "ulrikstrid/postgresql:package.json#193d72af7f2132e30ad429401232da6359b248a2",
     "@opam/conf-pkg-config": "esy-ocaml/yarn-pkg-config#db3a0b63883606dd57c54a7158d560d6cba8cd79",
     "@opam/session": "inhabitedtype/ocaml-session:session.opam#6180413"

frameworks/OCaml/morph/src/morph-tfb.opam

@@ -7,9 +7,11 @@ depends: [
   "ocaml" {>= "4.11.0"}
   "archi"
   "archi-lwt"
-  "pgx"
-  "pgx_lwt_unix"
-  "dune"
+  "caqti-driver-postgresql"
+  "caqti"
+  "caqti-lwt"
+  "conf-libev"
+  "dune" {>= "2.6"}
   "fmt"
   "logs"
   "lwt"
@@ -18,6 +20,10 @@ depends: [
   "yojson"
   "morph"
   "piaf"
+  "reason"
+  "tyxml"
+  "tyxml-jsx"
+  "ppx_rapper"
 ]
 build: [
   ["dune" "subst"] {pinned}

frameworks/OCaml/morph/src/server_io/dune

@@ -0,0 +1,6 @@
+(library
+ (name server_io)
+ (libraries lwt lwt.unix piaf unix)
+ (modules server_io)
+ (virtual_modules server_io)
+ (default_implementation server_io_single))

frameworks/OCaml/morph/src/server_io/server_io.mli

@@ -0,0 +1,11 @@
+val listen : request_handler:Unix.sockaddr Piaf.Server.Handler.t ->
+    error_handler:(Unix.sockaddr ->
+                   ?request:Piaf.Request.t ->
+                   respond:(headers:Piaf.Headers.t ->
+                            Piaf.Body.t -> Piaf.Server.Error_response.t) ->
+                   [ `Bad_gateway
+                   | `Bad_request
+                   | `Exn of exn
+                   | `Internal_server_error ] ->
+                   Piaf.Server.Error_response.t Lwt.t) ->
+    int -> unit

frameworks/OCaml/morph/src/server_io_nproc/dune

@@ -0,0 +1,4 @@
+(library
+ (name server_io_nproc)
+ (implements server_io)
+ (libraries lwt lwt.unix piaf unix))

frameworks/OCaml/morph/src/server_io_nproc/server_io.ml

@@ -0,0 +1,195 @@
+(* This code is borrowed from this PR: https://github.com/anmonteiro/piaf/pull/80 *)
+open Lwt.Infix
+
+type server = { shutdown: unit Lwt.t Lazy.t  }
+
+let dump_lwt () =
+  let options =
+    [ "fd_passing", `fd_passing
+    ; "fdatasync", `fdatasync
+    ; "get_affinity", `get_affinity
+    ; "get_cpu", `get_cpu
+    ; "get_credentials", `get_credentials
+    ; "libev", `libev
+    ; "madvise", `madvise
+    ; "mincore", `mincore
+    ; "recv_msg", `recv_msg
+    ; "send_msg", `send_msg
+    ; "set_affinity", `set_affinity
+    ; "wait4", `wait4
+    ]
+  in
+  Format.eprintf "Lwt:\n%a@."
+    (Format.pp_print_list
+      ~pp_sep:(fun fmt () -> Format.fprintf fmt "@\n")
+      (fun fmt (str, opt) -> Format.fprintf fmt "  %s = %b" str (Lwt_sys.have opt)))
+    options
+
+let close_socket fd =
+  Lwt.finalize
+    (fun () ->
+       Lwt.catch
+         (fun () ->
+            Lwt_unix.shutdown fd Unix.SHUTDOWN_ALL;
+            Lwt.return_unit)
+         (function
+           (* Occurs if the peer closes the connection first. *)
+           | Unix.Unix_error (Unix.ENOTCONN, _, _) -> Lwt.return_unit
+           | exn -> Lwt.fail exn))
+    (fun () ->
+       Lwt_unix.close fd)
+
+let establish_server_generic
+    ?fd:preexisting_socket_for_listening
+    listening_address
+    connection_handler_callback =
+
+  let listening_socket =
+    match preexisting_socket_for_listening with
+    | None ->
+      Lwt_unix.socket
+        (Unix.domain_of_sockaddr listening_address) Unix.SOCK_STREAM 0
+    | Some socket ->
+      socket
+  in
+  Lwt_unix.setsockopt listening_socket Unix.SO_REUSEADDR true;
+
+  (* This promise gets resolved with `Should_stop when the user calls
+     Lwt_io.shutdown_server. This begins the shutdown procedure. *)
+  let should_stop, notify_should_stop =
+    Lwt.wait () in
+
+  (* Some time after Lwt_io.shutdown_server is called, this function
+     establish_server_generic will actually close the listening socket. At that
+     point, this promise is resolved. This ends the shutdown procedure. *)
+  let wait_until_listening_socket_closed, notify_listening_socket_closed =
+    Lwt.wait () in
+
+  let rec accept_loop () =
+    let try_to_accept =
+      Lwt_unix.accept listening_socket >|= fun x ->
+      `Accepted x
+    in
+
+    Lwt.pick [try_to_accept; should_stop] >>= function
+    | `Accepted (client_socket, client_address) ->
+      begin
+        try Lwt_unix.set_close_on_exec client_socket
+        with Invalid_argument _ -> ()
+      end;
+
+      connection_handler_callback client_address client_socket;
+
+      accept_loop ()
+
+    | `Should_stop ->
+      Lwt_unix.close listening_socket >>= fun () ->
+
+      begin match listening_address with
+      | Unix.ADDR_UNIX path when path <> "" && path.[0] <> '\x00' ->
+        Unix.unlink path
+      | _ ->
+        ()
+      end [@ocaml.warning "-4"];
+
+      Lwt.wakeup_later notify_listening_socket_closed ();
+      Lwt.return_unit
+  in
+
+  let server =
+    {shutdown =
+      lazy begin
+        Lwt.wakeup_later notify_should_stop `Should_stop;
+        wait_until_listening_socket_closed
+      end}
+  in
+
+  (* Actually start the server. *)
+  let server_has_started =
+    (* bind_function listening_socket listening_address >>= fun () -> *)
+    (* Lwt_unix.listen listening_socket backlog; *)
+
+    Lwt.async accept_loop;
+
+    Lwt.return_unit
+  in
+
+  server, server_has_started
+
+let establish_server_with_client_socket
+    ?server_fd ?(no_close = false) sockaddr f
+  =
+  let handler client_address client_socket =
+    Lwt.async (fun () ->
+        (* Not using Lwt.finalize here, to make sure that exceptions from [f]
+           reach !Lwt.async_exception_hook before exceptions from closing the
+           channels. *)
+        Lwt.catch
+          (fun () -> f client_address client_socket)
+          (fun exn ->
+            !Lwt.async_exception_hook exn;
+            Lwt.return_unit)
+        >>= fun () ->
+        if no_close then
+          Lwt.return_unit
+        else if Lwt_unix.state client_socket = Lwt_unix.Closed then
+          Lwt.return_unit
+        else
+          Lwt.catch
+            (fun () -> close_socket client_socket)
+            (fun exn ->
+              !Lwt.async_exception_hook exn;
+              Lwt.return_unit))
+  in
+  let server, server_started =
+    establish_server_generic
+      ?fd:server_fd
+      sockaddr
+      handler
+  in
+  server_started >>= fun () -> Lwt.return server
+
+
+let listen ~request_handler ~error_handler port =
+  let nproc =
+    Unix.open_process_in "getconf _NPROCESSORS_ONLN"
+    |> input_line
+    |> int_of_string
+  in
+  Format.eprintf "Detected %d cores@." nproc;
+  let ulimit_n =
+    Unix.open_process_in "ulimit -n" |> input_line |> int_of_string
+  in
+  Format.eprintf "Detected %d max open files@." ulimit_n;
+  dump_lwt ();
+  let sockaddr = Unix.(ADDR_INET (inet_addr_any, port)) in
+  let socket =
+    Lwt_unix.socket (Unix.domain_of_sockaddr sockaddr) Unix.SOCK_STREAM 0
+  in
+  Lwt_unix.setsockopt socket Unix.SO_REUSEADDR true;
+  Lwt_main.run
+  @@ ( Lwt_unix.bind socket sockaddr >|= fun () ->
+  Lwt_unix.listen socket 10_000 );
+  for i = 1 to nproc do
+    flush_all ();
+    let pid = Lwt_unix.fork () in
+    if pid = 0 then (
+      (* child *)
+      (Lwt.async_exception_hook := fun exn -> raise exn);
+      Lwt.async (fun () ->
+          establish_server_with_client_socket
+            ~server_fd:socket
+            sockaddr
+            (Piaf.Server.create ?config:None ~error_handler request_handler)
+          >|= fun _server ->
+          Format.eprintf "Listening on localhost:%i (child %d)@." port i);
+      let forever, _ = Lwt.wait () in
+      Lwt_main.run forever;
+      exit 0)
+  done;
+  while true do
+    Unix.pause ()
+  done
+
+let listen ~request_handler ~error_handler port =
+  Unix.handle_unix_error (listen ~request_handler ~error_handler) port

frameworks/OCaml/morph/src/server_io_single/dune

@@ -0,0 +1,4 @@
+(library
+ (name server_io_single)
+ (implements server_io)
+ (libraries lwt lwt.unix piaf unix))

frameworks/OCaml/morph/src/server_io_single/server_io.ml

@@ -0,0 +1,10 @@
+let listen ~request_handler ~error_handler port =
+  let open Lwt.Infix in
+  let listen_address = Unix.(ADDR_INET (inet_addr_any, port)) in
+  Lwt.async (fun () ->
+      Lwt_io.establish_server_with_client_socket listen_address
+        (Piaf.Server.create ?config:None ~error_handler request_handler)
+      >|= fun _server ->
+      Printf.printf "Listening on port %i and echoing POST requests.\n%!" port);
+  let forever, _ = Lwt.wait () in
+  Lwt_main.run forever

frameworks/OCaml/opium/benchmark_config.json

@@ -1,6 +1,7 @@
 {
   "framework": "opium",
-  "tests": [{
+  "tests": [
+    {
       "default": {
         "json_url": "/json",
         "plaintext_url": "/plaintext",
@@ -47,5 +48,6 @@
         "notes": "",
         "versus": "httpaf"
       }
-  }]
+    }
+  ]
 }