modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsInputStreamImpl.java - ignite - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.ignite.internal.processors.igfs;

 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.events.IgfsEvent;
 import org.apache.ignite.igfs.IgfsCorruptedFileException;
 import org.apache.ignite.igfs.IgfsInputStream;
 import org.apache.ignite.igfs.IgfsPath;
 import org.apache.ignite.igfs.IgfsPathNotFoundException;
 import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystemPositionedReadable;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
 import org.apache.ignite.internal.util.GridConcurrentHashSet;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteInClosure;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;

 import java.io.EOFException;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;

 import static org.apache.ignite.events.EventType.EVT_IGFS_FILE_CLOSED_READ;

 /**
  * Input stream to read data from grid cache with separate blocks.
  */
 public class IgfsInputStreamImpl extends IgfsInputStream implements IgfsSecondaryFileSystemPositionedReadable {
     /** Empty chunks result. */
     private static final byte[][] EMPTY_CHUNKS = new byte[0][];

     /** IGFS context. */
     private final IgfsContext igfsCtx;

     /** Secondary file system reader. */
     @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
     private final IgfsSecondaryFileSystemPositionedReadable secReader;

     /** Logger. */
     private IgniteLogger log;

     /** Path to file. */
     protected final IgfsPath path;

     /** File descriptor. */
     private volatile IgfsEntryInfo fileInfo;

     /** The number of already read bytes. Important! Access to the property is guarded by this object lock. */
     private long pos;

     /** Local cache. */
     private final Map<Long, IgniteInternalFuture<byte[]>> locCache;

     /** Maximum local cache size. */
     private final int maxLocCacheSize;

     /** Pending data read futures which were evicted from the local cache before completion. */
     private final Set<IgniteInternalFuture<byte[]>> pendingFuts;

     /** Pending futures lock. */
     private final Lock pendingFutsLock = new ReentrantLock();

     /** Pending futures condition. */
     private final Condition pendingFutsCond = pendingFutsLock.newCondition();

     /** Closed flag. */
     private boolean closed;

     /** Number of blocks to prefetch asynchronously. */
     private int prefetchBlocks;

     /** Numbed of blocks that must be read sequentially before prefetch is triggered. */
     private int seqReadsBeforePrefetch;

     /** Bytes read. */
     private long bytes;

     /** Index of the previously read block. Initially it is set to -1 indicating that no reads has been made so far. */
     private long prevBlockIdx = -1;

     /** Amount of sequential reads performed. */
     private int seqReads;

     /** Time consumed on reading. */
     private long time;

     /** File Length. */
     private long len;

     /** Block size to read. */
     private int blockSize;

     /** Block size to read. */
     private long blocksCnt;

     /** Proxy mode. */
     private boolean proxy;

     /**
      * Constructs file output stream.
      *  @param igfsCtx IGFS context.
      * @param path Path to stored file.
      * @param fileInfo File info to write binary data to.
      * @param prefetchBlocks Number of blocks to prefetch.
      * @param seqReadsBeforePrefetch Amount of sequential reads before prefetch is triggered.
      * @param secReader Optional secondary file system reader.
      * @param len File length.
      * @param blockSize Block size.
      * @param blocksCnt Blocks count.
      * @param proxy Proxy mode flag.
      */
     IgfsInputStreamImpl(
         IgfsContext igfsCtx,
         IgfsPath path,
         @Nullable IgfsEntryInfo fileInfo,
         int prefetchBlocks,
         int seqReadsBeforePrefetch,
         @Nullable IgfsSecondaryFileSystemPositionedReadable secReader,
         long len,
         int blockSize,
         long blocksCnt,
         boolean proxy) {
         assert igfsCtx != null;
         assert path != null;

         this.igfsCtx = igfsCtx;
         this.path = path;
         this.fileInfo = fileInfo;
         this.prefetchBlocks = prefetchBlocks;
         this.seqReadsBeforePrefetch = seqReadsBeforePrefetch;
         this.secReader = secReader;
         this.len = len;
         this.blockSize = blockSize;
         this.blocksCnt = blocksCnt;
         this.proxy = proxy;

         log = igfsCtx.kernalContext().log(IgfsInputStream.class);

         maxLocCacheSize = (prefetchBlocks > 0 ? prefetchBlocks : 1) * 3 / 2;

         locCache = new LinkedHashMap<>(maxLocCacheSize, 1.0f);

         pendingFuts = new GridConcurrentHashSet<>(prefetchBlocks > 0 ? prefetchBlocks : 1);

         igfsCtx.metrics().incrementFilesOpenedForRead();
     }

     /**
      * Gets bytes read.
      *
      * @return Bytes read.
      */
     public synchronized long bytes() {
         return bytes;
     }

     /** {@inheritDoc} */
     @Override public long length() {
         return len;
     }

     /** {@inheritDoc} */
     @Override public synchronized int read() throws IOException {
         byte[] buf = new byte[1];

         int read = read(buf, 0, 1);

         if (read == -1)
             return -1; // EOF.

         return buf[0] & 0xFF; // Cast to int and cut to *unsigned* byte value.
     }

     /** {@inheritDoc} */
     @Override public synchronized int read(@NotNull byte[] b, int off, int len) throws IOException {
         int read = readFromStore(pos, b, off, len);

         if (read != -1)
             pos += read;

         return read;
     }

     /** {@inheritDoc} */
     @Override public synchronized void seek(long pos) throws IOException {
         if (pos < 0)
             throw new IOException("Seek position cannot be negative: " + pos);


         this.pos = pos;
     }

     /** {@inheritDoc} */
     @Override public synchronized long position() throws IOException {
         return pos;
     }

     /** {@inheritDoc} */
     @Override public synchronized int available() throws IOException {
         long l = len - pos;

         if (l < 0)
             return 0;

         if (l > Integer.MAX_VALUE)
             return Integer.MAX_VALUE;

         return (int)l;
     }

     /** {@inheritDoc} */
     @Override public synchronized void readFully(long pos, byte[] buf) throws IOException {
         readFully(pos, buf, 0, buf.length);
     }

     /** {@inheritDoc} */
     @Override public synchronized void readFully(long pos, byte[] buf, int off, int len) throws IOException {
         for (int readBytes = 0; readBytes < len; ) {
             int read = readFromStore(pos + readBytes, buf, off + readBytes, len - readBytes);

             if (read == -1)
                 throw new EOFException("Failed to read stream fully (stream ends unexpectedly)" +
                     "[pos=" + pos + ", buf.length=" + buf.length + ", off=" + off + ", len=" + len + ']');

             readBytes += read;
         }
     }

     /** {@inheritDoc} */
     @Override public synchronized int read(long pos, byte[] buf, int off, int len) throws IOException {
         return readFromStore(pos, buf, off, len);
     }

     /**
      * Reads bytes from given position.
      *
      * @param pos Position to read from.
      * @param len Number of bytes to read.
      * @return Array of chunks with respect to chunk file representation.
      * @throws IOException If read failed.
      */
     @SuppressWarnings("IfMayBeConditional")
     public synchronized byte[][] readChunks(long pos, int len) throws IOException {
         // Readable bytes in the file, starting from the specified position.
         long readable = this.len - pos;

         if (readable <= 0)
             return EMPTY_CHUNKS;

         long startTime = System.nanoTime();

         if (readable < len)
             len = (int)readable; // Truncate expected length to available.

         assert len > 0;

         bytes += len;

         int start = (int)(pos / blockSize);
         int end = (int)((pos + len - 1) / blockSize);

         int chunkCnt = end - start + 1;

         byte[][] chunks = new byte[chunkCnt][];

         for (int i = 0; i < chunkCnt; i++) {
             byte[] block = blockFragmentizerSafe(start + i);

             int blockOff = (int)(pos % blockSize);
             int blockLen = Math.min(len, block.length - blockOff);

             // If whole block can be used as result, do not do array copy.
             if (blockLen == block.length)
                 chunks[i] = block;
             else {
                 // Only first or last block can have non-full data.
                 assert i == 0 || i == chunkCnt - 1;

                 chunks[i] = Arrays.copyOfRange(block, blockOff, blockOff + blockLen);
             }

             len -= blockLen;
             pos += blockLen;
         }

         assert len == 0;

         time += System.nanoTime() - startTime;

         return chunks;
     }

     /** {@inheritDoc} */
     @Override public synchronized void close() throws IOException {
         if (!closed) {
             try {
                 if (secReader != null) {
                     // Close secondary input stream.
                     secReader.close();

                     // Ensuring local cache futures completion.
                     for (IgniteInternalFuture<byte[]> fut : locCache.values()) {
                         try {
                             fut.get();
                         }
                         catch (IgniteCheckedException ignore) {
                             // No-op.
                         }
                     }

                     // Ensuring pending evicted futures completion.
                     while (!pendingFuts.isEmpty()) {
                         pendingFutsLock.lock();

                         try {
                             pendingFutsCond.await(100, TimeUnit.MILLISECONDS);
                         }
                         catch (InterruptedException ignore) {
                             // No-op.
                         }
                         finally {
                             pendingFutsLock.unlock();
                         }
                     }
                 }
             }
             catch (Exception e) {
                 throw new IOException("File to close the file: " + path, e);
             }
             finally {
                 closed = true;

                 IgfsLocalMetrics metrics = igfsCtx.metrics();

                 metrics.addReadBytesTime(bytes, time);
                 metrics.decrementFilesOpenedForRead();

                 locCache.clear();

                 GridEventStorageManager evts = igfsCtx.kernalContext().event();

                 if (evts.isRecordable(EVT_IGFS_FILE_CLOSED_READ))
                     evts.record(new IgfsEvent(path, igfsCtx.localNode(), EVT_IGFS_FILE_CLOSED_READ, bytes()));
             }
         }
     }

     /**
      * @param pos Position to start reading from.
      * @param buf Data buffer to save read data to.
      * @param off Offset in the buffer to write data from.
      * @param len Length of the data to read from the stream.
      * @return Number of actually read bytes.
      * @throws IOException In case of any IO exception.
      */
     private int readFromStore(long pos, byte[] buf, int off, int len) throws IOException {
         if (pos < 0)
             throw new IllegalArgumentException("Read position cannot be negative: " + pos);

         if (buf == null)
             throw new NullPointerException("Destination buffer cannot be null.");

         if (off < 0 || len < 0 || buf.length < len + off)
             throw new IndexOutOfBoundsException("Invalid buffer boundaries " +
                 "[buf.length=" + buf.length + ", off=" + off + ", len=" + len + ']');

         if (len == 0)
             return 0; // Fully read done: read zero bytes correctly.

         // Readable bytes in the file, starting from the specified position.
         long readable = this.len - pos;

         if (readable <= 0)
             return -1; // EOF.

         long startTime = System.nanoTime();

         if (readable < len)
             len = (int)readable; // Truncate expected length to available.

         assert len > 0;

         byte[] block = blockFragmentizerSafe(pos / blockSize);

         // Skip bytes to expected position.
         int blockOff = (int)(pos % blockSize);

         len = Math.min(len, block.length - blockOff);

         U.arrayCopy(block, blockOff, buf, off, len);

         bytes += len;
         time += System.nanoTime() - startTime;

         return len;
     }

     /**
      * Method to safely retrieve file block. In case if file block is missing this method will check file map
      * and update file info. This may be needed when file that we are reading is concurrently fragmented.
      *
      * @param blockIdx Block index to read.
      * @return Block data.
      * @throws IOException If read failed.
      */
     private byte[] blockFragmentizerSafe(long blockIdx) throws IOException {
         try {
             try {
                 return block(blockIdx);
             }
             catch (IgfsCorruptedFileException e) {
                 if (log.isDebugEnabled())
                     log.debug("Failed to fetch file block [path=" + path + ", fileInfo=" + fileInfo +
                         ", blockIdx=" + blockIdx + ", errMsg=" + e.getMessage() + ']');

                 // This failure may be caused by file being fragmented.
                 if (fileInfo != null && fileInfo.fileMap() != null && !fileInfo.fileMap().ranges().isEmpty()) {
                     IgfsEntryInfo newInfo = igfsCtx.meta().info(fileInfo.id());

                     // File was deleted.
                     if (newInfo == null)
                         throw new IgfsPathNotFoundException("Failed to read file block (file was concurrently " +
                                 "deleted) [path=" + path + ", blockIdx=" + blockIdx + ']');

                     fileInfo = newInfo;

                     // Must clear cache as it may have failed futures.
                     locCache.clear();

                     if (log.isDebugEnabled())
                         log.debug("Updated input stream file info after block fetch failure [path=" + path
                             + ", fileInfo=" + fileInfo + ']');

                     return block(blockIdx);
                 }

                 throw new IOException(e.getMessage(), e);
             }
         }
         catch (IgniteCheckedException e) {
             throw new IOException(e.getMessage(), e);
         }
     }

     /**
      * @param blockIdx Block index.
      * @return File block data.
      * @throws IOException If failed.
      * @throws IgniteCheckedException If failed.
      */
     private byte[] block(long blockIdx) throws IOException, IgniteCheckedException {
         assert blockIdx >= 0;

         IgniteInternalFuture<byte[]> bytesFut = locCache.get(blockIdx);

         if (bytesFut == null) {
             if (closed)
                 throw new IOException("Stream is already closed: " + this);

             seqReads = (prevBlockIdx != -1 && prevBlockIdx + 1 == blockIdx) ? ++seqReads : 0;

             prevBlockIdx = blockIdx;

             bytesFut = dataBlock(blockIdx);

             assert bytesFut != null;

             addLocalCacheFuture(blockIdx, bytesFut);
         }

         // Schedule the next block(s) prefetch.
         if (prefetchBlocks > 0 && seqReads >= seqReadsBeforePrefetch - 1) {
             for (int i = 1; i <= prefetchBlocks; i++) {
                 // Ensure that we do not prefetch over file size.
                 if (blockSize * (i + blockIdx) >= len)
                     break;
                 else if (locCache.get(blockIdx + i) == null)
                     addLocalCacheFuture(blockIdx + i, dataBlock(blockIdx + i));
             }
         }

         byte[] bytes = bytesFut.get();

         if (bytes == null)
             throw new IgfsCorruptedFileException("Failed to retrieve file's data block (corrupted file?) " +
                 "[path=" + path + ", blockIdx=" + blockIdx + ']');

         int blockSize0 = blockSize;

         if (blockIdx == blocksCnt - 1)
             blockSize0 = (int)(len % blockSize0);

         // If part of the file was reserved for writing, but was not actually written.
         if (bytes.length < blockSize0)
             throw new IOException("Inconsistent file's data block (incorrectly written?)" +
                 " [path=" + path + ", blockIdx=" + blockIdx + ", blockSize=" + bytes.length +
                 ", expectedBlockSize=" + blockSize0 + ", fileBlockSize=" + blockSize +
                 ", fileLen=" + len + ']');

         return bytes;
     }

     /**
      * Add local cache future.
      *
      * @param idx Block index.
      * @param fut Future.
      */
     private void addLocalCacheFuture(long idx, IgniteInternalFuture<byte[]> fut) {
         assert Thread.holdsLock(this);

         if (!locCache.containsKey(idx)) {
             if (locCache.size() == maxLocCacheSize) {
                 final IgniteInternalFuture<byte[]> evictFut = locCache.remove(locCache.keySet().iterator().next());

                 if (!evictFut.isDone()) {
                     pendingFuts.add(evictFut);

                     evictFut.listen(new IgniteInClosure<IgniteInternalFuture<byte[]>>() {
                         @Override
                         public void apply(IgniteInternalFuture<byte[]> t) {
                             pendingFuts.remove(evictFut);

                             pendingFutsLock.lock();

                             try {
                                 pendingFutsCond.signalAll();
                             } finally {
                                 pendingFutsLock.unlock();
                             }
                         }
                     });
                 }
             }

             locCache.put(idx, fut);
         }
     }

     /**
      * Get data block for specified block index.
      *
      * @param blockIdx Block index.
      * @return Requested data block or {@code null} if nothing found.
      * @throws IgniteCheckedException If failed.
      */
     @Nullable protected IgniteInternalFuture<byte[]> dataBlock(final long blockIdx)
         throws IgniteCheckedException {
         if (proxy) {
             assert secReader != null;

             final GridFutureAdapter<byte[]> fut = new GridFutureAdapter<>();

             igfsCtx.runInIgfsThreadPool(new Runnable() {
                 @Override public void run() {
                     try {
                         fut.onDone(igfsCtx.data().secondaryDataBlock(path, blockIdx, secReader, blockSize));
                     }
                     catch (Throwable e) {
                         fut.onDone(null, e);
                     }
                 }
             });

             return fut;
         }
         else {
             assert fileInfo != null;

             return igfsCtx.data().dataBlock(fileInfo, path, blockIdx, secReader);
         }
     }

     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(IgfsInputStreamImpl.class, this);
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.ignite.internal.processors.igfs;

	import org.apache.ignite.IgniteCheckedException;
	import org.apache.ignite.IgniteLogger;
	import org.apache.ignite.events.IgfsEvent;
	import org.apache.ignite.igfs.IgfsCorruptedFileException;
	import org.apache.ignite.igfs.IgfsInputStream;
	import org.apache.ignite.igfs.IgfsPath;
	import org.apache.ignite.igfs.IgfsPathNotFoundException;
	import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystemPositionedReadable;
	import org.apache.ignite.internal.IgniteInternalFuture;
	import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
	import org.apache.ignite.internal.util.GridConcurrentHashSet;
	import org.apache.ignite.internal.util.future.GridFutureAdapter;
	import org.apache.ignite.internal.util.typedef.internal.S;
	import org.apache.ignite.internal.util.typedef.internal.U;
	import org.apache.ignite.lang.IgniteInClosure;
	import org.jetbrains.annotations.NotNull;
	import org.jetbrains.annotations.Nullable;

	import java.io.EOFException;
	import java.io.IOException;
	import java.util.Arrays;
	import java.util.LinkedHashMap;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.locks.Condition;
	import java.util.concurrent.locks.Lock;
	import java.util.concurrent.locks.ReentrantLock;

	import static org.apache.ignite.events.EventType.EVT_IGFS_FILE_CLOSED_READ;

	/**
	* Input stream to read data from grid cache with separate blocks.
	*/
	public class IgfsInputStreamImpl extends IgfsInputStream implements IgfsSecondaryFileSystemPositionedReadable {
	/** Empty chunks result. */
	private static final byte[][] EMPTY_CHUNKS = new byte[0][];

	/** IGFS context. */
	private final IgfsContext igfsCtx;

	/** Secondary file system reader. */
	@SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
	private final IgfsSecondaryFileSystemPositionedReadable secReader;

	/** Logger. */
	private IgniteLogger log;

	/** Path to file. */
	protected final IgfsPath path;

	/** File descriptor. */
	private volatile IgfsEntryInfo fileInfo;

	/** The number of already read bytes. Important! Access to the property is guarded by this object lock. */
	private long pos;

	/** Local cache. */
	private final Map<Long, IgniteInternalFuture<byte[]>> locCache;

	/** Maximum local cache size. */
	private final int maxLocCacheSize;

	/** Pending data read futures which were evicted from the local cache before completion. */
	private final Set<IgniteInternalFuture<byte[]>> pendingFuts;

	/** Pending futures lock. */
	private final Lock pendingFutsLock = new ReentrantLock();

	/** Pending futures condition. */
	private final Condition pendingFutsCond = pendingFutsLock.newCondition();

	/** Closed flag. */
	private boolean closed;

	/** Number of blocks to prefetch asynchronously. */
	private int prefetchBlocks;

	/** Numbed of blocks that must be read sequentially before prefetch is triggered. */
	private int seqReadsBeforePrefetch;

	/** Bytes read. */
	private long bytes;

	/** Index of the previously read block. Initially it is set to -1 indicating that no reads has been made so far. */
	private long prevBlockIdx = -1;

	/** Amount of sequential reads performed. */
	private int seqReads;

	/** Time consumed on reading. */
	private long time;

	/** File Length. */
	private long len;

	/** Block size to read. */
	private int blockSize;

	/** Block size to read. */
	private long blocksCnt;

	/** Proxy mode. */
	private boolean proxy;

	/**
	* Constructs file output stream.
	* @param igfsCtx IGFS context.
	* @param path Path to stored file.
	* @param fileInfo File info to write binary data to.
	* @param prefetchBlocks Number of blocks to prefetch.
	* @param seqReadsBeforePrefetch Amount of sequential reads before prefetch is triggered.
	* @param secReader Optional secondary file system reader.
	* @param len File length.
	* @param blockSize Block size.
	* @param blocksCnt Blocks count.
	* @param proxy Proxy mode flag.
	*/
	IgfsInputStreamImpl(
	IgfsContext igfsCtx,
	IgfsPath path,
	@Nullable IgfsEntryInfo fileInfo,
	int prefetchBlocks,
	int seqReadsBeforePrefetch,
	@Nullable IgfsSecondaryFileSystemPositionedReadable secReader,
	long len,
	int blockSize,
	long blocksCnt,
	boolean proxy) {
	assert igfsCtx != null;
	assert path != null;

	this.igfsCtx = igfsCtx;
	this.path = path;
	this.fileInfo = fileInfo;
	this.prefetchBlocks = prefetchBlocks;
	this.seqReadsBeforePrefetch = seqReadsBeforePrefetch;
	this.secReader = secReader;
	this.len = len;
	this.blockSize = blockSize;
	this.blocksCnt = blocksCnt;
	this.proxy = proxy;

	log = igfsCtx.kernalContext().log(IgfsInputStream.class);

	maxLocCacheSize = (prefetchBlocks > 0 ? prefetchBlocks : 1) * 3 / 2;

	locCache = new LinkedHashMap<>(maxLocCacheSize, 1.0f);

	pendingFuts = new GridConcurrentHashSet<>(prefetchBlocks > 0 ? prefetchBlocks : 1);

	igfsCtx.metrics().incrementFilesOpenedForRead();
	}

	/**
	* Gets bytes read.
	*
	* @return Bytes read.
	*/
	public synchronized long bytes() {
	return bytes;
	}

	/** {@inheritDoc} */
	@Override public long length() {
	return len;
	}

	/** {@inheritDoc} */
	@Override public synchronized int read() throws IOException {
	byte[] buf = new byte[1];

	int read = read(buf, 0, 1);

	if (read == -1)
	return -1; // EOF.

	return buf[0] & 0xFF; // Cast to int and cut to unsigned byte value.
	}

	/** {@inheritDoc} */
	@Override public synchronized int read(@NotNull byte[] b, int off, int len) throws IOException {
	int read = readFromStore(pos, b, off, len);

	if (read != -1)
	pos += read;

	return read;
	}

	/** {@inheritDoc} */
	@Override public synchronized void seek(long pos) throws IOException {
	if (pos < 0)
	throw new IOException("Seek position cannot be negative: " + pos);


	this.pos = pos;
	}

	/** {@inheritDoc} */
	@Override public synchronized long position() throws IOException {
	return pos;
	}

	/** {@inheritDoc} */
	@Override public synchronized int available() throws IOException {
	long l = len - pos;

	if (l < 0)
	return 0;

	if (l > Integer.MAX_VALUE)
	return Integer.MAX_VALUE;

	return (int)l;
	}

	/** {@inheritDoc} */
	@Override public synchronized void readFully(long pos, byte[] buf) throws IOException {
	readFully(pos, buf, 0, buf.length);
	}

	/** {@inheritDoc} */
	@Override public synchronized void readFully(long pos, byte[] buf, int off, int len) throws IOException {
	for (int readBytes = 0; readBytes < len; ) {
	int read = readFromStore(pos + readBytes, buf, off + readBytes, len - readBytes);

	if (read == -1)
	throw new EOFException("Failed to read stream fully (stream ends unexpectedly)" +
	"[pos=" + pos + ", buf.length=" + buf.length + ", off=" + off + ", len=" + len + ']');

	readBytes += read;
	}
	}

	/** {@inheritDoc} */
	@Override public synchronized int read(long pos, byte[] buf, int off, int len) throws IOException {
	return readFromStore(pos, buf, off, len);
	}

	/**
	* Reads bytes from given position.
	*
	* @param pos Position to read from.
	* @param len Number of bytes to read.
	* @return Array of chunks with respect to chunk file representation.
	* @throws IOException If read failed.
	*/
	@SuppressWarnings("IfMayBeConditional")
	public synchronized byte[][] readChunks(long pos, int len) throws IOException {
	// Readable bytes in the file, starting from the specified position.
	long readable = this.len - pos;

	if (readable <= 0)
	return EMPTY_CHUNKS;

	long startTime = System.nanoTime();

	if (readable < len)
	len = (int)readable; // Truncate expected length to available.

	assert len > 0;

	bytes += len;

	int start = (int)(pos / blockSize);
	int end = (int)((pos + len - 1) / blockSize);

	int chunkCnt = end - start + 1;

	byte[][] chunks = new byte[chunkCnt][];

	for (int i = 0; i < chunkCnt; i++) {
	byte[] block = blockFragmentizerSafe(start + i);

	int blockOff = (int)(pos % blockSize);
	int blockLen = Math.min(len, block.length - blockOff);

	// If whole block can be used as result, do not do array copy.
	if (blockLen == block.length)
	chunks[i] = block;
	else {
	// Only first or last block can have non-full data.
	assert i == 0 \|\| i == chunkCnt - 1;

	chunks[i] = Arrays.copyOfRange(block, blockOff, blockOff + blockLen);
	}

	len -= blockLen;
	pos += blockLen;
	}

	assert len == 0;

	time += System.nanoTime() - startTime;

	return chunks;
	}

	/** {@inheritDoc} */
	@Override public synchronized void close() throws IOException {
	if (!closed) {
	try {
	if (secReader != null) {
	// Close secondary input stream.
	secReader.close();

	// Ensuring local cache futures completion.
	for (IgniteInternalFuture<byte[]> fut : locCache.values()) {
	try {
	fut.get();
	}
	catch (IgniteCheckedException ignore) {
	// No-op.
	}
	}

	// Ensuring pending evicted futures completion.
	while (!pendingFuts.isEmpty()) {
	pendingFutsLock.lock();

	try {
	pendingFutsCond.await(100, TimeUnit.MILLISECONDS);
	}
	catch (InterruptedException ignore) {
	// No-op.
	}
	finally {
	pendingFutsLock.unlock();
	}
	}
	}
	}
	catch (Exception e) {
	throw new IOException("File to close the file: " + path, e);
	}
	finally {
	closed = true;

	IgfsLocalMetrics metrics = igfsCtx.metrics();

	metrics.addReadBytesTime(bytes, time);
	metrics.decrementFilesOpenedForRead();

	locCache.clear();

	GridEventStorageManager evts = igfsCtx.kernalContext().event();

	if (evts.isRecordable(EVT_IGFS_FILE_CLOSED_READ))
	evts.record(new IgfsEvent(path, igfsCtx.localNode(), EVT_IGFS_FILE_CLOSED_READ, bytes()));
	}
	}
	}

	/**
	* @param pos Position to start reading from.
	* @param buf Data buffer to save read data to.
	* @param off Offset in the buffer to write data from.
	* @param len Length of the data to read from the stream.
	* @return Number of actually read bytes.
	* @throws IOException In case of any IO exception.
	*/
	private int readFromStore(long pos, byte[] buf, int off, int len) throws IOException {
	if (pos < 0)
	throw new IllegalArgumentException("Read position cannot be negative: " + pos);

	if (buf == null)
	throw new NullPointerException("Destination buffer cannot be null.");

	if (off < 0 \|\| len < 0 \|\| buf.length < len + off)
	throw new IndexOutOfBoundsException("Invalid buffer boundaries " +
	"[buf.length=" + buf.length + ", off=" + off + ", len=" + len + ']');

	if (len == 0)
	return 0; // Fully read done: read zero bytes correctly.

	// Readable bytes in the file, starting from the specified position.
	long readable = this.len - pos;

	if (readable <= 0)
	return -1; // EOF.

	long startTime = System.nanoTime();

	if (readable < len)
	len = (int)readable; // Truncate expected length to available.

	assert len > 0;

	byte[] block = blockFragmentizerSafe(pos / blockSize);

	// Skip bytes to expected position.
	int blockOff = (int)(pos % blockSize);

	len = Math.min(len, block.length - blockOff);

	U.arrayCopy(block, blockOff, buf, off, len);

	bytes += len;
	time += System.nanoTime() - startTime;

	return len;
	}

	/**
	* Method to safely retrieve file block. In case if file block is missing this method will check file map
	* and update file info. This may be needed when file that we are reading is concurrently fragmented.
	*
	* @param blockIdx Block index to read.
	* @return Block data.
	* @throws IOException If read failed.
	*/
	private byte[] blockFragmentizerSafe(long blockIdx) throws IOException {
	try {
	try {
	return block(blockIdx);
	}
	catch (IgfsCorruptedFileException e) {
	if (log.isDebugEnabled())
	log.debug("Failed to fetch file block [path=" + path + ", fileInfo=" + fileInfo +
	", blockIdx=" + blockIdx + ", errMsg=" + e.getMessage() + ']');

	// This failure may be caused by file being fragmented.
	if (fileInfo != null && fileInfo.fileMap() != null && !fileInfo.fileMap().ranges().isEmpty()) {
	IgfsEntryInfo newInfo = igfsCtx.meta().info(fileInfo.id());

	// File was deleted.
	if (newInfo == null)
	throw new IgfsPathNotFoundException("Failed to read file block (file was concurrently " +
	"deleted) [path=" + path + ", blockIdx=" + blockIdx + ']');

	fileInfo = newInfo;

	// Must clear cache as it may have failed futures.
	locCache.clear();

	if (log.isDebugEnabled())
	log.debug("Updated input stream file info after block fetch failure [path=" + path
	+ ", fileInfo=" + fileInfo + ']');

	return block(blockIdx);
	}

	throw new IOException(e.getMessage(), e);
	}
	}
	catch (IgniteCheckedException e) {
	throw new IOException(e.getMessage(), e);
	}
	}

	/**
	* @param blockIdx Block index.
	* @return File block data.
	* @throws IOException If failed.
	* @throws IgniteCheckedException If failed.
	*/
	private byte[] block(long blockIdx) throws IOException, IgniteCheckedException {
	assert blockIdx >= 0;

	IgniteInternalFuture<byte[]> bytesFut = locCache.get(blockIdx);

	if (bytesFut == null) {
	if (closed)
	throw new IOException("Stream is already closed: " + this);

	seqReads = (prevBlockIdx != -1 && prevBlockIdx + 1 == blockIdx) ? ++seqReads : 0;

	prevBlockIdx = blockIdx;

	bytesFut = dataBlock(blockIdx);

	assert bytesFut != null;

	addLocalCacheFuture(blockIdx, bytesFut);
	}

	// Schedule the next block(s) prefetch.
	if (prefetchBlocks > 0 && seqReads >= seqReadsBeforePrefetch - 1) {
	for (int i = 1; i <= prefetchBlocks; i++) {
	// Ensure that we do not prefetch over file size.
	if (blockSize * (i + blockIdx) >= len)
	break;
	else if (locCache.get(blockIdx + i) == null)
	addLocalCacheFuture(blockIdx + i, dataBlock(blockIdx + i));
	}
	}

	byte[] bytes = bytesFut.get();

	if (bytes == null)
	throw new IgfsCorruptedFileException("Failed to retrieve file's data block (corrupted file?) " +
	"[path=" + path + ", blockIdx=" + blockIdx + ']');

	int blockSize0 = blockSize;

	if (blockIdx == blocksCnt - 1)
	blockSize0 = (int)(len % blockSize0);

	// If part of the file was reserved for writing, but was not actually written.
	if (bytes.length < blockSize0)
	throw new IOException("Inconsistent file's data block (incorrectly written?)" +
	" [path=" + path + ", blockIdx=" + blockIdx + ", blockSize=" + bytes.length +
	", expectedBlockSize=" + blockSize0 + ", fileBlockSize=" + blockSize +
	", fileLen=" + len + ']');

	return bytes;
	}

	/**
	* Add local cache future.
	*
	* @param idx Block index.
	* @param fut Future.
	*/
	private void addLocalCacheFuture(long idx, IgniteInternalFuture<byte[]> fut) {
	assert Thread.holdsLock(this);

	if (!locCache.containsKey(idx)) {
	if (locCache.size() == maxLocCacheSize) {
	final IgniteInternalFuture<byte[]> evictFut = locCache.remove(locCache.keySet().iterator().next());

	if (!evictFut.isDone()) {
	pendingFuts.add(evictFut);

	evictFut.listen(new IgniteInClosure<IgniteInternalFuture<byte[]>>() {
	@Override
	public void apply(IgniteInternalFuture<byte[]> t) {
	pendingFuts.remove(evictFut);

	pendingFutsLock.lock();

	try {
	pendingFutsCond.signalAll();
	} finally {
	pendingFutsLock.unlock();
	}
	}
	});
	}
	}

	locCache.put(idx, fut);
	}
	}

	/**
	* Get data block for specified block index.
	*
	* @param blockIdx Block index.
	* @return Requested data block or {@code null} if nothing found.
	* @throws IgniteCheckedException If failed.
	*/
	@Nullable protected IgniteInternalFuture<byte[]> dataBlock(final long blockIdx)
	throws IgniteCheckedException {
	if (proxy) {
	assert secReader != null;

	final GridFutureAdapter<byte[]> fut = new GridFutureAdapter<>();

	igfsCtx.runInIgfsThreadPool(new Runnable() {
	@Override public void run() {
	try {
	fut.onDone(igfsCtx.data().secondaryDataBlock(path, blockIdx, secReader, blockSize));
	}
	catch (Throwable e) {
	fut.onDone(null, e);
	}
	}
	});

	return fut;
	}
	else {
	assert fileInfo != null;

	return igfsCtx.data().dataBlock(fileInfo, path, blockIdx, secReader);
	}
	}

	/** {@inheritDoc} */
	@Override public String toString() {
	return S.toString(IgfsInputStreamImpl.class, this);
	}
	}