@Test public void testSingleRequestWithPrefixAndSuffix() throws IOException { byte[] buffer = new byte[25]; ReadRequest rr = new ReadRequest(100, 200, 150, 175, buffer, 0, backendFile.length()); String generatedTestData = DataGen.generateContent(); String expectedBufferOutput = generatedTestData.substring(150, 175); String expectedCacheOutput = generatedTestData.substring(100, 200); remoteReadRequestChain.addReadRequest(rr); remoteReadRequestChain.lock(); long readSize = remoteReadRequestChain.call(); assertEquals(readSize, expectedBufferOutput.length()); String actualBufferOutput = new String(buffer, Charset.defaultCharset()); assertEquals(actualBufferOutput, expectedBufferOutput, "Wrong data read, expected\n" + expectedBufferOutput + "\nBut got\n" + actualBufferOutput); byte[] filledBuffer = new byte[expectedCacheOutput.length()]; byte[] emptyBuffer = new byte[100]; readSize = 0; FileInputStream localFileInputStream = new FileInputStream(new File(localFileName)); for (int i = 0; i < 20; i++) { if (i == 1) { readSize += localFileInputStream.read(filledBuffer, 0, 100); } else { localFileInputStream.read(emptyBuffer, 0, 100); for (int j = 0; j < 100; j++) { assertEquals(emptyBuffer[j], 0, "Got data instead of hole: " + emptyBuffer[j]); } } } localFileInputStream.close(); assertEquals(readSize, expectedCacheOutput.length()); String actualCacheOutput = new String(filledBuffer, Charset.defaultCharset()); assertEquals(actualCacheOutput, expectedCacheOutput, "Wrong data read in local randomAccessFile, expected\n" + expectedCacheOutput + "\nBut got\n" + actualCacheOutput); }
public Long call() throws IOException { log.debug(String.format("Read Request threadName: %s, Remote read Executor threadName: %s", threadName, Thread.currentThread().getName())); Thread.currentThread().setName(threadName); checkState(isLocked, "Trying to execute Chain without locking"); if (readRequests.size() == 0) { return 0L; } File file = new File(localFile); if (!file.exists()) { throw new IOException(String.format("File does not exists %s", localFile)); } FileChannel fileChannel = new FileOutputStream(new RandomAccessFile(file, "rw").getFD()).getChannel(); ByteBuffer directBuffer = bufferPool.getBuffer(directBufferSize); checkState(directBuffer != null, "directBuffer could not be allocated"); try { for (ReadRequest readRequest : readRequests) { if (cancelled) { propagateCancel(this.getClass().getName()); } log.debug(String.format("Executing ReadRequest: [%d, %d, %d, %d, %d]", readRequest.getBackendReadStart(), readRequest.getBackendReadEnd(), readRequest.getActualReadStart(), readRequest.getActualReadEnd(), readRequest.getDestBufferOffset())); int prefixBufferLength = (int) (readRequest.getActualReadStart() - readRequest.getBackendReadStart()); int suffixBufferLength = (int) (readRequest.getBackendReadEnd() - readRequest.getActualReadEnd()); log.debug(String.format("PrefixLength: %d SuffixLength: %d", prefixBufferLength, suffixBufferLength)); if (prefixBufferLength > 0) { inputStream.seek(readRequest.backendReadStart); log.debug(String.format("Trying to Read %d bytes into prefix buffer", prefixBufferLength)); extraRead += readIntoBuffer(affixBuffer, 0, prefixBufferLength, inputStream); int written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, prefixBufferLength, readRequest.backendReadStart); log.debug(String.format("Copied %d prefix bytes into cache", written)); } log.debug(String.format("Trying to Read %d bytes into destination buffer", readRequest.getActualReadLengthIntUnsafe())); int readBytes; if (prefixBufferLength > 0 || suffixBufferLength > 0) { inputStream.seek(readRequest.actualReadStart); readBytes = readIntoBuffer(readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), inputStream); } else { inputStream.readFully(readRequest.actualReadStart, readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe()); readBytes = readRequest.getActualReadLengthIntUnsafe(); } int written = copyIntoCache(fileChannel, directBuffer, readRequest.destBuffer, readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), readRequest.actualReadStart); log.debug(String.format("Copied %d requested bytes into cache", written)); totalRequestedRead += readRequest.getActualReadLengthIntUnsafe(); if (suffixBufferLength > 0) { log.debug(String.format("Trying to Read %d bytes into suffix buffer", suffixBufferLength)); inputStream.seek(readRequest.actualReadEnd); extraRead += readIntoBuffer(affixBuffer, 0, suffixBufferLength, inputStream); written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, suffixBufferLength, readRequest.actualReadEnd); log.debug(String.format("Copied %d suffix bytes into cache", written)); } } log.debug(String.format("Read %d bytes from remote localFile, added %d to destination buffer", extraRead + totalRequestedRead, totalRequestedRead)); return totalRequestedRead; } finally { bufferPool.returnBuffer(directBuffer); fileChannel.close(); } }
RemoteReadRequestChain extends ReadRequestChain { public Long call() throws IOException { log.debug(String.format("Read Request threadName: %s, Remote read Executor threadName: %s", threadName, Thread.currentThread().getName())); Thread.currentThread().setName(threadName); checkState(isLocked, "Trying to execute Chain without locking"); if (readRequests.size() == 0) { return 0L; } File file = new File(localFile); if (!file.exists()) { throw new IOException(String.format("File does not exists %s", localFile)); } FileChannel fileChannel = new FileOutputStream(new RandomAccessFile(file, "rw").getFD()).getChannel(); ByteBuffer directBuffer = bufferPool.getBuffer(directBufferSize); checkState(directBuffer != null, "directBuffer could not be allocated"); try { for (ReadRequest readRequest : readRequests) { if (cancelled) { propagateCancel(this.getClass().getName()); } log.debug(String.format("Executing ReadRequest: [%d, %d, %d, %d, %d]", readRequest.getBackendReadStart(), readRequest.getBackendReadEnd(), readRequest.getActualReadStart(), readRequest.getActualReadEnd(), readRequest.getDestBufferOffset())); int prefixBufferLength = (int) (readRequest.getActualReadStart() - readRequest.getBackendReadStart()); int suffixBufferLength = (int) (readRequest.getBackendReadEnd() - readRequest.getActualReadEnd()); log.debug(String.format("PrefixLength: %d SuffixLength: %d", prefixBufferLength, suffixBufferLength)); if (prefixBufferLength > 0) { inputStream.seek(readRequest.backendReadStart); log.debug(String.format("Trying to Read %d bytes into prefix buffer", prefixBufferLength)); extraRead += readIntoBuffer(affixBuffer, 0, prefixBufferLength, inputStream); int written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, prefixBufferLength, readRequest.backendReadStart); log.debug(String.format("Copied %d prefix bytes into cache", written)); } log.debug(String.format("Trying to Read %d bytes into destination buffer", readRequest.getActualReadLengthIntUnsafe())); int readBytes; if (prefixBufferLength > 0 || suffixBufferLength > 0) { inputStream.seek(readRequest.actualReadStart); readBytes = readIntoBuffer(readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), inputStream); } else { inputStream.readFully(readRequest.actualReadStart, readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe()); readBytes = readRequest.getActualReadLengthIntUnsafe(); } int written = copyIntoCache(fileChannel, directBuffer, readRequest.destBuffer, readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), readRequest.actualReadStart); log.debug(String.format("Copied %d requested bytes into cache", written)); totalRequestedRead += readRequest.getActualReadLengthIntUnsafe(); if (suffixBufferLength > 0) { log.debug(String.format("Trying to Read %d bytes into suffix buffer", suffixBufferLength)); inputStream.seek(readRequest.actualReadEnd); extraRead += readIntoBuffer(affixBuffer, 0, suffixBufferLength, inputStream); written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, suffixBufferLength, readRequest.actualReadEnd); log.debug(String.format("Copied %d suffix bytes into cache", written)); } } log.debug(String.format("Read %d bytes from remote localFile, added %d to destination buffer", extraRead + totalRequestedRead, totalRequestedRead)); return totalRequestedRead; } finally { bufferPool.returnBuffer(directBuffer); fileChannel.close(); } } }
RemoteReadRequestChain extends ReadRequestChain { public Long call() throws IOException { log.debug(String.format("Read Request threadName: %s, Remote read Executor threadName: %s", threadName, Thread.currentThread().getName())); Thread.currentThread().setName(threadName); checkState(isLocked, "Trying to execute Chain without locking"); if (readRequests.size() == 0) { return 0L; } File file = new File(localFile); if (!file.exists()) { throw new IOException(String.format("File does not exists %s", localFile)); } FileChannel fileChannel = new FileOutputStream(new RandomAccessFile(file, "rw").getFD()).getChannel(); ByteBuffer directBuffer = bufferPool.getBuffer(directBufferSize); checkState(directBuffer != null, "directBuffer could not be allocated"); try { for (ReadRequest readRequest : readRequests) { if (cancelled) { propagateCancel(this.getClass().getName()); } log.debug(String.format("Executing ReadRequest: [%d, %d, %d, %d, %d]", readRequest.getBackendReadStart(), readRequest.getBackendReadEnd(), readRequest.getActualReadStart(), readRequest.getActualReadEnd(), readRequest.getDestBufferOffset())); int prefixBufferLength = (int) (readRequest.getActualReadStart() - readRequest.getBackendReadStart()); int suffixBufferLength = (int) (readRequest.getBackendReadEnd() - readRequest.getActualReadEnd()); log.debug(String.format("PrefixLength: %d SuffixLength: %d", prefixBufferLength, suffixBufferLength)); if (prefixBufferLength > 0) { inputStream.seek(readRequest.backendReadStart); log.debug(String.format("Trying to Read %d bytes into prefix buffer", prefixBufferLength)); extraRead += readIntoBuffer(affixBuffer, 0, prefixBufferLength, inputStream); int written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, prefixBufferLength, readRequest.backendReadStart); log.debug(String.format("Copied %d prefix bytes into cache", written)); } log.debug(String.format("Trying to Read %d bytes into destination buffer", readRequest.getActualReadLengthIntUnsafe())); int readBytes; if (prefixBufferLength > 0 || suffixBufferLength > 0) { inputStream.seek(readRequest.actualReadStart); readBytes = readIntoBuffer(readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), inputStream); } else { inputStream.readFully(readRequest.actualReadStart, readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe()); readBytes = readRequest.getActualReadLengthIntUnsafe(); } int written = copyIntoCache(fileChannel, directBuffer, readRequest.destBuffer, readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), readRequest.actualReadStart); log.debug(String.format("Copied %d requested bytes into cache", written)); totalRequestedRead += readRequest.getActualReadLengthIntUnsafe(); if (suffixBufferLength > 0) { log.debug(String.format("Trying to Read %d bytes into suffix buffer", suffixBufferLength)); inputStream.seek(readRequest.actualReadEnd); extraRead += readIntoBuffer(affixBuffer, 0, suffixBufferLength, inputStream); written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, suffixBufferLength, readRequest.actualReadEnd); log.debug(String.format("Copied %d suffix bytes into cache", written)); } } log.debug(String.format("Read %d bytes from remote localFile, added %d to destination buffer", extraRead + totalRequestedRead, totalRequestedRead)); return totalRequestedRead; } finally { bufferPool.returnBuffer(directBuffer); fileChannel.close(); } } RemoteReadRequestChain(FSDataInputStream inputStream,
      String remotePath,
      int generationNumber,
      DirectBufferPool bufferPool,
      Configuration conf,
      byte[] affixBuffer,
      BookKeeperFactory bookKeeperFactory); @VisibleForTesting  RemoteReadRequestChain(FSDataInputStream inputStream, String remoteFileName, int generationNumber, Configuration conf); }
