@@ -333,6 +333,9 @@ Write-only Transports
    This method does not block; it buffers the data and arranges for it
    to be sent out asynchronously.
 
+   The transport keeps a reference to *data* rather than copying. *data*
+   must not be mutated while being sent.
+
 .. method:: WriteTransport.writelines(list_of_data)
 
    Write a list (or any iterable) of data bytes to the transport.
 
@@ -1065,6 +1065,9 @@ The :mod:`!socket` module also offers various network-related services:
       for IPv6 multicast addresses, string representing an address will not
       contain ``%scope_id`` part.
 
+   .. versionchanged:: next
+      Null characters are now rejected in the *host* and *port* parameters.
+
 .. _Happy Eyeballs: https://en.wikipedia.org/wiki/Happy_Eyeballs
 
 .. function:: getfqdn([name])
 
@@ -663,14 +663,6 @@ class timedelta:
 
     def __new__(cls, days=0, seconds=0, microseconds=0,
                 milliseconds=0, minutes=0, hours=0, weeks=0):
-        # Doing this efficiently and accurately in C is going to be difficult
-        # and error-prone, due to ubiquitous overflow possibilities, and that
-        # C double doesn't have enough bits of precision to represent
-        # microseconds over 10K years faithfully.  The code here tries to make
-        # explicit where go-fast assumptions can be relied on, in order to
-        # guide the C implementation; it's way more convoluted than speed-
-        # ignoring auto-overflow-to-long idiomatic Python could be.
-
         for name, value in (
             ("days", days),
             ("seconds", seconds),
@@ -685,80 +677,35 @@ def __new__(cls, days=0, seconds=0, microseconds=0,
                     f"unsupported type for timedelta {name} component: {type(value).__name__}"
                 )
 
-        # Final values, all integer.
-        # s and us fit in 32-bit signed ints; d isn't bounded.
-        d = s = us = 0
-
-        # Normalize everything to days, seconds, microseconds.
-        days += weeks*7
-        seconds += minutes*60 + hours*3600
-        microseconds += milliseconds*1000
-
-        # Get rid of all fractions, and normalize s and us.
-        # Take a deep breath <wink>.
-        if isinstance(days, float):
-            dayfrac, days = _math.modf(days)
-            daysecondsfrac, daysecondswhole = _math.modf(dayfrac * (24.*3600.))
-            assert daysecondswhole == int(daysecondswhole)  # can't overflow
-            s = int(daysecondswhole)
-            assert days == int(days)
-            d = int(days)
-        else:
-            daysecondsfrac = 0.0
-            d = days
-        assert isinstance(daysecondsfrac, float)
-        assert abs(daysecondsfrac) <= 1.0
-        assert isinstance(d, int)
-        assert abs(s) <= 24 * 3600
-        # days isn't referenced again before redefinition
-
-        if isinstance(seconds, float):
-            secondsfrac, seconds = _math.modf(seconds)
-            assert seconds == int(seconds)
-            seconds = int(seconds)
-            secondsfrac += daysecondsfrac
-            assert abs(secondsfrac) <= 2.0
-        else:
-            secondsfrac = daysecondsfrac
-        # daysecondsfrac isn't referenced again
-        assert isinstance(secondsfrac, float)
-        assert abs(secondsfrac) <= 2.0
-
-        assert isinstance(seconds, int)
-        days, seconds = divmod(seconds, 24*3600)
-        d += days
-        s += int(seconds)    # can't overflow
-        assert isinstance(s, int)
-        assert abs(s) <= 2 * 24 * 3600
-        # seconds isn't referenced again before redefinition
-
-        usdouble = secondsfrac * 1e6
-        assert abs(usdouble) < 2.1e6    # exact value not critical
-        # secondsfrac isn't referenced again
-
-        if isinstance(microseconds, float):
-            microseconds = round(microseconds + usdouble)
-            seconds, microseconds = divmod(microseconds, 1000000)
-            days, seconds = divmod(seconds, 24*3600)
-            d += days
-            s += seconds
-        else:
-            microseconds = int(microseconds)
-            seconds, microseconds = divmod(microseconds, 1000000)
-            days, seconds = divmod(seconds, 24*3600)
-            d += days
-            s += seconds
-            microseconds = round(microseconds + usdouble)
-        assert isinstance(s, int)
-        assert isinstance(microseconds, int)
-        assert abs(s) <= 3 * 24 * 3600
-        assert abs(microseconds) < 3.1e6
-
-        # Just a little bit of carrying possible for microseconds and seconds.
-        seconds, us = divmod(microseconds, 1000000)
-        s += seconds
-        days, s = divmod(s, 24*3600)
-        d += days
+        # Sum the exact number of microseconds, like the C implementation:
+        # the integer part of each float is converted exactly, and only the
+        # fractional microseconds left over are added up as a float and
+        # rounded (half to even) at the end.
+        us = 0
+        leftover = 0.0
+        for value, factor in (
+            (microseconds, 1),
+            (milliseconds, 1000),
+            (seconds, 1000000),
+            (minutes, 60*1000000),
+            (hours, 3600*1000000),
+            (days, 24*3600*1000000),
+            (weeks, 7*24*3600*1000000)
+        ):
+            if isinstance(value, float):
+                frac, whole = _math.modf(value)
+                us += int(whole) * factor
+                frac, whole = _math.modf(frac * factor)
+                us += int(whole)
+                leftover += frac
+            else:
+                us += int(value * factor)
+        if leftover:
+            n, m = leftover.as_integer_ratio()
+            us = _divide_and_round(us * m + n, m)
+
+        s, us = divmod(us, 1000000)
+        d, s = divmod(s, 24*3600)
 
         assert isinstance(d, int)
         assert isinstance(s, int) and 0 <= s < 24*3600
 
@@ -1099,7 +1099,9 @@ def write(self, data):
             # Not all was written; register write handler.
             self._add_writer(self._sock_fd, self._write_ready)
 
-        # Add it to the buffer.
+        # Add it to the buffer as a view, so partial sends don't copy it.
+        if not isinstance(data, memoryview):
+            data = memoryview(data)
         self._buffer.append(data)
         self._buffer_size += len(data)
         self._maybe_pause_protocol()
 
@@ -308,6 +308,9 @@ def __init__(self, loop, app_protocol, sslcontext, waiter,
         # App data write buffering
         self._write_backlog = collections.deque()
         self._write_buffer_size = 0
+        # How much of self._outgoing was produced by _do_write(): encrypted
+        # app data, plus anything else OpenSSL flushed along with it
+        self._outgoing_app_size = 0
 
         self._waiter = waiter
         self._loop = loop
@@ -720,6 +723,7 @@ def _write_appdata(self, list_of_data):
             self._fatal_error(ex, 'Fatal error on SSL protocol')
 
     def _do_write(self):
+        pending = self._outgoing.pending
         try:
             while self._write_backlog:
                 data = self._write_backlog[0]
@@ -733,11 +737,13 @@ def _do_write(self):
                     self._write_buffer_size -= data_len
         except SSLAgainErrors:
             pass
+        self._outgoing_app_size += self._outgoing.pending - pending
         self._process_outgoing()
 
     def _process_outgoing(self):
         if not self._ssl_writing_paused:
             data = self._outgoing.read()
+            self._outgoing_app_size = 0
             if len(data):
                 self._transport.write(data)
         self._control_app_writing()
@@ -904,7 +910,15 @@ def resume():
     # Flow control for reads from SSL socket
 
     def _control_ssl_reading(self):
-        size = self._get_read_buffer_size()
+        # Also count the records OpenSSL writes on its own while reading
+        # (alerts, KeyUpdate replies, ...) which the peer does not read:
+        # the app is not involved in producing them, so pausing it does
+        # not bound them; only pausing reads from the peer does.  Whatever
+        # was produced while writing app data (_outgoing_app_size) is left
+        # out so that two peers which both write before they read cannot
+        # deadlock.
+        size = (self._get_read_buffer_size() +
+                self._outgoing.pending - self._outgoing_app_size)
         if size >= self._incoming_high_water and not self._ssl_reading_paused:
             self._ssl_reading_paused = True
             self._transport.pause_reading()
@@ -937,6 +951,7 @@ def resume_writing(self):
         assert self._ssl_writing_paused
         self._ssl_writing_paused = False
         self._process_outgoing()
+        self._control_ssl_reading()
 
     def _fatal_error(self, exc, message='Fatal error on transport'):
         if self._transport:
 
@@ -554,6 +554,13 @@ def test_constructor(self):
         eq(td(seconds=0.001), td(milliseconds=1))
         eq(td(milliseconds=0.001), td(microseconds=1))
 
+        # Large float args keep their sub-second part
+        eq(td(weeks=12345678.9), td(86419752, 25920, 225))
+        eq(td(hours=123456789.1), td(5144032, 75959, 999979))
+        eq(td(minutes=1234567890.1), td(857338, 70205, 999994))
+        eq(td(days=1.1, weeks=1000000.1), td(7000001, 69119, 999986))
+        eq(td(seconds=0.5, microseconds=2.0**60), td(13343998, 77407, 346976))
+
         # Check type of args to constructor
         ra(TypeError, lambda: td(weeks='1'))
         ra(TypeError, lambda: td(days='1'))
 
@@ -855,14 +855,23 @@ def test_write_partial_none(self):
         self.assertEqual(list_to_buffer([b'data']), transport._buffer)
 
     def test_write_tryagain(self):
-        self.sock.send.side_effect = BlockingIOError
-
         data = b'data'
-        transport = self.socket_transport()
-        transport.write(data)
-
-        self.loop.assert_writer(7, transport._write_ready)
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
+        for buf in data, bytearray(data), memoryview(data):
+            with self.subTest(type=type(buf).__name__):
+                self.sock.send.side_effect = BlockingIOError
+
+                transport = self.socket_transport()
+                transport.write(buf)
+
+                self.loop.assert_writer(7, transport._write_ready)
+                self.assertEqual(list_to_buffer([b'data']), transport._buffer)
+                # gh-159503: the data is buffered as a view, not a copy
+                self.assertIsInstance(transport._buffer[0], memoryview)
+                self.assertIs(transport._buffer[0].obj, memoryview(buf).obj)
+                if isinstance(buf, bytearray):
+                    # The buffered view pins the bytearray against resizing.
+                    with self.assertRaises(BufferError):
+                        buf.append(0)
 
     def test_write_sendmsg_no_data(self):
         self.sock.sendmsg = mock.Mock()
 
@@ -289,6 +289,46 @@ def test_write_after_closing(self):
         # should not raise
         self.assertIsNone(transp.write(b'data'))
 
+    def test_pause_reading_if_ssl_replies_are_not_read(self):
+        # Output which OpenSSL produces on its own while reading (alerts,
+        # KeyUpdate, ...) must not pile up if the peer does not read it.
+        ssl_proto = self.ssl_protocol()
+        transport = self.connection_made(ssl_proto, do_handshake=mock.Mock())
+        reply = b'x' * (ssl_proto._incoming_high_water // 2)
+
+        def read(*args):
+            ssl_proto._outgoing.write(reply)
+            raise ssl.SSLWantReadError
+        ssl_proto._sslobj.read.side_effect = read
+
+        ssl_proto.pause_writing()
+        ssl_proto.buffer_updated(0)
+        transport.pause_reading.assert_not_called()
+        ssl_proto.buffer_updated(0)
+        transport.pause_reading.assert_called_once_with()
+        transport.resume_reading.assert_not_called()
+        transport.write.assert_not_called()
+
+        ssl_proto.resume_writing()
+        transport.write.assert_called_once_with(reply * 2)
+        transport.resume_reading.assert_called_once_with()
+
+    def test_no_pause_reading_if_app_data_is_not_read(self):
+        # Peers which both write a lot before they read must not deadlock.
+        ssl_proto = self.ssl_protocol()
+        transport = self.connection_made(ssl_proto, do_handshake=mock.Mock())
+        ssl_proto._sslobj.write.side_effect = ssl_proto._outgoing.write
+        data = b'x' * ssl_proto._incoming_high_water
+
+        ssl_proto.pause_writing()
+        ssl_proto._app_transport.write(data)
+        ssl_proto.buffer_updated(0)
+        transport.pause_reading.assert_not_called()
+
+        ssl_proto.resume_writing()
+        transport.write.assert_called_once_with(data)
+        transport.resume_reading.assert_not_called()
+
 
 ##############################################################################
 # Start TLS Tests
 
@@ -416,6 +416,52 @@ async def notmuch():
         self.assertEqual(t.get_name(), '123456789')
         self.loop.run_until_complete(t)
 
+    # The free-threaded GC publishes allocation counts in batches, so the
+    # collection cannot be scheduled to run inside get_name().
+    @support.requires_gil_enabled("needs precise GC scheduling")
+    def test_task_get_name_reentered_during_formatting(self):
+        # gh-158159: formatting the lazy default name can run the GC, and a
+        # finalizer may call get_name() on the same task, replacing the name
+        # the outer call is still formatting.
+        loop = self.loop
+        observed = {}
+        keep = []
+        task = None
+
+        class Reenter:
+            def __init__(self):
+                self.cycle = self
+
+            def __del__(self):
+                observed['inner'] = task.get_name()
+                # Reuse the memory of a counter that was freed too early.
+                keep.extend(10**9 + i for i in range(8))
+
+        async def notmuch():
+            pass
+
+        async def scenario():
+            gc.collect()
+            Reenter()
+            old_threshold = gc.get_threshold()
+            gc.set_threshold(1)
+            try:
+                scratch = [[0], [0], [0]]  # schedule a collection
+                observed['outer'] = task.get_name()
+            finally:
+                gc.set_threshold(*old_threshold)
+            del scratch
+
+        # Names are only freeable once the counter is past the small ints.
+        for _ in range(1100):
+            loop.run_until_complete(self.new_task(loop, notmuch(), name=None))
+        task = self.new_task(loop, scenario(), name=None)
+        loop.run_until_complete(task)
+
+        self.assertRegex(observed['outer'], r'^Task-\d+$')
+        self.assertEqual(observed['outer'], observed['inner'])
+        self.assertEqual(task.get_name(), observed['outer'])
+
     def test_task_repr_name_not_str(self):
         async def notmuch():
             return 123
 
@@ -984,6 +984,21 @@ def __index__(self):
             finally:
                 size.result.release()
 
+        # Sizes above 2**31 - 1 must work where C long is 32-bit.
+        if sys.maxsize > 2**31:
+            stream = self.ioclass(b'abc')
+            self.assertEqual(stream.truncate(2**31), 2**31)
+            self.assertEqual(stream.getvalue(), b'abc')
+
+    def test_write_overflow(self):
+        # gh-160781: overallocation at a position near sys.maxsize
+        # must raise OverflowError, not SystemError.
+        memio = self.ioclass()
+        memio.seek(sys.maxsize - 1)
+        self.assertRaises(OverflowError, memio.write, b"x")
+        self.assertEqual(memio.getvalue(), b"")
+        self.assertEqual(memio.tell(), sys.maxsize - 1)
+
     def test_readinto_overlapping_buffer(self):
         with self.ioclass(b'abcd') as stream:
             with stream.getbuffer()[1:] as target:
@@ -1150,6 +1165,12 @@ def __index__(self):
         with self.assertRaises(ValueError):
             stream.truncate(Size())
 
+        # Sizes above 2**31 - 1 must work where C long is 32-bit.
+        if sys.maxsize > 2**31:
+            stream = self.ioclass("abc")
+            self.assertEqual(stream.truncate(2**31), 2**31)
+            self.assertEqual(stream.getvalue(), "abc")
+
     # XXX: For the Python version of io.StringIO, this is highly
     # dependent on the encoding used for the underlying buffer.
     def test_widechar(self):