SQLDict.py 18.7 KB
Newer Older
Jean-Paul Smets's avatar
Jean-Paul Smets committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28
##############################################################################
#
# Copyright (c) 2002 Nexedi SARL and Contributors. All Rights Reserved.
#                    Jean-Paul Smets-Solanes <jp@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
#
##############################################################################

29
import random
30
from DateTime import DateTime
Jean-Paul Smets's avatar
Jean-Paul Smets committed
31
from Products.CMFActivity.ActivityTool import registerActivity
32
from Queue import VALID, INVALID_ORDER, INVALID_PATH, EXCEPTION, MAX_PROCESSING_TIME, VALIDATION_ERROR_DELAY, SECONDS_IN_DAY
Jean-Paul Smets's avatar
Jean-Paul Smets committed
33
from RAMDict import RAMDict
34
from Products.CMFActivity.ActiveObject import DISTRIBUTABLE_STATE, INVOKE_ERROR_STATE, VALIDATE_ERROR_STATE
Jean-Paul Smets's avatar
Jean-Paul Smets committed
35 36 37

from zLOG import LOG

38
MAX_PRIORITY = 5
Jean-Paul Smets's avatar
Jean-Paul Smets committed
39

40 41 42 43 44 45
priority_weight = \
  [1] * 64 + \
  [2] * 20 + \
  [3] * 10 + \
  [4] * 5 + \
  [5] * 1
46

47 48 49
class ActivityFlushError(Exception):
    """Error during active message flush"""

Jean-Paul Smets's avatar
Jean-Paul Smets committed
50 51 52 53 54 55
class SQLDict(RAMDict):
  """
    A simple OOBTree based queue. It should be compatible with transactions
    and provide sequentiality. Should not create conflict
    because use of OOBTree.
  """
Jean-Paul Smets's avatar
Jean-Paul Smets committed
56
  # Transaction commit methods
57
  def prepareQueueMessage(self, activity_tool, m):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
58
    if m.is_registered:
59
      activity_tool.SQLDict_writeMessage( path = '/'.join(m.object_path) ,
Jean-Paul Smets's avatar
Jean-Paul Smets committed
60 61
                                          method_id = m.method_id,
                                          priority = m.activity_kw.get('priority', 1),
62
                                          broadcast = m.activity_kw.get('broadcast', 0),
63 64
                                          message = self.dumpMessage(m),
                                          date = m.activity_kw.get('at_date', DateTime()))
Jean-Paul Smets's avatar
Jean-Paul Smets committed
65
                                          # Also store uid of activity
Jean-Paul Smets's avatar
Jean-Paul Smets committed
66

67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
  def prepareQueueMessageList(self, activity_tool, message_list):
    registered_message_list = []
    for message in message_list:
      if message.is_registered:
        registered_message_list.append(message)
    if len(registered_message_list) > 0:
      path_list = ['/'.join(message.object_path) for message in registered_message_list]
      method_id_list = [message.method_id for message in registered_message_list]
      priority_list = [message.activity_kw.get('priority', 1) for message in registered_message_list]
      broadcast_list = [message.activity_kw.get('broadcast', 0) for message in registered_message_list]
      dumped_message_list = [self.dumpMessage(message) for message in registered_message_list]
      date_list = [DateTime()] * len(registered_message_list)
      activity_tool.SQLDict_writeMessageList( path_list = path_list,
                                              method_id_list = method_id_list,
                                              priority_list = priority_list,
                                              broadcast_list = broadcast_list,
                                              message_list = dumped_message_list,
                                              date_list = date_list)
                                                         
86 87
  def prepareDeleteMessage(self, activity_tool, m):
    # Erase all messages in a single transaction
Jean-Paul Smets's avatar
Jean-Paul Smets committed
88 89
    path = '/'.join(m.object_path)
    uid_list = activity_tool.SQLDict_readUidList(path=path, method_id=m.method_id,processing_node=None)
90
    uid_list = map(lambda x:x.uid, uid_list)
