accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mmil...@apache.org
Subject [accumulo] branch 1.9 updated: Use optimized writeV methods in Mutations (#669)
Date Wed, 03 Oct 2018 16:22:04 GMT
This is an automated email from the ASF dual-hosted git repository.

mmiller pushed a commit to branch 1.9
in repository https://gitbox.apache.org/repos/asf/accumulo.git


The following commit(s) were added to refs/heads/1.9 by this push:
     new 987ae65  Use optimized writeV methods in Mutations (#669)
987ae65 is described below

commit 987ae65cc606bdaf6865c2c6810a13c767b7a5c5
Author: Mike Miller <mmiller@apache.org>
AuthorDate: Tue Oct 2 11:03:05 2018 -0400

    Use optimized writeV methods in Mutations (#669)
    
    * Moved optimized writeVLong and writeVInt to UnsynchronizedBuffer
    to replace use of WriteableUtils methods. Each optimized method will
    only make one write call to the underlying outputstream.
---
 .../org/apache/accumulo/core/data/Mutation.java    |  13 +--
 .../accumulo/core/util/UnsynchronizedBuffer.java   | 101 +++++++++++++++------
 .../core/util/UnsynchronizedBufferTest.java        |  72 +++++++++++++++
 .../accumulo/server/data/ServerMutation.java       |   4 +-
 4 files changed, 157 insertions(+), 33 deletions(-)

diff --git a/core/src/main/java/org/apache/accumulo/core/data/Mutation.java b/core/src/main/java/org/apache/accumulo/core/data/Mutation.java
index 091d50d..338444d 100644
--- a/core/src/main/java/org/apache/accumulo/core/data/Mutation.java
+++ b/core/src/main/java/org/apache/accumulo/core/data/Mutation.java
@@ -1116,6 +1116,7 @@ public class Mutation implements Writable {
 
   @Override
   public void write(DataOutput out) throws IOException {
+    final byte[] integerBuffer = new byte[5];
     serialize();
     byte hasValues = (values == null) ? 0 : (byte) 1;
     if (!replicationSources.isEmpty()) {
@@ -1124,23 +1125,23 @@ public class Mutation implements Writable {
     }
     out.write((byte) (0x80 | hasValues));
 
-    WritableUtils.writeVInt(out, row.length);
+    UnsynchronizedBuffer.writeVInt(out, integerBuffer, row.length);
     out.write(row);
 
-    WritableUtils.writeVInt(out, data.length);
+    UnsynchronizedBuffer.writeVInt(out, integerBuffer, data.length);
     out.write(data);
-    WritableUtils.writeVInt(out, entries);
+    UnsynchronizedBuffer.writeVInt(out, integerBuffer, entries);
 
     if (0x01 == (0x01 & hasValues)) {
-      WritableUtils.writeVInt(out, values.size());
+      UnsynchronizedBuffer.writeVInt(out, integerBuffer, values.size());
       for (int i = 0; i < values.size(); i++) {
         byte val[] = values.get(i);
-        WritableUtils.writeVInt(out, val.length);
+        UnsynchronizedBuffer.writeVInt(out, integerBuffer, val.length);
         out.write(val);
       }
     }
     if (0x02 == (0x02 & hasValues)) {
-      WritableUtils.writeVInt(out, replicationSources.size());
+      UnsynchronizedBuffer.writeVInt(out, integerBuffer, replicationSources.size());
       for (String source : replicationSources) {
         WritableUtils.writeString(out, source);
       }
diff --git a/core/src/main/java/org/apache/accumulo/core/util/UnsynchronizedBuffer.java b/core/src/main/java/org/apache/accumulo/core/util/UnsynchronizedBuffer.java
index b2fd932..2993f8b 100644
--- a/core/src/main/java/org/apache/accumulo/core/util/UnsynchronizedBuffer.java
+++ b/core/src/main/java/org/apache/accumulo/core/util/UnsynchronizedBuffer.java
@@ -16,6 +16,8 @@
  */
 package org.apache.accumulo.core.util;
 
+import java.io.DataOutput;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 
 import org.apache.hadoop.io.WritableUtils;
@@ -136,32 +138,7 @@ public class UnsynchronizedBuffer {
      */
     public void writeVLong(long i) {
       reserve(9);
-      if (i >= -112 && i <= 127) {
-        data[offset++] = (byte) i;
-        return;
-      }
-
-      int len = -112;
-      if (i < 0) {
-        i ^= -1L; // take one's complement'
-        len = -120;
-      }
-
-      long tmp = i;
-      while (tmp != 0) {
-        tmp = tmp >> 8;
-        len--;
-      }
-
-      data[offset++] = (byte) len;
-
-      len = (len < -120) ? -(len + 120) : -(len + 112);
-
-      for (int idx = len; idx != 0; idx--) {
-        int shiftbits = (idx - 1) * 8;
-        long mask = 0xFFL << shiftbits;
-        data[offset++] = (byte) ((i & mask) >> shiftbits);
-      }
+      offset = UnsynchronizedBuffer.writeVLong(data, offset, i);
     }
   }
 
@@ -304,4 +281,76 @@ public class UnsynchronizedBuffer {
 
     return ret;
   }
+
+  /**
+   * Use the provided byte[] to buffer only the bytes used to write out the integer i to
the
+   * DataOutput out. This will only ever make one write call to the DataOutput. Use this
instead of
+   * {@link WritableUtils#writeVInt(DataOutput, int)} which could make up to 4 separate writes
to
+   * the underlying OutputStream. Is compatible with WritableUtils as it will write the same
data.
+   */
+  public static void writeVInt(DataOutput out, byte[] workBuffer, int i) throws IOException
{
+    int size = UnsynchronizedBuffer.writeVInt(workBuffer, 0, i);
+    out.write(workBuffer, 0, size);
+  }
+
+  /**
+   * Use the provided byte[] to buffer only the bytes used to write out the long i to the
DataOutput
+   * out. This will only ever make one write call to the DataOutput. Use this instead of
+   * {@link WritableUtils#writeVLong(DataOutput, long)} which could make up to 8 separate
writes to
+   * the underlying OutputStream. Is compatible with WritableUtils as it will write the same
data.
+   */
+  public static void writeVLong(DataOutput out, byte[] workBuffer, long i) throws IOException
{
+    int size = UnsynchronizedBuffer.writeVLong(workBuffer, 0, i);
+    out.write(workBuffer, 0, size);
+  }
+
+  /**
+   * Writes a variable int directly to a byte array. Is compatible with {@link WritableUtils}
as it
+   * will write the same data.
+   */
+  public static int writeVInt(byte[] dest, int offset, int i) {
+    return writeVLong(dest, offset, i);
+  }
+
+  /**
+   * Writes a variable long directly to a byte array. Is compatible with {@link WritableUtils}
as it
+   * will write the same data.
+   *
+   * @param dest
+   *          The destination array for the long to be written to
+   * @param offset
+   *          The location where to write the long to
+   * @param value
+   *          The long value being written into byte array
+   * @return Returns the new offset location
+   */
+  public static int writeVLong(byte[] dest, int offset, long value) {
+    if (value >= -112 && value <= 127) {
+      dest[offset++] = (byte) value;
+      return offset;
+    }
+
+    int len = -112;
+    if (value < 0) {
+      value ^= -1L; // take one's complement'
+      len = -120;
+    }
+
+    long tmp = value;
+    while (tmp != 0) {
+      tmp = tmp >> 8;
+      len--;
+    }
+
+    dest[offset++] = (byte) len;
+
+    len = (len < -120) ? -(len + 120) : -(len + 112);
+
+    for (int idx = len; idx != 0; idx--) {
+      int shiftbits = (idx - 1) * 8;
+      long mask = 0xFFL << shiftbits;
+      dest[offset++] = (byte) ((value & mask) >> shiftbits);
+    }
+    return offset;
+  }
 }
diff --git a/core/src/test/java/org/apache/accumulo/core/util/UnsynchronizedBufferTest.java
b/core/src/test/java/org/apache/accumulo/core/util/UnsynchronizedBufferTest.java
index 052d7b6..0038b06 100644
--- a/core/src/test/java/org/apache/accumulo/core/util/UnsynchronizedBufferTest.java
+++ b/core/src/test/java/org/apache/accumulo/core/util/UnsynchronizedBufferTest.java
@@ -18,9 +18,14 @@ package org.apache.accumulo.core.util;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 
+import org.apache.hadoop.io.WritableUtils;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
@@ -53,4 +58,71 @@ public class UnsynchronizedBufferTest {
     thrown.expect(ArrayIndexOutOfBoundsException.class);
     ub.readBytes(buf);
   }
+
+  @Test
+  public void testWriteVMethods() throws Exception {
+    // writeV methods use an extra byte for length, unless value is only one byte
+    // Integer.MAX_VALUE = 0x7fffffff
+    testInteger(0x7fffffff, 4 + 1);
+    testInteger(0x7fffff, 3 + 1);
+    testInteger(0x7fff, 2 + 1);
+    testInteger(0x7f, 1);
+
+    // Long.MAX_VALUE = 0x7fffffffffffffffL
+    testLong(0x7fffffffffffffffL, 8 + 1);
+    testLong(0x7fffffffffffffL, 7 + 1);
+    testLong(0x7fffffffffffL, 6 + 1);
+    testLong(0x7fffffffffL, 5 + 1);
+    testLong(0x7fffffffL, 4 + 1);
+    testLong(0x7fffffL, 3 + 1);
+    testLong(0x7fffL, 2 + 1);
+    testLong(0x7fL, 1);
+  }
+
+  private void testInteger(int value, int length) throws Exception {
+    byte[] integerBuffer = new byte[5];
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos)) {
+      UnsynchronizedBuffer.writeVInt(dos, integerBuffer, value);
+      dos.flush();
+      assertEquals(length, baos.toByteArray().length);
+    }
+  }
+
+  private void testLong(long value, int length) throws Exception {
+    byte[] longBuffer = new byte[9];
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos)) {
+      UnsynchronizedBuffer.writeVLong(dos, longBuffer, value);
+      dos.flush();
+      assertEquals(length, baos.toByteArray().length);
+    }
+  }
+
+  @Test
+  public void compareWithWritableUtils() throws Exception {
+    byte[] hadoopBytes;
+    byte[] accumuloBytes;
+    int oneByteInt = 0x7f;
+    int threeByteInt = 0x7fff;
+    long sixByteLong = 0x7fffffffffL;
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos)) {
+      WritableUtils.writeVInt(dos, oneByteInt);
+      WritableUtils.writeVInt(dos, threeByteInt);
+      WritableUtils.writeVLong(dos, sixByteLong);
+      dos.flush();
+      hadoopBytes = baos.toByteArray();
+    }
+    try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        DataOutputStream dos = new DataOutputStream(baos)) {
+      UnsynchronizedBuffer.writeVInt(dos, new byte[5], oneByteInt);
+      UnsynchronizedBuffer.writeVInt(dos, new byte[5], threeByteInt);
+      UnsynchronizedBuffer.writeVLong(dos, new byte[9], sixByteLong);
+      dos.flush();
+      accumuloBytes = baos.toByteArray();
+    }
+    assertTrue("The byte array written to by UnsynchronizedBuffer is not equal to WritableUtils",
+        Arrays.equals(hadoopBytes, accumuloBytes));
+  }
 }
diff --git a/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
b/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
index a86fd85..9dfd850 100644
--- a/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
+++ b/server/base/src/main/java/org/apache/accumulo/server/data/ServerMutation.java
@@ -23,6 +23,7 @@ import java.io.IOException;
 import org.apache.accumulo.core.data.ColumnUpdate;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.data.thrift.TMutation;
+import org.apache.accumulo.core.util.UnsynchronizedBuffer;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.WritableUtils;
 
@@ -57,8 +58,9 @@ public class ServerMutation extends Mutation {
 
   @Override
   public void write(DataOutput out) throws IOException {
+    final byte[] timeBuffer = new byte[9];
     super.write(out);
-    WritableUtils.writeVLong(out, systemTime);
+    UnsynchronizedBuffer.writeVLong(out, timeBuffer, systemTime);
   }
 
   public void setSystemTimestamp(long v) {


Mime
View raw message