CODEWITHSUNDEEP
pythonmysqlpython-multiprocessingtornadothreadpoolexecutor
hyp3rv1p3r
hyp3rv1p3r

Reputation: 21

Problem with two concurrent workers accessing mysql with tornado

I have this simple program composed by two workers: Worker1 inserts records that Workers2 should read. The problem is that during execution Workers2 reads 0 records. Launched separately from CLI they work correctly. The "culprit" seems to be tornado. Any idea?

import time
import munch
from tornado import concurrent
from tornado.ioloop import IOLoop
import logging
import pymysql

config = munch.munchify({"mysql_host": "127.0.0.1", "mysql_port": 3306, "mysql_user": "", "mysql_password": "", "mysql_db": ""})

executor = concurrent.futures.ThreadPoolExecutor(max_workers=8)

class Worker1():
    CONST_TYPE_ID = 'worker1'

    def __init__(self):
        self.conn = pymysql.connect(host=config.mysql_host, port=config.mysql_port, user=config.mysql_user,
                                    passwd=config.mysql_password, db=config.mysql_db, charset='UTF8MB4',
                                    local_infile=True)
        self.cursor = self.conn.cursor(pymysql.cursors.DictCursor)

    def single_run(self):
        print("Single run of worker:", self.CONST_TYPE_ID)
        query = "insert into readwrite (timestamp) values (now())"
        self.conn.ping(True)
        self.cursor.execute(query)
        self.conn.commit()

class Worker2():
    CONST_TYPE_ID = 'worker2'

    def __init__(self):
        self.conn = pymysql.connect(host=config.mysql_host, port=config.mysql_port, user=config.mysql_user,
                                    passwd=config.mysql_password, db=config.mysql_db, charset='UTF8MB4',
                                    local_infile=True)
        self.cursor = self.conn.cursor(pymysql.cursors.DictCursor)

    def single_run(self):
        print("Single run of worker:", self.CONST_TYPE_ID)
        query = "select * from readwrite"
        self.conn.ping(True)
        self.cursor.execute(query)
        for row in self.cursor.fetchall():
            print(row)

def run_worker(worker):
    i = 0
    instance = worker()
    while True:
        try:
            i += 1
            print("Starting {name} - run {iter}".format(name=instance.CONST_TYPE_ID, iter=i))
            instance.single_run()
        except Exception as e:
            logging.exception('Worker {} got error {!r}, errno is {}'.format(instance.CONST_TYPE_ID, e, e.args[0]))
        print("Waiting...")
        time.sleep(1)

def main():
    workers = [Worker1, Worker2]
    executor.map(run_worker, workers)
    IOLoop.current().start()

if __name__ == '__main__':
    main()

Upvotes: 1

Views: 56

Answers (1)

hyp3rv1p3r
hyp3rv1p3r

Reputation: 21

Solved by adding self.conn.commit() after self.cursor.execute(query) in Worker2.

Upvotes: 1

Related Questions