Larger pending queue depth.

docs/sphinx/configuration.rst

@@ -129,6 +129,11 @@ Defines basic plugin properties.
 
 File extensions just be (.conf|.json).
 
+[pending]
+
+- **depth** - The pending queue depth.  Default: 10K
+
+
 [model]
 -------
 

etc/gofer/plugins/builtin.conf

@@ -14,6 +14,11 @@
 #   forward
 #      Forward to.  A comma (,) separated list of plugin names (,=none|*=all).
 #
+# [pending]
+#
+#   depth
+#      The pending queue depth.  Default: 10K
+#
 # [messaging]
 #
 #   uuid

etc/gofer/plugins/package.conf

@@ -14,6 +14,11 @@
 #   forward
 #      Forward to.  A comma (,) separated list of plugin names (,=none|*=all).
 #
+# [pending]
+#
+#   depth
+#      The pending queue depth.  Default: 10K
+#
 # [messaging]
 #
 #   uuid

etc/gofer/plugins/system.conf

@@ -14,6 +14,11 @@
 #   forward
 #      Forward to.  A comma (,) separated list of plugin names (,=none|*=all).
 #
+# [pending]
+#
+#   depth
+#      The pending queue depth.  Default: 10K
+#
 # [messaging]
 #
 #   uuid

etc/gofer/plugins/virt.conf

@@ -14,6 +14,11 @@
 #   forward
 #      Forward to.  A comma (,) separated list of plugin names (,=none|*=all).
 #
+# [pending]
+#
+#   depth
+#      The pending queue depth.  Default: 10K
+#
 # [messaging]
 #
 #   uuid

src/gofer/agent/config.py

@@ -84,6 +84,11 @@
 #   authenticator
 #      The (optional) fully qualified Authenticator to be loaded from the PYTHON path.
 #
+# [pending]
+#
+#   depth
+#     The pending queue depth.  Default: 10K
+#
 # [model]
 #
 #   managed
@@ -121,6 +126,11 @@
             ('heartbeat', OPTIONAL, NUMBER),
         )
     ),
