Improve documentation of the `frequenz.channels.experimental.Pipe` #407

ela-kotulska-frequenz · 2025-04-08T15:21:13Z

Documentation shows how to create Pipe inside context manager to cleanup the resources when context is exited.

Copilot

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

Comments suppressed due to low confidence (1)

src/frequenz/channels/experimental/_pipe.py:33

The code snippet references 'Receiver[int]' without importing 'Receiver'. Consider adding the appropriate import (e.g., from frequenz.channels import Receiver) to avoid confusion.

async def create_forwarding_pipe(receiver: Receiver[int]) -> AsyncIterator[Broadcast[int]]:

llucax

All comments are minor, I think the example is good enough to go as it is, just ideas of possible improvements.

llucax · 2025-04-09T07:24:37Z

src/frequenz/channels/experimental/_pipe.py

+        from frequenz.channels import Broadcast, Pipe, Receiver
+
+        @asynccontextmanager
+        async def create_forwarding_pipe(


Nitpick: I would try to use the pattern used by the Python library, like closing. Maybe call this forwarding_channel (as you are also returning the channel, not the pipe)?

Then:

async with forwarding_channel( source_channel.new_receiver() ) as destination_channel: receiver = destination_channel.new_receiver() await source_sender.send(10) assert await receiver.receive() == 10

llucax · 2025-04-09T07:26:32Z

src/frequenz/channels/experimental/_pipe.py

+        async with create_forwarding_pipe(
+            source_channel.new_receiver()
+        ) as forwarding_channel:
+            receiver = forwarding_channel.new_receiver()


I guess here you don't close the receiver because you are closing the channel? Maybe this needs a note because it might not be obvious? I also wonder if, at least for this example, it wouldn't make more sense to return a new receiver instead of the whole channel, and explicitly close the receiver in the async context manager, to make it more clear/explicit.

I didn't close it because it goes with with not async with and I simply forgot...

llucax · 2025-04-09T07:29:34Z

RELEASE_NOTES.md

 ## Upgrading

 - Some minimal dependencies have been bumped, so you might need to adjust your dependencies accordingly.
+- Improve documentation of the frequenz.channels.experimental.Pipe


This is not really an upgrade step, so I would put it into New Features (even if it is not a new feature either :D, sometimes I create a new section # Enhancements in these cases, but I think it can also be left out of the release notes, as I don't think anyone will go check out the Pipe docs just because it is in the changelog).

llucax · 2025-04-09T07:48:47Z

If we merge this, we can even simplify the example further and only use async with aclosing(...) 😉

Rename close() to aclose() in Broadcast and Anycast #408

ela-kotulska-frequenz · 2025-04-16T08:33:18Z

Ok prepared 2 examples:

with asynccontextmanager and all objects properly closed:
Example:

        from contextlib import asynccontextmanager, closing, aclosing
        from typing import AsyncIterator
        from frequenz.channels import Broadcast, Pipe, Receiver

        @asynccontextmanager
        async def new_forwarding_channel(
            receiver: Receiver[int],
        ) -> AsyncIterator[Broadcast[int]]:
            forwarding_channel: Broadcast[int] = Broadcast(name="forwarded channel")
            forwarding_sender = forwarding_channel.new_sender()
            try:
                async with Pipe(receiver, forwarding_sender):
                    yield forwarding_channel
            finally:
                receiver.close()
                await forwarding_channel.aclose()

        async with (
            aclosing(Broadcast(name="source channel")) as source_channel,
            new_forwarding_channel(source_channel.new_receiver()) as forwarding_channel,
            ):
            with closing(forwarding_channel.new_receiver()) as receiver:
                source_sender = source_channel.new_sender()
                await source_sender.send(10)
                assert await receiver.receive() == 10

Without asynccontextmanager and all object properly closed

        from contextlib import closing, aclosing
        from frequenz.channels import Broadcast, Pipe, Receiver

        async with aclosing(Broadcast(name="source channel")) as source_channel:
            with closing(source_channel.new_receiver()) as source_receiver:
                async with (
                    aclosing(Broadcast(name="forwarded channel")) as forwarding_channel,
                    Pipe(source_receiver, forwarding_channel.new_sender())
                    ):
                    with closing(forwarding_channel.new_receiver()) as receiver:
                        source_sender = source_channel.new_sender()
                        await source_sender.send(10)
                        assert await receiver.receive() == 10

The first one has more lines, but took me much less time to write and understand

ela-kotulska-frequenz · 2025-04-16T09:15:53Z

It is a new version with AsyncExitStack :)

Copilot

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

Comments suppressed due to low confidence (1)

src/frequenz/channels/experimental/_pipe.py:38

Consider using 'aclosing' instead of 'closing' when entering the async context for 'new_receiver()' (if it is an asynchronous context manager) to ensure proper async cleanup.

source_receiver = await stack.enter_async_context(closing(source_channel.new_receiver()))

src/frequenz/channels/experimental/_pipe.py

llucax · 2025-04-16T09:30:42Z

src/frequenz/channels/experimental/_pipe.py

+            source_receiver = await stack.enter_async_context(
+                closing(source_channel.new_receiver()))


Does this really works? AsycnExitStack() also provides the methods provided by ExitStack, so you should probably use enter_context() here that the context manager is sync.

Suggested change

source_receiver = await stack.enter_async_context(

closing(source_channel.new_receiver()))

source_receiver = await stack.enter_context(

closing(source_channel.new_receiver()))

Yea sorry I assumed the code in Example is executed. (because they are showed in the pytest summary as tests.
I just found out only imports are checked. Fixed

pylint is also ran, but the code itself isn't, nor mypy or other lint tools.

llucax · 2025-04-16T09:31:07Z

src/frequenz/channels/experimental/_pipe.py

+                aclosing(Broadcast(name="forwarding channel")))
+            await stack.enter_async_context(Pipe(source_receiver, forwarding_channel.new_sender()))
+
+            receiver = await stack.enter_async_context(closing(forwarding_channel.new_receiver()))


And here.

Suggested change

receiver = await stack.enter_async_context(closing(forwarding_channel.new_receiver()))

receiver = await stack.enter_context(closing(forwarding_channel.new_receiver()))

llucax

Thanks for taking care of this!

Documentation shows how to create Pipe inside context manager to cleanup the resources when context is exited. Signed-off-by: Elzbieta Kotulska <[email protected]>

ela-kotulska-frequenz added the part:docs Affects the documentation label Apr 8, 2025

ela-kotulska-frequenz self-assigned this Apr 8, 2025

Copilot AI review requested due to automatic review settings April 8, 2025 15:21

ela-kotulska-frequenz requested a review from a team as a code owner April 8, 2025 15:21

ela-kotulska-frequenz requested a review from daniel-zullo-frequenz April 8, 2025 15:21

github-actions bot added the part:experimental Affects the experimental package label Apr 8, 2025

Copilot AI reviewed Apr 8, 2025

View reviewed changes

ela-kotulska-frequenz force-pushed the pipe_doc branch 2 times, most recently from 66559d7 to aca4789 Compare April 8, 2025 15:22

ela-kotulska-frequenz requested review from cyiallou and shsms and removed request for cyiallou April 8, 2025 16:30

llucax previously approved these changes Apr 9, 2025

View reviewed changes

ela-kotulska-frequenz dismissed llucax’s stale review via e7069da April 16, 2025 09:15

ela-kotulska-frequenz force-pushed the pipe_doc branch from aca4789 to e7069da Compare April 16, 2025 09:15

ela-kotulska-frequenz force-pushed the pipe_doc branch from e7069da to 8398261 Compare April 16, 2025 09:20

ela-kotulska-frequenz requested a review from Copilot April 16, 2025 09:20

Copilot AI reviewed Apr 16, 2025

View reviewed changes

src/frequenz/channels/experimental/_pipe.py Outdated Show resolved Hide resolved

ela-kotulska-frequenz requested a review from llucax April 16, 2025 09:23

llucax reviewed Apr 16, 2025

View reviewed changes

ela-kotulska-frequenz force-pushed the pipe_doc branch from 8398261 to b2e2f66 Compare April 16, 2025 10:59

github-actions bot added the part:tests Affects the unit, integration and performance (benchmarks) tests label Apr 16, 2025

ela-kotulska-frequenz requested a review from llucax April 16, 2025 11:02

llucax previously approved these changes Apr 16, 2025

View reviewed changes

Improve documentation of the frequenz.channels.experimental.Pipe

9b4b7d7

Documentation shows how to create Pipe inside context manager to cleanup the resources when context is exited. Signed-off-by: Elzbieta Kotulska <[email protected]>

ela-kotulska-frequenz dismissed llucax’s stale review via 9b4b7d7 April 17, 2025 06:35

ela-kotulska-frequenz force-pushed the pipe_doc branch from b2e2f66 to 9b4b7d7 Compare April 17, 2025 06:35

llucax approved these changes Apr 17, 2025

View reviewed changes

ela-kotulska-frequenz added this pull request to the merge queue Apr 17, 2025

Merged via the queue into frequenz-floss:v1.x.x with commit 9e67b90 Apr 17, 2025
5 checks passed

ela-kotulska-frequenz deleted the pipe_doc branch April 17, 2025 06:52

llucax added this to the v1.8.0 milestone Jun 25, 2025

		source_receiver = await stack.enter_async_context(
		closing(source_channel.new_receiver()))

	receiver = await stack.enter_async_context(closing(forwarding_channel.new_receiver()))
	receiver = await stack.enter_context(closing(forwarding_channel.new_receiver()))

Improve documentation of the frequenz.channels.experimental.Pipe #407

Improve documentation of the frequenz.channels.experimental.Pipe #407

Conversation

ela-kotulska-frequenz commented Apr 8, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

llucax left a comment

Choose a reason for hiding this comment

Uh oh!

llucax Apr 9, 2025

Choose a reason for hiding this comment

Uh oh!

llucax Apr 9, 2025

Choose a reason for hiding this comment

Uh oh!

ela-kotulska-frequenz Apr 16, 2025

Choose a reason for hiding this comment

Uh oh!

llucax Apr 9, 2025

Choose a reason for hiding this comment

Uh oh!

llucax commented Apr 9, 2025

Uh oh!

ela-kotulska-frequenz commented Apr 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

ela-kotulska-frequenz commented Apr 16, 2025

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

llucax Apr 16, 2025

Choose a reason for hiding this comment

Uh oh!

ela-kotulska-frequenz Apr 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

llucax Apr 16, 2025

Choose a reason for hiding this comment

Uh oh!

llucax Apr 16, 2025

Choose a reason for hiding this comment

Uh oh!

llucax left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Improve documentation of the `frequenz.channels.experimental.Pipe` #407

Improve documentation of the `frequenz.channels.experimental.Pipe` #407

ela-kotulska-frequenz commented Apr 16, 2025 •

edited

Loading

ela-kotulska-frequenz Apr 16, 2025 •

edited

Loading