alluxio-master-2.x/integration/fuse/src/main/java/alluxio/fuse/file/FuseFileOutStream.java
/*
 * 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.fuse.file;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.URIStatus;
import alluxio.concurrent.LockMode;
import alluxio.exception.PreconditionMessage;
import alluxio.exception.runtime.AlluxioRuntimeException;
import alluxio.exception.runtime.AlreadyExistsRuntimeException;
import alluxio.exception.runtime.FailedPreconditionRuntimeException;
import alluxio.exception.runtime.UnimplementedRuntimeException;
import alluxio.fuse.AlluxioFuseOpenUtils;
import alluxio.fuse.AlluxioFuseUtils;
import alluxio.fuse.auth.AuthPolicy;
import alluxio.fuse.lock.FuseReadWriteLockManager;
import alluxio.resource.CloseableResource;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.locks.Lock;
import javax.annotation.concurrent.ThreadSafe;

/**
 * An implementation for {@link FuseFileStream} for sequential write only operations
 * against an Alluxio uri.
 */
@ThreadSafe
public class FuseFileOutStream implements FuseFileStream {
  private static final Logger LOG = LoggerFactory.getLogger(FuseFileOutStream.class);
  private static final int DEFAULT_BUFFER_SIZE = Constants.MB * 4;
  private final AuthPolicy mAuthPolicy;
  private final FileSystem mFileSystem;
  private final CloseableResource<Lock> mLockResource;
  private final AlluxioURI mURI;
  private final CreateFileStatus mFileStatus;

  private volatile boolean mClosed = false;
  private Optional<FileOutStream> mOutStream;

  /**
   * Creates a {@link FuseFileInOrOutStream}.
   *
   * @param fileSystem the Alluxio file system
   * @param authPolicy the Authentication policy
   * @param lockManager the lock manager
   * @param uri the alluxio uri
   * @param flags the fuse create/open flags
   * @param mode the filesystem mode, -1 if not set
   * @return a {@link FuseFileInOrOutStream}
   */
  public static FuseFileOutStream create(FileSystem fileSystem, AuthPolicy authPolicy,
      FuseReadWriteLockManager lockManager, AlluxioURI uri, int flags, long mode) {
    Preconditions.checkNotNull(fileSystem);
    Preconditions.checkNotNull(authPolicy);
    Preconditions.checkNotNull(lockManager);
    Preconditions.checkNotNull(uri);
    // Make sure file is not being read/written by current FUSE
    CloseableResource<Lock> lockResource = lockManager.tryLock(uri.toString(), LockMode.WRITE);

    try {
      // Make sure file is not being written by other clients outside current FUSE
      Optional<URIStatus> status = AlluxioFuseUtils.getPathStatus(fileSystem, uri);
      if (status.isPresent() && !status.get().isCompleted()) {
        status = AlluxioFuseUtils.waitForFileCompleted(fileSystem, uri);
        if (!status.isPresent()) {
          throw new UnimplementedRuntimeException(String.format(
              "Failed to create fuse file out stream for %s: cannot concurrently write same file",
              uri));
        }
      }
      if (mode == AlluxioFuseUtils.MODE_NOT_SET_VALUE && status.isPresent()) {
        mode = status.get().getMode();
      }
      long fileLen = status.map(URIStatus::getLength).orElse(0L);
      CreateFileStatus createFileStatus = CreateFileStatus.create(authPolicy, mode, fileLen);
      if (status.isPresent()) {
        if (AlluxioFuseOpenUtils.containsTruncate(flags) || fileLen == 0) {
          // support OPEN(O_WRONLY | O_RDONLY) existing file + O_TRUNC to write
          // support create empty file then open for write/read_write workload
          AlluxioFuseUtils.deletePath(fileSystem, uri);
          createFileStatus.setFileLength(0L);
          if (LOG.isDebugEnabled()) {
            LOG.debug(String.format("Open path %s with flag 0x%x for overwriting. "
                + "Alluxio deleted the old file and created a new file for writing", uri, flags));
          }
        } else {
          // Support open(O_WRONLY | O_RDWR flag) - truncate(0) - write() workflow
          return new FuseFileOutStream(fileSystem, authPolicy, uri, createFileStatus, lockResource,
              Optional.empty());
        }
      }
      return new FuseFileOutStream(fileSystem, authPolicy, uri,
          createFileStatus, lockResource,
          Optional.of(AlluxioFuseUtils.createFile(fileSystem, authPolicy, uri, createFileStatus)));
    } catch (Throwable t) {
      lockResource.close();
      throw t;
    }
  }

  private FuseFileOutStream(FileSystem fileSystem, AuthPolicy authPolicy,
      AlluxioURI uri, CreateFileStatus fileStatus, CloseableResource<Lock> lockResource,
      Optional<FileOutStream> outStream) {
    mFileSystem = Preconditions.checkNotNull(fileSystem);
    mAuthPolicy = Preconditions.checkNotNull(authPolicy);
    mFileStatus = Preconditions.checkNotNull(fileStatus);
    mURI = Preconditions.checkNotNull(uri);
    mLockResource = Preconditions.checkNotNull(lockResource);
    mOutStream = Preconditions.checkNotNull(outStream);
  }

