GitHub PullRequestReviewCommentEvent at 2014-05-10 17:23:40
Repository: msgpack-java
Actor: xerial
Payload: {"comment":{"url":"https://api.github.com/repos/msgpack/msgpack-java/pulls/comments/12505228","id":12505228,"diff_hunk":"@@ -1,735 +1,412 @@\n-//\n-// MessagePack for Java\n-//\n-//    Licensed under the Apache License, Version 2.0 (the \"License\");\n-//    you may not use this file except in compliance with the License.\n-//    You may obtain a copy of the License at\n-//\n-//        http://www.apache.org/licenses/LICENSE-2.0\n-//\n-//    Unless required by applicable law or agreed to in writing, software\n-//    distributed under the License is distributed on an \"AS IS\" BASIS,\n-//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n-//    See the License for the specific language governing permissions and\n-//    limitations under the License.\n-//\n package org.msgpack.core;\n \n-import java.util.List;\n-import java.util.LinkedList;\n-import java.util.Iterator;\n-import java.math.BigInteger;\n-import java.io.IOException;\n-import java.io.EOFException;\n+import sun.misc.Unsafe;\n+import sun.nio.ch.DirectBuffer;\n+\n+import java.lang.reflect.Constructor;\n+import java.lang.reflect.Field;\n+import java.nio.BufferOverflowException;\n import java.nio.ByteBuffer;\n-import java.nio.channels.ReadableByteChannel;\n-import java.nio.channels.WritableByteChannel;\n+import java.util.concurrent.atomic.AtomicInteger;\n+\n+import static sun.misc.Unsafe.ARRAY_BYTE_BASE_OFFSET;\n+import static sun.misc.Unsafe.ARRAY_BYTE_INDEX_SCALE;\n \n /**\n+ * MessageBuffer class is an abstraction of memory for reading/writing message packed data.\n+ * This MessageBuffers ensures short/int/float/long/double values are written in the big-endian order.\n  *\n- *   <table>\n- *     <tr>\n- *       <th>Category</th><th>Producer</th><th>Consumer</th>\n- *     </tr>\n- *\n- *     <tr>\n- *       <th rowspan=\"1\">At-most bulk IO</th>\n- *       <td>write(src)</td><td>read(dst)</td>\n- *     </tr>\n- *\n- *     <tr>\n- *       <th rowspan=\"2\">All bulk IO</th>\n- *       <td>writeAll(src)</td><td>readAll(dst)</td>\n- *     </tr>\n- *     <tr><td>writeAll(b, off, len)</td><td>readAll(size)</td></tr>\n- *\n- *     <tr>\n- *       <th rowspan=\"1\">Direct transfer bulk IO</th>\n- *       <td>transferFrom(size, channel)</td><td>transferTo(size, channel)</td>\n- *     </tr>\n- *     <td></td><td>transferAllTo(size, channel)</td>\n- *\n- *     <tr>\n- *       <th rowspan=\"6\">Composite primitive type IO</th>\n- *       <td>writeByte(v)</td><td>readByte()</td>\n- *     </tr>\n- *     <tr><td>writeShort(v)</td><td>readShort()</td></tr>\n- *     <tr><td>writeInt(v)</td><td>readInt()</td></tr>\n- *     <tr><td>writeLong(v)</td><td>readLong()</td></tr>\n- *     <tr><td>writeFloat(v)</td><td>readFloat()</td></tr>\n- *     <tr><td>writeDouble(v)</td><td>readDouble()</td></tr>\n- *\n- *     <tr>\n- *       <th rowspan=\"6\">Primitive type IO</th>\n- *       <td>writeByteAndByte(v)</td><td>\n- *     </tr>\n- *     <tr><td>writeByteAndShort(v)</td><td></tr>\n- *     <tr><td>writeByteAndInt(v)</td><td></tr>\n- *     <tr><td>writeByteAndLong(v)</td><td></tr>\n- *     <tr><td>writeByteAndFloat(v)</td><td></tr>\n- *     <tr><td>writeByteAndDouble(v)</td><td></tr>\n- *\n- *     <tr>\n- *       <th rowspan=\"2\">Zero-copy IO</th>\n- *       <td rowspan=\"2\">add(bb)</td><td>get()</td>\n- *     </tr>\n- *     <tr><td>remove()</td></tr>\n- *   </table>\n+ * This class is optimized for fast memory access, so many methods are\n+ * implemented without using any interface method that produces invokeinterface call in JVM.\n+ * Compared to invokevirtual, invokeinterface is 30% slower in general because it needs to find a target function from the table.\n  *\n  */\n-public class MessageBuffer implements MessagePackerChannel, MessageUnpackerChannel {\n-    public static class Builder {\n-        private ReadableByteChannel inputChannel = null;\n-        private WritableByteChannel outputChannel = null;\n-        private int writeBufferChunkSize = 1024;\n-        private int readReferenceThreshold = 1024;\n-\n-        public MessageBuffer build() {\n-            return new MessageBuffer(this);\n-        }\n-\n-        public ReadableByteChannel getInputChannel() {\n-            return inputChannel;\n-        }\n-\n-        public void setInputChannel(ReadableByteChannel channel) {\n-            this.inputChannel = channel;\n-        }\n-\n-        public Builder withInputChannel(ReadableByteChannel channel) {\n-            setInputChannel(channel);\n-            return this;\n-        }\n-\n-        public WritableByteChannel getOutputChannel() {\n-            return outputChannel;\n-        }\n-\n-        public void setOutputChannel(WritableByteChannel channel) {\n-            this.outputChannel = channel;\n-        }\n-\n-        public Builder withOutputChannel(WritableByteChannel channel) {\n-            setOutputChannel(channel);\n-            return this;\n-        }\n-\n-        public int getReadReferenceThreshold() {\n-            return readReferenceThreshold;\n-        }\n+public class MessageBuffer {\n+\n+    static final Unsafe unsafe;\n+    static final Constructor byteBufferConstructor;\n+\n+    static {\n+        try {\n+            // Fetch theUnsafe object for Orackle JDK and OpenJDK\n+            Field field = Unsafe.class.getDeclaredField(\"theUnsafe\");\n+            field.setAccessible(true);\n+            unsafe = (Unsafe) field.get(null);\n+            if (unsafe == null) {\n+                throw new RuntimeException(\"Unsafe is unavailable\");\n+            }\n+            // TODO Finding Unsafe instance for Android JVM\n \n-        public void setReadReferenceThreshold(int size) {\n-            this.readReferenceThreshold = size;\n-        }\n+            // Make sure the VM thinks bytes are only one byte wide\n+            if (sun.misc.Unsafe.ARRAY_BYTE_INDEX_SCALE != 1) {\n+                throw new IllegalStateException(\"Byte array index scale must be 1, but is \" + ARRAY_BYTE_INDEX_SCALE);\n+            }\n \n-        public Builder withReadReferenceThreshold(int size) {\n-            setReadReferenceThreshold(size);\n-            return this;\n-        }\n+            // Find the hidden constructor for DirectByteBuffer\n+            Class<?> directByteBufferClass = ClassLoader.getSystemClassLoader().loadClass(\"java.nio.DirectByteBuffer\");\n+            byteBufferConstructor = directByteBufferClass.getDeclaredConstructor(long.class, int.class, Object.class);\n+            byteBufferConstructor.setAccessible(true);\n \n-        public int getWriteBufferChunkSize() {\n-            return writeBufferChunkSize;\n-        }\n+            // Check the endian of this CPU\n+            boolean isLittleEndian = true;\n+            long a = unsafe.allocateMemory(8);\n+            try {\n+                unsafe.putLong(a, 0x0102030405060708L);\n+                byte b = unsafe.getByte(a);\n+                switch (b) {\n+                    case 0x01:\n+                        isLittleEndian = false;\n+                        break;\n+                    case 0x08:\n+                        isLittleEndian = true;\n+                        break;\n+                    default:\n+                        assert false;\n+                }\n+            } finally {\n+                unsafe.freeMemory(a);\n+            }\n \n-        public void setWriteBufferChunkSize(int size) {\n-            this.writeBufferChunkSize = size;\n+            String bufferClsName = isLittleEndian ? \"org.msgpack.core.MessageBuffer\" : \"org.msgpack.core.MessageBufferBE\";\n+            msgBufferClass = Class.forName(bufferClsName);\n         }\n-\n-        public Builder withWriteBufferChunkSize(int size) {\n-            setWriteBufferChunkSize(size);\n-            return this;\n+        catch (Exception e) {\n+            throw new RuntimeException(e);\n         }\n     }\n \n-    private static final class Chunk {\n-        ByteBuffer bb;\n-        boolean reusable;\n \n-        Chunk(ByteBuffer bb, boolean reusable) {\n-            this.bb = bb;\n-            this.reusable = reusable;\n-        }\n-    }\n+    /**\n+     * MessageBuffer class to use. If this machine is big-endian, it uses MessageBufferBE, which overrides some methods in this class that translate endians. If not, uses MessageBuffer.\n+     */\n+    private final static Class<?> msgBufferClass;\n \n-    private final LinkedList<Chunk> list = new LinkedList<Chunk>();\n-    private ByteBuffer lastWritableBuffer = null;\n-    private int lastWritableBufferReadPosition = 0;\n+    /**\n+     * Base object for resolving the relative address of the raw byte array.\n+     * If base == null, the address value is a raw memory address\n+     */\n+    protected final Object base;\n \n-    private final byte[] castBuffer = new byte[8];\n-    private final ByteBuffer castByteBuffer = ByteBuffer.wrap(castBuffer);\n+    /**\n+     * Head address of the underlying memory. If base is null, the address is a direct memory address, and if not,\n+     * it is the relative address within an array object (base)\n+     */\n+    protected final long address;\n \n-    protected ReadableByteChannel inputChannel;\n-    protected WritableByteChannel outputChannel;\n+    /**\n+     * Size of the underlying memory\n+     */\n+    protected final int size;\n \n-    private int writeBufferChunkSize;\n-    private int readReferenceThreshold;\n+    /**\n+     * The limit index of the end of the data contained in this buffer\n+     */\n+    private int limit;\n \n-    // TODO configurable buffer allocator\n+    /**\n+     * Reference is used to hold a reference to an object that holds the underlying memory so that it cannot be\n+     * released by the garbage collector.\n+     */\n+    private final ByteBuffer reference;\n \n-    public MessageBuffer() {\n-        this(new Builder());\n-    }\n \n-    MessageBuffer(Builder builder) {\n-        this.inputChannel = builder.getInputChannel();\n-        this.outputChannel = builder.getOutputChannel();\n-        this.writeBufferChunkSize = builder.getWriteBufferChunkSize();\n-        this.readReferenceThreshold = builder.getReadReferenceThreshold();\n-    }\n+    private AtomicInteger referenceCounter;\n \n-    public void close() throws IOException {\n-        if (inputChannel != null) {\n-            inputChannel.close();\n-        }\n-        if (outputChannel != null) {\n-            if (outputChannel.isOpen()) {\n-                flush();\n-            }\n-            outputChannel.close();\n-        }\n-    }\n-\n-    public boolean isOpen() {\n-        if (inputChannel != null && !inputChannel.isOpen()) {\n-            return false;\n-        }\n-        if (outputChannel != null && !outputChannel.isOpen()) {\n-            return false;\n-        }\n-        return true;\n-    }\n-\n-    private static int transferByteBuffer(ByteBuffer src, ByteBuffer dst) {\n-        int pos = dst.position();\n-\n-        int srcrem = src.remaining();\n-        int dstrem = dst.remaining();\n-        if (dstrem < srcrem) {\n-            int lim = src.limit();\n-            try {\n-                src.limit(src.position() + dstrem);\n-                dst.put(src);\n-            } finally {\n-                src.limit(lim);\n-            }\n-        } else {\n-            dst.put(src);\n-        }\n \n-        return dst.position() - pos;\n+    static MessageBuffer newOffHeapBuffer(int length) {\n+        long address = unsafe.allocateMemory(length);\n+        return new MessageBuffer(address, length);\n     }\n \n-    private void consumedFirstChunk() {\n-        Chunk c = list.removeFirst();\n-        if (c.reusable) {\n-            if (lastWritableBuffer == null) {\n-                lastWritableBuffer = c.bb;\n-                lastWritableBuffer.position(0);\n-                lastWritableBuffer.limit(lastWritableBuffer.capacity());\n-                lastWritableBufferReadPosition = 0;\n-            }\n-        }\n+    public static MessageBuffer newDirectBuffer(int length) {\n+        ByteBuffer m = ByteBuffer.allocateDirect(length);\n+        return newMessageBuffer(m);\n     }\n \n-    public void add(ByteBuffer bb) throws IOException {\n-        list.addLast(new Chunk(bb, true));\n+    public static MessageBuffer newBuffer(int length) {\n+        ByteBuffer m = ByteBuffer.allocate(length);\n+        return newMessageBuffer(m);\n     }\n \n-    public ByteBuffer get() throws IOException {\n-        if (list.isEmpty() && !tryReadMore()) {\n-            return null;\n-        }\n-        return list.getFirst().bb;\n+    public static MessageBuffer wrap(byte[] array) {\n+        return newMessageBuffer(array);\n     }\n \n-    public ByteBuffer remove() throws IOException {\n-        if (list.isEmpty() && !tryReadMore()) {\n-            return null;\n-        }\n-        Chunk c = list.getFirst();\n-        c.reusable = false;\n-        consumedFirstChunk();\n-        return c.bb;\n+    public static MessageBuffer wrap(ByteBuffer bb) {\n+        return newMessageBuffer(bb);\n     }\n \n-    private void producedLastChunk() throws IOException {\n-        lastWritableBuffer.limit(lastWritableBuffer.position());\n-        lastWritableBuffer.position(lastWritableBufferReadPosition);\n-        list.addLast(new Chunk(lastWritableBuffer, true));\n-        lastWritableBuffer = null;\n-        tryFlush();\n+    /**\n+     * Creates a new MessageBuffer instance bakeed by ByteBuffer","path":"msgpack-core/src/main/java/org/msgpack/core/MessageBuffer.java","position":364,"original_position":364,"commit_id":"fa1a9ebc36382d7911b91dcb6e086891697e39b1","original_commit_id":"fa1a9ebc36382d7911b91dcb6e086891697e39b1","user":{"login":"xerial","id":57538,"avatar_url":"https://avatars.githubusercontent.com/u/57538?","gravatar_id":"04657be0eb87c725c2a2468f13322932","url":"https://api.github.com/users/xerial","html_url":"https://github.com/xerial","followers_url":"https://api.github.com/users/xerial/followers","following_url":"https://api.github.com/users/xerial/following{/other_user}","gists_url":"https://api.github.com/users/xerial/gists{/gist_id}","starred_url":"https://api.github.com/users/xerial/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/xerial/subscriptions","organizations_url":"https://api.github.com/users/xerial/orgs","repos_url":"https://api.github.com/users/xerial/repos","events_url":"https://api.github.com/users/xerial/events{/privacy}","received_events_url":"https://api.github.com/users/xerial/received_events","type":"User","site_admin":false},"body":"thx ","created_at":"2014-05-10T17:23:40Z","updated_at":"2014-05-10T17:23:40Z","html_url":"https://github.com/msgpack/msgpack-java/pull/82#discussion_r12505228","pull_request_url":"https://api.github.com/repos/msgpack/msgpack-java/pulls/82","_links":{"self":{"href":"https://api.github.com/repos/msgpack/msgpack-java/pulls/comments/12505228"},"html":{"href":"https://github.com/msgpack/msgpack-java/pull/82#discussion_r12505228"},"pull_request":{"href":"https://api.github.com/repos/msgpack/msgpack-java/pulls/82"}}}}