You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fory.apache.org by "zhan7236 (via GitHub)" <gi...@apache.org> on 2025/12/01 03:09:25 UTC
[PR] [Java] Add optimized serializers for ArrayBlockingQueue and LinkedBlockingQueue [fory]
zhan7236 opened a new pull request, #2955:
URL: https://github.com/apache/fory/pull/2955
## Why?
The JDK default serialization for `ArrayBlockingQueue` and `LinkedBlockingQueue` is inefficient and produces warnings like:
```text
WARN ObjectStreamSerializer - class java.util.concurrent.ArrayBlockingQueue customized jdk serialization, which is inefficient.
```
These blocking queues are commonly used in concurrent applications, and optimized serializers will improve performance.
## What does this PR do?
Add dedicated serializers for `java.util.concurrent.ArrayBlockingQueue` and `java.util.concurrent.LinkedBlockingQueue`:
- `ArrayBlockingQueueSerializer`: Extends `ConcurrentCollectionSerializer` to handle thread-safe serialization with capacity preservation.
- `LinkedBlockingQueueSerializer`: Extends `ConcurrentCollectionSerializer` to handle thread-safe serialization with capacity preservation.
Both serializers:
- Use `CollectionSnapshot` for safe concurrent iteration.
- Preserve queue capacity during serialization/deserialization.
- Support both serialization and copy operations.
- Are registered as default serializers in `registerDefaultSerializers()`.
## Related issues
- Fixes #2663 (partial - blocking queue serializers)
## Does this PR introduce any user-facing change?
- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?
No public API changes. Users will automatically benefit from optimized serialization for `ArrayBlockingQueue` and `LinkedBlockingQueue`.
## Benchmark
Not applicable - This PR replaces the slow JDK `ObjectStreamSerializer` with a specialized binary serializer, which is a well-known performance improvement pattern already used for other collection types in this codebase.
--
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: commits-unsubscribe@fory.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@fory.apache.org
For additional commands, e-mail: commits-help@fory.apache.org
Re: [PR] feat(java): add optimized serializers for blocking queues [fory]
Posted by "zhan7236 (via GitHub)" <gi...@apache.org>.
zhan7236 commented on code in PR #2955:
URL: https://github.com/apache/fory/pull/2955#discussion_r2577543023
##########
java/fory-core/src/main/java/org/apache/fory/serializer/collection/CollectionSerializers.java:
##########
@@ -690,6 +692,89 @@ public PriorityQueue newCollection(MemoryBuffer buffer) {
}
}
+ /**
+ * Serializer for {@link ArrayBlockingQueue}.
+ *
+ * <p>This serializer handles the capacity field which is essential for ArrayBlockingQueue since
+ * it's a bounded queue. The capacity is stored in the items array length and needs to be
+ * preserved during serialization.
+ */
+ public static class ArrayBlockingQueueSerializer
+ extends ConcurrentCollectionSerializer<ArrayBlockingQueue> {
+
+ public ArrayBlockingQueueSerializer(Fory fory, Class<ArrayBlockingQueue> cls) {
+ super(fory, cls, true);
+ }
+
+ @Override
+ public CollectionSnapshot onCollectionWrite(MemoryBuffer buffer, ArrayBlockingQueue value) {
+ CollectionSnapshot snapshot = super.onCollectionWrite(buffer, value);
+ // Write the capacity (remaining capacity + current size = total capacity)
+ int capacity = value.remainingCapacity() + value.size();
+ buffer.writeVarUint32Small7(capacity);
+ return snapshot;
+ }
+
+ @Override
+ public ArrayBlockingQueue newCollection(MemoryBuffer buffer) {
+ int numElements = buffer.readVarUint32Small7();
+ setNumElements(numElements);
+ int capacity = buffer.readVarUint32Small7();
+ ArrayBlockingQueue queue = new ArrayBlockingQueue<>(capacity);
+ fory.getRefResolver().reference(queue);
+ return queue;
+ }
+
+ @Override
+ public Collection newCollection(Collection collection) {
+ ArrayBlockingQueue abq = (ArrayBlockingQueue) collection;
+ // Capacity is remaining + current size
+ int capacity = abq.remainingCapacity() + abq.size();
Review Comment:
You're right. I've opened a new PR https://github.com/apache/fory/pull/2956 that handles the race condition. Thanks!
--
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: commits-unsubscribe@fory.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@fory.apache.org
For additional commands, e-mail: commits-help@fory.apache.org
Re: [PR] feat(java): add optimized serializers for blocking queues [fory]
Posted by "chaokunyang (via GitHub)" <gi...@apache.org>.
chaokunyang commented on code in PR #2955:
URL: https://github.com/apache/fory/pull/2955#discussion_r2577221568
##########
java/fory-core/src/main/java/org/apache/fory/serializer/collection/CollectionSerializers.java:
##########
@@ -690,6 +692,89 @@ public PriorityQueue newCollection(MemoryBuffer buffer) {
}
}
+ /**
+ * Serializer for {@link ArrayBlockingQueue}.
+ *
+ * <p>This serializer handles the capacity field which is essential for ArrayBlockingQueue since
+ * it's a bounded queue. The capacity is stored in the items array length and needs to be
+ * preserved during serialization.
+ */
+ public static class ArrayBlockingQueueSerializer
+ extends ConcurrentCollectionSerializer<ArrayBlockingQueue> {
+
+ public ArrayBlockingQueueSerializer(Fory fory, Class<ArrayBlockingQueue> cls) {
+ super(fory, cls, true);
+ }
+
+ @Override
+ public CollectionSnapshot onCollectionWrite(MemoryBuffer buffer, ArrayBlockingQueue value) {
+ CollectionSnapshot snapshot = super.onCollectionWrite(buffer, value);
+ // Write the capacity (remaining capacity + current size = total capacity)
+ int capacity = value.remainingCapacity() + value.size();
+ buffer.writeVarUint32Small7(capacity);
+ return snapshot;
+ }
+
+ @Override
+ public ArrayBlockingQueue newCollection(MemoryBuffer buffer) {
+ int numElements = buffer.readVarUint32Small7();
+ setNumElements(numElements);
+ int capacity = buffer.readVarUint32Small7();
+ ArrayBlockingQueue queue = new ArrayBlockingQueue<>(capacity);
+ fory.getRefResolver().reference(queue);
+ return queue;
+ }
+
+ @Override
+ public Collection newCollection(Collection collection) {
+ ArrayBlockingQueue abq = (ArrayBlockingQueue) collection;
+ // Capacity is remaining + current size
+ int capacity = abq.remainingCapacity() + abq.size();
Review Comment:
is it possible this creates a race condition?
After we read `remainingCapacity`, the queue has been added/removed some elements?
--
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: commits-unsubscribe@fory.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@fory.apache.org
For additional commands, e-mail: commits-help@fory.apache.org
Re: [PR] feat(java): add optimized serializers for blocking queues [fory]
Posted by "chaokunyang (via GitHub)" <gi...@apache.org>.
chaokunyang merged PR #2955:
URL: https://github.com/apache/fory/pull/2955
--
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: commits-unsubscribe@fory.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@fory.apache.org
For additional commands, e-mail: commits-help@fory.apache.org