Skip to content
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

Jump to bottom

Implement streaming client backpressure with configurable batch size #636

Merged
merged 3 commits into from
Jul 14, 2024
Merged

Implement streaming client backpressure with configurable batch size #636

merged 3 commits into from
Jul 14, 2024

Conversation

joroKr21
Copy link
Contributor

@joroKr21 joroKr21 commented Jun 7, 2024

Closes #631

@joroKr21 joroKr21 force-pushed the client-backpressure branch 3 times, most recently from 6dc935b to 6fc4b28 Compare June 7, 2024 16:09
@joroKr21 joroKr21 mentioned this pull request Jun 7, 2024
Closed
thesamet
thesamet reviewed Jun 10, 2024
View reviewed changes
Copy link
Contributor

@thesamet thesamet left a comment

Choose a reason for hiding this comment

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

Can you add changelog item, and update backpressure.md in the docs about this functionality?

thesamet
thesamet reviewed Jun 10, 2024
View reviewed changes
Copy link
Contributor

@thesamet thesamet left a comment

Choose a reason for hiding this comment

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

can you add tests for the new functionality?

@joroKr21
Copy link
Contributor Author

can you add tests for the new functionality?

I don't quite understand the testing machinery here. How can we test backpressure?

@joroKr21
Copy link
Contributor Author

@thesamet it's not possible to test backpressure reliably because the server just keeps sending messages. See:

zio-grpc/core/src/main/scalajvm/scalapb/zio_grpc/server/ZServerCallHandler.scala

Line 115 in 5d29b6e

ZChannel.suspend(if (call.call.isReady()) go else ZChannel.fromZIO(call.awaitReady) *> go),

The check to call.call.isReady succeeds and is not related to call.awaitReady:

  • def isReady: UIO[Boolean] = ZIO.succeed(call.isReady())
  • private[zio_grpc] def awaitReady: UIO[Unit] = canSend.acquire.commit
  • private[zio_grpc] def setReady(): UIO[Unit] = canSend.release.commit

I don't know if that's intended or not but it means that test service keeps sending me messages I didn't request in the client.

@joroKr21
Copy link
Contributor Author

@thesamet wdyt about this one?

@thesamet thesamet merged commit 03d623d into scalapb:master Jul 14, 2024
1 check passed
@joroKr21 joroKr21 deleted the client-backpressure branch July 15, 2024 03:30
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@thesamet thesamet thesamet left review comments

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.

Streaming gRPC client has no backpressure
2 participants
@joroKr21 @thesamet