diff options
| -rw-r--r-- | synapse/http/endpoint.py | 70 | ||||
| -rw-r--r-- | synapse/http/matrixfederationclient.py | 7 |
2 files changed, 70 insertions, 7 deletions
diff --git a/synapse/http/endpoint.py b/synapse/http/endpoint.py index 442696d393..8c64339a7c 100644 --- a/synapse/http/endpoint.py +++ b/synapse/http/endpoint.py @@ -14,7 +14,7 @@ # limitations under the License. from twisted.internet.endpoints import SSL4ClientEndpoint, TCP4ClientEndpoint -from twisted.internet import defer +from twisted.internet import defer, reactor from twisted.internet.error import ConnectError from twisted.names import client, dns from twisted.names.error import DNSNameError, DomainError @@ -66,13 +66,75 @@ def matrix_federation_endpoint(reactor, destination, ssl_context_factory=None, default_port = 8448 if port is None: - return SRVClientEndpoint( + return _WrappingEndpointFac(SRVClientEndpoint( reactor, "matrix", domain, protocol="tcp", default_port=default_port, endpoint=transport_endpoint, endpoint_kw_args=endpoint_kw_args - ) + )) else: - return transport_endpoint(reactor, domain, port, **endpoint_kw_args) + return _WrappingEndpointFac(transport_endpoint( + reactor, domain, port, **endpoint_kw_args + )) + + +class _WrappingEndpointFac(object): + def __init__(self, endpoint_fac): + self.endpoint_fac = endpoint_fac + + @defer.inlineCallbacks + def connect(self, protocolFactory): + conn = yield self.endpoint_fac.connect(protocolFactory) + conn = _WrappedConnection(conn) + defer.returnValue(conn) + + +class _WrappedConnection(object): + """Wraps a connection and calls abort on it if it hasn't seen any action + for 2.5-3 minutes. + """ + __slots__ = ["conn", "last_request"] + + def __init__(self, conn): + object.__setattr__(self, "conn", conn) + object.__setattr__(self, "last_request", time.time()) + + def __getattr__(self, name): + return getattr(self.conn, name) + + def __setattr__(self, name, value): + setattr(self.conn, name, value) + + def _time_things_out_maybe(self): + # We use a slightly shorter timeout here just in case the callLater is + # triggered early. Paranoia ftw. + # TODO: Cancel the previous callLater rather than comparing time.time()? + if time.time() - self.last_request >= 2.5 * 60: + self.abort() + # Abort the underlying TLS connection. The abort() method calls + # loseConnection() on the underlying TLS connection which tries to + # shutdown the connection cleanly. We call abortConnection() + # since that will promptly close the underlying TCP connection. + self.transport.abortConnection() + + def request(self, request): + self.last_request = time.time() + + # Time this connection out if we haven't send a request in the last + # N minutes + # TODO: Cancel the previous callLater? + reactor.callLater(3 * 60, self._time_things_out_maybe) + + d = self.conn.request(request) + + def update_request_time(res): + self.last_request = time.time() + # TODO: Cancel the previous callLater? + reactor.callLater(3 * 60, self._time_things_out_maybe) + return res + + d.addCallback(update_request_time) + + return d class SpiderEndpoint(object): diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py index d5970c05a8..78b92cef36 100644 --- a/synapse/http/matrixfederationclient.py +++ b/synapse/http/matrixfederationclient.py @@ -88,7 +88,8 @@ class MatrixFederationHttpClient(object): self.signing_key = hs.config.signing_key[0] self.server_name = hs.hostname pool = HTTPConnectionPool(reactor) - pool.maxPersistentPerHost = 10 + pool.maxPersistentPerHost = 5 + pool.cachedConnectionTimeout = 2 * 60 self.agent = Agent.usingEndpointFactory( reactor, MatrixFederationEndpointFactory(hs), pool=pool ) @@ -299,7 +300,7 @@ class MatrixFederationHttpClient(object): defer.returnValue(json.loads(body)) @defer.inlineCallbacks - def post_json(self, destination, path, data={}, long_retries=True, + def post_json(self, destination, path, data={}, long_retries=False, timeout=None): """ Sends the specifed json data using POST @@ -332,7 +333,7 @@ class MatrixFederationHttpClient(object): path.encode("ascii"), body_callback=body_callback, headers_dict={"Content-Type": ["application/json"]}, - long_retries=True, + long_retries=long_retries, timeout=timeout, ) |