+    ('pending', OPTIONAL,
+        (
+            ('depth', REQUIRED, NUMBER),
+        )
+    ),
     ('model', OPTIONAL,
         (
             ('managed', OPTIONAL, '(0|1|2)'),
@@ -155,6 +165,9 @@
     'messaging': {
         'heartbeat': '10'
     },
+    'pending': {
+        'depth': '10000'
+    },
     'model': {
         'managed': '2'
     }

src/gofer/agent/plugin.py

@@ -248,7 +248,7 @@ def __init__(self, descriptor, path):
         self.actions = []
         self.dispatcher = Dispatcher()
         self.whiteboard = Whiteboard()
-        self.scheduler = Scheduler(self)
+        self.scheduler = Scheduler(self, int(descriptor.pending.depth))
         self.delegate = Delegate()
         self.authenticator = None
         self.consumer = None

src/gofer/agent/rmi.py

@@ -146,14 +146,16 @@ class Scheduler(Thread):
     Processes the *pending* queue.
     """
 
-    def __init__(self, plugin):
+    def __init__(self, plugin, depth):
         """
         :param plugin: A plugin.
         :type plugin: gofer.agent.plugin.Plugin
+        :param depth: The max queue depth.
+        :type depth: int
         """
         Thread.__init__(self, name='scheduler:%s' % plugin.name)
         self.plugin = plugin
-        self.pending = Pending(plugin.name)
+        self.pending = Pending(plugin.name, depth)
         self.builtin = Builtin(plugin)
         self.setDaemon(True)
 

src/gofer/rmi/store.py

@@ -23,7 +23,9 @@
 from logging import getLogger
 from Queue import Queue, Empty
 
+
 from gofer import NAME, Thread
+from gofer.common import utf8
 from gofer.common import mkdir, rmdir, unlink
 from gofer.messaging import Document
 from gofer.rmi.tracker import Tracker
@@ -32,13 +34,36 @@
 log = getLogger(__name__)
 
 
+class FileCorrupted(Exception):
+    """
+    File corrupted and likely discarded.
+    """
+
+    def __init__(self, path):
+        super(FileCorrupted, self).__init__(path)
+
+    @property
+    def path(self):
+        return self.args[0]
+
+    def __str__(self):
+        return "File %s corrupted.  Discarded" % self.path
+
+
 class Pending(object):
     """
     Persistent store and queuing for pending requests.
     """
 
     PENDING = '/var/lib/%s/messaging/pending' % NAME
 
+    # The queue depth
+    MAX_DEPTH = 10000
+
+    # The soft threshold determines when the journal
+    # file path is queued instead of the actual request
+    SOFT_THRESHOLD = 50
+
     @staticmethod
     def _write(request, path):
         """
@@ -52,7 +77,7 @@ def _write(request, path):
         try:
             body = request.dump()
             fp.write(body)
-            log.debug('wrote [%s]: %s', path, body)
+            log.debug('Writing [%s]: %s', path, body)
         finally:
             fp.close()
 
@@ -64,18 +89,19 @@ def _read(path):
         :type path: str
         :return: The read request.
         :rtype: Document
+        :raise FileCorrupted:
         """
         fp = open(path)
         try:
             try:
                 request = Document()
                 body = fp.read()
                 request.load(body)
-                log.debug('read [%s]: %s', path, body)
+                log.debug('Read [%s]: %s', path, body)
                 return request
             except ValueError:
-                log.error('%s corrupt (discarded)', path)
                 os.unlink(path)
+                raise FileCorrupted(path)
         finally:
             fp.close()
 
@@ -89,13 +115,15 @@ def _list(self):
         paths = [os.path.join(path, name) for name in os.listdir(path)]
         return sorted(paths)
 
-    def __init__(self, stream):
+    def __init__(self, stream, depth=MAX_DEPTH):
         """
         :param stream: The stream name.
         :type stream: str
+        :param depth: The queue depth.
+        :type depth: int
         """
         self.stream = stream
-        self.queue = Queue(maxsize=100)
+        self.queue = Queue(maxsize=depth)
         self.is_open = False
         self.sequential = Sequential()
         self.journal = {}
@@ -114,11 +142,11 @@ def _open(self):
         log.info('Using: %s', path)
         for path in self._list():
             log.info('Restoring: %s', path)
-            request = Pending._read(path)
-            if not request:
-                # read failed
-                continue
-            self._put(request, path)
+            try:
+                request = Pending._read(path)
+                self._put(request, path)
+            except FileCorrupted, fe:
+                log.debug(utf8(fe))
         self.is_open = True
 
     def put(self, request):
@@ -145,7 +173,7 @@ def get(self):
         """
         while not Thread.aborted():
             try:
-                return self.queue.get(timeout=10)
+                return self._get(timeout=10)
             except Empty:
                 pass
 
@@ -159,9 +187,9 @@ def commit(self, sn):
         try:
             path = self.journal[sn]
             unlink(path)
-            log.debug('%s committed', sn)
+            log.debug('Request %s committed', sn)
         except KeyError:
-            log.warn('%s not found for commit', sn)
+            log.warn('Request %s not found for commit', sn)
 
     def delete(self):
         """
@@ -173,7 +201,7 @@ def delete(self):
         self._drain()
         path = os.path.join(Pending.PENDING, self.stream)
         rmdir(path)
-        log.info('%s, deleted', path)
+        log.info('File %s deleted', path)
 
     def _drain(self):
         """
@@ -182,24 +210,55 @@ def _drain(self):
         self.is_open = False
         while not Thread.aborted():
             try:
-                request = self.queue.get(timeout=1)
+                request = self._get(timeout=1)
                 self.commit(request.sn)
             except Empty:
                 break
 
+    def _get(self, timeout=10):
+        """
+        Get the next pending request to be dispatched.
+        The queued *thing* can be either a request or the path to
+        a journal file.  When a path is detected, the file is read
+        and the actual request is read.
+        :return: The next pending request.
+        :rtype: Document
+        :raise Empty: when queue empty.
+        """
+        while True:
+            if Thread.aborted():
+                raise Empty()
+            thing = self.queue.get(timeout=timeout)
+            if isinstance(thing, str):
+                try:
+                    request = Pending._read(thing)
+                except (IOError, OSError, FileCorrupted), fe:
+                    log.warn(utf8(fe))
+                    continue
+            else:
+                request = thing
+            request.ts = time()
+            return request
+
     def _put(self, request, jnl_path):
         """
         Enqueue the request.
+        When the queue depth threshold is exceeded, the path to the
+        journal file is queued instead of the actual request. This
+        is done to reduce memory footprint for long backlogs.
         :param request: An AMQP request.
         :type request: Document
         :param jnl_path: Path to the associated journal file.
         :type jnl_path: str
         """
-        request.ts = time()
         tracker = Tracker()
         tracker.add(request.sn, request.data)
         self.journal[request.sn] = jnl_path
-        self.queue.put(request)
+        if self.queue.qsize() > Pending.SOFT_THRESHOLD:
+            thing = jnl_path
+        else:
+            thing = request
+        self.queue.put(thing)
 
 
 class Sequential(object):

test/functional/plugins/testplugin.conf

@@ -27,6 +27,11 @@
 #   host_validation
 #      The (optional) flag indicates SSL host validation should be performed.
 #
+# [pending]
+#
+#   depth
+#      The pending queue depth.  Default: 100K
+#
 # [model]
 #
 #   managed
@@ -47,6 +52,9 @@
 enabled=1
 forward=*
 
+[pending]
+depth=500
+
 [messaging]
 uuid=TEST
 url=