Can't pickle

Question

I am encountering a picklability problem in this code (also attached below). I have read relevant posts [1] [2] but I can not find usefull correspodences. Could you please give an explanation or solution of this error?

Below these is the part of the code that returns the error:

pickle.PicklingError: Can't pickle <class '__main__.JobQueueManager'>: it's not found as __main__.JobQueueManager

Thanks!

def make_server_manager(port, authkey):

    job_q = Queue.Queue()
    result_q = Queue.Queue()

    class JobQueueManager(SyncManager):
        pass

    JobQueueManager.register('get_job_q', callable=lambda: job_q)
    JobQueueManager.register('get_result_q', callable=lambda: result_q)

    manager = JobQueueManager(address=('', port), authkey=authkey)
    manager.start()
    print 'Server started at port %s' % port
    return manager

^{PS: Python 2.7.7, Win 7}

Answer 1

As best as I can tell, to make this pattern work on Windows, you need to create a picklable queue.Queue. You can do that by creating a child class of Queue that defines __setstate__ and __getstate__, and have it only pickle the pieces of state that we actually need to send between processes, and leave the other stuff (unpicklable internal locks) out.

The other changes we need to make are to move the custom Manager class definitions to the top-level, and to not use lambda functions as the argument to callable. Instead, we use a partial and a top-level function, because that can be pickled. Here's the final code:

import sys
from multiprocessing.managers import SyncManager
from functools import partial
import multiprocessing
from Queue import Queue as _Queue

class Queue(_Queue):
    """ A picklable queue. """   
    def __getstate__(self):
        # Only pickle the state we care about
        return (self.maxsize, self.queue, self.unfinished_tasks)

    def __setstate__(self, state):
        # Re-initialize the object, then overwrite the default state with
        # our pickled state.
        Queue.__init__(self)
        self.maxsize = state[0]
        self.queue = state[1]
        self.unfinished_tasks = state[2]


def get_q(q):
    return q

class JobQueueManager(SyncManager):
    pass


def make_server_manager(port, authkey):
    job_q = Queue()
    result_q = Queue()

    job_q.put("hey")
    JobQueueManager.register('get_job_q', callable=partial(get_q, job_q))
    JobQueueManager.register('get_result_q', callable=partial(get_q, result_q))

    manager = JobQueueManager(address=('', port), authkey=authkey)
    #manager.start()
    print('Server started at port %s' % port)
    return manager

def make_client_manager(port, authkey):
    JobQueueManager.register('get_job_q')
    JobQueueManager.register('get_result_q')
    manager = JobQueueManager(address=('localhost', port), authkey=authkey)
    manager.connect()
    queue = manager.get_job_q()
    print("got queue {}".format(queue))
    print(queue.get_nowait())

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--client":
        make_client_manager(50000, 'abcdefg')
    else:
        manager = make_server_manager(50000, "abcdefg")
        server = manager.get_server()
        server.serve_forever()

Answer 2

You need to have Queue.Queue pickleable, as well as your lambda functions, and your JobQueueManager.

To do that, I think you can be very lazy about it, and all you need to do is to get the dill package and import dill.

I didn't test on windows, but it should work as below. dill is available here: https://github.com/uqfoundation.

>>> import dill
>>> import Queue
>>> from multiprocessing.managers import SyncManager
>>> 
>>> def make_server_manager(port, authkey):
...   job_q = Queue.Queue()
...   result_q = Queue.Queue()
...   class JobQueueManager(SyncManager):
...     pass
...   JobQueueManager.register('get_job_q', callable=lambda: job_q)
...   JobQueueManager.register('get_result_q', callable=lambda: result_q)
...   manager = JobQueueManager(address=('',port), authkey=authkey)
...   manager.start()
...   print "server started at port %s" % port
...   return manager
... 
>>> sm = make_server_manager(12345, 'foofoo')
server started at port 12345

Answer 3

The multiprocessing library gives you a solution out of the box - multiprocessing.Queue which should be automatically picklable everywhere, even on Windows (and works as far back as 2.7).

Trying to make Queue.Queue picklable seems like a bad idea to me. You aren't going to get one queue that you can use from two different processes - you're going to get an independent copy of that queue in the other process.

If you wanted to have a copy of the current state of the queue in another process, it'd be much less work to extract all the elements in the queue as a plain old list which pickles for free (if all the elements are picklable), send the list over, and then reconstitute a new Queue.Queue on the other side.

Also, as I imagine you have discovered by now, you cannot pickle local lambdas - how would that even work? Instead, create a function global to that namespace, and send that global function over, with the required data.

Answer 4

Try:

class JobQueueManager(SyncManager):
        pass

def make_server_manager(port, authkey):

    job_q = Queue.Queue()
    result_q = Queue.Queue()

    JobQueueManager.register('get_job_q', callable=lambda: job_q)
    JobQueueManager.register('get_result_q', callable=lambda: result_q)

    manager = JobQueueManager(address=('', port), authkey=authkey)
    manager.start()
    print 'Server started at port %s' % port
    return manager

Moving the definition of the class to where pickle can find it should allow pickling. Pickle will look in the __main__ module for the class, but with your code, it cannot find it, as it is inside the function. However, as pointed out in the comments, the manager shouldn't need to be pickled, so another object must be dragging it in, such as a function containing the manager in its globals.