Merge epoll feature branch to master (xapi-project#6005)

Pending some final testing by QA, these commits have already been
reviewed on the feature branch.

This switches `select` calls to `Unixext.select` (which is implemented
using `epoll`), and implements the few performance sensitive parts using
epoll directly.
It also enables some more tests with >1024 fds.

Author: edwintorok

ezxenstore.opam

@@ -14,6 +14,7 @@ depends: [
   "cmdliner" {with-test & >= "1.1.0"}
   "logs"
   "uuidm"
+  "xapi-stdext-unix"
   "xenctrl"
   "xenstore"
   "xenstore_transport"

ezxenstore.opam.template

@@ -12,6 +12,7 @@ depends: [
   "cmdliner" {with-test & >= "1.1.0"}
   "logs"
   "uuidm"
+  "xapi-stdext-unix"
   "xenctrl"
   "xenstore"
   "xenstore_transport"

ocaml/database/block_device_io.ml

@@ -328,7 +328,7 @@ let accept_conn s latest_response_time =
   let now = Unix.gettimeofday () in
   let timeout = latest_response_time -. now in
   (* Await an incoming connection... *)
-  let ready_to_read, _, _ = Unix.select [s] [] [] timeout in
+  let ready_to_read, _, _ = Xapi_stdext_unix.Unixext.select [s] [] [] timeout in
   R.info "Finished selecting" ;
   if List.mem s ready_to_read then
     (* We've received a connection. Accept it and return the socket. *)

ocaml/database/master_connection.ml

@@ -171,7 +171,11 @@ let open_secure_connection () =
     ~write_to_log:(fun x -> debug "stunnel: %s\n" x)
     ~verify_cert host port
   @@ fun st_proc ->
-  let fd_closed = Thread.wait_timed_read Unixfd.(!(st_proc.Stunnel.fd)) 5. in
+  let fd_closed =
+    Xapi_stdext_threads.Threadext.wait_timed_read
+      Unixfd.(!(st_proc.Stunnel.fd))
+      5.
+  in
   let proc_quit =
     try
       Unix.kill (Stunnel.getpid st_proc.Stunnel.pid) 0 ;

ocaml/forkexecd/src/child.ml

@@ -61,7 +61,9 @@ let handle_comms_sock comms_sock state =
 
 let handle_comms_no_fd_sock2 comms_sock fd_sock state =
   debug "Selecting in handle_comms_no_fd_sock2" ;
-  let ready, _, _ = Unix.select [comms_sock; fd_sock] [] [] (-1.0) in
+  let ready, _, _ =
+    Xapi_stdext_unix.Unixext.select [comms_sock; fd_sock] [] [] (-1.0)
+  in
   debug "Done" ;
   if List.mem fd_sock ready then (
     debug "fd sock" ;
@@ -74,7 +76,9 @@ let handle_comms_no_fd_sock2 comms_sock fd_sock state =
 
 let handle_comms_with_fd_sock2 comms_sock _fd_sock fd_sock2 state =
   debug "Selecting in handle_comms_with_fd_sock2" ;
-  let ready, _, _ = Unix.select [comms_sock; fd_sock2] [] [] (-1.0) in
+  let ready, _, _ =
+    Xapi_stdext_unix.Unixext.select [comms_sock; fd_sock2] [] [] (-1.0)
+  in
   debug "Done" ;
   if List.mem fd_sock2 ready then (
     debug "fd sock2" ;

ocaml/libs/ezxenstore/core/dune

@@ -9,5 +9,6 @@
     (re_export xenstore)
     (re_export xenstore_transport)
     threads.posix
+    xapi-stdext-unix
     (re_export xenstore.unix))
 )

ocaml/libs/ezxenstore/core/watch.ml

@@ -50,7 +50,7 @@ let wait_for ~xs ?(timeout = 300.) (x : 'a t) =
       let thread =
         Thread.create
           (fun () ->
-            let r, _, _ = Unix.select [p1] [] [] timeout in
+            let r, _, _ = Xapi_stdext_unix.Unixext.select [p1] [] [] timeout in
             if r <> [] then
               ()
             else

ocaml/libs/http-lib/buf_io.ml

@@ -74,21 +74,21 @@ let is_full ic = ic.cur = 0 && ic.max = Bytes.length ic.buf
 let fill_buf ~buffered ic timeout =
   let buf_size = Bytes.length ic.buf in
   let fill_no_exc timeout len =
-    let l, _, _ = Unix.select [ic.fd] [] [] timeout in
-    if l <> [] then (
+    Xapi_stdext_unix.Unixext.with_socket_timeout ic.fd timeout @@ fun () ->
+    try
       let n = Unix.read ic.fd ic.buf ic.max len in
       ic.max <- n + ic.max ;
       if n = 0 && len <> 0 then raise Eof ;
       n
-    ) else
-      -1
+    with Unix.Unix_error (Unix.(EAGAIN | EWOULDBLOCK), _, _) -> -1
   in
   (* If there's no space to read, shift *)
   if ic.max = buf_size then shift ic ;
   let space_left = buf_size - ic.max in
   (* Read byte one by one just do make sure we don't buffer too many chars *)
   let n =
-    fill_no_exc timeout (if buffered then space_left else min space_left 1)
+    fill_no_exc (Some timeout)
+      (if buffered then space_left else min space_left 1)
   in
   (* Select returned nothing to read *)
   if n = -1 then raise Timeout ;
@@ -97,7 +97,11 @@ let fill_buf ~buffered ic timeout =
     let tofillsz =
       if buffered then buf_size - ic.max else min (buf_size - ic.max) 1
     in
-    ignore (fill_no_exc 0.0 tofillsz)
+    (* cannot use 0 here, for select that'd mean timeout immediately, for
+       setsockopt it would mean no timeout.
+       So use a very short timeout instead
+    *)
+    ignore (fill_no_exc (Some 1e-6) tofillsz)
   )
 
 (** Input one line terminated by \n *)

ocaml/libs/http-lib/http.ml

@@ -320,14 +320,8 @@ let read_frame_header buf =
   let prefix = Bytes.sub_string buf 0 frame_header_length in
   try Scanf.sscanf prefix "FRAME %012d" (fun x -> Some x) with _ -> None
 
-let set_socket_timeout fd t =
-  try Unix.(setsockopt_float fd SO_RCVTIMEO t)
-  with Unix.Unix_error (Unix.ENOTSOCK, _, _) ->
-    (* In the unit tests, the fd comes from a pipe... ignore *)
-    ()
-
 let read_http_request_header ~read_timeout ~total_timeout ~max_length fd =
-  Option.iter (fun t -> set_socket_timeout fd t) read_timeout ;
+  Unixext.with_socket_timeout fd read_timeout @@ fun () ->
   let buf = Bytes.create (Option.value ~default:1024 max_length) in
   let deadline =
     Option.map
@@ -372,7 +366,6 @@ let read_http_request_header ~read_timeout ~total_timeout ~max_length fd =
         check_timeout_and_read 0 length ;
         (true, length)
   in
-  set_socket_timeout fd 0. ;
   (frame, Bytes.sub_string buf 0 headers_length, proxy)
 
 let read_http_response_header buf fd =

ocaml/libs/xapi-stdext/lib/xapi-stdext-unix/unixext.ml

@@ -1062,3 +1062,18 @@ module Daemon = struct
       true
     with Unix.Unix_error _ -> false
 end
+
+let set_socket_timeout fd t =
+  try Unix.(setsockopt_float fd SO_RCVTIMEO t)
+  with Unix.Unix_error (Unix.ENOTSOCK, _, _) ->
+    (* In the unit tests, the fd comes from a pipe... ignore *)
+    ()
+
+let with_socket_timeout fd timeout_opt f =
+  match timeout_opt with
+  | Some t ->
+      if t < 1e-6 then invalid_arg (Printf.sprintf "Timeout too short: %g" t) ;
+      let finally () = set_socket_timeout fd 0. in
+      set_socket_timeout fd t ; Fun.protect ~finally f
+  | None ->
+      f ()