Issue: pool.map cannot pickle database or results object
Author: tsikerdekis
State: closed
Created: 2019-03-12 04:54:39

I have a db file:
```
from datetime import datetime, timezone
import pickle
import os
from peewee import *

# Global database object
DB = SqliteDatabase(None)

class Database:
    # db is meant to be a dict
    # contains object User and indexed by a user's user_id
    # Why set it as None? See below:
    # http://effbot.org/zone/default-values.htm
    db = None
    same_url_limit = 1  # Defines how many same urls are allowed, e.g., in cases of group projects this may be 2
    logger = None
    db_filename = None

    def __init__(self, logger=None, db_filename='database.sqlite3'):
        self.db_filename = db_filename
        self.logger = logger
        self.db = DB
        self.db.init(db_filename)
        self.db.connect()
        self.db.create_tables([Database.Users])
        os.chmod(db_filename, 0o666)

    def __del__(self):

        self.db.close()


    class Users(Model):
        """
        The class containing local information obtained through canvas about each user, assignment status, last commit,
        plagiarism status and other important information for ATHINA to function.
        """
        user_id = BigIntegerField(primary_key=True)
        course_id = BigIntegerField(default=0)
        user_fullname = TextField(default="")
        secondary_id = TextField(default="")
        repository_url = TextField(default="")
        url_date = DateTimeField(default=datetime(1, 1, 1, 0, 0))  # When a new url was found
        new_url = BooleanField(default=False)  # Switched when new url is discovered on e-learning site
        commit_date = DateTimeField(default=datetime(1, 1, 1, 0, 0))  # Day of the last commit
        same_url_flag = BooleanField(default=False)  # Is repo url found to be similar with N other students?
        plagiarism_to_grade = BooleanField(default=False)  # Signifies whether a user received a new grade (plagiarism)
        last_plagiarism_check = DateTimeField(default=datetime.now())
        last_graded = DateTimeField(default=datetime(1, 1, 1, 0, 0))
        last_grade = SmallIntegerField(null=True)
        changed_state = BooleanField(default=False)

        class Meta:
            database = DB

```

I have a class in another file that when it is called, it gets the user object as a variable

```
class Tester:
    user_data = None

    def __init__(self, user_data):
        self.user_data = user_data

    def parallel_map(self, user_ids):
        compute_pool = multiprocessing.Pool(processes=self.configuration.processes)
        user_object_results = compute_pool.map(self.process_student_assignment, user_ids)
        return user_object_results

```

Problem is, I get the following error:

`multiprocessing.pool.MaybeEncodingError: Error sending result: '<multiprocessing.pool.ExceptionWithTraceback object at 0x7f3c1b55b198>'. Reason: 'PicklingError("Can't pickle <class 'athina.users.UsersDoesNotExist'>: attribute lookup UsersDoesNotExist on athina.users failed",)'
`

Pickle doesn't like the fact that you've nested your `User` model inside the `class Database`. Consider reorganizing your code. It will work fine to pickle Peewee model classes and instances (though this is really not advisable!), but not in the way you have written your code.

A couple other things to mention:

* `BigIntegerField(primary_key=True)` requires you to manually specify the IDs (assuming you plan to create new rows). For auto-incrementing you would use `BigAutoField()` instead.
* Peewee model classes can already be used like dictionaries. So, `User[1]` would return the user with id=1.

Hi @coleifer , thank you for your reply, I made some changes to the code but it is still not working. I made a test code for you to evaluate:

users.py
```
from datetime import datetime
import os
from peewee import *

# Global database object
DB = SqliteDatabase(None)

class Database:
    db = None
    same_url_limit = 1  # Defines how many same urls are allowed, e.g., in cases of group projects this may be 2
    logger = None
    db_filename = None

    def __init__(self, logger=None, db_filename='database.sqlite3'):
        self.db_filename = db_filename
        self.logger = logger
        self.db = DB
        self.db.init(db_filename)
        self.db.connect()
        self.db.create_tables([Users])
        os.chmod(db_filename, 0o666)

    def __del__(self):

        self.db.close()


class Users(Model):
    """
            The class containing local information obtained through canvas about each user, assignment status, last commit,
            plagiarism status and other important information for ATHINA to function.
            """
    user_id = BigIntegerField(primary_key=True)
    course_id = BigIntegerField(default=0)

    class Meta:
        database = DB
```

