class_docstr: Manages state and performs updates for IMPALA learner.
docstr: Elementary data pipeline.
docstr_tok: ["Elementary","data","pipeline","."]
code: def host_to_device_worker(self):
      """Elementary data pipeline."""
      batch = []
      while not self._done:
          # Try to get a batch. Skip the iteration if we couldn't.
          try:
              for _ in range(len(batch), self._batch_size):
                  # As long as possible while keeping learner_test time reasonable.
                  batch.append(self._host_q.get(timeout=10))
          except queue.Empty:
              continue
  
          assert len(batch) == self._batch_size
          # Prepare for consumption, then put batch onto device.
          stacked_batch = jax.tree_util.tree_map(
              lambda *xs: np.stack(xs, axis=1), *batch
          )
          self._device_q.put(jax.device_put(stacked_batch))
  
          # Clean out the built-up batch.
          batch = []
  
imports: import functools
  import itertools
  import queue
  import threading
  import warnings
  import dm_env
  import haiku
  from examples.impala import agent
  from examples.impala import util
  import jax
  from jax.example_libraries import optimizers
  import jax.numpy
  import numpy
  import optax
  import rlax
label_desc: Use image node_id 6 for calling the Learner obj's underlying member method code with example usage: obj.host_to_device_worker() without return types