summary refs log tree commit diff
path: root/synapse/util/async.py
diff options
context:
space:
mode:
authorMatthew Hodgson <matthew@matrix.org>2016-04-04 00:38:21 +0100
committerMatthew Hodgson <matthew@matrix.org>2016-04-04 00:38:21 +0100
commit9f7dc2bef7cd39645ae74d96f2919a3f9fdb65ac (patch)
treeb818eb0b962de7e860584bb5286f98e60e12ac75 /synapse/util/async.py
parentreport image size (bytewise) in OG meta (diff)
parentMerge pull request #686 from matrix-org/markjh/doc_strings (diff)
downloadsynapse-9f7dc2bef7cd39645ae74d96f2919a3f9fdb65ac.tar.xz
Merge branch 'develop' into matthew/preview_urls
Diffstat (limited to 'synapse/util/async.py')
-rw-r--r--synapse/util/async.py32
1 files changed, 31 insertions, 1 deletions
diff --git a/synapse/util/async.py b/synapse/util/async.py
index 640fae3890..cd4d90f3cf 100644
--- a/synapse/util/async.py
+++ b/synapse/util/async.py
@@ -16,7 +16,8 @@
 
 from twisted.internet import defer, reactor
 
-from .logcontext import PreserveLoggingContext
+from .logcontext import PreserveLoggingContext, preserve_fn
+from synapse.util import unwrapFirstError
 
 
 @defer.inlineCallbacks
@@ -107,3 +108,32 @@ class ObservableDeferred(object):
         return "<ObservableDeferred object at %s, result=%r, _deferred=%r>" % (
             id(self), self._result, self._deferred,
         )
+
+
+def concurrently_execute(func, args, limit):
+    """Executes the function with each argument conncurrently while limiting
+    the number of concurrent executions.
+
+    Args:
+        func (func): Function to execute, should return a deferred.
+        args (list): List of arguments to pass to func, each invocation of func
+            gets a signle argument.
+        limit (int): Maximum number of conccurent executions.
+
+    Returns:
+        deferred: Resolved when all function invocations have finished.
+    """
+    it = iter(args)
+
+    @defer.inlineCallbacks
+    def _concurrently_execute_inner():
+        try:
+            while True:
+                yield func(it.next())
+        except StopIteration:
+            pass
+
+    return defer.gatherResults([
+        preserve_fn(_concurrently_execute_inner)()
+        for _ in xrange(limit)
+    ], consumeErrors=True).addErrback(unwrapFirstError)