ocaml opium process forking (#6163)

* opium forks

* start date refreshes

* add missing changes

frameworks/OCaml/opium/README.md

@@ -24,6 +24,17 @@
 It doesn't have a dependency on any web server to make it more portable and allow easier addition to new webservers.
 Feel free to copy paste when adding additional servers.
 
+## Local development
+
+Either use the docker images or `make pin && make install` and then `make run` or `make run-forks` 
+
+## Variants
+
+- opium - base implementation, uses a single process
+- opium-haproxy - starts multiple processes on different ports and uses haproxy to distribute the load
+- opium-fedora-forks - starts multiple processes listening on the same socket
+- opium-alpine-forks - same as the above, but uses `alpine` instead of `fedora`
+
 ## Test URLs
 
 ### PLAINTEXT

frameworks/OCaml/opium/benchmark_config.json

@@ -25,6 +25,52 @@
         "notes": "",
         "versus": "httpaf"
       },
+      "fedora-forks": {
+        "json_url": "/json",
+        "plaintext_url": "/plaintext",
+        "db_url": "/db",
+        "query_url": "/queries/",
+        "update_url": "/updates/",
+        "fortune_url": "/fortunes",
+        "port": 8080,
+        "approach": "Realistic",
+        "classification": "Micro",
+        "database": "postgres",
+        "framework": "opium",
+        "language": "OCaml",
+        "flavor": "None",
+        "orm": "Micro",
+        "platform": "httpaf",
+        "webserver": "None",
+        "os": "Linux",
+        "database_os": "Linux",
+        "display_name": "opium-fedora-forks",
+        "notes": "",
+        "versus": "httpaf"
+      },
+      "alpine-forks": {
+        "json_url": "/json",
+        "plaintext_url": "/plaintext",
+        "db_url": "/db",
+        "query_url": "/queries/",
+        "update_url": "/updates/",
+        "fortune_url": "/fortunes",
+        "port": 8080,
+        "approach": "Realistic",
+        "classification": "Micro",
+        "database": "postgres",
+        "framework": "opium",
+        "language": "OCaml",
+        "flavor": "None",
+        "orm": "Micro",
+        "platform": "httpaf",
+        "webserver": "None",
+        "os": "Linux",
+        "database_os": "Linux",
+        "display_name": "opium-alpine-forks",
+        "notes": "",
+        "versus": "httpaf"
+      },
       "haproxy": {
         "json_url": "/json",
         "plaintext_url": "/plaintext",

frameworks/OCaml/opium/opium-alpine-forks.dockerfile

@@ -0,0 +1,21 @@
+FROM ocurrent/opam:alpine-3.12-ocaml-4.11
+
+ENV DIR web
+# 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 add --no-cache make m4 postgresql-dev libev-dev libffi-dev linux-headers
+
+WORKDIR /${DIR}
+
+COPY src/opi.opam src/Makefile ./
+
+RUN make install-ci
+
+COPY ./src ./
+
+RUN sudo chown -R opam: . && make build
+
+ENTRYPOINT _build/default/bin/main_forks.exe

frameworks/OCaml/opium/opium-fedora-forks.dockerfile

@@ -0,0 +1,21 @@
+FROM ocurrent/opam:fedora-32-ocaml-4.11
+
+ENV DIR web
+# 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 dnf install --assumeyes postgresql-devel libev-devel libffi-devel
+
+WORKDIR /${DIR}
+
+COPY src/opi.opam src/Makefile ./
+
+RUN make install-ci
+
+COPY ./src ./
+
+RUN sudo chown -R opam: . && make build
+
+ENTRYPOINT _build/default/bin/main_forks.exe

frameworks/OCaml/opium/opium-haproxy.dockerfile

@@ -14,6 +14,8 @@ COPY src/opi.opam src/Makefile ./
 
 RUN make install-ci
 
+ENV APP_INSTANCES 1
+
 COPY ./src ./
 
 RUN sudo chown -R opam: . && make build

frameworks/OCaml/opium/src/Makefile

@@ -19,10 +19,10 @@ gen-opam:
 	dune build @install
 
 build:
-	opam exec -- dune build --profile release bin/$(BINARY).exe
+	opam exec -- dune build --profile release
 
 run:
-	./_build/default/bin/$(BINARY).exe
-
-run-debug:
 	dune exec $(project_name) -- --debug
+
+run-forks:
+	dune exec $(project_name)_forks -- --debug

frameworks/OCaml/opium/src/bin/dune

@@ -1,4 +1,11 @@
 (executable
  (public_name opi)
  (name main)
- (libraries opi opium caqti caqti-driver-postgresql caqti-lwt tyxml lwt.unix))
+ (modules main)
+ (libraries server lwt.unix))
+
+(executable
+ (public_name opi_forks)
+ (name main_forks)
+ (modules main_forks)
+ (libraries server lwt.unix))