91
    if len(uid_list)>0:
92 93 94
      activity_tool.SQLDict_delMessage(uid = uid_list)

  # Registration management
Jean-Paul Smets's avatar
Jean-Paul Smets committed
95
  def registerActivityBuffer(self, activity_buffer):
96
    class_name = self.__class__.__name__
97 98 99
    setattr(activity_buffer, '_%s_uid_dict' % class_name, {})
    setattr(activity_buffer, '_%s_message_list' % class_name, [])

Jean-Paul Smets's avatar
Jean-Paul Smets committed
100
  def isMessageRegistered(self, activity_buffer, activity_tool, m):
101 102 103
    class_name = self.__class__.__name__
    uid_dict = getattr(activity_buffer,'_%s_uid_dict' % class_name)
    return uid_dict.has_key((tuple(m.object_path), m.method_id))
104

Jean-Paul Smets's avatar
Jean-Paul Smets committed
105 106
  def registerMessage(self, activity_buffer, activity_tool, m):
    m.is_registered = 1
107 108 109 110
    class_name = self.__class__.__name__
    uid_dict = getattr(activity_buffer,'_%s_uid_dict' % class_name)
    uid_dict[(tuple(m.object_path), m.method_id)] = 1
    getattr(activity_buffer,'_%s_message_list' % class_name).append(m)
111

Jean-Paul Smets's avatar
Jean-Paul Smets committed
112 113
  def unregisterMessage(self, activity_buffer, activity_tool, m):
    m.is_registered = 0 # This prevents from inserting deleted messages into the queue
114 115 116 117
    class_name = self.__class__.__name__
    uid_dict = getattr(activity_buffer,'_%s_uid_dict' % class_name)
    if uid_dict.has_key((tuple(m.object_path), m.method_id)):
      del uid_dict[(tuple(m.object_path), m.method_id)]
Jean-Paul Smets's avatar
Jean-Paul Smets committed
118 119

  def getRegisteredMessageList(self, activity_buffer, activity_tool):
120 121 122 123
    class_name = self.__class__.__name__
    if hasattr(activity_buffer,'_%s_message_list' % class_name):
      message_list = getattr(activity_buffer,'_%s_message_list' % class_name)
      return filter(lambda m: m.is_registered, message_list)
124 125
    else:
      return ()
126

Jean-Paul Smets's avatar
Jean-Paul Smets committed
127
  # Queue semantic
Jean-Paul Smets's avatar
Jean-Paul Smets committed
128
  def dequeueMessage(self, activity_tool, processing_node):
129 130 131 132
    if hasattr(activity_tool,'SQLDict_readMessage'):
      now_date = DateTime()
      # Next processing date in case of error
      next_processing_date = now_date + VALIDATION_ERROR_DELAY
133
      priority = random.choice(priority_weight)
134
      # Try to find a message at given priority level which is scheduled for now
135
      result = activity_tool.SQLDict_readMessage(processing_node=processing_node, priority=priority,
136
                                                 to_date=now_date)
137
      if len(result) == 0:
138
        # If empty, take any message which is scheduled for now
139
        priority = None
140
        result = activity_tool.SQLDict_readMessage(processing_node=processing_node, priority=priority, to_date=now_date)
141 142 143 144
      if len(result) > 0:
        line = result[0]
        path = line.path
        method_id = line.method_id
145
        uid_list = activity_tool.SQLDict_readUidList( path=path, method_id= method_id, processing_node = None, to_date=now_date )
146 147 148
        uid_list = map(lambda x:x.uid, uid_list)
        # Make sure message can not be processed anylonger
        if len(uid_list) > 0:
149
          # Set selected messages to processing
150 151 152 153
          activity_tool.SQLDict_processMessage(uid = uid_list)
        get_transaction().commit() # Release locks before starting a potentially long calculation
        # This may lead (1 for 1,000,000 in case of reindexing) to messages left in processing state
        m = self.loadMessage(line.message, uid = line.uid)
