diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py
index 2f16f23d91..7253ba120f 100644
--- a/synapse/util/async_helpers.py
+++ b/synapse/util/async_helpers.py
@@ -156,6 +156,25 @@ def concurrently_execute(func, args, limit):
], consumeErrors=True)).addErrback(unwrapFirstError)
+def yieldable_gather_results(func, iter, *args, **kwargs):
+ """Executes the function with each argument concurrently.
+
+ Args:
+ func (func): Function to execute that returns a Deferred
+ iter (iter): An iterable that yields items that get passed as the first
+ argument to the function
+ *args: Arguments to be passed to each call to func
+
+ Returns
+ Deferred[list]: Resolved when all functions have been invoked, or errors if
+ one of the function calls fails.
+ """
+ return logcontext.make_deferred_yieldable(defer.gatherResults([
+ run_in_background(func, item, *args, **kwargs)
+ for item in iter
+ ], consumeErrors=True)).addErrback(unwrapFirstError)
+
+
class Linearizer(object):
"""Limits concurrent access to resources based on a key. Useful to ensure
only a few things happen at a time on a given resource.
diff --git a/synapse/util/distributor.py b/synapse/util/distributor.py
index 194da87639..e14c8bdfda 100644
--- a/synapse/util/distributor.py
+++ b/synapse/util/distributor.py
@@ -27,6 +27,7 @@ def user_left_room(distributor, user, room_id):
distributor.fire("user_left_room", user=user, room_id=room_id)
+# XXX: this is no longer used. We should probably kill it.
def user_joined_room(distributor, user, room_id):
distributor.fire("user_joined_room", user=user, room_id=room_id)
diff --git a/synapse/util/ratelimitutils.py b/synapse/util/ratelimitutils.py
index 7deb38f2a7..b146d137f4 100644
--- a/synapse/util/ratelimitutils.py
+++ b/synapse/util/ratelimitutils.py
@@ -30,31 +30,14 @@ logger = logging.getLogger(__name__)
class FederationRateLimiter(object):
- def __init__(self, clock, window_size, sleep_limit, sleep_msec,
- reject_limit, concurrent_requests):
+ def __init__(self, clock, config):
"""
Args:
clock (Clock)
- window_size (int): The window size in milliseconds.
- sleep_limit (int): The number of requests received in the last
- `window_size` milliseconds before we artificially start
- delaying processing of requests.
- sleep_msec (int): The number of milliseconds to delay processing
- of incoming requests by.
- reject_limit (int): The maximum number of requests that are can be
- queued for processing before we start rejecting requests with
- a 429 Too Many Requests response.
- concurrent_requests (int): The number of concurrent requests to
- process.
+ config (FederationRateLimitConfig)
"""
self.clock = clock
-
- self.window_size = window_size
- self.sleep_limit = sleep_limit
- self.sleep_msec = sleep_msec
- self.reject_limit = reject_limit
- self.concurrent_requests = concurrent_requests
-
+ self._config = config
self.ratelimiters = {}
def ratelimit(self, host):
@@ -76,25 +59,25 @@ class FederationRateLimiter(object):
host,
_PerHostRatelimiter(
clock=self.clock,
- window_size=self.window_size,
- sleep_limit=self.sleep_limit,
- sleep_msec=self.sleep_msec,
- reject_limit=self.reject_limit,
- concurrent_requests=self.concurrent_requests,
+ config=self._config,
)
).ratelimit()
class _PerHostRatelimiter(object):
- def __init__(self, clock, window_size, sleep_limit, sleep_msec,
- reject_limit, concurrent_requests):
+ def __init__(self, clock, config):
+ """
+ Args:
+ clock (Clock)
+ config (FederationRateLimitConfig)
+ """
self.clock = clock
- self.window_size = window_size
- self.sleep_limit = sleep_limit
- self.sleep_sec = sleep_msec / 1000.0
- self.reject_limit = reject_limit
- self.concurrent_requests = concurrent_requests
+ self.window_size = config.window_size
+ self.sleep_limit = config.sleep_limit
+ self.sleep_sec = config.sleep_delay / 1000.0
+ self.reject_limit = config.reject_limit
+ self.concurrent_requests = config.concurrent
# request_id objects for requests which have been slept
self.sleeping_requests = set()
|
