Jonathan Lange wrote:
> Review: Needs Fixing
> On Thu, Aug 6, 2009 at 10:18 AM, Michael Hudson<email address hidden> wrote:
>> You have been requested to review the proposed merge of lp:~mwhudson/launchpad/more-task-scheduled-bug-408638 into lp:launchpad/devel.
>>
>> EPIGRAMS IN PROGRAMMING:
>>
>> 58. Fools ignore complexity. Pragmatists suffer it. Some can avoid it. Geniuses remove it.
>>
>> I hope I'm being a pragmatist here.
>>
> 
> I think so.

Good :)  Some genius can clean up my mess later.

>> Hi Jono,
>>
>> I hope you can make time to have a look at the branch some time today.  If you don't have time for a full review, I'll get Tim to look at it tomorrow.  But you already know what a DeferredLock is :)
>>
>> If the code and tests don't make sense, then I've failed in my mission, but there are two issues addressed here:
>>
>> 1) Some race conditions around the puller exiting while a request for a job is pending (basically, change ITaskSource.stop to return deferred)
>> 2) Addressing the behaviour where only one job gets pulled per run of the puller (see bug 408638)
>>
> 
> Wow, concurrency really is quite hard!

Yeah, no kidding.

> However, I really do think that this branch is as simple as possible, barring any paradigm breakthroughs. Thanks for doing such a good job with it.
> 
> I've got a few comments and questions that I'd like you to address before this lands.
> 
>> === modified file 'lib/canonical/twistedsupport/task.py'
>> --- lib/canonical/twistedsupport/task.py	2009-07-17 00:26:05 +0000
>> +++ lib/canonical/twistedsupport/task.py	2009-08-06 09:08:47 +0000
>> @@ -37,6 +37,12 @@
>>          """Stop generating tasks.
>>  
>>          Any subsequent calls to `stop` are silently ignored.
>> +
>> +        :return: A Deferred that will fire when the source is stopped.  It is
>> +            possible that tasks may be produced until this deferred fires.
>> +            The deferred will fire with a boolean; True if the source is still
>> +            stopped, False if the source has been restarted since stop() was
>> +            called.
> 
> Do we know this for sure?

Well it's an interface docstring, so we can say so.

> What I mean is, is it possible for the source to have been restarted but for
> this to return False. It smells like a possible race condition.

I think in a twisted world, it makes sense to talk about the state of
the source at the moment the deferred is fired.  I guess you have to be
careful about any callbacks you add that themselves return deferred.

But anyway, until I did it like this, I had no idea how I was going to
make this all work.

>>          """
>>  
>>  
>> @@ -100,10 +106,13 @@
>>              clock = reactor
>>          self._clock = clock
>>          self._looping_call = None
>> +        self._polling_lock = defer.DeferredLock()
> 
> I think it's worth adding a comment on how this lock is used & why.
> 
>> +        self._started = False
>>
> 
> And maybe for this one too.

I realized that "_loopingcall is not None" was identical with "_started"
so I deleted _started.

>>      def start(self, task_consumer):
>>          """See `ITaskSource`."""
>>          self.stop()
>> +        self._started = True
>>          self._looping_call = LoopingCall(self._poll, task_consumer)
>>          self._looping_call.clock = self._clock
>>          self._looping_call.start(self._interval)
>> @@ -122,15 +131,21 @@
>>              # If task production fails, we inform the consumer of this, but we
>>              # don't let any deferred it returns delay subsequent polls.
>>              task_consumer.taskProductionFailed(reason)
>> -        d = defer.maybeDeferred(self._task_producer)
>> -        d.addCallbacks(got_task, task_failed)
>> -        return d
>> +        def poll():
>> +            if self._started:
>> +                d = defer.maybeDeferred(self._task_producer)
>> +                return d.addCallbacks(got_task, task_failed)
>> +        return self._polling_lock.run(poll)
>>  
>>      def stop(self):
>>          """See `ITaskSource`."""
>>          if self._looping_call is not None:
>>              self._looping_call.stop()
>>              self._looping_call = None
>> +        self._started = False
>> +        def _return_still_stopped():
>> +            return not self._started
>> +        return self._polling_lock.run(_return_still_stopped)
>>
> 
> Reading this and the function above it, the logic seems essentially to say,
> "don't return from stop while we're still polling".
> 
> This makes me think of my comment on the interface docstring above. Is this
> behaviour a part of the contract, or is it an implementation detail?

I think it's part of the contract: in a message queue version,
presumably unsubscribing from the queue involves the network, i.e. a
deferred will be in there somewhere (even stopListening() returns
deferred!), so you can't call stop() and be sure that a task won't be
arriving just as you do so.

