async def _setup_runtime_env(serialized_runtime_env, serialized_allocated_resource_instances):

    def run_setup_with_logger():
        runtime_env = RuntimeEnv(serialized_runtime_env=serialized_runtime_env)
        allocated_resource: dict = json.loads((serialized_allocated_resource_instances or '{}'))
        per_job_logger = self.get_or_create_logger(request.job_id)
        per_job_logger.debug(f'Worker has resource :{allocated_resource}')
        context = RuntimeEnvContext(env_vars=runtime_env.env_vars())
        self._conda_manager.setup(runtime_env, context, logger=per_job_logger)
        self._py_modules_manager.setup(runtime_env, context, logger=per_job_logger)
        self._working_dir_manager.setup(runtime_env, context, logger=per_job_logger)
        self._container_manager.setup(runtime_env, context, logger=per_job_logger)
        if runtime_env.working_dir_uri():
            uri = runtime_env.working_dir_uri()
            self._uris_to_envs[uri].add(serialized_runtime_env)
        if runtime_env.py_modules_uris():
            for uri in runtime_env.py_modules_uris():
                self._uris_to_envs[uri].add(serialized_runtime_env)
        if runtime_env.conda_uri():
            uri = runtime_env.conda_uri()
            self._uris_to_envs[uri].add(serialized_runtime_env)
        if runtime_env.plugin_uris():
            for uri in runtime_env.plugin_uris():
                self._uris_to_envs[uri].add(serialized_runtime_env)
        for (plugin_class_path, config) in runtime_env.plugins():
            logger.debug(f'Setting up runtime env plugin {plugin_class_path}')
            plugin_class = import_attr(plugin_class_path)
            plugin_class.create('uri not implemented', json.loads(config), context)
            plugin_class.modify_context('uri not implemented', json.loads(config), context)
        return context
    loop = asyncio.get_event_loop()
    return (await loop.run_in_executor(None, run_setup_with_logger))
async def _setup_runtime_env(serialized_runtime_env, serialized_allocated_resource_instances):

    def run_setup_with_logger():
        runtime_env = RuntimeEnv(serialized_runtime_env=serialized_runtime_env)
        allocated_resource: dict = json.loads((serialized_allocated_resource_instances or '{}'))
        per_job_logger = self.get_or_create_logger(request.job_id)
        per_job_logger.debug(f'Worker has resource :{allocated_resource}')
        context = RuntimeEnvContext(env_vars=runtime_env.env_vars())
        self._conda_manager.setup(runtime_env, context, logger=per_job_logger)
        self._py_modules_manager.setup(runtime_env, context, logger=per_job_logger)
        self._working_dir_manager.setup(runtime_env, context, logger=per_job_logger)
        self._container_manager.setup(runtime_env, context, logger=per_job_logger)
        if runtime_env.working_dir_uri():
            uri = runtime_env.working_dir_uri()
            self._uris_to_envs[uri].add(serialized_runtime_env)
        if runtime_env.py_modules_uris():
            for uri in runtime_env.py_modules_uris():
                self._uris_to_envs[uri].add(serialized_runtime_env)
        if runtime_env.conda_uri():
            uri = runtime_env.conda_uri()
            self._uris_to_envs[uri].add(serialized_runtime_env)
        if runtime_env.plugin_uris():
            for uri in runtime_env.plugin_uris():
                self._uris_to_envs[uri].add(serialized_runtime_env)
        for (plugin_class_path, config) in runtime_env.plugins():
            logger.debug(f'Setting up runtime env plugin {plugin_class_path}')
            plugin_class = import_attr(plugin_class_path)
            plugin_class.create('uri not implemented', json.loads(config), context)
            plugin_class.modify_context('uri not implemented', json.loads(config), context)
        return context
    loop = asyncio.get_event_loop()
    return (await loop.run_in_executor(None, run_setup_with_logger))