def testBlockingTasks(self):
    # The pool starts off with the minimum number of threads.
    self.assertEqual(len(self.test_pool), self.NUMBER_OF_THREADS)

    done_event = threading.Event()
    self.lock = threading.Lock()
    res = []

    def Block(done):
      done.wait()

    def Insert(list_obj, element):
      with self.lock:
        list_obj.append(element)

    # Ensure that the thread pool is able to add the correct number of threads
    # deterministically.
    with utils.Stubber(self.test_pool, "CPUUsage", lambda: 0):
      # Schedule the maximum number of threads of blocking tasks and the same of
      # insert tasks. The threads are now all blocked, and the inserts are
      # waiting in the queue.
      for _ in range(self.MAXIMUM_THREADS):
        self.test_pool.AddTask(Block, (done_event,), "Blocking")

      # Wait until the threadpool picks up the task.
      self.WaitUntil(
          lambda: self.test_pool.busy_threads == self.NUMBER_OF_THREADS)

      # Now there are minimum number of threads active and the rest are sitting
      # on the queue.
      self.assertEqual(self.test_pool.pending_tasks,
                       self.MAXIMUM_THREADS - self.NUMBER_OF_THREADS)

      # Now as we push these tasks on the queue, new threads will be created and
      # they will receive the blocking tasks from the queue.
      for i in range(self.MAXIMUM_THREADS):
        self.test_pool.AddTask(Insert, (res, i), "Insert", blocking=True,
                               inline=False)

      # There should be 20 workers created and they should consume all the
      # blocking tasks.
      self.WaitUntil(
          lambda: self.test_pool.busy_threads == self.MAXIMUM_THREADS)

      # No Insert tasks are running yet.
      self.assertEqual(res, [])

      # There are 20 tasks waiting on the queue.
      self.assertEqual(self.test_pool.pending_tasks, self.MAXIMUM_THREADS)

      # Inserting more tasks than the queue can hold should lead to processing
      # the tasks inline. This effectively causes these tasks to skip over the
      # tasks which are waiting in the queue.
      for i in range(10, 20):
        self.test_pool.AddTask(Insert, (res, i), "Insert", inline=True)

      res.sort()
      self.assertEqual(res, range(10, 20))

      # This should release all the busy tasks. It will also cause the workers
      # to process all the Insert tasks in the queue.
      done_event.set()

      self.test_pool.Join()

      # Now the rest of the tasks should have been processed as well.
      self.assertEqual(sorted(res[10:]), range(20))
dataset/ETHPy150Open google/grr/grr/lib/threadpool_test.py/ThreadPoolTest.testBlockingTasks/original