Repository: twisted/twisted

Issue:
Lots of small writes to TLS transport use a lot of CPU
**Describe the incorrect behavior you saw**

If you do lots of small writes to a TLS transport, it uses a really significant amount of CPU and bandwidth compared to larger writes. The impact seems far more significant in TLS than it would be in TCP, e.g. TCP would combine the writes due to Nagle and therefore usually won't have 23× bandwidth increase!

**Describe how to cause this behavior**

See upstream pyOpenSSL issue for reproducer and numbers (10× CPU and 23× bandwidth): https://github.com/pyca/pyopenssl/issues/1250

I discovered this in the wild in Foolscap, which writes many things one byte at a time... with code basically copy/pasted from Twisted: https://github.com/twisted/twisted/blob/trunk/src/twisted/spread/banana.py#L31 and later code.

**Describe the correct behavior you'd like to see**

Fixing this in pyOpenSSL is likely difficult; there is the upstream bug but it's not clear to me an extra buffering layer there makes sense. In contrast, `twisted/protocols/tls.py` already has a buffering layer, albeit for other use cases (wantread/wantwrite errors from OpenSSL).

The solution I'm imagining is having the TLS protocol register as a pull producer with the underlying TCP transport and aggregate small writes until it gets a pull message.

Another alternative is to just document this behavior, and fix upstream protocol implementations to be less chatty.

**Testing environment**
- Twisted checkout as of Sep 12, 2023
- Ubuntu 22.04



Discussion:

Patch:
diff --git a/src/twisted/internet/_newtls.py b/src/twisted/internet/_newtls.py
index dbc12e007a..5c27f31eb9 100644
--- a/src/twisted/internet/_newtls.py
+++ b/src/twisted/internet/_newtls.py
@@ -15,7 +15,7 @@ from zope.interface import directlyProvides
 
 from twisted.internet.abstract import FileDescriptor
 from twisted.internet.interfaces import ISSLTransport
-from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol
+from twisted.protocols.tls import TLSMemoryBIOFactory
 
 
 class _BypassTLS:
@@ -128,7 +128,8 @@ def startTLS(transport, contextFactory, normal, bypass):
         transport.unregisterProducer()
 
     tlsFactory = TLSMemoryBIOFactory(contextFactory, client, None)
-    tlsProtocol = TLSMemoryBIOProtocol(tlsFactory, transport.protocol, False)
+    tlsProtocol = tlsFactory.protocol(tlsFactory, transport.protocol, False)
+    # Hook up the new TLS protocol to the transport:
     transport.protocol = tlsProtocol
 
     transport.getHandle = tlsProtocol.getHandle
diff --git a/src/twisted/newsfragments/11989.bugfix b/src/twisted/newsfragments/11989.bugfix
new file mode 100644
index 0000000000..bdf2171b4d
--- /dev/null
+++ b/src/twisted/newsfragments/11989.bugfix
@@ -0,0 +1,1 @@
+Fixed significant performance overhead (CPU and bandwidth) when doing small writes to a TLS transport. Specifically, small writes to a TLS transport are now buffered until the next reactor iteration.
\ No newline at end of file
diff --git a/src/twisted/protocols/tls.py b/src/twisted/protocols/tls.py
index e042b9113a..d2ac2d2cf8 100644
--- a/src/twisted/protocols/tls.py
+++ b/src/twisted/protocols/tls.py
@@ -36,6 +36,9 @@ implement onion routing.  It can also be used to run TLS over unusual
 transports, such as UNIX sockets and stdio.
 """
 
+from __future__ import annotations
+
+from typing import Callable, Iterable, Optional, cast
 
 from zope.interface import directlyProvides, implementer, providedBy
 
@@ -44,14 +47,18 @@ from OpenSSL.SSL import Connection, Error, SysCallError, WantReadError, ZeroRetu
 from twisted.internet._producer_helpers import _PullToPush
 from twisted.internet._sslverify import _setAcceptableProtocols
 from twisted.internet.interfaces import (
+    IDelayedCall,
     IHandshakeListener,
     ILoggingContext,
     INegotiated,
     IOpenSSLClientConnectionCreator,
     IOpenSSLServerConnectionCreator,
+    IProtocol,
     IProtocolNegotiationFactory,
     IPushProducer,
+    IReactorTime,
     ISystemHandle,
+    ITransport,
 )
 from twisted.internet.main import CONNECTION_LOST
 from twisted.internet.protocol import Protocol
@@ -116,7 +123,7 @@ def _representsEOF(exceptionObject: Error) -> bool:
     return reasonString.casefold().startswith("unexpected eof")
 
 
-@implementer(ISystemHandle, INegotiated)
+@implementer(ISystemHandle, INegotiated, ITransport)
 class TLSMemoryBIOProtocol(ProtocolWrapper):
     """
     L{TLSMemoryBIOProtocol} is a protocol wrapper which uses OpenSSL via a
