This is an automated email from the ASF dual-hosted git repository.
chaokunyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-fury.git
The following commit(s) were added to refs/heads/main by this push:
new 124bf0ec fix(java): fix bytebuffer no such method error (#1580)
124bf0ec is described below
commit 124bf0ec3fbe6138e2e2c5aa7f1bc14c3d3b8055
Author: Shawn Yang <[email protected]>
AuthorDate: Sat Apr 27 00:19:11 2024 +0800
fix(java): fix bytebuffer no such method error (#1580)
## What does this PR do?
This PR add a bytebuffer util and fix bytebuffer no such method error
## Related issues
Closes #1577
#1579
## Does this PR introduce any user-facing change?
<!--
If any user-facing interface changes, please [open an
issue](https://github.com/apache/incubator-fury/issues/new/choose)
describing the need to do so and update the document if necessary.
-->
- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?
## Benchmark
<!--
When the PR has an impact on performance (if you don't know whether the
PR will have an impact on performance, you can submit the PR first, and
if it will have impact on performance, the code reviewer will explain
it), be sure to attach a benchmark data here.
-->
---
.../fury/benchmark/UserTypeDeserializeSuite.java | 4 +-
.../fury/benchmark/UserTypeSerializeSuite.java | 4 +-
.../org/apache/fury/benchmark/ZeroCopySuite.java | 6 +-
.../fury/benchmark/state/FlatBuffersState.java | 4 +-
.../org/apache/fury/benchmark/state/FstState.java | 6 +-
.../apache/fury/benchmark/state/JsonbState.java | 8 +-
.../org/apache/fury/io/FuryReadableChannel.java | 3 +-
.../org/apache/fury/memory/ByteBufferUtil.java | 108 +++++++++++++++++++++
.../java/org/apache/fury/memory/MemoryBuffer.java | 18 ++--
.../main/java/org/apache/fury/memory/Platform.java | 81 ----------------
.../src/test/java/org/apache/fury/StreamTest.java | 28 ------
.../org/apache/fury/memory/MemoryBufferTest.java | 7 +-
.../fury/serializer/BufferSerializersTest.java | 5 +-
.../java/org/apache/fury/util/PlatformTest.java | 11 ++-
14 files changed, 148 insertions(+), 145 deletions(-)
diff --git
a/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeDeserializeSuite.java
b/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeDeserializeSuite.java
index 94d24784..f18acc5a 100644
---
a/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeDeserializeSuite.java
+++
b/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeDeserializeSuite.java
@@ -30,7 +30,7 @@ import org.apache.fury.benchmark.state.KryoState;
import org.apache.fury.benchmark.state.ObjectType;
import org.apache.fury.benchmark.state.ProtoBuffersState;
import org.apache.fury.benchmark.state.ProtostuffState;
-import org.apache.fury.memory.Platform;
+import org.apache.fury.memory.ByteBufferUtil;
import org.openjdk.jmh.Main;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -125,7 +125,7 @@ public class UserTypeDeserializeSuite {
@Benchmark
public Object
flatbuffers_deserialize(FlatBuffersState.FlatBuffersUserTypeState state) {
- Platform.clearBuffer(state.deserializedData);
+ ByteBufferUtil.clearBuffer(state.deserializedData);
if (state.objectType == ObjectType.SAMPLE) {
return FlatBuffersState.deserializeSample(state.deserializedData);
} else {
diff --git
a/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeSerializeSuite.java
b/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeSerializeSuite.java
index c563f099..b6697025 100644
---
a/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeSerializeSuite.java
+++
b/java/benchmark/src/main/java/org/apache/fury/benchmark/UserTypeSerializeSuite.java
@@ -33,7 +33,7 @@ import org.apache.fury.benchmark.state.KryoState;
import org.apache.fury.benchmark.state.ObjectType;
import org.apache.fury.benchmark.state.ProtoBuffersState;
import org.apache.fury.benchmark.state.ProtostuffState;
-import org.apache.fury.memory.Platform;
+import org.apache.fury.memory.ByteBufferUtil;
import org.openjdk.jmh.Main;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -133,7 +133,7 @@ public class UserTypeSerializeSuite {
@Benchmark
public Object
flatbuffers_serialize(FlatBuffersState.FlatBuffersUserTypeState state) {
- Platform.clearBuffer(state.directBuffer);
+ ByteBufferUtil.clearBuffer(state.directBuffer);
if (state.objectType == ObjectType.SAMPLE) {
return FlatBuffersState.serializeSample((Sample) state.object,
state.directBuffer);
} else {
diff --git
a/java/benchmark/src/main/java/org/apache/fury/benchmark/ZeroCopySuite.java
b/java/benchmark/src/main/java/org/apache/fury/benchmark/ZeroCopySuite.java
index 9861519c..0e06f450 100644
--- a/java/benchmark/src/main/java/org/apache/fury/benchmark/ZeroCopySuite.java
+++ b/java/benchmark/src/main/java/org/apache/fury/benchmark/ZeroCopySuite.java
@@ -41,9 +41,9 @@ import org.apache.fury.benchmark.data.ComparableByteArray;
import org.apache.fury.benchmark.data.SerializableByteBuffer;
import org.apache.fury.benchmark.state.BufferType;
import org.apache.fury.config.Language;
+import org.apache.fury.memory.ByteBufferUtil;
import org.apache.fury.memory.MemoryBuffer;
import org.apache.fury.memory.MemoryUtils;
-import org.apache.fury.memory.Platform;
import org.apache.fury.serializer.BufferObject;
import org.apache.fury.test.bean.ArraysData;
import org.apache.fury.util.Preconditions;
@@ -314,7 +314,7 @@ public class ZeroCopySuite {
public static Object jsonbSerialize(JsonBState state, Blackhole bh) {
byte[] bytes = JSONB.toBytes(state.data, state.jsonbWriteFeatures);
if (state.bufferType == BufferType.directBuffer) {
- Platform.clearBuffer(state.directBuffer);
+ ByteBufferUtil.clearBuffer(state.directBuffer);
state.directBuffer.put(bytes);
}
if (bh != null) {
@@ -331,7 +331,7 @@ public class ZeroCopySuite {
public static Object jsonbDeserialize(JsonBState state, Blackhole bh) {
if (state.bufferType == BufferType.directBuffer) {
- Platform.rewind(state.directBuffer);
+ ByteBufferUtil.rewind(state.directBuffer);
byte[] bytes = new byte[state.buffer.length];
state.directBuffer.get(bytes);
Object newObj = JSONB.parseObject(bytes, Object.class,
state.jsonbReaderFeatures);
diff --git
a/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FlatBuffersState.java
b/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FlatBuffersState.java
index d99ad052..ba6ae8c5 100644
---
a/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FlatBuffersState.java
+++
b/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FlatBuffersState.java
@@ -39,7 +39,7 @@ import org.apache.fury.benchmark.state.generated.FBSImage;
import org.apache.fury.benchmark.state.generated.FBSMedia;
import org.apache.fury.benchmark.state.generated.FBSMediaContent;
import org.apache.fury.benchmark.state.generated.FBSSample;
-import org.apache.fury.memory.Platform;
+import org.apache.fury.memory.ByteBufferUtil;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Param;
@@ -439,7 +439,7 @@ public class FlatBuffersState {
} else {
deserializedData = ByteBuffer.wrap(data);
}
- Platform.clearBuffer(deserializedData);
+ ByteBufferUtil.clearBuffer(deserializedData);
Object newObj = deserializeFunc.apply(deserializedData);
Preconditions.checkArgument(object.equals(newObj));
}
diff --git
a/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FstState.java
b/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FstState.java
index 2c87cce1..32a84d38 100644
--- a/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FstState.java
+++ b/java/benchmark/src/main/java/org/apache/fury/benchmark/state/FstState.java
@@ -29,7 +29,7 @@ import org.apache.fury.benchmark.data.Data;
import org.apache.fury.benchmark.data.Image;
import org.apache.fury.benchmark.data.Media;
import org.apache.fury.benchmark.data.MediaContent;
-import org.apache.fury.memory.Platform;
+import org.apache.fury.memory.ByteBufferUtil;
import org.apache.fury.util.Preconditions;
import org.nustaq.serialization.FSTConfiguration;
import org.openjdk.jmh.annotations.CompilerControl;
@@ -93,7 +93,7 @@ public class FstState {
Blackhole blackhole) {
byte[] bytes = fst.asSharedByteArray(value, out);
if (bufferType == BufferType.directBuffer) {
- Platform.clearBuffer(directBuffer);
+ ByteBufferUtil.clearBuffer(directBuffer);
directBuffer.put(bytes, 0, out[0]);
}
if (blackhole != null) {
@@ -116,7 +116,7 @@ public class FstState {
ByteBuffer directBuffer,
Blackhole blackhole) {
if (bufferType == BufferType.directBuffer) {
- Platform.rewind(directBuffer);
+ ByteBufferUtil.rewind(directBuffer);
byte[] bytes = new byte[out[0]];
directBuffer.get(bytes);
Object newObj = fst.asObject(bytes);
diff --git
a/java/benchmark/src/main/java/org/apache/fury/benchmark/state/JsonbState.java
b/java/benchmark/src/main/java/org/apache/fury/benchmark/state/JsonbState.java
index bec24c63..87da1081 100644
---
a/java/benchmark/src/main/java/org/apache/fury/benchmark/state/JsonbState.java
+++
b/java/benchmark/src/main/java/org/apache/fury/benchmark/state/JsonbState.java
@@ -38,7 +38,7 @@ import org.apache.fury.Fury;
import org.apache.fury.benchmark.data.CustomJDKSerialization;
import org.apache.fury.logging.Logger;
import org.apache.fury.logging.LoggerFactory;
-import org.apache.fury.memory.Platform;
+import org.apache.fury.memory.ByteBufferUtil;
import org.apache.fury.util.Preconditions;
import org.openjdk.jmh.annotations.CompilerControl;
import org.openjdk.jmh.annotations.Fork;
@@ -90,7 +90,7 @@ public class JsonbState {
"======> Jsonb | {} | {} | {} | {} |", objectType, references,
bufferType, buffer.length);
if (bufferType == BufferType.directBuffer) {
directBuffer.put(buffer);
- Platform.clearBuffer(directBuffer);
+ ByteBufferUtil.clearBuffer(directBuffer);
}
Preconditions.checkArgument(object.equals(deserialize(null, this)));
}
@@ -127,7 +127,7 @@ public class JsonbState {
public static byte[] serialize(Blackhole blackhole, JsonbBenchmarkState
state, Object value) {
byte[] bytes = JSONB.toBytes(value, state.jsonbWriteFeatures);
if (state.bufferType == BufferType.directBuffer) {
- Platform.clearBuffer(state.directBuffer);
+ ByteBufferUtil.clearBuffer(state.directBuffer);
state.directBuffer.put(bytes);
}
if (blackhole != null) {
@@ -139,7 +139,7 @@ public class JsonbState {
public static Object deserialize(Blackhole blackhole, JsonbBenchmarkState
state) {
if (state.bufferType == BufferType.directBuffer) {
- Platform.rewind(state.directBuffer);
+ ByteBufferUtil.rewind(state.directBuffer);
byte[] bytes = new byte[state.buffer.length];
state.directBuffer.get(bytes);
Object newObj = JSONB.parseObject(bytes, Object.class,
state.jsonbReaderFeatures);
diff --git
a/java/fury-core/src/main/java/org/apache/fury/io/FuryReadableChannel.java
b/java/fury-core/src/main/java/org/apache/fury/io/FuryReadableChannel.java
index fd4d28c8..293620b7 100644
--- a/java/fury-core/src/main/java/org/apache/fury/io/FuryReadableChannel.java
+++ b/java/fury-core/src/main/java/org/apache/fury/io/FuryReadableChannel.java
@@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.ReadableByteChannel;
import javax.annotation.concurrent.NotThreadSafe;
import org.apache.fury.exception.DeserializationException;
+import org.apache.fury.memory.ByteBufferUtil;
import org.apache.fury.memory.MemoryBuffer;
import org.apache.fury.memory.Platform;
import org.apache.fury.util.Preconditions;
@@ -60,7 +61,7 @@ public class FuryReadableChannel implements FuryStreamReader,
ReadableByteChanne
byteBuf.position(0);
newByteBuf.put(byteBuf);
byteBuf = byteBuffer = newByteBuf;
- memoryBuf.initDirectBuffer(Platform.getAddress(byteBuf), position,
byteBuf);
+ memoryBuf.initDirectBuffer(ByteBufferUtil.getAddress(byteBuf),
position, byteBuf);
}
byteBuf.limit(newLimit);
int readCount = channel.read(byteBuf);
diff --git
a/java/fury-core/src/main/java/org/apache/fury/memory/ByteBufferUtil.java
b/java/fury-core/src/main/java/org/apache/fury/memory/ByteBufferUtil.java
new file mode 100644
index 00000000..94f5bc81
--- /dev/null
+++ b/java/fury-core/src/main/java/org/apache/fury/memory/ByteBufferUtil.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.fury.memory;
+
+import java.lang.reflect.Field;
+import java.nio.Buffer;
+import java.nio.ByteBuffer;
+import org.apache.fury.util.Preconditions;
+
+public class ByteBufferUtil {
+ private static final long BUFFER_ADDRESS_FIELD_OFFSET;
+ private static final long BUFFER_CAPACITY_FIELD_OFFSET;
+
+ static {
+ try {
+ Field addressField = Buffer.class.getDeclaredField("address");
+ BUFFER_ADDRESS_FIELD_OFFSET = Platform.objectFieldOffset(addressField);
+ Preconditions.checkArgument(BUFFER_ADDRESS_FIELD_OFFSET != 0);
+ Field capacityField = Buffer.class.getDeclaredField("capacity");
+ BUFFER_CAPACITY_FIELD_OFFSET = Platform.objectFieldOffset(capacityField);
+ Preconditions.checkArgument(BUFFER_CAPACITY_FIELD_OFFSET != 0);
+ } catch (NoSuchFieldException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ public static long getAddress(ByteBuffer buffer) {
+ Preconditions.checkNotNull(buffer, "buffer is null");
+ Preconditions.checkArgument(buffer.isDirect(), "Can't get address of a
non-direct ByteBuffer.");
+ long offHeapAddress;
+ try {
+ offHeapAddress = Platform.getLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET);
+ } catch (Throwable t) {
+ throw new Error("Could not access direct byte buffer address field.", t);
+ }
+ return offHeapAddress;
+ }
+
+ private static final ByteBuffer localBuffer = ByteBuffer.allocateDirect(0);
+
+ /** Create a direct buffer from native memory represented by address
[address, address + size). */
+ public static ByteBuffer createDirectByteBufferFromNativeAddress(long
address, int size) {
+ try {
+ // ByteBuffer.allocateDirect(0) is about 30x slower than
`localBuffer.duplicate()`.
+ ByteBuffer buffer = localBuffer.duplicate();
+ Platform.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);
+ Platform.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);
+ buffer.clear();
+ return buffer;
+ } catch (Throwable t) {
+ throw new Error("Failed to wrap unsafe off-heap memory with ByteBuffer",
t);
+ }
+ }
+
+ /** Wrap a buffer [address, address + size) into provided
<code>buffer</code>. */
+ public static void wrapDirectByteBufferFromNativeAddress(
+ ByteBuffer buffer, long address, int size) {
+ Preconditions.checkArgument(
+ buffer.isDirect(), "Can't wrap native memory into a non-direct
ByteBuffer.");
+ Platform.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);
+ Platform.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);
+ buffer.clear();
+ }
+
+ public static ByteBuffer wrapDirectBuffer(long address, int size) {
+ return createDirectByteBufferFromNativeAddress(address, size);
+ }
+
+ /** Wrap a buffer [address, address + size) into provided
<code>buffer</code>. */
+ public static void wrapDirectBuffer(ByteBuffer buffer, long address, int
size) {
+ Platform.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);
+ Platform.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);
+ buffer.clear();
+ }
+
+ public static void clearBuffer(Buffer buffer) {
+ buffer.clear();
+ }
+
+ public static void flipBuffer(Buffer buffer) {
+ buffer.flip();
+ }
+
+ public static void rewind(Buffer buffer) {
+ buffer.rewind();
+ }
+
+ public static void position(Buffer buffer, int pos) {
+ buffer.position(pos);
+ }
+}
diff --git
a/java/fury-core/src/main/java/org/apache/fury/memory/MemoryBuffer.java
b/java/fury-core/src/main/java/org/apache/fury/memory/MemoryBuffer.java
index 24e0a224..f5b1eea4 100644
--- a/java/fury-core/src/main/java/org/apache/fury/memory/MemoryBuffer.java
+++ b/java/fury-core/src/main/java/org/apache/fury/memory/MemoryBuffer.java
@@ -329,7 +329,7 @@ public final class MemoryBuffer {
}
final int targetPos = target.position();
if (target.isDirect()) {
- final long targetAddr = Platform.getAddress(target) + targetPos;
+ final long targetAddr = ByteBufferUtil.getAddress(target) + targetPos;
final long sourceAddr = address + offset;
if (sourceAddr <= addressLimit - numBytes) {
Platform.copyMemory(heapMemory, sourceAddr, null, targetAddr,
numBytes);
@@ -340,7 +340,7 @@ public final class MemoryBuffer {
assert target.hasArray();
get(offset, target.array(), targetPos + target.arrayOffset(), numBytes);
}
- target.position(targetPos + numBytes);
+ ByteBufferUtil.position(target, targetPos + numBytes);
}
public void put(int offset, ByteBuffer source, int numBytes) {
@@ -350,7 +350,7 @@ public final class MemoryBuffer {
}
final int sourcePos = source.position();
if (source.isDirect()) {
- final long sourceAddr = Platform.getAddress(source) + sourcePos;
+ final long sourceAddr = ByteBufferUtil.getAddress(source) + sourcePos;
final long targetAddr = address + offset;
if (targetAddr <= addressLimit - numBytes) {
Platform.copyMemory(null, sourceAddr, heapMemory, targetAddr,
numBytes);
@@ -361,7 +361,7 @@ public final class MemoryBuffer {
assert source.hasArray();
put(offset, source.array(), sourcePos + source.arrayOffset(), numBytes);
}
- source.position(sourcePos + numBytes);
+ ByteBufferUtil.position(source, sourcePos + numBytes);
}
public void put(int index, byte[] src) {
@@ -2465,12 +2465,12 @@ public final class MemoryBuffer {
ByteBuffer offHeapBuffer = this.offHeapBuffer;
if (offHeapBuffer != null) {
ByteBuffer duplicate = offHeapBuffer.duplicate();
- int start = (int) (address - Platform.getAddress(duplicate));
- duplicate.position(start + offset);
+ int start = (int) (address - ByteBufferUtil.getAddress(duplicate));
+ ByteBufferUtil.position(duplicate, start + offset);
duplicate.limit(start + offset + length);
return duplicate.slice();
} else {
- return Platform.createDirectByteBufferFromNativeAddress(address +
offset, length);
+ return ByteBufferUtil.createDirectByteBufferFromNativeAddress(address
+ offset, length);
}
}
}
@@ -2553,7 +2553,7 @@ public final class MemoryBuffer {
public static MemoryBuffer fromByteBuffer(ByteBuffer buffer) {
if (buffer.isDirect()) {
return new MemoryBuffer(
- Platform.getAddress(buffer) + buffer.position(), buffer.remaining(),
buffer);
+ ByteBufferUtil.getAddress(buffer) + buffer.position(),
buffer.remaining(), buffer);
} else {
int offset = buffer.arrayOffset() + buffer.position();
return new MemoryBuffer(buffer.array(), offset, buffer.remaining());
@@ -2562,7 +2562,7 @@ public final class MemoryBuffer {
public static MemoryBuffer fromDirectByteBuffer(
ByteBuffer buffer, int size, FuryStreamReader streamReader) {
- long offHeapAddress = Platform.getAddress(buffer) + buffer.position();
+ long offHeapAddress = ByteBufferUtil.getAddress(buffer) +
buffer.position();
return new MemoryBuffer(offHeapAddress, size, buffer, streamReader);
}
diff --git a/java/fury-core/src/main/java/org/apache/fury/memory/Platform.java
b/java/fury-core/src/main/java/org/apache/fury/memory/Platform.java
index d8d49e0b..aa9464a2 100644
--- a/java/fury-core/src/main/java/org/apache/fury/memory/Platform.java
+++ b/java/fury-core/src/main/java/org/apache/fury/memory/Platform.java
@@ -19,10 +19,8 @@ package org.apache.fury.memory;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
-import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
-import org.apache.fury.util.Preconditions;
import org.apache.fury.util.unsafe._JDKAccess;
import sun.misc.Unsafe;
@@ -107,24 +105,6 @@ public final class Platform {
unaligned = unalign;
}
- // Access fields and constructors once and store them, for performance:
-
- private static final long BUFFER_ADDRESS_FIELD_OFFSET;
- private static final long BUFFER_CAPACITY_FIELD_OFFSET;
-
- static {
- try {
- Field addressField = Buffer.class.getDeclaredField("address");
- BUFFER_ADDRESS_FIELD_OFFSET = UNSAFE.objectFieldOffset(addressField);
- Preconditions.checkArgument(BUFFER_ADDRESS_FIELD_OFFSET != 0);
- Field capacityField = Buffer.class.getDeclaredField("capacity");
- BUFFER_CAPACITY_FIELD_OFFSET = UNSAFE.objectFieldOffset(capacityField);
- Preconditions.checkArgument(BUFFER_CAPACITY_FIELD_OFFSET != 0);
- } catch (NoSuchFieldException e) {
- throw new IllegalStateException(e);
- }
- }
-
/**
* Returns true when running JVM is having sun's Unsafe package available in
it and underlying
* system having unaligned-access capability.
@@ -316,65 +296,4 @@ public final class Platform {
}
throw new IllegalStateException("unreachable");
}
-
- public static long getAddress(ByteBuffer buffer) {
- Preconditions.checkNotNull(buffer, "buffer is null");
- Preconditions.checkArgument(buffer.isDirect(), "Can't get address of a
non-direct ByteBuffer.");
- long offHeapAddress;
- try {
- offHeapAddress = UNSAFE.getLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET);
- } catch (Throwable t) {
- throw new Error("Could not access direct byte buffer address field.", t);
- }
- return offHeapAddress;
- }
-
- private static final ByteBuffer localBuffer = ByteBuffer.allocateDirect(0);
-
- /** Create a direct buffer from native memory represented by address
[address, address + size). */
- public static ByteBuffer createDirectByteBufferFromNativeAddress(long
address, int size) {
- try {
- // ByteBuffer.allocateDirect(0) is about 30x slower than
`localBuffer.duplicate()`.
- ByteBuffer buffer = localBuffer.duplicate();
- UNSAFE.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);
- UNSAFE.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);
- buffer.clear();
- return buffer;
- } catch (Throwable t) {
- throw new Error("Failed to wrap unsafe off-heap memory with ByteBuffer",
t);
- }
- }
-
- /** Wrap a buffer [address, address + size) into provided
<code>buffer</code>. */
- public static void wrapDirectByteBufferFromNativeAddress(
- ByteBuffer buffer, long address, int size) {
- Preconditions.checkArgument(
- buffer.isDirect(), "Can't wrap native memory into a non-direct
ByteBuffer.");
- UNSAFE.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);
- UNSAFE.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);
- buffer.clear();
- }
-
- public static ByteBuffer wrapDirectBuffer(long address, int size) {
- return createDirectByteBufferFromNativeAddress(address, size);
- }
-
- /** Wrap a buffer [address, address + size) into provided
<code>buffer</code>. */
- public static void wrapDirectBuffer(ByteBuffer buffer, long address, int
size) {
- UNSAFE.putLong(buffer, BUFFER_ADDRESS_FIELD_OFFSET, address);
- UNSAFE.putInt(buffer, BUFFER_CAPACITY_FIELD_OFFSET, size);
- buffer.clear();
- }
-
- public static void clearBuffer(Buffer buffer) {
- buffer.clear();
- }
-
- public static void flipBuffer(Buffer buffer) {
- buffer.flip();
- }
-
- public static void rewind(Buffer buffer) {
- buffer.rewind();
- }
}
diff --git a/java/fury-core/src/test/java/org/apache/fury/StreamTest.java
b/java/fury-core/src/test/java/org/apache/fury/StreamTest.java
index 0d075e8d..95d783ac 100644
--- a/java/fury-core/src/test/java/org/apache/fury/StreamTest.java
+++ b/java/fury-core/src/test/java/org/apache/fury/StreamTest.java
@@ -28,7 +28,6 @@ import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
import org.apache.fury.io.FuryInputStream;
@@ -40,33 +39,6 @@ import org.apache.fury.util.ReflectionUtils;
import org.testng.annotations.Test;
public class StreamTest {
- // Note that Fury instances should be reused between
- // multiple serializations of different objects.
- private static final ThreadSafeFury fury =
- new ThreadLocalFury(
- classLoader -> {
- Fury f =
-
Fury.builder().withClassLoader(classLoader).requireClassRegistration(false).build();
- return f;
- });
-
- public static byte[] encoder(Object object) {
- return fury.serialize(object);
- }
-
- public static <T> T decoder(byte[] bytes) {
- return (T) fury.deserialize(bytes);
- }
-
- public static void main(String[] args) {
- byte len = 10;
- ByteBuffer byteBuffer = ByteBuffer.allocate(len);
- for (int i = 0; i < len; i++) {
- byteBuffer.put((byte) i);
- }
- System.out.println(encoder(byteBuffer).length);
- }
-
@Test
public void testBufferStream() {
MemoryBuffer buffer0 = MemoryBuffer.newHeapBuffer(10);
diff --git
a/java/fury-core/src/test/java/org/apache/fury/memory/MemoryBufferTest.java
b/java/fury-core/src/test/java/org/apache/fury/memory/MemoryBufferTest.java
index bc65ce74..48ce9962 100644
--- a/java/fury-core/src/test/java/org/apache/fury/memory/MemoryBufferTest.java
+++ b/java/fury-core/src/test/java/org/apache/fury/memory/MemoryBufferTest.java
@@ -174,19 +174,20 @@ public class MemoryBufferTest {
MemoryBuffer buffer = MemoryUtils.wrap(direct);
assertEquals(buffer.sliceAsByteBuffer(), direct);
Assert.assertEquals(
- Platform.getAddress(buffer.sliceAsByteBuffer()),
Platform.getAddress(direct) + 5);
+ ByteBufferUtil.getAddress(buffer.sliceAsByteBuffer()),
+ ByteBufferUtil.getAddress(direct) + 5);
}
{
long address = 0;
try {
address = Platform.allocateMemory(10);
- ByteBuffer direct = Platform.wrapDirectBuffer(address, 10);
+ ByteBuffer direct = ByteBufferUtil.wrapDirectBuffer(address, 10);
direct.put(data);
direct.flip();
direct.position(5);
MemoryBuffer buffer = MemoryUtils.wrap(direct);
assertEquals(buffer.sliceAsByteBuffer(), direct);
- assertEquals(Platform.getAddress(buffer.sliceAsByteBuffer()), address
+ 5);
+ assertEquals(ByteBufferUtil.getAddress(buffer.sliceAsByteBuffer()),
address + 5);
} finally {
Platform.freeMemory(address);
}
diff --git
a/java/fury-core/src/test/java/org/apache/fury/serializer/BufferSerializersTest.java
b/java/fury-core/src/test/java/org/apache/fury/serializer/BufferSerializersTest.java
index 675d1235..b4f9301c 100644
---
a/java/fury-core/src/test/java/org/apache/fury/serializer/BufferSerializersTest.java
+++
b/java/fury-core/src/test/java/org/apache/fury/serializer/BufferSerializersTest.java
@@ -22,6 +22,7 @@ package org.apache.fury.serializer;
import java.nio.ByteBuffer;
import org.apache.fury.Fury;
import org.apache.fury.FuryTestBase;
+import org.apache.fury.memory.ByteBufferUtil;
import org.testng.annotations.Test;
public class BufferSerializersTest extends FuryTestBase {
@@ -31,11 +32,11 @@ public class BufferSerializersTest extends FuryTestBase {
Fury fury = Fury.builder().build();
ByteBuffer buffer1 = ByteBuffer.allocate(32);
buffer1.putLong(1000L);
- buffer1.rewind();
+ ByteBufferUtil.rewind(buffer1);
serDeCheck(fury, buffer1);
ByteBuffer buffer2 = ByteBuffer.allocateDirect(32);
buffer2.putDouble(1.0 / 3);
- buffer2.rewind();
+ ByteBufferUtil.rewind(buffer2);
serDeCheck(fury, buffer2);
}
}
diff --git
a/java/fury-core/src/test/java/org/apache/fury/util/PlatformTest.java
b/java/fury-core/src/test/java/org/apache/fury/util/PlatformTest.java
index 7ced538b..262ad923 100644
--- a/java/fury-core/src/test/java/org/apache/fury/util/PlatformTest.java
+++ b/java/fury-core/src/test/java/org/apache/fury/util/PlatformTest.java
@@ -27,6 +27,7 @@ import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import org.apache.fury.logging.Logger;
import org.apache.fury.logging.LoggerFactory;
+import org.apache.fury.memory.ByteBufferUtil;
import org.apache.fury.memory.Platform;
import org.testng.annotations.Test;
@@ -104,7 +105,7 @@ public class PlatformTest {
try {
int size = 16;
address = Platform.allocateMemory(size);
- ByteBuffer buffer = Platform.wrapDirectBuffer(address, size);
+ ByteBuffer buffer = ByteBufferUtil.wrapDirectBuffer(address, size);
buffer.putLong(0, 1);
assertEquals(1, buffer.getLong(0));
} finally {
@@ -122,11 +123,11 @@ public class PlatformTest {
ByteBuffer buffer = null;
{
for (int i = 0; i < nums; i++) {
- buffer = Platform.wrapDirectBuffer(address, size);
+ buffer = ByteBufferUtil.wrapDirectBuffer(address, size);
}
long startTime = System.nanoTime();
for (int i = 0; i < nums; i++) {
- buffer = Platform.wrapDirectBuffer(address, size);
+ buffer = ByteBufferUtil.wrapDirectBuffer(address, size);
}
long duration = System.nanoTime() - startTime;
buffer.putLong(0, 1);
@@ -134,11 +135,11 @@ public class PlatformTest {
}
{
for (int i = 0; i < nums; i++) {
- Platform.wrapDirectBuffer(buffer, address, size);
+ ByteBufferUtil.wrapDirectBuffer(buffer, address, size);
}
long startTime = System.nanoTime();
for (int i = 0; i < nums; i++) {
- Platform.wrapDirectBuffer(buffer, address, size);
+ ByteBufferUtil.wrapDirectBuffer(buffer, address, size);
}
long duration = System.nanoTime() - startTime;
buffer.putLong(0, 1);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
