content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.underfs.cephfs;
  
  import com.ceph.fs.CephMount;
  
  import java.io.IOException;
  import java.io.InputStream;
  import java.util.concurrent.atomic.AtomicBoolean;
  import javax.annotation.concurrent.NotThreadSafe;
  
  /**
   * A stream for reading a file from CephFS.
   */
  @NotThreadSafe
  public final class CephInputStream extends InputStream {
    private CephMount mMount;
    private final int mFileHandle;
    private final long mFileLength;
    private long mCephPos = 0;
    private byte[] mBuf;
    private int mBufPos = 0;
    private int mBufValid = 0;
    private byte[] mOneByteBuf = new byte[1];
    /** Flag to indicate this stream has been closed, to ensure close is only done once. */
    private final AtomicBoolean mClosed = new AtomicBoolean(false);
  
    /**
     * Create a new CephInputStream.
     *
     * @param mount CephFileSystem mount
     * @param fh The filehandle provided by Ceph to reference
     * @param flength The current length of the file. If the length changes
     *                you will need to close and re-open it to access the new data
     */
    public CephInputStream(CephMount mount, int fh, long flength) {
      mFileLength = flength;
      mFileHandle = fh;
      mMount = mount;
      mBuf = new byte[1 << 21];
    }
  
    /**
     * Gets the current position of the stream.
     *
     * @return current positon of the stream
     */
    public synchronized long getPos() throws IOException {
      return mCephPos - mBufValid + mBufPos;
    }
  
    @Override
    public synchronized int available() throws IOException {
      if (mClosed.get()) {
        throw new IOException("file is closed");
      }
      return (int) (mFileLength - getPos());
    }
  
    /**
     * Seek.
     *
     * @param targetPos Position
     * @throws IOException throws
     */
    public synchronized void seek(long targetPos) throws IOException {
      if (targetPos > mFileLength) {
        throw new IOException(
          String.format("CephInputStream.seek: failed seek to position %d "
            + "on fd %d : Cannot seek after EOF fileSize %d",
            targetPos, mFileHandle, mFileLength));
      }
  
      long oldPos = mCephPos;
      mCephPos = mMount.lseek(mFileHandle, targetPos, CephMount.SEEK_SET);
      if (mCephPos < 0) {
        int ret = (int) mCephPos;
        mCephPos = oldPos;
        throw new IOException(String.format("ceph.lseek: failed ret = %d", ret));
      }
      mBufValid = 0;
      mBufPos = 0;
    }
  
    @Override
    public synchronized long skip(long n) throws IOException {
      if (n <= 0) {
        return 0;
      }
      long targetPos = getPos() + n;
      seek(targetPos);
      return n;
    }
  
    @Override
    public synchronized int read() throws IOException {
      if (getPos() >= mFileLength) {
        return -1;
      }
      if (-1 == read(mOneByteBuf, 0, 1)) {
        return -1;
      }
      if (mOneByteBuf[0] < 0) {
        return 256 + (int) mOneByteBuf[0];
      } else {
        return mOneByteBuf[0];
      }
    }
  
    @Override
    public synchronized int read(byte[] buf, int off, int len) throws IOException {
      if (mClosed.get()) {
        throw new IOException(String.format("CephInputStream.read: cannot read %d "
          + "bytes from fd %d : stream closed",
          len, mFileHandle));
      }
      if (buf == null) {
        throw new NullPointerException();
      } else if (off < 0 || len < 0 || len > buf.length - off) {
        throw new IndexOutOfBoundsException();
      } else if (len == 0) {
        return 0;
      }
  
      // ensure we're not past the end of the file
      if (getPos() >= mFileLength) {
        return -1;
      }
      int totalRead = 0;
      int initialLen = len;
      int read;
  
      do {
        read = Math.min(len, mBufValid - mBufPos);
        try {
          System.arraycopy(mBuf, mBufPos, buf, off, read);
        } catch (IndexOutOfBoundsException ie) {
          throw new IOException(String.format("CephInputStream.read: Indices out of bounds: "
            + "read length is %d, buffer offset is %d, and buffer size is %d",
            len, off, buf.length));
        } catch (ArrayStoreException ae) {
          throw new IOException(String.format("CephInputStream.read: failed to do an array"
            + "copy due to type mismatch..."));
        } catch (NullPointerException ne) {
          throw new IOException(String.format("CephInputStream.read: cannot read %d "
            + "bytes from fd %d: mBuf is null",
            len, mFileHandle));
        }
        mBufPos += read;
        len -= read;
        off += read;
        totalRead += read;
      } while (len > 0 && fillBuffer());
  
      return totalRead;
    }
  
    @Override
    public void close() throws IOException {
      if (mClosed.getAndSet(true)) {
        return;
      }
      mMount.close(mFileHandle);
    }
  
    private synchronized boolean fillBuffer() throws IOException {
      mBufValid = (int) mMount.read(mFileHandle, mBuf, mBuf.length, -1);
      mBufPos = 0;
      if (mBufValid < 0) {
        int err = mBufValid;
        mBufValid = 0;
        // attempt to reset to old position. If it fails, too bad.
        mMount.lseek(mFileHandle, mCephPos, CephMount.SEEK_SET);
        throw new IOException(String.format("Failed to fill read buffer! Error code: %d", err));
      }
      mCephPos += mBufValid;
      return (mBufValid != 0);
    }
  }
  
  