frameworks/OCaml/opium/src/bin/main.ml

@@ -1,42 +1,17 @@
 open Opium.Std
 
-module Tfb_headers = struct
-  let middleware =
-    let open Lwt.Syntax in
-    let filter handler req =
-      let+ res = handler req in
-      let headers = ["Server", "opium"; "Date", Opi.Time.now ()] in
-      Response.add_headers_or_replace headers res
-    in
-    Rock.Middleware.create ~name:"TFB Headers" ~filter
-end
-
 let main () =
   let port =
     match Sys.getenv_opt "PORT" with
     | Some x -> int_of_string x
     | None -> 8080
   in
-  let routes =
-    [
-      "/plaintext", Routes.plaintext;
-      "/json", Routes.json;
-      "/db", Routes.single_query;
-      "/fortunes", Routes.fortunes;
-      "/queries/", Routes.multiple_queries;
-      "/queries/:count", Routes.multiple_queries;
-      "/updates/", Routes.updates;
-      "/updates/:count", Routes.updates
-    ]
-  in
-  let add_routes app = List.fold_left (fun app (route,handler) -> (get route handler) app) app routes in
-  let app : Opium.App.t =
-    App.empty
-    |> App.cmd_name "Opium"
-    |> App.port port
-    |> middleware Middleware.content_length
-    |> middleware Tfb_headers.middleware
-    |> add_routes in
+
+  Server.dump_lwt ();
+
+  let app = Server.create_app ~port in
+
+  Server.start_refreshing_date ();
 
   match App.run_command' app with
   | `Ok (app : unit Lwt.t ) ->

frameworks/OCaml/opium/src/bin/main_forks.ml

@@ -0,0 +1,69 @@
+open Opium.Std
+open Lwt.Syntax
+
+let run_app app ~instances ~port =
+  let listen_address =
+    let inet_addr = Unix.inet_addr_any in
+    Unix.ADDR_INET (inet_addr, port)
+  in
+  let socket =
+    Lwt_unix.socket (Unix.domain_of_sockaddr listen_address) Unix.SOCK_STREAM 0
+  in
+  Lwt_unix.setsockopt socket Unix.SO_REUSEADDR true;
+
+  Lwt_main.run (
+    let+ () = Lwt_unix.bind socket listen_address in
+    Lwt_unix.listen socket (Lwt_unix.somaxconn () [@ocaml.warning "-3"])
+  );
+
+  let rec accept_loop socket handler instance =
+    let* (socket', sockaddr') = Lwt_unix.accept socket in
+    Lwt.async (fun () -> handler sockaddr' socket');
+    accept_loop socket handler instance
+  in
+
+  let rock = App.to_rock app in
+
+  for i = 1 to instances do
+    flush_all ();
+    if Lwt_unix.fork () = 0 then (
+      (* child *)
+      Server.start_refreshing_date ();
+      Lwt.async (fun () ->
+          let* () = Lwt_io.eprintf "Listening on %d (child %d)\n" port i in
+          let handler = Server.create_connection_handler rock in
+          accept_loop socket handler i
+        );
+      let forever, _ = Lwt.wait () in
+      Lwt_main.run forever;
+      exit 0)
+  done;
+
+  while true do
+    Unix.pause ()
+  done
+
+
+let main () =
+  let port =
+    match Sys.getenv_opt "PORT" with
+    | Some x -> int_of_string x
+    | None -> 8080
+  in
+  let instances =
+    match Sys.getenv_opt "APP_INSTANCES" with
+    | Some x -> int_of_string x
+    | None ->
+      let ic = Unix.open_process_in "getconf _NPROCESSORS_ONLN" in
+      let cores = int_of_string (input_line ic) in
+      ignore (Unix.close_process_in ic);
+      cores
+  in
+
+  Server.dump_lwt ();
+  Printf.eprintf "Starting %d instances\n" instances;
+
+  let app = Server.create_app ~port in
+  run_app app ~instances ~port
+
+let () = main ()

frameworks/OCaml/opium/src/lib/routes.ml

@@ -41,7 +41,7 @@ let updates count =
       let+ result = Db.Update.update_world updated_world_req in
       match result with
       | Error _ -> failwith "failed queries"
-      | Ok w -> {world with randomNumber=updated_random_number}
+      | Ok () -> {world with randomNumber=updated_random_number}
     ) in
   let+ updated_worlds = Lwt.all results in
   Models.World.list_response_to_yojson updated_worlds

frameworks/OCaml/opium/src/server/dune

@@ -0,0 +1,3 @@
+(library
+ (name server)
+ (libraries opi opium caqti caqti-driver-postgresql caqti-lwt tyxml lwt.unix))

frameworks/OCaml/opium/src/server/server.ml

@@ -0,0 +1,80 @@
+open Opium.Std
+open Lwt.Syntax
+
+module Tfb_headers = struct
+  let memo_date = ref (Opi.Time.now ())
+
+  let start_refreshing_date () = 
+    let refresh_date _sig =
+      memo_date := Opi.Time.now ();
+      ignore (Unix.alarm 1)
+    in
+    ignore (Sys.(signal sigalrm (Signal_handle refresh_date)));
+    refresh_date ()
+
+  let middleware =
+    let filter handler req =
+      let+ res = handler req in
+      let headers = ["Server", "opium"; "Date", !memo_date] in
+      Response.add_headers_or_replace headers res
+    in
+
+    Rock.Middleware.create ~name:"TFB Headers" ~filter
+end
+
+let start_refreshing_date = Tfb_headers.start_refreshing_date
+
+(* lwt debugging information *)
+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
+  Printf.eprintf "Lwt:\n";
+  List.iter
+    (fun (str, opt) -> Printf.eprintf "  %s = %b\n" str (Lwt_sys.have opt))
+    options
+
+let create_connection_handler t addr fd=
+  let f ~request_handler ~error_handler =
+    Httpaf_lwt_unix.Server.create_connection_handler
+      ~request_handler:(fun _ -> request_handler)
+      ~error_handler:(fun _ -> error_handler)
+      addr
+      fd
+  in
+  Opium_kernel.Server_connection.run f t
+
+let create_app ~port =
+  let routes =
+    [
+      "/plaintext", Routes.plaintext;
+      "/json", Routes.json;
+      "/db", Routes.single_query;
+      "/fortunes", Routes.fortunes;
+      "/queries/", Routes.multiple_queries;
+      "/queries/:count", Routes.multiple_queries;
+      "/updates/", Routes.updates;
+      "/updates/:count", Routes.updates
+    ]
+  in
+  let add_routes app = List.fold_left (fun app (route,handler) -> (get route handler) app) app routes in
+
+  App.empty
+  |> App.cmd_name "Opium"
+  |> App.port port
+  |> middleware Middleware.content_length
+  |> middleware Tfb_headers.middleware
+  |> add_routes