def run(args: 'argparse.Namespace', name: str, runtime_cls, envs: Dict[(str, str)], is_started: Union[('multiprocessing.Event', 'threading.Event')], is_shutdown: Union[('multiprocessing.Event', 'threading.Event')], is_ready: Union[('multiprocessing.Event', 'threading.Event')], cancel_event: Union[('multiprocessing.Event', 'threading.Event')]):
    "Method representing the :class:`BaseRuntime` activity.\n\n    This method is the target for the Pea's `thread` or `process`\n\n    .. note::\n        :meth:`run` is running in subprocess/thread, the exception can not be propagated to the main process.\n        Hence, please do not raise any exception here.\n\n    .. note::\n        Please note that env variables are process-specific. Subprocess inherits envs from\n        the main process. But Subprocess's envs do NOT affect the main process. It does NOT\n        mess up user local system envs.\n\n    .. warning::\n        If you are using ``thread`` as backend, envs setting will likely be overidden by others\n\n    :param args: namespace args from the Pea\n    :param name: name of the Pea to have proper logging\n    :param runtime_cls: the runtime class to instantiate\n    :param envs: a dictionary of environment variables to be set in the new Process\n    :param is_started: concurrency event to communicate runtime is properly started. Used for better logging\n    :param is_shutdown: concurrency event to communicate runtime is terminated\n    :param is_ready: concurrency event to communicate runtime is ready to receive messages\n    :param cancel_event: concurrency event to receive cancelling signal from the Pea. Needed by some runtimes\n    "
    logger = JinaLogger(name, **vars(args))

    def _unset_envs():
        if (envs and (args.runtime_backend != RuntimeBackendType.THREAD)):
            for k in envs.keys():
                os.unsetenv(k)

    def _set_envs():
        if args.env:
            if (args.runtime_backend == RuntimeBackendType.THREAD):
                logger.warning('environment variables should not be set when runtime="thread".')
            else:
                os.environ.update({k: str(v) for (k, v) in envs.items()})
    try:
        _set_envs()
        runtime = runtime_cls(args=args, cancel_event=cancel_event)
    except Exception as ex:
        logger.error(((f'{ex!r} during {runtime_cls!r} initialization' + f'''
 add "--quiet-error" to suppress the exception details''') if (not args.quiet_error) else ''), exc_info=(not args.quiet_error))
    else:
        is_started.set()
        with runtime:
            is_ready.set()
            runtime.run_forever()
    finally:
        _unset_envs()
        is_shutdown.set()
Method representing the :class:`BaseRuntime` activity.

This method is the target for the Pea's `thread` or `process`

.. note::
    :meth:`run` is running in subprocess/thread, the exception can not be propagated to the main process.
    Hence, please do not raise any exception here.

.. note::
    Please note that env variables are process-specific. Subprocess inherits envs from
    the main process. But Subprocess's envs do NOT affect the main process. It does NOT
    mess up user local system envs.

.. warning::
    If you are using ``thread`` as backend, envs setting will likely be overidden by others

:param args: namespace args from the Pea
:param name: name of the Pea to have proper logging
:param runtime_cls: the runtime class to instantiate
:param envs: a dictionary of environment variables to be set in the new Process
:param is_started: concurrency event to communicate runtime is properly started. Used for better logging
:param is_shutdown: concurrency event to communicate runtime is terminated
:param is_ready: concurrency event to communicate runtime is ready to receive messages
:param cancel_event: concurrency event to receive cancelling signal from the Pea. Needed by some runtimes
def run(args: 'argparse.Namespace', name: str, runtime_cls, envs: Dict[(str, str)], is_started: Union[('multiprocessing.Event', 'threading.Event')], is_shutdown: Union[('multiprocessing.Event', 'threading.Event')], is_ready: Union[('multiprocessing.Event', 'threading.Event')], cancel_event: Union[('multiprocessing.Event', 'threading.Event')]):
    "Method representing the :class:`BaseRuntime` activity.\n\n    This method is the target for the Pea's `thread` or `process`\n\n    .. note::\n        :meth:`run` is running in subprocess/thread, the exception can not be propagated to the main process.\n        Hence, please do not raise any exception here.\n\n    .. note::\n        Please note that env variables are process-specific. Subprocess inherits envs from\n        the main process. But Subprocess's envs do NOT affect the main process. It does NOT\n        mess up user local system envs.\n\n    .. warning::\n        If you are using ``thread`` as backend, envs setting will likely be overidden by others\n\n    :param args: namespace args from the Pea\n    :param name: name of the Pea to have proper logging\n    :param runtime_cls: the runtime class to instantiate\n    :param envs: a dictionary of environment variables to be set in the new Process\n    :param is_started: concurrency event to communicate runtime is properly started. Used for better logging\n    :param is_shutdown: concurrency event to communicate runtime is terminated\n    :param is_ready: concurrency event to communicate runtime is ready to receive messages\n    :param cancel_event: concurrency event to receive cancelling signal from the Pea. Needed by some runtimes\n    "
    logger = JinaLogger(name, **vars(args))

    def _unset_envs():
        if (envs and (args.runtime_backend != RuntimeBackendType.THREAD)):
            for k in envs.keys():
                os.unsetenv(k)

    def _set_envs():
        if args.env:
            if (args.runtime_backend == RuntimeBackendType.THREAD):
                logger.warning('environment variables should not be set when runtime="thread".')
            else:
                os.environ.update({k: str(v) for (k, v) in envs.items()})
    try:
        _set_envs()
        runtime = runtime_cls(args=args, cancel_event=cancel_event)
    except Exception as ex:
        logger.error(((f'{ex!r} during {runtime_cls!r} initialization' + f'
 add "--quiet-error" to suppress the exception details') if (not args.quiet_error) else ), exc_info=(not args.quiet_error))
    else:
        is_started.set()
        with runtime:
            is_ready.set()
            runtime.run_forever()
    finally:
        _unset_envs()
        is_shutdown.set()