queue.enqueue raises "RuntimeError: dictionary changed size during iteration"

[marcelkornblum]

We queue various functions, often passing in Django model instances as parameters, and this error is being raised by many or all of the places we call queue.enqueue.

We started seeing this error in production after upgrading to Django 1.8.2.

We can't see what's changed in Django 1.8 that's causing us to suddenly see this exception, but a solution might be to copy.deepcopy the objects we're about to pickle before calling dumps - or before passing them to rq for processing.

I'm cross-posting this issue on the rq repo as well. I'd be happy to submit a pull request with the copy.deepcopy amendment if that'd help.

Here's part of a sample stack trace:

rq/queue.py in enqueue:

        return self.enqueue_call(func=f, args=args, kwargs=kwargs,
                                 timeout=timeout, result_ttl=result_ttl, ttl=ttl,
                                 description=description, depends_on=depends_on,
                                 job_id=job_id, at_front=at_front)

    def enqueue_job(self, job, at_front=False):

django_rq/queues.py in enqueue_call:

    def enqueue_call(self, *args, **kwargs):
        if self._autocommit:
            return self.original_enqueue_call(*args, **kwargs)
        else:
            thread_queue.add(self, args, kwargs)

django_rq/queues.py in original_enqueue_call:

        return super(DjangoRQ, self).__init__(*args, **kwargs)

    def original_enqueue_call(self, *args, **kwargs):
        return super(DjangoRQ, self).enqueue_call(*args, **kwargs)

    def enqueue_call(self, *args, **kwargs):

rq/queue.py in enqueue_call:

                    except WatchError:
                        continue

        return self.enqueue_job(job, at_front=at_front)

    def enqueue(self, f, *args, **kwargs):

rq/queue.py in enqueue_job:

            if job.timeout is None:
                job.timeout = self.DEFAULT_TIMEOUT
            job.save(pipeline=pipeline)

            pipeline.execute()

rq/job.py in save:

        key = self.key
        connection = pipeline if pipeline is not None else self.connection

        connection.hmset(key, self.to_dict())
        self.cleanup(self.ttl)

rq/job.py in to_dict:

        """Returns a serialization of the current job instance"""
        obj = {}
        obj['created_at'] = utcformat(self.created_at or utcnow())
        obj['data'] = self.data

        if self.origin is not None:

rq/job.py in data:

                self._kwargs = {}

            job_tuple = self._func_name, self._instance, self._args, self._kwargs
            self._data = dumps(job_tuple)
        return self._data

[selwin]

Python's pickling of instance methods are iffy and we recommend against doing that. deepcopy-ing objects before pickling is also inefficient.

If you want to do that, I think the best way forward is to allow the use of custom Queue class like so:

RQ_QUEUES = {
    'default': {
        'URL': 'redis://localhost:6379',
        'QUEUE_CLASS': 'path.to.MyQueue',
    },
}

[lmortimer]

I've run into this bug passing a model instance to the RQ job. An easy work around is to instead pass the Primary Key of the model instance to the job, and line one of the job query for the model.