gh-135444: fix DatagramTransport buffer_size accounting

Commit 73e8637 added 8 to the
buffer_size when send could not be called right away.  However, it did
not complete this accounting by removing 8 from the buffer size when
sending did finally complete.

Author: jsbronder

Lib/asyncio/proactor_events.py

@@ -526,7 +526,7 @@ def _loop_writing(self, fut=None):
                 return
 
             data, addr = self._buffer.popleft()
-            self._buffer_size -= len(data)
+            self._buffer_size -= len(data) + 8 # include header bytes
             if self._address is not None:
                 self._write_fut = self._loop._proactor.send(self._sock,
                                                             data)

Lib/asyncio/selector_events.py

@@ -1291,15 +1291,15 @@ def sendto(self, data, addr=None):
     def _sendto_ready(self):
         while self._buffer:
             data, addr = self._buffer.popleft()
-            self._buffer_size -= len(data)
+            self._buffer_size -= len(data) + 8 # include header bytes
             try:
                 if self._extra['peername']:
                     self._sock.send(data)
                 else:
                     self._sock.sendto(data, addr)
             except (BlockingIOError, InterruptedError):
                 self._buffer.appendleft((data, addr))  # Try again later.
-                self._buffer_size += len(data)
+                self._buffer_size += len(data) + 8 # include header bytes
                 break
             except OSError as exc:
                 self._protocol.error_received(exc)

Lib/test/test_asyncio/test_proactor_events.py

@@ -566,6 +566,8 @@ def test_sendto(self):
         self.assertTrue(self.proactor.sendto.called)
         self.proactor.sendto.assert_called_with(
             self.sock, data, addr=('0.0.0.0', 1234))
+        self.assertFalse(transport._buffer)
+        self.assertEqual(0, transport._buffer_size)
 
     def test_sendto_bytearray(self):
         data = bytearray(b'data')

Lib/test/test_asyncio/test_selector_events.py

@@ -1497,6 +1497,47 @@ def test_sendto_closing(self):
         transport.sendto(b'data', (1,))
         self.assertEqual(transport._conn_lost, 2)
 
+    def test_sendto_sendto_ready(self):
+        data = b'data'
+
+        # First queue up a buffer by having the socket block
+        self.sock.sendto.side_effect = BlockingIOError
+        transport = self.datagram_transport()
+        transport.sendto(data, ('0.0.0.0', 12345))
+        self.loop.assert_writer(7, transport._sendto_ready)
+        self.assertEqual(1, len(transport._buffer))
+        self.assertEqual(transport._buffer_size, len(data) + 8)
+
+        # Now let the socket send the buffer
+        self.sock.sendto.side_effect = None
+        transport._sendto_ready()
+        self.assertTrue(self.sock.sendto.called)
+        self.assertEqual(
+            self.sock.sendto.call_args[0], (data, ('0.0.0.0', 12345)))
+        self.assertFalse(self.loop.writers)
+        self.assertFalse(transport._buffer)
+        self.assertEqual(transport._buffer_size, 0)
+
+    def test_sendto_sendto_ready_blocked(self):
+        data = b'data'
+
+        # First queue up a buffer by having the socket block
+        self.sock.sendto.side_effect = BlockingIOError
+        transport = self.datagram_transport()
+        transport.sendto(data, ('0.0.0.0', 12345))
+        self.loop.assert_writer(7, transport._sendto_ready)
+        self.assertEqual(1, len(transport._buffer))
+        self.assertEqual(transport._buffer_size, len(data) + 8)
+
+        # Now try to send the buffer and let it get requeued
+        transport._sendto_ready()
+        self.assertTrue(self.sock.sendto.called)
+        self.assertEqual(
+            self.sock.sendto.call_args[0], (data, ('0.0.0.0', 12345)))
+        self.assertTrue(self.loop.writers)
+        self.assertEqual(1, len(transport._buffer))
+        self.assertEqual(transport._buffer_size, len(data) + 8)
+
     def test_sendto_ready(self):
         data = b'data'
         self.sock.sendto.return_value = len(data)