mirror of
https://github.com/element-hq/synapse
synced 2024-08-20 05:00:20 +00:00
Merge pull request #1725 from matrix-org/erikj/timeout_conn
Wrap connections in an N minute timeout to ensure they get reaped correctly
This commit is contained in:
commit
828c58522e
2 changed files with 70 additions and 7 deletions
|
@ -14,7 +14,7 @@
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
|
|
||||||
from twisted.internet.endpoints import SSL4ClientEndpoint, TCP4ClientEndpoint
|
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.internet.error import ConnectError
|
||||||
from twisted.names import client, dns
|
from twisted.names import client, dns
|
||||||
from twisted.names.error import DNSNameError, DomainError
|
from twisted.names.error import DNSNameError, DomainError
|
||||||
|
@ -66,13 +66,75 @@ def matrix_federation_endpoint(reactor, destination, ssl_context_factory=None,
|
||||||
default_port = 8448
|
default_port = 8448
|
||||||
|
|
||||||
if port is None:
|
if port is None:
|
||||||
return SRVClientEndpoint(
|
return _WrappingEndpointFac(SRVClientEndpoint(
|
||||||
reactor, "matrix", domain, protocol="tcp",
|
reactor, "matrix", domain, protocol="tcp",
|
||||||
default_port=default_port, endpoint=transport_endpoint,
|
default_port=default_port, endpoint=transport_endpoint,
|
||||||
endpoint_kw_args=endpoint_kw_args
|
endpoint_kw_args=endpoint_kw_args
|
||||||
)
|
))
|
||||||
else:
|
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):
|
class SpiderEndpoint(object):
|
||||||
|
|
|
@ -88,7 +88,8 @@ class MatrixFederationHttpClient(object):
|
||||||
self.signing_key = hs.config.signing_key[0]
|
self.signing_key = hs.config.signing_key[0]
|
||||||
self.server_name = hs.hostname
|
self.server_name = hs.hostname
|
||||||
pool = HTTPConnectionPool(reactor)
|
pool = HTTPConnectionPool(reactor)
|
||||||
pool.maxPersistentPerHost = 10
|
pool.maxPersistentPerHost = 5
|
||||||
|
pool.cachedConnectionTimeout = 2 * 60
|
||||||
self.agent = Agent.usingEndpointFactory(
|
self.agent = Agent.usingEndpointFactory(
|
||||||
reactor, MatrixFederationEndpointFactory(hs), pool=pool
|
reactor, MatrixFederationEndpointFactory(hs), pool=pool
|
||||||
)
|
)
|
||||||
|
@ -299,7 +300,7 @@ class MatrixFederationHttpClient(object):
|
||||||
defer.returnValue(json.loads(body))
|
defer.returnValue(json.loads(body))
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def post_json(self, destination, path, data={}, long_retries=True,
|
def post_json(self, destination, path, data={}, long_retries=False,
|
||||||
timeout=None):
|
timeout=None):
|
||||||
""" Sends the specifed json data using POST
|
""" Sends the specifed json data using POST
|
||||||
|
|
||||||
|
@ -332,7 +333,7 @@ class MatrixFederationHttpClient(object):
|
||||||
path.encode("ascii"),
|
path.encode("ascii"),
|
||||||
body_callback=body_callback,
|
body_callback=body_callback,
|
||||||
headers_dict={"Content-Type": ["application/json"]},
|
headers_dict={"Content-Type": ["application/json"]},
|
||||||
long_retries=True,
|
long_retries=long_retries,
|
||||||
timeout=timeout,
|
timeout=timeout,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue