fix callback chaining benchmarking example

Greg Banks · Greg Banks · commit 5381488086e5 · 2014-02-04T09:34:41.000-08:00
if there was an error in any query, callbacks_full_pipeline.py would
hang because "self.event" (or "finished_event" in the docs) would never
get set. this also fixes some minor pep8 issues.
diff --git a/benchmarks/callback_full_pipeline.py b/benchmarks/callback_full_pipeline.py
@@ -1,12 +1,16 @@
-from itertools import count
 import logging
+
+from itertools import count
 from threading import Event
 
 from base import benchmark, BenchmarkThread
 
+
 log = logging.getLogger(__name__)
 
-initial = object()
+
+sentinal = object()
+
 
 class Runner(BenchmarkThread):
 
@@ -16,26 +20,22 @@ def __init__(self, *args, **kwargs):
         self.num_finished = count()
         self.event = Event()
 
-    def handle_error(self, exc):
-        log.error("Error on insert: %r", exc)
-
-    def insert_next(self, previous_result):
-        current_num = self.num_started.next()
-
-        if previous_result is not initial:
-            num = next(self.num_finished)
-            if num >= self.num_queries:
+    def insert_next(self, previous_result=sentinal):
+        if previous_result is not sentinal:
+            if isinstance(previous_result, BaseException):
+                log.error("Error on insert: %r", previous_result)
+            if self.num_finished.next() >= self.num_queries:
                 self.event.set()
 
-        if current_num <= self.num_queries:
+        if self.num_started.next() <= self.num_queries:
             future = self.session.execute_async(self.query, self.values)
-            future.add_callbacks(self.insert_next, self.handle_error)
+            future.add_callbacks(self.insert_next, self.insert_next)
 
     def run(self):
         self.start_profile()
 
-        for i in range(120):
-            self.insert_next(initial)
+        for _ in xrange(min(120, self.num_queries)):
+            self.insert_next()
 
         self.event.wait()
 
diff --git a/docs/performance.rst b/docs/performance.rst
@@ -182,28 +182,26 @@ flight at any time:
     from itertools import count
     from threading import Event
 
+    sentinal = object()
+    num_queries = 100000
     num_started = count()
     num_finished = count()
-    initial = object()
     finished_event = Event()
 
-    def handle_error(exc):
-        log.error("Error on insert: %r", exc)
-
-    def insert_next(previous_result):
-        current_num = num_started.next()
-
-        if previous_result is not initial:
-            num = next(num_finished)
-            if num >= 100000:
+    def insert_next(previous_result=sentinal):
+        if previous_result is not sentinal:
+            if isinstance(previous_result, BaseException):
+                log.error("Error on insert: %r", previous_result)
+            if num_finished.next() >= num_queries:
                 finished_event.set()
 
-        if current_num <= 100000:
+        if num_started.next() <= num_queries:
             future = session.execute_async(query)
-            future.add_callbacks(insert_next, handle_error)
+            # NOTE: this callback also handles errors
+            future.add_callbacks(insert_next, insert_next)
 
-    for i in range(120):
-        insert_next(initial)
+    for i in range(min(120, num_queries)):
+        insert_next()
 
     finished_event.wait()
 
