class_docstr: Step a single env in a separate process for lock free paralellism.
class_docstr_tok: ["Step","a","single","env","in","a","separate","process","for","lock","free","paralellism","."]
docstr: The process waits for actions and sends back environment results.
  
  Args:
    conn: Connection for communication to the main process.
  
  Raises:
    KeyError: When receiving a message of unknown type.
docstr_tok: ["The","process","waits","for","actions","and","sends","back","environment","results",".","Args",":","conn",":","Connection","for","communication","to","the","main","process",".","Raises",":","KeyError",":","When","receiving","a","message","of","unknown","type","."]
code: def _worker(self, conn):
      """The process waits for actions and sends back environment results.
  
      Args:
        conn: Connection for communication to the main process.
  
      Raises:
        KeyError: When receiving a message of unknown type.
      """
      try:
          env = cloudpickle.loads(self._pickled_env_constructor)()
          action_spec = env.action_spec()
          conn.send(self._READY)  # Ready.
          while True:
              try:
                  # Only block for short times to have keyboard exceptions be raised.
                  if not conn.poll(_POLLING_PERIOD):
                      continue
                  message, payload = conn.recv()
              except (EOFError, KeyboardInterrupt):
                  break
              if message == self._ACCESS:
                  name = payload
                  result = getattr(env, name)
                  conn.send((self._RESULT, result))
                  continue
              if message == self._CALL:
                  name, args, kwargs = payload
                  if self._flatten and name == "step":
                      args = [
                          tf.nest.pack_sequence_as(action_spec, args[0])
                      ]
                  result = getattr(env, name)(*args, **kwargs)
                  if self._flatten and name in ["step", "reset"]:
                      result = tf.nest.flatten(result)
                  conn.send((self._RESULT, result))
                  continue
              if message == self._CLOSE:
                  assert payload is None
                  env.close()
                  break
              raise KeyError(
                  "Received message of unknown type {}".format(message)
              )
      except Exception:  # pylint: disable=broad-except
          etype, evalue, tb = sys.exc_info()
          stacktrace = "".join(
              traceback.format_exception(etype, evalue, tb)
          )
          message = "Error in environment process: {}".format(
              stacktrace
          )
          logging.error(message)
          conn.send((self._EXCEPTION, stacktrace))
      finally:
          conn.close()
  
imports: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  import atexit
  import sys
  import traceback
  from typing import Any, Callable, Sequence, Text, Union
  from absl import logging
  import cloudpickle
  import gin
  import numpy
  import tensorflow
  from tf_agents.environments import py_environment
  from tf_agents.system import system_multiprocessing
  from tf_agents.trajectories import time_step
  from tf_agents.typing import types
  from tf_agents.utils import nest_utils
label_desc: Use image node_id 15 for calling the ProcessPyEnvironment obj's underlying member method code with example usage: obj._worker(conn) without return types