154
        # Validate message (make sure object exists, priority OK, etc.)
155 156
        validation_state = m.validate(self, activity_tool)
        if validation_state is not VALID:
157
          if validation_state in (EXCEPTION, INVALID_PATH):
158
            # There is a serious validation error - we must lower priority
159 160
            if line.priority > MAX_PRIORITY:
              # This is an error
161
              if len(uid_list) > 0:
162 163 164 165 166 167 168 169 170 171
                activity_tool.SQLDict_assignMessage(uid = uid_list, processing_node = VALIDATE_ERROR_STATE)
                                                                                # Assign message back to 'error' state
              #m.notifyUser(activity_tool)                                      # Notify Error
              get_transaction().commit()                                        # and commit
            else:
              # Lower priority
              if len(uid_list) > 0: # Add some delay before new processing
                activity_tool.SQLDict_setPriority(uid = uid_list, date = next_processing_date,
                                                  priority = line.priority + 1)
              get_transaction().commit() # Release locks before starting a potentially long calculation
172
          else:
173 174 175 176
            # We do not lower priority for INVALID_ORDER errors but we do postpone execution
            if len(uid_list) > 0: # Add some delay before new processing
              activity_tool.SQLDict_setPriority(uid = uid_list, date = next_processing_date,
                                                priority = line.priority)
177
            get_transaction().commit() # Release locks before starting a potentially long calculation
178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201
        else:
          # Try to invoke
          activity_tool.invoke(m) # Try to invoke the message - what happens if read conflict error restarts transaction ?
          if m.is_executed:                                          # Make sure message could be invoked
            if len(uid_list) > 0:
              activity_tool.SQLDict_delMessage(uid = uid_list)                # Delete it
            get_transaction().commit()                                        # If successful, commit
            if m.active_process:
              active_process = activity_tool.unrestrictedTraverse(m.active_process)
              if not active_process.hasActivity():
                # Not more activity
                m.notifyUser(activity_tool, message="Process Finished") # XXX commit bas ???
          else:
            get_transaction().abort()                                         # If not, abort transaction and start a new one
            if line.priority > MAX_PRIORITY:
              # This is an error
              if len(uid_list) > 0:
                activity_tool.SQLDict_assignMessage(uid = uid_list, processing_node = INVOKE_ERROR_STATE)
                                                                                # Assign message back to 'error' state
              m.notifyUser(activity_tool)                                       # Notify Error
              get_transaction().commit()                                        # and commit
            else:
              # Lower priority
              if len(uid_list) > 0:
202
                activity_tool.SQLDict_setPriority(uid = uid_list, date = next_processing_date,
203 204 205 206
                                                  priority = line.priority + 1)
              get_transaction().commit() # Release locks before starting a potentially long calculation
        return 0
      get_transaction().commit() # Release locks before starting a potentially long calculation
Jean-Paul Smets's avatar
Jean-Paul Smets committed
207 208
    return 1

209
  def hasActivity(self, activity_tool, object, **kw):
210 211 212 213 214 215 216 217
    if hasattr(activity_tool,'SQLDict_readMessageList'):
      if object is not None:
        my_object_path = '/'.join(object.getPhysicalPath())
        result = activity_tool.SQLDict_hasMessage(path=my_object_path, **kw)
        if len(result) > 0:
          return result[0].message_count > 0
      else:
        return 1 # Default behaviour if no object specified is to return 1 until active_process implemented
Jean-Paul Smets's avatar
Jean-Paul Smets committed
218 219
    return 0

Jean-Paul Smets's avatar
Jean-Paul Smets committed
220
  def flush(self, activity_tool, object_path, invoke=0, method_id=None, commit=0, **kw):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
