Track max buffer usage in memory channel statistics (#3474)

* Track max buffer usage in memory channel statistics

Add a max_buffer_used field to MemoryChannelStatistics that reports
the largest number of items the channel's buffer has held at once
since it was created. This lets users pick a sensible
max_buffer_size based on data from a real run instead of guessing.

Closes #1723

* Rename max_buffer_used to peak_buffer_used per review feedback

max_buffer_used read as a configured limit rather than a runtime metric,
easily confused with max_buffer_size. peak_buffer_used pairs more clearly
with the existing current_buffer_used field.

Author: pctablet505

newsfragments/1723.feature.rst

@@ -0,0 +1,4 @@
+`MemoryChannelStatistics` now has a ``peak_buffer_used`` field, which reports
+the largest number of items that have ever been in the channel's buffer at
+once. This makes it easier to pick a sensible ``max_buffer_size`` based on
+data from a real run, instead of guessing.

src/trio/_channel.py

@@ -96,6 +96,8 @@ class open_memory_channel(tuple["MemorySendChannel[T]", "MemoryReceiveChannel[T]
       channel (summing over all clones).
     * ``tasks_waiting_receive``: The number of tasks blocked in ``receive`` on
       this channel (summing over all clones).
+    * ``peak_buffer_used``: The largest number of items that have been in the
+      buffer at once since the channel was created.
     """
 
     def __new__(  # type: ignore[misc]  # "must return a subtype"
@@ -142,6 +144,10 @@ class MemoryChannelStatistics:
     tasks_waiting_receive: int
     """The number of tasks currently blocked waiting to receive."""
 
+    peak_buffer_used: int
+    """The largest number of items that have been in the buffer at once
+    since the channel was created."""
+
 
 @attrs.define
 class MemoryChannelState(Generic[T]):
@@ -154,6 +160,8 @@ class MemoryChannelState(Generic[T]):
     send_tasks: OrderedDict[Task, T] = attrs.Factory(OrderedDict)
     # {task: None}
     receive_tasks: OrderedDict[Task, None] = attrs.Factory(OrderedDict)
+    # The largest len(self.data) has ever been
+    peak_buffer_used: int = 0
 
     def statistics(self) -> MemoryChannelStatistics:
         return MemoryChannelStatistics(
@@ -163,6 +171,7 @@ def statistics(self) -> MemoryChannelStatistics:
             open_receive_channels=self.open_receive_channels,
             tasks_waiting_send=len(self.send_tasks),
             tasks_waiting_receive=len(self.receive_tasks),
+            peak_buffer_used=self.peak_buffer_used,
         )
 
 
@@ -211,6 +220,10 @@ def send_nowait(self, value: SendType) -> None:
             trio.lowlevel.reschedule(task, Value(value))
         elif len(self._state.data) < self._state.max_buffer_size:
             self._state.data.append(value)
+            self._state.peak_buffer_used = max(
+                self._state.peak_buffer_used,
+                len(self._state.data),
+            )
         else:
             raise trio.WouldBlock
 

src/trio/_tests/test_channel.py

@@ -320,9 +320,11 @@ async def test_statistics() -> None:
     assert stats.open_receive_channels == 1
     assert stats.tasks_waiting_send == 0
     assert stats.tasks_waiting_receive == 0
+    assert stats.peak_buffer_used == 0
 
     s.send_nowait(None)
     assert s.statistics().current_buffer_used == 1
+    assert s.statistics().peak_buffer_used == 1
 
     s2 = s.clone()
     assert s.statistics().open_send_channels == 2
@@ -337,6 +339,7 @@ async def test_statistics() -> None:
     async with trio.open_nursery() as nursery:
         s2.send_nowait(None)  # fill up the buffer
         assert s.statistics().current_buffer_used == 2
+        assert s.statistics().peak_buffer_used == 2
         nursery.start_soon(s2.send, None)
         nursery.start_soon(s2.send, None)
         await wait_all_tasks_blocked()
@@ -351,6 +354,10 @@ async def test_statistics() -> None:
     except trio.WouldBlock:
         pass
 
+    # draining the buffer doesn't reset the high-water mark
+    assert s.statistics().current_buffer_used == 0
+    assert s.statistics().peak_buffer_used == 2
+
     async with trio.open_nursery() as nursery:
         nursery.start_soon(r.receive)
         await wait_all_tasks_blocked()