Fixed behavior of concurrent.futures.as_completed() for duplicate arguments

CHANGES

@@ -5,7 +5,8 @@
 - Removed the deprecated "futures" top level package
 - Applied patch for issue 15015 (accessing an non-existing attribute)
 - Applied patch for issue 16284 (memory leak)
-
+- Applied patch for issue 20367 (behavior of concurrent.futures.as_completed()
+                                 for duplicate arguments)
 
 2.2.0
 =====

concurrent/futures/_base.py

@@ -181,7 +181,8 @@ def as_completed(fs, timeout=None):
 
     Returns:
         An iterator that yields the given Futures as they complete (finished or
-        cancelled).
+        cancelled). If any given Futures are duplicated, they will be returned
+        once.
 
     Raises:
         TimeoutError: If the entire result iterator could not be generated
@@ -190,11 +191,12 @@ def as_completed(fs, timeout=None):
     if timeout is not None:
         end_time = timeout + time.time()
 
+    fs = set(fs)
     with _AcquireFutures(fs):
         finished = set(
                 f for f in fs
                 if f._state in [CANCELLED_AND_NOTIFIED, FINISHED])
-        pending = set(fs) - finished
+        pending = fs - finished
         waiter = _create_and_install_waiters(fs, _AS_COMPLETED)
 
     try:

docs/index.rst

@@ -335,11 +335,13 @@ Module Functions
 
 .. function:: as_completed(fs, timeout=None)
 
-   Returns an iterator over the :class:`Future` instances  (possibly created
-   by different :class:`Executor` instances) given by *fs* that yields futures
-   as they complete (finished or were cancelled). Any futures that completed
-   before :func:`as_completed()` was called will be yielded first. The returned
-   iterator raises a :exc:`TimeoutError` if :meth:`__next__()` is called and
-   the result isn't available after *timeout* seconds from the original call
-   to :func:`as_completed()`. *timeout* can be an int or float. If *timeout*
-   is not specified or ``None`` then there is no limit to the wait time.
+   Returns an iterator over the :class:`Future` instances (possibly created by
+   different :class:`Executor` instances) given by *fs* that yields futures as
+   they complete (finished or were cancelled). Any futures given by *fs* that
+   are duplicated will be returned once. Any futures that completed
+   before :func:`as_completed` is called will be yielded first.  The returned
+   iterator raises a :exc:`TimeoutError` if :meth:`~iterator.__next__` is
+   called and the result isn't available after *timeout* seconds from the
+   original call to :func:`as_completed`.  *timeout* can be an int or float.
+   If *timeout* is not specified or ``None``, there is no limit to the wait
+   time.

test_futures.py

@@ -423,6 +423,13 @@ class AsCompletedTests(unittest.TestCase):
                               SUCCESSFUL_FUTURE]),
                          completed_futures)
 
+    def test_duplicate_futures(self):
+        # Issue 20367. Duplicate futures should not raise exceptions or give
+        # duplicate responses.
+        future1 = self.executor.submit(time.sleep, 2)
+        completed = [f for f in futures.as_completed([future1,future1])]
+        self.assertEqual(len(completed), 1)
+
 
 class ThreadPoolAsCompletedTests(ThreadPoolMixin, AsCompletedTests):
     pass