Replace Thread with ThreadPoolExecutor

[culhatsker]

Description

Use concurrent.futures.ThreadPoolExecutor instead of an own threading.Thread-based implementation.

Fixes hanging when connection is not closed before exiting Python.

[culhatsker]

Turned out tortoise-orm uses .start() method that vanished when I removed Thread base.

[amyreese]

Can you provide a repro case that shows the behavior you're trying to fix with this? I'm not fond of using the thread pool, and there might be a simpler way to fix bad behavior.

[Adminiuga]

I'm not positive, but very likely I'm experiencing same issue: when a test in a test suite throws, it actually never exits the test, unless database connection is properly closed.

[Adminiuga]

I can confirm that with ThreadExecutor pool throwing an exception in the test does not hang up the test.

[Adminiuga]

how to reproduce:

import os
import aiosqlite


async def test_aiothread(loop, tmpdir):
    dbname = os.path.join(tmpdir, "test.db")
    conn = await aiosqlite.connect(dbname)

    assert False

running with pytest test_aiosqlite.py hangs indefinitely at the end:

============================================= test session starts =============================================
platform linux -- Python 3.7.3, pytest-5.4.3, py-1.8.1, pluggy-0.13.1
rootdir: /home/doh/src/zigpy
plugins: aiohttp-0.3.0, forked-1.2.0, asyncio-0.12.0, cov-2.10.0, xdist-1.33.0
collected 1 item                                                                                              

test_aiosqlite.py F                                                                                     [100%]

================================================== FAILURES ===================================================
___________________________________________ test_aiothread[pyloop] ____________________________________________

loop = <_UnixSelectorEventLoop running=False closed=False debug=False>
tmpdir = local('/tmp/pytest-of-lex/pytest-3/test_aiothread_pyloop_0')

    async def test_aiothread(loop, tmpdir):
        dbname = os.path.join(tmpdir, "test.db")
        conn = await aiosqlite.connect(dbname)
    
>       assert False
E       assert False

test_aiosqlite.py:10: AssertionError
=========================================== short test summary info ===========================================
FAILED test_aiosqlite.py::test_aiothread[pyloop] - assert False
============================================== 1 failed in 0.20s ==============================================

[amyreese]

The reason it hangs is because you should either be using the connection as a context manager, or having something await the close() method in cases where you have to abort execution. Otherwise, using daemon threads (like I believe thread pool executor does) would result in potential loss of information at worst, or leaving transactions in weird states at best.

I'm pretty sure that if you instead constructed your tests like one of the two examples below, you should not see hung threads (and if you do, please give a repro, as that is definitely a bug):

async def test_whatever(...):
    async with aiosqlite.connect(...) as conn:
        assert False

class TestWhatever(AsyncTestCase):
    async def setUp(self):
        self.conn = await aiosqlite.connect(...)

    async def tearDown(self):
        await self.conn.close()

    async def test_whatever(self):
        assert False

[amyreese]

Another alternative that I would be alright adding is something like a __del__ method on the aiosqlite.Connection object that signals a graceful shutdown to the separate thread as a last resort, but whether that would work with reference counting across threads is a question that would need to be answered and verified.

[Adminiuga]

I'm pretty sure that if you instead constructed your tests like one of the two examples below, you should not see hung threads

Yes, I agree completely, but that was just an example. Unfortunately the app in question does not use sqlite with context managers and I'm testing the app per se, not the aiosqlite. And while the tests are expected to pass, when they do not pass, it hangs up until a timeout.

[amyreese]

For purposes of testing, ideally you would be mocking out the db connection entirely so that it wouldn't be a factor, but if you absolutely can't do that, then you can always set .daemon=True on the aiosqlite connection object before awaiting it. This will allow the Python runtime to forcibly terminate the connection when all non-daemon threads have finished executing.

Do note, however, that you should not use this option in production instances, as that can result in aforementioned dataloss if there are pending transactions when the thread is terminated. Production instances should always make sure to await conn.close() or use the connection as a context manager to ensure that close is correctly awaited in event of errors that abort execution.