@Test public void positionedReadPartialPage() throws Exception { int fileSize = PAGE_SIZE; byte[] testData = BufferUtils.getIncreasingByteArray(fileSize); ByteArrayCacheManager manager = new ByteArrayCacheManager(); LocalCacheFileInStream stream = setupWithSingleFile(testData, manager); int partialReadSize = fileSize / 5; int offset = fileSize / 5; byte[] cacheMiss = new byte[partialReadSize]; Assert.assertEquals(partialReadSize, stream.positionedRead(offset, cacheMiss, 0, cacheMiss.length)); Assert.assertArrayEquals( Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheMiss); Assert.assertEquals(0, manager.mPagesServed); Assert.assertEquals(1, manager.mPagesCached); byte[] cacheHit = new byte[partialReadSize]; Assert.assertEquals(partialReadSize, stream.positionedRead(offset, cacheHit, 0, cacheHit.length)); Assert.assertArrayEquals( Arrays.copyOfRange(testData, offset, offset + partialReadSize), cacheHit); Assert.assertEquals(1, manager.mPagesServed); }
@Override public int positionedRead(long pos, byte[] b, int off, int len) throws IOException { Preconditions.checkArgument(len >= 0, "length should be non-negative"); Preconditions.checkArgument(off >= 0, "offset should be non-negative"); Preconditions.checkArgument(pos >= 0, "position should be non-negative"); if (len == 0) { return 0; } if (pos < 0 || pos >= mStatus.getLength()) { return -1; } int totalBytesRead = 0; long currentPosition = pos; long lengthToRead = Math.min(len, mStatus.getLength() - pos); while (totalBytesRead < lengthToRead) { long currentPage = currentPosition / mPageSize; int currentPageOffset = (int) (currentPosition % mPageSize); int bytesLeftInPage = (int) Math.min(mPageSize - currentPageOffset, lengthToRead - totalBytesRead); PageId pageId = new PageId(mStatus.getFileIdentifier(), currentPage); int bytesRead = mCacheManager.get(pageId, currentPageOffset, bytesLeftInPage, b, off + totalBytesRead); if (bytesRead > 0) { totalBytesRead += bytesRead; currentPosition += bytesRead; Metrics.BYTES_READ_CACHE.mark(bytesRead); } else { byte[] page = readExternalPage(currentPosition); if (page.length > 0) { System.arraycopy(page, currentPageOffset, b, off + totalBytesRead, bytesLeftInPage); totalBytesRead += bytesLeftInPage; currentPosition += bytesLeftInPage; Metrics.BYTES_REQUESTED_EXTERNAL.mark(bytesLeftInPage); mCacheManager.put(pageId, page); } } } if (totalBytesRead > len || (totalBytesRead < len && currentPosition < mStatus.getLength())) { throw new IOException(String.format( "Invalid number of bytes positionread - read from position = %d, " + "bytes to read = %d, actual bytes read = %d, bytes remains in file %d", pos, len, totalBytesRead, mStatus.getLength() - currentPosition) ); } return totalBytesRead; }
LocalCacheFileInStream extends FileInStream { @Override public int positionedRead(long pos, byte[] b, int off, int len) throws IOException { Preconditions.checkArgument(len >= 0, "length should be non-negative"); Preconditions.checkArgument(off >= 0, "offset should be non-negative"); Preconditions.checkArgument(pos >= 0, "position should be non-negative"); if (len == 0) { return 0; } if (pos < 0 || pos >= mStatus.getLength()) { return -1; } int totalBytesRead = 0; long currentPosition = pos; long lengthToRead = Math.min(len, mStatus.getLength() - pos); while (totalBytesRead < lengthToRead) { long currentPage = currentPosition / mPageSize; int currentPageOffset = (int) (currentPosition % mPageSize); int bytesLeftInPage = (int) Math.min(mPageSize - currentPageOffset, lengthToRead - totalBytesRead); PageId pageId = new PageId(mStatus.getFileIdentifier(), currentPage); int bytesRead = mCacheManager.get(pageId, currentPageOffset, bytesLeftInPage, b, off + totalBytesRead); if (bytesRead > 0) { totalBytesRead += bytesRead; currentPosition += bytesRead; Metrics.BYTES_READ_CACHE.mark(bytesRead); } else { byte[] page = readExternalPage(currentPosition); if (page.length > 0) { System.arraycopy(page, currentPageOffset, b, off + totalBytesRead, bytesLeftInPage); totalBytesRead += bytesLeftInPage; currentPosition += bytesLeftInPage; Metrics.BYTES_REQUESTED_EXTERNAL.mark(bytesLeftInPage); mCacheManager.put(pageId, page); } } } if (totalBytesRead > len || (totalBytesRead < len && currentPosition < mStatus.getLength())) { throw new IOException(String.format( "Invalid number of bytes positionread - read from position = %d, " + "bytes to read = %d, actual bytes read = %d, bytes remains in file %d", pos, len, totalBytesRead, mStatus.getLength() - currentPosition) ); } return totalBytesRead; } }
LocalCacheFileInStream extends FileInStream { @Override public int positionedRead(long pos, byte[] b, int off, int len) throws IOException { Preconditions.checkArgument(len >= 0, "length should be non-negative"); Preconditions.checkArgument(off >= 0, "offset should be non-negative"); Preconditions.checkArgument(pos >= 0, "position should be non-negative"); if (len == 0) { return 0; } if (pos < 0 || pos >= mStatus.getLength()) { return -1; } int totalBytesRead = 0; long currentPosition = pos; long lengthToRead = Math.min(len, mStatus.getLength() - pos); while (totalBytesRead < lengthToRead) { long currentPage = currentPosition / mPageSize; int currentPageOffset = (int) (currentPosition % mPageSize); int bytesLeftInPage = (int) Math.min(mPageSize - currentPageOffset, lengthToRead - totalBytesRead); PageId pageId = new PageId(mStatus.getFileIdentifier(), currentPage); int bytesRead = mCacheManager.get(pageId, currentPageOffset, bytesLeftInPage, b, off + totalBytesRead); if (bytesRead > 0) { totalBytesRead += bytesRead; currentPosition += bytesRead; Metrics.BYTES_READ_CACHE.mark(bytesRead); } else { byte[] page = readExternalPage(currentPosition); if (page.length > 0) { System.arraycopy(page, currentPageOffset, b, off + totalBytesRead, bytesLeftInPage); totalBytesRead += bytesLeftInPage; currentPosition += bytesLeftInPage; Metrics.BYTES_REQUESTED_EXTERNAL.mark(bytesLeftInPage); mCacheManager.put(pageId, page); } } } if (totalBytesRead > len || (totalBytesRead < len && currentPosition < mStatus.getLength())) { throw new IOException(String.format( "Invalid number of bytes positionread - read from position = %d, " + "bytes to read = %d, actual bytes read = %d, bytes remains in file %d", pos, len, totalBytesRead, mStatus.getLength() - currentPosition) ); } return totalBytesRead; } LocalCacheFileInStream(AlluxioURI path, OpenFilePOptions options, FileSystem externalFs,
      CacheManager cacheManager);  LocalCacheFileInStream(URIStatus status, OpenFilePOptions options, FileSystem externalFs,
      CacheManager cacheManager); }
