Establish server with Lwt_io and rename benchmark.

benchmarks/jbuild

@@ -7,5 +7,5 @@
 
 (executables
  ((libraries (httpaf httpaf-lwt-unix lwt lwt.unix))
-  (modules (wrk_lwt_benchmark))
-  (names (wrk_lwt_benchmark))))
+  (modules (wrk_lwt_unix_benchmark))
+  (names (wrk_lwt_unix_benchmark))))

benchmarks/wrk_lwt_benchmark.ml → benchmarks/wrk_lwt_unix_benchmark.ml

@@ -6,6 +6,7 @@ let text = "CHAPTER I. Down the Rabbit-Hole  Alice was beginning to get very tir
 let text = Bigstring.of_string text
 
 let headers = Headers.of_list ["content-length", string_of_int (Bigstring.length text)]
+
 let error_handler _ ?request:_ error start_response =
   let response_body = start_response Headers.empty in
   begin match error with
@@ -16,7 +17,6 @@ let error_handler _ ?request:_ error start_response =
     Body.write_string response_body (Status.default_reason_phrase error)
   end;
   Body.close_writer response_body
-;;
 
 let request_handler _ reqd =
   let {Request.target; _} = Reqd.request reqd in
@@ -26,46 +26,35 @@ let request_handler _ reqd =
   | "/" -> Reqd.respond_with_bigstring reqd (Response.create ~headers `OK) text;
   | _   -> Reqd.respond_with_string    reqd (Response.create `Not_found) "Route not found"
 
-let main port max_accepts_per_batch =
-  let conn_count = ref 0 in
-  let sock = Lwt_unix.socket Lwt_unix.PF_INET Lwt_unix.SOCK_STREAM 0 in
+let connection_handler =
+  Httpaf_lwt_unix.Server.create_connection_handler
+    ~error_handler ~request_handler
+
+let main port =
   let sockaddr = Lwt_unix.ADDR_INET (Unix.inet_addr_loopback, port) in
-  Lwt_unix.bind sock sockaddr >>= fun () ->
-  Lwt_unix.listen sock 11_000;
-  let h = Httpaf_lwt_unix.Server.create_connection_handler ~error_handler ~request_handler in
+  let conn_count = ref 0 in
 
   let rec monitor () =
     Lwt_unix.sleep 0.5 >>= fun () ->
     Lwt_io.printlf "conns: %d" !conn_count >>= fun () ->
     monitor () in
   Lwt.async monitor;
 
-  let rec serve () =
-    Lwt_unix.accept_n sock max_accepts_per_batch >>= fun (accepts, exn) ->
-    begin match exn with
-    | None -> Lwt.return_unit
-    | Some exn -> Lwt_io.eprintlf "Accept failed: %s." (Printexc.to_string exn)
-    end >>= fun () ->
-    conn_count := !conn_count + List.length accepts;
-    accepts |> List.iter begin fun (sa, fd) ->
-      Lwt.async begin fun () ->
-        Lwt.catch
-          (fun () -> h fd sa)
-          (fun exn ->
-            Lwt_io.eprintlf "Failure while serving client: %s."
-                            (Printexc.to_string exn))
-        >|= fun () -> decr conn_count
-      end
-    end;
-    serve () in
-  serve ()
+  let handler sockaddr fd =
+    incr conn_count;
+    connection_handler sockaddr fd >|= fun () ->
+    decr conn_count in
+  Lwt.async begin fun () ->
+    Lwt_io.establish_server_with_client_socket ~backlog:10_000 sockaddr handler
+      >|= ignore
+  end;
+
+  fst (Lwt.wait ())
 
 let () =
   let port = ref 8080 in
-  let max_accepts_per_batch = ref 1 in
   Arg.parse
-    [ "-p", Arg.Set_int port, "int Source port to listen on";
-      "-a", Arg.Set_int max_accepts_per_batch, "int Maximum accepts per batch" ]
+    ["-p", Arg.Set_int port, "int Source port to listen on"]
     (fun _ -> raise (Arg.Bad "positional arg"))
     "Start a hello world Lwt server";
-  Lwt_main.run (main !port !max_accepts_per_batch)
+  Lwt_main.run (main !port)

examples/lwt_unix_echo_post.ml

@@ -39,30 +39,21 @@ let request_handler _ reqd =
   | _ ->
     Reqd.respond_with_string reqd (Response.create `Method_not_allowed) ""
 
-let main port max_accepts_per_batch =
-  let sock = Lwt_unix.socket Lwt_unix.PF_INET Lwt_unix.SOCK_STREAM 0 in
+let main port =
   let sockaddr = Lwt_unix.ADDR_INET (Unix.inet_addr_loopback, port) in
-  Lwt_unix.bind sock sockaddr >>= fun () ->
-  Lwt_unix.listen sock 10_000;
-  let h = Server.create_connection_handler ~error_handler ~request_handler in
-  let rec serve () =
-    Lwt_unix.accept_n sock max_accepts_per_batch >>= fun (accepts, exn) ->
-    begin match exn with
-    | None -> ()
-    | Some exn -> prerr_endline ("Accept failed: " ^ Printexc.to_string exn)
-    end;
-    List.iter (fun (sa, fd) -> Lwt.async (fun () -> h fd sa)) accepts;
-    serve () in
-  serve ()
+  let handler = Server.create_connection_handler ~error_handler ~request_handler in
+  Lwt.async begin fun () ->
+    Lwt_io.establish_server_with_client_socket ~backlog:10_000 sockaddr handler
+      >|= ignore
+  end;
+  fst (Lwt.wait ())
 
 let () =
   let port = ref 8080 in
-  let batch_capacity = ref 100 in
   Arg.parse
-    ["-p", Arg.Set_int port, " Port number to listen on.";
-     "-a", Arg.Set_int batch_capacity, " Maximum number of accepts per batch."]
+    ["-p", Arg.Set_int port, " Port number to listen on."]
     (fun _ ->
       prerr_endline "No posititonal arguments accepted.";
       exit 64)
     "lwt_unix_echo_post [-p PORT] [-a N-ACCEPT-PER-BATCH]";
-  Lwt_main.run (main !port !batch_capacity)
+  Lwt_main.run (main !port)

httpaf-lwt-unix.opam

@@ -18,6 +18,6 @@ depends: [
   "angstrom-lwt-unix"
   "faraday-lwt-unix"
   "httpaf"
-  "lwt"
+  "lwt" {>= "4.1.0"}
 ]
 available: [ ocaml-version >= "4.03.0" ]