221 222
    """
      object_path is a tuple
Jean-Paul Smets's avatar
Jean-Paul Smets committed
223 224 225 226 227 228

      commit allows to choose mode
        - if we commit, then we make sure no locks are taken for too long
        - if we do not commit, then we can use flush in a larger transaction

      commit should in general not be used
229 230

      NOTE: commiting is very likely nonsenses here. We should just avoid to flush as much as possible
Jean-Paul Smets's avatar
Jean-Paul Smets committed
231 232
    """
    path = '/'.join(object_path)
233
    # LOG('Flush', 0, str((path, invoke, method_id)))
234
    method_dict = {}
235 236 237 238 239
    if hasattr(activity_tool,'SQLDict_readMessageList'):
      # Parse each message in registered
      for m in activity_tool.getRegisteredMessageList(self):
        if list(m.object_path) == list(object_path) and (method_id is None or method_id == m.method_id):
          activity_tool.unregisterMessage(self, m)
240
          #if not method_dict.has_key(method_id or m.method_id):
241 242
          if not method_dict.has_key(m.method_id):
            method_dict[m.method_id] = 1 # Prevents calling invoke twice
243 244
            if invoke:
              # First Validate
245 246
              validate_value = m.validate(self, activity_tool)
              if validate_value is VALID:
247 248 249 250
                activity_tool.invoke(m) # Try to invoke the message - what happens if invoke calls flushActivity ??
                if not m.is_executed:                                                 # Make sure message could be invoked
                  # The message no longer exists
                  raise ActivityFlushError, (
251
                      'Could not evaluate %s on %s' % (m.method_id , path))
252
              elif validate_value is INVALID_PATH:
253 254
                # The message no longer exists
                raise ActivityFlushError, (
255
                    'The document %s does not exist' % path)
256 257 258 259 260
      # Parse each message in SQL dict
      result = activity_tool.SQLDict_readMessageList(path=path, method_id=method_id,processing_node=None)
      for line in result:
        path = line.path
        method_id = line.method_id
Jean-Paul Smets's avatar
Jean-Paul Smets committed
261
        if not method_dict.has_key(method_id):
262
          # Only invoke once (it would be different for a queue)
263 264
          # This is optimisation with the goal to process objects on the same
          # node and minimize network traffic with ZEO server
265 266 267
          method_dict[method_id] = 1
          m = self.loadMessage(line.message, uid = line.uid)
          self.deleteMessage(activity_tool, m)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
268 269
          if invoke:
            # First Validate
270
            validate_value = m.validate(self, activity_tool)
Romain Courteaud's avatar
Romain Courteaud committed
271
#             LOG('SQLDict.flush validate_value',0,validate_value)
272
            if validate_value is VALID:
Jean-Paul Smets's avatar
Jean-Paul Smets committed
273
              activity_tool.invoke(m) # Try to invoke the message - what happens if invoke calls flushActivity ??
Romain Courteaud's avatar
Romain Courteaud committed
274
#               LOG('SQLDict.flush m.is_executed',0,m.is_executed)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
275 276 277
              if not m.is_executed:                                                 # Make sure message could be invoked
                # The message no longer exists
                raise ActivityFlushError, (
278
                    'Could not evaluate %s on %s' % (m.method_id , path))
279
            if validate_value is INVALID_PATH:
Jean-Paul Smets's avatar
Jean-Paul Smets committed
280 281
              # The message no longer exists
              raise ActivityFlushError, (
282
                  'The document %s does not exist' % path)
Jean-Paul Smets's avatar
Jean-Paul Smets committed
283

Jean-Paul Smets's avatar
Jean-Paul Smets committed
284
  def getMessageList(self, activity_tool, processing_node=None):
Jean-Paul Smets's avatar
Jean-Paul Smets committed
285
    # YO: reading all lines might cause a deadlock
Jean-Paul Smets's avatar
Jean-Paul Smets committed
286
    message_list = []
287
    if hasattr(activity_tool,'SQLDict_readMessageList'):
288
      result = activity_tool.SQLDict_readMessageList(path=None, method_id=None, processing_node=None, to_processing_date=None)
