fixup! CP-52225 - poll: Reduce unnecessary allocations

Author: Andrii Sultanov

oxenstored/connections.ml

@@ -22,7 +22,7 @@ let debug fmt = Logging.debug "connections" fmt
 type t = {
     anonymous: (Unix.file_descr, Connection.t * int) Hashtbl.t
         (* (fd -> Connection.t, index) where index maps to the poll_status array *)
-  ; mutable poll_status: (Unix.file_descr * Poll.event) array
+  ; mutable poll_status: Poll.poll_status_t array
   ; domains: (int, Connection.t) Hashtbl.t
   ; ports: (Xeneventchn.t, Connection.t) Hashtbl.t
   ; mutable watches: Connection.watch list Trie.t
@@ -46,7 +46,15 @@ let get_capacity () =
   ; maxwatchevents= !Define.maxwatchevents
   }
 
-let default_poll_status () = (Unix.stdin, Poll.init_event ())
+let default_poll_status () = Poll.{fd= Unix.stdin; event= init_event ()}
+
+let spec_poll_status_new () = Poll.{read= true; write= false; except= false}
+
+let spec_poll_status event =
+  let open Poll in
+  event.read <- true ;
+  event.write <- false ;
+  event.except <- false
 
 let add_anonymous cons fd =
   let capacity = get_capacity () in
@@ -67,20 +75,23 @@ let add_domain cons dom =
   Hashtbl.replace cons.domains (Domain.get_id dom) con ;
   Hashtbl.replace cons.ports (Domain.get_local_port dom) con
 
-let refresh_poll_status ?(only_if = fun _ -> true) cons =
+let refresh_poll_status ?(only_if = fun _ -> true) cons spec_fds =
+  (* special fds are always read=true, but get overwritten by select_stubs, so we
+     need to reset the event we are polling for *)
+  List.iteri
+    (fun index fd ->
+      cons.poll_status.(index).fd <- fd ;
+      spec_poll_status cons.poll_status.(index).event
+    )
+    spec_fds ;
   Hashtbl.iter
     (fun _ (con, index) ->
       let only = only_if con in
-      let fd = Connection.get_fd con in
+      cons.poll_status.(index).fd <- Connection.get_fd con ;
       let open Poll in
-      let event =
-        {
-          read= only && Connection.can_input con
-        ; write= only && Connection.has_output con
-        ; except= false
-        }
-      in
-      cons.poll_status.(index) <- (fd, event)
+      cons.poll_status.(index).event.read <- only && Connection.can_input con ;
+      cons.poll_status.(index).event.write <- only && Connection.has_output con ;
+      cons.poll_status.(index).event.except <- false
     )
     cons.anonymous
 
@@ -107,19 +118,33 @@ let del_watches cons con =
     |> Connection.Watch.Set.filter @@ fun w -> Connection.get_con w != con
     )
 
-let del_anonymous cons con =
+let del_anonymous cons con spec_fds =
   try
     Hashtbl.remove cons.anonymous (Connection.get_fd con) ;
     (* Reallocate the poll_status array, update indices pointing to it *)
     cons.poll_status <-
-      Array.make (Hashtbl.length cons.anonymous) (default_poll_status ()) ;
+      Array.make
+        (Hashtbl.length cons.anonymous + List.length spec_fds)
+        (default_poll_status ()) ;
+
+    (* Keep the special fds at the beginning *)
+    let i =
+      List.fold_left
+        (fun index fd ->
+          cons.poll_status.(index).fd <- fd ;
+          spec_poll_status cons.poll_status.(index).event ;
+          index + 1
+        )
+        0 spec_fds
+    in
+
     let _ =
       Hashtbl.fold
         (fun key (con, _) i ->
           Hashtbl.replace cons.anonymous key (con, i) ;
           i + 1
         )
-        cons.anonymous 0
+        cons.anonymous i
     in
 
     del_watches cons con ; Connection.close con

oxenstored/poll.ml

@@ -17,9 +17,11 @@
    readfds, writefds, exceptfds concept as in select. *)
 type event = {mutable read: bool; mutable write: bool; mutable except: bool}
 
