Issue: Dealys with multiprocessing
Author: codez266
State: closed
Created: 2019-11-28 08:13:41

Hi, I have a code that uses peewee to access a MySQL backend. The code needs to look for the data related to a bunch of wikipedia pages first in the MySQL table, and if not present fetch it from the internet using an API. I'm multiprocessing to fetch data for multiple pages simultaneously. Each process handles one page. My database access just contains read and join operations. However, with peewee, the processes seem to be stuck waiting for each other, even though i'm doing no writes at all. I'm not sure what underlying locking it uses.

I'm sorry, this is not something I can help with. No code, extremely vague description, no information about where things are stuck.

You might just make sure that before you fork new processes you are creating a new database connection, as an open connection in the parent-process might get inherited by the child process which could be buggy / problematic.

filter.py:

```
from mwapilib import get_qual_samples, get_mwapi_session,\
    get_page_revs_with_quals,get_user_info, chunkify, main_header,\
    talk_header, user_header, get_page_edits_from_db
from revscoring.utilities.util import dump_observation, read_observations
from concurrent.futures import ThreadPoolExecutor

def get_all_page_revs(labelings, workers = 5):
    all_edits = pd.DataFrame([], columns = main_header)
    executor = ThreadPoolExecutor(max_workers=workers)
    session = get_mwapi_session()
    partial_get_page_revs = partial(get_page_revs_with_quals, session)
    #edits_from_db = get_page_edits_from_db(labelings)
    for edits in executor.map(partial_get_page_revs, labelings):
    for labeling in labelings:
        edits = partial_get_page_revs(labeling)
        all_edits = all_edits.append(edits, sort = False)
    pdb.set_trace()
    return all_edits
```


mwapilib:


```
import mwapi
import traceback
import numpy as np
import sys
import pdb
from itertools import islice
import pandas as pd
import random
import mwreverts.api
from mw.lib import title as mwtitle
import traceback
from multiprocessing import Pool
from models import Page, Quality, Revision, User
from models import database as db
from functools import partial
from revscoring.utilities.util import dump_observation, read_observations
import peewee
import json
import re

def get_page_edits(session, labeling):
    pageid = labeling['pageid']
    page_revs_arr = []
    # First, try to look for the pageid in the database
    page_present = Revision.select().where(Revision.rev_page ==
            pageid).count()
    page_present = 0
    #if False:
    if page_present > 0:
        page_revs = []
        # get results
        revs = (Page
                .select(Page, Revision)
                .join(Revision, on=(Page.page_id==Revision.rev_page).alias('rev'))
                .where(Page.page_id == pageid
                ))
        for page in revs:
            rev = page.rev
            page_revs.append([rev.rev_id, rev.rev_page, rev.rev_comment,
                    rev.rev_user, rev.rev_user_text, rev.rev_timestamp,
                    rev.rev_minor_edit, rev.rev_len,
                    rev.rev_sha1, rev.rev_quality, page.page_title])

        return pd.DataFrame(page_revs, columns = main_header)

    elif page_present == 0:
        page_revs = []
        doc = session.get(
            action='query', prop='revisions', formatversion=2,
            pageids = pageid, rvprop=['ids', 'timestamp', 'user', 'size', 'comment',
                'sha1', 'flags', 'userid'],
            continuation=True, rvlimit=500, rvslots='main', rvdir='newer')
        title = ''
        for result in doc:
            pages = result['query']['pages']
            for page in pages:
                if 'revisions' not in page:
                    continue
                for rev in page['revisions']:
                    revid = rev['revid']
                    pageid = page['pageid']
                    title = mwtitle.normalize(page['title'])
                    ts = pd.Timestamp(rev.get('timestamp', ''))

                    rev = (
                            revid,
                            pageid,
                            rev.get('comment', ''),
                           rev.get('userid'),
                            rev.get('user'),
                            rev.get('timestamp', ''),
                            1 if rev.get('minor', False) else 0,
                            rev.get('size', 0),
                            rev.get('sha1', '')
                            )
                    page_revs.append(rev)

        talk_id = labeling['talk_page_id']
        talk_edits = get_page_qualities(session, talk_id)
        page_revs_with_quals = []
        for rev in page_revs:
            ts = rev[5]
            t_filter = talk_edits['rev_timestamp'] < ts
            rev_new = list(rev)
            rev_new[5] = pd.Timestamp(ts).strftime('%Y%m%d%H%M%S')
            try:
                max_ts = talk_edits.where(t_filter)['rev_timestamp'].idxmax()
                quals.append(talk_edits.loc[max_ts, 'quality'])
                rev_new = list(rev)
                rev_new[5] = pd.Timestamp(ts).strftime('%Y%m%d%H%M%S')
                page_revs_with_quals.append(tuple(rev_new) + (class_map[quals[-1]],))
            except:
                sys.stderr.write('#')
                sys.stderr.flush()
                page_revs_with_quals.append(tuple(rev_new) + (0,))
                quals.append('non-def')

    Revision.insert_many(page_revs_with_quals,fields=['rev_id', 'rev_page',
        'rev_comment', 'rev_user', 'rev_user_text', 'rev_timestamp',
        'rev_minor_edit', 'rev_len', 'rev_sha1' ,'rev_quality']).execute()
    Page.insert(page_id = pageid, page_namespace = 0, page_title =
            title).execute()
    edits = pd.DataFrame([rev+(title,) for rev in page_revs_with_quals], columns = main_header)
    return edits
```

the filter class creates multiple processes and calls the get_page_revs from mwapilib.py in each of them. The get_page_revs function uses the Page, and Revision db models to query existing data and inserts new data after fetching from api.

The processes are getting stuck in deadlock in the get_page_edits function.

Models:
```
from peewee import *

database = MySQLDatabase(....})

class UnknownField(object):
    def __init__(self, *_, **__): pass

class BaseModel(Model):
    class Meta:
        database = database

class Page(BaseModel):
    page_id = IntegerField()
    page_namespace = IntegerField()
    page_title = CharField()

    class Meta:
        table_name = 'page'
        primary_key = False

class Revision(BaseModel):
    rev_comment = CharField()
    rev_id = IntegerField()
    rev_len = IntegerField(null=True)
    rev_minor_edit = IntegerField()
    rev_page = IntegerField()
    rev_quality = IntegerField()
    rev_sha1 = CharField(null=True)
    rev_timestamp = CharField()
    rev_user = IntegerField()
    rev_user_text = CharField(null=True)

    class Meta:
        table_name = 'revision'
        primary_key = False
```

It's called "**Thread**PoolExecutor" - meaning threads, not processes.

> The processes are getting stuck in deadlock in the get_page_edits function.

Where exactly?

@codez266 you fell into a trap that many Python noobs are falling into.
Threads in Python are not CPU threads, but rather execution threads internal to the CPython interpreter. All your threads are running on a single CPU. Running alot of threads with IO will
cause a lot of context switching and slow execution. 
Consider re-designing your software in such way that you delegate the heavy processing to the database, not the Python client. If you have no other choice, consider using 
```
from multiprocessing import Pool
```
But this comes with some caveats.