bpo-29603 New Queue.unfinished() method

Doc/library/queue.rst

@@ -190,6 +190,17 @@ Example of how to wait for enqueued tasks to be completed::
     for t in threads:
         t.join()
 
+One more method provides information about progress of tasks completion.
+
+.. method:: Queue.unfinished()
+
+   Return number of unfinished tasks i.e. number of such tasks which were queued but
+   consumer threads hadn't yet indicated that task is completed (via call of
+   :meth:`task_done`).
+
+   To get the correct value from this method you have to call :meth:`task_done`
+   for each call to :meth:`get` used to fetch a task to indicate that the item was
+   retrieved and all work on it is complete.
 
 .. seealso::
 

Lib/queue.py

@@ -82,6 +82,13 @@ def join(self):
             while self.unfinished_tasks:
                 self.all_tasks_done.wait()
 
+    def unfinished(self):
+        '''Return number of unfinished tasks i.e. number of such tasks which
+        were queued but consumer threads hadn't yet indicated that task is
+        completed (via call of method task_done()).
+        '''
+        return self.unfinished_tasks
+
     def qsize(self):
         '''Return the approximate size of the queue (not reliable!).'''
         with self.mutex:

Lib/test/test_queue.py

@@ -193,6 +193,15 @@ def test_queue_join(self):
         else:
             self.fail("Did not detect task count going negative")
 
+    def test_queue_unfinished(self):
+        q = self.type2test()
+        q.put(1)
+        q.put(2)
+        self.assertEqual(q.unfinished(), 2)
+        q.get()
+        q.task_done()
+        self.assertEqual(q.unfinished(), 1)
+
     def test_simple_queue(self):
         # Do it a couple of times on the same queue.
         # Done twice to make sure works with same instance reused.