Made the OCR cache backend used for locking configurable, move ocr locking to queued document from periodic task, added again a random delay fallback in case no cache backend is used

apps/ocr/tasks.py

@@ -1,10 +1,12 @@
 from datetime import timedelta, datetime
 import platform
+from time import sleep
+from random import random
 
 from django.db.models import Q
 from django.utils.translation import ugettext as _
-from django.core.cache.backends.locmem import CacheClass
-
+from django.core.cache import get_cache
+    
 from celery.decorators import task, periodic_task
 from celery.task.control import inspect
 
@@ -16,25 +18,51 @@ from ocr.models import QueueDocument, DocumentQueue
 from ocr.conf.settings import NODE_CONCURRENT_EXECUTION
 from ocr.conf.settings import REPLICATION_DELAY
 from ocr.conf.settings import QUEUE_PROCESSING_INTERVAL
+from ocr.conf.settings import CACHE_URI
 
-LOCK_EXPIRE = 60 * 5  # Lock expires in 5 minutes
-local_cache = CacheClass([], {})
+LOCK_EXPIRE = 60 * 10  # Lock expires in 10 minutes
+# TODO: Tie LOCK_EXPIRATION with hard task timeout
+
+if CACHE_URI:
+    try:
+        cache_backend = get_cache(CACHE_URI)
+    except ImportError:
+        # TODO: display or log error
+        cache_backend = None        
+else:
+    cache_backend = None
+
+
+def random_delay():
+    sleep(random() * (QUEUE_PROCESSING_INTERVAL - 1))
+    return True
+
+
+if cache_backend:
+    acquire_lock = lambda lock_id: cache_backend.add(lock_id, u'true', LOCK_EXPIRE)
+    release_lock = lambda lock_id: cache_backend.delete(lock_id)
+else:
+    acquire_lock = lambda lock_id: True
+    release_lock = lambda lock_id: True
 
 
 @task
 def task_process_queue_document(queue_document_id):
-    queue_document = QueueDocument.objects.get(id=queue_document_id)
-    queue_document.state = QUEUEDOCUMENT_STATE_PROCESSING
-    queue_document.node_name = platform.node()
-    queue_document.result = task_process_queue_document.request.id
-    queue_document.save()
-    try:
-        do_document_ocr(queue_document.document)
-        queue_document.delete()
-    except Exception, e:
-        queue_document.state = QUEUEDOCUMENT_STATE_ERROR
-        queue_document.result = e
+    lock_id = u'%s-lock-%d' % (u'task_process_queue_document', queue_document_id)
+    if acquire_lock(lock_id):
+        queue_document = QueueDocument.objects.get(pk=queue_document_id)
+        queue_document.state = QUEUEDOCUMENT_STATE_PROCESSING
+        queue_document.node_name = platform.node()
+        queue_document.result = task_process_queue_document.request.id
         queue_document.save()
+        try:
+            do_document_ocr(queue_document.document)
+            queue_document.delete()
+        except Exception, e:
+            queue_document.state = QUEUEDOCUMENT_STATE_ERROR
+            queue_document.result = e
+            queue_document.save()
+        release_lock(lock_id)
 
 
 def reset_orphans():
@@ -62,26 +90,24 @@ def reset_orphans():
 
 @periodic_task(run_every=timedelta(seconds=QUEUE_PROCESSING_INTERVAL))
 def task_process_document_queues():
-    lock_id = '%s-lock-%s' % ('task_process_document_queues', platform.node())
-    acquire_lock = lambda: local_cache.add(lock_id, 'true', LOCK_EXPIRE)
-    release_lock = lambda: local_cache.delete(lock_id)
+    if not cache_backend:
+        random_delay()
+        
+    reset_orphans()
+    q_pending = Q(state=QUEUEDOCUMENT_STATE_PENDING)
+    q_delayed = Q(delay=True)
+    q_delay_interval = Q(datetime_submitted__lt=datetime.now() - timedelta(seconds=REPLICATION_DELAY))
+    for document_queue in DocumentQueue.objects.filter(state=DOCUMENTQUEUE_STATE_ACTIVE):
+        current_local_processing_count = QueueDocument.objects.filter(
+            state=QUEUEDOCUMENT_STATE_PROCESSING).filter(
+            node_name=platform.node()).count()
+        if current_local_processing_count < NODE_CONCURRENT_EXECUTION:
+            try:
+                oldest_queued_document_qs = document_queue.queuedocument_set.filter(
+                    (q_pending & ~q_delayed) | (q_pending & q_delayed & q_delay_interval))
 
-    if acquire_lock():
-        reset_orphans()
-        q_pending = Q(state=QUEUEDOCUMENT_STATE_PENDING)
-        q_delayed = Q(delay=True)
-        q_delay_interval = Q(datetime_submitted__lt=datetime.now() - timedelta(seconds=REPLICATION_DELAY))
-        for document_queue in DocumentQueue.objects.filter(state=DOCUMENTQUEUE_STATE_ACTIVE):
-            if QueueDocument.objects.filter(
-                state=QUEUEDOCUMENT_STATE_PROCESSING).filter(
-                node_name=platform.node()).count() < NODE_CONCURRENT_EXECUTION:
-                try:
-                    oldest_queued_document_qs = document_queue.queuedocument_set.filter(
-                        (q_pending & ~q_delayed) | (q_pending & q_delayed & q_delay_interval))
-
-                    if oldest_queued_document_qs:
-                        oldest_queued_document = oldest_queued_document_qs.order_by('datetime_submitted')[0]
-                        task_process_queue_document.delay(oldest_queued_document.id)
-                except Exception, e:
-                    print 'DocumentQueueWatcher exception: %s' % e
-        release_lock()
+                if oldest_queued_document_qs:
+                    oldest_queued_document = oldest_queued_document_qs.order_by('datetime_submitted')[0]
+                    task_process_queue_document.delay(oldest_queued_document.pk)
+            except Exception, e:
+                print 'DocumentQueueWatcher exception: %s' % e