Merge pull request #82 from meejah/issue81-negative-call-later-time

Issue81 negative call_later time

docs/releases.rst

@@ -1,6 +1,12 @@
 txio releases
 =============
 
+2.6.0
+-----
+
+- avoid giving negative times to `callLater` with batched timers (issue #81)
+
+
 2.5.2
 -----
 

test/test_batched_timers_tx.py

@@ -192,3 +192,24 @@ def test_batched_chunks_with_errors(framework_tx):
             assert False, "Should get exception"
         except RuntimeError as e:
             assert "processing call_later" in str(e)
+
+
+def test_batched_close_to_now(framework_tx):
+    '''
+    if our current time is fractional, and we make a call_later with a
+    tiny delay that's still within the same second, we'll produce a
+    negative call_later when adding a bucket; see issue #81
+    '''
+    from twisted.internet.task import Clock
+
+    class FakeClock(Clock):
+        def callLater(self, delay, *args, **kw):
+            # 'real' reactors do this, but Clock doesn't assert on
+            # this.
+            assert delay >= 0
+            return Clock.callLater(self, delay, *args, **kw)
+
+    with replace_loop(FakeClock()) as clock:
+        clock.advance(0.5)
+        batched = txaio.make_batched_timer(1, chunk_size=2)
+        batched.call_later(0.1, lambda: None)

txaio/_common.py

@@ -39,6 +39,10 @@ class _BatchedTimer(IBatchedTimer):
 
     def __init__(self, bucket_milliseconds, chunk_size,
                  seconds_provider, delayed_call_creator):
+        if bucket_milliseconds <= 0.0:
+            raise ValueError(
+                "bucket_milliseconds must be > 0.0"
+            )
         self._bucket_milliseconds = float(bucket_milliseconds)
         self._chunk_size = chunk_size
         self._get_seconds = seconds_provider
@@ -50,15 +54,20 @@ class _BatchedTimer(IBatchedTimer):
         IBatchedTimer API
         """
         # "quantize" the delay to the nearest bucket
-        real_time = int(self._get_seconds() + delay) * 1000
+        now = self._get_seconds()
+        real_time = int(now + delay) * 1000
         real_time -= int(real_time % self._bucket_milliseconds)
         call = _BatchedCall(self, real_time, lambda: func(*args, **kwargs))
         try:
             self._buckets[real_time][1].append(call)
         except KeyError:
             # new bucket; need to add "actual" underlying IDelayedCall
+            diff = (real_time / 1000.0) - now
+            # we need to clamp this because if we quantized to the
+            # nearest second, but that second is actually (slightly)
+            # less than the current time 'diff' will be negative.
             delayed_call = self._create_delayed_call(
-                (real_time / 1000.0) - self._get_seconds(),
+                max(0.0, diff),
                 self._notify_bucket, real_time,
             )
             self._buckets[real_time] = (delayed_call, [call])
@@ -96,7 +105,9 @@ class _BatchedTimer(IBatchedTimer):
         # ceil()ing because we want the number of chunks, and a
         # partial chunk is still a chunk
         delay_ms = self._bucket_milliseconds / math.ceil(float(len(calls)) / self._chunk_size)
-        notify_one_chunk(calls, self._chunk_size, delay_ms)
+        # I can't imagine any scenario in which chunk_delay_ms is
+        # actually less than zero, but just being safe here
+        notify_one_chunk(calls, self._chunk_size, max(0.0, delay_ms))
 
     def _remove_call(self, real_time, call):
         """

txaio/_unframework.py

@@ -44,6 +44,7 @@ def _throw_usage_error(*args, **kw):
         "with .use_twisted() or .use_asyncio()"
     )
 
+
 # all the txaio API methods just raise the error
 create_future = _throw_usage_error
 create_future_success = _throw_usage_error

txaio/_version.py

@@ -1 +1 @@
-__version__ = u'2.5.2'
+__version__ = u'2.6.0'