  @Override
  public int read(ByteBuffer buf, long size, long offset) {
    throw new FailedPreconditionRuntimeException("Cannot read from write only stream");
  }

  @Override
  public synchronized void write(ByteBuffer buf, long size, long offset) {
    Preconditions.checkArgument(size >= 0 && offset >= 0 && size <= buf.capacity(),
        PreconditionMessage.ERR_BUFFER_STATE.toString(), buf.capacity(), offset, size);
    if (!mOutStream.isPresent()) {
      throw new AlreadyExistsRuntimeException(
          "Cannot overwrite/extending existing file without O_TRUNC flag or truncate(0) operation");
    }
    if (size == 0) {
      return;
    }
    int sz = (int) size;
    long bytesWritten = mOutStream.get().getBytesWritten();
    if (offset != bytesWritten && offset + sz > bytesWritten) {
      throw new UnimplementedRuntimeException(String.format("Only sequential write is supported. "
          + "Cannot write bytes of size %s to offset %s when %s bytes have written to path %s",
          size, offset, bytesWritten, mURI));
    }
    if (offset + sz <= bytesWritten) {
      LOG.warn("Skip writing to file {} offset={} size={} when {} bytes has written to file",
          mURI, offset, sz, bytesWritten);
      // To fulfill vim :wq
    }
    final byte[] dest = new byte[sz];
    buf.get(dest, 0, sz);
    try {
      mOutStream.get().write(dest);
    } catch (IOException e) {
      throw AlluxioRuntimeException.from(e);
    }
  }

  @Override
  public synchronized FileStatus getFileStatus() {
    if (mOutStream.isPresent()) {
      if (mOutStream.get().getBytesWritten() > mFileStatus.getFileLength()) {
        mFileStatus.setFileLength(mOutStream.get().getBytesWritten());
      }
    }
    return mFileStatus;
  }

  @Override
  public synchronized void flush() {
    if (!mOutStream.isPresent()) {
      return;
    }
    try {
      mOutStream.get().flush();
    } catch (IOException e) {
      throw AlluxioRuntimeException.from(e);
    }
  }

  @Override
  public synchronized void truncate(long size) {
    long currentSize = getFileStatus().getFileLength();
    if (size == currentSize) {
      return;
    }
    if (size == 0) {
      closeStreams();
      AlluxioFuseUtils.deletePath(mFileSystem, mURI);
      mOutStream = Optional.of(AlluxioFuseUtils
          .createFile(mFileSystem, mAuthPolicy, mURI, mFileStatus));
      mFileStatus.setFileLength(0);
      return;
    }
    if (mOutStream.isPresent() && size >= mOutStream.get().getBytesWritten()) {
      // support setting file length to a value bigger than current file length
      // but do not support opening an existing file and append on top.
      // e.g. support "create() -> sequential write
      // -> truncate(to larger value) -> sequential write"
      // do not support "file exist -> open(W or RW) -> truncate(to a larger value)"
      mFileStatus.setFileLength(size);
      return;
    }
    throw new UnimplementedRuntimeException(
        String.format("Cannot truncate file %s from size %s to size %s", mURI, currentSize, size));
  }

  @Override
  public synchronized void close() {
    if (mClosed) {
      return;
    }
    mClosed = true;
    try {
      closeStreams();
    } finally {
      mLockResource.close();
    }
  }

  private void closeStreams() {
    try {
      writeToFileLengthIfNeeded();
      if (mOutStream.isPresent()) {
        mOutStream.get().close();
      }
    } catch (IOException e) {
      throw AlluxioRuntimeException.from(e);
    }
  }

  /**
   * Fills zero bytes to file if file length is set to a value larger
   * than bytes written by truncate() operation.
   */
  private void writeToFileLengthIfNeeded() throws IOException {
    if (!mOutStream.isPresent()) {
      return;
    }
    long bytesWritten = mOutStream.get().getBytesWritten();
    if (bytesWritten >= mFileStatus.getFileLength()) {
      return;
    }
    long bytesGap = mFileStatus.getFileLength() - bytesWritten;
    final long originalBytesGap = bytesGap;
    int bufferSize = bytesGap >= DEFAULT_BUFFER_SIZE
        ? DEFAULT_BUFFER_SIZE : (int) bytesGap;
    byte[] buffer = new byte[bufferSize];
    Arrays.fill(buffer, (byte) 0);
    while (bytesGap > 0) {
      int bytesToWrite = bytesGap >= DEFAULT_BUFFER_SIZE
          ? DEFAULT_BUFFER_SIZE : (int) bytesGap;
      mOutStream.get().write(buffer, 0, bytesToWrite);
      bytesGap -= DEFAULT_BUFFER_SIZE;
    }
    LOG.debug("Filled {} zero bytes to file {} to fulfill the extended file length of {}",
        originalBytesGap, mURI, mFileStatus.getFileLength());
  }
}