RemoteReadRequestChain extends ReadRequestChain { public Long call() throws IOException { log.debug(String.format("Read Request threadName: %s, Remote read Executor threadName: %s", threadName, Thread.currentThread().getName())); Thread.currentThread().setName(threadName); checkState(isLocked, "Trying to execute Chain without locking"); if (readRequests.size() == 0) { return 0L; } File file = new File(localFile); if (!file.exists()) { throw new IOException(String.format("File does not exists %s", localFile)); } FileChannel fileChannel = new FileOutputStream(new RandomAccessFile(file, "rw").getFD()).getChannel(); ByteBuffer directBuffer = bufferPool.getBuffer(directBufferSize); checkState(directBuffer != null, "directBuffer could not be allocated"); try { for (ReadRequest readRequest : readRequests) { if (cancelled) { propagateCancel(this.getClass().getName()); } log.debug(String.format("Executing ReadRequest: [%d, %d, %d, %d, %d]", readRequest.getBackendReadStart(), readRequest.getBackendReadEnd(), readRequest.getActualReadStart(), readRequest.getActualReadEnd(), readRequest.getDestBufferOffset())); int prefixBufferLength = (int) (readRequest.getActualReadStart() - readRequest.getBackendReadStart()); int suffixBufferLength = (int) (readRequest.getBackendReadEnd() - readRequest.getActualReadEnd()); log.debug(String.format("PrefixLength: %d SuffixLength: %d", prefixBufferLength, suffixBufferLength)); if (prefixBufferLength > 0) { inputStream.seek(readRequest.backendReadStart); log.debug(String.format("Trying to Read %d bytes into prefix buffer", prefixBufferLength)); extraRead += readIntoBuffer(affixBuffer, 0, prefixBufferLength, inputStream); int written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, prefixBufferLength, readRequest.backendReadStart); log.debug(String.format("Copied %d prefix bytes into cache", written)); } log.debug(String.format("Trying to Read %d bytes into destination buffer", readRequest.getActualReadLengthIntUnsafe())); int readBytes; if (prefixBufferLength > 0 || suffixBufferLength > 0) { inputStream.seek(readRequest.actualReadStart); readBytes = readIntoBuffer(readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), inputStream); } else { inputStream.readFully(readRequest.actualReadStart, readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe()); readBytes = readRequest.getActualReadLengthIntUnsafe(); } int written = copyIntoCache(fileChannel, directBuffer, readRequest.destBuffer, readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), readRequest.actualReadStart); log.debug(String.format("Copied %d requested bytes into cache", written)); totalRequestedRead += readRequest.getActualReadLengthIntUnsafe(); if (suffixBufferLength > 0) { log.debug(String.format("Trying to Read %d bytes into suffix buffer", suffixBufferLength)); inputStream.seek(readRequest.actualReadEnd); extraRead += readIntoBuffer(affixBuffer, 0, suffixBufferLength, inputStream); written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, suffixBufferLength, readRequest.actualReadEnd); log.debug(String.format("Copied %d suffix bytes into cache", written)); } } log.debug(String.format("Read %d bytes from remote localFile, added %d to destination buffer", extraRead + totalRequestedRead, totalRequestedRead)); return totalRequestedRead; } finally { bufferPool.returnBuffer(directBuffer); fileChannel.close(); } } RemoteReadRequestChain(FSDataInputStream inputStream,
      String remotePath,
      int generationNumber,
      DirectBufferPool bufferPool,
      Configuration conf,
      byte[] affixBuffer,
      BookKeeperFactory bookKeeperFactory); @VisibleForTesting  RemoteReadRequestChain(FSDataInputStream inputStream, String remoteFileName, int generationNumber, Configuration conf); Long call(); static int readIntoBuffer(byte[] destBuffer, int destBufferOffset, int length, FSDataInputStream inputStream); ReadRequestChainStats getStats(); @Override void updateCacheStatus(String remotePath, long fileSize, long lastModified, int blockSize, Configuration conf); }
