Speed up experimental worker

josegonzalez · josegonzalez · commit b6d10cc67300 · 2013-04-18T15:24:18.000-04:00
- Removed inline sleep call, which slowed down passes n*0.1 seconds, where n is the number of files being tailed
- Inline methods that update data structures which should speed up larger installations
- Make self.active() an attribute lookup instead of a method call
diff --git a/beaver/worker/tail.py b/beaver/worker/tail.py
@@ -20,7 +20,7 @@ class Tail(BaseLog):
     def __init__(self, filename, callback, position="end", logger=None, beaver_config=None, file_config=None):
         super(Tail, self).__init__(logger=logger)
 
-        self._active = False
+        self.active = False
         self._callback = callback
         self._fid = None
         self._file = None
@@ -69,22 +69,18 @@ def open(self, encoding=None):
 
     def close(self):
         """Closes all currently open file pointers"""
-        self._active = False
+        self.active = False
         if self._file:
             self._file.close()
 
-    def active(self):
-        return self._active
-
     def run(self, once=False):
-        while self._active:
-            current_time = int(time.time())
-            self._run_pass(current_time=current_time)
-            self._log_debug('Iteration took {0:.6f}'.format(time.time() - current_time))
-            time.sleep(0.1)
+        while self.active:
+            current_time = time.time()
+            self._run_pass()
 
             self._ensure_file_is_good(current_time=current_time)
 
+            self._log_debug('Iteration took {0:.6f}'.format(time.time() - current_time))
             if once:
                 break
 
@@ -99,7 +95,7 @@ def _ensure_file_is_good(self, current_time):
         if self._last_file_mapping_update and current_time - self._last_file_mapping_update <= self._stat_interval:
             return
 
-        self._last_file_mapping_update = int(time.time())
+        self._last_file_mapping_update = time.time()
 
         try:
             st = os.stat(self._filename)
@@ -128,15 +124,15 @@ def _ensure_file_is_good(self, current_time):
             self._update_file(seek_to_end=False)
             self._file.seek(position, os.SEEK_SET)
 
-    def _run_pass(self, current_time):
+    def _run_pass(self):
         """Read lines from a file and performs a callback against them"""
         line_count = 0
         while True:
             try:
                 lines = self._file.readlines(4096)
             except IOError, e:
                 if e.errno == errno.ESTALE:
-                    self._active = False
+                    self.active = False
                     return False
 
             if not lines:
@@ -321,7 +317,7 @@ def _update_file(self, seek_to_end=True):
         except IOError:
             pass
         else:
-            self._active = True
+            self.active = True
             try:
                 st = os.stat(self._filename)
             except EnvironmentError, err:
diff --git a/beaver/worker/tail_manager.py b/beaver/worker/tail_manager.py
@@ -49,7 +49,7 @@ def watch(self, paths=[]):
 
             self._tails[tail.fid()] = tail
 
-    def run(self):
+    def run(self, interval=0.1,):
         while self._active:
             for fid in self._tails.keys():
                 if not (self._proc and self._proc.is_alive()):
@@ -63,9 +63,11 @@ def run(self):
 
                 self._tails[fid].run(once=True)
 
-                if not self._tails[fid].active():
+                if not self._tails[fid].active:
                     del self._tails[fid]
 
+            time.sleep(interval)
+
     def update_files(self):
         """Ensures all files are properly loaded.
         Detects new files, file removals, file rotation, and truncation.
@@ -80,14 +82,16 @@ def update_files(self):
         possible_files = []
         files = []
         if len(self._beaver_config.get('globs')) > 0:
+            extend_files = files.extend
             for name, exclude in self._beaver_config.get('globs').items():
                 globbed = [os.path.realpath(filename) for filename in eglob(name, exclude)]
-                files.extend(globbed)
                 self._file_config.addglob(name, globbed)
+                extend_files(globbed)
                 self._callback(("addglob", (name, globbed)))
         else:
+            append_files = files.append
             for name in self.listdir():
-                files.append(os.path.realpath(os.path.join(self._folder, name)))
+                append_files(os.path.realpath(os.path.join(self._folder, name)))
 
         for absname in files:
             try:
@@ -98,8 +102,9 @@ def update_files(self):
             else:
                 if not stat.S_ISREG(st.st_mode):
                     continue
+                append_possible_files = possible_files.append
                 fid = self.get_file_id(st)
-                possible_files.append((fid, absname))
+                append_possible_files((fid, absname))
 
         # add new ones
         new_files = [fname for fid, fname in possible_files if fid not in self._tails]