@@ -130,6 +137,9 @@ class TLSMemoryBIOProtocol(ProtocolWrapper):
     merged using the L{_ProducerMembrane} wrapper.  Non-streaming (pull)
     producers are supported by wrapping them with L{_PullToPush}.
 
+    Because TLS may need to wait for reads before writing, some writes may be
+    buffered until a read occurs.
+
     @ivar _tlsConnection: The L{OpenSSL.SSL.Connection} instance which is
         encrypted and decrypting this connection.
 
@@ -449,8 +459,6 @@ class TLSMemoryBIOProtocol(ProtocolWrapper):
         If C{loseConnection} was called, subsequent calls to C{write} will
         drop the bytes on the floor.
         """
-        if isinstance(bytes, str):
-            raise TypeError("Must write bytes to a TLS transport, not str.")
         # Writes after loseConnection are not supported, unless a producer has
         # been registered, in which case writes can happen until the producer
         # is unregistered:
@@ -682,6 +690,117 @@ class _ContextFactoryToConnectionFactory:
         return self._connectionForTLS(protocol)
 
 
+class _AggregateSmallWrites:
+    """
+    Aggregate small writes so they get written in large batches.
+
+    If this is used as part of a transport, the transport needs to call
+    ``flush()`` immediately when ``loseConnection()`` is called, otherwise any
+    buffered writes will never get written.
+
+    @cvar MAX_BUFFER_SIZE: The maximum amount of bytes to buffer before writing
+        them out.
+    """
+
+    MAX_BUFFER_SIZE = 64_000
+
+    def __init__(self, write: Callable[[bytes], object], clock: IReactorTime):
+        self._write = write
+        self._clock = clock
+        self._buffer: list[bytes] = []
+        self._bufferLen = 0
+        self._scheduled: Optional[IDelayedCall] = None
+
+    def write(self, data: bytes) -> None:
+        """
+        Buffer the data, or write it immediately if we've accumulated enough to
+        make it worth it.
+
+        Accumulating too much data can result in higher memory usage.
+        """
+        self._buffer.append(data)
+        self._bufferLen += len(data)
+
+        if self._bufferLen > self.MAX_BUFFER_SIZE:
+            # We've accumulated enough we should just write it out. No need to
+            # schedule a flush, since we just flushed everything.
+            self.flush()
+            return
+
+        if self._scheduled:
+            # We already have a scheduled send, so with the data in the buffer,
+            # there is nothing more to do here.
+            return
+
+        # Schedule the write of the accumulated buffer for the next reactor
+        # iteration.
+        self._scheduled = self._clock.callLater(0, self._scheduledFlush)
+
+    def _scheduledFlush(self) -> None:
+        """Called in next reactor iteration."""
+        self._scheduled = None
+        self.flush()
+
+    def flush(self) -> None:
+        """Flush any buffered writes."""
+        if self._buffer:
+            self._bufferLen = 0
+            self._write(b"".join(self._buffer))
+            del self._buffer[:]
+
+
+def _get_default_clock() -> IReactorTime:
+    """
+    Return the default reactor.
+
+    This is a function so it can be monkey-patched in tests, specifically
+    L{twisted.web.test.test_agent}.
+    """
+    from twisted.internet import reactor
+
+    return cast(IReactorTime, reactor)
+
+
+class BufferingTLSTransport(TLSMemoryBIOProtocol):
+    """
+    A TLS transport implemented by wrapping buffering around a
+    ``TLSMemoryBIOProtocol``.
+
+    Doing many small writes directly to a ``OpenSSL.SSL.Connection``, as
+    implemented in ``TLSMemoryBIOProtocol``, can add significant CPU and
+    bandwidth overhead.  Thus, even when writing is possible, small writes will
+    get aggregated and written as a single write at the next reactor iteration.
+    """
+
+    # Note: An implementation based on composition would be nicer, but there's
+    # close integration between ``ProtocolWrapper`` subclasses like
+    # ``TLSMemoryBIOProtocol`` and the corresponding factory. Composition broke
+    # things like ``TLSMemoryBIOFactory.protocols`` having the correct
+    # instances, whereas subclassing makes that work.
+
+    def __init__(
+        self,
+        factory: TLSMemoryBIOFactory,
+        wrappedProtocol: IProtocol,
+        _connectWrapped: bool = True,
+    ):
+        super().__init__(factory, wrappedProtocol, _connectWrapped)
+        actual_write = super().write
+        self._aggregator = _AggregateSmallWrites(actual_write, factory._clock)
+
+    def write(self, data: bytes) -> None:
+        if isinstance(data, str):  # type: ignore[unreachable]
+            raise TypeError("Must write bytes to a TLS transport, not str.")
+        self._aggregator.write(data)
+
+    def writeSequence(self, sequence: Iterable[bytes]) -> None:
+        self._aggregator.write(b"".join(sequence))
+
+    def loseConnection(self) -> None:
+        self._aggregator.flush()
+        super().loseConnection()
+
+
 class TLSMemoryBIOFactory(WrappingFactory):
     """
     L{TLSMemoryBIOFactory} adds TLS to connections.