RemoteReadRequestChain extends ReadRequestChain { public Long call() throws IOException { log.debug(String.format("Read Request threadName: %s, Remote read Executor threadName: %s", threadName, Thread.currentThread().getName())); Thread.currentThread().setName(threadName); checkState(isLocked, "Trying to execute Chain without locking"); if (readRequests.size() == 0) { return 0L; } File file = new File(localFile); if (!file.exists()) { throw new IOException(String.format("File does not exists %s", localFile)); } FileChannel fileChannel = new FileOutputStream(new RandomAccessFile(file, "rw").getFD()).getChannel(); ByteBuffer directBuffer = bufferPool.getBuffer(directBufferSize); checkState(directBuffer != null, "directBuffer could not be allocated"); try { for (ReadRequest readRequest : readRequests) { if (cancelled) { propagateCancel(this.getClass().getName()); } log.debug(String.format("Executing ReadRequest: [%d, %d, %d, %d, %d]", readRequest.getBackendReadStart(), readRequest.getBackendReadEnd(), readRequest.getActualReadStart(), readRequest.getActualReadEnd(), readRequest.getDestBufferOffset())); int prefixBufferLength = (int) (readRequest.getActualReadStart() - readRequest.getBackendReadStart()); int suffixBufferLength = (int) (readRequest.getBackendReadEnd() - readRequest.getActualReadEnd()); log.debug(String.format("PrefixLength: %d SuffixLength: %d", prefixBufferLength, suffixBufferLength)); if (prefixBufferLength > 0) { inputStream.seek(readRequest.backendReadStart); log.debug(String.format("Trying to Read %d bytes into prefix buffer", prefixBufferLength)); extraRead += readIntoBuffer(affixBuffer, 0, prefixBufferLength, inputStream); int written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, prefixBufferLength, readRequest.backendReadStart); log.debug(String.format("Copied %d prefix bytes into cache", written)); } log.debug(String.format("Trying to Read %d bytes into destination buffer", readRequest.getActualReadLengthIntUnsafe())); int readBytes; if (prefixBufferLength > 0 || suffixBufferLength > 0) { inputStream.seek(readRequest.actualReadStart); readBytes = readIntoBuffer(readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), inputStream); } else { inputStream.readFully(readRequest.actualReadStart, readRequest.getDestBuffer(), readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe()); readBytes = readRequest.getActualReadLengthIntUnsafe(); } int written = copyIntoCache(fileChannel, directBuffer, readRequest.destBuffer, readRequest.destBufferOffset, readRequest.getActualReadLengthIntUnsafe(), readRequest.actualReadStart); log.debug(String.format("Copied %d requested bytes into cache", written)); totalRequestedRead += readRequest.getActualReadLengthIntUnsafe(); if (suffixBufferLength > 0) { log.debug(String.format("Trying to Read %d bytes into suffix buffer", suffixBufferLength)); inputStream.seek(readRequest.actualReadEnd); extraRead += readIntoBuffer(affixBuffer, 0, suffixBufferLength, inputStream); written = copyIntoCache(fileChannel, directBuffer, affixBuffer, 0, suffixBufferLength, readRequest.actualReadEnd); log.debug(String.format("Copied %d suffix bytes into cache", written)); } } log.debug(String.format("Read %d bytes from remote localFile, added %d to destination buffer", extraRead + totalRequestedRead, totalRequestedRead)); return totalRequestedRead; } finally { bufferPool.returnBuffer(directBuffer); fileChannel.close(); } } RemoteReadRequestChain(FSDataInputStream inputStream,
      String remotePath,
      int generationNumber,
      DirectBufferPool bufferPool,
      Configuration conf,
      byte[] affixBuffer,
      BookKeeperFactory bookKeeperFactory); @VisibleForTesting  RemoteReadRequestChain(FSDataInputStream inputStream, String remoteFileName, int generationNumber, Configuration conf); Long call(); static int readIntoBuffer(byte[] destBuffer, int destBufferOffset, int length, FSDataInputStream inputStream); ReadRequestChainStats getStats(); @Override void updateCacheStatus(String remotePath, long fileSize, long lastModified, int blockSize, Configuration conf);  }