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 46db7e0a feat(java): optimize string serialization by concating coder
and length (#1486)
46db7e0a is described below
commit 46db7e0ae4948f2a2924fa7993d351a8fa205b6d
Author: Shawn Yang <[email protected]>
AuthorDate: Fri Apr 12 14:21:58 2024 +0800
feat(java): optimize string serialization by concating coder and length
(#1486)
<!--
**Thanks for contributing to Fury.**
**If this is your first time opening a PR on fury, you can refer to
[CONTRIBUTING.md](https://github.com/apache/incubator-fury/blob/main/CONTRIBUTING.md).**
Contribution Checklist
- The **Apache Fury (incubating)** community has restrictions on the
naming of pr titles. You can also find instructions in
[CONTRIBUTING.md](https://github.com/apache/incubator-fury/blob/main/CONTRIBUTING.md).
- Fury has a strong focus on performance. If the PR you submit will have
an impact on performance, please benchmark it first and provide the
benchmark result here.
-->
## What does this PR do?
This PR optimize string serialization by concating coder and length into
a long and serialize it together, it can save one byte for small string
which the length is less than 32
<!-- Describe the purpose of this PR. -->
## Related issues
#1240
<!--
Is there any related issue? Please attach here.
- #xxxx0
- #xxxx1
- #xxxx2
-->
## 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
Before
```
Benchmark (bufferType) (objectType)
(references) Mode Cnt Score Error Units
UserTypeDeserializeSuite.fury_deserialize array MEDIA_CONTENT
false thrpt 50 2576580.395 ± 55698.851 ops/s
```
This PR:
```java
Benchmark (bufferType) (objectType)
(references) Mode Cnt Score Error Units
UserTypeDeserializeSuite.fury_deserialize array MEDIA_CONTENT
false thrpt 50 2591695.102 ± 48174.940 ops/s
```
<!--
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.
-->
---
.../integration_tests/JDKCompatibilityTest.java | 3 +-
.../java/org/apache/fury/memory/MemoryBuffer.java | 263 ++++++++++++---------
.../java/org/apache/fury/memory/MemoryUtils.java | 8 +-
.../org/apache/fury/serializer/Serializers.java | 11 +-
.../apache/fury/serializer/StringSerializer.java | 166 ++++++-------
.../org/apache/fury/memory/MemoryBufferTest.java | 36 +++
.../fury/serializer/StringSerializerTest.java | 13 +-
7 files changed, 292 insertions(+), 208 deletions(-)
diff --git
a/integration_tests/jdk_compatibility_tests/src/test/java/org/apache/fury/integration_tests/JDKCompatibilityTest.java
b/integration_tests/jdk_compatibility_tests/src/test/java/org/apache/fury/integration_tests/JDKCompatibilityTest.java
index f191277f..53d2ee09 100644
---
a/integration_tests/jdk_compatibility_tests/src/test/java/org/apache/fury/integration_tests/JDKCompatibilityTest.java
+++
b/integration_tests/jdk_compatibility_tests/src/test/java/org/apache/fury/integration_tests/JDKCompatibilityTest.java
@@ -29,7 +29,6 @@ import org.apache.fury.Fury;
import org.apache.fury.config.CompatibleMode;
import org.apache.fury.config.FuryBuilder;
import org.apache.fury.config.Language;
-import org.apache.fury.test.bean.BeanA;
import org.apache.fury.util.Platform;
import org.testng.Assert;
import org.testng.annotations.Test;
@@ -42,7 +41,7 @@ public class JDKCompatibilityTest {
Object createObject() {
// test non latin1 string
- return Arrays.asList("Hello", "Hello，你好", BeanA.createBeanA(2));
+ return Arrays.asList("Hello", "Hello，你好");
}
@Test
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 68630df1..7f2be270 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
@@ -1252,7 +1252,7 @@ public final class MemoryBuffer {
* to avoid using two memory operations.
*/
public int unsafeWritePositiveVarInt(int v) {
- int varintBytes = unsafePutPositiveVarInt(writerIndex, v);
+ int varintBytes = unsafePutVarUint36Small(writerIndex, v);
writerIndex += varintBytes;
return varintBytes;
}
@@ -1260,40 +1260,110 @@ public final class MemoryBuffer {
/**
* Caller must ensure there must be at least 8 bytes for writing, otherwise
the crash may occur.
*/
- public int unsafePutPositiveVarInt(int index, int value) {
- // The encoding algorithm are based on kryo UnsafeMemoryOutput.writeVarInt
- // varint are written using little endian byte order.
- // This version should have better performance since it remove an index
update.
- long varInt = (value & 0x7F);
+ public int unsafePutVarUint36Small(int index, long value) {
+ long encoded = (value & 0x7F);
if (value >>> 7 == 0) {
- UNSAFE.putByte(heapMemory, address + index, (byte) varInt);
+ UNSAFE.putByte(heapMemory, address + index, (byte) value);
return 1;
}
// bit 8 `set` indicates have next data bytes.
// 0x3f80: 0b1111111 << 7
- varInt |= (((value & 0x3f80) << 1) | 0x80);
+ encoded |= (((value & 0x3f80) << 1) | 0x80);
if (value >>> 14 == 0) {
- unsafePutInt(index, (int) varInt);
+ unsafePutInt(index, (int) encoded);
return 2;
}
+ return continuePutVarInt36(index, encoded, value);
+ }
+
+ private int continuePutVarInt36(int index, long encoded, long value) {
// 0x1fc000: 0b1111111 << 14
- varInt |= (((value & 0x1fc000) << 2) | 0x8000);
+ encoded |= (((value & 0x1fc000) << 2) | 0x8000);
if (value >>> 21 == 0) {
- unsafePutInt(index, (int) varInt);
+ unsafePutInt(index, (int) encoded);
return 3;
}
// 0xfe00000: 0b1111111 << 21
- varInt |= ((value & 0xfe00000) << 3) | 0x800000;
+ encoded |= ((value & 0xfe00000) << 3) | 0x800000;
if (value >>> 28 == 0) {
- unsafePutInt(index, (int) varInt);
+ unsafePutInt(index, (int) encoded);
return 4;
}
- // 0xfe00000: 0b1111111 << 28
- varInt |= ((value & 0x7f0000000L) << 4) | 0x80000000L;
- unsafePutLong(index, varInt);
+ // 0xff0000000: 0b11111111 << 28. Note eight `1` here instead of seven.
+ encoded |= ((value & 0xff0000000L) << 4) | 0x80000000L;
+ unsafePutLong(index, encoded);
return 5;
}
+ public long readVarUint36Small() {
+ // Duplicate and manual inline for performance.
+ // noinspection Duplicates
+ int readIdx = readerIndex;
+ if (size - readIdx >= 9) {
+ long bulkValue = unsafeGetLong(readIdx++);
+ // noinspection Duplicates
+ long result = bulkValue & 0x7F;
+ if ((bulkValue & 0x80) != 0) {
+ readIdx++;
+ // 0x3f80: 0b1111111 << 7
+ result |= (bulkValue >>> 1) & 0x3f80;
+ // 0x8000: 0b1 << 15
+ if ((bulkValue & 0x8000) != 0) {
+ return continueReadVarLong36(readIdx, bulkValue, result);
+ }
+ }
+ readerIndex = readIdx;
+ return result;
+ } else {
+ return readVarUint36Slow();
+ }
+ }
+
+ private long continueReadVarLong36(int readIdx, long bulkValue, long result)
{
+ readIdx++;
+ // 0x1fc000: 0b1111111 << 14
+ result |= (bulkValue >>> 2) & 0x1fc000;
+ // 0x800000: 0b1 << 23
+ if ((bulkValue & 0x800000) != 0) {
+ readIdx++;
+ // 0xfe00000: 0b1111111 << 21
+ result |= (bulkValue >>> 3) & 0xfe00000;
+ if ((bulkValue & 0x80000000L) != 0) {
+ readIdx++;
+ // 0xff0000000: 0b11111111 << 28
+ result |= (bulkValue >>> 4) & 0xff0000000L;
+ }
+ }
+ readerIndex = readIdx;
+ return result;
+ }
+
+ private long readVarUint36Slow() {
+ long b = readByte();
+ long result = b & 0x7F;
+ // Note:
+ // Loop are not used here to improve performance.
+ // We manually unroll the loop for better performance.
+ // noinspection Duplicates
+ if ((b & 0x80) != 0) {
+ b = readByte();
+ result |= (b & 0x7F) << 7;
+ if ((b & 0x80) != 0) {
+ b = readByte();
+ result |= (b & 0x7F) << 14;
+ if ((b & 0x80) != 0) {
+ b = readByte();
+ result |= (b & 0x7F) << 21;
+ if ((b & 0x80) != 0) {
+ b = readByte();
+ result |= (b & 0x7F) << 28;
+ }
+ }
+ }
+ }
+ return result;
+ }
+
/** Reads the 1-5 byte int part of a non-negative varint. */
public int readPositiveVarInt() {
int readIdx = readerIndex;
@@ -1767,46 +1837,19 @@ public final class MemoryBuffer {
result = readPositiveVarLongSlow();
} else {
long address = this.address;
- long value = UNSAFE.getLong(heapMemory, address + readIdx);
+ long bulkValue = UNSAFE.getLong(heapMemory, address + readIdx);
// Duplicate and manual inline for performance.
// noinspection Duplicates
readIdx++;
- result = value & 0x7F;
- if ((value & 0x80) != 0) {
+ result = bulkValue & 0x7F;
+ if ((bulkValue & 0x80) != 0) {
readIdx++;
// 0x3f80: 0b1111111 << 7
- result |= (value >>> 1) & 0x3f80;
+ result |= (bulkValue >>> 1) & 0x3f80;
// 0x8000: 0b1 << 15
- if ((value & 0x8000) != 0) {
- readIdx++;
- // 0x1fc000: 0b1111111 << 14
- result |= (value >>> 2) & 0x1fc000;
- // 0x800000: 0b1 << 23
- if ((value & 0x800000) != 0) {
- readIdx++;
- // 0xfe00000: 0b1111111 << 21
- result |= (value >>> 3) & 0xfe00000;
- if ((value & 0x80000000L) != 0) {
- readIdx++;
- result |= (value >>> 4) & 0x7f0000000L;
- if ((value & 0x8000000000L) != 0) {
- readIdx++;
- result |= (value >>> 5) & 0x3f800000000L;
- if ((value & 0x800000000000L) != 0) {
- readIdx++;
- result |= (value >>> 6) & 0x1fc0000000000L;
- if ((value & 0x80000000000000L) != 0) {
- readIdx++;
- result |= (value >>> 7) & 0xfe000000000000L;
- if ((value & 0x8000000000000000L) != 0) {
- long b = UNSAFE.getByte(heapMemory, address + readIdx++);
- result |= b << 56;
- }
- }
- }
- }
- }
- }
+ if ((bulkValue & 0x8000) != 0) {
+ result = continueReadVarLong64(readIdx, bulkValue, result);
+ return ((result >>> 1) ^ -(result & 1));
}
}
readerIndex = readIdx;
@@ -1822,46 +1865,19 @@ public final class MemoryBuffer {
result = readPositiveVarLongSlow();
} else {
long address = this.address;
- long value = Long.reverseBytes(UNSAFE.getLong(heapMemory, address +
readIdx));
+ long bulkValue = Long.reverseBytes(UNSAFE.getLong(heapMemory, address +
readIdx));
// Duplicate and manual inline for performance.
// noinspection Duplicates
readIdx++;
- result = value & 0x7F;
- if ((value & 0x80) != 0) {
+ result = bulkValue & 0x7F;
+ if ((bulkValue & 0x80) != 0) {
readIdx++;
// 0x3f80: 0b1111111 << 7
- result |= (value >>> 1) & 0x3f80;
+ result |= (bulkValue >>> 1) & 0x3f80;
// 0x8000: 0b1 << 15
- if ((value & 0x8000) != 0) {
- readIdx++;
- // 0x1fc000: 0b1111111 << 14
- result |= (value >>> 2) & 0x1fc000;
- // 0x800000: 0b1 << 23
- if ((value & 0x800000) != 0) {
- readIdx++;
- // 0xfe00000: 0b1111111 << 21
- result |= (value >>> 3) & 0xfe00000;
- if ((value & 0x80000000L) != 0) {
- readIdx++;
- result |= (value >>> 4) & 0x7f0000000L;
- if ((value & 0x8000000000L) != 0) {
- readIdx++;
- result |= (value >>> 5) & 0x3f800000000L;
- if ((value & 0x800000000000L) != 0) {
- readIdx++;
- result |= (value >>> 6) & 0x1fc0000000000L;
- if ((value & 0x80000000000000L) != 0) {
- readIdx++;
- result |= (value >>> 7) & 0xfe000000000000L;
- if ((value & 0x8000000000000000L) != 0) {
- long b = UNSAFE.getByte(heapMemory, address + readIdx++);
- result |= b << 56;
- }
- }
- }
- }
- }
- }
+ if ((bulkValue & 0x8000) != 0) {
+ result = continueReadVarLong64(readIdx, bulkValue, result);
+ return ((result >>> 1) ^ -(result & 1));
}
}
readerIndex = readIdx;
@@ -1876,42 +1892,48 @@ public final class MemoryBuffer {
return readPositiveVarLongSlow();
}
// varint are written using little endian byte order, so read by little
endian byte order.
- long value = unsafeGetLong(readIdx);
+ long bulkValue = unsafeGetLong(readIdx);
// Duplicate and manual inline for performance.
// noinspection Duplicates
readIdx++;
- long result = value & 0x7F;
- if ((value & 0x80) != 0) {
+ long result = bulkValue & 0x7F;
+ if ((bulkValue & 0x80) != 0) {
readIdx++;
// 0x3f80: 0b1111111 << 7
- result |= (value >>> 1) & 0x3f80;
+ result |= (bulkValue >>> 1) & 0x3f80;
// 0x8000: 0b1 << 15
- if ((value & 0x8000) != 0) {
+ if ((bulkValue & 0x8000) != 0) {
+ return continueReadVarLong64(readIdx, bulkValue, result);
+ }
+ }
+ readerIndex = readIdx;
+ return result;
+ }
+
+ private long continueReadVarLong64(int readIdx, long bulkValue, long result)
{
+ readIdx++;
+ // 0x1fc000: 0b1111111 << 14
+ result |= (bulkValue >>> 2) & 0x1fc000;
+ // 0x800000: 0b1 << 23
+ if ((bulkValue & 0x800000) != 0) {
+ readIdx++;
+ // 0xfe00000: 0b1111111 << 21
+ result |= (bulkValue >>> 3) & 0xfe00000;
+ if ((bulkValue & 0x80000000L) != 0) {
readIdx++;
- // 0x1fc000: 0b1111111 << 14
- result |= (value >>> 2) & 0x1fc000;
- // 0x800000: 0b1 << 23
- if ((value & 0x800000) != 0) {
+ result |= (bulkValue >>> 4) & 0x7f0000000L;
+ if ((bulkValue & 0x8000000000L) != 0) {
readIdx++;
- // 0xfe00000: 0b1111111 << 21
- result |= (value >>> 3) & 0xfe00000;
- if ((value & 0x80000000L) != 0) {
+ result |= (bulkValue >>> 5) & 0x3f800000000L;
+ if ((bulkValue & 0x800000000000L) != 0) {
readIdx++;
- result |= (value >>> 4) & 0x7f0000000L;
- if ((value & 0x8000000000L) != 0) {
+ result |= (bulkValue >>> 6) & 0x1fc0000000000L;
+ if ((bulkValue & 0x80000000000000L) != 0) {
readIdx++;
- result |= (value >>> 5) & 0x3f800000000L;
- if ((value & 0x800000000000L) != 0) {
- readIdx++;
- result |= (value >>> 6) & 0x1fc0000000000L;
- if ((value & 0x80000000000000L) != 0) {
- readIdx++;
- result |= (value >>> 7) & 0xfe000000000000L;
- if ((value & 0x8000000000000000L) != 0) {
- long b = UNSAFE.getByte(heapMemory, address + readIdx++);
- result |= b << 56;
- }
- }
+ result |= (bulkValue >>> 7) & 0xfe000000000000L;
+ if ((bulkValue & 0x8000000000000000L) != 0) {
+ long b = UNSAFE.getByte(heapMemory, address + readIdx++);
+ result |= b << 56;
}
}
}
@@ -2528,6 +2550,11 @@ public final class MemoryBuffer {
final int numBytes = readBinarySize();
int readerIdx = readerIndex;
final byte[] arr = new byte[numBytes];
+ // use subtract to avoid overflow
+ if (readerIdx > size - numBytes) {
+ streamReader.readTo(arr, 0, numBytes);
+ return arr;
+ }
byte[] heapMemory = this.heapMemory;
if (heapMemory != null) {
System.arraycopy(heapMemory, heapOffset + readerIdx, arr, 0, numBytes);
@@ -2580,6 +2607,22 @@ public final class MemoryBuffer {
readerIndex = readerIdx + numBytes;
}
+ @CodegenInvoke
+ public char[] readCharsAndSize() {
+ final int numBytes = readBinarySize();
+ int readerIdx = readerIndex;
+ final char[] arr = new char[numBytes >> 1];
+ // use subtract to avoid overflow
+ if (readerIdx > size - numBytes) {
+ streamReader.readToUnsafe(arr, 0, numBytes);
+ return arr;
+ }
+ Platform.UNSAFE.copyMemory(
+ heapMemory, address + readerIdx, arr, Platform.CHAR_ARRAY_OFFSET,
numBytes);
+ readerIndex = readerIdx + numBytes;
+ return arr;
+ }
+
public char[] readCharsWithAlignedSize() {
final int numBytes = readPositiveAlignedVarInt();
return readChars(numBytes);
diff --git
a/java/fury-core/src/main/java/org/apache/fury/memory/MemoryUtils.java
b/java/fury-core/src/main/java/org/apache/fury/memory/MemoryUtils.java
index dd4fdf6c..cac4a247 100644
--- a/java/fury-core/src/main/java/org/apache/fury/memory/MemoryUtils.java
+++ b/java/fury-core/src/main/java/org/apache/fury/memory/MemoryUtils.java
@@ -130,9 +130,7 @@ public class MemoryUtils {
buffer.readerIndex(pos);
}
- public static int writePositiveVarInt(byte[] arr, int index, int v) {
- // The encoding algorithm are based on kryo UnsafeMemoryOutput.writeVarInt
- // varint are written using little endian byte order.
+ public static int putVarUint36Small(byte[] arr, int index, long v) {
if (v >>> 7 == 0) {
arr[index] = (byte) v;
return 1;
@@ -142,6 +140,10 @@ public class MemoryUtils {
arr[index] = (byte) (v >>> 7);
return 2;
}
+ return bigWriteUint36(arr, index, v);
+ }
+
+ private static int bigWriteUint36(byte[] arr, int index, long v) {
if (v >>> 21 == 0) {
arr[index++] = (byte) ((v & 0x7F) | 0x80);
arr[index++] = (byte) (v >>> 7 | 0x80);
diff --git
a/java/fury-core/src/main/java/org/apache/fury/serializer/Serializers.java
b/java/fury-core/src/main/java/org/apache/fury/serializer/Serializers.java
index 72992cba..a38a0f20 100644
--- a/java/fury-core/src/main/java/org/apache/fury/serializer/Serializers.java
+++ b/java/fury-core/src/main/java/org/apache/fury/serializer/Serializers.java
@@ -246,15 +246,16 @@ public class Serializers {
if (GET_CODER != null) {
int coder = GET_CODER.applyAsInt(value);
byte[] v = (byte[]) GET_VALUE.apply(value);
- buffer.writeByte(coder);
- if (coder == 0) {
- buffer.writePrimitiveArrayWithSize(v, Platform.BYTE_ARRAY_OFFSET,
value.length());
- } else {
+ int bytesLen = value.length();
+ if (coder != 0) {
if (coder != 1) {
throw new UnsupportedOperationException("Unsupported coder " +
coder);
}
- buffer.writePrimitiveArrayWithSize(v, Platform.BYTE_ARRAY_OFFSET,
value.length() << 1);
+ bytesLen <<= 1;
}
+ long header = ((long) bytesLen << 2) | coder;
+ buffer.writePositiveVarLong(header);
+ buffer.writeBytes(v, 0, bytesLen);
} else {
char[] v = (char[]) GET_VALUE.apply(value);
if (StringSerializer.isLatin(v)) {
diff --git
a/java/fury-core/src/main/java/org/apache/fury/serializer/StringSerializer.java
b/java/fury-core/src/main/java/org/apache/fury/serializer/StringSerializer.java
index 6de440cf..4ceedb0e 100644
---
a/java/fury-core/src/main/java/org/apache/fury/serializer/StringSerializer.java
+++
b/java/fury-core/src/main/java/org/apache/fury/serializer/StringSerializer.java
@@ -211,8 +211,7 @@ public final class StringSerializer extends
Serializer<String> {
if (compressString) {
return new Invoke(strSerializer, "readCompressedCharsString",
STRING_TYPE, buffer);
} else {
- Expression chars =
- new Invoke(buffer, "readCharsWithSizeEmbedded",
PRIMITIVE_CHAR_ARRAY_TYPE);
+ Expression chars = new Invoke(buffer, "readCharsAndSize",
PRIMITIVE_CHAR_ARRAY_TYPE);
return new StaticInvoke(
StringSerializer.class, "newCharsStringZeroCopy", STRING_TYPE,
chars);
}
@@ -224,9 +223,10 @@ public final class StringSerializer extends
Serializer<String> {
@CodegenInvoke
public String readBytesString(MemoryBuffer buffer) {
- byte coder = buffer.readByte();
- // may fill heap array, if array not big enough, a new array will be
allocated.
- final int numBytes = buffer.readBinarySize();
+ long header = buffer.readVarUint36Small();
+ byte coder = (byte) (header & 0b11);
+ int numBytes = (int) (header >>> 2);
+ buffer.checkReadableBytes(numBytes);
byte[] bytes;
byte[] heapMemory = buffer.getHeapMemory();
if (heapMemory != null) {
@@ -246,14 +246,24 @@ public final class StringSerializer extends
Serializer<String> {
@CodegenInvoke
public String readCompressedCharsString(MemoryBuffer buffer) {
- byte coder = buffer.readByte();
+ long header = buffer.readVarUint36Small();
+ byte coder = (byte) (header & 0b11);
+ int numBytes = (int) (header >>> 2);
if (coder == LATIN1) {
- return newCharsStringZeroCopy(readLatinChars(buffer));
+ return newCharsStringZeroCopy(readLatinChars(buffer, numBytes));
+ } else if (coder == UTF16) {
+ return newCharsStringZeroCopy(readUTF16Chars(buffer, numBytes));
} else {
- return newCharsStringZeroCopy(readUTF16Chars(buffer, coder));
+ return readUtf8(buffer, coder, numBytes);
}
}
+ private String readUtf8(MemoryBuffer buffer, byte coder, int numBytes) {
+ Preconditions.checkArgument(coder == UTF8, UTF8);
+ byte[] bytes = buffer.readBytes(numBytes);
+ return new String(bytes, 0, numBytes, StandardCharsets.UTF_8);
+ }
+
private byte[] getByteArray(int numElements) {
byte[] byteArray = this.byteArray;
if (byteArray.length < numElements) {
@@ -275,9 +285,7 @@ public final class StringSerializer extends
Serializer<String> {
if (STRING_VALUE_FIELD_IS_BYTES) {
writeBytesString(buffer, value);
} else {
- if (!STRING_VALUE_FIELD_IS_CHARS) {
- throw new UnsupportedOperationException();
- }
+ assert STRING_VALUE_FIELD_IS_CHARS;
final char[] chars = (char[]) Platform.getObject(value,
STRING_VALUE_FIELD_OFFSET);
if (compressString) {
if (isLatin(chars)) {
@@ -323,31 +331,20 @@ public final class StringSerializer extends
Serializer<String> {
if (STRING_VALUE_FIELD_IS_BYTES) {
return readBytesString(buffer);
} else {
- if (!STRING_VALUE_FIELD_IS_CHARS) {
- throw new UnsupportedOperationException();
- }
+ assert STRING_VALUE_FIELD_IS_CHARS;
if (compressString) {
- byte coder = buffer.readByte();
- if (coder == LATIN1) {
- return newCharsStringZeroCopy(readLatinChars(buffer));
- } else if (coder == UTF16) {
- return newCharsStringZeroCopy(readUTF16Chars(buffer, coder));
- } else {
- if (coder != UTF8) {
- throw new UnsupportedOperationException("Unsupported encoding: " +
coder);
- }
- return readUTF8String(buffer);
- }
+ return readCompressedCharsString(buffer);
} else {
- return
newCharsStringZeroCopy(buffer.readChars(buffer.readVarUintSmall()));
+ return newCharsStringZeroCopy(buffer.readCharsAndSize());
}
}
}
public static void writeBytesString(MemoryBuffer buffer, String value) {
byte[] bytes = (byte[]) Platform.getObject(value,
STRING_VALUE_FIELD_OFFSET);
- byte coder = Platform.getByte(value, Offset.STRING_CODER_FIELD_OFFSET);
int bytesLen = bytes.length;
+ long header =
+ ((long) bytesLen << 2) | Platform.getByte(value,
Offset.STRING_CODER_FIELD_OFFSET);
int writerIndex = buffer.writerIndex();
// The `ensure` ensure next operations are safe without bound checks,
// and inner heap buffer doesn't change.
@@ -358,19 +355,16 @@ public final class StringSerializer extends
Serializer<String> {
// jvm doesn't eliminate well in some jdk.
final int targetIndex = buffer.unsafeHeapWriterIndex();
int arrIndex = targetIndex;
- targetArray[arrIndex++] = coder;
- arrIndex += MemoryUtils.writePositiveVarInt(targetArray, arrIndex,
bytesLen);
+ arrIndex += MemoryUtils.putVarUint36Small(targetArray, arrIndex, header);
writerIndex += arrIndex - targetIndex;
System.arraycopy(bytes, 0, targetArray, arrIndex, bytesLen);
- writerIndex += bytesLen;
} else {
- buffer.unsafePut(writerIndex++, coder);
- writerIndex += buffer.unsafePutPositiveVarInt(writerIndex, bytesLen);
+ writerIndex += buffer.unsafePutVarUint36Small(writerIndex, header);
long offHeapAddress = buffer.getUnsafeAddress();
Platform.copyMemory(
bytes, Platform.BYTE_ARRAY_OFFSET, null, offHeapAddress +
writerIndex, bytesLen);
- writerIndex += bytesLen;
}
+ writerIndex += bytesLen;
buffer.unsafeWriterIndex(writerIndex);
}
@@ -379,20 +373,19 @@ public final class StringSerializer extends
Serializer<String> {
// The `ensure` ensure next operations are safe without bound checks,
// and inner heap buffer doesn't change.
buffer.ensure(writerIndex + 9 + strLen);
+ long header = ((long) strLen << 2) | LATIN1;
final byte[] targetArray = buffer.getHeapMemory();
if (targetArray != null) {
- final int targetIndex = buffer.unsafeHeapWriterIndex();
- int arrIndex = targetIndex;
- targetArray[arrIndex++] = LATIN1;
- arrIndex += MemoryUtils.writePositiveVarInt(targetArray, arrIndex,
strLen);
- writerIndex += arrIndex - targetIndex + strLen;
+ int arrIndex = buffer.unsafeHeapWriterIndex();
+ int written = MemoryUtils.putVarUint36Small(targetArray, arrIndex,
header);
+ arrIndex += written;
+ writerIndex += written + strLen;
for (int i = 0; i < strLen; i++) {
targetArray[arrIndex + i] = (byte) chars[i];
}
buffer.unsafeWriterIndex(writerIndex);
} else {
- buffer.unsafePut(writerIndex++, LATIN1);
- writerIndex += buffer.unsafePutPositiveVarInt(writerIndex, strLen);
+ writerIndex += buffer.unsafePutVarUint36Small(writerIndex, header);
final byte[] tmpArray = getByteArray(strLen);
// Write to heap memory then copy is 60% faster than unsafe write to
direct memory.
for (int i = 0; i < strLen; i++) {
@@ -406,48 +399,61 @@ public final class StringSerializer extends
Serializer<String> {
public void writeCharsUTF16(MemoryBuffer buffer, char[] chars, int strLen) {
int numBytes = MathUtils.doubleExact(strLen);
- if (Platform.IS_LITTLE_ENDIAN) {
- buffer.writeByte(UTF16);
- // FIXME JDK11 utf16 string uses little-endian order.
- buffer.writePrimitiveArrayWithSize(chars, Platform.CHAR_ARRAY_OFFSET,
numBytes);
- } else {
- // The `ensure` ensure next operations are safe without bound checks,
- // and inner heap buffer doesn't change.
- int writerIndex = buffer.writerIndex();
- buffer.ensure(writerIndex + 9 + numBytes);
- byte[] targetArray = buffer.getHeapMemory();
- if (targetArray != null) {
- final int targetIndex = buffer.unsafeHeapWriterIndex();
- int arrIndex = targetIndex;
- targetArray[arrIndex++] = UTF16;
- arrIndex += MemoryUtils.writePositiveVarInt(targetArray, arrIndex,
strLen);
- // Write to heap memory then copy is 250% faster than unsafe write to
direct memory.
- int charIndex = 0;
- for (int i = arrIndex, end = i + numBytes; i < end; i += 2) {
- char c = chars[charIndex++];
- targetArray[i] = (byte) (c >> StringUTF16.HI_BYTE_SHIFT);
- targetArray[i + 1] = (byte) (c >> StringUTF16.LO_BYTE_SHIFT);
- }
- writerIndex += arrIndex - targetIndex + numBytes;
+ long header = ((long) numBytes << 2) | UTF16;
+ // The `ensure` ensure next operations are safe without bound checks,
+ // and inner heap buffer doesn't change.
+ int writerIndex = buffer.writerIndex();
+ buffer.ensure(writerIndex + 9 + numBytes);
+ byte[] targetArray = buffer.getHeapMemory();
+ if (targetArray != null) {
+ int arrIndex = buffer.unsafeHeapWriterIndex();
+ int written = MemoryUtils.putVarUint36Small(targetArray, arrIndex,
header);
+ arrIndex += written;
+ writerIndex += written + numBytes;
+ if (Platform.IS_LITTLE_ENDIAN) {
+ // FIXME JDK11 utf16 string uses little-endian order.
+ Platform.UNSAFE.copyMemory(
+ chars,
+ Platform.CHAR_ARRAY_OFFSET,
+ targetArray,
+ Platform.BYTE_ARRAY_OFFSET + arrIndex,
+ numBytes);
} else {
- buffer.unsafePut(writerIndex++, UTF16);
- writerIndex += buffer.unsafePutPositiveVarInt(writerIndex, numBytes);
- byte[] tmpArray = getByteArray(strLen);
- int charIndex = 0;
- for (int i = 0; i < numBytes; i += 2) {
- char c = chars[charIndex++];
- tmpArray[i] = (byte) (c >> StringUTF16.HI_BYTE_SHIFT);
- tmpArray[i + 1] = (byte) (c >> StringUTF16.LO_BYTE_SHIFT);
- }
- buffer.put(writerIndex, tmpArray, 0, numBytes);
- writerIndex += numBytes;
+ heapWriteCharsUTF16BE(chars, arrIndex, numBytes, targetArray);
}
- buffer.unsafeWriterIndex(writerIndex);
+ } else {
+ writerIndex = offHeapWriteCharsUTF16(buffer, chars, writerIndex, header,
numBytes);
}
+ buffer.unsafeWriterIndex(writerIndex);
}
- private char[] readLatinChars(MemoryBuffer buffer) {
- final int numBytes = buffer.readVarUintSmall();
+ private static void heapWriteCharsUTF16BE(
+ char[] chars, int arrIndex, int numBytes, byte[] targetArray) {
+ // Write to heap memory then copy is 250% faster than unsafe write to
direct memory.
+ int charIndex = 0;
+ for (int i = arrIndex, end = i + numBytes; i < end; i += 2) {
+ char c = chars[charIndex++];
+ targetArray[i] = (byte) (c >> StringUTF16.HI_BYTE_SHIFT);
+ targetArray[i + 1] = (byte) (c >> StringUTF16.LO_BYTE_SHIFT);
+ }
+ }
+
+ private int offHeapWriteCharsUTF16(
+ MemoryBuffer buffer, char[] chars, int writerIndex, long header, int
numBytes) {
+ writerIndex += buffer.unsafePutVarUint36Small(writerIndex, header);
+ byte[] tmpArray = getByteArray(numBytes);
+ int charIndex = 0;
+ for (int i = 0; i < numBytes; i += 2) {
+ char c = chars[charIndex++];
+ tmpArray[i] = (byte) (c >> StringUTF16.HI_BYTE_SHIFT);
+ tmpArray[i + 1] = (byte) (c >> StringUTF16.LO_BYTE_SHIFT);
+ }
+ buffer.put(writerIndex, tmpArray, 0, numBytes);
+ writerIndex += numBytes;
+ return writerIndex;
+ }
+
+ private char[] readLatinChars(MemoryBuffer buffer, int numBytes) {
char[] chars = new char[numBytes];
buffer.checkReadableBytes(numBytes);
byte[] targetArray = buffer.getHeapMemory();
@@ -467,11 +473,7 @@ public final class StringSerializer extends
Serializer<String> {
return chars;
}
- private char[] readUTF16Chars(MemoryBuffer buffer, byte coder) {
- if (coder != UTF16) {
- throw new UnsupportedOperationException(String.format("Unsupported coder
%s", coder));
- }
- int numBytes = buffer.readVarUintSmall();
+ private char[] readUTF16Chars(MemoryBuffer buffer, int numBytes) {
char[] chars = new char[numBytes >> 1];
if (Platform.IS_LITTLE_ENDIAN) {
// FIXME JDK11 utf16 string uses little-endian order.
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 f6f15e16..d032dadc 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
@@ -578,4 +578,40 @@ public class MemoryBufferTest {
assertEquals(value, varLong);
assertEquals(buf.slice(readerIndex, buf.readerIndex() -
readerIndex).readSliLong(), value);
}
+
+ @Test
+ public void testVarUint36Small() {
+ MemoryBuffer buf = MemoryUtils.buffer(80);
+ int index = 0;
+ {
+ int diff = MemoryUtils.putVarUint36Small(buf.getHeapMemory(), index, 10);
+ assertEquals(buf.readVarUint36Small(), 10);
+ buf.increaseReaderIndex(-diff);
+ index += buf.unsafePutVarUint36Small(index, 10);
+ assertEquals(buf.readVarUint36Small(), 10);
+ }
+ {
+ int diff = MemoryUtils.putVarUint36Small(buf.getHeapMemory(), index,
Short.MAX_VALUE);
+ assertEquals(buf.readVarUint36Small(), Short.MAX_VALUE);
+ buf.increaseReaderIndex(-diff);
+ index += buf.unsafePutVarUint36Small(index, Short.MAX_VALUE);
+ assertEquals(buf.readVarUint36Small(), Short.MAX_VALUE);
+ }
+ {
+ int diff = MemoryUtils.putVarUint36Small(buf.getHeapMemory(), index,
Integer.MAX_VALUE);
+ assertEquals(buf.readVarUint36Small(), Integer.MAX_VALUE);
+ buf.increaseReaderIndex(-diff);
+ index += buf.unsafePutVarUint36Small(index, Integer.MAX_VALUE);
+ assertEquals(buf.readVarUint36Small(), Integer.MAX_VALUE);
+ }
+ {
+ int diff =
+ MemoryUtils.putVarUint36Small(
+ buf.getHeapMemory(), index,
0b111111111111111111111111111111111111L);
+ assertEquals(buf.readVarUint36Small(),
0b111111111111111111111111111111111111L);
+ buf.increaseReaderIndex(-diff);
+ buf.unsafePutVarUint36Small(index,
0b1000000000000000000000000000000000000L);
+ assertEquals(buf.readVarUint36Small(), 0); // overflow
+ }
+ }
}
diff --git
a/java/fury-core/src/test/java/org/apache/fury/serializer/StringSerializerTest.java
b/java/fury-core/src/test/java/org/apache/fury/serializer/StringSerializerTest.java
index a6c56e88..46f5832d 100644
---
a/java/fury-core/src/test/java/org/apache/fury/serializer/StringSerializerTest.java
+++
b/java/fury-core/src/test/java/org/apache/fury/serializer/StringSerializerTest.java
@@ -95,9 +95,10 @@ public class StringSerializerTest extends FuryTestBase {
}
static String readJDK11String(MemoryBuffer buffer) {
- byte coder = buffer.readByte();
- byte[] value = buffer.readBytesAndSize();
- return newBytesStringZeroCopy(coder, value);
+ long header = buffer.readVarUint36Small();
+ byte coder = (byte) (header & 0b11);
+ int numBytes = (int) (header >>> 2);
+ return newBytesStringZeroCopy(coder, buffer.readBytes(numBytes));
}
private static boolean writeJavaStringZeroCopy(MemoryBuffer buffer, String
value) {
@@ -381,10 +382,10 @@ public class StringSerializerTest extends FuryTestBase {
StringSerializer serializer = new StringSerializer(fury);
serializer.write(buffer, "abc你好");
assertEquals(serializer.read(buffer), "abc你好");
+ byte[] bytes = "abc你好".getBytes(StandardCharsets.UTF_8);
byte UTF8 = 2;
- buffer.writeByte(UTF8);
-
buffer.writePositiveVarInt("abc你好".getBytes(StandardCharsets.UTF_8).length);
- buffer.writeBytes("abc你好".getBytes(StandardCharsets.UTF_8));
+ buffer.writePositiveVarLong(((long) bytes.length) << 2 | UTF8);
+ buffer.writeBytes(bytes);
assertEquals(serializer.read(buffer), "abc你好");
assertEquals(buffer.readerIndex(), buffer.writerIndex());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
