This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch cluster_add_snappy
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/cluster_add_snappy by this
push:
new 2aba5fb abstract TCompressedElasticFramedTransport
2aba5fb is described below
commit 2aba5fb5c3248c99aef45f4f9a334fdc21db7601
Author: jt <[email protected]>
AuthorDate: Thu Dec 3 13:48:15 2020 +0800
abstract TCompressedElasticFramedTransport
---
.../apache/iotdb/cluster/server/NodeReport.java | 12 +-
.../main/java/org/apache/iotdb/rpc/RpcStat.java | 50 +++++++++
...java => TCompressedElasticFramedTransport.java} | 73 +++---------
.../iotdb/rpc/TSnappyElasticFramedTransport.java | 122 ++-------------------
4 files changed, 85 insertions(+), 172 deletions(-)
diff --git
a/cluster/src/main/java/org/apache/iotdb/cluster/server/NodeReport.java
b/cluster/src/main/java/org/apache/iotdb/cluster/server/NodeReport.java
index 0d84b0f..0b6816c 100644
--- a/cluster/src/main/java/org/apache/iotdb/cluster/server/NodeReport.java
+++ b/cluster/src/main/java/org/apache/iotdb/cluster/server/NodeReport.java
@@ -19,13 +19,11 @@
package org.apache.iotdb.cluster.server;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.iotdb.cluster.rpc.thrift.Node;
+import org.apache.iotdb.rpc.RpcStat;
import org.apache.iotdb.rpc.RpcTransportFactory;
-import org.apache.iotdb.rpc.TSnappyElasticFramedTransport;
-import org.xerial.snappy.Snappy;
/**
* A node report collects the current runtime information of the local node,
which contains:
@@ -112,10 +110,10 @@ public class NodeReport {
@Override
public String toString() {
- long readBytes = TSnappyElasticFramedTransport.getReadBytes();
- long readCompressedBytes =
TSnappyElasticFramedTransport.getReadCompressedBytes();
- long writeBytes = TSnappyElasticFramedTransport.getWriteBytes();
- long writeCompressedBytes =
TSnappyElasticFramedTransport.getWriteCompressedBytes();
+ long readBytes = RpcStat.getReadBytes();
+ long readCompressedBytes = RpcStat.getReadCompressedBytes();
+ long writeBytes = RpcStat.getWriteBytes();
+ long writeCompressedBytes = RpcStat.getWriteCompressedBytes();
double readCompressionRatio
= (double) readBytes / readCompressedBytes;
double writeCompressionRatio = (double) writeBytes /
writeCompressedBytes;
diff --git a/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcStat.java
b/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcStat.java
new file mode 100644
index 0000000..3bf5f76
--- /dev/null
+++ b/service-rpc/src/main/java/org/apache/iotdb/rpc/RpcStat.java
@@ -0,0 +1,50 @@
+/*
+ * 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.iotdb.rpc;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+public class RpcStat {
+
+ private RpcStat() {
+ // pure static class
+ }
+
+ static final AtomicLong writeBytes = new AtomicLong();
+ static final AtomicLong writeCompressedBytes = new AtomicLong();
+ static final AtomicLong readBytes = new AtomicLong();
+ static final AtomicLong readCompressedBytes = new AtomicLong();
+
+ public static long getReadBytes() {
+ return readBytes.get();
+ }
+
+ public static long getReadCompressedBytes() {
+ return readCompressedBytes.get();
+ }
+
+ public static long getWriteBytes() {
+ return writeBytes.get();
+ }
+
+ public static long getWriteCompressedBytes() {
+ return writeCompressedBytes.get();
+ }
+}
diff --git
a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
similarity index 67%
copy from
service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
copy to
service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
index cc1b01e..de6f79a 100644
---
a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
+++
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TCompressedElasticFramedTransport.java
@@ -20,54 +20,23 @@ package org.apache.iotdb.rpc;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.thrift.transport.TByteBuffer;
import org.apache.thrift.transport.TFastFramedTransport;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
-import org.apache.thrift.transport.TTransportFactory;
-import org.xerial.snappy.Snappy;
-public class TSnappyElasticFramedTransport extends TFastFramedTransport {
-
- private static final AtomicLong writeBytes = new AtomicLong();
- private static final AtomicLong writeCompressedBytes = new AtomicLong();
- private static final AtomicLong readBytes = new AtomicLong();
- private static final AtomicLong readCompressedBytes = new AtomicLong();
+public abstract class TCompressedElasticFramedTransport extends
TFastFramedTransport {
private TByteBuffer writeCompressBuffer;
private TByteBuffer readCompressBuffer;
- public static class Factory extends TTransportFactory {
-
- private final int initialCapacity;
- private final int maxLength;
-
- public Factory() {
- this(DEFAULT_BUF_CAPACITY, DEFAULT_MAX_LENGTH);
- }
-
- public Factory(int initialCapacity) {
- this(initialCapacity, DEFAULT_MAX_LENGTH);
- }
-
- public Factory(int initialCapacity, int maxLength) {
- this.initialCapacity = initialCapacity;
- this.maxLength = maxLength;
- }
-
- @Override
- public TTransport getTransport(TTransport trans) {
- return new TSnappyElasticFramedTransport(trans, initialCapacity,
maxLength);
- }
- }
-
- public TSnappyElasticFramedTransport(TTransport underlying) {
+ public TCompressedElasticFramedTransport(TTransport underlying) {
this(underlying, DEFAULT_BUF_CAPACITY, DEFAULT_MAX_LENGTH);
}
- public TSnappyElasticFramedTransport(TTransport underlying, int
initialBufferCapacity, int maxLength) {
+ public TCompressedElasticFramedTransport(TTransport underlying, int
initialBufferCapacity,
+ int maxLength) {
super(underlying, initialBufferCapacity, maxLength);
this.underlying = underlying;
this.maxLength = maxLength;
@@ -107,12 +76,12 @@ public class TSnappyElasticFramedTransport extends
TFastFramedTransport {
}
readBuffer.fill(underlying, size);
- readCompressedBytes.addAndGet(size);
+ RpcStat.readCompressedBytes.addAndGet(size);
try {
- int uncompressedLength =
Snappy.uncompressedLength(readBuffer.getBuffer(), 0, size);
- readBytes.addAndGet(uncompressedLength);
+ int uncompressedLength = uncompressedLength(readBuffer.getBuffer(), 0,
size);
+ RpcStat.readBytes.addAndGet(uncompressedLength);
readCompressBuffer = resizeCompressBuf(uncompressedLength,
readCompressBuffer);
- Snappy.uncompress(readBuffer.getBuffer(), 0, size,
readCompressBuffer.getByteBuffer().array(), 0);
+ uncompress(readBuffer.getBuffer(), 0, size,
readCompressBuffer.getByteBuffer().array(), 0);
readCompressBuffer.getByteBuffer().limit(uncompressedLength);
readCompressBuffer.getByteBuffer().position(0);
@@ -137,13 +106,13 @@ public class TSnappyElasticFramedTransport extends
TFastFramedTransport {
@Override
public void flush() throws TTransportException {
int length = writeBuffer.getPos();
- writeBytes.addAndGet(length);
+ RpcStat.writeBytes.addAndGet(length);
try {
- int maxCompressedLength = Snappy.maxCompressedLength(length);
+ int maxCompressedLength = maxCompressedLength(length);
writeCompressBuffer = resizeCompressBuf(maxCompressedLength,
writeCompressBuffer);
- int compressedLength = Snappy.compress(writeBuffer.getBuf().array(), 0,
length,
+ int compressedLength = compress(writeBuffer.getBuf().array(), 0, length,
writeCompressBuffer.getByteBuffer().array(), 0);
- writeCompressedBytes.addAndGet(compressedLength);
+ RpcStat.writeCompressedBytes.addAndGet(compressedLength);
TFramedTransport.encodeFrameSize(compressedLength, i32buf);
underlying.write(i32buf, 0, 4);
@@ -164,19 +133,13 @@ public class TSnappyElasticFramedTransport extends
TFastFramedTransport {
writeBuffer.write(buf, off, len);
}
- public static long getReadBytes() {
- return readBytes.get();
- }
+ protected abstract int uncompressedLength(byte[] but, int off, int len)
throws IOException;
- public static long getReadCompressedBytes() {
- return readCompressedBytes.get();
- }
+ protected abstract int maxCompressedLength(int len);
- public static long getWriteBytes() {
- return writeBytes.get();
- }
+ protected abstract int compress(byte[] input, int inOff, int len, byte[]
output,
+ int outOff) throws IOException;
- public static long getWriteCompressedBytes() {
- return writeCompressedBytes.get();
- }
+ protected abstract void uncompress(byte[] input, int inOff, int size, byte[]
output,
+ int outOff) throws IOException;
}
diff --git
a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
index cc1b01e..d938fb6 100644
---
a/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
+++
b/service-rpc/src/main/java/org/apache/iotdb/rpc/TSnappyElasticFramedTransport.java
@@ -19,25 +19,12 @@
package org.apache.iotdb.rpc;
import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.concurrent.atomic.AtomicLong;
-import org.apache.thrift.transport.TByteBuffer;
-import org.apache.thrift.transport.TFastFramedTransport;
-import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
import org.apache.thrift.transport.TTransportFactory;
import org.xerial.snappy.Snappy;
-public class TSnappyElasticFramedTransport extends TFastFramedTransport {
+public class TSnappyElasticFramedTransport extends
TCompressedElasticFramedTransport {
- private static final AtomicLong writeBytes = new AtomicLong();
- private static final AtomicLong writeCompressedBytes = new AtomicLong();
- private static final AtomicLong readBytes = new AtomicLong();
- private static final AtomicLong readCompressedBytes = new AtomicLong();
-
- private TByteBuffer writeCompressBuffer;
- private TByteBuffer readCompressBuffer;
public static class Factory extends TTransportFactory {
@@ -69,114 +56,29 @@ public class TSnappyElasticFramedTransport extends
TFastFramedTransport {
public TSnappyElasticFramedTransport(TTransport underlying, int
initialBufferCapacity, int maxLength) {
super(underlying, initialBufferCapacity, maxLength);
- this.underlying = underlying;
- this.maxLength = maxLength;
- readBuffer = new AutoScalingBufferReadTransport(initialBufferCapacity);
- writeBuffer = new AutoScalingBufferWriteTransport(initialBufferCapacity);
- writeCompressBuffer = new
TByteBuffer(ByteBuffer.allocate(initialBufferCapacity));
- readCompressBuffer = new
TByteBuffer(ByteBuffer.allocate(initialBufferCapacity));
}
- private final int maxLength;
- private final TTransport underlying;
- private AutoScalingBufferReadTransport readBuffer;
- private AutoScalingBufferWriteTransport writeBuffer;
- private final byte[] i32buf = new byte[4];
@Override
- public int read(byte[] buf, int off, int len) throws TTransportException {
- int got = readBuffer.read(buf, off, len);
- if (got > 0) {
- return got;
- }
-
- // Read another frame of data
- readFrame();
- return readBuffer.read(buf, off, len);
- }
-
- @SuppressWarnings("java:S2177") // no better name
- private void readFrame() throws TTransportException {
- underlying.readAll(i32buf, 0, 4);
- int size = TFramedTransport.decodeFrameSize(i32buf);
-
- if (size < 0) {
- close();
- throw new TTransportException(TTransportException.CORRUPTED_DATA,
- "Read a negative frame size (" + size + ")!");
- }
-
- readBuffer.fill(underlying, size);
- readCompressedBytes.addAndGet(size);
- try {
- int uncompressedLength =
Snappy.uncompressedLength(readBuffer.getBuffer(), 0, size);
- readBytes.addAndGet(uncompressedLength);
- readCompressBuffer = resizeCompressBuf(uncompressedLength,
readCompressBuffer);
- Snappy.uncompress(readBuffer.getBuffer(), 0, size,
readCompressBuffer.getByteBuffer().array(), 0);
- readCompressBuffer.getByteBuffer().limit(uncompressedLength);
- readCompressBuffer.getByteBuffer().position(0);
-
- readBuffer.fill(readCompressBuffer, uncompressedLength);
- } catch (IOException e) {
- throw new TTransportException(e);
- }
- }
-
- private TByteBuffer resizeCompressBuf(int size, TByteBuffer byteBuffer) {
- double expandFactor = 1.5;
- double loadFactor = 0.6;
- if (byteBuffer.getByteBuffer().capacity() < size) {
- int newCap = (int) Math.min(size * expandFactor, maxLength);
- byteBuffer = new TByteBuffer(ByteBuffer.allocate(newCap));
- } else if (byteBuffer.getByteBuffer().capacity() * loadFactor > size) {
- byteBuffer = new TByteBuffer(ByteBuffer.allocate(size));
- }
- return byteBuffer;
+ protected int uncompressedLength(byte[] buf, int off, int len) throws
IOException {
+ return Snappy.uncompressedLength(buf, off, len);
}
@Override
- public void flush() throws TTransportException {
- int length = writeBuffer.getPos();
- writeBytes.addAndGet(length);
- try {
- int maxCompressedLength = Snappy.maxCompressedLength(length);
- writeCompressBuffer = resizeCompressBuf(maxCompressedLength,
writeCompressBuffer);
- int compressedLength = Snappy.compress(writeBuffer.getBuf().array(), 0,
length,
- writeCompressBuffer.getByteBuffer().array(), 0);
- writeCompressedBytes.addAndGet(compressedLength);
- TFramedTransport.encodeFrameSize(compressedLength, i32buf);
- underlying.write(i32buf, 0, 4);
-
- underlying.write(writeCompressBuffer.getByteBuffer().array(), 0,
compressedLength);
- } catch (IOException e) {
- throw new TTransportException(e);
- }
-
- writeBuffer.reset();
- if (maxLength < length) {
- writeBuffer.resizeIfNecessary(maxLength);
- }
- underlying.flush();
+ protected int maxCompressedLength(int len) {
+ return Snappy.maxCompressedLength(len);
}
@Override
- public void write(byte[] buf, int off, int len) {
- writeBuffer.write(buf, off, len);
- }
-
- public static long getReadBytes() {
- return readBytes.get();
- }
-
- public static long getReadCompressedBytes() {
- return readCompressedBytes.get();
+ protected int compress(byte[] input, int inOff, int len, byte[] output, int
outOff)
+ throws IOException {
+ return Snappy.compress(input, inOff, len, output, outOff);
}
- public static long getWriteBytes() {
- return writeBytes.get();
+ @Override
+ protected void uncompress(byte[] input, int inOff, int size, byte[] output,
int outOff)
+ throws IOException {
+ Snappy.uncompress(input, inOff, size, output, outOff);
}
- public static long getWriteCompressedBytes() {
- return writeCompressedBytes.get();
- }
}
