();
+ for (final NetworkType type : NetworkType.values()) {
+ list.add(new Object[] { type });
+ }
+ return list;
+ }
+
@Test
public void readPartialTest1() throws InvalidPathException, FileAlreadyExistException,
IOException {
int fileId = TestUtils.createByteFile(mTFS, "/testFile", WriteType.MUST_CACHE, 10);
- long blockId = mTFS.getBlockId(fileId, 0);
- DataServerMessage sendMsg;
- sendMsg = DataServerMessage.createBlockRequestMessage(blockId, 0, 6);
- NetAddress firstBlock = mTFS.getFileBlocks(fileId).get(0).getLocations().get(0);
- SocketChannel socketChannel =
- SocketChannel.open(new InetSocketAddress(firstBlock.mHost, firstBlock.mSecondaryPort));
- while (!sendMsg.finishSending()) {
- sendMsg.send(socketChannel);
- }
- DataServerMessage recvMsg = DataServerMessage.createBlockResponseMessage(false, blockId, 0, 6);
- while (!recvMsg.isMessageReady()) {
- int numRead = recvMsg.recv(socketChannel);
- if (numRead == -1) {
- break;
- }
- }
- socketChannel.close();
- Assert.assertEquals(TestUtils.getIncreasingByteBuffer(6), recvMsg.getReadOnlyData());
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ final int offset = 0;
+ final int length = 6;
+ DataServerMessage recvMsg = request(block, offset, length);
+ assertValid(recvMsg, length, block.getBlockId(), offset, length);
}
@Test
public void readPartialTest2() throws InvalidPathException, FileAlreadyExistException,
IOException {
int fileId = TestUtils.createByteFile(mTFS, "/testFile", WriteType.MUST_CACHE, 10);
- long blockId = mTFS.getBlockId(fileId, 0);
- DataServerMessage sendMsg;
- sendMsg = DataServerMessage.createBlockRequestMessage(blockId, 2, 6);
- SocketChannel socketChannel =
- SocketChannel
- .open(new InetSocketAddress(
- mTFS.getFileBlocks(fileId).get(0).getLocations().get(0).mHost, mTFS
- .getFileBlocks(fileId).get(0).getLocations().get(0).mSecondaryPort));
- while (!sendMsg.finishSending()) {
- sendMsg.send(socketChannel);
- }
- DataServerMessage recvMsg = DataServerMessage.createBlockResponseMessage(false, blockId, 2, 6);
- while (!recvMsg.isMessageReady()) {
- int numRead = recvMsg.recv(socketChannel);
- if (numRead == -1) {
- break;
- }
- }
- socketChannel.close();
- Assert.assertEquals(TestUtils.getIncreasingByteBuffer(2, 6), recvMsg.getReadOnlyData());
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ final int offset = 2;
+ final int length = 6;
+ DataServerMessage recvMsg = request(block, offset, length);
+ assertValid(recvMsg, TestUtils.getIncreasingByteBuffer(offset, length), block.getBlockId(),
+ offset, length);
}
@Test
public void readTest() throws InvalidPathException, FileAlreadyExistException, IOException {
- int fileId = TestUtils.createByteFile(mTFS, "/testFile", WriteType.MUST_CACHE, 10);
- long blockId = mTFS.getBlockId(fileId, 0);
- DataServerMessage sendMsg = DataServerMessage.createBlockRequestMessage(blockId);
- SocketChannel socketChannel =
- SocketChannel
- .open(new InetSocketAddress(
- mTFS.getFileBlocks(fileId).get(0).getLocations().get(0).mHost, mTFS
- .getFileBlocks(fileId).get(0).getLocations().get(0).mSecondaryPort));
- while (!sendMsg.finishSending()) {
- sendMsg.send(socketChannel);
+ final int length = 10;
+ int fileId = TestUtils.createByteFile(mTFS, "/testFile", WriteType.MUST_CACHE, length);
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ DataServerMessage recvMsg = request(block);
+ assertValid(recvMsg, length, block.getBlockId(), 0, length);
+ }
+
+ @Test
+ public void multiReadTest() throws IOException {
+ final int length = 20;
+ int fileId = TestUtils.createByteFile(mTFS, "/multiReadTest", WriteType.MUST_CACHE, length);
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ for (int i = 0; i < 10; i ++) {
+ DataServerMessage recvMsg = request(block);
+ assertValid(recvMsg, length, block.getBlockId(), 0, length);
}
- DataServerMessage recvMsg = DataServerMessage.createBlockResponseMessage(false, blockId);
- while (!recvMsg.isMessageReady()) {
- int numRead = recvMsg.recv(socketChannel);
- if (numRead == -1) {
- break;
+ }
+
+ @Test
+ public void readTooLarge() throws IOException {
+ final int length = 20;
+ int fileId = TestUtils.createByteFile(mTFS, "/readTooLarge", WriteType.MUST_CACHE, length);
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ DataServerMessage recvMsg = request(block, 0, length * 2);
+ assertError(recvMsg, block.blockId);
+ }
+
+ @Test
+ public void lengthTooSmall() throws IOException {
+ final int length = 20;
+ int fileId = TestUtils.createByteFile(mTFS, "/readTooLarge", WriteType.MUST_CACHE, length);
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ DataServerMessage recvMsg = request(block, 0, length * -2);
+ assertError(recvMsg, block.blockId);
+ }
+
+ @Test
+ public void tooLargeOffset() throws IOException {
+ final int length = 10;
+ int fileId = TestUtils.createByteFile(mTFS, "/readTooLarge", WriteType.MUST_CACHE, length);
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ DataServerMessage recvMsg = request(block, length * 2, 1);
+ assertError(recvMsg, block.blockId);
+ }
+
+ @Test
+ public void negativeOffset() throws IOException {
+ final int length = 10;
+ int fileId = TestUtils.createByteFile(mTFS, "/readTooLarge", WriteType.MUST_CACHE, length);
+ ClientBlockInfo block = mTFS.getFileBlocks(fileId).get(0);
+ DataServerMessage recvMsg = request(block, length * -2, 1);
+ assertError(recvMsg, block.blockId);
+ }
+
+ /**
+ * Asserts that the message back matches the block response protocols for the error case.
+ */
+ private void assertError(final DataServerMessage msg, final long blockId) {
+ assertValid(msg, 0, -1 * blockId, 0, 0);
+ }
+
+ /**
+ * Asserts that the message back matches the block response protocols.
+ */
+ private void assertValid(final DataServerMessage msg, final int expectedSize,
+ final long blockId, final long offset, final long length) {
+ assertValid(msg, TestUtils.getIncreasingByteBuffer(expectedSize), blockId, offset, length);
+ }
+
+ /**
+ * Asserts that the message back matches the block response protocols.
+ */
+ private void assertValid(final DataServerMessage msg, final ByteBuffer expectedData,
+ final long blockId, final long offset, final long length) {
+ Assert.assertEquals(expectedData, msg.getReadOnlyData());
+ Assert.assertEquals(blockId, msg.getBlockId());
+ Assert.assertEquals(offset, msg.getOffset());
+ Assert.assertEquals(length, msg.getLength());
+ }
+
+ /**
+ * Requests a block from the server. This call will read the full block.
+ */
+ private DataServerMessage request(final ClientBlockInfo block) throws IOException {
+ return request(block, 0, -1);
+ }
+
+ /**
+ * Create a new socket to the data port and send a block request. The returned value is
+ * the response from the server.
+ */
+ private DataServerMessage request(final ClientBlockInfo block, final long offset,
+ final long length) throws IOException {
+ DataServerMessage sendMsg =
+ DataServerMessage.createBlockRequestMessage(block.blockId, offset, length);
+ SocketChannel socketChannel =
+ SocketChannel.open(new InetSocketAddress(block.getLocations().get(0).mHost, block
+ .getLocations().get(0).mSecondaryPort));
+ try {
+ while (!sendMsg.finishSending()) {
+ sendMsg.send(socketChannel);
+ }
+ DataServerMessage recvMsg =
+ DataServerMessage.createBlockResponseMessage(false, block.blockId, offset, length);
+ while (!recvMsg.isMessageReady()) {
+ int numRead = recvMsg.recv(socketChannel);
+ if (numRead == -1) {
+ break;
+ }
}
+ return recvMsg;
+ } finally {
+ socketChannel.close();
}
- socketChannel.close();
- Assert.assertEquals(TestUtils.getIncreasingByteBuffer(10), recvMsg.getReadOnlyData());
}
}
diff --git a/docs/Configuration-Settings.md b/docs/Configuration-Settings.md
index b6119e6e2ad5..1d8df69c13c8 100644
--- a/docs/Configuration-Settings.md
+++ b/docs/Configuration-Settings.md
@@ -171,6 +171,47 @@ number.
| 128 MB |
Memory capacity of each worker node. |
+
+ | tachyon.worker.network.type |
+ NETTY |
+ Selects networking stack to run the worker with. Valid options are NETTY and NIO. |
+
+
+ | tachyon.worker.network.netty.channel |
+ EPOLL |
+ Selects netty's channel implementation. On linux, epoll is used; valid options are NIO and EPOLL. |
+
+
+ | tachyon.worker.network.netty.file.transfer |
+ MAPPED |
+ When returning files to the user, select how the data is transferred; valid options are MAPPED (uses java MappedByteBuffer) and TRANSFER (uses Java FileChannel.transferTo). |
+
+
+
+ | tachyon.worker.network.netty.watermark.high |
+ 32768 |
+ Determines how many bytes can be in the write queue before channels isWritable is set to false. |
+
+
+ | tachyon.worker.network.netty.watermark.low |
+ 8192 |
+ Once the high watermark limit is reached, the queue must be flushed down to the low watermark before switching back to writable. |
+
+
+ | tachyon.worker.network.netty.backlog |
+ 128 on linux |
+ How many requests can be queued up before new requests are rejected; this value is platform dependent. |
+
+
+ | tachyon.worker.network.netty.buffer.send |
+ platform specific |
+ Sets SO_SNDBUF for the socket; more details can be found in the socket man page. |
+
+
+ | tachyon.worker.network.netty.buffer.recieve |
+ platform specific |
+ Sets SO_RCVBUF for the socket; more details can be found in the socket man page. |
+
# User Configuration
@@ -204,4 +245,9 @@ The user configuration specifies values regarding file system access.
1 MB |
The size of the file buffer to read data from remote Tachyon worker. |
+
+ | tachyon.worker.network.netty.process.threads |
+ 16 |
+ How many threads to use to process block requests. |
+