pymysql failures w/ greenlets, sqla edition

test.py

import eventlet
eventlet.monkey_patch()

from sqlalchemy  import exc

import itertools
import pymysql
import random
from eventlet.green import time
import greenlet
import traceback

greenlet_id = itertools.count(1)


def do_work(engine):
    our_id = next(greenlet_id)
    conn = engine.connect()
    while True:
        try:
            print("greenlet %d working..." % our_id)
            result = conn.execute("select 1")
            result.scalar()
            with conn.begin():
                try:
                    result = conn.execute(
                        "select sleep(%s)", (random.random() * 2, ))
                    time.sleep(random.random())
                    result.fetchall()[0]

                    for i in range(10):
                        conn.execute(
                            "insert into stuff (data) values (%s)",
                            (("some_data_%f" % random.random()), )
                        )
                except greenlet.GreenletExit as ex:
                    print("exit exception, we need to invalidate: %r" % ex)
                    if ensure_greenlet_exit_handled:
                        conn.invalidate()
                    raise
        except exc.DBAPIError as err:
            print("DBAPI error occurred, SQLA should "
                  "invalidate the connection: %s" % err)
            traceback.print_exc()
        except Exception as err:
            print("totally unexpected error occurred: %r" % err)
        except greenlet.GreenletExit as ex:
            print("exit exception, we need to invalidate: %r" % ex)
            if ensure_greenlet_exit_handled:
                conn.invalidate()
            break

if __name__ == '__main__':
    ensure_greenlet_exit_handled = True

    num = 10

    from sqlalchemy import create_engine
    engine = create_engine(
        "mysql+pymysql://scott:[email protected]/test",
        pool_size=num, max_overflow=0)

    conn = engine.connect()
    conn.execute("drop table if exists stuff")
    conn.execute(
        "create table stuff(id integer primary key auto_increment, "
        "data varchar(200))")
    conn.close()

    greenlets = [
        eventlet.spawn(do_work, engine)
        for idx in range(num)]

    while True:
        for idx in range(num):
            time.sleep(random.random())
            print("kill...%d" % idx)

            eventlet.greenthread.kill(greenlets[idx])
            time.sleep(.5)
            greenlets[idx] = eventlet.spawn(do_work, engine)

see also https://bitbucket.org/zzzeek/sqlalchemy/issues/3803/dbapi-connections-go-invalid-on