hama-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From edwardy...@apache.org
Subject svn commit: r1682806 - in /hama/trunk: ./ core/src/main/java/org/apache/hama/util/ graph/src/main/java/org/apache/hama/graph/
Date Sun, 31 May 2015 23:21:20 GMT
Author: edwardyoon
Date: Sun May 31 23:21:20 2015
New Revision: 1682806

URL: http://svn.apache.org/r1682806
Log:
HAMA-955: Support UnsafeByteArrayInputStream and UnSafeByteArrayOutputStream

Added:
    hama/trunk/core/src/main/java/org/apache/hama/util/ByteUtils.java
    hama/trunk/core/src/main/java/org/apache/hama/util/ExtendedDataOutput.java
    hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayInputStream.java
    hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayOutputStream.java
Modified:
    hama/trunk/CHANGES.txt
    hama/trunk/core/src/main/java/org/apache/hama/util/WritableUtils.java
    hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJob.java
    hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java
    hama/trunk/graph/src/main/java/org/apache/hama/graph/MapVerticesInfo.java
    hama/trunk/graph/src/main/java/org/apache/hama/graph/Vertex.java

Modified: hama/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hama/trunk/CHANGES.txt?rev=1682806&r1=1682805&r2=1682806&view=diff
==============================================================================
--- hama/trunk/CHANGES.txt (original)
+++ hama/trunk/CHANGES.txt Sun May 31 23:21:20 2015
@@ -25,7 +25,8 @@ Release 0.7.0 (unreleased changes)
    HAMA-885: Semi-Clustering is not producing expected output (Renil J via edwardyoon)
 
   IMPROVEMENTS
-
+  
+   HAMA-955: Support UnsafeByteArrayInputStream and UnSafeByteArrayOutputStream (Minho Kim
via edwardyoon)
    HAMA-944: Add JSON format option to fastgen command (Minho Kim via edwardyoon)
    HAMA-919: Manage messages per Vertex (edwardyoon)
    HAMA-923: add a toString() method for FloatArrayWritable and TextArrayWritable classes
(edwardyoon)

Added: hama/trunk/core/src/main/java/org/apache/hama/util/ByteUtils.java
URL: http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/util/ByteUtils.java?rev=1682806&view=auto
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/util/ByteUtils.java (added)
+++ hama/trunk/core/src/main/java/org/apache/hama/util/ByteUtils.java Sun May 31 23:21:20
2015
@@ -0,0 +1,40 @@
+/**
+ * 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.hama.util;
+
+/**
+ * Utilities class for byte operations and constants
+ */
+public class ByteUtils {
+  /** Bytes used in a boolean */
+  public static final int SIZE_OF_BOOLEAN = 1;
+  /** Bytes used in a byte */
+  public static final int SIZE_OF_BYTE = 1;
+  /** Bytes used in a char */
+  public static final int SIZE_OF_CHAR = Character.SIZE / Byte.SIZE;
+  /** Bytes used in a short */
+  public static final int SIZE_OF_SHORT = Short.SIZE / Byte.SIZE;
+  /** Bytes used in an int */
+  public static final int SIZE_OF_INT = Integer.SIZE / Byte.SIZE;
+  /** Bytes used in a long */
+  public static final int SIZE_OF_LONG = Long.SIZE / Byte.SIZE;
+  /** Bytes used in a float */
+  public static final int SIZE_OF_FLOAT = Float.SIZE / Byte.SIZE;
+  /** Bytes used in a double */
+  public static final int SIZE_OF_DOUBLE = Double.SIZE / Byte.SIZE;
+}

