[Erp5-report] r16828 - in /erp5/trunk/products/CMFActivity/Activity: SQLDict.py SQLQueue.py

[nobody at svn.erp5.org]

Author: vincent
Date: Thu Oct  4 17:50:59 2007
New Revision: 16828

URL: http://svn.erp5.org?rev=16828&view=rev
Log:
Save last used processing node in a global variable. This avoids the case where all activities are assigned to first node when there is just one activity to distribute per distribute call.

Modified:
    erp5/trunk/products/CMFActivity/Activity/SQLDict.py
    erp5/trunk/products/CMFActivity/Activity/SQLQueue.py

Modified: erp5/trunk/products/CMFActivity/Activity/SQLDict.py
URL: http://svn.erp5.org/erp5/trunk/products/CMFActivity/Activity/SQLDict.py?rev=16828&r1=16827&r2=16828&view=diff
==============================================================================
--- erp5/trunk/products/CMFActivity/Activity/SQLDict.py (original)
+++ erp5/trunk/products/CMFActivity/Activity/SQLDict.py Thu Oct  4 17:50:59 2007
@@ -54,6 +54,7 @@
   [4] * 5 + \
   [5] * 1
 
+LAST_PROCESSING_NODE = 1
 
 class SQLDict(RAMDict):
   """
@@ -470,6 +471,7 @@
   def distribute(self, activity_tool, node_count):
     readMessageList = getattr(activity_tool, 'SQLDict_readMessageList', None)
     if readMessageList is not None:
+      global LAST_PROCESSING_NODE
       now_date = DateTime()
       result = readMessageList(path=None, method_id=None, processing_node=-1,
                                to_date=now_date, include_processing=0)
@@ -486,7 +488,7 @@
       # XXX probably this below can be optimized by assigning multiple messages at a time.
       path_dict = {}
       assignMessage = activity_tool.SQLDict_assignMessage
-      processing_node = 1
+      processing_node = LAST_PROCESSING_NODE
       id_tool = activity_tool.getPortalObject().portal_ids
       for message in message_dict.itervalues():
         path = '/'.join(message.object_path)
@@ -535,6 +537,7 @@
 
           assignMessage(processing_node=node, uid=[message.uid], broadcast=0)
           get_transaction().commit() # Release locks immediately to allow processing of messages
+      LAST_PROCESSING_NODE = processing_node
 
   # Validation private methods
   def _validate(self, activity_tool, method_id=None, message_uid=None, path=None, tag=None):

Modified: erp5/trunk/products/CMFActivity/Activity/SQLQueue.py
URL: http://svn.erp5.org/erp5/trunk/products/CMFActivity/Activity/SQLQueue.py?rev=16828&r1=16827&r2=16828&view=diff
==============================================================================
--- erp5/trunk/products/CMFActivity/Activity/SQLQueue.py (original)
+++ erp5/trunk/products/CMFActivity/Activity/SQLQueue.py Thu Oct  4 17:50:59 2007
@@ -52,6 +52,8 @@
   [3] * 10 + \
   [4] * 5 + \
   [5] * 1
+
+LAST_PROCESSING_NODE = 1
 
 class SQLQueue(RAMQueue):
   """
@@ -302,6 +304,7 @@
   def distribute(self, activity_tool, node_count):
     readMessageList = getattr(activity_tool, 'SQLQueue_readMessageList', None)
     if readMessageList is not None:
+      global LAST_PROCESSING_NODE
       now_date = DateTime()
       result = readMessageList(path=None, method_id=None,
                                processing_node=-1, to_date=now_date)
@@ -318,7 +321,7 @@
       # XXX probably this below can be optimized by assigning multiple messages at a time.
       path_dict = {}
       assignMessage = activity_tool.SQLQueue_assignMessage
-      processing_node = 1
+      processing_node = LAST_PROCESSING_NODE
       id_tool = activity_tool.getPortalObject().portal_ids
       for message in message_dict.itervalues():
         path = '/'.join(message.object_path)
@@ -358,6 +361,7 @@
 
           assignMessage(processing_node=node, uid=message.uid, broadcast=0)
           get_transaction().commit() # Release locks immediately to allow processing of messages
+      LAST_PROCESSING_NODE = processing_node
 
   # Validation private methods
   def _validate(self, activity_tool, method_id=None, message_uid=None, path=None, tag=None):