CA-410782: Add receive_memory_queues for VM_receive_memory operations

    Migration spawns 2 operations which depend on each other so we need to
    ensure there is always space for both of them to prevent a deadlock.
    Adding VM_receive_memory to a new queue ensures that there will always
    be a worker for the receive operation so the paired send will never be
    blocked.

    Signed-off-by: Steven Woods <[email protected]>

Author: snwoods

ocaml/xenopsd/lib/xenops_server.ml

@@ -916,6 +916,12 @@ module Redirector = struct
      Parallel atoms, creating a deadlock. *)
   let parallel_queues = {queues= Queues.create (); mutex= Mutex.create ()}
 
+  (* We create another queue only for VM_receive_memory operations for the same reason as Parallel atoms.
+     Migration spawns 2 operations, send and receive, so if there is limited available worker space
+     a deadlock can happen when VMs are migrating between hosts or on localhost migration
+     as the receiver has no free workers to receive memory. *)
+  let receive_memory_queues = {queues= Queues.create (); mutex= Mutex.create ()}
+
   (* When a thread is actively processing a queue, items are redirected to a
      thread-private queue *)
   let overrides = ref StringMap.empty
@@ -1035,6 +1041,7 @@ module Redirector = struct
           List.concat_map one
             (default.queues
             :: parallel_queues.queues
+            :: receive_memory_queues.queues
             :: List.map snd (StringMap.bindings !overrides)
             )
       )
@@ -1268,7 +1275,8 @@ module WorkerPool = struct
   let start size =
     for _i = 1 to size do
       incr Redirector.default ;
-      incr Redirector.parallel_queues
+      incr Redirector.parallel_queues ;
+      incr Redirector.receive_memory_queues
     done
 
   let set_size size =
@@ -1283,7 +1291,8 @@ module WorkerPool = struct
       done
     in
     inner Redirector.default ;
-    inner Redirector.parallel_queues
+    inner Redirector.parallel_queues ;
+    inner Redirector.receive_memory_queues
 end
 
 (* Keep track of which VMs we're rebooting so we avoid transient glitches where
@@ -3276,19 +3285,20 @@ let uses_mxgpu id =
     )
     (VGPU_DB.ids id)
 
-let queue_operation_int ?traceparent dbg id op =
+let queue_operation_int ?traceparent ?(redirector = Redirector.default) dbg id
+    op =
   let task =
     Xenops_task.add ?traceparent tasks dbg
       (let r = ref None in
        fun t -> perform ~result:r op t ; !r
       )
   in
   let tag = if uses_mxgpu id then "mxgpu" else id in
-  Redirector.push Redirector.default tag (op, task) ;
+  Redirector.push redirector tag (op, task) ;
   task
 
-let queue_operation ?traceparent dbg id op =
-  let task = queue_operation_int ?traceparent dbg id op in
+let queue_operation ?traceparent ?redirector dbg id op =
+  let task = queue_operation_int ?traceparent ?redirector dbg id op in
   Xenops_task.id_of_handle task
 
 let queue_operation_and_wait dbg id op =
@@ -3737,7 +3747,12 @@ module VM = struct
                 ; vmr_compressed= compressed_memory
                 }
             in
-            let task = Some (queue_operation ?traceparent dbg id op) in
+            let task =
+              Some
+                (queue_operation ?traceparent
+                   ~redirector:Redirector.receive_memory_queues dbg id op
+                )
+            in
             Option.iter
               (fun t -> t |> Xenops_client.wait_for_task dbg |> ignore)
               task