Streaming gRPC client has no backpressure

[joroKr21]

It uses an unbounded queue:

zio-grpc/core/src/main/scalajvm/scalapb/zio_grpc/client/StreamingClientCallListener.scala

Line 51 in def8d5d

queue <- Queue.unbounded[ResponseFrame[Res]]

But there is no external backpressure mechanism, we request the next message as soon as we receive one:

zio-grpc/core/src/main/scalajvm/scalapb/zio_grpc/client/StreamingClientCallListener.scala

Line 27 in def8d5d

runtime.unsafe.run(queue.offer(ResponseFrame.Message(message)) *> call.request(1)).getOrThrowFiberFailure()

In addition, it would be beneficial to use a (configurable) batch size > 1 when requesting messages from the server.

If that's ok in terms of IP, I can port the solution from fs2-grpc to zio-grpc.

[thesamet]

A PR that adds backpressure (disabled by default) will be accepted. Generally, porting an implementation should be ok. If unsure please check directly with fs2-grpc team.

…

On Tue, May 28, 2024, 2:12 AM Georgi Krastev ***@***.***> wrote: It uses an unbounded queue: https://github.com/scalapb/zio-grpc/blob/def8d5d99b51330627d7b0b4f84b887de5b03034/core/src/main/scalajvm/scalapb/zio_grpc/client/StreamingClientCallListener.scala#L51 But there is no external backpressure mechanism, we request the next message as soon as we receive one: https://github.com/scalapb/zio-grpc/blob/def8d5d99b51330627d7b0b4f84b887de5b03034/core/src/main/scalajvm/scalapb/zio_grpc/client/StreamingClientCallListener.scala#L27 In addition, to use a (configurable) batch size > 1 when requesting messages from the server. If that's ok in terms of IP, I can port the solution from fs2-grpc to zio-grpc. — Reply to this email directly, view it on GitHub <#631>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACLBLJWVBHD6CIDF6VYNILZERDBDAVCNFSM6AAAAABIMQB7KCVHI2DSMVQWIX3LMV43ASLTON2WKOZSGMZDANJUGA4DGMA> . You are receiving this because you are subscribed to this thread.Message ID: ***@***.***>

[joroKr21]

A PR that adds backpressure (disabled by default) will be accepted.

Where can we configure it? I can't find a good place.

[thesamet]

See

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

Lines 41 to 48 in def8d5d

	private[zio_grpc] val queueSizeProp = "zio_grpc.backpressure_queue_size"
	val backpressureQueueSize: IO[StatusException, Int] =
	ZIO
	.config(zio.Config.int("backpressure_queue_size").nested("zio_grpc").withDefault(16))
	.mapError { (e: zio.Config.Error) =>
	Status.INTERNAL.withDescription(s"$queueSizeProp: ${e.getMessage}").asException()
	}

[joroKr21]

I added it as a parameter to ZChannel: #636
If you tell me how to call the setting, I could also make it a property.

[thesamet]

Can you rephrase your question, I am not sure exactly what are you looking for.

[joroKr21]

In my PR I just added a constructor parameter to ZChannel - should I convert it to a property and if so how should we call it?

[joroKr21]

Or do you mean that I should reuse backpressure_queue_size for the client?

[thesamet]

Since it's a constructor parameter, no need to introduce a property.