Added: hama/trunk/core/src/main/java/org/apache/hama/util/ExtendedDataOutput.java
URL: http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/util/ExtendedDataOutput.java?rev=1682806&view=auto
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/util/ExtendedDataOutput.java (added)
+++ hama/trunk/core/src/main/java/org/apache/hama/util/ExtendedDataOutput.java Sun May 31
23:21:20 2015
@@ -0,0 +1,84 @@
+/**
+ * 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.hama.util;
+
+import java.io.DataOutput;
+
+/**
+ * Add a few features to data output
+ */
+public interface ExtendedDataOutput extends DataOutput {
+  /**
+   * Ensure that backing byte structure has at least minSize
+   * additional bytes
+   *
+   * @param minSize additional size required
+   */
+  void ensureWritable(int minSize);
+
+  /**
+   * Skip some number of bytes.
+   *
+   * @param  bytesToSkip Number of bytes to skip
+   */
+  void skipBytes(int bytesToSkip);
+
+  /**
+   * In order to write a size as a first part of an data output, it is
+   * useful to be able to write an int at an arbitrary location in the stream
+   *
+   * @param pos Byte position in the output stream
+   * @param value Value to write
+   */
+  void writeInt(int pos, int value);
+
+  /**
+   * Get the position in the output stream
+   *
+   * @return Position in the output stream
+   */
+  int getPos();
+
+  /**
+   * Get the internal byte array (if possible), read-only
+   *
+   * @return Internal byte array (do not modify)
+   */
+  byte[] getByteArray();
+
+  /**
+   * Copies the internal byte array
+   *
+   * @return Copied byte array
+   */
+  byte[] toByteArray();
+
+  /**
+   * Return a copy of slice of byte array
+   *
+   * @param offset offset of array
+   * @param length length of slice
+   * @return byte array
+   */
+  byte[] toByteArray(int offset, int length);
+
+  /**
+   * Clears the buffer
+   */
+  void reset();
+}

Added: hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayInputStream.java
URL: http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayInputStream.java?rev=1682806&view=auto
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayInputStream.java (added)
+++ hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayInputStream.java Sun
May 31 23:21:20 2015
@@ -0,0 +1,139 @@
+/**
+ * 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.hama.util;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+
+/**
+ * This class, much like {@link ByteArrayInputStream} uses a given buffer as a
+ * source of an InputStream. Unlike ByteArrayInputStream, this class does not
+ * "waste" memory by creating a local copy of the given buffer, but rather uses
+ * the given buffer as is. Hence the name Unsafe. While using this class one
+ * should remember that the byte[] buffer memory is shared and might be changed
+ * from outside.
+ *
+ * For reuse-ability, a call for {@link #reInit(byte[])} can be called, and
+ * initialize the stream with a new buffer.
+ *
+ */
+public class UnsafeByteArrayInputStream extends InputStream {
+  private byte[] buffer;
+  private int markIndex;
+  private int upperLimit;
+  private int index;
+
+  /**
+   * Creates a new instance by not using any byte[] up front. If you use this
+   * constructor, you MUST call either of the {@link #reInit(byte[]) reInit}
+   * methods before you consume any byte from this instance.<br>
+   * This constructor is for convenience purposes only, so that if one does not
+   * have the byte[] at the moment of creation, one is not forced to pass a
+   * <code>new byte[0]</code> or something. Obviously in that case, one will
+   * call either {@link #reInit(byte[]) reInit} methods before using the class.
+   */
+  public UnsafeByteArrayInputStream() {
+    markIndex = upperLimit = index = 0;
+  }
+
+  /**
+   * Creates an UnsafeByteArrayInputStream which uses a given byte array as the
+   * source of the stream. Default range is [0 , buffer.length)
+   *
+   * @param buffer byte array used as the source of this stream
+   */
+  public UnsafeByteArrayInputStream(byte[] buffer) {
+    reInit(buffer, 0, buffer.length);
+  }
+
+  /**
+   * Creates an UnsafeByteArrayInputStream which uses a given byte array as the
+   * source of the stream, at the specific range: [startPos, endPos)
+   *
+   * @param buffer byte array used as the source of this stream
+   * @param startPos first index (inclusive) to the data lying in the given
+   *          buffer
+   * @param endPos an index (exclusive) where the data ends. data @
+   *          buffer[endPos] will never be read
+   */
+  public UnsafeByteArrayInputStream(byte[] buffer, int startPos, int endPos) {
+    reInit(buffer, startPos, endPos);
+  }
+
+  @Override
+  public void mark(int readlimit) {
+    markIndex = index;
+  }
+
+  @Override
+  public boolean markSupported() {
+    return true;
+  }
+
+  /**
+   * Initialize the stream with a given buffer, using the default limits of [0,
+   * buffer.length)
+   *
+   * @param buffer byte array used as the source of this stream
+   */
+  public void reInit(byte[] buffer) {
+    reInit(buffer, 0, buffer.length);
+  }
+
+  /**
+   * Initialize the stream with a given byte array as the source of the stream,
+   * at the specific range: [startPos, endPos)
+   *
+   * @param buffer byte array used as the source of this stream
+   * @param startPos first index (inclusive) to the data lying in the given
+   *          buffer
+   * @param endPos an index (exclusive) where the data ends. data @
+   *          buffer[endPos] will never be read
+   */
+  public void reInit(byte[] buffer, int startPos, int endPos) {
+    this.buffer = buffer;
+    markIndex = startPos;
+    upperLimit = endPos;
+    index = markIndex;
+  }
+
+  @Override
+  public int available() throws IOException {
+    return upperLimit - index;
+  }
+
+  /**
+   * Read a byte. Data returned as an integer [0,255] If end of stream reached,
+   * returns -1
+   */
+  @Override
+  public int read() throws IOException {
+    return index < upperLimit ? buffer[index++] & 0xff : -1;
+  }
+
+  /**
+   * Resets the stream back to its original state. Basically - moving the index
+   * back to start position.
+   */
+  @Override
+  public void reset() throws IOException {
+    index = markIndex;
+  }
+}

