Merge 5cb6744f1e into 20a36919ce

examples/fuseau/dune

@@ -0,0 +1,5 @@
+(executable
+ (name echo)
+ (flags :standard -warn-error -a+8)
+ (modules echo)
+ (libraries tiny_httpd tiny_httpd.fuseau fuseau.unix logs tiny_httpd_camlzip))

examples/fuseau/echo.ml

@@ -0,0 +1,236 @@
+module S = Tiny_httpd
+module Log = Tiny_httpd.Log
+
+let ( let@ ) = ( @@ )
+let now_ = Unix.gettimeofday
+
+let alice_text =
+  "CHAPTER I. Down the Rabbit-Hole  Alice was beginning to get very tired of \
+   sitting by her sister on the bank, and of having nothing to do: once or \
+   twice she had peeped into the book her sister was reading, but it had no \
+   pictures or conversations in it, <and what is the use of a book,> thought \
+   Alice <without pictures or conversations?> So she was considering in her \
+   own mind (as well as she could, for the hot day made her feel very sleepy \
+   and stupid), whether the pleasure of making a daisy-chain would be worth \
+   the trouble of getting up and picking the daisies, when suddenly a White \
+   Rabbit with pink eyes ran close by her. There was nothing so very \
+   remarkable in that; nor did Alice think it so very much out of the way to \
+   hear the Rabbit say to itself, <Oh dear! Oh dear! I shall be late!> (when \
+   she thought it over afterwards, it occurred to her that she ought to have \
+   wondered at this, but at the time it all seemed quite natural); but when \
+   the Rabbit actually took a watch out of its waistcoat-pocket, and looked at \
+   it, and then hurried on, Alice started to her feet, for it flashed across \
+   her mind that she had never before seen a rabbit with either a \
+   waistcoat-pocket, or a watch to take out of it, and burning with curiosity, \
+   she ran across the field after it, and fortunately was just in time to see \
+   it pop down a large rabbit-hole under the hedge. In another moment down \
+   went Alice after it, never once considering how in the world she was to get \
+   out again. The rabbit-hole went straight on like a tunnel for some way, and \
+   then dipped suddenly down, so suddenly that Alice had not a moment to think \
+   about stopping herself before she found herself falling down a very deep \
+   well. Either the well was very deep, or she fell very slowly, for she had \
+   plenty of time as she went down to look about her and to wonder what was \
+   going to happen next. First, she tried to look down and make out what she \
+   was coming to, but it was too dark to see anything; then she looked at the \
+   sides of the well, and noticed that they were filled with cupboards......"
+
+(* util: a little middleware collecting statistics *)
+let middleware_stat () : S.Middleware.t * (unit -> string) =
+  let n_req = ref 0 in
+  let total_time_ = ref 0. in
+  let parse_time_ = ref 0. in
+  let build_time_ = ref 0. in
+  let write_time_ = ref 0. in
+
+  let m h req ~resp =
+    incr n_req;
+    let t1 = S.Request.start_time req in
+    let t2 = now_ () in
+    h req ~resp:(fun response ->
+        let t3 = now_ () in
+        resp response;
+        let t4 = now_ () in
+        total_time_ := !total_time_ +. (t4 -. t1);
+        parse_time_ := !parse_time_ +. (t2 -. t1);
+        build_time_ := !build_time_ +. (t3 -. t2);
+        write_time_ := !write_time_ +. (t4 -. t3))
+  and get_stat () =
+    Printf.sprintf
+      "%d requests (average response time: %.3fms = %.3fms + %.3fms + %.3fms)"
+      !n_req
+      (!total_time_ /. float !n_req *. 1e3)
+      (!parse_time_ /. float !n_req *. 1e3)
+      (!build_time_ /. float !n_req *. 1e3)
+      (!write_time_ /. float !n_req *. 1e3)
+  in
+  m, get_stat
+
+let setup_logging () =
+  Logs.set_reporter @@ Logs.format_reporter ();
+  Logs.set_level ~all:true (Some Logs.Debug)
+
+let () =
+  let@ () = Fuseau_unix.main in
+  let port_ = ref 8080 in
+  Arg.parse
+    (Arg.align
+       [
+         "--port", Arg.Set_int port_, " set port";
+         "-p", Arg.Set_int port_, " set port";
+         "--debug", Arg.Unit setup_logging, " enable debug";
+       ])
+    (fun _ -> raise (Arg.Bad ""))
+    "echo [option]*";
+
+  let buf_pool = S.create_buf_pool () in
+  let server =
+    S.create_from ~buf_pool
+      ~backend:(Tiny_httpd_fuseau.io_backend ~buf_pool ~port:!port_ ())
+      ()
+  in
+
+  Tiny_httpd_camlzip.setup ~compress_above:1024 ~buf_size:(16 * 1024) server;
+  let m_stats, get_stats = middleware_stat () in
+  S.add_middleware server ~stage:(`Stage 1) m_stats;
+
+  (* say hello *)
+  S.add_route_handler ~meth:`GET server
+    S.Route.(exact "hello" @/ string @/ return)
+    (fun name _req -> S.Response.make_string (Ok ("hello " ^ name ^ "!\n")));
+
+  (* compressed file access *)
+  S.add_route_handler ~meth:`GET server
+    S.Route.(exact "zcat" @/ string_urlencoded @/ return)
+    (fun path _req ->
+      let ic = open_in path in
+      let str = S.Byte_stream.of_chan ic in
+      let mime_type =
+        try
+          let p = Unix.open_process_in (Printf.sprintf "file -i -b %S" path) in
+          try
+            let s = [ "Content-Type", String.trim (input_line p) ] in
+            ignore @@ Unix.close_process_in p;
+            s
+          with _ ->
+            ignore @@ Unix.close_process_in p;
+            []
+        with _ -> []
+      in
+      S.Response.make_stream ~headers:mime_type (Ok str));
+
+  (* echo request *)
+  S.add_route_handler server
+    S.Route.(exact "echo" @/ return)
+    (fun req ->
+      let q =
+        S.Request.query req
+        |> List.map (fun (k, v) -> Printf.sprintf "%S = %S" k v)
+        |> String.concat ";"
+      in
+      S.Response.make_string
+        (Ok (Format.asprintf "echo:@ %a@ (query: %s)@." S.Request.pp req q)));
+
+  (* file upload *)
+  S.add_route_handler_stream ~meth:`PUT server
+    S.Route.(exact "upload" @/ string @/ return)
+    (fun path req ->
+      Log.debug (fun k ->
+          k "start upload %S, headers:\n%s\n\n%!" path
+            (Format.asprintf "%a" S.Headers.pp (S.Request.headers req)));
+      try
+        let oc = open_out @@ "/tmp/" ^ path in
+        S.Byte_stream.to_chan oc req.S.Request.body;
+        flush oc;
+        S.Response.make_string (Ok "uploaded file")
+      with e ->
+        S.Response.fail ~code:500 "couldn't upload file: %s"
+          (Printexc.to_string e));
+
+  (* logout *)
+  S.add_route_handler server
+    S.Route.(exact "logout" @/ return)
+    (fun _req -> S.Response.fail ~code:401 "logged out");
+
+  (* stats *)
+  S.add_route_handler server
+    S.Route.(exact "stats" @/ return)
+    (fun _req ->
+      let stats = get_stats () in
+      S.Response.make_string @@ Ok stats);
+
+  S.add_route_handler server
+    S.Route.(exact "alice" @/ return)
+    (fun _req -> S.Response.make_string (Ok alice_text));
+
+  (* main page *)
+  S.add_route_handler server
+    S.Route.(return)
+    (fun _req ->
+      let open Tiny_httpd_html in
+      let h =
+        html []
+          [
+            head [] [ title [] [ txt "index of echo" ] ];
+            body []
+              [
+                h3 [] [ txt "welcome!" ];
+                p [] [ b [] [ txt "endpoints are:" ] ];
+                ul []
+                  [
+                    li [] [ pre [] [ txt "/hello/:name (GET)" ] ];
+                    li []
+                      [
+                        pre []
+                          [
+                            a [ A.href "/echo/" ] [ txt "echo" ];
+                            txt " echo back query";
+                          ];
+                      ];
+                    li []
+                      [ pre [] [ txt "/upload/:path (PUT) to upload a file" ] ];
+                    li []
+                      [
+                        pre []
+                          [
+                            txt
+                              "/zcat/:path (GET) to download a file (deflate \
+                               transfer-encoding)";
+                          ];
+                      ];
+                    li []
+                      [
+                        pre []
+                          [
+                            a [ A.href "/stats/" ] [ txt "/stats/" ];
+                            txt " (GET) to access statistics";
+                          ];
+                      ];
+                    li []
+                      [
+                        pre []
+                          [
+                            a [ A.href "/protected" ] [ txt "/protected" ];
+                            txt
+                              " (GET) to see a protected page (login: user, \
+                               password: foobar)";
+                          ];
+                      ];
+                    li []
+                      [
+                        pre []
+                          [
+                            a [ A.href "/logout" ] [ txt "/logout" ];
+                            txt " (POST) to log out";
+                          ];
+                      ];
+                  ];
+              ];
+          ]
+      in
+      let s = to_string_top h in
+      S.Response.make_string ~headers:[ "content-type", "text/html" ] @@ Ok s);
+
+  Printf.printf "listening on http://%s:%d\n%!" (S.addr server) (S.port server);
+  match S.run server with
+  | Ok () -> ()
+  | Error e -> raise e

src/Tiny_httpd_server.ml

@@ -1,5 +1,6 @@
 type buf = Tiny_httpd_buf.t
 type byte_stream = Tiny_httpd_stream.t
+type buf_pool = buf Tiny_httpd_pool.t
 
 module Buf = Tiny_httpd_buf
 module Byte_stream = Tiny_httpd_stream
@@ -871,7 +872,20 @@ let get_max_connection_ ?(max_connections = 64) () : int =
   let max_connections = max 4 max_connections in
   max_connections
 
-let create_from ?(buf_size = 16 * 1_024) ?(middlewares = []) ~backend () : t =
+let _default_buf_size = 16 * 1024
+
+let create_buf_pool ?(buf_size = _default_buf_size) () : buf_pool =
+  Pool.create ~clear:Buf.clear_and_zero
+    ~mk_item:(fun () -> Buf.create ~size:buf_size ())
+    ()
+
+let create_from ?(buf_size = _default_buf_size) ?buf_pool ?(middlewares = [])
+    ~backend () : t =
+  let buf_pool =
+    match buf_pool with
+    | Some p -> p
+    | None -> create_buf_pool ~buf_size ()
+  in
   let handler _req = Response.fail ~code:404 "no top handler" in
   let self =
     {
@@ -882,10 +896,7 @@ let create_from ?(buf_size = 16 * 1_024) ?(middlewares = []) ~backend () : t =
       path_handlers = [];
       middlewares = [];
       middlewares_sorted = lazy [];
-      buf_pool =
-        Pool.create ~clear:Buf.clear_and_zero
-          ~mk_item:(fun () -> Buf.create ~size:buf_size ())
-          ();
+      buf_pool;
     }
   in
   List.iter (fun (stage, m) -> add_middleware self ~stage m) middlewares;
@@ -1223,6 +1234,7 @@ let client_handle_for (self : t) ~client_addr ic oc : unit =
            try
              if Headers.get "connection" r.Response.headers = Some "close" then
                continue := false;
+             Log.debug (fun k -> k "got response: %a" Response.pp r);
              log_response req r;
              Response.output_ ~buf:buf_res oc r
            with Sys_error _ -> continue := false

src/Tiny_httpd_server.mli

@@ -10,6 +10,7 @@
 
 type buf = Tiny_httpd_buf.t
 type byte_stream = Tiny_httpd_stream.t
+type buf_pool = buf Tiny_httpd_pool.t
 
 (** {2 HTTP Methods} *)
 
@@ -469,8 +470,13 @@ module type IO_BACKEND = sig
       on a port and handle clients. *)
 end
 
