From e6401c688b93ed3a688d13e32b6844dfcb22ce2f Mon Sep 17 00:00:00 2001 From: Ruslan Lesiutin Date: Mon, 3 Nov 2025 11:46:41 -0800 Subject: [PATCH 1/3] Limit WebSocket queue size for packager connection (#54300) Summary: # Changelog: [Internal] Establishes a queue mechanism on top of the OkHttp's WebSocket implementation. This mechanism will control the queue size and guarantee that we don't have more than 16MB scheduled. This prevents the scenario of when OkHttp forces WS disconnection because of this threshold. Reviewed By: motiz88 Differential Revision: D85581509 --- .../CxxInspectorPackagerConnection.kt | 108 ++++++++++++++++-- ...xxInspectorPackagerConnectionWebSocket.cpp | 31 ++++- .../CxxInspectorPackagerConnectionTest.kt | 28 +++++ 3 files changed, 156 insertions(+), 11 deletions(-) create mode 100644 packages/react-native/ReactAndroid/src/test/java/com/facebook/react/devsupport/CxxInspectorPackagerConnectionTest.kt diff --git a/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/CxxInspectorPackagerConnection.kt b/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/CxxInspectorPackagerConnection.kt index a92d4ac26c62..5e3be8b14413 100644 --- a/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/CxxInspectorPackagerConnection.kt +++ b/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/CxxInspectorPackagerConnection.kt @@ -9,11 +9,17 @@ package com.facebook.react.devsupport import android.os.Handler import android.os.Looper +import com.facebook.common.logging.FLog import com.facebook.jni.HybridData import com.facebook.proguard.annotations.DoNotStrip import com.facebook.proguard.annotations.DoNotStripAny +import com.facebook.react.common.annotations.VisibleForTesting import com.facebook.soloader.SoLoader import java.io.Closeable +import java.nio.ByteBuffer +import java.nio.charset.StandardCharsets +import java.util.ArrayDeque +import java.util.Queue import java.util.concurrent.TimeUnit import okhttp3.OkHttpClient import okhttp3.Request @@ -67,7 +73,7 @@ internal class CxxInspectorPackagerConnection( */ @DoNotStripAny private interface IWebSocket : Closeable { - fun send(message: String) + fun send(chunk: ByteBuffer) /** * Close the WebSocket connection. NOTE: There is no close() method in the C++ interface. @@ -76,6 +82,95 @@ internal class CxxInspectorPackagerConnection( override fun close() } + /** + * A simple WebSocket wrapper that prevents having more than 16MiB of messages queued + * simultaneously. This is done to stop OkHttp from closing the WebSocket connection. + * + * https://github.com/facebook/react-native/issues/39651. + * https://github.com/square/okhttp/blob/4e7dbec1ea6c9cf8d80422ac9d44b9b185c749a3/okhttp/src/commonJvmAndroid/kotlin/okhttp3/internal/ws/RealWebSocket.kt#L684. + */ + private class InspectorPackagerWebSocketImpl( + private val nativeWebSocket: WebSocket, + private val handler: Handler, + ) : IWebSocket { + private val messageQueue: Queue> = ArrayDeque() + private val queueLock = Any() + private val drainRunnable = + object : Runnable { + override fun run() { + FLog.d(TAG, "Attempting to drain the message queue after ${drainDelayMs}ms") + tryDrainQueue() + } + } + + /** + * We are providing a String to OkHttp's WebSocket, because there is no guarantee that all CDP + * clients will support binary data format. + */ + override fun send(chunk: ByteBuffer) { + synchronized(queueLock) { + val messageSize = chunk.capacity() + val message = StandardCharsets.UTF_8.decode(chunk).toString() + val currentQueueSize = nativeWebSocket.queueSize() + + if (currentQueueSize + messageSize > MAX_QUEUE_SIZE) { + FLog.d(TAG, "Reached queue size limit. Queueing the message.") + messageQueue.offer(Pair(message, messageSize)) + scheduleDrain() + } else { + if (messageQueue.isEmpty()) { + nativeWebSocket.send(message) + } else { + messageQueue.offer(Pair(message, messageSize)) + tryDrainQueue() + } + } + } + } + + override fun close() { + synchronized(queueLock) { + handler.removeCallbacks(drainRunnable) + messageQueue.clear() + nativeWebSocket.close(1000, "End of session") + } + } + + private fun tryDrainQueue() { + synchronized(queueLock) { + while (messageQueue.isNotEmpty()) { + val (nextMessage, nextMessageSize) = messageQueue.peek() ?: break + val currentQueueSize = nativeWebSocket.queueSize() + + if (currentQueueSize + nextMessageSize <= MAX_QUEUE_SIZE) { + messageQueue.poll() + if (!nativeWebSocket.send(nextMessage)) { + // The WebSocket is closing, closed, or cancelled. + handler.removeCallbacks(drainRunnable) + messageQueue.clear() + + break + } + } else { + scheduleDrain() + break + } + } + } + } + + private fun scheduleDrain() { + FLog.d(TAG, "Scheduled a task to drain messages queue.") + handler.removeCallbacks(drainRunnable) + handler.postDelayed(drainRunnable, drainDelayMs) + } + + companion object { + private val TAG: String = InspectorPackagerWebSocketImpl::class.java.simpleName + private const val drainDelayMs: Long = 100 + } + } + /** Java implementation of the C++ InspectorPackagerConnectionDelegate interface. */ private class DelegateImpl { private val httpClient = @@ -130,15 +225,8 @@ internal class CxxInspectorPackagerConnection( } }, ) - return object : IWebSocket { - override fun send(message: String) { - webSocket.send(message) - } - override fun close() { - webSocket.close(1000, "End of session") - } - } + return InspectorPackagerWebSocketImpl(webSocket, handler) } @DoNotStrip @@ -152,6 +240,8 @@ internal class CxxInspectorPackagerConnection( SoLoader.loadLibrary("react_devsupportjni") } + @VisibleForTesting internal const val MAX_QUEUE_SIZE = 16L * 1024 * 1024 // 16MiB + @JvmStatic private external fun initHybrid( url: String, diff --git a/packages/react-native/ReactAndroid/src/main/jni/react/devsupport/JCxxInspectorPackagerConnectionWebSocket.cpp b/packages/react-native/ReactAndroid/src/main/jni/react/devsupport/JCxxInspectorPackagerConnectionWebSocket.cpp index 7de472736303..efd1948f7aa5 100644 --- a/packages/react-native/ReactAndroid/src/main/jni/react/devsupport/JCxxInspectorPackagerConnectionWebSocket.cpp +++ b/packages/react-native/ReactAndroid/src/main/jni/react/devsupport/JCxxInspectorPackagerConnectionWebSocket.cpp @@ -5,6 +5,8 @@ * LICENSE file in the root directory of this source tree. */ +#include + #include "JCxxInspectorPackagerConnectionWebSocket.h" using namespace facebook::jni; @@ -12,10 +14,35 @@ using namespace facebook::react::jsinspector_modern; namespace facebook::react::jsinspector_modern { +namespace { + +local_ref getReadOnlyByteBufferFromStringView( + std::string_view sv) { + auto buffer = JByteBuffer::wrapBytes( + const_cast(reinterpret_cast(sv.data())), + sv.size()); + + /** + * Return a read-only buffer that shares the underlying contents. + * This guards from accidential mutations on the Java side, since we did + * casting above. + * + * https://docs.oracle.com/javase/8/docs/api/java/nio/ByteBuffer.html#asReadOnlyBuffer-- + */ + static auto method = + buffer->javaClassStatic()->getMethod( + "asReadOnlyBuffer"); + return method(buffer); +} + +} // namespace + void JCxxInspectorPackagerConnectionWebSocket::send(std::string_view message) { static auto method = - javaClassStatic()->getMethod("send"); - method(self(), std::string(message)); + javaClassStatic()->getMethod)>( + "send"); + auto byteBuffer = getReadOnlyByteBufferFromStringView(message); + method(self(), byteBuffer); } void JCxxInspectorPackagerConnectionWebSocket::close() { diff --git a/packages/react-native/ReactAndroid/src/test/java/com/facebook/react/devsupport/CxxInspectorPackagerConnectionTest.kt b/packages/react-native/ReactAndroid/src/test/java/com/facebook/react/devsupport/CxxInspectorPackagerConnectionTest.kt new file mode 100644 index 000000000000..462317018ec7 --- /dev/null +++ b/packages/react-native/ReactAndroid/src/test/java/com/facebook/react/devsupport/CxxInspectorPackagerConnectionTest.kt @@ -0,0 +1,28 @@ +/* + * Copyright (c) Meta Platforms, Inc. and affiliates. + * + * This source code is licensed under the MIT license found in the + * LICENSE file in the root directory of this source tree. + */ + +package com.facebook.react.devsupport + +import okhttp3.internal.ws.RealWebSocket +import org.assertj.core.api.Assertions.assertThat +import org.junit.Test + +class CxxInspectorPackagerConnectionTest { + + @Test + fun testMaxQueueSizeEquality() { + val okHttpRealWebSocketClass = RealWebSocket::class.java + val okHttpMaxQueueSizeField = okHttpRealWebSocketClass.getDeclaredField("MAX_QUEUE_SIZE") + okHttpMaxQueueSizeField.isAccessible = true + + val okHttpMaxQueueSize = okHttpMaxQueueSizeField.getLong(null) + assertThat(okHttpMaxQueueSize).isNotNull + + assertThat(okHttpMaxQueueSize) + .isEqualTo(CxxInspectorPackagerConnection.Companion.MAX_QUEUE_SIZE) + } +} From 323ef8e1c7a81695e2cbc306605bd30751bb2eb3 Mon Sep 17 00:00:00 2001 From: Ruslan Lesiutin Date: Mon, 3 Nov 2025 11:46:41 -0800 Subject: [PATCH 2/3] Bump chunk size for android (#54301) Summary: # Changelog: [Internal] Previously, we would read every single kilobyte and do Java -> C++ call with `jni`. For big objects, like source maps, this means that we were doing at least 1024 calls for a single megabyte of incoming data. From my observations, some source maps on Twilight could reach 30Mb+. There is a trade-off between how much of memory we want to allocate while reading a stream and a runtime. I didn't notice any differences while changing the chunk size from 8Kb to 1Mb and some values in between; in the end it purely depends on the OkHttp's or Okio's implementation of the stream, looks like it uses 8Kb as a chunk size by default: {F1983042734} Reviewed By: huntie Differential Revision: D85652217 --- .../react/devsupport/inspector/InspectorNetworkHelper.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/inspector/InspectorNetworkHelper.kt b/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/inspector/InspectorNetworkHelper.kt index a7ba76b09e03..f8459b4b51e3 100644 --- a/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/inspector/InspectorNetworkHelper.kt +++ b/packages/react-native/ReactAndroid/src/main/java/com/facebook/react/devsupport/inspector/InspectorNetworkHelper.kt @@ -66,7 +66,7 @@ internal object InspectorNetworkHelper { response.body().use { responseBody -> if (responseBody != null) { val inputStream = responseBody.byteStream() - val chunkSize = 1024 + val chunkSize = 8 * 1024 // 8Kb val buffer = ByteArray(chunkSize) var bytesRead: Int From f676b8d7073b145d3ec0d69093136b357cc6a93f Mon Sep 17 00:00:00 2001 From: Ruslan Lesiutin Date: Mon, 3 Nov 2025 11:46:41 -0800 Subject: [PATCH 3/3] Increase the size of the chunk of ProfileChunk events (#54352) Summary: # Changelog: [Internal] Since we can now avoid disconnections, it should be safe to increase the chunk size. This should improve the trace loading time. Reviewed By: huntie Differential Revision: D85937959 --- .../ReactCommon/jsinspector-modern/TracingAgent.cpp | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/packages/react-native/ReactCommon/jsinspector-modern/TracingAgent.cpp b/packages/react-native/ReactCommon/jsinspector-modern/TracingAgent.cpp index e661b0e741a5..9700a1633a74 100644 --- a/packages/react-native/ReactCommon/jsinspector-modern/TracingAgent.cpp +++ b/packages/react-native/ReactCommon/jsinspector-modern/TracingAgent.cpp @@ -26,12 +26,8 @@ const uint16_t TRACE_EVENT_CHUNK_SIZE = 1000; /** * The maximum number of ProfileChunk trace events * that will be sent in a single CDP Tracing.dataCollected message. - * TODO(T219394401): Increase the size once we manage the queue on OkHTTP - side - * properly and avoid WebSocket disconnections when sending a message larger - * than 16MB. */ -const uint16_t PROFILE_TRACE_EVENT_CHUNK_SIZE = 1; +const uint16_t PROFILE_TRACE_EVENT_CHUNK_SIZE = 10; } // namespace