Added: hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayOutputStream.java
URL: http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayOutputStream.java?rev=1682806&view=auto
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayOutputStream.java (added)
+++ hama/trunk/core/src/main/java/org/apache/hama/util/UnsafeByteArrayOutputStream.java Sun
May 31 23:21:20 2015
@@ -0,0 +1,311 @@
+/**
+ * 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.hama.util;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.util.Arrays;
+
+/**
+ * Byte array output stream that uses Unsafe methods to serialize/deserialize
+ * much faster
+ */
+@SuppressWarnings("restriction")
+public class UnsafeByteArrayOutputStream extends OutputStream implements
+    ExtendedDataOutput {
+  static {
+    try {
+      Field field = sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
+      field.setAccessible(true);
+      UNSAFE = (sun.misc.Unsafe) field.get(null);
+      // Checkstyle exception due to needing to check if unsafe is allowed
+      // CHECKSTYLE: stop IllegalCatch
+    } catch (Exception e) {
+      // CHECKSTYLE: resume IllegalCatch
+      throw new RuntimeException("UnsafeByteArrayOutputStream: Failed to "
+          + "get unsafe", e);
+    }
+  }
+
+  /** Default number of bytes */
+  private static final int DEFAULT_BYTES = 32;
+  /** Access to the unsafe class */
+  private static final sun.misc.Unsafe UNSAFE;
+
+  /** Offset of a byte array */
+  private static final long BYTE_ARRAY_OFFSET = UNSAFE
+      .arrayBaseOffset(byte[].class);
+
+  /** Byte buffer */
+  private byte[] buf;
+  /** Position in the buffer */
+  private int pos = 0;
+
+  /**
+   * Constructor
+   */
+  public UnsafeByteArrayOutputStream() {
+    this(DEFAULT_BYTES);
+  }
+
+  /**
+   * Constructor
+   *
+   * @param size Initial size of the underlying byte array
+   */
+  public UnsafeByteArrayOutputStream(int size) {
+    buf = new byte[size];
+  }
+
+  /**
+   * Constructor to take in a buffer
+   *
+   * @param buf Buffer to start with, or if null, create own buffer
+   */
+  public UnsafeByteArrayOutputStream(byte[] buf) {
+    if (buf == null) {
+      this.buf = new byte[DEFAULT_BYTES];
+    } else {
+      this.buf = buf;
+    }
+  }
+
+  /**
+   * Constructor to take in a buffer with a given position into that buffer
+   *
+   * @param buf Buffer to start with
+   * @param pos Position to write at the buffer
+   */
+  public UnsafeByteArrayOutputStream(byte[] buf, int pos) {
+    this(buf);
+    this.pos = pos;
+  }
+
+  /**
+   * Ensure that this buffer has enough remaining space to add the size. Creates
+   * and copies to a new buffer if necessary
+   *
+   * @param size Size to add
+   */
+  private void ensureSize(int size) {
+    if (pos + size > buf.length) {
+      byte[] newBuf = new byte[(buf.length + size) << 1];
+      System.arraycopy(buf, 0, newBuf, 0, pos);
+      buf = newBuf;
+    }
+  }
+
+  @Override
+  public byte[] getByteArray() {
+    return buf;
+  }
+
+  @Override
+  public byte[] toByteArray() {
+    return Arrays.copyOf(buf, pos);
+  }
+
+  @Override
+  public byte[] toByteArray(int offset, int length) {
+    if (offset + length > pos) {
+      throw new IndexOutOfBoundsException(String.format("Offset: %d + "
+          + "Length: %d exceeds the size of buf : %d", offset, length, pos));
+    }
+    return Arrays.copyOfRange(buf, offset, length);
+  }
+
+  @Override
+  public void reset() {
+    pos = 0;
+  }
+
+  @Override
+  public int getPos() {
+    return pos;
+  }
+
+  @Override
+  public void write(int b) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_BYTE);
+    buf[pos] = (byte) b;
+    pos += ByteUtils.SIZE_OF_BYTE;
+  }
+
+  @Override
+  public void write(byte[] b) throws IOException {
+    ensureSize(b.length);
+    System.arraycopy(b, 0, buf, pos, b.length);
+    pos += b.length;
+  }
+
+  @Override
+  public void write(byte[] b, int off, int len) throws IOException {
+    ensureSize(len);
+    System.arraycopy(b, off, buf, pos, len);
+    pos += len;
+  }
+
+  @Override
+  public void writeBoolean(boolean v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_BOOLEAN);
+    UNSAFE.putBoolean(buf, BYTE_ARRAY_OFFSET + pos, v);
+    pos += ByteUtils.SIZE_OF_BOOLEAN;
+  }
+
+  @Override
+  public void writeByte(int v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_BYTE);
+    UNSAFE.putByte(buf, BYTE_ARRAY_OFFSET + pos, (byte) v);
+    pos += ByteUtils.SIZE_OF_BYTE;
+  }
+
+  @Override
+  public void writeShort(int v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_SHORT);
+    UNSAFE.putShort(buf, BYTE_ARRAY_OFFSET + pos, (short) v);
+    pos += ByteUtils.SIZE_OF_SHORT;
+  }
+
+  @Override
+  public void writeChar(int v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_CHAR);
+    UNSAFE.putChar(buf, BYTE_ARRAY_OFFSET + pos, (char) v);
+    pos += ByteUtils.SIZE_OF_CHAR;
+  }
+
+  @Override
+  public void writeInt(int v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_INT);
+    UNSAFE.putInt(buf, BYTE_ARRAY_OFFSET + pos, v);
+    pos += ByteUtils.SIZE_OF_INT;
+  }
+
+  @Override
+  public void ensureWritable(int minSize) {
+    if ((pos + minSize) > buf.length) {
+      buf = Arrays.copyOf(buf, Math.max(buf.length << 1, pos + minSize));
+    }
+  }
+
+  @Override
+  public void skipBytes(int bytesToSkip) {
+    ensureWritable(bytesToSkip);
+    pos += bytesToSkip;
+  }
+
+  @Override
+  public void writeInt(int pos, int value) {
+    if (pos + ByteUtils.SIZE_OF_INT > this.pos) {
+      throw new IndexOutOfBoundsException(
+          "writeInt: Tried to write int to position " + pos
+              + " but current length is " + this.pos);
+    }
+    UNSAFE.putInt(buf, BYTE_ARRAY_OFFSET + pos, value);
+  }
+
+  @Override
+  public void writeLong(long v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_LONG);
+    UNSAFE.putLong(buf, BYTE_ARRAY_OFFSET + pos, v);
+    pos += ByteUtils.SIZE_OF_LONG;
+  }
+
+  @Override
+  public void writeFloat(float v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_FLOAT);
+    UNSAFE.putFloat(buf, BYTE_ARRAY_OFFSET + pos, v);
+    pos += ByteUtils.SIZE_OF_FLOAT;
+  }
+
+  @Override
+  public void writeDouble(double v) throws IOException {
+    ensureSize(ByteUtils.SIZE_OF_DOUBLE);
+    UNSAFE.putDouble(buf, BYTE_ARRAY_OFFSET + pos, v);
+    pos += ByteUtils.SIZE_OF_DOUBLE;
+  }
+
+  @Override
+  public void writeBytes(String s) throws IOException {
+    // Note that this code is mostly copied from DataOutputStream
+    int len = s.length();
+    ensureSize(len);
+    for (int i = 0; i < len; i++) {
+      int v = s.charAt(i);
+      writeByte(v);
+    }
+  }
+
+  @Override
+  public void writeChars(String s) throws IOException {
+    // Note that this code is mostly copied from DataOutputStream
+    int len = s.length();
+    ensureSize(len * ByteUtils.SIZE_OF_CHAR);
+    for (int i = 0; i < len; i++) {
+      int v = s.charAt(i);
+      writeChar(v);
+    }
+  }
+
+  @Override
+  public void writeUTF(String s) throws IOException {
+    // Note that this code is mostly copied from DataOutputStream
+    int strlen = s.length();
+    int utflen = 0;
+    int c;
+
+    /* use charAt instead of copying String to char array */
+    for (int i = 0; i < strlen; i++) {
+      c = s.charAt(i);
+      if ((c >= 0x0001) && (c <= 0x007F)) {
+        utflen++;
+      } else if (c > 0x07FF) {
+        utflen += 3;
+      } else {
+        utflen += 2;
+      }
+    }
+
+    ensureSize(utflen + ByteUtils.SIZE_OF_SHORT);
+    writeShort(utflen);
+
+    int i = 0;
+    for (i = 0; i < strlen; i++) {
+      c = s.charAt(i);
+      if (!((c >= 0x0001) && (c <= 0x007F))) {
+        break;
+      }
+      buf[pos++] = (byte) c;
+    }
+
+    for (; i < strlen; i++) {
+      c = s.charAt(i);
+      if ((c >= 0x0001) && (c <= 0x007F)) {
+        buf[pos++] = (byte) c;
+
+      } else if (c > 0x07FF) {
+        buf[pos++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
+        buf[pos++] = (byte) (0x80 | ((c >> 6) & 0x3F));
+        buf[pos++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+      } else {
+        buf[pos++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
+        buf[pos++] = (byte) (0x80 | ((c >> 0) & 0x3F));
+      }
+    }
+  }
+}

Modified: hama/trunk/core/src/main/java/org/apache/hama/util/WritableUtils.java
URL: http://svn.apache.org/viewvc/hama/trunk/core/src/main/java/org/apache/hama/util/WritableUtils.java?rev=1682806&r1=1682805&r2=1682806&view=diff
==============================================================================
--- hama/trunk/core/src/main/java/org/apache/hama/util/WritableUtils.java (original)
+++ hama/trunk/core/src/main/java/org/apache/hama/util/WritableUtils.java Sun May 31 23:21:20
2015
@@ -47,4 +47,24 @@ public class WritableUtils {
       e.printStackTrace();
     }
   }
+
+  public static byte[] unsafeSerialize(Writable w) {
+    UnsafeByteArrayOutputStream out = new UnsafeByteArrayOutputStream();
+    DataOutput output = new DataOutputStream(out);
+    try {
+      w.write(output);
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+    return out.toByteArray();
+  }
+
+  public static void unsafeDeserialize(byte[] bytes, Writable obj) {
+    DataInputStream in = new DataInputStream(new UnsafeByteArrayInputStream(bytes));
+    try {
+      obj.readFields(in);
+    } catch (IOException e) {
+      e.printStackTrace();
+    }
+  }
 }

Modified: hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJob.java
URL: http://svn.apache.org/viewvc/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJob.java?rev=1682806&r1=1682805&r2=1682806&view=diff
==============================================================================
--- hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJob.java (original)
+++ hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJob.java Sun May 31 23:21:20
2015
@@ -60,6 +60,7 @@ public class GraphJob extends BSPJob {
         OutgoingVertexMessageManager.class, OutgoingMessageManager.class);
     conf.setBoolean(Constants.FORCE_SET_BSP_TASKS, true);
     conf.setBoolean(Constants.ENABLE_RUNTIME_PARTITIONING, false);
+    conf.setBoolean("hama.use.unsafeserialization", true);
     
     this.setBspClass(GraphJobRunner.class);
     this.setJarByClass(exampleClass);

Modified: hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java
URL: http://svn.apache.org/viewvc/hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java?rev=1682806&r1=1682805&r2=1682806&view=diff
==============================================================================
--- hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java (original)
+++ hama/trunk/graph/src/main/java/org/apache/hama/graph/GraphJobRunner.java Sun May 31 23:21:20
2015
@@ -22,6 +22,7 @@ import java.io.DataInputStream;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executors;
@@ -49,6 +50,7 @@ import org.apache.hama.bsp.Partitioner;
 import org.apache.hama.bsp.sync.SyncException;
 import org.apache.hama.commons.util.KeyValuePair;
 import org.apache.hama.util.ReflectionUtils;
+import org.apache.hama.util.UnsafeByteArrayInputStream;
 import org.apache.hama.util.WritableUtils;
 
 /**
@@ -112,7 +114,7 @@ public final class GraphJobRunner<V exte
   // global counter for thread exceptions
   // TODO find more graceful way to handle thread exceptions.
   private AtomicInteger errorCount = new AtomicInteger(0);
-  
+
   private AggregationRunner<V, E, M> aggregationRunner;
   private VertexOutputWriter<Writable, Writable, V, E, M> vertexOutputWriter;
   private Combiner<Writable> combiner;
@@ -121,7 +123,8 @@ public final class GraphJobRunner<V exte
 
   private RejectedExecutionHandler retryHandler = new RetryRejectedExecutionHandler();
 
-  // Below maps are used for grouping messages into single GraphJobMessage, based on vertex
ID.
+  // Below maps are used for grouping messages into single GraphJobMessage,
+  // based on vertex ID.
   private final ConcurrentHashMap<Integer, GraphJobMessage> partitionMessages = new
ConcurrentHashMap<Integer, GraphJobMessage>();
   private final ConcurrentHashMap<V, GraphJobMessage> vertexMessages = new ConcurrentHashMap<V,
GraphJobMessage>();
 
@@ -259,7 +262,7 @@ public final class GraphJobRunner<V exte
 
     ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
         .newCachedThreadPool();
-    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 256));
+    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 64));
     executor.setRejectedExecutionHandler(retryHandler);
 
     long loopStartTime = System.currentTimeMillis();
@@ -319,7 +322,7 @@ public final class GraphJobRunner<V exte
 
     ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
         .newCachedThreadPool();
-    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 256));
+    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 64));
     executor.setRejectedExecutionHandler(retryHandler);
 
     for (V v : vertices.keySet()) {
@@ -337,7 +340,7 @@ public final class GraphJobRunner<V exte
       throw new IOException("there were " + errorCount
           + " exceptions during compute vertices.");
     }
-    
+
     getAggregationRunner().sendAggregatorValues(peer, 1, this.changedVertexCnt);
     iteration++;
     finishSuperstep();
@@ -455,10 +458,11 @@ public final class GraphJobRunner<V exte
 
     ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors
         .newCachedThreadPool();
-    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 256));
+    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 64));
     executor.setRejectedExecutionHandler(retryHandler);
 
     KeyValuePair<Writable, Writable> next = null;
+
     while ((next = peer.readNext()) != null) {
       Vertex<V, E, M> vertex = GraphJobRunner
           .<V, E, M> newVertexInstance(VERTEX_CLASS);
@@ -496,7 +500,7 @@ public final class GraphJobRunner<V exte
     peer.sync();
 
     executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();
-    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 256));
+    executor.setMaximumPoolSize(conf.getInt(DEFAULT_THREAD_POOL_SIZE, 64));
     executor.setRejectedExecutionHandler(retryHandler);
 
     GraphJobMessage msg;
@@ -695,26 +699,50 @@ public final class GraphJobRunner<V exte
     vertices.finishAdditions();
   }
 
-  public void sendMessage(V vertexID, byte[] msg) throws IOException {
+  public void sendMessage(V vertexID, M msg) throws IOException {
     if (!vertexMessages.containsKey(vertexID)) {
-      // To save bit memory we don't set vertexID twice
       vertexMessages.putIfAbsent(vertexID, new GraphJobMessage());
     }
-    vertexMessages.get(vertexID).add(msg);
+    if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+      vertexMessages.get(vertexID).add(WritableUtils.serialize(msg));
+    } else {
+      vertexMessages.get(vertexID).add(WritableUtils.unsafeSerialize(msg));
+    }
+  }
+
+  public void sendMessage(List<Edge<V, E>> outEdges, M msg) throws IOException
{
+    byte[] serialized;
+    if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+      serialized = WritableUtils.serialize(msg);
+    } else {
+      serialized = WritableUtils.unsafeSerialize(msg);
+    }
+
+    for (Edge<V, E> e : outEdges) {
+      if (!vertexMessages.containsKey(e.getDestinationVertexID())) {
+        vertexMessages.putIfAbsent(e.getDestinationVertexID(),
+            new GraphJobMessage());
+      }
+
+      vertexMessages.get(e.getDestinationVertexID()).add(serialized);
+    }
   }
 
   public void finishSuperstep() throws IOException {
     vertices.finishSuperstep();
 
-    Iterator<Entry<V, GraphJobMessage>> it = vertexMessages.entrySet().iterator();
+    Iterator<Entry<V, GraphJobMessage>> it = vertexMessages.entrySet()
+        .iterator();
     while (it.hasNext()) {
       Entry<V, GraphJobMessage> e = it.next();
       it.remove();
 
       if (combiner != null && e.getValue().getNumOfValues() > 1) {
-        GraphJobMessage combined = new GraphJobMessage(e.getKey(),
+        GraphJobMessage combined;
+        combined = new GraphJobMessage(e.getKey(),
             WritableUtils.serialize(combiner.combine(getIterableMessages(e
                 .getValue().getValuesBytes(), e.getValue().getNumOfValues()))));
+
         combined.setFlag(GraphJobMessage.VERTEX_FLAG);
         peer.send(getHostName(e.getKey()), combined);
       } else {
@@ -732,13 +760,19 @@ public final class GraphJobRunner<V exte
 
   public Iterable<Writable> getIterableMessages(final byte[] valuesBytes,
       final int numOfValues) {
-
+    
     return new Iterable<Writable>() {
+      DataInputStream dis;
+      
       @Override
       public Iterator<Writable> iterator() {
+        if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+          dis = new DataInputStream(new ByteArrayInputStream(valuesBytes));
+        } else {
+          dis = new DataInputStream(new UnsafeByteArrayInputStream(valuesBytes));
+        }
+        
         return new Iterator<Writable>() {
-          ByteArrayInputStream bis = new ByteArrayInputStream(valuesBytes);
-          DataInputStream dis = new DataInputStream(bis);
           int index = 0;
 
           @Override

Modified: hama/trunk/graph/src/main/java/org/apache/hama/graph/MapVerticesInfo.java
URL: http://svn.apache.org/viewvc/hama/trunk/graph/src/main/java/org/apache/hama/graph/MapVerticesInfo.java?rev=1682806&r1=1682805&r2=1682806&view=diff
==============================================================================
--- hama/trunk/graph/src/main/java/org/apache/hama/graph/MapVerticesInfo.java (original)
+++ hama/trunk/graph/src/main/java/org/apache/hama/graph/MapVerticesInfo.java Sun May 31 23:21:20
2015
@@ -44,26 +44,36 @@ public final class MapVerticesInfo<V ext
   private final ConcurrentHashMap<V, byte[]> vertices = new ConcurrentHashMap<V,
byte[]>();
 
   private GraphJobRunner<V, E, M> runner;
-
+  private HamaConfiguration conf;
   private AtomicInteger activeVertices = new AtomicInteger(0);
 
   @Override
   public void init(GraphJobRunner<V, E, M> runner, HamaConfiguration conf,
       TaskAttemptID attempt) throws IOException {
     this.runner = runner;
+    this.conf = conf;
   }
 
   @Override
   public void put(Vertex<V, E, M> vertex) throws IOException {
     if (!vertices.containsKey(vertex.getVertexID())) {
-      vertices.putIfAbsent(vertex.getVertexID(),
-          WritableUtils.serialize(vertex));
+      if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+        vertices.putIfAbsent(vertex.getVertexID(),
+            WritableUtils.serialize(vertex));
+      } else {
+        vertices.putIfAbsent(vertex.getVertexID(),
+            WritableUtils.unsafeSerialize(vertex));
+      }
     } else {
       Vertex<V, E, M> v = this.get(vertex.getVertexID());
       for (Edge<V, E> e : vertex.getEdges()) {
         v.addEdge(e);
       }
-      vertices.put(vertex.getVertexID(), WritableUtils.serialize(v));
+      if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+        vertices.put(vertex.getVertexID(), WritableUtils.serialize(v));
+      } else {
+        vertices.put(vertex.getVertexID(), WritableUtils.unsafeSerialize(v));
+      }
     }
   }
 
@@ -85,7 +95,11 @@ public final class MapVerticesInfo<V ext
   public Vertex<V, E, M> get(V vertexID) throws IOException {
     Vertex<V, E, M> v = GraphJobRunner
         .<V, E, M> newVertexInstance(GraphJobRunner.VERTEX_CLASS);
-    WritableUtils.deserialize(vertices.get(vertexID), v);
+    if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+      WritableUtils.deserialize(vertices.get(vertexID), v);
+    } else {
+      WritableUtils.unsafeDeserialize(vertices.get(vertexID), v);
+    }
     v.setRunner(runner);
 
     return v;
@@ -107,7 +121,13 @@ public final class MapVerticesInfo<V ext
       public Vertex<V, E, M> next() {
         Vertex<V, E, M> v = GraphJobRunner
             .<V, E, M> newVertexInstance(GraphJobRunner.VERTEX_CLASS);
-        WritableUtils.deserialize(it.next(), v);
+
+        if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+          WritableUtils.deserialize(it.next(), v);
+        } else {
+          WritableUtils.unsafeDeserialize(it.next(), v);
+        }
+
         v.setRunner(runner);
         return v;
       }
@@ -130,7 +150,11 @@ public final class MapVerticesInfo<V ext
       throws IOException {
     incrementCount();
     vertex.setComputed();
-    vertices.put(vertex.getVertexID(), WritableUtils.serialize(vertex));
+    if (!conf.getBoolean("hama.use.unsafeserialization", false)) {
+      vertices.put(vertex.getVertexID(), WritableUtils.serialize(vertex));
+    } else {
+      vertices.put(vertex.getVertexID(), WritableUtils.unsafeSerialize(vertex));
+    }
   }
 
   public void incrementCount() {

Modified: hama/trunk/graph/src/main/java/org/apache/hama/graph/Vertex.java
URL: http://svn.apache.org/viewvc/hama/trunk/graph/src/main/java/org/apache/hama/graph/Vertex.java?rev=1682806&r1=1682805&r2=1682806&view=diff
==============================================================================
--- hama/trunk/graph/src/main/java/org/apache/hama/graph/Vertex.java (original)
+++ hama/trunk/graph/src/main/java/org/apache/hama/graph/Vertex.java Sun May 31 23:21:20 2015
@@ -30,7 +30,6 @@ import org.apache.hadoop.io.WritableComp
 import org.apache.hama.HamaConfiguration;
 import org.apache.hama.bsp.BSPPeer;
 import org.apache.hama.bsp.Counters.Counter;
-import org.apache.hama.util.WritableUtils;
 
 /**
  * Vertex is a abstract definition of Google Pregel Vertex. For implementing a
@@ -61,7 +60,7 @@ public abstract class Vertex<V extends W
 
   private boolean votedToHalt = false;
   private long lastComputedSuperstep = 0;
-  
+
   public HamaConfiguration getConf() {
     return runner.getPeer().getConfiguration();
   }
@@ -80,22 +79,17 @@ public abstract class Vertex<V extends W
 
   @Override
   public void sendMessage(Edge<V, E> e, M msg) throws IOException {
-    runner.sendMessage(e.getDestinationVertexID(),
-        WritableUtils.serialize(msg));
+    runner.sendMessage(e.getDestinationVertexID(), msg);
   }
 
   @Override
   public void sendMessage(V destinationVertexID, M msg) throws IOException {
-    runner.sendMessage(destinationVertexID, WritableUtils.serialize(msg));
+    runner.sendMessage(destinationVertexID, msg);
   }
 
   @Override
   public void sendMessageToNeighbors(M msg) throws IOException {
-    final List<Edge<V, E>> outEdges = this.getEdges();
-    byte[] serialized = WritableUtils.serialize(msg);
-    for (Edge<V, E> e : outEdges) {
-      runner.sendMessage(e.getDestinationVertexID(), serialized);
-    }
+    runner.sendMessage(this.getEdges(), msg);
   }
 
   private void alterVertexCounter(int i) throws IOException {
@@ -204,11 +198,11 @@ public abstract class Vertex<V extends W
   void setComputed() {
     this.lastComputedSuperstep = this.getSuperstepCount();
   }
-  
+
   public boolean isComputed() {
     return (lastComputedSuperstep == this.getSuperstepCount()) ? true : false;
   }
-  
+
   void setVotedToHalt(boolean votedToHalt) {
     this.votedToHalt = votedToHalt;
   }
@@ -257,7 +251,7 @@ public abstract class Vertex<V extends W
     }
 
     this.lastComputedSuperstep = in.readLong();
-    
+
     this.edges = new ArrayList<Edge<V, E>>();
     if (in.readBoolean()) {
       int num = in.readInt();
@@ -288,16 +282,16 @@ public abstract class Vertex<V extends W
       out.writeBoolean(true);
       vertexID.write(out);
     }
-    
+
     if (value == null) {
       out.writeBoolean(false);
     } else {
       out.writeBoolean(true);
       value.write(out);
     }
-    
+
     out.writeLong(lastComputedSuperstep);
-    
+
     if (this.edges == null) {
       out.writeBoolean(false);
     } else {



Mime
View raw message