RPCProtoMessage. Use that in the server

bin/alluxio

@@ -310,6 +310,10 @@ function main {
     for src_file in $(find ${BIN}/../core/server/src/proto -name '*.proto' -type f); do
       protoc --java_out=${BIN}/../core/server/src/main/java --proto_path=`dirname ${src_file}` ${src_file}
     done
+    rm -rf ${BIN}/../core/common/src/main/java/alluxio/proto
+    for src_file in $(find ${BIN}/../core/common/src/proto -name '*.proto' -type f); do
+      protoc --java_out=${BIN}/../core/common/src/main/java --proto_path=`dirname ${src_file}` ${src_file}
+    done
   ;;
   "clearCache")
     sync; echo 3 > /proc/sys/vm/drop_caches ;

build/findbugs/findbugs-exclude.xml

@@ -5,6 +5,7 @@
       <Package name="org.apache.thrift"/>
       <Package name="alluxio.thrift"/>
       <Package name="alluxio.proto.journal.*"/>
+      <Package name="alluxio.proto.dataserver.*"/>
     </Or>
   </Match>
 

core/client/src/main/java/alluxio/client/block/stream/PacketInStream.java

@@ -22,7 +22,6 @@
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.nio.ByteBuffer;
 
 import javax.annotation.concurrent.NotThreadSafe;
 
@@ -183,13 +182,6 @@ private void closePacketReader() {
    * @param packet the packet
    */
   private void destroyPacket(ByteBuf packet) {
-    // TODO(peis): Investigate whether we can get rid of this.
-    if (packet.nioBufferCount() > 0) {
-      ByteBuffer buffer = packet.nioBuffer();
-      if (buffer.isDirect()) {
-        BufferUtils.cleanDirectBuffer(buffer);
-      }
-    }
     ReferenceCountUtil.release(packet);
   }
 

core/common/src/main/java/alluxio/network/protocol/RPCMessage.java

@@ -41,6 +41,12 @@ public enum Type implements EncodedMessage {
     RPC_FILE_READ_RESPONSE(6),
     RPC_FILE_WRITE_REQUEST(7),
     RPC_FILE_WRITE_RESPONSE(8),
+
+    RPC_READ_REQUEST(9),
+    RPC_WRITE_REQUEST(10),
+    RPC_RESPONSE(11),
+
+    RPC_UNKNOWN(1000),
     ;
 
     private final int mId;

core/common/src/main/java/alluxio/network/protocol/RPCMessageEncoder.java

@@ -73,6 +73,5 @@ protected void encode(ChannelHandlerContext ctx, RPCMessage in, List<Object> out
           "The payload must be a ByteBuf or a FileRegion.");
       out.add(output);
     }
-
   }
 }

core/server/src/main/java/alluxio/worker/netty/DataServerBlockReadHandler.java

@@ -12,29 +12,32 @@
 package alluxio.worker.netty;
 
 import alluxio.Constants;
-import alluxio.network.protocol.RPCBlockReadRequest;
+import alluxio.network.protocol.RPCProtoMessage;
 import alluxio.network.protocol.databuffer.DataBuffer;
-import alluxio.network.protocol.databuffer.DataByteBuffer;
 import alluxio.network.protocol.databuffer.DataFileChannel;
+import alluxio.network.protocol.databuffer.DataNettyBuffer;
+import alluxio.proto.dataserver.Protocol;
 import alluxio.worker.block.BlockWorker;
 import alluxio.worker.block.io.BlockReader;
 
 import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
 import java.util.concurrent.ExecutorService;
 
 import javax.annotation.concurrent.NotThreadSafe;
 
 /**
- * This class handles {@link RPCBlockReadRequest}s.
+ * This handler handles block read request. Check more information in {@link DataServerReadHandler}.
  */
 @NotThreadSafe