tester.py
```
from users import *
import multiprocessing


class Tester:
    user_data = None

    def __init__(self, user_data):
        self.user_data = user_data

    def test(self, test):
        print(test)

    def parallel_map(self, user_ids):
        compute_pool = multiprocessing.Pool(processes=2)
        user_object_results = compute_pool.map(self.test, user_ids)
        return user_object_results

tester = Tester(Database('file.sqlite3'))
tester.parallel_map([1,2,3])
```

I am getting the following error:
```
Traceback (most recent call last):
  File "tester.py", line 20, in <module>
    tester.parallel_map([1,2,3])
  File "tester.py", line 16, in parallel_map
    user_object_results = compute_pool.map(self.test, user_ids)
  File "/usr/lib/python3.5/multiprocessing/pool.py", line 260, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/lib/python3.5/multiprocessing/pool.py", line 608, in get
    raise self._value
  File "/usr/lib/python3.5/multiprocessing/pool.py", line 385, in _handle_tasks
    put(task)
  File "/usr/lib/python3.5/multiprocessing/connection.py", line 206, in send
    self._send_bytes(ForkingPickler.dumps(obj))
  File "/usr/lib/python3.5/multiprocessing/reduction.py", line 50, in dumps
    cls(buf, protocol).dump(obj)
TypeError: can't pickle _thread.lock objects

```

Is that `_thread.lock` coming from peewee? It's not clear to me. Your code organization is kind of fucked.

Look: here is a minimal example with none of this crazy shit, and it works fine:

```python
from multiprocessing import Pool
from peewee import *

db = SqliteDatabase('/tmp/test-mp.db')

class Register(Model):
    value = IntegerField()
    class Meta:
        database = db

with db:
    db.drop_tables([Register])
    db.create_tables([Register])


def create_register(i):
    with db:
        Register.create(value=i)

Pool(processes=4).map(create_register, range(10))

with db:
    for row in Register.select().order_by(Register.id):
        print(row.id, row.value)
```

I suspect that thread_lock comes from peewee. It makes sense that the sqlite connection won't be sharable to other threads. The issue is that I need to have separate modules and separate the db logic with my tester module. The problem is that if the self is passed on the mapped function and that contains the connection then you get that error. There is a work around.

Before calling pool, I close the database (del has an alias that closes it) and I connect to the db from within the function. I get multiple connections to the same file but it would be the same like running several python scripts that do operations on the same sqlite db.

```
from users import *
import multiprocessing


class Tester:
    user_data = None

    def __init__(self, user_data):
        self.user_data = user_data

    def test(self, test):
        self.user_data = Database('file.sqlite3')
        print(Users[1])
        print(test)
        Users[1].course_id = 1
        Users[1].save()
        del self.user_data
        return True

    def parallel_map(self, user_ids):
        compute_pool = multiprocessing.Pool(processes=2)
        user_object_results = compute_pool.map(self.test, user_ids)
        return user_object_results

tester = Tester(Database('file.sqlite3'))
#Users(user_id=1).save(force_insert=True)
Users[1]
del tester.user_data
tester.parallel_map([1, 2, 3])
tester.user_data = Database('file.sqlite3')
```

I *feel* like I'm speaking english and my code was clear and concise.

One more thought: when you fork, any open connections are carried into the new process. Which usually results in bad things happening when you try to use them from the child process. So: **always** close connections before fork. The same applies to pickling connections.

Basically: if you want to do multiple processes, always try to minimize the database state (eg open conns) before fork.