addressing comments

Signed-off-by: Anthony Petrov <anthony@swirldslabs.com>

Author: anthony-swirldslabs

pbj-core/pbj-grpc-client-helidon/src/main/java/com/hedera/pbj/grpc/client/helidon/PbjGrpcCall.java

@@ -21,8 +21,6 @@
 import io.helidon.http.http2.Http2StreamState;
 import io.helidon.webclient.http2.Http2ClientStream;
 import io.helidon.webclient.http2.StreamTimeoutException;
-import java.io.UncheckedIOException;
-import java.net.SocketException;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -197,29 +195,13 @@ private void receiveRepliesLoop() {
             // read data from stream
             final PbjGrpcDatagramReader datagramReader =
                     new PbjGrpcDatagramReader(grpcClient.getConfig().maxIncomingBufferSize());
-            boolean repliesReceived = false;
             while (isStreamOpen() && !clientStream.trailers().isDone() && clientStream.hasEntity()) {
                 final Http2FrameData frameData;
                 try {
                     frameData = clientStream.readOne(grpcClient.getConfig().readTimeout());
                 } catch (StreamTimeoutException e) {
                     // Check if the connection is alive. See a comment above about the KeepAlive timeout.
-                    try {
-                        clientStream.sendPing();
-                    } catch (UncheckedIOException uioe) {
-                        // And the connection may in fact be closed.
-                        if (repliesReceived
-                                && uioe.getCause() instanceof SocketException se
-                                && se.getMessage() != null
-                                && se.getMessage().contains("Socket closed")) {
-                            // We won't be able to read trailers anyway because the connection is closed.
-                            // Since at least one reply has been received and processed, complete the call:
-                            pipeline.onComplete();
-                            return;
-                        }
-                        // Either we've never received a single reply yet, or this isn't a "Socket closed".
-                        throw uioe;
-                    }
+                    clientStream.sendPing();
                     // FUTURE WORK: implement an uber timeout to return
                     continue;
                 }
@@ -254,7 +236,6 @@ private void receiveRepliesLoop() {
                                     Codec.DEFAULT_MAX_DEPTH,
                                     grpcClient.getConfig().maxSize());
                             pipeline.onNext(reply);
-                            repliesReceived = true;
                         } catch (ParseException e) {
                             pipeline.onError(e);
                             // We won't be able to proceed probably because parsing failed.

pbj-core/pbj-runtime/src/main/java/com/hedera/pbj/runtime/grpc/GrpcCompression.java

@@ -90,12 +90,12 @@ public Bytes decompress(Bytes bytes) {
     private static final Map<String, Decompressor> DECOMPRESSOR_MAP = new HashMap<>();
 
     /** Register a Compressor, potentially overwriting an existing registration for `name`. */
-    public static void registerCompressor(String name, Compressor compressor) {
+    public static synchronized void registerCompressor(String name, Compressor compressor) {
         COMPRESSOR_MAP.put(name, compressor);
     }
 
     /** Register a Decompressor, potentially overwriting an existing registration for `name`. */
-    public static void registerDecompressor(String name, Decompressor decompressor) {
+    public static synchronized void registerDecompressor(String name, Decompressor decompressor) {
         DECOMPRESSOR_MAP.put(name, decompressor);
     }
 

pbj-integration-tests/src/jmh/java/com/hedera/pbj/integration/jmh/grpc/NetworkLatencySimulator.java

@@ -12,6 +12,11 @@
  * instead of leveraging the fast network.
  * A slow network would show that compression may be useful sometimes. Specifically, for larger, compressible
  * payloads. Smaller payloads (<8K) never benefit from compression.
+ * <p>
+ * The network latency simulator uses the PbjGrpcNetworkBytesInspector feature of the PbjGrpcCall, which is
+ * a static entity, meaning that it affects any and all PbjGrpcCall objects in the same JVM. This is by design
+ * for performance reasons. However, users of this Simulator should be aware of this and not try to use it
+ * from multiple threads with different latency parameters at once.
  */
 public class NetworkLatencySimulator {
     // ms 1e-3, us 1e-6, ns 1e-9:
@@ -31,9 +36,9 @@ public static void simulate(final long networkSpeedMbitPerSecond, final boolean
 
             private void sleep(long bytes) {
                 final long nanos = nanosPerByte * bytes;
-                try {
-                    Thread.sleep(nanos / NANOS_IN_MILLI, (int) (nanos % NANOS_IN_MILLI));
-                } catch (InterruptedException ignore) {
+                final long deadline = System.nanoTime() + nanos;
+                while (System.nanoTime() < deadline) {
+                    Thread.onSpinWait();
                 }
             }
 

pbj-integration-tests/src/jmh/java/com/hedera/pbj/integration/jmh/grpc/block/TestBlockGrpcBench.java

@@ -7,8 +7,11 @@
 import com.hedera.pbj.integration.jmh.grpc.NetworkLatencySimulator;
 import com.hedera.pbj.integration.jmh.grpc.PbjGrpcBench;
 import com.hedera.pbj.integration.jmh.grpc.ZstdGrpcTransformer;
+import com.hedera.pbj.runtime.Codec;
 import com.hedera.pbj.runtime.grpc.GrpcClient;
 import com.hedera.pbj.runtime.grpc.Pipeline;
+import java.io.UncheckedIOException;
+import java.net.SocketException;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Flow;
@@ -54,12 +57,12 @@
 @OutputTimeUnit(TimeUnit.SECONDS)
 @BenchmarkMode(Mode.Throughput)
 public class TestBlockGrpcBench {
-    private static final int INVOCATIONS = 2_000;
+    private static final int INVOCATIONS = 400;
 
     static {
         new ZstdGrpcTransformer(-5).register("zstd-5");
-        new ZstdGrpcTransformer(0).register("zstd0");
         new ZstdGrpcTransformer(3).register("zstd"); // the default level
+        new ZstdGrpcTransformer(10).register("zstd10");
 
         // 1Gbps network:
         NetworkLatencySimulator.simulate(1_000, true);
@@ -70,8 +73,13 @@ static TestBlockStreamerInterface.TestBlockStreamerClient createClient(final int
         if (encodings == null || encodings.length == 0) {
             grpcClient = GrpcTestUtils.createGrpcClient(port, GrpcTestUtils.PROTO_OPTIONS);
         } else {
-            grpcClient =
-                    GrpcTestUtils.createGrpcClient(port, GrpcTestUtils.PROTO_OPTIONS, encodings[0], Set.of(encodings));
+            grpcClient = GrpcTestUtils.createGrpcClient(
+                    port,
+                    GrpcTestUtils.PROTO_OPTIONS,
+                    encodings[0],
+                    Set.of(encodings),
+                    Codec.DEFAULT_MAX_SIZE,
+                    Codec.DEFAULT_MAX_SIZE * 5);
         }
 
         return new TestBlockStreamerInterface.TestBlockStreamerClient(grpcClient, GrpcTestUtils.PROTO_OPTIONS);
@@ -82,7 +90,7 @@ public static class BenchState {
         @Param({"102400", "524288", "2048000"})
         int maxBlockSize;
 
-        @Param({"identity", "gzip", "zstd", "zstd0", "zstd-5"})
+        @Param({"identity", "gzip", "zstd", "zstd10", "zstd-5"})
         String encodings;
 
         PortsAllocator.Port port;
@@ -156,8 +164,19 @@ public void onSubscribe(Flow.Subscription subscription) {
 
                 @Override
                 public void onError(Throwable throwable) {
-                    new RuntimeException(throwable).printStackTrace();
                     latch.countDown();
+                    if (throwable instanceof UncheckedIOException uioe
+                            && uioe.getCause() instanceof SocketException se
+                            && se.getMessage() != null
+                            && se.getMessage().contains("Socket closed")) {
+                        // A streaming server may close its connection sometimes before
+                        // the client has received and processed all the replies. However, the client's
+                        // PbjGrpcCall may try and ping the server during the processing. This results in
+                        // calling this method. And this seems to happen often enough to print this error
+                        // more often than we'd like. So we just ignore this particular error altogether here.
+                        return;
+                    }
+                    new RuntimeException(throwable).printStackTrace();
                 }
 
                 @Override