-final public class DataServerBlockReadHandler extends DataServerReadHandler {
+public final class DataServerBlockReadHandler extends DataServerReadHandler {
   private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
 
   /** The Block Worker which handles blocks stored in the Alluxio storage of the worker. */
@@ -54,10 +57,10 @@ private final class BlockReadRequestInternal extends ReadRequestInternal {
      * @param request the block read request
      * @throws Exception if it fails to create the object
      */
-    public BlockReadRequestInternal(RPCBlockReadRequest request) throws Exception {
+    public BlockReadRequestInternal(Protocol.ReadRequest request) throws Exception {
       mBlockReader = mWorker
-          .readBlockRemote(request.getSessionId(), request.getBlockId(), request.getLockId());
-      mId = request.getBlockId();
+          .readBlockRemote(request.getSessionId(), request.getId(), request.getLockId());
+      mId = request.getId();
       mWorker.accessBlock(request.getSessionId(), mId);
 
       mStart = request.getOffset();
@@ -91,21 +94,38 @@ public DataServerBlockReadHandler(ExecutorService executorService, BlockWorker b
   }
 
   @Override
-  protected void initializeRequest(RPCBlockReadRequest request) throws Exception {
+  protected boolean acceptMessage(Object object) {
+    if (!super.acceptMessage(object)) {
+      return false;
+    }
+    Protocol.ReadRequest request = (Protocol.ReadRequest) ((RPCProtoMessage) object).getMessage();
+    return request.getType() == Protocol.RequestType.ALLUXIO_BLOCK;
+  }
+
+  @Override
+  protected void initializeRequest(Protocol.ReadRequest request) throws Exception {
     mRequest = new BlockReadRequestInternal(request);
   }
 
   @Override
-  protected DataBuffer getDataBuffer(long offset, int len) throws IOException {
+  protected DataBuffer getDataBuffer(Channel channel, long offset, int len) throws IOException {
     BlockReader blockReader = ((BlockReadRequestInternal) mRequest).mBlockReader;
+    Preconditions.checkArgument(blockReader.getChannel() instanceof FileChannel,
+        "Only FileChannel is supported!");
     switch (mTransferType) {
       case MAPPED:
-        ByteBuffer data = blockReader.read(offset, len);
-        return new DataByteBuffer(data, len);
+        ByteBuf buf = channel.alloc().buffer(len, len);
+        try {
+          while (buf.writableBytes() > 0
+              && buf.writeBytes((FileChannel) blockReader.getChannel(), buf.writableBytes()) != -1)
+            ;
+        } catch (Throwable e) {
+          ReferenceCountUtil.release(buf);
+          throw e;
+        }
+        return new DataNettyBuffer(buf, buf.readableBytes());
       case TRANSFER: // intend to fall through as TRANSFER is the default type.
       default:
-        Preconditions.checkArgument(blockReader.getChannel() instanceof FileChannel,
-            "Only FileChannel is supported!");
         return new DataFileChannel((FileChannel) blockReader.getChannel(), offset, len);
     }
   }

core/server/src/main/java/alluxio/worker/netty/DataServerBlockWriteHandler.java

@@ -13,7 +13,8 @@
 
 import alluxio.StorageTierAssoc;
 import alluxio.WorkerStorageTierAssoc;
-import alluxio.network.protocol.RPCBlockWriteRequest;
+import alluxio.network.protocol.RPCProtoMessage;
+import alluxio.proto.dataserver.Protocol;
 import alluxio.worker.block.BlockWorker;
 import alluxio.worker.block.io.BlockWriter;
 
@@ -27,20 +28,11 @@
 import javax.annotation.concurrent.NotThreadSafe;
 
 /**
- * This class handles {@link RPCBlockWriteRequest}s.
- *
- * Protocol: Check {@link alluxio.client.block.stream.NettyBlockWriter} for more information.
- * 1. The netty channel handler streams packets from the channel and buffers them. The netty
- *    reader is paused if the buffer is full by turning off the auto read, and is resumed when
- *    the buffer is not full.
- * 2. The {@link PacketWriter} polls packets from the buffer and writes to the block worker. The
- *    writer becomes inactive if there is nothing on the buffer to free up the executor. It is
- *    resumed when the buffer becomes non-empty.
- * 3. When an error occurs, the channel is closed. All the buffered packets are released when the
- *    channel is deregistered.
+ * This handler handles block write request. Check more information in
+ * {@link DataServerWriteHandler}.
  */
 @NotThreadSafe
-public abstract class DataServerBlockWriteHandler extends DataServerWriteHandler {
+public final class DataServerBlockWriteHandler extends DataServerWriteHandler {
   /** The Block Worker which handles blocks stored in the Alluxio storage of the worker. */
   private final BlockWorker mWorker;
   /** An object storing the mapping of tier aliases to ordinals. */
@@ -49,10 +41,10 @@ public abstract class DataServerBlockWriteHandler extends DataServerWriteHandler
   private class BlockWriteRequestInternal extends WriteRequestInternal {
     public BlockWriter mBlockWriter;
 
-    public BlockWriteRequestInternal(RPCBlockWriteRequest request) throws Exception {
-      mBlockWriter = mWorker.getTempBlockWriterRemote(request.getSessionId(), request.getBlockId());
+    public BlockWriteRequestInternal(Protocol.WriteRequest request) throws Exception {
+      mBlockWriter = mWorker.getTempBlockWriterRemote(request.getSessionId(), request.getId());
       mSessionId = request.getSessionId();
-      mId = request.getBlockId();
+      mId = request.getId();
     }
 
     @Override
@@ -77,10 +69,11 @@ public DataServerBlockWriteHandler(ExecutorService executorService, BlockWorker
    * @param msg the block write request
    * @throws Exception if it fails to initialize
    */
-  protected void initializeRequest(RPCBlockWriteRequest msg) throws Exception {
+  protected void initializeRequest(RPCProtoMessage msg) throws Exception {
     super.initializeRequest(msg);
     if (mRequest == null) {
-      mRequest = new BlockWriteRequestInternal(msg);
+      Protocol.WriteRequest request = (Protocol.WriteRequest) (msg.getMessage());
+      mRequest = new BlockWriteRequestInternal(request);
     }
   }
 

core/server/src/main/java/alluxio/worker/netty/DataServerFileReadHandler.java

@@ -12,12 +12,15 @@
 package alluxio.worker.netty;
 
 import alluxio.Constants;
-import alluxio.network.protocol.RPCBlockReadRequest;
+import alluxio.network.protocol.RPCProtoMessage;
 import alluxio.network.protocol.databuffer.DataBuffer;
 import alluxio.network.protocol.databuffer.DataNettyBuffer;
+import alluxio.proto.dataserver.Protocol;
 import alluxio.worker.file.FileSystemWorker;
 
-import io.netty.buffer.Unpooled;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.util.ReferenceCountUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,10 +31,10 @@
 import javax.annotation.concurrent.NotThreadSafe;
 
 /**
- * This class handles {@link RPCBlockReadRequest}s.
+ * This handler handles file read request. Check more information in {@link DataServerReadHandler}.
  */
 @NotThreadSafe
-final public class DataServerFileReadHandler extends DataServerReadHandler {
+public final class DataServerFileReadHandler extends DataServerReadHandler {
   private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
 
   /** The Block Worker which handles blocks stored in the Alluxio storage of the worker. */
@@ -50,8 +53,8 @@ private final class FileReadRequestInternal extends ReadRequestInternal {
      * @param request the block read request
      * @throws Exception if it fails to create the object
      */
-    public FileReadRequestInternal(RPCBlockReadRequest request) throws Exception {
-      mInputStream = mWorker.getUfsInputStream(request.getBlockId(), request.getOffset());
+    public FileReadRequestInternal(Protocol.ReadRequest request) throws Exception {
+      mInputStream = mWorker.getUfsInputStream(request.getId(), request.getOffset());
 
       mStart = request.getOffset();
       mEnd = mStart + request.getLength();
@@ -73,26 +76,37 @@ public DataServerFileReadHandler(ExecutorService executorService, FileSystemWork
   }
 
   @Override
-  protected void initializeRequest(RPCBlockReadRequest request) throws Exception {
+  protected boolean acceptMessage(Object object) {
+    if (!super.acceptMessage(object)) {
+      return false;
+    }
+    Protocol.ReadRequest request = (Protocol.ReadRequest) ((RPCProtoMessage) object).getMessage();
+    return request.getType() == Protocol.RequestType.UFS_FILE;
+  }
+
+  @Override
+  protected void initializeRequest(Protocol.ReadRequest request) throws Exception {
     mRequest = new FileReadRequestInternal(request);
   }
 
   @Override
-  protected DataBuffer getDataBuffer(long offset, int len) throws IOException {
-    byte[] data = new byte[len];
+  protected DataBuffer getDataBuffer(Channel channel, long offset, int len) throws IOException {
+    ByteBuf buf = channel.alloc().buffer(len, len);
     InputStream in = ((FileReadRequestInternal) mRequest).mInputStream;
 
-    int bytesRead = 0;
-    if (in != null) { // if we have not reached the end of the file
-      while (bytesRead < len) {
-        int read = in.read(data, bytesRead, len - bytesRead);
-        if (read == -1) {
-          break;
-        }
-        bytesRead += read;
+    try {
+      if (in != null) { // if we have not reached the end of the file
+        while (buf.writableBytes() > 0 && buf.writeBytes(in, buf.writableBytes()) != -1);
       }
+    } catch (Throwable e) {
+      ReferenceCountUtil.release(buf);
+      throw e;
+    }
+
+    if (buf.readableBytes() == 0) {
+      ReferenceCountUtil.release(buf);
+      return null;
     }
-    return bytesRead != 0 ?
-        new DataNettyBuffer(Unpooled.wrappedBuffer(data, 0, bytesRead), bytesRead) : null;
+    return new DataNettyBuffer(buf, buf.readableBytes());
   }
 }

core/server/src/main/java/alluxio/worker/netty/DataServerFileWriteHandler.java

@@ -11,7 +11,8 @@
 
 package alluxio.worker.netty;
 
-import alluxio.network.protocol.RPCBlockWriteRequest;
+import alluxio.network.protocol.RPCProtoMessage;
+import alluxio.proto.dataserver.Protocol;
 import alluxio.worker.file.FileSystemWorker;
 
 import io.netty.buffer.ByteBuf;
@@ -24,29 +25,20 @@
 import javax.annotation.concurrent.NotThreadSafe;
 
 /**
- * This class handles {@link RPCBlockWriteRequest}s.
- *
- * Protocol: Check {@link alluxio.client.block.stream.NettyBlockWriter} for more information.
- * 1. The netty channel handler streams packets from the channel and buffers them. The netty
- *    reader is paused if the buffer is full by turning off the auto read, and is resumed when
- *    the buffer is not full.
- * 2. The {@link PacketWriter} polls packets from the buffer and writes to the block worker. The
- *    writer becomes inactive if there is nothing on the buffer to free up the executor. It is
- *    resumed when the buffer becomes non-empty.
- * 3. When an error occurs, the channel is closed. All the buffered packets are released when the
- *    channel is deregistered.
+ * This handler handles file write request. Check more information in
+ * {@link DataServerWriteHandler}.
  */
 @NotThreadSafe
-public abstract class DataServerFileWriteHandler extends DataServerWriteHandler {
+public final class DataServerFileWriteHandler extends DataServerWriteHandler {
   /** Filesystem worker which handles file level operations for the worker. */
   private final FileSystemWorker mWorker;
 
   private class FileWriteRequestInternal extends WriteRequestInternal {
     public OutputStream mOutputStream;
 
-    public FileWriteRequestInternal(RPCBlockWriteRequest request) throws Exception {
-      mOutputStream = mWorker.getUfsOutputStream(request.getBlockId());
-      mId = request.getBlockId();
+    public FileWriteRequestInternal(Protocol.WriteRequest request) throws Exception {
+      mOutputStream = mWorker.getUfsOutputStream(request.getId());
+      mId = request.getId();
     }
 
     @Override
@@ -69,10 +61,11 @@ public DataServerFileWriteHandler(ExecutorService executorService, FileSystemWor
    * @param msg the block write request
    * @throws Exception if it fails to initialize
    */
-  protected void initializeRequest(RPCBlockWriteRequest msg) throws Exception {
+  @Override
+  protected void initializeRequest(RPCProtoMessage msg) throws Exception {
     super.initializeRequest(msg);
     if (mRequest == null) {
-      mRequest = new FileWriteRequestInternal(msg);
+      mRequest = new FileWriteRequestInternal((Protocol.WriteRequest) msg.getMessage());
     }
   }