docstr: Distributes sampling requests among processes running sampling tasks.
  
  Parameters
  ----------
  anchors, batch_size
      See :py:meth:`alibi.explainers.anchors.anchor_base.AnchorBaseBeam.draw_samples` implementation.
  
  Returns
  -------
  See :py:meth:`alibi.explainers.anchors.anchor_base.AnchorBaseBeam.draw_samples` implementation.
docstr_tok: ["Distributes","sampling","requests","among","processes","running","sampling","tasks",".","Parameters","--","--","--","--","--","anchors",",","batch_size","See",":","py",":","meth",":","`","alibi.explainers.anchors.anchor_base.AnchorBaseBeam.draw_samples","`","implementation",".","Returns","--","--","--","-","See",":","py",":","meth",":","`","alibi.explainers.anchors.anchor_base.AnchorBaseBeam.draw_samples","`","implementation","."]
code: def draw_samples(self, anchors: list, batch_size: int) -> Tuple[np.ndarray, np.ndarray]:  # type: ignore[override]
      """
      Distributes sampling requests among processes running sampling tasks.
  
      Parameters
      ----------
      anchors, batch_size
          See :py:meth:`alibi.explainers.anchors.anchor_base.AnchorBaseBeam.draw_samples` implementation.
  
      Returns
      -------
      See :py:meth:`alibi.explainers.anchors.anchor_base.AnchorBaseBeam.draw_samples` implementation.
      """
  
      # partial anchors not generated by propose_anchors are not in the order dictionary
      for anchor in anchors:
          if anchor not in self.state["t_order"]:
              self.state["t_order"][anchor] = list(anchor)
  
      pos, total = np.zeros((len(anchors),)), np.zeros((len(anchors),))
      order_map = [
          (i, tuple(self.state["t_order"][anchor]))
          for i, anchor in enumerate(anchors)
      ]
      samples_iter = self.pool.map_unordered(
          partial(self.sample_fcn, n_samples=batch_size),
          order_map,
          self.chunksize,
      )
      for samples_batch in samples_iter:
          for samples in samples_batch:
              (
                  covered_true,
                  covered_false,
                  labels,
                  *additionals,
                  anchor_idx,
              ) = samples
              positives, n_samples = self.update_state(
                  covered_true,
                  covered_false,
                  labels,
                  additionals,
                  anchors[anchor_idx],
              )
              # return statistics in the same order as the requests
              pos[anchor_idx], total[anchor_idx] = positives, n_samples
  
      return pos, total
  
imports: import logging
  from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
  import numpy
  import ray
  from alibi.api.interfaces import Explanation
  from alibi.utils.discretizer import Discretizer
  from alibi.utils.mapping import ohe_to_ord
  from .anchor_base import AnchorBaseBeam
  from .anchor_tabular import AnchorTabular, TabularSampler
  from alibi.utils.distributed import ActorPool
  from functools import partial
label_desc: Use image node_id 3 for calling the DistributedAnchorBaseBeam obj's underlying member method code with example usage: obj.draw_samples(anchors, batch_size) and returns: pos,  total