code: def _test_queue_recv(
      start_method,
      worker_params,
      capacity,
      send_msgs,
      recv_msgs,
      send_one_by_one,
  ):
      count = 0
  
      def next_i():
          nonlocal count
          count += 1
          return count
  
      with setup_queue_and_worker(
          start_method, capacity, copy_callback, worker_params
      ) as (
          task_queue,
          res_queue,
      ):
          all_msgs = []
          received = 0
          for send_msg, recv_msg in zip(send_msgs, recv_msgs):
              msgs = [
                  ShmMessageDesc(
                      next_i(), -next_i(), next_i(), next_i(), next_i()
                  )
                  for i in range(send_msg)
              ]
              all_msgs.extend(msgs)
              _put_msgs(task_queue, msgs, send_one_by_one)
              for _ in range(recv_msg):
                  [recv_msg] = res_queue.get()
                  msg_values = all_msgs[received].get_values()
                  received += 1
                  recv_msg_values = recv_msg.get_values()
                  assert len(msg_values) == len(recv_msg_values)
                  assert all(
                      msg == recv_msg
                      for msg, recv_msg in zip(
                          msg_values, recv_msg_values
                      )
                  )
  
imports: import os
  import multiprocessing
  import socket
  from contextlib import closing, contextmanager
  import numpy
  from nvidia.dali._multiproc.shared_batch import BufShmChunk, SharedBatchWriter, SharedBatchMeta, deserialize_batch
  from nvidia.dali._multiproc.shared_queue import ShmQueue
  from nvidia.dali._multiproc.messages import ShmMessageDesc
  from test_utils import RandomlyShapedDataIterator
  from nose_utils import raises
label_desc: Use image node_id 9 for calling a global function with example usage: _test_queue_recv(start_method, worker_params, capacity, send_msgs, recv_msgs, send_one_by_one) and returns: count