From 92093efd46d65e65f70951174d75c03454e647bc Mon Sep 17 00:00:00 2001 From: pctablet505 Date: Thu, 16 Jul 2026 21:58:54 +0530 Subject: [PATCH] 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 --- newsfragments/1723.feature.rst | 4 ++++ src/trio/_channel.py | 13 +++++++++++++ src/trio/_tests/test_channel.py | 7 +++++++ 3 files changed, 24 insertions(+) create mode 100644 newsfragments/1723.feature.rst diff --git a/newsfragments/1723.feature.rst b/newsfragments/1723.feature.rst new file mode 100644 index 0000000000..6826723b96 --- /dev/null +++ b/newsfragments/1723.feature.rst @@ -0,0 +1,4 @@ +`MemoryChannelStatistics` now has a ``max_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. diff --git a/src/trio/_channel.py b/src/trio/_channel.py index 8cf435f4e5..02fcff9b01 100644 --- a/src/trio/_channel.py +++ b/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). + * ``max_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.""" + max_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 + max_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), + max_buffer_used=self.max_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.max_buffer_used = max( + self._state.max_buffer_used, + len(self._state.data), + ) else: raise trio.WouldBlock diff --git a/src/trio/_tests/test_channel.py b/src/trio/_tests/test_channel.py index c8b2efddc6..9e41335815 100644 --- a/src/trio/_tests/test_channel.py +++ b/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.max_buffer_used == 0 s.send_nowait(None) assert s.statistics().current_buffer_used == 1 + assert s.statistics().max_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().max_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().max_buffer_used == 2 + async with trio.open_nursery() as nursery: nursery.start_soon(r.receive) await wait_all_tasks_blocked()