So, I just had an interesting idea... what if ThreadPool.run() returned a Task instead of a coroutine object?

With the current version of asyncio.ThreadPool, if a user wants to create a Task, they would have to do something like this:

async with asyncio.ThreadPool() as pool:
    task = asyncio.create_task(pool.run(io_blocking_func, 10, kwarg1='test'))
    other_task = asyncio.create_task(pool.run(io_blocking_func, 12))
    if some_conditional:
        task.cancel()
    results = await asyncio.gather(task, other_task, return_exceptions=True)
    ...

To me, this looks like excessive boilerplate, particularly for a higher level API. Even for rather straightforward behavior, it requires nested function calls. If we were to return a task directly, this would be significantly cleaner:

async with asyncio.ThreadPool() as pool:
    task = pool.run(io_blocking_func, 10, kwarg1='test')
    other_task = pool.run(io_blocking_func, 12)
    if some_conditional:
        task.cancel()
    results = await asyncio.gather(task, other_task, return_exceptions=True)
    ...

Since asyncio.ThreadPool is intended to be a high-level API, I don't think it's an issue to return a Task from it's run() method. It would make it significantly easier and more convenient to work with from a user perspective.

Thoughts?