ERP5Type.Core.Folder: Split _count on large containers.

Divide conflict hot-spot to improve write performance: now, conflict risk
will not be proportional to the number of zope processes, but only to the
number of threads within considered process (...because there is no
stable thread name, if there is one day and conflict splitting is needed,
it will be easy to implement: just concatenate that name to node name in
FragmentedLength._map).
Automatically migrate to FragmentedCount on containers larger than 1k
objects (this threshold may evolve).

product/ERP5Type/Core/Folder.py

@@ -33,11 +33,13 @@ from functools import wraps
 from AccessControl import ClassSecurityInfo, getSecurityManager
 from AccessControl.ZopeGuards import NullIter, guarded_getattr
 from Acquisition import aq_base, aq_parent, aq_inner
+from BTrees.Length import Length
 from OFS.Folder import Folder as OFSFolder
 from OFS.ObjectManager import ObjectManager, checkValidId
 from zExceptions import BadRequest
 from OFS.History import Historical
 import ExtensionClass
+from Persistence import Persistent
 from Products.CMFCore.exceptions import AccessControl_Unauthorized
 from Products.CMFCore.CMFCatalogAware import CMFCatalogAware
 from Products.CMFCore.PortalFolder import ContentFilter
@@ -75,6 +77,7 @@ from zLOG import LOG, WARNING
 import warnings
 from urlparse import urlparse
 from Products.ERP5Type.Message import translateString
+from ZODB.POSException import ConflictError
 
 # Dummy Functions for update / upgrade
 def dummyFilter(object,REQUEST=None):
@@ -98,6 +101,57 @@ class ExceptionRaised(object):
         raise
     return wraps(func)(wrapper)
 
+# Above this many subobjects, migrate _count from Length to FragmentedLength
+# to accomodate concurrent accesses.
+FRAGMENTED_LENGTH_THRESHOLD = 1000
+class FragmentedLength(Persistent):
+  """
+  Drop-in replacement for BTrees.Length, which splits storage by zope node.
+  The intent is that per-node conflicts should be roughly constant, but adding
+  more nodes should not increase overall conflict rate.
+
+  Inherit from Persistent in order to be able to resolve our own conflicts
+  (first time a node touches an instance of this class), which should be a rare
+  event per-instance.
+  Contain BTrees.Length instances for intra-node conflict resolution
+  (inter-threads).
+  """
+  def __init__(self, legacy=None):
+    self._map = {}
+    if legacy is not None:
+      # Key does not matter as long as it is independent from the node
+      # constructing this instance.
+      self._map[None] = legacy
+
+  def set(self, new):
+    self._map.clear()
+    self.change(new)
+
+  def change(self, delta):
+    try:
+      self._map[getCurrentNode()].change(delta)
+    except KeyError:
+      self._map[getCurrentNode()] = Length(delta)
+      # _map is mutable, notify persistence that we have to be serialised.
+      self._p_changed = 1
+
+  def __call__(self):
+    return sum(x() for x in self._map.values())
+
+  @staticmethod
+  def _p_resolveConflict(old_state, current_state, my_state):
+    # Minimal implementation for sanity: only handle addition of one by "me" as
+    # long as current_state does not contain the same key. Anything else is a
+    # conflict.
+    try:
+      my_added_key, = set(my_state['_map']).difference(old_state['_map'])
+    except ValueError:
+      raise ConflictError
+    if my_added_key in current_state:
+      raise ConflictError
+    current_state['_map'][my_added_key] = my_state['_map'][my_added_key]
+    return current_state
+
 class FolderMixIn(ExtensionClass.Base):
   """A mixin class for folder operations, add content, delete content etc.
   """
@@ -663,6 +717,25 @@ class Folder(OFSFolder2, CMFBTreeFolder, CMFHBTreeFolder, Base, FolderMixIn):
   def __init__(self, id):
     self.id = id
 
+  @property
+  def _count(self):
+    count = self.__dict__.get('_count')
+    if isinstance(count, Length) and count() > FRAGMENTED_LENGTH_THRESHOLD:
+      count = self._count = FragmentedLength(count)
+    return count
+
+  @_count.setter
+  def _count(self, value):
+    if isinstance(value, Length) and value() > FRAGMENTED_LENGTH_THRESHOLD:
+      value = FragmentedLength(value)
+    self.__dict__['_count'] = value
+    self._p_changed = 1
+
+  @_count.deleter
+  def _count(self):
+    del self.__dict__['_count']
+    self._p_changed = 1
+
   security.declarePublic('newContent')
   def newContent(self, *args, **kw):
     """ Create a new content """

product/ERP5Type/tests/testFolder.py

@@ -28,11 +28,13 @@
 
 import unittest
 
+from BTrees.Length import Length
 from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
 from Products.ERP5Type.tests.utils import LogInterceptor
 from Products.ERP5Type.tests.utils import createZODBPythonScript
 from Products.ERP5Type.ERP5Type import ERP5TypeInformation
 from Products.ERP5Type.Cache import clearCache
+from Products.ERP5Type.Core.Folder import FragmentedLength, FRAGMENTED_LENGTH_THRESHOLD
 from AccessControl.ZopeGuards import guarded_getattr
 from zExceptions import Unauthorized
 
@@ -266,6 +268,34 @@ class TestFolder(ERP5TypeTestCase, LogInterceptor):
       self.assertTrue(obj.getId() in self.folder.objectIds())
       self.assertEqual(302, response.getStatus())
 
+    def test_fragmentedLength(self):
+      """Test Folder._count type and behaviour"""
+      type_list = ['Folder']
+      self._setAllowedContentTypesForFolderType(type_list)
+      folder = self.folder
+      folder_dict = folder.__dict__
+      folder.newContent(portal_type='Folder')
+      self.assertEqual(len(folder), 1)
+      self.assertIsInstance(folder_dict['_count'], Length)
+      original_length_oid = folder_dict['_count']._p_oid
+      for _ in xrange(FRAGMENTED_LENGTH_THRESHOLD - len(folder) - 1):
+        folder.newContent(portal_type='Folder')
+      self.assertEqual(len(folder), FRAGMENTED_LENGTH_THRESHOLD - 1)
+      self.assertIsInstance(folder_dict['_count'], Length)
+      # Generate 3 to completely clear the threshold, as we do not care whether
+      # the change happens when reaching the threshold or when going over it.
+      folder.newContent(portal_type='Folder')
+      folder.newContent(portal_type='Folder')
+      folder.newContent(portal_type='Folder')
+      self.assertEqual(len(folder), FRAGMENTED_LENGTH_THRESHOLD + 2)
+      fragmented_length = folder_dict['_count']
+      self.assertIsInstance(fragmented_length, FragmentedLength)
+      self.assertEqual(len(fragmented_length._map), 2, fragmented_length._map)
+      original_length = fragmented_length._map[None]
+      self.assertEqual(original_length_oid, original_length._p_oid)
+      self.assertGreater(original_length(), FRAGMENTED_LENGTH_THRESHOLD - 1)
+      self.assertGreater(len(folder), original_length())
+
 def test_suite():
   suite = unittest.TestSuite()
   suite.addTest(unittest.makeSuite(TestFolder))