Synnefo » snf-ganeti » ganeti-local

    self._active = True

  def SetActive(self, active):

    """Enable/disable processing of tasks.

    This is different from L{Quiesce} in the sense that this function just

    changes an internal flag and doesn't wait for the queue to be empty. Tasks

    already being processed continue normally, but no new tasks will be

    started. New tasks can still be added.

    @type active: bool

    @param active: Whether tasks should be processed

    """

    self._lock.acquire()

    try:

      self._active = active

      if active:

        # Tell all workers to continue processing

        self._pool_to_worker.notifyAll()

    finally:

      self._lock.release()

    if not (self._active and self._tasks):

        # Just loop if pool is not processing tasks at this time

        if self._active and self._tasks:

  def testActive(self):

    ctx = CountingContext()

    wp = workerpool.WorkerPool("TestActive", 5, CountingBaseWorker)

    try:

      self._CheckWorkerCount(wp, 5)

      self.assertTrue(wp._active)

      # Process some tasks

      for _ in range(10):

        wp.AddTask((ctx, None))

      wp.Quiesce()

      self._CheckNoTasks(wp)

      self.assertEquals(ctx.GetDoneTasks(), 10)

      # Repeat a few times

      for count in range(10):

        # Deactivate pool

        wp.SetActive(False)

        self._CheckNoTasks(wp)

        # Queue some more tasks

        for _ in range(10):

          wp.AddTask((ctx, None))

        for _ in range(5):

          # Short delays to give other threads a chance to cause breakage

          time.sleep(.01)

          wp.AddTask((ctx, "Hello world %s" % 999))

          self.assertFalse(wp._active)

        self.assertEquals(ctx.GetDoneTasks(), 10 + (count * 15))

        # Start processing again

        wp.SetActive(True)

        self.assertTrue(wp._active)

        # Wait for tasks to finish

        wp.Quiesce()

        self._CheckNoTasks(wp)

        self.assertEquals(ctx.GetDoneTasks(), 10 + (count * 15) + 15)

        self._CheckWorkerCount(wp, 5)

    finally:

      wp.TerminateWorkers()

      self._CheckWorkerCount(wp, 0)