Issue: peewee and multiprocess
Author: pw717
State: closed
Created: 2017-06-12 13:04:41

Hi,

I'm trying to use a mysql+pool connexion within a multiprocess treatment.
I have always the errors below :
ERROR : (2013, 'Lost connection to MySQL server during query')
ERROR : (2006, 'MySQL server has gone away')

I understand that the processes tries to use the same connection already in use.

I thought the use of a mysql pool would have prevented this.

I'm not sure wether it's an issue, a wrong usage of the peewee lib or a request.
When each process runs it's own MySQLdb connection than it's just fine (see test number 4 of the code below).
But I'd like to stay in the "peewee" pool of connexions, I don't know if this will be possible. 
So I would rather say that I have a wrong usage of the lib, but could you be kind enough to guide me?

Here is below a piece of code in order to reproduce the behaviour, please see in particular test number 3 which leads to the MySQL errors.

I'm running this test under mac osx 10.12.5 with sierra.

Thank you in advance for your help.

```python
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
from datetime import date, datetime
from os import path
from os.path import expanduser
import traceback
import multiprocessing
from multiprocessing import Process, Pool, Value, Manager

import MySQLdb
import MySQLdb.cursors

from peewee import *
from playhouse.db_url import connect
from playhouse.shortcuts import model_to_dict
from playhouse.pool import *

def InitDB(base, usr, pwd, nb_conn, mode):
    db = None
    usr = None
    pwd = None
   
    port = 3306
    cnx_args = {"autocommit": True, "autorollback": False}

    args = ""
    srv = "mysql+pool"
    args = "?max_connections=%d&stale_timeout=300" % nb_conn

    if usr and pwd:
        cnx = "%s://%s:%s@localhost%s/%s%s" % (srv, usr, pwd, ":%s" % port if port else "", base, args)
        db = connect(cnx, **cnx_args)
    elif usr:
        cnx = "%s://%s@localhost%s/%s%s" % (srv, usr, ":%s" % port if port else "", base, args)
        db = connect(cnx, read_default_file = '~/.my.cnf', **cnx_args)
    else:
        cnx = "%s://localhost%s/%s%s" % (srv, ":%s" % port if port else "", base, args)
        if "mysql" in srv:
            db = connect(cnx, read_default_file = '~/.my.cnf', **cnx_args)
        else:
            db = connect(cnx, **cnx_args)

    return db

def MySqlConn(base, usr=None, pwd=None):
    db = None
    if usr and pwd:
        db = MySQLdb.connect(host="localhost", port=3306, user=usr, passwd=pwd, db=base, cursorclass=MySQLdb.cursors.DictCursor)
    elif usr:
        db = MySQLdb.connect(host="localhost", port=3306, user=usr, read_default_file = '~/.my.cnf', db=base, cursorclass=MySQLdb.cursors.DictCursor)
    else:
        db = MySQLdb.connect(host="localhost", port=3306, read_default_file = '~/.my.cnf', db=base, cursorclass=MySQLdb.cursors.DictCursor)

    return db

def InitModels(db):
    global _MODELS

    # ------- #
    # Persons #
    # ------- #
    class Person(Model):
        name		= CharField(max_length = 60, null = False)
        birthday	= DateField(null = False)

        class Meta:
            db_table = 'persons'
            database = db
    _MODELS['p'] = Person

    # Create table :
    if db.is_closed():
        db.connect()
    if Person.table_exists():
        db.drop_table(Person)
    db.create_table(Person)

    # Populate table :
    for i, name in enumerate(['Alfred', 'Bernard', 'Charles', 'Dimitri', 'Eric', 'Franck', 'Gloria', 'Hope', 'Isadora', 'Junon',
                            'Kevin', 'Leonora', 'Marcel', 'Nora', 'Oscar', 'Patrick', 'Quentin', 'Romain', 'Samuel', 'Tadeus',
                            'Ursula', 'Virginia', 'Wilfrid', 'Xavier', 'Yann', 'Zoé']):
        p = Person(name=name, birthday=date(1960, 1, i+1))
        p.save()
    db.close()
    return

def PurgeExamples():
    global _MODELS

    for k in _MODELS.keys():
        m = _MODELS[k]
        db = m._meta.database
        if db.is_closed():
            db.connect()
        if m.table_exists():
            db.drop_table(m)
        db.close()
    return

def ReadModel(counter=None):
    global _MODELS, _STOP

    # Get recordsets info :
    if counter and _STOP.value:
        PrintWarn("%sERROR IN POOL - Function has been stopped !!!" % "counter = %s - " % counter if counter else "")
        return
    m = _MODELS['p']
    if m:
        db = m._meta.database
        try:
            if db.is_closed():
                db.connect()

            rs = m.select()
            nb = len(rs)
            msg = "%s%s records found in Modele %s - table = %s" % ("counter = %s - " % counter if counter else "", nb, m.__name__, m._meta.db_table)
            if nb == 0:
                PrintWarn(msg)
            else:
                PrintOk(msg)
            #for p in rs:
            #    print "\t%s was born on %s" % (p.name, p.birthday)

        except Exception as exc:
            PrintError("%sERROR : %s" % ("counter = %s - " % counter if counter else "", exc))
            print traceback.format_exc()
            if counter:
                _STOP.value = True
        finally:
            if not db.is_closed():
                db.close()
    return

def ReadTable(counter=None):
    global _MODELS, _STOP

    # Get recordsets info :
    if counter and _STOP.value:
        PrintWarn("%sERROR IN POOL - Function has been stopped !!!" % "counter = %s - " % counter if counter else "")
        return
    conn = MySqlConn("tests")
    if conn:
        cursor = None
        try:
            cursor = conn.cursor()
            query = "SELECT name, birthday FROM persons;"
            cursor.execute(query)
            nb = cursor.rowcount
            msg = "%s%s records found in table = persons" % ("counter = %s - " % counter if counter else "", nb)
            if nb == 0:
                PrintWarn(msg)
            else:
                PrintOk(msg)
            #rs = cursor.fetchall()
            #for r in rs:
            #    print "\t%s was born on %s" % (r["name"], r["birthday"])

        except Exception as exc:
            PrintError("%sERROR : %s" % ("counter = %s - " % counter if counter else "", exc))
            print traceback.format_exc()
            if counter:
                _STOP.value = True
        finally:
            # Close the connection when it is open.
            if cursor:
                cursor.close()
            if conn:
                conn.close()
    return

def MultiProcRead(mode="peewee"):
    global _STOP

    _STOP = Value('b', False)
    trace_fct =  "MultiProcRead(mode=\"%s\")" % mode
    PrintInfo("\n%s - BEGIN %s" % (datetime.now(), trace_fct))
    if mode == "peewee":
        fct = ReadModel
    else:
        fct = ReadTable
    nb_procs = 10
    jobs = []
    msg = "load processes loop"
    PrintBold("Begin %s - %s" % (datetime.now(), msg))
    for i in range(0, nb_procs):
        out_list = list()
        process = multiprocessing.Process(target=fct, args=(i, ))
        jobs.append(process)
    PrintBold("End %s - %s" % (datetime.now(), msg))

    # Start the processes (i.e. calculate the random number lists)          
    msg = "start jobs loop"
    PrintBold("Begin %s - %s" % (datetime.now(), msg))
    for j in jobs:
        j.start()
    PrintBold("End %s - %s" % (datetime.now(), msg))

    # Ensure all of the processes have finished
    msg = "\"join\" operation loop"
    PrintBold("Begin %s - %s" % (datetime.now(), msg))
    for j in jobs:
        j.join()
    PrintBold("End %s - %s" % (datetime.now(), msg))

    PrintInfo("\n%s - END %s" % (datetime.now(), trace_fct))
    return

def PrintBold(string):
    pref = u"\x1b[1m"
    suff = u"\x1b[0m"
    print "{0}{1}{2}".format(pref, string, suff)
    return

def PrintInfo(string):
    pref = u"\x1b[1;37;44m"
    suff = u"\x1b[0m"
    print "{0}{1}{2}".format(pref, string, suff)
    return

def PrintOk(string):
    pref = u"\x1b[1;30;42m"
    suff = u"\x1b[0m"
    print "{0}{1}{2}".format(pref, string, suff)
    return

def PrintWarn(string):
    pref = u"\x1b[1;37;45m"
    suff = u"\x1b[0m"
    print "{0}{1}{2}".format(pref, string, suff)
    return

def PrintError(string):
    pref = u"\x1b[1;37;41m"
    suff = u"\x1b[0m"
    print "{0}{1}{2}".format(pref, string, suff)
    return

# ------- #
# GLOBALS #
# ------- #
_MODELS = {}
_MNGR = Manager()
_STOP = Value('b', False)
_LOCK = multiprocessing.Manager().Lock()

# ------------- #
# Fonction main #
# ------------- #
def main(argv):

    db = None
    try:
        db = InitDB("tests", usr=None, pwd=None, nb_conn=5, mode="url")
        InitModels(db)

        # OK : 1st test : simple read with peewee connect
        ReadModel()

        # NOK : 2nd test : "with pool of processes within peewee"
        MultiProcRead(mode="peewee")

        # OK : 3rd test : simple read with MySQLdb cursor
        ReadTable()

        # OK : 4th test : "with pool of processes within MySQLdb"
        # Each process runs it's own connection : then it's fine
        MultiProcRead(mode="table")

        PurgeExamples()
    except Exception as exc:
        PrintError("ERROR : %s" % exc)
        print traceback.format_exc()

    finally:
        if db and not db.is_closed():
            db.close()
    
    return

if __name__ == "__main__":
    main(sys.argv[1:])
```