+type poll_status_t = {mutable fd: Unix.file_descr; event: event}
+
 let init_event () = {read= false; write= false; except= false}
 
-external select_on_poll : (Unix.file_descr * event) array -> int -> int
+external select_on_poll : poll_status_t array -> int -> int
   = "stub_select_on_poll"
 
 external set_fd_limit : int -> unit = "stub_set_fd_limit"
@@ -33,14 +35,14 @@ let get_sys_fs_nr_open () =
     close_in_noerr ch ; v
   with _ -> 1024 * 1024
 
-let poll_select fdarr timeout =
+let poll_select (fdarr : poll_status_t array) timeout =
   let n = select_on_poll fdarr (int_of_float (timeout *. 1000.)) in
   let r = ([], [], []) in
   if n = 0 then
     r
   else
     Array.fold_right
-      (fun (fd, event) (r, w, x) ->
+      (fun {fd; event} (r, w, x) ->
         ( (if event.read then fd :: r else r)
         , (if event.write then fd :: w else w)
         , if event.except then fd :: x else x

oxenstored/poll.mli

@@ -14,9 +14,11 @@
 
 type event = {mutable read: bool; mutable write: bool; mutable except: bool}
 
+type poll_status_t = {mutable fd: Unix.file_descr; event: event}
+
 val init_event : unit -> event
 
 val poll_select :
-     (Unix.file_descr * event) array
+     poll_status_t array
   -> float
   -> Unix.file_descr list * Unix.file_descr list * Unix.file_descr list

oxenstored/xenstored.ml

@@ -29,20 +29,20 @@ let debug fmt = Logging.debug "xenstored" fmt
 let info fmt = Logging.info "xenstored" fmt
 
 (*------------ event klass processors --------------*)
-let process_connection_fds store cons domains rset wset =
+let process_connection_fds store cons domains rset wset spec_fds =
   let try_fct fct c =
     try fct store cons domains c with
     | Unix.Unix_error (err, "write", _) ->
-        Connections.del_anonymous cons c ;
+        Connections.del_anonymous cons c spec_fds ;
         error "closing socket connection: write error: %s"
           (Unix.error_message err)
     | Unix.Unix_error (err, "read", _) ->
-        Connections.del_anonymous cons c ;
+        Connections.del_anonymous cons c spec_fds ;
         if err <> Unix.ECONNRESET then
           error "closing socket connection: read error: %s"
             (Unix.error_message err)
     | Xenbus.Xb.End_of_file ->
-        Connections.del_anonymous cons c ;
+        Connections.del_anonymous cons c spec_fds ;
         debug "closing socket connection"
   in
   let process_fdset_with fds fct =
@@ -528,6 +528,16 @@ let () =
     (match rw_sock with None -> [] | Some x -> [x])
     @ if cf.domain_init then [Event.fd eventchn] else []
   in
+  (* Always positioned at the start of cons.poll_status, maintained during
+     reallocations *)
+  cons.poll_status <-
+    Array.append cons.poll_status
+      (Array.of_list
+         (List.map
+            Connections.(fun fd -> Poll.{fd; event= spec_poll_status_new ()})
+            spec_fds
+         )
+      ) ;
 
   let process_special_fds rset =
     let accept_connection fd =
@@ -681,7 +691,7 @@ let () =
       in
       if peaceful_mw <> [] then 0. else until_next_activity
     in
-    Connections.refresh_poll_status ~only_if:is_peaceful cons ;
+    Connections.refresh_poll_status ~only_if:is_peaceful cons spec_fds ;
     let rset, wset, _ =
       try Poll.poll_select cons.poll_status timeout
       with Unix.Unix_error (Unix.EINTR, _, _) -> ([], [], [])
@@ -691,7 +701,7 @@ let () =
       process_special_fds sfds ;
 
     if cfds <> [] || wset <> [] then
-      process_connection_fds store cons domains cfds wset ;
+      process_connection_fds store cons domains cfds wset spec_fds ;
     ( if timeout <> 0. then
         let now = Unix.gettimeofday () in
         if now > !period_start +. period_ops_interval then (