// Try to get the file metadata first. The entire file must be
// available before it can be downloaded.
// If not available then we back off and retry using the provided
// retry policy.
ObjectMetadata attributes;
RetryPolicy retry = retryPolicy.duplicate();
do {
try {
attributes = provider.readMetadata(objectName);
if (attributes.isValidForRead())
break;
if (!retry.allowRetry())
throw new NotFoundException("File doesn't exists or isn't ready to be read: " + objectName);
}
catch (Exception e) {
LOG.warn(e.getMessage());
if (!retry.allowRetry())
throw e;
}
} while (true);
final AtomicReference<Exception> exception = new AtomicReference<Exception>();
final AtomicLong totalBytesRead = new AtomicLong();
final AtomicLong totalBytesRead2 = new AtomicLong();
// Iterate sequentially building up the batches. Once a complete
// batch of ids is ready
// randomize fetching the chunks and then download them in parallel
List<Integer> idsToRead = Lists.newArrayList();
for (int block = 0; block < attributes.getChunkCount(); block++) {
idsToRead.add(block);
// Got a batch, or reached the end
if (idsToRead.size() == batchSize || block == attributes.getChunkCount() - 1) {
// Read blocks in random order
final int firstBlockId = idsToRead.get(0);
Collections.shuffle(idsToRead);
final AtomicReferenceArray<ByteBuffer> chunks = new AtomicReferenceArray<ByteBuffer>(
idsToRead.size());
ExecutorService executor = Executors.newFixedThreadPool(
concurrencyLevel,
new ThreadFactoryBuilder().setDaemon(true)
.setNameFormat("ChunkReader-" + objectName + "-%d").build());
try {
for (final int chunkId : idsToRead) {
executor.submit(new Runnable() {
@Override
public void run() {
// Do the fetch
RetryPolicy retry = retryPolicy.duplicate();
while (exception.get() == null) {
try {
ByteBuffer chunk = provider.readChunk(objectName, chunkId);
totalBytesRead.addAndGet(chunk.remaining());
chunks.set(chunkId - firstBlockId, chunk);
callback.onChunk(chunkId, chunk);
break;
}
catch (Exception e) {
callback.onChunkException(chunkId, e);
if (retry.allowRetry())
continue;
exception.compareAndSet(null, e);
}
}
}