class_docstr: Performs prefetching from an iterable in separate threads.
  
  Its interface is additionally extended with `ready` method which tells whether
  there is any data waiting for processing and a `retrieved_elements` method
  specifying number of elements retrieved from the iterator.
  
  Yields:
    Prefetched elements from the original iterable.
  
  Raises:
    ValueError: if the buffer_size < 1.
    StopIteration: If the iterable contains no more items.
    Any error thrown by the iterable_function. Note this is not raised inside
      the producer, but after it finishes executing.
class_docstr_tok: ["Performs","prefetching","from","an","iterable","in","separate","threads",".","Its","interface","is","additionally","extended","with","`","ready","`","method","which","tells","whether","there","is","any","data","waiting","for","processing","and","a","`","retrieved_elements","`","method","specifying","number","of","elements","retrieved","from","the","iterator",".","Yields",":","Prefetched","elements","from","the","original","iterable",".","Raises",":","ValueError",":","if","the","buffer_size","<","1",".","StopIteration",":","If","the","iterable","contains","no","more","items",".","Any","error","thrown","by","the","iterable_function",".","Note","this","is","not","raised","inside","the","producer",",","but","after","it","finishes","executing","."]
docstr: Enqueues items from `iterable` on a given thread.
code: def producer(self):
      """Enqueues items from `iterable` on a given thread."""
      try:
          # Build a new iterable for each thread. This is crucial if working with
          # tensorflow datasets because tf.Graph objects are thread local.
          for item in self.iterable:
              if self.device:
                  jax.device_put(item, self.device)
              self.buffer.put(item)
      except Exception as e:  # pylint: disable=broad-except
          logging.exception(
              "Error in producer thread for %s", self.iterable
          )
          self.producer_error.append(e)
      finally:
          self.buffer.put(self.end)
  
imports: import functools
  import itertools
  import queue
  import threading
  from typing import Callable, Iterable, Iterator, NamedTuple, Optional, Sequence, Tuple, TypeVar
  from absl import logging
  from acme import core
  from acme import types
  from acme.jax import types
  import haiku
  import jax
  import jax.numpy
  import numpy
  import reverb
  import tree
label_desc: Use image node_id 2 for calling the PrefetchIterator obj's underlying member method code with example usage: obj.producer() without return types