@Test public void skipTest() throws IOException { for (int k = MIN_LEN + DELTA; k <= MAX_LEN; k += DELTA) { WriteType op = WriteType.THROUGH; int fileId = TestUtils.createByteFile(mTfs, "/root/testFile_" + k + "_" + op, op, k); TachyonFile file = mTfs.getFile(fileId); InStream is = file.getInStream(ReadType.CACHE); Assert.assertTrue(is instanceof RemoteBlockInStream); Assert.assertEquals(k / 2, is.skip(k / 2)); Assert.assertEquals(k / 2, is.read()); is.close(); Assert.assertFalse(file.isInMemory()); if (k >= 3) { is = file.getInStream(ReadType.CACHE); Assert.assertTrue(is instanceof RemoteBlockInStream); int t = k / 3; Assert.assertEquals(t, is.skip(t)); Assert.assertEquals(t, is.read()); Assert.assertEquals(t, is.skip(t)); Assert.assertEquals(2 * t + 1, is.read()); is.close(); Assert.assertFalse(file.isInMemory()); } } }
@Override public long skip(long n) throws IOException { if (n <= 0) { return 0; } long ret = mBlockInfo.length - mReadByte; if (ret > n) { ret = n; } if (mCurrentBuffer != null) { if (mCurrentBuffer.remaining() < ret) { mBufferStartPosition = mReadByte + ret; updateCurrentBuffer(); } if (mCurrentBuffer != null) { if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return (int) ret; } setupStreamFromUnderFs(mBlockInfo.offset + mReadByte, mUFSConf); } long tmp = mCheckpointInputStream.skip(ret); ret = Math.min(ret, tmp); mReadByte += ret; if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return ret; }
RemoteBlockInStream extends BlockInStream { @Override public long skip(long n) throws IOException { if (n <= 0) { return 0; } long ret = mBlockInfo.length - mReadByte; if (ret > n) { ret = n; } if (mCurrentBuffer != null) { if (mCurrentBuffer.remaining() < ret) { mBufferStartPosition = mReadByte + ret; updateCurrentBuffer(); } if (mCurrentBuffer != null) { if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return (int) ret; } setupStreamFromUnderFs(mBlockInfo.offset + mReadByte, mUFSConf); } long tmp = mCheckpointInputStream.skip(ret); ret = Math.min(ret, tmp); mReadByte += ret; if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return ret; } }
RemoteBlockInStream extends BlockInStream { @Override public long skip(long n) throws IOException { if (n <= 0) { return 0; } long ret = mBlockInfo.length - mReadByte; if (ret > n) { ret = n; } if (mCurrentBuffer != null) { if (mCurrentBuffer.remaining() < ret) { mBufferStartPosition = mReadByte + ret; updateCurrentBuffer(); } if (mCurrentBuffer != null) { if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return (int) ret; } setupStreamFromUnderFs(mBlockInfo.offset + mReadByte, mUFSConf); } long tmp = mCheckpointInputStream.skip(ret); ret = Math.min(ret, tmp); mReadByte += ret; if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return ret; } RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex);   RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf); }
RemoteBlockInStream extends BlockInStream { @Override public long skip(long n) throws IOException { if (n <= 0) { return 0; } long ret = mBlockInfo.length - mReadByte; if (ret > n) { ret = n; } if (mCurrentBuffer != null) { if (mCurrentBuffer.remaining() < ret) { mBufferStartPosition = mReadByte + ret; updateCurrentBuffer(); } if (mCurrentBuffer != null) { if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return (int) ret; } setupStreamFromUnderFs(mBlockInfo.offset + mReadByte, mUFSConf); } long tmp = mCheckpointInputStream.skip(ret); ret = Math.min(ret, tmp); mReadByte += ret; if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return ret; } RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex);   RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf); @Override void close(); @Override int read(); @Override int read(byte b[]); @Override int read(byte b[], int off, int len); @Override void seek(long pos); @Override long skip(long n); }
RemoteBlockInStream extends BlockInStream { @Override public long skip(long n) throws IOException { if (n <= 0) { return 0; } long ret = mBlockInfo.length - mReadByte; if (ret > n) { ret = n; } if (mCurrentBuffer != null) { if (mCurrentBuffer.remaining() < ret) { mBufferStartPosition = mReadByte + ret; updateCurrentBuffer(); } if (mCurrentBuffer != null) { if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return (int) ret; } setupStreamFromUnderFs(mBlockInfo.offset + mReadByte, mUFSConf); } long tmp = mCheckpointInputStream.skip(ret); ret = Math.min(ret, tmp); mReadByte += ret; if (ret > 0) { if (mRecache) { mBlockOutStream.cancel(); } mRecache = false; } return ret; } RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex);   RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf); @Override void close(); @Override int read(); @Override int read(byte b[]); @Override int read(byte b[], int off, int len); @Override void seek(long pos); @Override long skip(long n);  }