22XbHMLK

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.DEBUG)
15
16import pymysql as dbapi
17
18
19class SimplePool(object):
20    def __init__(self):
21        self.checkedin = collections.deque([
22            self._connect() for i in range(50)
23        ])
24        self.checkout_lock = threading.Lock()
25        self.checkin_lock = threading.Lock()
26
27    def _connect(self):
28        return dbapi.connect(
29            user="scott", passwd="tiger",
30            host="localhost", db="test")
31
32    def get(self):
33        with self.checkout_lock:
34            while not self.checkedin:
35                time.sleep(.1)
36            return self.checkedin.pop()
37
38    def return_conn(self, conn):
39        try:
40            conn.rollback()
41        except:
42            log.error("Exception during rollback", exc_info=True)
43            try:
44                conn.close()
45            except:
46                log.error("Exception during close", exc_info=True)
47
48            # recycle to a new connection
49            conn = self._connect()
50        with self.checkin_lock:
51            self.checkedin.append(conn)
52
53
54def execute_sql(conn, sql, params=()):
55    cursor = conn.cursor()
56    cursor.execute(sql, params)
57    lastrowid = cursor.lastrowid
58    cursor.close()
59    return lastrowid
60
61
62pool = SimplePool()
63
64# SELECT * FROM table_b WHERE a_id not in
65# (SELECT id FROM table_a) ORDER BY a_id DESC;
66
67PREPARE_SQL = """
68DROP TABLE IF EXISTS table_b;
69DROP TABLE IF EXISTS table_a;
70CREATE TABLE table_a (
71    id INT NOT NULL AUTO_INCREMENT,
72    data VARCHAR (256) NOT NULL,
73    PRIMARY KEY (id)
74) engine='InnoDB';
75
76CREATE TABLE table_b (
77    id INT NOT NULL AUTO_INCREMENT,
78    a_id INT NOT NULL,
79    data VARCHAR (256) NOT NULL,
80    -- uncomment this to illustrate where the driver is attempting
81    -- to INSERT the row during ROLLBACK
82    -- FOREIGN KEY (a_id) REFERENCES table_a(id),
83    PRIMARY KEY (id)
84)  engine='InnoDB';
85"""
86
87connection = pool.get()
88execute_sql(connection, PREPARE_SQL)
89connection.commit()
90pool.return_conn(connection)
91print("Table prepared...")
92
93
94def transaction_kill_worker():
95    while True:
96        try:
97            connection = None
98            with gevent.Timeout(0.1):
99                connection = pool.get()
100                rowid = execute_sql(
101                    connection,
102                    "INSERT INTO table_a (data) VALUES (%s)", ("a",))
103                gevent.sleep(random.random() * 0.2)
104
105                try:
106                    execute_sql(
107                        connection,
108                        "INSERT INTO table_b (a_id, data) VALUES (%s, %s)",
109                        (rowid, "b",))
110                    connection.commit()
111                    pool.return_conn(connection)
112                except Exception:
113                    #log.error("error", exc_info=True)
114                    connection.rollback()
115                    pool.return_conn(connection)
116                    sys.stdout.write("$")
117        except gevent.Timeout:
118            # try to return the connection anyway
119            if connection is not None:
120                pool.return_conn(connection)
121            sys.stdout.write("#")
122        except Exception:
123            # logger.exception(e)
124            sys.stdout.write("@")
125        else:
126            sys.stdout.write(".")
127        finally:
128            if connection is not None:
129                pool.return_conn(connection)
130
131
132def main():
133    for i in range(50):
134        gevent.spawn(transaction_kill_worker)
135
136    gevent.sleep(3)
137
138    while True:
139        gevent.sleep(5)
140
141
142if __name__ == "__main__":
143    main()