[RFC] Initial port to Eio

src/dream.mli

@@ -2079,15 +2079,14 @@ val catch : (error -> response) -> middleware
 val run :
   ?interface:string ->
   ?port:int ->
-  ?stop:unit promise ->
   ?error_handler:error_handler ->
   ?https:bool ->
   ?certificate_file:string ->
   ?key_file:string ->
   ?builtins:bool ->
   ?greeting:bool ->
   ?adjust_terminal:bool ->
-  < clock: Eio.Time.clock; ..> ->
+  < clock:#Eio.Time.clock; net:#Eio.Net.t; ..> ->
     handler -> unit
 (** Runs the Web application represented by the {!handler}, by default at
     {{:http://localhost:8080} http://localhost:8080}.
@@ -2099,10 +2098,6 @@ val run :
     - [~interface] is the network interface to listen on. Defaults to
       ["localhost"]. Use ["0.0.0.0"] to listen on all interfaces.
     - [~port] is the port to listen on. Defaults to [8080].
-    - [~stop] is a promise that causes the server to stop accepting new
-      requests, and {!Dream.run} to return. Requests that have already entered
-      the Web application continue to be processed. The default value is a
-      promise that never resolves. However, see also [~stop_on_input].
     - [~debug:true] enables debug information in error templates. See
       {!Dream.error_template}. The default is [false], to prevent accidental
       deployment with debug output turned on. See example
@@ -2139,12 +2134,12 @@ val run :
 val serve :
   ?interface:string ->
   ?port:int ->
-  ?stop:unit promise ->
   ?error_handler:error_handler ->
   ?https:bool ->
   ?certificate_file:string ->
   ?key_file:string ->
   ?builtins:bool ->
+  net:#Eio.Net.t ->
     handler -> unit
 (** Like {!Dream.run}, but returns a promise that does not resolve until the
     server stops listening, instead of calling

src/http/http.ml

@@ -518,24 +518,26 @@ type tls_library = {
     key_file:string ->
     handler:Message.handler ->
     error_handler:Catch.error_handler ->
+    sw:Switch.t ->
       Unix.sockaddr ->
       Lwt_unix.file_descr ->
         unit Lwt.t;
 }
 
-let no_tls ~sw = {
+let no_tls = {
   create_handler = begin fun
       ~certificate_file:_ ~key_file:_
       ~handler
-      ~error_handler ->
+      ~error_handler
+      ~sw ->
     Httpaf_lwt_unix.Server.create_connection_handler
       ?config:None
       ~request_handler:(wrap_handler ~sw false error_handler handler)
       ~error_handler:(Error_handler.httpaf error_handler)
   end;
 }
 
-let openssl ~sw:_ = {
+let openssl = {
   create_handler = fun ~certificate_file:_ -> failwith "https://github.com/savonet/ocaml-ssl/issues/76"
 (*
   create_handler = begin fun
@@ -597,11 +599,12 @@ let openssl ~sw:_ = {
 }
 
 (* TODO LATER Add ALPN + HTTP/2.0 with ocaml-tls, too. *)
-let ocaml_tls ~sw = {
+let ocaml_tls = {
   create_handler = fun
       ~certificate_file ~key_file
       ~handler
-      ~error_handler ->
+      ~error_handler
+      ~sw ->
     Httpaf_lwt_unix.Server.TLS.create_connection_handler_with_default
       ~certfile:certificate_file ~keyfile:key_file
       ?config:None
@@ -620,12 +623,22 @@ let built_in_middleware error_handler =
 
 
 
+let of_unix_addr = function
+  | Unix.ADDR_INET (host, port) -> `Tcp (host, port)
+  | Unix.ADDR_UNIX path -> `Unix path
+
+let to_unix_addr = function
+  | `Tcp (host, port) -> Unix.ADDR_INET (host, port)
+  | `Unix path -> Unix.ADDR_UNIX path
+
+
+
 let serve_with_details
     caller_function_for_error_messages
     tls_library
+    ~net
     ~interface
     ~port
-    ~stop
     ~error_handler
     ~certificate_file
     ~key_file
@@ -669,36 +682,40 @@ let serve_with_details
      be pattern matching on the exception (but that might introduce dependency
      coupling), or the upstream should be patched to distinguish the errors in
      some useful way. *)
-  let httpaf_connection_handler client_address socket =
-    Lwt.catch
-      (fun () ->
-         httpaf_connection_handler client_address socket)
-      (fun exn ->
-         tls_error_handler client_address exn;
-         Lwt.return_unit)
+  let httpaf_connection_handler ~sw flow client_address =
+    let client_address = to_unix_addr client_address in
+    try
+      let fd = Eio_unix.FD.take flow |> Option.get in
+      let socket = Lwt_unix.of_unix_file_descr fd in
+      Lwt_eio.Promise.await_lwt @@
+      httpaf_connection_handler ~sw client_address socket
+    with exn ->
+      tls_error_handler client_address exn
   in
 
-  (* Look up the low-level address corresponding to the interface. Hopefully,
-     this is a local interface. *)
-  let%lwt addresses = Lwt_unix.getaddrinfo interface (string_of_int port) [] in
-  match addresses with
-  | [] ->
-    Printf.ksprintf failwith "Dream.%s: no interface with address %s"
-      caller_function_for_error_messages interface
-  | address::_ ->
-    let listen_address = Lwt_unix.(address.ai_addr) in
-
-
-    (* Bring up the HTTP server. Wait for the server to actually get started.
-       Then, wait for the ~stop promise. If the ~stop promise ever resolves, stop
-       the server. *)
-    let%lwt server =
-      Lwt_io.establish_server_with_client_socket
-        listen_address
-        httpaf_connection_handler in
+  let listen_address = Lwt_eio.Promise.await_lwt @@
+    (* Look up the low-level address corresponding to the interface. Hopefully,
+       this is a local interface. *)
+    let%lwt addresses = Lwt_unix.getaddrinfo interface (string_of_int port) [] in
+    match addresses with
+    | [] ->
+      Printf.ksprintf failwith "Dream.%s: no interface with address %s"
+        caller_function_for_error_messages interface
+    | address::_ ->
+      Lwt.return (of_unix_addr Lwt_unix.(address.ai_addr))
+  in
 
-    let%lwt () = stop in
-    Lwt_io.shutdown_server server
+  (* Bring up the HTTP server. *)
+  Switch.run @@ fun sw ->
+  let socket =
+    Eio.Net.listen ~sw net listen_address
+      ~reuse_addr:true
+      ~backlog:(Lwt_unix.somaxconn () [@ocaml.warning "-3"])
+  in
+  while true do
+    Eio.Net.accept_sub ~sw socket httpaf_connection_handler
+      ~on_error:(fun ex -> !Lwt.async_exception_hook ex)
+  done
 
 
 
@@ -709,17 +726,15 @@ let serve_with_maybe_https
     caller_function_for_error_messages
     ~interface
     ~port
-    ~stop
     ~error_handler
     ~https
     ?certificate_file ?key_file
     ?certificate_string ?key_string
     ~builtins
+    ~net
     user's_dream_handler =
 
-  Switch.run @@ fun sw ->
   try
-    Lwt_eio.Promise.await_lwt @@
     (* This check will at least catch secrets like "foo" when used on a public
        interface. *)
     (* if not (is_localhost interface) then
@@ -734,10 +749,10 @@ let serve_with_maybe_https
     | `No ->
       serve_with_details
         caller_function_for_error_messages
-        (no_tls ~sw)
+        no_tls
+        ~net
         ~interface
         ~port
-        ~stop
         ~error_handler
         ~certificate_file:""
         ~key_file:""
@@ -791,25 +806,26 @@ let serve_with_maybe_https
 
       let tls_library =
         match tls_library with
-        | `OpenSSL -> openssl ~sw
-        | `OCaml_TLS -> ocaml_tls ~sw
+        | `OpenSSL -> openssl
+        | `OCaml_TLS -> ocaml_tls
       in
 
       match certificate_and_key with
       | `File (certificate_file, key_file) ->
         serve_with_details
           caller_function_for_error_messages
           tls_library
+          ~net
           ~interface
           ~port
-          ~stop
           ~error_handler
           ~certificate_file
           ~key_file
           ~builtins
           user's_dream_handler
 
       | `Memory (certificate_string, key_string, verbose_or_silent) ->
+        Lwt_eio.Promise.await_lwt @@
         Lwt_io.with_temp_file begin fun (certificate_file, certificate_stream) ->
         Lwt_io.with_temp_file begin fun (key_file, key_stream) ->
 
@@ -825,16 +841,17 @@ let serve_with_maybe_https
         let%lwt () = Lwt_io.close certificate_stream in
         let%lwt () = Lwt_io.close key_stream in
 
+        Lwt_eio.run_eio @@ fun () ->
         serve_with_details
           caller_function_for_error_messages
           tls_library
           ~interface
           ~port
-          ~stop
           ~error_handler
           ~certificate_file
           ~key_file
           ~builtins
+          ~net
           user's_dream_handler
 
         end
@@ -853,26 +870,25 @@ let serve_with_maybe_https
 
 let default_interface = "localhost"
 let default_port = 8080
-let never = fst (Lwt.wait ())
 
 
 
 let serve
     ?(interface = default_interface)
     ?(port = default_port)
-    ?(stop = never)
     ?(error_handler = Error_handler.default)
     ?(https = false)
     ?certificate_file
     ?key_file
     ?(builtins = true)
+    ~net
     user's_dream_handler =
 
   serve_with_maybe_https
     "serve"
+    ~net
     ~interface
     ~port
-    ~stop
     ~error_handler
     ~https:(if https then `OCaml_TLS else `No)
     ?certificate_file
@@ -887,7 +903,6 @@ let serve
 let run
     ?(interface = default_interface)
     ?(port = default_port)
-    ?(stop = never)
     ?(error_handler = Error_handler.default)
     ?(https = false)
     ?certificate_file
@@ -964,9 +979,9 @@ let run
       Lwt_eio.with_event_loop ~clock:env#clock @@ fun () ->
       serve_with_maybe_https
         "run"
+        ~net:env#net
         ~interface
         ~port
-        ~stop
         ~error_handler
         ~https:(if https then `OCaml_TLS else `No)
         ?certificate_file ?key_file

src/server/helpers.ml

@@ -99,15 +99,6 @@ let redirect ?status ?code ?headers _request location =
   Message.set_header response "Location" location;
   response
 
-(* Ideally, we'd create a new Eio switch for each connection.
-   But connection creation happens from Lwt at the moment, so we just
-   push the exception back to Lwt to keep things working as before. *)
-let fork_from_lwt ~sw fn =
-  Fibre.fork ~sw (fun () ->
-      try fn ()
-      with ex -> !Lwt.async_exception_hook ex
-    )
-
 let get_switch request =
   match Message.field request switch_field with
   | Some sw -> sw
@@ -122,7 +113,7 @@ let stream ?status ?code ?headers request callback =
     Message.response ?status ?code ?headers client_stream server_stream in
   (* TODO Should set up an error handler for this. YES. *)
   (* TODO Make sure the request id is propagated to the callback. *)
-  let wrapped_callback _ = fork_from_lwt ~sw (fun () -> callback response) in
+  let wrapped_callback _ = Fibre.fork ~sw (fun () -> callback response) in
   Stream.ready server_stream ~close:wrapped_callback wrapped_callback;
   response
 
@@ -149,7 +140,7 @@ let websocket ?headers request callback =
       ~status:`Switching_Protocols ?headers client_stream server_stream in
   Message.set_field response websocket_field true;
   (* TODO Make sure the request id is propagated to the callback. *)
-  let wrapped_callback _ = fork_from_lwt ~sw (fun () -> callback response) in
+  let wrapped_callback _ = Fibre.fork ~sw (fun () -> callback response) in
   Stream.ready server_stream ~close:wrapped_callback wrapped_callback;
   response