async def CreateRuntimeEnv(self, request, context):

    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))
    serialized_env = request.serialized_runtime_env
    if (serialized_env not in self._env_locks):
        self._env_locks[serialized_env] = asyncio.Lock()
    async with self._env_locks[serialized_env]:
        if (serialized_env in self._env_cache):
            serialized_context = self._env_cache[serialized_env]
            result = self._env_cache[serialized_env]
            if result.success:
                context = result.result
                logger.info(f'Runtime env already created successfully. Env: {serialized_env}, context: {context}')
                return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_OK, serialized_runtime_env_context=context)
            else:
                error_message = result.result
                logger.info(f'Runtime env already failed. Env: {serialized_env}, err: {error_message}')
                return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_FAILED, error_message=error_message)
        if SLEEP_FOR_TESTING_S:
            logger.info(f'Sleeping for {SLEEP_FOR_TESTING_S}s.')
            time.sleep(int(SLEEP_FOR_TESTING_S))
        logger.info(f'Creating runtime env: {serialized_env}')
        runtime_env_context: RuntimeEnvContext = None
        error_message = None
        for _ in range(runtime_env_consts.RUNTIME_ENV_RETRY_TIMES):
            try:
                runtime_env_context = (await _setup_runtime_env(serialized_env, request.serialized_allocated_resource_instances))
                break
            except Exception as ex:
                logger.exception('Runtime env creation failed.')
                error_message = str(ex)
                (await asyncio.sleep((runtime_env_consts.RUNTIME_ENV_RETRY_INTERVAL_MS / 1000)))
        if error_message:
            logger.error("Runtime env creation failed for %d times, don't retry any more.", runtime_env_consts.RUNTIME_ENV_RETRY_TIMES)
            self._env_cache[serialized_env] = CreatedEnvResult(False, error_message)
            return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_FAILED, error_message=error_message)
        serialized_context = runtime_env_context.serialize()
        self._env_cache[serialized_env] = CreatedEnvResult(True, serialized_context)
        logger.info('Successfully created runtime env: %s, the context: %s', serialized_env, serialized_context)
        return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_OK, serialized_runtime_env_context=serialized_context)
async def CreateRuntimeEnv(self, request, context):

    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))
    serialized_env = request.serialized_runtime_env
    if (serialized_env not in self._env_locks):
        self._env_locks[serialized_env] = asyncio.Lock()
    async with self._env_locks[serialized_env]:
        if (serialized_env in self._env_cache):
            serialized_context = self._env_cache[serialized_env]
            result = self._env_cache[serialized_env]
            if result.success:
                context = result.result
                logger.info(f'Runtime env already created successfully. Env: {serialized_env}, context: {context}')
                return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_OK, serialized_runtime_env_context=context)
            else:
                error_message = result.result
                logger.info(f'Runtime env already failed. Env: {serialized_env}, err: {error_message}')
                return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_FAILED, error_message=error_message)
        if SLEEP_FOR_TESTING_S:
            logger.info(f'Sleeping for {SLEEP_FOR_TESTING_S}s.')
            time.sleep(int(SLEEP_FOR_TESTING_S))
        logger.info(f'Creating runtime env: {serialized_env}')
        runtime_env_context: RuntimeEnvContext = None
        error_message = None
        for _ in range(runtime_env_consts.RUNTIME_ENV_RETRY_TIMES):
            try:
                runtime_env_context = (await _setup_runtime_env(serialized_env, request.serialized_allocated_resource_instances))
                break
            except Exception as ex:
                logger.exception('Runtime env creation failed.')
                error_message = str(ex)
                (await asyncio.sleep((runtime_env_consts.RUNTIME_ENV_RETRY_INTERVAL_MS / 1000)))
        if error_message:
            logger.error("Runtime env creation failed for %d times, don't retry any more.", runtime_env_consts.RUNTIME_ENV_RETRY_TIMES)
            self._env_cache[serialized_env] = CreatedEnvResult(False, error_message)
            return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_FAILED, error_message=error_message)
        serialized_context = runtime_env_context.serialize()
        self._env_cache[serialized_env] = CreatedEnvResult(True, serialized_context)
        logger.info('Successfully created runtime env: %s, the context: %s', serialized_env, serialized_context)
        return runtime_env_agent_pb2.CreateRuntimeEnvReply(status=agent_manager_pb2.AGENT_RPC_STATUS_OK, serialized_runtime_env_context=serialized_context)