Added a max_keep_alive and moved to a non blocking reading of request not to loose connection slot.

Default is max_keep_alive = -1.0 which keeps the original behaviour.

src/Tiny_httpd.ml

@@ -83,6 +83,27 @@ module Byte_stream = struct
   let of_chan = of_chan_ ~close:close_in
   let of_chan_close_noerr = of_chan_ ~close:close_in_noerr
 
+  exception Timeout
+
+  let of_descr_ ?(timeout=(-1.0)) ~close ic : t =
+    let i = ref 0 in
+    let len = ref 0 in
+    let buf = Bytes.make 4096 ' ' in
+    { bs_fill_buf=(fun () ->
+      if !i >= !len then (
+        i := 0;
+        let (to_read,_,_) = Unix.select [ic] [] [] timeout in
+        if to_read = [] then raise Timeout;
+        try len := Unix.read ic buf 0 (Bytes.length buf)
+        with Unix.Unix_error ((EAGAIN | EWOULDBLOCK), _, _) -> ();
+      );
+      buf, !i,!len - !i);
+      bs_consume=(fun n -> i := !i + n);
+      bs_close=(fun () -> close ic)
+    }
+
+  let of_descr = of_descr_ ~close:Unix.close
+
   let rec iter f (self:t) : unit =
     let s, i, len = self.bs_fill_buf () in
     if len=0 then (
@@ -511,6 +532,9 @@ module Request = struct
                 headers; body=()})
     with
     | End_of_file | Sys_error _ -> Ok None
+    | Byte_stream.Timeout ->
+       _debug (fun k -> k"Timeout");
+       Ok None
     | Bad_req (c,s) -> Error (c,s)
     | e ->
       Error (400, Printexc.to_string e)
@@ -697,8 +721,10 @@ module Sem_ = struct
   let release m t =
     Mutex.lock t.mutex;
     t.n <- t.n + m;
+    let r = t.n in
     Condition.broadcast t.cond;
-    Mutex.unlock t.mutex
+    Mutex.unlock t.mutex;
+    r
 end
 
 module Route = struct
@@ -814,6 +840,9 @@ type t = {
   sem_max_connections: Sem_.t;
   (* semaphore to restrict the number of active concurrent connections *)
 
+  max_keep_alive: float;
+  (* maximum time in second before closing the client connections *)
+
   new_thread: (unit -> unit) -> unit;
   (* a function to run the given callback in a separate thread (or thread pool) *)
 
@@ -948,13 +977,14 @@ let add_route_server_sent_handler ?accept self route f =
 let create
     ?(masksigpipe=true)
     ?(max_connections=32)
+    ?(max_keep_alive=(-1.0))
     ?(new_thread=(fun f -> ignore (Thread.create f () : Thread.t)))
     ?(addr="127.0.0.1") ?(port=8080) ?sock () : t =
   let handler _req = Response.fail ~code:404 "no top handler" in
   let max_connections = max 4 max_connections in
   { new_thread; addr; port; sock; masksigpipe; handler;
     running= true; sem_max_connections=Sem_.create max_connections;
-    path_handlers=[];
+    path_handlers=[]; max_keep_alive;
     cb_encode_resp=[]; cb_decode_req=[];
   }
 
@@ -970,10 +1000,10 @@ let find_map f l =
   in aux f l
 
 let handle_client_ (self:t) (client_sock:Unix.file_descr) : unit =
-  let ic = Unix.in_channel_of_descr client_sock in
+  let _ = Unix.set_nonblock client_sock in
   let oc = Unix.out_channel_of_descr client_sock in
   let buf = Buf_.create() in
-  let is = Byte_stream.of_chan ic in
+  let is = Byte_stream.of_descr ~timeout:self.max_keep_alive client_sock in
   let continue = ref true in
   while !continue && self.running do
     _debug (fun k->k "read next request");
@@ -1087,17 +1117,20 @@ let run (self:t) : (unit,_) result =
     end;
     while self.running do
       (* limit concurrency *)
-      Sem_.acquire 1 self.sem_max_connections;
       try
         let client_sock, _ = Unix.accept sock in
+        Sem_.acquire 1 self.sem_max_connections;
         self.new_thread
           (fun () ->
             try
               handle_client_ self client_sock;
-              Sem_.release 1 self.sem_max_connections;
+              let avail = Sem_.release 1 self.sem_max_connections in
-            with e ->
+              _debug (fun k -> k"closing inactive connections (%d connections available)" avail)
+            with
+            | e ->
               (try Unix.close client_sock with _ -> ());
-              Sem_.release 1 self.sem_max_connections;
+              let avail = Sem_.release 1 self.sem_max_connections in
+              _debug (fun k -> k"closing connections on error (%d connections available)" avail);
               raise e
           );
       with e ->

src/Tiny_httpd.mli

@@ -434,6 +434,7 @@ type t
 val create :
   ?masksigpipe:bool ->
   ?max_connections:int ->
+  ?max_keep_alive:float ->
   ?new_thread:((unit -> unit) -> unit) ->
   ?addr:string ->
   ?port:int ->
@@ -453,7 +454,10 @@ val create :
     new client connection. By default it is {!Thread.create} but one
     could use a thread pool instead.
 
-    @param max_connections maximum number of simultaneous connections.
+    @param max_connections maximum number of simultaneous connections. Default 32.
+    @param max_keep_alive maximum number of seconds a thread in maintened for
+      a client with nothing to read. Default: -1.0, meaning threads are maintened
+      until client close the socket.
     @param addr address (IPv4 or IPv6) to listen on. Default ["127.0.0.1"].
     @param port to listen on. Default [8080].
     @param sock an existing socket given to the server to listen on, e.g. by
@@ -639,4 +643,3 @@ val _debug : ((('a, out_channel, unit, unit, unit, unit) format6 -> 'a) -> unit)
 val _enable_debug: bool -> unit
 
 (**/**)
-