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.

> 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!

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?

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.

>          """
>  
>  
> @@ -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.

>      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?

>  
>  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?

> +            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.

> +        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?

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

> === 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.

> +        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.

cheers,
jml