Copilot commented on code in PR #16369:
URL: https://github.com/apache/dubbo/pull/16369#discussion_r3611828850
##########
dubbo-serialization/dubbo-serialization-fastjson2/src/test/java/org/apache/dubbo/common/serialize/fastjson2/FastJson2SerializationTest.java:
##########
@@ -620,4 +678,95 @@ void testLimit5() throws IOException,
ClassNotFoundException {
frameworkModel.destroy();
}
}
+
+ private byte[] serializeRefOuter(Serialization serialization, URL url)
throws IOException {
+ RefOuter outer = new RefOuter();
+ List<RefInner> items = new ArrayList<>();
+ List<Long> sharedIds = new ArrayList<>();
+ sharedIds.add(1L);
+ sharedIds.add(2L);
+ for (int i = 0; i < 20; i++) {
+ RefInner inner = new RefInner();
+ inner.setName("item-" + i);
+ inner.setIds(sharedIds);
+ items.add(inner);
+ }
+ outer.setItems(items);
+
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ ObjectOutput objectOutput = serialization.serialize(url, outputStream);
+ objectOutput.writeObject(outer);
+ objectOutput.flushBuffer();
+ return outputStream.toByteArray();
+ }
+
+ private int countConcurrentNullIds(Serialization serialization, URL url,
byte[] bytes) throws Exception {
+ int rounds = 10;
+ int threadCount = 200;
+ AtomicInteger totalNullTasks = new AtomicInteger();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ for (int round = 0; round < rounds; round++) {
+ clearObjectReaderCache();
+ CyclicBarrier barrier = new CyclicBarrier(threadCount);
+ CountDownLatch endLatch = new CountDownLatch(threadCount);
+ AtomicInteger roundNullTasks = new AtomicInteger();
+ for (int i = 0; i < threadCount; i++) {
+ Thread thread = new Thread(() -> {
+ try {
+ barrier.await();
+ if (countNullIds(serialization, url, bytes) > 0) {
+ roundNullTasks.incrementAndGet();
+ }
+ } catch (Throwable throwable) {
+ failure.compareAndSet(null, throwable);
+ } finally {
+ endLatch.countDown();
+ }
+ });
+ thread.start();
+ }
+ endLatch.await();
+ if (failure.get() != null) {
+ throw new AssertionError("Concurrent deserialization failed",
failure.get());
+ }
Review Comment:
`CountDownLatch#await()` / `CyclicBarrier#await()` are currently unbounded.
If a worker thread stalls (e.g., due to GC pressure or an internal fastjson2
deadlock), this test can hang the entire build indefinitely. Add timeouts and
fail fast with a clear error when the concurrent round does not complete in
time.
##########
dubbo-serialization/dubbo-serialization-fastjson2/src/test/java/org/apache/dubbo/common/serialize/fastjson2/FastJson2SerializationTest.java:
##########
@@ -620,4 +678,95 @@ void testLimit5() throws IOException,
ClassNotFoundException {
frameworkModel.destroy();
}
}
+
+ private byte[] serializeRefOuter(Serialization serialization, URL url)
throws IOException {
+ RefOuter outer = new RefOuter();
+ List<RefInner> items = new ArrayList<>();
+ List<Long> sharedIds = new ArrayList<>();
+ sharedIds.add(1L);
+ sharedIds.add(2L);
+ for (int i = 0; i < 20; i++) {
+ RefInner inner = new RefInner();
+ inner.setName("item-" + i);
+ inner.setIds(sharedIds);
+ items.add(inner);
+ }
+ outer.setItems(items);
+
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ ObjectOutput objectOutput = serialization.serialize(url, outputStream);
+ objectOutput.writeObject(outer);
+ objectOutput.flushBuffer();
+ return outputStream.toByteArray();
+ }
+
+ private int countConcurrentNullIds(Serialization serialization, URL url,
byte[] bytes) throws Exception {
+ int rounds = 10;
+ int threadCount = 200;
+ AtomicInteger totalNullTasks = new AtomicInteger();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+
+ for (int round = 0; round < rounds; round++) {
+ clearObjectReaderCache();
+ CyclicBarrier barrier = new CyclicBarrier(threadCount);
+ CountDownLatch endLatch = new CountDownLatch(threadCount);
+ AtomicInteger roundNullTasks = new AtomicInteger();
+ for (int i = 0; i < threadCount; i++) {
+ Thread thread = new Thread(() -> {
+ try {
Review Comment:
This test creates 200 new `Thread` instances per round (2,000 total) which
adds significant overhead and can make the test flaky on constrained CI
runners. Consider using an `ExecutorService` (fixed thread pool) and reusing
worker threads across rounds to reduce scheduling/creation cost while
preserving the concurrency stress.
##########
dubbo-serialization/dubbo-serialization-fastjson2/src/main/java/org/apache/dubbo/common/serialize/fastjson2/FastJson2ObjectInput.java:
##########
@@ -205,4 +175,32 @@ private int readLength() throws IOException {
}
return value;
}
+
+ private <T> T parseObject(byte[] bytes, Class<T> cls,
Fastjson2SecurityManager.Handler securityFilter) {
+ synchronized (getParseLock(cls)) {
+ if (securityFilter.isCheckSerializable()) {
Review Comment:
`parseObject(...)` now synchronizes for the full duration of
`JSONB.parseObject`. When callers use the no-arg `ObjectInput#readObject()`
path, `FastJson2ObjectInput#readObject()` delegates to
`readObject(Object.class)`, which means all generic decodes contend on a single
`Object.class` lock. ExchangeCodec decodes request/response data via
`in.readObject()` (dubbo-remoting-api/.../ExchangeCodec.java:435-449), so this
can effectively serialize fastjson2 decoding across threads and cause a
significant throughput regression. Consider narrowing the lock for the
`Object.class` case by extracting the concrete type from the JSONB payload
header (or other fastjson2 metadata) and locking on that class instead of
`Object.class`, so unrelated message types can still deserialize concurrently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