LocalCacheFileInStream extends FileInStream { @Override public int positionedRead(long pos, byte[] b, int off, int len) throws IOException { Preconditions.checkArgument(len >= 0, "length should be non-negative"); Preconditions.checkArgument(off >= 0, "offset should be non-negative"); Preconditions.checkArgument(pos >= 0, "position should be non-negative"); if (len == 0) { return 0; } if (pos < 0 || pos >= mStatus.getLength()) { return -1; } int totalBytesRead = 0; long currentPosition = pos; long lengthToRead = Math.min(len, mStatus.getLength() - pos); while (totalBytesRead < lengthToRead) { long currentPage = currentPosition / mPageSize; int currentPageOffset = (int) (currentPosition % mPageSize); int bytesLeftInPage = (int) Math.min(mPageSize - currentPageOffset, lengthToRead - totalBytesRead); PageId pageId = new PageId(mStatus.getFileIdentifier(), currentPage); int bytesRead = mCacheManager.get(pageId, currentPageOffset, bytesLeftInPage, b, off + totalBytesRead); if (bytesRead > 0) { totalBytesRead += bytesRead; currentPosition += bytesRead; Metrics.BYTES_READ_CACHE.mark(bytesRead); } else { byte[] page = readExternalPage(currentPosition); if (page.length > 0) { System.arraycopy(page, currentPageOffset, b, off + totalBytesRead, bytesLeftInPage); totalBytesRead += bytesLeftInPage; currentPosition += bytesLeftInPage; Metrics.BYTES_REQUESTED_EXTERNAL.mark(bytesLeftInPage); mCacheManager.put(pageId, page); } } } if (totalBytesRead > len || (totalBytesRead < len && currentPosition < mStatus.getLength())) { throw new IOException(String.format( "Invalid number of bytes positionread - read from position = %d, " + "bytes to read = %d, actual bytes read = %d, bytes remains in file %d", pos, len, totalBytesRead, mStatus.getLength() - currentPosition) ); } return totalBytesRead; } LocalCacheFileInStream(AlluxioURI path, OpenFilePOptions options, FileSystem externalFs,
      CacheManager cacheManager);  LocalCacheFileInStream(URIStatus status, OpenFilePOptions options, FileSystem externalFs,
      CacheManager cacheManager); @Override int read(); @Override int read(byte[] b); @Override int read(byte[] b, int off, int len); @Override long skip(long n); @Override void close(); @Override long remaining(); @Override int positionedRead(long pos, byte[] b, int off, int len); @Override long getPos(); @Override void seek(long pos); }