289 290 291 292 293
      for line in result:
        m = self.loadMessage(line.message, uid = line.uid)
        m.processing_node = line.processing_node
        m.priority = line.priority
        message_list.append(m)
294 295
    return message_list

Jean-Paul Smets's avatar
Jean-Paul Smets committed
296 297
  def distribute(self, activity_tool, node_count):
    processing_node = 1
298
    if hasattr(activity_tool,'SQLDict_readMessageList'):
299 300 301 302 303 304
      now_date = DateTime()
      if (now_date - self.max_processing_date) > MAX_PROCESSING_TIME:
        # Sticky processing messages should be set back to non processing
        max_processing_date = now_date - MAX_PROCESSING_TIME
        self.max_processing_date = now_date
      else:
305
        max_processing_date = None
306 307
      result = activity_tool.SQLDict_readMessageList(path=None, method_id=None, processing_node = -1,
                                                     to_processing_date = max_processing_date) # Only assign non assigned messages
308 309 310 311
      get_transaction().commit() # Release locks before starting a potentially long calculation
      path_dict = {}
      for line in result:
        path = line.path
312 313 314 315 316
        broadcast = line.broadcast
        if broadcast:
          # Broadcast messages must be distributed into all nodes.
          uid = line.uid
          activity_tool.SQLDict_assignMessage(processing_node=1, uid=[uid])
Yoshinori Okuji's avatar
Yoshinori Okuji committed
317 318 319 320 321 322 323 324 325
          if node_count > 1:
            for node in range(2, node_count+1):
              activity_tool.SQLDict_writeMessage( path = path,
                                                  method_id = line.method_id,
                                                  priority = line.priority,
                                                  broadcast = 1,
                                                  processing_node = node,
                                                  message = line.message,
                                                  date = line.date)
326
        elif not path_dict.has_key(path):
327 328
          # Only assign once (it would be different for a queue)
          path_dict[path] = 1
329
          activity_tool.SQLDict_assignMessage(path=path, processing_node=processing_node, uid=None, broadcast=0)
330 331 332 333
          get_transaction().commit() # Release locks immediately to allow processing of messages
          processing_node = processing_node + 1
          if processing_node > node_count:
            processing_node = 1 # Round robin
Jean-Paul Smets's avatar
Jean-Paul Smets committed
334

335 336 337
  # Validation private methods
  def _validate_after_method_id(self, activity_tool, message, value):
    # Count number of occurances of method_id
338 339
    if type(value) == type(''):
      value = [value]
340
    result = activity_tool.SQLDict_validateMessageList(method_id=value, message_uid=None, path=None)
Romain Courteaud's avatar
Romain Courteaud committed
341 342
#     LOG('SQLDict._validate_after_method_id, method_id',0,value)
#     LOG('SQLDict._validate_after_method_id, result[0].uid_count',0,result[0].uid_count)
343 344
    if result[0].uid_count > 0:
      return INVALID_ORDER
345
    return VALID
346

347 348 349 350 351 352
  def _validate_after_path(self, activity_tool, message, value):
    # Count number of occurances of path
    result = activity_tool.SQLDict_validateMessageList(method_id=None, message_uid=None, path=value)
    if result[0].uid_count > 0:
      return INVALID_ORDER
    return VALID
353

354 355 356 357 358 359 360
  def _validate_after_message_uid(self, activity_tool, message, value):
    # Count number of occurances of message_uid
    result = activity_tool.SQLDict_validateMessageList(method_id=None, message_uid=value, path=None)
    if result[0].uid_count > 0:
      return INVALID_ORDER
    return VALID

361 362
  # Required for tests (time shift)
  def timeShift(self, activity_tool, delay):
363 364 365 366
    """
      To simulate timeShift, we simply substract delay from
      all dates in SQLDict message table
    """
367
    activity_tool.SQLDict_timeShift(delay = delay * SECONDS_IN_DAY)
368

Jean-Paul Smets's avatar
Jean-Paul Smets committed
369
registerActivity(SQLDict)