Changeset 472:f7bf19c86baa

Timestamp:

08/28/08 18:37:04 (4 months ago)

Author:

Brian Warner <warner@allmydata.com>

branch:

default

Message:

incident-gatherer: only fetch one incident at a time. Closes #85.

Files:

• ChangeLog (modified) (1 diff)
• foolscap/logging/gatherer.py (modified) (7 diffs)
• foolscap/test/test_logging.py (modified) (2 diffs)

ChangeLog

@@ -1 +1 @@
-2008-08-28  Brian Warner  <warner@lothar.com>
+
+        * foolscap/logging/gatherer.py (IncidentObserver): only fetch one
+        incident at a time, to limit the size of the sender's outbound
+        queue. This should help close #85.
+        (IncidentGathererService.new_incident): include the classification
+        results in the per-incident log messages
+        * foolscap/test/test_logging.py (IncidentGatherer.test_emit): tests
-
-        * foolscap/logging/log.py (FoolscapLogger.declare_incident):

foolscap/logging/gatherer.py

@@ -8 +8 @@
-from zope.interface import implements
-from twisted.internet import reactor, utils, defer
-
+, log as tw_log
-from twisted.application import service, internet
-import foolscap
@@ -310 +310 @@
-        self.stdout = stdout
-        self.caught_up_d = defer.Deferred()
+        self.incidents_wanted = []
+        self.incident_fetch_outstanding = False
-
-    def connect(self):
@@ -332 +334 @@
-
-    def remote_new_incident(self, name, trigger):
-got
+new
-        # name= should look like "incident-2008-07-29-204211-aspkxoi". We
-        # prevent name= from containing path metacharacters like / or : by
-        # using FilePath later on.
+        self.incidents_wanted.append( (name, trigger) )
+        self.maybe_fetch_incident()
+
+    def maybe_fetch_incident(self):
+        # only fetch one incident at a time, to keep the sender's outbound
+        # memory usage to a reasonable level
+        if self.incident_fetch_outstanding:
+            return
+        if not self.incidents_wanted:
+            return
+        self.incident_fetch_outstanding = True
+        (name, trigger) = self.incidents_wanted.pop(0)
+        print >>self.stdout, "fetching incident", name
-        d = self.publisher.callRemote("get_incident", name)
-        d.addCallback(self._got_incident, name, trigger)
-
-
+
+
+
-    def _got_incident(self, incident, name, trigger):
+        self.incident_fetch_outstanding = False
-        # We always save the incident to a .bz2 file.
-        abs_fn = self.basedir.child(name).path # this prevents evil
@@ -350 +367 @@
-        self.update_latest(name)
-        self.gatherer.new_incident(abs_fn, rel_fn, self.tubid_s, incident)
+        self.maybe_fetch_incident()
-
-    def save_incident(self, filename, incident):
@@ -405 +423 @@
-        self.classifiers.extend(classifiers)
-        self.stdout = stdout
+        self.incidents_received = 0 # for tests
-
-    def add_classifier(self, f):
@@ -461 +480 @@
-    def new_incident(self, abs_fn, rel_fn, tubid_s, incident):
-        stdout = self.stdout or sys.stdout
-
-
+
+
+
-
-    def classify_incident(self, rel_fn, tubid_s, incident):
@@ -479 +499 @@
-            f.write(rel_fn + "\n")
-            f.close()
+        return categories
-
-

foolscap/test/test_logging.py

@@ -1060 +1060 @@
-            ig3 = self.create_incident_gatherer(basedir)
-            ig3.setServiceParent(self.parent)
+            self.ig3 = ig3
-
-            unknowns_fn = os.path.join(ig.basedir, "classified", "unknown")
@@ -1074 +1075 @@
-        # give the call to remote_logport a chance to retire
-        d.addCallback(self.stall, 0.5)
+        d.addCallback(lambda res: self.ig3.disownServiceParent())
+
+        # and if we remove all the stored incidents (and the 'latest'
+        # record), the gatherer will grab everything. This exercises the
+        # only-grab-one-at-a-time code. I verified this manually, by adding a
+        # print to the avoid-duplicate clause of
+        # IncidentObserver.maybe_fetch_incident .
+
+        def _create_ig4(res):
+            ig4 = self.create_incident_gatherer(basedir)
+            for nodeid in os.listdir(os.path.join(ig4.basedir, "incidents")):
+                nodedir = os.path.join(ig4.basedir, "incidents", nodeid)
+                for fn in os.listdir(nodedir):
+                    os.unlink(os.path.join(nodedir, fn))
+                os.rmdir(nodedir)
+            ig4.setServiceParent(self.parent)
+            self.ig4 = ig4
+        d.addCallback(_create_ig4)
+        d.addCallback(lambda res:
+                      self.poll(lambda : self.ig4.incidents_received == 2))
+
+        d.addCallback(self.stall, 0.5)
+
-        return d
-