diff --git a/grpc_p2p_client/cmd/multi-publish/main.go b/grpc_p2p_client/cmd/multi-publish/main.go index f59e048..c54f958 100644 --- a/grpc_p2p_client/cmd/multi-publish/main.go +++ b/grpc_p2p_client/cmd/multi-publish/main.go @@ -130,6 +130,19 @@ func sendMessages(ctx context.Context, ip string, datasize int, write bool, data return fmt.Errorf("[%s] ListenCommands failed: %w", ip, err) } + // Drain the response side of the bidi stream so the per-stream HTTP/2 + // flow-control window does not fill up and block stream.Send. Without + // this drain the publisher deadlocks after ~7-8 publications at + // payload sizes >=100KB, because each publish triggers several + // trace-event Response messages from the server. + go func() { + for { + if _, err := stream.Recv(); err != nil { + return + } + } + }() + println(fmt.Sprintf("Connected to node at: %s…", ip)) for i := 0; i < *count; i++ { diff --git a/grpc_p2p_client/cmd/single/main.go b/grpc_p2p_client/cmd/single/main.go index 1b26ac1..d524b83 100644 --- a/grpc_p2p_client/cmd/single/main.go +++ b/grpc_p2p_client/cmd/single/main.go @@ -115,6 +115,18 @@ func publish(ctx context.Context, stream protobuf.CommandStream_ListenCommandsCl log.Fatal("-msg is required in publish mode") } + // Drain the response side of the bidi stream so the per-stream HTTP/2 + // flow-control window does not fill up and block stream.Send. Without + // this drain a multi-message publish (-count >> 1 with large -msg) + // deadlocks after ~7-8 publications at payload sizes >=100KB. + go func() { + for { + if _, err := stream.Recv(); err != nil { + return + } + } + }() + for i := 0; i < count; i++ { start := time.Now() var data []byte