> Also in this case run awaits and returns the result. Yury suggested earlier just to return the future and not await.

Yeah that's roughly what my initial version was doing. I'm personally leaning a bit more towards returning the future rather than the result, but I'm okay with either option. What are your thoughts on this Yury and Andrew?

> I agree that shutdown_default_executor and _do_shutdown should be changed to accept an executor argument so that any executor can be shutdown asynchronously

We could potentially add an internal method _shutdown_executor, but this would also require modification of _do_shutdown (so that it shuts down the executor passed, rather than the default one). I mentioned this in an earlier example, but this one shows all three together and changes _shutdown_executor to a method of BaseEventLoop:
        
    async def shutdown_default_executor(self):
        """Schedule the shutdown of the default executor."""
        self._executor_shutdown_called = True
        executor = self._default_executor
        await self._shutdown_executor(executor)

    async def _shutdown_executor(self, executor):
        if executor is None:
            return
        future = self.create_future()
        thread = threading.Thread(target=self._do_shutdown, args=(executor,future))
        thread.start()
        try:
            await future
        finally:
            thread.join()

    def _do_shutdown(self, executor, future):
        try:
            executor.shutdown(wait=True)
            self.call_soon_threadsafe(future.set_result, None)
        except Exception as ex:
            self.call_soon_threadsafe(future.set_exception, ex)

Functionally, it works the same for shutdown_default_executor(), but allows _shutdown_executor to be used for asyncio.ThreadPool as well. Since GH-16360 (adding timeout param) also makes changes to shutdown_default_executor(), it will be blocking this issue.