Minor polish on cfexecutor
@@ -128,10 +128,9 @@ class ClientExecutor(cf.Executor):
fs = self._client.map(fn, *iterables, **self._kwargs)
- if isinstance(fs, list):
- # Below iterator relies on this being a generator to cancel
- # remaining futures
- fs = (val for val in fs)
+ # Below iterator relies on fs being an iterator itself, and not just an iterable
+ # (such as a list), in order to cancel remaining futures
+ fs = iter(fs)
# Yield must be hidden in closure so that the tasks are submitted
# before the first iterator value is required.
@@ -148,8 +147,7 @@ class ClientExecutor(cf.Executor):
yield future.result()
finally:
remaining = list(fs)
- for future in remaining:
- self._futures.add(future)
+ self._futures.update(remaining)
self._client.cancel(remaining)
return result_iterator()
