Skip to content

Client streaming tests #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Jun 16, 2020
Merged

Client streaming tests #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
159c30d
Fix close not awaitable, fix done is callable, fix return async next …
boukeversteegh Jun 15, 2020
f7aa615
Add test-cases for client stream-stream
boukeversteegh Jun 15, 2020
0814729
Add cases for send()
boukeversteegh Jun 15, 2020
50bb67b
Fix bugs and remove footgun feature in AsyncChannel
nat-n Jun 15, 2020
4e78fe9
Merge branch 'client-streaming' into client-streaming-tests
nat-n Jun 15, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 9 additions & 15 deletions betterproto/grpc/util/async_channel.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,17 +81,10 @@ class AsyncChannel(AsyncIterable[T]):
"""

def __init__(
self,
source: Union[Iterable[T], AsyncIterable[T]] = tuple(),
*,
buffer_limit: int = 0,
close: bool = False,
self, *, buffer_limit: int = 0, close: bool = False,
):
self._queue: asyncio.Queue[Union[T, object]] = asyncio.Queue(buffer_limit)
self._closed = False
self._sending_task = (
asyncio.ensure_future(self.send_from(source, close)) if source else None
)
self._waiting_recievers: int = 0
# Track whether flush has been invoked so it can only happen once
self._flushed = False
Expand All @@ -100,13 +93,14 @@ def __aiter__(self) -> AsyncIterator[T]:
return self

async def __anext__(self) -> T:
if self.done:
if self.done():
raise StopAsyncIteration
self._waiting_recievers += 1
try:
result = await self._queue.get()
if result is self.__flush:
raise StopAsyncIteration
return result
finally:
self._waiting_recievers -= 1
self._queue.task_done()
Expand All @@ -131,7 +125,7 @@ def done(self) -> bool:

async def send_from(
self, source: Union[Iterable[T], AsyncIterable[T]], close: bool = False
):
) -> "AsyncChannel[T]":
"""
Iterates the given [Async]Iterable and sends all the resulting items.
If close is set to True then subsequent send calls will be rejected with a
Expand All @@ -151,24 +145,26 @@ async def send_from(
await self._queue.put(item)
if close:
# Complete the closing process
await self.close()
self.close()
return self

async def send(self, item: T):
async def send(self, item: T) -> "AsyncChannel[T]":
"""
Send a single item over this channel.
:param item: The item to send
"""
if self._closed:
raise ChannelClosed("Cannot send through a closed channel")
await self._queue.put(item)
return self

async def recieve(self) -> Optional[T]:
"""
Returns the next item from this channel when it becomes available,
or None if the channel is closed before another item is sent.
:return: An item from the channel
"""
if self.done:
if self.done():
raise ChannelDone("Cannot recieve from a closed channel")
self._waiting_recievers += 1
try:
Expand All @@ -184,8 +180,6 @@ def close(self):
"""
Close this channel to new items
"""
if self._sending_task is not None:
self._sending_task.cancel()
self._closed = True
asyncio.ensure_future(self._flush_queue())

Expand Down
22 changes: 13 additions & 9 deletions betterproto/tests/grpc/test_grpclib_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import asyncio
from betterproto.tests.output_betterproto.service.service import (
DoThingResponse,
DoThingRequest,
Expand Down Expand Up @@ -129,7 +130,10 @@ async def test_async_gen_for_stream_stream_request():
# Use an AsyncChannel to decouple sending and recieving, it'll send some_things
# immediately and we'll use it to send more_things later, after recieving some
# results
request_chan = AsyncChannel(GetThingRequest(name) for name in some_things)
request_chan = AsyncChannel()
send_initial_requests = asyncio.ensure_future(
request_chan.send_from(GetThingRequest(name) for name in some_things)
)
response_index = 0
async for response in client.get_different_things(request_chan):
assert response.name == expected_things[response_index]
Expand All @@ -138,13 +142,13 @@ async def test_async_gen_for_stream_stream_request():
if more_things:
# Send some more requests as we recieve reponses to be sure coordination of
# send/recieve events doesn't matter
another_response = await request_chan.send(
GetThingRequest(more_things.pop(0))
)
if another_response is not None:
assert another_response.name == expected_things[response_index]
assert another_response.version == response_index
response_index += 1
await request_chan.send(GetThingRequest(more_things.pop(0)))
elif not send_initial_requests.done():
# Make sure the sending task it completed
await send_initial_requests
else:
# No more things to send make sure channel is closed
await request_chan.close()
request_chan.close()
assert response_index == len(
expected_things
), "Didn't recieve all exptected responses"
151 changes: 151 additions & 0 deletions betterproto/tests/grpc/test_stream_stream.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

import pytest

import betterproto
from betterproto.grpc.util.async_channel import AsyncChannel


@dataclass
class Message(betterproto.Message):
body: str = betterproto.string_field(1)


@pytest.fixture
def expected_responses():
return [Message("Hello world 1"), Message("Hello world 2"), Message("Done")]


class ClientStub:
async def connect(self, requests: AsyncIterator):
await asyncio.sleep(0.1)
async for request in requests:
await asyncio.sleep(0.1)
yield request
await asyncio.sleep(0.1)
yield Message("Done")


async def to_list(generator: AsyncIterator):
lis = []
async for value in generator:
lis.append(value)
return lis


@pytest.fixture
def client():
# channel = Channel(host='127.0.0.1', port=50051)
# return ClientStub(channel)
return ClientStub()


@pytest.mark.asyncio
async def test_from_list_close_automatically(client, expected_responses):
requests = AsyncChannel(
[Message(body="Hello world 1"), Message(body="Hello world 2")], close=True
)

responses = client.connect(requests)

assert await to_list(responses) == expected_responses


@pytest.mark.asyncio
async def test_from_list_close_manually_immediately(client, expected_responses):
requests = AsyncChannel(
[Message(body="Hello world 1"), Message(body="Hello world 2")], close=False
)

requests.close()

responses = client.connect(requests)

assert await to_list(responses) == expected_responses


@pytest.mark.asyncio
async def test_from_list_close_manually_after_connect(client, expected_responses):
requests = AsyncChannel(
[Message(body="Hello world 1"), Message(body="Hello world 2")], close=False
)

responses = client.connect(requests)

requests.close()

assert await to_list(responses) == expected_responses
Copy link
Owner

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Technically this is an incorrect usage of the API, (though it could be the API that's wrong...).

It looks like the sending task is canceled on the call to close, having not been given much chance to complete. and so to_list(responses) doesn't get all the responses because the request was cut short.

Now I can see how this is surprising, but it's not obvious whether the solution to this is a design change or just some documentation of the kind of sharp edge that is common with async/await.

It's not possible to await on a construction of the channel, and the intention is that the calling coro can continue while requests are sent. So the user either needs to accept that this can happen with items provided to the constructor or there needs to be a way to await the initial sending task, like await requests.wait(). The close method can't have this role IMO because I think it's important to be able to close the channel synchronously.

Maybe it's not worth being able to construct the channel with an initial source of items to send 😞, which is a pity cos it's slick.

Maybe we could get comparable convenience by adding support for chaining methods so:

requests = await AsyncChannel().send_from([...])
...
requests.close()

This makes some use cases less convenient like:

requests = await AsyncChannel()
sending_task = asyncio.ensure_future(requests.send_from([...]))
... # get some responses
await sending_task
requests.close()

But probably the simplicity and removal of a sharp edge is worth it.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see, I didn't realize or expect that the sending was in anyway influenced by close. I thought it was just to mark that the stream will have no more data.

That is something you really need, otherwise the connection will stay open forever. But I don't think it should cancel unsent messages.

I experimented a bit with just using a vanilla asyncio.queues.Queue, but quickly found you do need some way to convert it to an async generator (at least for the client as you've proposed it, which i think is good). So actually, if you just have a way to end the stream, and to async generate, it's all you need.

class MessageQueue(asyncio.queues.Queue):
    __close = object()

    async def close(self):
        await self.put(self.__close)

    def close_nowait(self):
        self.put_nowait(self.__close)

    async def __aiter__(self):
        while True:
            element = await self.get()
            if element is self.__close:
                self.task_done()
                break
            yield element
            self.task_done()

Test scenario:

@pytest.mark.asyncio
async def test_send_from_async_queue(client, expected_responses):
    requests = MessageQueue()
    await requests.put(Message(body="Hello world 1"))
    await requests.put(Message(body="Hello world 2"))

    responses = client.connect(requests)

    await requests.close()
    assert await to_list(responses) == expected_responses

This little class does the job for me.

If I want to send from another async generator, I don't need this class, I can just pass the generator directly as an argument to the client.

If I want to send from a list, I can do the same. If I want some complex combination of both, I can call put manually in a loop or asynchronously and it will work.

On its own AsyncChannel is a good looking and reusable abstraction, but in the context of a client that already accepts lists and async generators, I think its use will be limited.

Copy link
Owner

@nat-n nat-n Jun 16, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think sending has to cancel unsent messages, because theres' no guarantee that the number of unsent messages might be large or indefinite.

That's clever. And I'd probably prefer that kind of solution for its simplicity within the scope of a single app. But I think a library should have more clearly defined abstractions, and work as expected with a broader range of edge cases. A channel is conceptually a better fit than a queue for an object with the required set of basic operations: FIFO add/remove, concurrent iteration, and close. It's less code to implement it as a subtype of Queue, but I think naming/semantics/encapsulation are important too, and so abstracting the actual queue as an implementation detail is worthwhile.

I also think send_from is an important enough use case to support as first class, rather than requiring the user to write async for loops to connect things together. I image if I were using this pattern a few times within a project I'd end up writing a pipe_this_iterable_into_that_channel helper function and stashing it in a utils module somewhere, which is a pity when the library can support it natively without any obvious compromise to the scope or design.



@pytest.mark.asyncio
async def test_send_from_before_connect_and_close_automatically(
client, expected_responses
):
requests = AsyncChannel()

await requests.send_from(
[Message(body="Hello world 1"), Message(body="Hello world 2")], close=True
)

responses = client.connect(requests)

assert await to_list(responses) == expected_responses


@pytest.mark.asyncio
async def test_send_from_after_connect_and_close_automatically(
client, expected_responses
):
requests = AsyncChannel()

responses = client.connect(requests)

await requests.send_from(
[Message(body="Hello world 1"), Message(body="Hello world 2")], close=True
)

assert await to_list(responses) == expected_responses


@pytest.mark.asyncio
async def test_send_from_close_manually_immediately(client, expected_responses):
requests = AsyncChannel()

responses = client.connect(requests)

await requests.send_from(
[Message(body="Hello world 1"), Message(body="Hello world 2")], close=False
)

requests.close()

assert await to_list(responses) == expected_responses


@pytest.mark.asyncio
async def test_send_individually_and_close_before_connect(client, expected_responses):
requests = AsyncChannel()

await requests.send(Message(body="Hello world 1"))
await requests.send(Message(body="Hello world 2"))
requests.close()

responses = client.connect(requests)

assert await to_list(responses) == expected_responses


@pytest.mark.asyncio
async def test_send_individually_and_close_after_connect(client, expected_responses):
requests = AsyncChannel()

await requests.send(Message(body="Hello world 1"))
await requests.send(Message(body="Hello world 2"))

responses = client.connect(requests)

requests.close()

assert await to_list(responses) == expected_responses