Last question if you may : 
I don't see where I should "play" your 

```python
def before_request_handler():
    database.connect()

def after_request_handler():
    database.close()
```

Is it reserved to web context only ?


I apologize but I don't have the time to read all that code.

Depending on where you are fork-ing the process, you could definitely run into some problems. It looks from a quick read that you may be forking after you've initialized the connection, etc. This is probably what's causing the problems. You'll need to be sure that you don't initialize any connections until after you fork, as connections cannot be shared among multiple processes...even if you're using the pool.

Thank you very much four your answer!
I understand you.
It is true that the connections are created first, then I launch the pool of processes.
When I create the connection inside each process, it's just fine !

```python
def ProxyModel():
    db = InitDB("tests", usr=None, pwd=None, nb_conn=5, mode="url")
    proxy = Proxy()
    proxy.initialize(db)
    return proxy

def ReadModelProxy(counter=None):
    global _MODELS, _STOP

    proxy = None
    # Get recordsets info :
    if counter and _STOP.value:
        PrintWarn("%sERROR IN POOL - Function has been stopped !!!" % "counter = %s - " % counter if counter else "")
        return
    m = _MODELS['p']
    if m:
        try:
            proxy = ProxyModel()
            with Using(proxy, [m]):

                rs = m.select()
                nb = len(rs)
                msg = "%s%s records found in Modele %s - table = %s" % ("counter = %s - " % counter if counter else "", nb, m.__name__, m._meta.db_table)
                if nb == 0:
                    PrintWarn(msg)
                else:
                    PrintOk(msg)
                #for p in rs:
                #    print "\t%s was born on %s" % (p.name, p.birthday)

        except Exception as exc:
            PrintError("%sERROR : %s" % ("counter = %s - " % counter if counter else "", exc))
            print traceback.format_exc()
            if counter:
                _STOP.value = True
        finally:
            if proxy and not proxy.is_closed():
                proxy.close()
    return
```
and in main :
```python
        # OK : 5th test : "with pool of processes within proxy peewee"
        MultiProcRead(mode="proxy")
```

Thank-you very much again for all your work!
It's a great lib!

PS : Yes, the code seems a bit long: I thought you could just copy/paste it and run as it is!
It only creates a `persons`  table in `tests`mysql` database and  adds some data in it.

Then it does a first simple read of the table test.
After that the same read called within a pool of 10 processes.
This is where I have the errors.

The other 2 tests left to show that when I each process runs it's own connection it's ok.