Merge pull request #65 from radixdlt/feature/backpressure-fix

Add backpressure buffer to TCP/UDP channels processor

radixdlt-core/radixdlt/src/main/java/com/radixdlt/network/transport/tcp/NettyTCPTransportImpl.java

@@ -27,6 +27,7 @@
 import com.radixdlt.counters.SystemCounters;
 import com.radixdlt.network.messaging.InboundMessage;
 import hu.akarnokd.rxjava3.operators.FlowableTransformers;
+import io.reactivex.rxjava3.core.BackpressureOverflowStrategy;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.processors.PublishProcessor;
 import io.reactivex.rxjava3.schedulers.Schedulers;
@@ -60,6 +61,8 @@
 final class NettyTCPTransportImpl implements NettyTCPTransport {
 	private static final Logger log = LogManager.getLogger();
 
+	private static final int CHANNELS_BUFFER_SIZE = 128;
+
 	// Set this to true to see a detailed hexdump of sent/received data at runtime
 	private final boolean debugData;
 
@@ -203,6 +206,11 @@ public void initChannel(SocketChannel ch) {
 		}
 
 		return channels
+			.onBackpressureBuffer(
+				CHANNELS_BUFFER_SIZE,
+				() -> log.error("TCP channels buffer overflow!"),
+				BackpressureOverflowStrategy.DROP_LATEST
+			)
 			.compose(FlowableTransformers.flatMapAsync(v -> v, Schedulers.single(), false));
 	}
 

radixdlt-core/radixdlt/src/main/java/com/radixdlt/network/transport/udp/NettyUDPTransportImpl.java

@@ -26,8 +26,11 @@
 
 import com.radixdlt.counters.SystemCounters;
 import com.radixdlt.network.messaging.InboundMessage;
+import hu.akarnokd.rxjava3.operators.FlowableTransformers;
+import io.reactivex.rxjava3.core.BackpressureOverflowStrategy;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.processors.PublishProcessor;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -54,6 +57,8 @@
 final class NettyUDPTransportImpl implements Transport {
 	private static final Logger log = LogManager.getLogger();
 
+	private static final int CHANNELS_BUFFER_SIZE = 128;
+
 	// Set this to true to see a dump of packet data
 	protected static final boolean DEBUG_DATA = false;
 
@@ -187,7 +192,13 @@ public void initChannel(NioDatagramChannel ch) {
 	    	throw new UncheckedIOException("Error while opening channel", e);
 		}
 
-	    return Flowable.mergeDelayError(channels);
+		return channels
+			.onBackpressureBuffer(
+				CHANNELS_BUFFER_SIZE,
+				() -> log.error("UDP channels buffer overflow!"),
+				BackpressureOverflowStrategy.DROP_LATEST
+			)
+			.compose(FlowableTransformers.flatMapAsync(v -> v, Schedulers.single(), false));
 	}
 
 	@Override