@@ -696,11 +815,17 @@ class TLSMemoryBIOFactory(WrappingFactory):
         L{TLSMemoryBIOProtocol} and returning L{OpenSSL.SSL.Connection}.
     """
 
-    protocol = TLSMemoryBIOProtocol
+    protocol = BufferingTLSTransport
 
     noisy = False  # disable unnecessary logging.
 
-    def __init__(self, contextFactory, isClient, wrappedFactory):
+    def __init__(
+        self,
+        contextFactory,
+        isClient,
+        wrappedFactory,
+        clock=None,
+    ):
         """
         Create a L{TLSMemoryBIOFactory}.
 
@@ -753,6 +878,10 @@ class TLSMemoryBIOFactory(WrappingFactory):
             contextFactory = _ContextFactoryToConnectionFactory(contextFactory)
         self._connectionCreator = contextFactory
 
+        if clock is None:
+            clock = _get_default_clock()
+        self._clock = clock
+
     def logPrefix(self):
         """
         Annotate the wrapped factory's log prefix with some text indicating TLS


Test patch:
diff --git a/src/twisted/internet/test/test_newtls.py b/src/twisted/internet/test/test_newtls.py
index 38ab05f3fe..9b92b8fc0b 100644
--- a/src/twisted/internet/test/test_newtls.py
+++ b/src/twisted/internet/test/test_newtls.py
@@ -91,7 +91,7 @@ class ProducerProtocol(ConnectableProtocol):
         self.result = result
 
     def handshakeCompleted(self):
-        if not isinstance(self.transport.protocol, tls.TLSMemoryBIOProtocol):
+        if not isinstance(self.transport.protocol, tls.BufferingTLSTransport):
             # Either the test or the code have a bug...
             raise RuntimeError("TLSMemoryBIOProtocol not hooked up.")
 
diff --git a/src/twisted/protocols/test/test_tls.py b/src/twisted/protocols/test/test_tls.py
index 828158d748..f8caad7b06 100644
--- a/src/twisted/protocols/test/test_tls.py
+++ b/src/twisted/protocols/test/test_tls.py
@@ -5,12 +5,17 @@
 Tests for L{twisted.protocols.tls}.
 """
 