>>  
>>  class AlreadyRunningError(Exception):
>> @@ -164,6 +179,19 @@
>>          self._worker_limit = worker_limit
>>          self._worker_count = 0
>>          self._terminationDeferred = None
>> +        self._stopping_lock = None
>> +
>> +    def _stop(self):
>> +        def _release_or_stop(still_stopped):
>> +            if still_stopped and self._worker_count == 0:
>> +                self._terminationDeferred.callback(None)
>> +                # Note that in this case we don't release the lock: we don't
>> +                # want to try to fire the _terminationDeferred twice!
> 
> I don't quite follow this. Why would releasing the lock fire the
> _terminationDeferred again?

I'm not completely sure about this, but if you manage to call _stop()
twice, you could end up here twice.

>> +            else:
>> +                self._stopping_lock.release()
>> +        def _call_stop(ignored):
>> +            return self._task_source.stop().addCallback(_release_or_stop)
> 
> Does this callback have to be added here? Couldn't it also be added directly
> to the _stopping_lock.acquire() callback? I think that adding the callbacks in
> sequence would make the code slightly easier to read.

Yeah, I think you're right.

>> +        return self._stopping_lock.acquire().addCallback(_call_stop)
>>
> 
>>      def consume(self, task_source):
>>          """Start consuming tasks from 'task_source'.
>> @@ -178,9 +206,7 @@
>>              raise AlreadyRunningError(self, self._task_source)
>>          self._task_source = task_source
>>          self._terminationDeferred = defer.Deferred()
>> -        # This merely begins polling. This means that we acquire our initial
>> -        # batch of work at the rate of one task per polling interval. As long
>> -        # as the polling interval is small, this is probably OK.
> 
> Why isn't this comment relevant any more?

Because of the change in taskStarted:

>> +        self._stopping_lock = defer.DeferredLock()
>>          task_source.start(self)
>>          return self._terminationDeferred

This one.

> 
>> === modified file 'lib/canonical/twistedsupport/tests/test_task.py'
>> --- lib/canonical/twistedsupport/tests/test_task.py	2009-07-17 00:26:05 +0000
>> +++ lib/canonical/twistedsupport/tests/test_task.py	2009-08-06 09:08:47 +0000
>> @@ -7,7 +7,7 @@
>>  
>>  import unittest
>>  
>> -from twisted.internet.defer import Deferred
>> +from twisted.internet.defer import Deferred, succeed
>>  from twisted.internet.task import Clock
>>  
>>  from zope.interface import implements
>> @@ -49,14 +49,19 @@
>>  
>>      implements(ITaskSource)
>>  
>> -    def __init__(self, log):
>> +    def __init__(self, log, stop_deferred=None):
>>          self._log = log
>> +        if stop_deferred is None:
>> +            self.stop_deferred = succeed(True)
>> +        else:
>> +            self.stop_deferred = stop_deferred
>>  
>>      def start(self, consumer):
>>          self._log.append(('start', consumer))
>>  
>>      def stop(self):
>>          self._log.append('stop')
>> +        return self.stop_deferred
>>  
>>  
>>  class TestPollingTaskSource(TestCase):
>> @@ -144,6 +149,16 @@
>>          # No more calls were made.
>>          self.assertEqual(0, self._num_task_producer_calls)
>>  
>> +    def test_stop_deferred_fires_immediately_if_no_polling(self):
>> +        # Calling stop when the source is not polling returns a deferred that
>> +        # fires immediately with True.
>> +        task_source = self.makeTaskSource()
>> +        task_source.start(NoopTaskConsumer())
>> +        stop_deferred = task_source.stop()
>> +        stop_calls = []
>> +        stop_deferred.addCallback(stop_calls.append)
>> +        self.assertEqual([True], stop_calls)
>> +
>>      def test_start_multiple_times_polls_immediately(self):
>>          # Starting a task source multiple times polls immediately.
>>          clock = Clock()
>> @@ -241,6 +256,74 @@
>>          clock.advance(interval)
>>          self.assertEqual(len(produced_deferreds), 2)
>>  
>> +    def test_stop_deferred_doesnt_fire_until_polling_finished(self):
>> +        # If there is a call to the task producer outstanding when stop() is
>> +        # called, stop() returns a deferred that fires when the poll finishes.
>> +        # The value fired with is True if the source is still stopped when the
>> +        # deferred fires.
>> +        produced_deferred = Deferred()
>> +        def producer():
>> +            return produced_deferred
>> +        task_source = self.makeTaskSource(task_producer=producer)
>> +        task_source.start(NoopTaskConsumer())
>> +        # The call to start calls producer.  It returns produced_deferred
>> +        # which has not been fired, so stop returns a deferred that has not
>> +        # been fired.
>> +        stop_deferred = task_source.stop()
>> +        stop_called = []
>> +        stop_deferred.addCallback(stop_called.append)
>> +        self.assertEqual([], stop_called)
>> +        # When the task producing deferred fires, the stop deferred fires with
>> +        # 'True' to indicate that the source is still stopped.
>> +        produced_deferred.callback(None)
>> +        self.assertEqual([True], stop_called)
>> +
>> +    def test_stop_deferred_fires_with_false_if_source_restarted(self):
>> +        # If there is a call to the task producer outstanding when stop() is
>> +        # called, stop() returns a deferred that fires when the poll finishes.
>> +        # The value fired with is False if the source is no longer stopped
>> +        # when the deferred fires.
>> +        produced_deferred = Deferred()
>> +        def producer():
>> +            return produced_deferred
>> +        task_source = self.makeTaskSource(task_producer=producer)
>> +        task_source.start(NoopTaskConsumer())
>> +        # The call to start calls producer.  It returns produced_deferred
>> +        # which has not been fired so stop returns a deferred that has not
>> +        # been fired.
>> +        stop_deferred = task_source.stop()
>> +        stop_called = []
>> +        stop_deferred.addCallback(stop_called.append)
>> +        # Now we restart the source.
>> +        task_source.start(NoopTaskConsumer())
>> +        self.assertEqual([], stop_called)
>> +        # When the task producing deferred fires, the stop deferred fires with
>> +        # 'False' to indicate that the source has been restarted.
>> +        produced_deferred.callback(None)
>> +        self.assertEqual([False], stop_called)
>> +
>> +    def test_stop_start_stop_when_polling_doesnt_poll_again(self):
>> +        # XXX
> 
> Docstring here.

Dammit, I thought I'd got all of them.  Added.

>> +        produced_deferreds = []
>> +        def producer():
>> +            d = Deferred()
>> +            produced_deferreds.append(d)
>> +            return d
>> +        task_source = self.makeTaskSource(task_producer=producer)
>> +        # Start the source.  This calls the producer.
>> +        task_source.start(NoopTaskConsumer())
>> +        self.assertEqual(1, len(produced_deferreds))
>> +        task_source.stop()
>> +        # If we start it again, this does not call the producer because
>> +        # the above call is still in process.
>> +        task_source.start(NoopTaskConsumer())
>> +        self.assertEqual(1, len(produced_deferreds))
>> +        # If we now stop the source and the initial poll for a task completes,
>> +        # we don't poll again.
>> +        task_source.stop()
>> +        produced_deferreds[0].callback(None)
>> +        self.assertEqual(1, len(produced_deferreds))
>> +
>>      def test_taskStarted_deferred_doesnt_delay_polling(self):
>>          # If taskStarted returns a deferred, we don't wait for it to fire
>>          # before polling again.
>> @@ -354,7 +437,7 @@
>>          consumer.consume(source)
>>          self.assertRaises(AlreadyRunningError, consumer.consume, source)
>>  
>> -    def test_consume_returns_deferred_doesnt_fire_until_tasks(self):
>> +    def test_consumer_doesnt_finish_until_tasks_finish(self):
>>          # `consume` returns a Deferred that fires when no more tasks are
>>          # running, but only after we've actually done something.
>>          consumer = self.makeConsumer()
>> @@ -363,7 +446,7 @@
>>          d.addCallback(log.append)
>>          self.assertEqual([], log)
>>  
>> -    def test_consume_returns_deferred_fires_when_tasks_done(self):
>> +    def test_consumer_finishes_when_tasks_done(self):
>>          # `consume` returns a Deferred that fires when no more tasks are
>>          # running.
>>          consumer = self.makeConsumer()
>> @@ -373,7 +456,7 @@
>>          consumer.taskStarted(lambda: None)
>>          self.assertEqual([None], task_log)
>>  
>> -    def test_consume_returns_deferred_fires_if_no_tasks_found(self):
>> +    def test_consumer_finishes_if_no_tasks_found(self):
>>          # `consume` returns a Deferred that fires if no tasks are found when
>>          # no tasks are running.
>>          consumer = self.makeConsumer()
>> @@ -383,7 +466,37 @@
>>          consumer.noTasksFound()
>>          self.assertEqual([None], task_log)
> 
> These method renames are nice, thanks.

Without the renaming some of my new tests had names that got over the
80th column all by themselves :)

Cheers,
mwh