Merge pull request #15 from Koed00/dev

Adds cpu affinity to workers

django_q/cluster.py

@@ -22,6 +22,12 @@ from multiprocessing import Queue, Event, Process, Value, current_process
 # external
 import arrow
 
+# optional
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
 # Django
 from django.utils import timezone
 from django.utils.translation import ugettext_lazy as _
@@ -203,10 +209,15 @@ class Sentinel(object):
     def spawn_cluster(self):
         self.pool = []
         Stat(self).save()
+        # spawn worker pool
         for i in range(self.pool_size):
             self.spawn_worker()
+        # spawn auxiliary
         self.monitor = self.spawn_monitor()
         self.pusher = self.spawn_pusher()
+        # set worker cpu affinity if needed
+        if psutil and Conf.CPU_AFFINITY:
+            set_cpu_affinity(Conf.CPU_AFFINITY, [w.pid for w in self.pool])
 
     def guard(self):
         logger.info(_('{} guarding cluster at {}').format(current_process().name, self.pid))
@@ -460,3 +471,35 @@ def scheduler(list_key=Conf.Q_LIST):
             s.repeats = 0
         # save the schedule
         s.save()
+
+
+def set_cpu_affinity(n, process_ids, actual=not Conf.TESTING):
+    """
+    Sets the cpu affinity for the supplied processes.
+    Requires the optional psutil module.
+    :param int n:
+    :param list process_ids: a list of pids
+    :param bool actual: Test workaround for Travis not supporting cpu affinity
+    """
+    # check if we have the psutil module
+    if not psutil:
+        return
+    # get the available processors
+    cpu_list = list(range(psutil.cpu_count()))
+    # affinities of 0 or gte cpu_count, equals to no affinity
+    if not n or n >= len(cpu_list):
+        return
+    # spread the workers over the available processors.
+    index = 0
+    for pid in process_ids:
+        affinity = []
+        for k in range(n):
+            if index == len(cpu_list):
+                index = 0
+            affinity.append(cpu_list[index])
+            index += 1
+        if psutil.pid_exists(pid):
+            p = psutil.Process(pid)
+            if actual:
+                p.cpu_affinity(affinity)
+            logger.info('{} will use cpu {}'.format(pid, affinity))

django_q/conf.py

@@ -45,6 +45,9 @@ class Conf(object):
     # The Django Admin label for this app
     LABEL = conf.get('label', 'Django Q')
 
+    # Sets the number of processors for each worker, defaults to all.
+    CPU_AFFINITY = conf.get('cpu_affinity', 0)
+
     # Use the secret key for package signing
     # Django itself should raise an error if it's not configured
     SECRET_KEY = settings.SECRET_KEY
@@ -64,6 +67,9 @@ class Conf(object):
     STOPPED = _('Stopped')
     STOPPING = _('Stopping')
 
+    # to manage workarounds during testing
+    TESTING = conf.get('testing', False)
+
 
 # logger
 logger = logging.getLogger('django-q')

django_q/monitor.py

@@ -123,8 +123,7 @@ class Stat(Status):
         self.task_q_size = sentinel.task_queue.qsize()
         if sentinel.pusher:
             self.pusher = sentinel.pusher.pid
-        for w in sentinel.pool:
-            self.workers.append(w.pid)
+        self.workers = [w.pid for w in sentinel.pool]
 
     def uptime(self):
         return (timezone.now() - self.tob).total_seconds()

django_q/tests/settings.py

@@ -102,4 +102,6 @@ LOGGING = {
 STATIC_URL = '/static/'
 
 # Django Q specific
-Q_CLUSTER = {'name': 'django_q_test'}
+Q_CLUSTER = {'name': 'django_q_test',
+             'cpu_affinity': 1,
+             'testing': True}

docs/install.rst

@@ -132,6 +132,50 @@ of the cache connection you want to use::
 .. tip::
     Django Q uses your `SECRET_KEY` to encrypt task packages and prevent task crossover. So make sure you have it set up in your Django settings.
 
+cpu_affinity
+~~~~~~~~~~~~
+
+Sets the number of processor each worker can use. This does not affect auxiliary process like the sentinel or monitor and is only useful for tweaking the performance of very high traffic clusters.
+The affinity number has to be higher than zero and less than the total number of processors to have any effect. Defaults to using all processors::
+
+    # processor affinity example.
+
+    4 processors, 4 workers, cpu_affinity: 1
+
+    worker 1 cpu [0]
+    worker 2 cpu [1]
+    worker 3 cpu [2]
+    worker 4 cpu [3]
+
+    4 processors, 4 workers, cpu_affinity: 2
+
+    worker 1 cpu [0, 1]
+    worker 2 cpu [2, 3]
+    worker 3 cpu [0, 1]
+    worker 4 cpu [2, 3]
+
+    8 processors, 8 workers, cpu_affinity: 3
+
+    worker 1 cpu [0, 1, 2]
+    worker 2 cpu [3, 4, 5]
+    worker 3 cpu [6, 7, 0]
+    worker 4 cpu [1, 2, 3]
+    worker 5 cpu [4, 5, 6]
+    worker 6 cpu [7, 0, 1]
+    worker 7 cpu [2, 3, 4]
+    worker 8 cpu [5, 6, 7]
+
+
+In some cases, setting the cpu affinity for your workers can lead to performance improvements, especially if the load is high and consists of many repeating small tasks.
+Start with an affinity of 1 and work your way up. You will have to experiment with what works best for you.
+As a rule of thumb; cpu_affinity 1 favors repetitive short running tasks,while no affinity benefits longer running tasks.
+
+.. note::
+
+    The `cpu_affinity` setting requires the optional `psutil <https://github.com/giampaolo/psutil>`__ module by Giampaolo Rodola'.
+    You can install it with:
+    ``pip install psutil``
+
 Requirements
 ------------
 

requirements.txt

@@ -5,4 +5,6 @@ Django>=1.7.8
 future==0.14.3
 hiredis==0.2.0
 redis==2.10.3
+psutil==3.1.1
+