+from __future__ import annotations
 
 import gc
+from typing import Union
 
 from zope.interface import Interface, directlyProvides, implementer
 from zope.interface.verify import verifyObject
 
+from hypothesis import given, strategies as st
+
+from twisted.internet.task import Clock
 from twisted.python.compat import iterbytes
 
 try:
@@ -30,6 +35,7 @@ try:
     from twisted.protocols.tls import (
         TLSMemoryBIOFactory,
         TLSMemoryBIOProtocol,
+        _AggregateSmallWrites,
         _ProducerMembrane,
         _PullToPush,
     )
@@ -164,7 +170,11 @@ def buildTLSProtocol(
             contextFactory = ServerTLSContext(method=serverMethod)
         else:
             contextFactory = ClientTLSContext()
-    wrapperFactory = TLSMemoryBIOFactory(contextFactory, not server, clientFactory)
+
+    clock = Clock()
+    wrapperFactory = TLSMemoryBIOFactory(
+        contextFactory, not server, clientFactory, clock
+    )
     sslProtocol = wrapperFactory.buildProtocol(None)
 
     if transport is None:
@@ -235,6 +245,7 @@ def handshakingClientAndServer(
     @return: 3-tuple of client, server, L{twisted.test.iosim.IOPump}
     """
     authCert, serverCert = certificatesForAuthorityAndServer()
+    clock = Clock()
 
     @implementer(IHandshakeListener)
     class Client(AccumulatingProtocol):
@@ -269,16 +280,19 @@ def handshakingClientAndServer(
         optionsForClientTLS("example.com", trustRoot=authCert),
         isClient=True,
         wrappedFactory=ClientFactory.forProtocol(lambda: Client(999999)),
+        clock=clock,
     )
     serverF = TLSMemoryBIOFactory(
         serverCert.options(),
         isClient=False,
         wrappedFactory=ServerFactory.forProtocol(lambda: Server(999999)),
+        clock=clock,
     )
     client, server, pump = connectedServerAndClient(
         lambda: serverF.buildProtocol(None),
         lambda: clientF.buildProtocol(None),
         greet=False,
+        clock=clock,
     )
     return client, server, pump
 
@@ -319,6 +333,24 @@ class DeterministicTLSMemoryBIOTests(SynchronousTestCase):
         pump.flush()
         self.assertEqual(wrappedServerProtocol.received, [])
 
+    def test_smalWriteBuffering(self):
+        """
+        If a small amount data is written to the TLS transport, it is only
+        delivered if time passes, indicating small-write buffering is in
+        effect.
+        """
+        client, server, pump = handshakingClientAndServer()
+        wrappedServerProtocol = server.wrappedProtocol
+        pump.flush()
+        self.assertEqual(wrappedServerProtocol.received, [])
+        client.write(b"hel")
+        client.write(b"lo")
+        self.assertEqual(wrappedServerProtocol.received, [])
+        pump.flush(advanceClock=False)
+        self.assertEqual(wrappedServerProtocol.received, [])
+        pump.flush(advanceClock=True)
+        self.assertEqual(b"".join(wrappedServerProtocol.received), b"hello")
+
 
 class TLSMemoryBIOTests(TestCase):
     """
@@ -1212,6 +1244,7 @@ class TLSProducerTests(TestCase):
         # cannot be written. Thus writing bytes before the handshake should
         # cause the producer to be paused:
         clientProtocol.transport.write(b"hello")
+        tlsProtocol.factory._clock.advance(0)
         self.assertEqual(producer.producerState, "paused")
         self.assertEqual(producer.producerHistory, ["pause"])
         self.assertTrue(tlsProtocol._producer._producerPaused)
@@ -1302,6 +1335,7 @@ class TLSProducerTests(TestCase):
         # haven't unregistered producer yet:
         clientProtocol.transport.write(b"hello")
         clientProtocol.transport.writeSequence([b" ", b"world"])
+        tlsProtocol.factory._clock.advance(0)
 
         # Unregister producer; this should trigger TLS shutdown:
         clientProtocol.transport.unregisterProducer()
@@ -1311,6 +1345,7 @@ class TLSProducerTests(TestCase):
         # Additional writes should not go through:
         clientProtocol.transport.write(b"won't")
         clientProtocol.transport.writeSequence([b"won't!"])
+        tlsProtocol.factory._clock.advance(0)
 
         # Finish TLS close handshake:
         self.flushTwoTLSProtocols(tlsProtocol, serverTLSProtocol)
@@ -1394,6 +1429,7 @@ class TLSProducerTests(TestCase):
         # WantReadError will be thrown, triggering the TLS transport's
         # producer code path.
         clientProtocol.transport.write(b"hello")
+        tlsProtocol.factory._clock.advance(0)
         self.assertEqual(producer.producerState, "paused")
         self.assertEqual(producer.producerHistory, ["pause"])
 
@@ -1819,3 +1855,82 @@ class ServerNegotiationFactory(ServerFactory):
         @rtype: L{list} of L{bytes}
         """
         return self._acceptableProtocols
+
+
+class _AggregateSmallWritesTests(SynchronousTestCase):
+    """Tests for ``_AggregateSmallWrites``."""
+
+    # Hypothesis will repeatedly generate different lists which will be a
+    # combination of instances of ``bytes`` and ``None``, and pass that as the
+    # ``writes`` parameter. The ``bytes`` should be passed to
+    # L{_AggregateSmallWrites.write}, a ``None`` indicates that some time has
+    # passed, which should cause the aggregator to write data out.
+    @given(
+        st.lists(
+            st.one_of(
+                st.none(),
+                # We could use st.bytes(min_size=1, max_size=100_000), but if
+                # we did we'd waste time exploring different byte contents,
+                # which in this particular case don't matter. We just care
+                # about lengths, so long as misordering or truncating is likely
+                # to be noticed.
+                st.integers(min_value=1, max_value=100_000).map(
+                    lambda length: (b"0123456789ABCDEFGHIJ" * ((length // 20) + 1))[
+                        :length
+                    ]
+                ),
+            ),
+            max_size=1_000,
+        )
+    )
+    def test_writes_get_aggregated(self, writes: list[Union[bytes, None]]) -> None:
+        """
+        A L{_AggregateSmallWrites} correctly aggregates data for the given
+        sequence of writes (indicated by bytes) and increments in the clock
+        (indicated by C{None}).
+
+        If multiple writes happen in between reactor iterations, they should
+        get written in a batch at the start of the next reactor iteration.
+        """
+        # Whenever the aggregate is flushed, it will write the bytes to this
+        # list:
+        result: list[bytes] = []
+
+        clock = Clock()
+        aggregate = _AggregateSmallWrites(result.append, clock)
+        for value in writes:
+            if value is None:
+                clock.advance(0)
+            else:
+                aggregate.write(value)
+
+        # Flush any remaining bytes:
+        aggregate.flush()
+
+        # Now, check the invariants:
+
+        # 1. All bytes passed to aggregate.write() should have been written
+        # out, in the correct order:
+        self.assertEqual(
+            b"".join(result), b"".join(value for value in writes if value is not None)
+        )
+
+        # 2. All writes happened either when the max buffer size was reached,
+        # or when the clock advanced:
+        small_writes = writes[:]
+        for chunk in result:
+            combined_length = len(chunk)
+            # Initial flushes have no side-effects:
+            while small_writes and small_writes[0] is None:
+                small_writes.pop(0)
+            small_writes_length = 0
+            while small_writes:
+                next_original_maybe_write = small_writes.pop(0)
+                if next_original_maybe_write is None:
+                    self.assertEqual(combined_length, small_writes_length)
+                    break
+                else:
+                    small_writes_length += len(next_original_maybe_write)
+                    if small_writes_length > aggregate.MAX_BUFFER_SIZE:
+                        self.assertEqual(combined_length, small_writes_length)
+                        break
diff --git a/src/twisted/test/iosim.py b/src/twisted/test/iosim.py
index 04c909324d..7e14e9a0c3 100644
--- a/src/twisted/test/iosim.py
+++ b/src/twisted/test/iosim.py
@@ -6,7 +6,6 @@
 Utilities and helpers for simulating a network
 """
 
-
 import itertools
 
 try:
@@ -284,22 +283,25 @@ class IOPump:
     Perhaps this is a utility worthy of being in protocol.py?
     """
 
-    def __init__(self, client, server, clientIO, serverIO, debug):
+    def __init__(self, client, server, clientIO, serverIO, debug, clock=None):
         self.client = client
         self.server = server
         self.clientIO = clientIO
         self.serverIO = serverIO
         self.debug = debug
+        if clock is None:
+            clock = MemoryReactorClock()
+        self.clock = clock
 
-    def flush(self, debug=False):
+    def flush(self, debug=False, advanceClock=True):
         """
         Pump until there is no more input or output.
 
         Returns whether any data was moved.
         """
         result = False
-        for x in range(1000):
-            if self.pump(debug):
+        for _ in range(1000):
+            if self.pump(debug, advanceClock):
                 result = True
             else:
                 break
@@ -307,12 +309,15 @@ class IOPump:
             assert 0, "Too long"
         return result
 
-    def pump(self, debug=False):
+    def pump(self, debug=False, advanceClock=True):
         """
-        Move data back and forth.
+        Move data back and forth, while also triggering any currently pending
+        scheduled calls (i.e. C{callLater(0, f)}).
 
         Returns whether any data was moved.
         """
+        if advanceClock:
+            self.clock.advance(0)
         if self.debug or debug:
             print("-- GLUG --")
         sData = self.serverIO.getOutBuffer()
@@ -356,6 +361,7 @@ def connect(
     clientTransport,
     debug=False,
     greet=True,
+    clock=None,
 ):
     """
     Create a new L{IOPump} connecting two protocols.
@@ -383,6 +389,9 @@ def connect(
         post-server-greeting state?
     @type greet: L{bool}
 
+    @param clock: An optional L{Clock}. Pumping the resulting L{IOPump} will
+        also increase clock time by a small increment.
+
     @return: An L{IOPump} which connects C{serverProtocol} and
         C{clientProtocol} and delivers bytes between them when it is pumped.
     @rtype: L{IOPump}
@@ -390,7 +399,12 @@ def connect(
     serverProtocol.makeConnection(serverTransport)
     clientProtocol.makeConnection(clientTransport)
     pump = IOPump(
-        clientProtocol, serverProtocol, clientTransport, serverTransport, debug
+        clientProtocol,
+        serverProtocol,
+        clientTransport,
+        serverTransport,
+        debug,
+        clock=clock,
     )
     if greet:
         # Kick off server greeting, etc
@@ -405,6 +419,7 @@ def connectedServerAndClient(
     serverTransportFactory=makeFakeServer,
     debug=False,
     greet=True,
+    clock=None,
 ):
     """
     Connect a given server and client class to each other.
@@ -435,6 +450,9 @@ def connectedServerAndClient(
         post-server-greeting state?
     @type greet: L{bool}
 
+    @param clock: An optional L{Clock}. Pumping the resulting L{IOPump} will
+        also increase clock time by a small increment.
+
     @return: the client protocol, the server protocol, and an L{IOPump} which,
         when its C{pump} and C{flush} methods are called, will move data
         between the created client and server protocol instances.
@@ -444,7 +462,7 @@ def connectedServerAndClient(
     s = ServerClass()
     cio = clientTransportFactory(c)
     sio = serverTransportFactory(s)
-    return c, s, connect(s, sio, c, cio, debug, greet)
+    return c, s, connect(s, sio, c, cio, debug, greet, clock=clock)
 
 
 def _factoriesShouldConnect(clientInfo, serverInfo):
diff --git a/src/twisted/test/test_iosim.py b/src/twisted/test/test_iosim.py
index 27abfbded3..9f8793eaa4 100644
--- a/src/twisted/test/test_iosim.py
+++ b/src/twisted/test/test_iosim.py
@@ -12,7 +12,8 @@ from zope.interface import implementer
 
 from twisted.internet.interfaces import IPushProducer
 from twisted.internet.protocol import Protocol
-from twisted.test.iosim import FakeTransport, connect
+from twisted.internet.task import Clock
+from twisted.test.iosim import FakeTransport, connect, connectedServerAndClient
 from twisted.trial.unittest import TestCase
 
 
@@ -299,3 +300,15 @@ class IOPumpTests(TestCase):
         (stream producer) registered with the client transport.
         """
         self._testStreamingProducer(mode="client")
+
+    def test_timeAdvances(self) -> None:
+        """
+        L{IOPump.pump} advances time in the given L{Clock}.
+        """
+        time_passed = []
+        clock = Clock()
+        _, _, pump = connectedServerAndClient(Protocol, Protocol, clock=clock)
+        clock.callLater(0, lambda: time_passed.append(True))
+        self.assertFalse(time_passed)
+        pump.pump()
+        self.assertTrue(time_passed)
diff --git a/src/twisted/test/test_sslverify.py b/src/twisted/test/test_sslverify.py
index 2aa4d6814a..8f59fa837a 100644
--- a/src/twisted/test/test_sslverify.py
+++ b/src/twisted/test/test_sslverify.py
@@ -19,6 +19,7 @@ from incremental import Version
 from twisted.internet import defer, interfaces, protocol, reactor
 from twisted.internet._idna import _idnaText
 from twisted.internet.error import CertificateError, ConnectionClosed, ConnectionLost
+from twisted.internet.task import Clock
 from twisted.python.compat import nativeString
 from twisted.python.filepath import FilePath
 from twisted.python.modules import getModule
@@ -288,17 +289,21 @@ def _loopbackTLSConnection(serverOpts, clientOpts):
     plainServerFactory = protocol.Factory()
     plainServerFactory.protocol = lambda: serverWrappedProto
 
+    clock = Clock()
     clientFactory = TLSMemoryBIOFactory(
-        clientOpts, isClient=True, wrappedFactory=plainServerFactory
+        clientOpts, isClient=True, wrappedFactory=plainServerFactory, clock=clock
     )
     serverFactory = TLSMemoryBIOFactory(
-        serverOpts, isClient=False, wrappedFactory=plainClientFactory
+        serverOpts, isClient=False, wrappedFactory=plainClientFactory, clock=clock
     )
 
     sProto, cProto, pump = connectedServerAndClient(
         lambda: serverFactory.buildProtocol(None),
         lambda: clientFactory.buildProtocol(None),
+        clock=clock,
     )
+    pump.flush()
+
     return sProto, cProto, serverWrappedProto, clientWrappedProto, pump
 
 
@@ -2065,17 +2070,21 @@ class ServiceIdentityTests(SynchronousTestCase):
         self.serverOpts = serverOpts
         self.clientOpts = clientOpts
 
+        clock = Clock()
         clientTLSFactory = TLSMemoryBIOFactory(
-            clientOpts, isClient=True, wrappedFactory=clientFactory
+            clientOpts, isClient=True, wrappedFactory=clientFactory, clock=clock
         )
         serverTLSFactory = TLSMemoryBIOFactory(
-            serverOpts, isClient=False, wrappedFactory=serverFactory
+            serverOpts, isClient=False, wrappedFactory=serverFactory, clock=clock
         )
 
         cProto, sProto, pump = connectedServerAndClient(
             lambda: serverTLSFactory.buildProtocol(None),
             lambda: clientTLSFactory.buildProtocol(None),
+            clock=clock,
         )
+        pump.flush()
+
         return cProto, sProto, clientWrappedProto, serverWrappedProto, pump
 
     def test_invalidHostname(self):
diff --git a/src/twisted/web/test/test_agent.py b/src/twisted/web/test/test_agent.py
index f143fc0916..9cee5b75a0 100644
--- a/src/twisted/web/test/test_agent.py
+++ b/src/twisted/web/test/test_agent.py
@@ -95,6 +95,7 @@ else:
     sslPresent = True
     from twisted.internet._sslverify import ClientTLSOptions, IOpenSSLTrustRoot
     from twisted.internet.ssl import optionsForClientTLS
+    from twisted.protocols import tls
     from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol
 
     @implementer(IOpenSSLTrustRoot)
@@ -833,7 +834,7 @@ class IntegrationTestingMixin:
         hostName,
         expectedAddress,
         addressType,
-        serverWrapper=lambda server: server,
+        serverWrapper=lambda server, _: server,
         createAgent=client.Agent,
         scheme=b"http",
     ):
@@ -851,9 +852,9 @@ class IntegrationTestingMixin:
         @param addressType: The class to construct an address out of.
         @type addressType: L{type}
 
-        @param serverWrapper: A callable that takes a protocol factory and
-            returns a protocol factory; used to wrap the server / responder
-            side in a TLS server.
+        @param serverWrapper: A callable that takes a protocol factory and a
+            ``Clock`` and returns a protocol factory; used to wrap the server /
+            responder side in a TLS server.
         @type serverWrapper:
             serverWrapper(L{twisted.internet.interfaces.IProtocolFactory}) ->
             L{twisted.internet.interfaces.IProtocolFactory}
@@ -867,6 +868,10 @@ class IntegrationTestingMixin:
         @type scheme: L{bytes}
         """
         reactor = self.createReactor()
+        if sslPresent:
+            # We have no way to tell the client to use our test reactor so we
+            # have to patch it.
+            self.patch(tls, "_get_default_clock", lambda: reactor)
         agent = createAgent(reactor)
         deferred = agent.request(b"GET", scheme + b"://" + hostName + b"/")
         host, port, factory, timeout, bind = reactor.tcpClients[0]
@@ -884,10 +889,17 @@ class IntegrationTestingMixin:
 
         accumulator.currentProtocol = None
         accumulator.protocolConnectionMade = None
-        wrapper = serverWrapper(accumulator).buildProtocol(None)
+        wrapper = serverWrapper(accumulator, reactor).buildProtocol(None)
         serverTransport = FakeTransport(wrapper, True)
         wrapper.makeConnection(serverTransport)
-        pump = IOPump(clientProtocol, wrapper, clientTransport, serverTransport, False)
+        pump = IOPump(
+            clientProtocol,
+            wrapper,
+            clientTransport,
+            serverTransport,
+            False,
+            clock=reactor,
+        )
         pump.flush()
         self.assertNoResult(deferred)
         lines = accumulator.currentProtocol.data.split(b"\r\n")
@@ -1570,8 +1582,8 @@ class AgentHTTPSTests(TestCase, FakeReactorAndConnectMixin, IntegrationTestingMi
             certHostName.decode("ascii")
         )
 
-        def tlsify(serverFactory):
-            return TLSMemoryBIOFactory(server.options(), False, serverFactory)
+        def tlsify(serverFactory, reactor):
+            return TLSMemoryBIOFactory(server.options(), False, serverFactory, reactor)
 
         def tlsagent(reactor):
             from zope.interface import implementer