LocalCacheFileInStream extends FileInStream { @Override public int positionedRead(long pos, byte[] b, int off, int len) throws IOException { Preconditions.checkArgument(len >= 0, "length should be non-negative"); Preconditions.checkArgument(off >= 0, "offset should be non-negative"); Preconditions.checkArgument(pos >= 0, "position should be non-negative"); if (len == 0) { return 0; } if (pos < 0 || pos >= mStatus.getLength()) { return -1; } int totalBytesRead = 0; long currentPosition = pos; long lengthToRead = Math.min(len, mStatus.getLength() - pos); while (totalBytesRead < lengthToRead) { long currentPage = currentPosition / mPageSize; int currentPageOffset = (int) (currentPosition % mPageSize); int bytesLeftInPage = (int) Math.min(mPageSize - currentPageOffset, lengthToRead - totalBytesRead); PageId pageId = new PageId(mStatus.getFileIdentifier(), currentPage); int bytesRead = mCacheManager.get(pageId, currentPageOffset, bytesLeftInPage, b, off + totalBytesRead); if (bytesRead > 0) { totalBytesRead += bytesRead; currentPosition += bytesRead; Metrics.BYTES_READ_CACHE.mark(bytesRead); } else { byte[] page = readExternalPage(currentPosition); if (page.length > 0) { System.arraycopy(page, currentPageOffset, b, off + totalBytesRead, bytesLeftInPage); totalBytesRead += bytesLeftInPage; currentPosition += bytesLeftInPage; Metrics.BYTES_REQUESTED_EXTERNAL.mark(bytesLeftInPage); mCacheManager.put(pageId, page); } } } if (totalBytesRead > len || (totalBytesRead < len && currentPosition < mStatus.getLength())) { throw new IOException(String.format( "Invalid number of bytes positionread - read from position = %d, " + "bytes to read = %d, actual bytes read = %d, bytes remains in file %d", pos, len, totalBytesRead, mStatus.getLength() - currentPosition) ); } return totalBytesRead; } LocalCacheFileInStream(AlluxioURI path, OpenFilePOptions options, FileSystem externalFs,
      CacheManager cacheManager);  LocalCacheFileInStream(URIStatus status, OpenFilePOptions options, FileSystem externalFs,
      CacheManager cacheManager); @Override int read(); @Override int read(byte[] b); @Override int read(byte[] b, int off, int len); @Override long skip(long n); @Override void close(); @Override long remaining(); @Override int positionedRead(long pos, byte[] b, int off, int len); @Override long getPos(); @Override void seek(long pos);  }