Merge pull request #104 from msgpack/v07-unpack-binary

Optimize unpackBinary

Author: xerial

msgpack-core/src/main/java/org/msgpack/core/MessageUnpacker.java

@@ -117,7 +117,7 @@ public MessageUnpacker(byte[] arr) {
      * @param in
      */
     public MessageUnpacker(InputStream in) {
-        this(new InputStreamBufferInput(in));
+        this(InputStreamBufferInput.newBufferInput(in));
     }
 
     /**
@@ -992,6 +992,10 @@ public void readPayload(ByteBuffer dst) throws IOException {
         }
     }
 
+    public void readPayload(byte[] dst) throws IOException {
+        readPayload(dst, 0, dst.length);
+    }
+
     /**
      * Read up to len bytes of data into the destination array
      *

msgpack-core/src/main/java/org/msgpack/core/buffer/ArrayBufferInput.java

@@ -12,17 +12,21 @@ public class ArrayBufferInput implements MessageBufferInput {
     private boolean isRead = false;
 
     public ArrayBufferInput(byte[] arr) {
-        this.buffer = MessageBuffer.wrap(checkNotNull(arr, "input array is null"));
+        this(arr, 0, arr.length);
     }
 
+    public ArrayBufferInput(byte[] arr, int offset, int length) {
+        checkArgument(offset + length <= arr.length);
+        this.buffer = MessageBuffer.wrap(checkNotNull(arr, "input array is null")).slice(offset, length);
+    }
+
+
     @Override
     public MessageBuffer next() throws IOException {
-        if(isRead) {
+        if(isRead)
             return null;
-        } else {
-            isRead = true;
-            return buffer;
-        }
+        isRead = true;
+        return buffer;
     }
 
     @Override

msgpack-core/src/main/java/org/msgpack/core/buffer/ByteBufferInput.java

@@ -0,0 +1,34 @@
+package org.msgpack.core.buffer;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import static org.msgpack.core.Preconditions.*;
+
+/**
+ * {@link MessageBufferInput} adapter for {@link java.nio.ByteBuffer}
+ */
+public class ByteBufferInput implements MessageBufferInput {
+
+    private final ByteBuffer input;
+    private boolean isRead = false;
+
+    public ByteBufferInput(ByteBuffer input) {
+        this.input = checkNotNull(input, "input ByteBuffer is null");
+    }
+
+
+    @Override
+    public MessageBuffer next() throws IOException {
+        if(isRead)
+            return null;
+
+        isRead = true;
+        return MessageBuffer.wrap(input);
+    }
+
+
+    @Override
+    public void close() throws IOException {
+        // Nothing to do
+    }
+}

msgpack-core/src/main/java/org/msgpack/core/buffer/ChannelBufferInput.java

@@ -4,30 +4,43 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
 
-import static org.msgpack.core.Preconditions.checkNotNull;
+import static org.msgpack.core.Preconditions.*;
 
 /**
  * {@link MessageBufferInput} adapter for {@link java.nio.channels.ReadableByteChannel}
  */
 public class ChannelBufferInput implements MessageBufferInput {
 
     private final ReadableByteChannel channel;
+    private boolean reachedEOF = false;
+    private final int bufferSize;
 
     public ChannelBufferInput(ReadableByteChannel channel) {
+        this(channel, 8192);
+    }
+
+    public ChannelBufferInput(ReadableByteChannel channel, int bufferSize) {
         this.channel = checkNotNull(channel, "input channel is null");
+        checkArgument(bufferSize > 0, "buffer size must be > 0: " + bufferSize);
+        this.bufferSize = bufferSize;
     }
 
     @Override
     public MessageBuffer next() throws IOException {
-        MessageBuffer m = MessageBuffer.newBuffer(8192);
-        ByteBuffer b = m.toByteBuffer(0, m.size);
-        for(int ret = 0; (ret = channel.read(b)) != -1; ) {
+
+        if(reachedEOF)
+            return null;
+
+        MessageBuffer m = MessageBuffer.newBuffer(bufferSize);
+        ByteBuffer b = m.toByteBuffer();
+        while(!reachedEOF && b.remaining() > 0) {
+            int ret = channel.read(b);
+            if(ret == -1) {
+                reachedEOF = true;
+            }
         }
         b.flip();
-        if(b.remaining() < m.size)
-            return m.slice(0, b.remaining());
-        else
-            return m;
+        return b.remaining() == 0 ? null : m.slice(0, b.limit());
     }
 
     @Override

msgpack-core/src/main/java/org/msgpack/core/buffer/InputStreamBufferInput.java

@@ -1,7 +1,10 @@
 package org.msgpack.core.buffer;
 
+import java.io.ByteArrayInputStream;
+import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.lang.reflect.Field;
 
 import static org.msgpack.core.Preconditions.checkNotNull;
 
@@ -10,29 +13,82 @@
  */
 public class InputStreamBufferInput implements MessageBufferInput {
 
+    private static Field bufField;
+    private static Field bufPosField;
+    private static Field bufCountField;
+
+    private static Field getField(String name) {
+        Field f = null;
+        try {
+            f = ByteArrayInputStream.class.getDeclaredField(name);
+            f.setAccessible(true);
+        }
+        catch(Exception e) {
+            e.printStackTrace();
+        }
+        return f;
+    }
+
+    static {
+        bufField = getField("buf");
+        bufPosField = getField("pos");
+        bufCountField = getField("count");
+    }
+
     private final InputStream in;
-    private byte[] buffer = new byte[8192];
+    private final int bufferSize;
+    private boolean reachedEOF = false;
+
+    public static MessageBufferInput newBufferInput(InputStream in) {
+        checkNotNull(in, "InputStream is null");
+        if(in.getClass() == ByteArrayInputStream.class) {
+            ByteArrayInputStream b = (ByteArrayInputStream) in;
+            try {
+                // Extract a raw byte array from the ByteArrayInputStream
+                byte[] buf = (byte[]) bufField.get(b);
+                int pos = (Integer) bufPosField.get(b);
+                int length = (Integer) bufCountField.get(b);
+                return new ArrayBufferInput(buf, pos, length);
+            }
+            catch(Exception e) {
+                // Failed to retrieve the raw byte array
+            }
+        } else if (in instanceof FileInputStream) {
+            return new ChannelBufferInput(((FileInputStream) in).getChannel());
+        }
+
+        return new InputStreamBufferInput(in);
+    }
 
     public InputStreamBufferInput(InputStream in) {
+        this(in, 8192);
+    }
+
+    public InputStreamBufferInput(InputStream in, int bufferSize) {
         this.in = checkNotNull(in, "input is null");
+        this.bufferSize = bufferSize;
     }
 
     @Override
     public MessageBuffer next() throws IOException {
-        // Manage the allocated buffers
-        MessageBuffer m = MessageBuffer.newBuffer(buffer.length);
+        if(reachedEOF)
+            return null;
 
-        // TODO reduce the number of memory copy
+        byte[] buffer = null;
         int cursor = 0;
-        while(cursor < buffer.length) {
-            int readLen = in.read(buffer, cursor, buffer.length - cursor);
+        while(!reachedEOF && cursor < bufferSize) {
+            if(buffer == null)
+                buffer = new byte[bufferSize];
+
+            int readLen = in.read(buffer, cursor, bufferSize - cursor);
             if(readLen == -1) {
+                reachedEOF = true;
                 break;
             }
             cursor += readLen;
         }
-        m.putBytes(0, buffer, 0, cursor);
-        return m;
+
+        return buffer == null ? null : MessageBuffer.wrap(buffer).slice(0, cursor);
     }
 
     @Override
@@ -41,7 +97,7 @@ public void close() throws IOException {
             in.close();
         }
         finally {
-            buffer = null;
+
         }
     }
 }

msgpack-core/src/main/java/org/msgpack/core/buffer/MessageBuffer.java

@@ -114,10 +114,10 @@ public class MessageBuffer {
      * Reference is used to hold a reference to an object that holds the underlying memory so that it cannot be
      * released by the garbage collector.
      */
-    private final ByteBuffer reference;
+    protected final ByteBuffer reference;
 
     // TODO life-time managment of this buffer
-    private AtomicInteger referenceCounter;
+    //private AtomicInteger referenceCounter;
 
 
     static MessageBuffer newOffHeapBuffer(int length) {
@@ -243,7 +243,7 @@ else if(bb.hasArray()) {
         this.reference = null;
     }
 
-    private MessageBuffer(Object base, long address, int length, ByteBuffer reference) {
+    protected MessageBuffer(Object base, long address, int length, ByteBuffer reference) {
         this.base = base;
         this.address = address;
         this.size = length;
@@ -259,7 +259,12 @@ private MessageBuffer(Object base, long address, int length, ByteBuffer referenc
 
     public MessageBuffer slice(int offset, int length) {
         // TODO ensure deleting this slice does not collapse this MessageBuffer
-        return new MessageBuffer(base, address + offset, length, reference);
+        if(offset == 0 && length == size())
+            return this;
+        else {
+            checkArgument(offset + length <= size());
+            return new MessageBuffer(base, address + offset, length, reference);
+        }
     }
 
     public byte getByte(int index) {
@@ -388,6 +393,17 @@ public ByteBuffer toByteBuffer(int index, int length) {
         }
     }
 
+    public ByteBuffer toByteBuffer() {
+        return toByteBuffer(0, size());
+    }
+
+    public byte[] toByteArray() {
+        byte[] b = new byte[size()];
+        unsafe.copyMemory(base, address, b, ARRAY_BYTE_BASE_OFFSET, size());
+        return b;
+    }
+
+
     public void relocate(int offset, int length, int dst) {
         unsafe.copyMemory(base, address + offset, base, address+dst, length);
     }

msgpack-core/src/main/java/org/msgpack/core/buffer/MessageBufferBE.java

@@ -2,6 +2,8 @@
 
 import java.nio.ByteBuffer;
 
+import static org.msgpack.core.Preconditions.checkArgument;
+
 /**
  * MessageBufferBE is a {@link MessageBuffer} implementation tailored to big-endian machines.
  * The specification of Message Pack demands writing short/int/float/long/double values in the big-endian format.
@@ -13,6 +15,20 @@ public class MessageBufferBE extends MessageBuffer {
         super(bb);
     }
 
+    private MessageBufferBE(Object base, long address, int length, ByteBuffer reference) {
+        super(base, address, length, reference);
+    }
+
+    @Override
+    public MessageBufferBE slice(int offset, int length) {
+        if(offset == 0 && length == size())
+            return this;
+        else {
+            checkArgument(offset + length <= size());
+            return new MessageBufferBE(base, address + offset, length, reference);
+        }
+    }
+
     @Override
     public short getShort(int index) {
         return unsafe.getShort(base, address + index);

msgpack-core/src/test/scala/org/msgpack/core/MessagePackSpec.scala

@@ -4,8 +4,17 @@ import org.scalatest._
 import xerial.core.log.{LogLevel, Logger}
 import xerial.core.util.{TimeReport, Timer}
 import scala.language.implicitConversions
-
-trait MessagePackSpec extends WordSpec with Matchers with GivenWhenThen with OptionValues with BeforeAndAfter with Benchmark with Logger {
+import org.scalatest.prop.PropertyChecks
+
+trait MessagePackSpec
+  extends WordSpec
+  with Matchers
+  with GivenWhenThen
+  with OptionValues
+  with BeforeAndAfter
+  with PropertyChecks
+  with Benchmark
+  with Logger {
 
   implicit def toTag(s:String) : Tag = Tag(s)
 

msgpack-core/src/test/scala/org/msgpack/core/MessagePackTest.scala

@@ -11,7 +11,7 @@ import java.nio.CharBuffer
 /**
  * Created on 2014/05/07.
  */
-class MessagePackTest extends MessagePackSpec with PropertyChecks {
+class MessagePackTest extends MessagePackSpec  {
 
   def isValidUTF8(s: String) = {
     MessagePack.UTF8.newEncoder().canEncode(s)