Fix connect errors leaking threads

[ryban]

Errors from connect() would not clean up their worker thread. This script will reproduce the leak by creating an engine with a pool_size of 1 and no timeout or overflow so the second connect call with fail with the error QueuePool limit of size 1 overflow 0 reached, connection timed out, timeout 0.

import threading

from alchimia import wrap_engine
from sqlalchemy import create_engine

from twisted.internet.defer import inlineCallbacks, DeferredList
from twisted.internet.task import react, deferLater


@inlineCallbacks
def main(reactor):
    print('baseline', threading.enumerate())
    db_url = "postgresql+pg8000://postgres:password@localhost"
    engine = wrap_engine(reactor, create_engine(db_url, pool_size=1, max_overflow=0, pool_timeout=0))
    # One thread created for the _engine_worker
    print('engine created', threading.enumerate())

    @inlineCallbacks
    def connect(i):
        # Creates and closes one thread
        print('making connection', i, threading.enumerate())
        conn = yield engine.connect()
        print('connection made', i, threading.enumerate())
        yield conn.close()

    # First connect() will succeed, the second will fail.
    # Without the fix the second thread will leak
    results = yield DeferredList([connect(0), connect(1)], consumeErrors=True)
    for success, result in results:
        if not success:
            print(result.getErrorMessage())

    # Wait a bit for threads to shutdown
    yield deferLater(reactor, 0.1, lambda: None)
    # should only be the main thread and the _engine_worker thread
    print('connections done', threading.enumerate())


if __name__ == "__main__":
    react(main, [])

Example output showing Thread-3 leak

('baseline', [<_MainThread(MainThread, started 139895101339392)>])
('engine created', [<_MainThread(MainThread, started 139895101339392)>, <Thread(Thread-1, started daemon 139895036782336)>])
('making connection', 0, [<_MainThread(MainThread, started 139895101339392)>, <Thread(Thread-1, started daemon 139895036782336)>])
('making connection', 1, [<_MainThread(MainThread, started 139895101339392)>, <Thread(Thread-1, started daemon 139895036782336)>, <Thread(Thread-2, started daemon 139895028389632)>])
('connection made', 0, [<_MainThread(MainThread, started 139895101339392)>, <Thread(Thread-1, started daemon 139895036782336)>, <Thread(Thread-3, started daemon 139895019996928)>, <Thread(Thread-2, started daemon 139895028389632)>])
QueuePool limit of size 1 overflow 0 reached, connection timed out, timeout 0 (Background on this error at: http://sqlalche.me/e/3o7r)
('connections done', [<_MainThread(MainThread, started 139895101339392)>, <Thread(Thread-1, started daemon 139895036782336)>, <Thread(Thread-3, started daemon 139895019996928)>])