+val create_buf_pool : ?buf_size:int -> unit -> buf_pool
+(** [create_buf_pool ()] creates a new buffer pool.
+   @since NEXT_RELEASE *)
+
 val create_from :
   ?buf_size:int ->
+  ?buf_pool:buf_pool ->
   ?middlewares:([ `Encoding | `Stage of int ] * Middleware.t) list ->
   backend:(module IO_BACKEND) ->
   unit ->

src/fuseau/dune

@@ -0,0 +1,7 @@
+
+(library
+  (name tiny_httpd_fuseau)
+  (public_name tiny_httpd.fuseau)
+  (synopsis "Tiny_httpd running on Fuseau")
+  (optional)
+  (libraries tiny_httpd mtime mtime.clock.os fuseau fuseau.unix))

src/fuseau/tiny_httpd_fuseau.ml

@@ -0,0 +1,174 @@
+module F = Fuseau_unix
+module IO = Tiny_httpd_io
+module Pool = Tiny_httpd_pool
+module Buf = Tiny_httpd_buf
+
+let ( let@ ) = ( @@ )
+
+module Server = struct
+  type t = {
+    addr: Unix.inet_addr;
+    port: int;
+    server_sock: Unix.file_descr;
+    buf_pool: Buf.t Pool.t;
+    mutable active: bool;
+    mutable n_connections: int;
+  }
+
+  let create ~buf_pool ~addr ~port () : t =
+    let sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
+    Unix.setsockopt sock Unix.SO_REUSEADDR true;
+    Unix.setsockopt sock Unix.SO_REUSEPORT true;
+    Unix.bind sock (Unix.ADDR_INET (addr, port));
+    Unix.set_nonblock sock;
+    Unix.listen sock 32;
+    let self =
+      {
+        addr;
+        port;
+        server_sock = sock;
+        active = true;
+        n_connections = 0;
+        buf_pool;
+      }
+    in
+    self
+
+  let ic_of_fd ~(buf : Buf.t) ~close fd : IO.Input.t =
+    let buf = Buf.bytes_slice buf in
+    let buf_i = ref 0 in
+    let buf_len = ref 0 in
+    let eof = ref false in
+
+    let refill () =
+      if not !eof then (
+        buf_i := 0;
+        buf_len := F.IO_unix.read fd buf 0 (Bytes.length buf);
+        if !buf_len = 0 then eof := true
+      )
+    in
+
+    let input bs i len =
+      if !buf_len = 0 then refill ();
+      let n = min len !buf_len in
+      Bytes.blit buf !buf_i bs i n;
+      buf_i := !buf_i + n;
+      buf_len := !buf_len - n;
+      n
+    in
+
+    { input; close }
+
+  let oc_of_fd ~buf ~close fd : IO.Output.t =
+    let buf = Buf.bytes_slice buf in
+    let off = ref 0 in
+
+    let flush () =
+      if !off > 0 then (
+        F.IO_unix.write fd buf 0 !off;
+        off := 0
+      )
+    in
+    let[@inline] maybe_flush () = if !off = Bytes.length buf then flush () in
+
+    let output_char c =
+      maybe_flush ();
+      Bytes.set buf !off c;
+      incr off
+    in
+
+    let output bs i len =
+      let i = ref i in
+      let len = ref len in
+      while !len > 0 do
+        maybe_flush ();
+        let n = min !len (Bytes.length buf - !off) in
+        Bytes.blit bs !i buf !off n;
+        off := !off + n;
+        i := !i + n;
+        len := !len - n
+      done
+    in
+    { output; output_char; flush; close }
+
+  type conn_handler = Tiny_httpd_io.TCP_server.conn_handler
+
+  let loop_client self ~(handler : conn_handler) client_sock client_addr : unit
+      =
+    Unix.set_nonblock client_sock;
+    Unix.setsockopt client_sock Unix.TCP_NODELAY true;
+
+    (* idempotent close *)
+    let closed = ref false in
+    let close () =
+      if not !closed then (
+        closed := true;
+        Unix.shutdown client_sock Unix.SHUTDOWN_ALL;
+        Unix.close client_sock
+      )
+    in
+
+    let@ buf_ic = Pool.with_resource self.buf_pool in
+    let@ buf_oc = Pool.with_resource self.buf_pool in
+    let ic = ic_of_fd ~buf:buf_ic ~close client_sock in
+    let oc = oc_of_fd ~buf:buf_oc ~close client_sock in
+    let finally () =
+      self.n_connections <- self.n_connections - 1;
+      close ()
+    in
+    let@ () = Fun.protect ~finally in
+    handler.handle ~client_addr ic oc
+
+  let loop (self : t) ~(handler : Tiny_httpd_io.TCP_server.conn_handler) : unit
+      =
+    while self.active do
+      match Unix.accept self.server_sock with
+      | client_sock, client_addr ->
+        self.n_connections <- 1 + self.n_connections;
+        ignore
+          (Fuseau.spawn ~propagate_cancel_to_parent:false (fun () ->
+               loop_client self ~handler client_sock client_addr)
+            : unit F.Fiber.t)
+      | exception Unix.Unix_error ((Unix.EAGAIN | Unix.EWOULDBLOCK), _, _) ->
+        (* FIXME: possible race condition: the socket became readable
+            in the mid-time and we won't get notified. We need to call
+            [accept] after subscribing to [on_readable]. *)
+        F.IO_unix.await_readable self.server_sock
+    done
+
+  let close self =
+    if self.active then (
+      self.active <- false;
+      try Unix.close self.server_sock with _ -> ()
+    )
+end
+
+let io_backend ~buf_pool ?(addr = Unix.inet_addr_loopback) ~port () :
+    (module Tiny_httpd_server.IO_BACKEND) =
+  let module M = struct
+    let init_addr () = Unix.string_of_inet_addr addr
+    let init_port () = port
+
+    let get_time_s () =
+      let t_ns = Mtime_clock.now () |> Mtime.to_uint64_ns in
+      Int64.to_float t_ns *. 1e-9
+
+    let serve ~after_init ~(handle : Tiny_httpd_io.TCP_server.conn_handler) () :
+        unit =
+      let server = Server.create ~buf_pool ~addr ~port () in
+      let server' : Tiny_httpd_io.TCP_server.t =
+        {
+          endpoint = (fun () -> Unix.string_of_inet_addr addr, port);
+          active_connections = (fun () -> server.n_connections);
+          running = (fun () -> server.active);
+          stop = (fun () -> server.active <- false);
+        }
+      in
+
+      after_init server';
+      Server.loop server ~handler:handle;
+      ()
+
+    let tcp_server () : Tiny_httpd_io.TCP_server.builder = { serve }
+  end in
+  (module M)

src/fuseau/tiny_httpd_fuseau.mli

@@ -0,0 +1,6 @@
+val io_backend :
+  buf_pool:Tiny_httpd_buf.t Tiny_httpd_pool.t ->
+  ?addr:Unix.inet_addr ->
+  port:int ->
+  unit ->
+  (module Tiny_httpd_server.IO_BACKEND)