class_docstr: Managing class for the acute evaluator process.
  
  Relevant args are parsed in the `setup_args` function above.
class_docstr_tok: ["Managing","class","for","the","acute","evaluator","process",".","Relevant","args","are","parsed","in","the","`","setup_args","`","function","above","."]
docstr: Poll task queue for tasks for a worker.
  
  :param worker_id:
      id for worker
  
  :param task_data:
      list of potential tasks already for worker
  
  :return task_data:
      a list of tasks for a worker to complete
docstr_tok: ["Poll","task","queue","for","tasks","for","a","worker",".",":","param","worker_id",":","id","for","worker",":","param","task_data",":","list","of","potential","tasks","already","for","worker",":","return","task_data",":","a","list","of","tasks","for","a","worker","to","complete"]
code: def _poll_task_queue(
      self, worker_id: str, task_data: List[Dict[str, Any]]
  ) -> List[PairingsDict]:
      """
      Poll task queue for tasks for a worker.
  
      :param worker_id:
          id for worker
  
      :param task_data:
          list of potential tasks already for worker
  
      :return task_data:
          a list of tasks for a worker to complete
      """
      worker_data = self._get_worker_data(worker_id)
      num_attempts = 0
      while (
          not self.task_queue.empty()
      ) and num_attempts < self.task_queue.qsize():
          try:
              next_task = self.task_queue.get()
          except queue.Empty:
              break
          num_attempts += 1
  
          pair_id = next_task["pair_id"]
          dialogue_ids = self._get_dialogue_ids(next_task)
  
          # make sure worker has not seen these conversations before
          if pair_id not in worker_data["tasks_completed"] and all(
              d_id not in worker_data["conversations_seen"]
              for d_id in dialogue_ids
          ):
              # track tasks and conversations seen
              worker_data["tasks_completed"].append(pair_id)
              worker_data["conversations_seen"].extend(dialogue_ids)
              task_data.append(next_task)
              if (
                  len(task_data)
                  == self.args.blueprint.subtasks_per_unit
              ):
                  return task_data
          else:
              self.task_queue.put(next_task)
  
      return task_data
  
imports: import json
  import queue
  import random
  from typing import List, Any, Dict, Tuple, Set, TYPE_CHECKING
  from mephisto.utils.logger_core import get_logger
  from mephisto.abstractions.blueprint import SharedTaskState, TaskRunner
  from omegaconf import DictConfig
label_desc: Use image node_id 7 for calling the AcuteEvalRunner obj's underlying member method code with example usage: obj._poll_task_queue(worker_id, task_data) and returns: task_data, task_data