You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sd...@apache.org on 2022/01/11 23:56:52 UTC
[ignite-3] 24/27: - use varints for descriptor/command/object ID and container serialization
This is an automated email from the ASF dual-hosted git repository.
sdanilov pushed a commit to branch uos-network
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 2432830b6dd55d431e06f4af342d9e6f93b7d73d
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Tue Jan 11 17:53:52 2022 +0400
- use varints for descriptor/command/object ID and container serialization
---
.../serialization/marshal/BuiltInMarshalling.java | 2 +-
.../serialization/marshal/ProtocolMarshalling.java | 30 +----
.../network/serialization/marshal/VarInts.java | 66 +++++++++++
...erObjectMarshallerWithArbitraryObjectsTest.java | 2 +-
.../network/serialization/marshal/VarIntsTest.java | 127 +++++++++++++++++++++
5 files changed, 201 insertions(+), 26 deletions(-)
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/BuiltInMarshalling.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/BuiltInMarshalling.java
index 0d48982..a2f02db 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/BuiltInMarshalling.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/BuiltInMarshalling.java
@@ -304,7 +304,7 @@ class BuiltInMarshalling {
static <T> T[] preInstantiateGenericRefArray(DataInput input) throws IOException, UnmarshalException {
IntFunction<T[]> arrayFactory = readTypeAndCreateArrayFactory(input);
- int length = input.readInt();
+ int length = readLength(input);
return arrayFactory.apply(length);
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/ProtocolMarshalling.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/ProtocolMarshalling.java
index 57574c5..32f5788 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/ProtocolMarshalling.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/ProtocolMarshalling.java
@@ -29,46 +29,28 @@ class ProtocolMarshalling {
static final int MAX_LENGTH_BYTE_COUNT = 4;
static void writeDescriptorOrCommandId(int id, DataOutput output) throws IOException {
- output.writeInt(id);
+ VarInts.writeUnsignedInt(id, output);
}
static int readDescriptorOrCommandId(DataInput input) throws IOException {
- return input.readInt();
+ return VarInts.readUnsignedInt(input);
}
static void writeObjectId(int id, DataOutput output) throws IOException {
- output.writeInt(id);
+ VarInts.writeUnsignedInt(id, output);
}
static int readObjectId(DataInput input) throws IOException {
- return input.readInt();
+ return VarInts.readUnsignedInt(input);
}
static void writeLength(int length, DataOutput output) throws IOException {
- writeUnsignedInt(length, output);
- }
-
- private static void writeUnsignedInt(int value, DataOutput output) throws IOException {
- if (value < 0) {
- throw new IllegalArgumentException(value + " is negative" );
- }
-
- output.writeInt(value);
+ VarInts.writeUnsignedInt(length, output);
}
static int readLength(DataInput input) throws IOException {
- return readUnsignedInt(input);
- }
-
- private static int readUnsignedInt(DataInput input) throws IOException {
- int value = input.readInt();
-
- if (value < 0) {
- throw new IllegalStateException(value + " is negative");
- }
-
- return value;
+ return VarInts.readUnsignedInt(input);
}
private ProtocolMarshalling() {
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/VarInts.java b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/VarInts.java
new file mode 100644
index 0000000..fe7ff4e
--- /dev/null
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/serialization/marshal/VarInts.java
@@ -0,0 +1,66 @@
+/*
+ * 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.ignite.internal.network.serialization.marshal;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ *
+ */
+class VarInts {
+ private VarInts() {
+ }
+
+ static void writeUnsignedInt(int value, DataOutput output) throws IOException {
+ if (value < 0) {
+ throw new IllegalArgumentException(value + " is negative" );
+ }
+
+ if (value < 0xFF) {
+ output.writeByte(value);
+ } else if (value < 0xFFFF) {
+ output.writeByte(0xFF);
+ output.writeShort(value);
+ } else {
+ output.writeByte(0xFF);
+ output.writeShort(0xFFFF);
+ output.writeInt(value);
+ }
+ }
+
+ static int readUnsignedInt(DataInput input) throws IOException {
+ int first = input.readUnsignedByte();
+ if (first < 0xFF) {
+ return first;
+ }
+
+ int second = input.readUnsignedShort();
+ if (second < 0xFFFF) {
+ return second;
+ }
+
+ int third = input.readInt();
+ if (third < 0) {
+ throw new IllegalStateException(third + " is negative");
+ }
+
+ return third;
+ }
+}
diff --git a/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/DefaultUserObjectMarshallerWithArbitraryObjectsTest.java b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/DefaultUserObjectMarshallerWithArbitraryObjectsTest.java
index 625d940..4edf235 100644
--- a/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/DefaultUserObjectMarshallerWithArbitraryObjectsTest.java
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/DefaultUserObjectMarshallerWithArbitraryObjectsTest.java
@@ -101,7 +101,7 @@ class DefaultUserObjectMarshallerWithArbitraryObjectsTest {
private int readType(MarshalledObject marshalled) throws IOException {
try (var dis = new DataInputStream(new ByteArrayInputStream(marshalled.bytes()))) {
- return dis.readInt();
+ return ProtocolMarshalling.readDescriptorOrCommandId(dis);
}
}
diff --git a/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/VarIntsTest.java b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/VarIntsTest.java
new file mode 100644
index 0000000..d7b5100
--- /dev/null
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/serialization/marshal/VarIntsTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.ignite.internal.network.serialization.marshal;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThan;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class VarIntsTest {
+ @ParameterizedTest
+ @MethodSource("intsBetween0And254")
+ void writesIntsBetween0And254As1Byte(int value) throws Exception {
+ byte[] bytes = writeToBytes(output -> VarInts.writeUnsignedInt(value, output));
+
+ assertThat(bytes.length, is(1));
+ }
+
+ private static Stream<Arguments> intsBetween0And254() {
+ return IntStream.rangeClosed(0, 254)
+ .boxed()
+ .map(Arguments::of);
+ }
+
+ private byte[] writeToBytes(StreamWriter streamWriter) throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (var dos = new DataOutputStream(baos)) {
+ streamWriter.write(dos);
+ }
+ return baos.toByteArray();
+ }
+
+ @ParameterizedTest
+ @MethodSource("intsHigherThan254")
+ void writesIntsHigherThan254WithCorrectLengths(IntWriteSpec spec) throws Exception {
+ byte[] bytes = writeToBytes(output -> VarInts.writeUnsignedInt(spec.value, output));
+
+ assertThat(bytes.length, is(spec.expectedLength));
+ }
+
+ private static Stream<Arguments> intsHigherThan254() {
+ return Stream.of(
+ new IntWriteSpec(0xFF, 3),
+ new IntWriteSpec(0xFFFE, 3),
+ new IntWriteSpec(0xFFFF, 7),
+ new IntWriteSpec(Integer.MAX_VALUE, 7)
+ ).map(Arguments::of);
+ }
+
+ @ParameterizedTest
+ @MethodSource("intsBetween0And254")
+ void writesAndReadsIntsBetween0And254(int value) throws Exception {
+ byte[] bytes = writeToBytes(output -> VarInts.writeUnsignedInt(value, output));
+ int result = readIntFromBytesConsuming(bytes, VarInts::readUnsignedInt);
+
+ assertThat(result, is(value));
+ }
+
+ private int readIntFromBytesConsuming(byte[] bytes, StreamReader streamReader) throws IOException {
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ try (var dos = new DataInputStream(bais)) {
+ int result = streamReader.read(dos);
+ assertThat("Stream was not fully consumed", dos.read(), is(lessThan(0)));
+ return result;
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("intsHigherThan254")
+ void writesAndReadsIntsHigherThan254(IntWriteSpec spec) throws Exception {
+ byte[] bytes = writeToBytes(output -> VarInts.writeUnsignedInt(spec.value, output));
+ int result = readIntFromBytesConsuming(bytes, VarInts::readUnsignedInt);
+
+ assertThat(result, is(spec.value));
+ }
+
+ private interface StreamWriter {
+ void write(DataOutput output) throws IOException;
+ }
+
+ private interface StreamReader {
+ int read(DataInput input) throws IOException;
+ }
+
+ private static class IntWriteSpec {
+ private final int value;
+ private final int expectedLength;
+
+ private IntWriteSpec(int value, int expectedLength) {
+ this.value = value;
+ this.expectedLength = expectedLength;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String toString() {
+ return Integer.toString(value);
+ }
+ }
+}