class_docstr: Internal autograd-friendly function that calls multiple experts on a batch of inputs and awaits responses
  This function that can recover from individual failures during forward and/or backward pass as long as at least
  one expert succeeds for each input. For user-friendly version of this function, use RemoteMixtureOfExperts module.
  
  Note: experts that failed during forward will be assigned zero outputs and marked as mask[i, j] = 0,
        experts that failed during backward will be treated as constants (i.e. gradients through them are zeros)
class_docstr_tok: ["Internal","autograd-friendly","function","that","calls","multiple","experts","on","a","batch","of","inputs","and","awaits","responses","This","function","that","can","recover","from","individual","failures","during","forward","and\/or","backward","pass","as","long","as","at","least","one","expert","succeeds","for","each","input",".","For","user-friendly","version","of","this","function",",","use","RemoteMixtureOfExperts","module",".","Note",":","experts","that","failed","during","forward","will","be","assigned","zero","outputs","and","marked","as","mask","[","i",",","j","]","=","0",",","experts","that","failed","during","backward","will","be","treated","as","constants","(","i.e",".","gradients","through","them","are","zeros",")"]
docstr: await up to k_min results and any result submitted within timeout_after_k_min, cancel stragglers
docstr_tok: ["await","up","to","k_min","results","and","any","result","submitted","within","timeout_after_k_min",",","cancel","stragglers"]
code: def _collect_responses(
      task_to_indices: Dict[Future, Tuple[int, int]],
      num_samples: int,
      k_min: int,
      timeout_total: Optional[float],
      timeout_after_k_min: Optional[float],
      detect_anomalies: bool,
  ) -> Tuple[List[Tuple[int, int]], List[Tuple[torch.Tensor, ...]]]:
      """await up to k_min results and any result submitted within timeout_after_k_min, cancel stragglers"""
      timeout_total = (
          float("inf") if timeout_total is None else timeout_total
      )
      timeout_after_k_min = (
          float("inf")
          if timeout_after_k_min is None
          else timeout_after_k_min
      )
      num_successful_tasks = [0 for _ in range(num_samples)]
      pending_samples = num_samples  # samples for which we have less than k_min results
      finished_indices, finished_outputs = [], []
      t_finish = time.perf_counter() + timeout_total
      pending_tasks = set(task_to_indices.keys())
      finished_tasks = Queue()
  
      try:
          # the algorithm below is essentially futures.as_completed, but for grpc.Future
          for task in pending_tasks:
              task.add_done_callback(finished_tasks.put)
  
          for _ in range(len(task_to_indices)):
              timeout = (
                  max(0.0, t_finish - time.perf_counter())
                  if t_finish != float("inf")
                  else None
              )
              task = finished_tasks.get(timeout=timeout)
              pending_tasks.discard(task)
  
              task_output = _process_dispatched_task(
                  task, detect_anomalies
              )
              if task_output is not None:
                  finished_indices.append(task_to_indices[task])
                  finished_outputs.append(task_output)
  
                  # count how many successes we have for each input sample
                  sample_index = task_to_indices[task][0]
                  num_successful_tasks[sample_index] += 1
                  if num_successful_tasks[sample_index] == k_min:
                      pending_samples -= 1
                      if (
                          pending_samples <= 0
                      ):  # all tasks finished, await stragglers for at most timeout_after_k_min
                          t_finish = min(
                              t_finish,
                              time.perf_counter() + timeout_after_k_min,
                          )
  
      except Empty:
          pass  # we reached t_finish, this is normal behavior
      finally:
          for task in pending_tasks:
              task.cancel()
      return finished_indices, finished_outputs
  
imports: from __future__ import annotations
  import time
  from concurrent.futures import Future
  from queue import Empty, Queue
  from typing import Any, Dict, List, Optional, Tuple
  import torch
  import torch.nn
  from torch.autograd.function import once_differentiable
  from hivemind.compression import serialize_torch_tensor
  from hivemind.dht import DHT
  from hivemind.moe.client.beam_search import MoEBeamSearcher
  from hivemind.moe.client.expert import DUMMY, RemoteExpert, expert_backward, expert_forward, get_server_stub
  from hivemind.moe.client.remote_expert_worker import RemoteExpertWorker
  from hivemind.moe.expert_uid import UID_DELIMITER
  from hivemind.p2p.p2p_daemon_bindings.control import P2PDaemonError
  from hivemind.utils import nested_flatten, nested_map, nested_pack
  from hivemind.utils.logging import get_logger
label_desc: Use image node_id 3 for calling the _RemoteCallMany obj's underlying member method code with example usage: obj._collect_responses(task_to_indices, num_samples, k_min, timeout_total, timeout_after_k_min, detect_anomalies) and returns: finished_indices,  finished_outputs