From dd4f02f1226cdac0f18b50b8d4ef6abfcc0d8be3 Mon Sep 17 00:00:00 2001 From: Ronald van Zon Date: Wed, 13 Dec 2017 14:36:12 +0100 Subject: [PATCH] Add custom SharedCounter / Queue This will not fix anything related to Django 2.0 but will allow MAC users to run Django-Q allowing tests to be run locally. --- django_q/cluster.py | 4 +- django_q/conf.py | 5 ++- django_q/queues.py | 69 ++++++++++++++++++++++++++++++++ django_q/tasks.py | 3 +- django_q/tests/test_cached.py | 3 +- django_q/tests/test_cluster.py | 3 +- django_q/tests/test_scheduler.py | 3 +- 7 files changed, 83 insertions(+), 7 deletions(-) create mode 100644 django_q/queues.py diff --git a/django_q/cluster.py b/django_q/cluster.py index 23db963..4dd61f6 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -10,7 +10,7 @@ import signal import socket import ast from time import sleep -from multiprocessing import Queue, Event, Process, Value, current_process +from multiprocessing import Event, Process, Value, current_process # external import arrow @@ -30,7 +30,7 @@ from django_q.models import Task, Success, Schedule from django_q.status import Stat, Status from django_q.brokers import get_broker from django_q.signals import pre_execute - +from django_q.queues import Queue class Cluster(object): diff --git a/django_q/conf.py b/django_q/conf.py index 6e955a1..e8afac1 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -1,7 +1,7 @@ import logging from copy import deepcopy from signal import signal -from multiprocessing import cpu_count, Queue +from multiprocessing import cpu_count # django from django.utils.translation import ugettext_lazy as _ @@ -11,6 +11,9 @@ from django.conf import settings import os import pkg_resources +# local +from django_q.queues import Queue + # optional try: import psutil diff --git a/django_q/queues.py b/django_q/queues.py new file mode 100644 index 0000000..ad729a1 --- /dev/null +++ b/django_q/queues.py @@ -0,0 +1,69 @@ +""" +The code is derived from https://github.com/althonos/pronto/commit/3384010dfb4fc7c66a219f59276adef3288a886b +""" + +import multiprocessing +import multiprocessing.queues + + +class SharedCounter(object): + """ A synchronized shared counter. + + The locking done by multiprocessing.Value ensures that only a single + process or thread may read or write the in-memory ctypes object. However, + in order to do n += 1, Python performs a read followed by a write, so a + second process may read the old value before the new one is written by + the first process. The solution is to use a multiprocessing.Lock to + guarantee the atomicity of the modifications to Value. + + This class comes almost entirely from Eli Bendersky's blog: + http://eli.thegreenplace.net/2012/01/04/shared-counter-with-pythons-multiprocessing/ + """ + + def __init__(self, n=0): + self.count = multiprocessing.Value('i', n) + + def increment(self, n=1): + """ Increment the counter by n (default = 1) """ + with self.count.get_lock(): + self.count.value += n + + @property + def value(self): + """ Return the value of the counter """ + return self.count.value + + +class Queue(multiprocessing.queues.Queue): + """ A portable implementation of multiprocessing.Queue. + + Because of multithreading / multiprocessing semantics, Queue.qsize() may + raise the NotImplementedError exception on Unix platforms like Mac OS X + where sem_getvalue() is not implemented. This subclass addresses this + problem by using a synchronized shared counter (initialized to zero) and + increasing / decreasing its value every time the put() and get() methods + are called, respectively. This not only prevents NotImplementedError from + being raised, but also allows us to implement a reliable version of both + qsize() and empty(). + """ + + def __init__(self, *args, **kwargs): + super(Queue, self).__init__(*args, ctx=multiprocessing.get_context(), **kwargs) + self.size = SharedCounter(0) + + def put(self, *args, **kwargs): + super(Queue, self).put(*args, **kwargs) + self.size.increment(1) + + def get(self, *args, **kwargs): + x = super(Queue, self).get(*args, **kwargs) + self.size.increment(-1) + return x + + def qsize(self): + """ Reliable implementation of multiprocessing.Queue.qsize() """ + return self.size.value + + def empty(self): + """ Reliable implementation of multiprocessing.Queue.empty() """ + return not self.qsize() > 0 diff --git a/django_q/tasks.py b/django_q/tasks.py index 5f1c46b..e64bfc9 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -1,5 +1,5 @@ """Provides task functionality.""" -from multiprocessing import Queue, Value +from multiprocessing import Value # django from django.db import IntegrityError @@ -14,6 +14,7 @@ from django_q.models import Schedule, Task from django_q.humanhash import uuid from django_q.brokers import get_broker from django_q.signals import pre_enqueue +from django_q.queues import Queue def async(func, *args, **kwargs): diff --git a/django_q/tests/test_cached.py b/django_q/tests/test_cached.py index 1cf9980..e32c60c 100644 --- a/django_q/tests/test_cached.py +++ b/django_q/tests/test_cached.py @@ -1,4 +1,4 @@ -from multiprocessing import Event, Queue, Value +from multiprocessing import Event, Value import pytest @@ -8,6 +8,7 @@ from django_q.conf import Conf from django_q.tasks import async, result, fetch, count_group, result_group, fetch_group, delete_group, delete_cached, \ async_iter, Chain, async_chain, Iter, Async from django_q.brokers import get_broker +from django_q.queues import Queue @pytest.fixture diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index d16d0ac..15461d0 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -1,6 +1,6 @@ import sys import threading -from multiprocessing import Queue, Event, Value +from multiprocessing import Event, Value from time import sleep from django.utils import timezone @@ -19,6 +19,7 @@ from django_q.conf import Conf from django_q.status import Stat from django_q.brokers import get_broker from django_q.tests.tasks import multiply +from django_q.queues import Queue class WordClass(object): diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 47a0860..5160776 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -1,5 +1,5 @@ from datetime import timedelta -from multiprocessing import Queue, Event, Value +from multiprocessing import Event, Value import arrow import pytest @@ -10,6 +10,7 @@ from django_q.brokers import get_broker from django_q.cluster import pusher, worker, monitor, scheduler from django_q.conf import Conf from django_q.tasks import Schedule, fetch, schedule as create_schedule +from django_q.queues import Queue @pytest.fixture