HB334bxS

1# -*- coding: utf-8 -*-
2import gevent.monkey
3gevent.monkey.patch_all()
4
5import collections
6import threading
7import time
8import random
9import sys
10
11import logging
12logging.basicConfig()
13log = logging.getLogger('foo')
14log.setLevel(logging.CRITICAL)
15
16import pymysql as dbapi
17#from mysql import connector as dbapi
18
19
20class SimplePool(object):
21    def __init__(self):
22        self.checkedin = collections.deque([
23            self._connect() for i in range(50)
24        ])
25        self.checkout_lock = threading.Lock()
26        self.checkin_lock = threading.Lock()
27
28        # alternate form
29        # self.makeup_count = 0
30
31    def _connect(self):
32        return dbapi.connect(
33            user="scott", passwd="tiger",
34            host="localhost", db="test")
35
36    def get(self):
37        with self.checkout_lock:
38            while not self.checkedin:
39
40                # reconnect here if we did the "defer" version
41                # if self.makeup_count:
42                #    self.makeup_count -= 1
43                #    return self._connect()
44
45                time.sleep(.1)
46            return self.checkedin.pop()
47
48    def chuck_conn(self, conn):
49        try:
50            conn.close()
51        except:
52            log.error("Exception during close", exc_info=True)
53
54        # defer reconnection..
55        # self.makeup_count += 1
56
57        # or do it now, but check size of pool
58        if len(self.checkedin) < 50:
59            conn = self._connect()
60            with self.checkin_lock:
61                self.checkedin.append(conn)
62
63    def return_conn(self, conn):
64        try:
65            conn.rollback()
66        except:
67            log.error("Exception during rollback", exc_info=True)
68            self.chuck_conn(conn)
69        else:
70            with self.checkin_lock:
71                self.checkedin.append(conn)
72
73
74def verify_connection_id(conn):
75    cursor = conn.cursor()
76    try:
77        cursor.execute("select connection_id()")
78        row = cursor.fetchone()
79        return row[0]
80    except:
81        return None
82    finally:
83        cursor.close()
84
85
86def execute_sql(conn, sql, params=()):
87    cursor = conn.cursor()
88    cursor.execute(sql, params)
89    lastrowid = cursor.lastrowid
90    cursor.close()
91    return lastrowid
92
93
94pool = SimplePool()
95
96# SELECT * FROM table_b WHERE a_id not in
97# (SELECT id FROM table_a) ORDER BY a_id DESC;
98
99PREPARE_SQL = [
100    "DROP TABLE IF EXISTS table_b",
101    "DROP TABLE IF EXISTS table_a",
102    """CREATE TABLE table_a (
103        id INT NOT NULL AUTO_INCREMENT,
104        data VARCHAR (256) NOT NULL,
105        PRIMARY KEY (id)
106    ) engine='InnoDB'""",
107    """CREATE TABLE table_b (
108        id INT NOT NULL AUTO_INCREMENT,
109        a_id INT NOT NULL,
110        data VARCHAR (256) NOT NULL,
111        -- uncomment this to illustrate where the driver is attempting
112        -- to INSERT the row during ROLLBACK
113        -- FOREIGN KEY (a_id) REFERENCES table_a(id),
114        PRIMARY KEY (id)
115    )  engine='InnoDB'
116    """]
117
118connection = pool.get()
119for sql in PREPARE_SQL:
120    execute_sql(connection, sql)
121connection.commit()
122pool.return_conn(connection)
123print("Table prepared...")
124
125
126def transaction_kill_worker():
127    while True:
128        try:
129            connection = None
130            with gevent.Timeout(0.1):
131                connection = pool.get()
132                rowid = execute_sql(
133                    connection,
134                    "INSERT INTO table_a (data) VALUES (%s)", ("a",))
135                gevent.sleep(random.random() * 0.2)
136
137                try:
138                    execute_sql(
139                        connection,
140                        "INSERT INTO table_b (a_id, data) VALUES (%s, %s)",
141                        (rowid, "b",))
142
143                    # this version prevents the commit from
144                    # proceeding on a bad connection
145                    # if verify_connection_id(connection):
146                    #    connection.commit()
147
148                    # this version does not.  It will commit the
149                    # row for table_b without the table_a being present.
150                    connection.commit()
151
152                    pool.return_conn(connection)
153                except Exception:
154                    connection.rollback()
155                    pool.return_conn(connection)
156                    sys.stdout.write("$")
157        except gevent.Timeout:
158            # try to return the connection anyway
159            if connection is not None:
160                pool.chuck_conn(connection)
161            sys.stdout.write("#")
162        except Exception:
163            # logger.exception(e)
164            sys.stdout.write("@")
165        else:
166            sys.stdout.write(".")
167        finally:
168            if connection is not None:
169                pool.return_conn(connection)
170
171
172def main():
173    for i in range(50):
174        gevent.spawn(transaction_kill_worker)
175
176    gevent.sleep(3)
177
178    while True:
179        gevent.sleep(5)
180
181
182if __name__ == "__main__":
183    main()