<commit_before>/*
 * 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.client.file;

import alluxio.AlluxioURI;
import alluxio.Configuration;
import alluxio.Constants;
import alluxio.PropertyKey;
import alluxio.annotation.PublicApi;
import alluxio.client.AbstractOutStream;
import alluxio.client.AlluxioStorageType;
import alluxio.client.UnderStorageType;
import alluxio.client.block.BufferedBlockOutStream;
import alluxio.client.file.options.CancelUfsFileOptions;
import alluxio.client.file.options.CompleteFileOptions;
import alluxio.client.file.options.CompleteUfsFileOptions;
import alluxio.client.file.options.CreateUfsFileOptions;
import alluxio.client.file.options.OutStreamOptions;
import alluxio.client.file.policy.FileWriteLocationPolicy;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.PreconditionMessage;
import alluxio.metrics.MetricsSystem;
import alluxio.security.authorization.Permission;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.CreateOptions;
import alluxio.util.IdUtils;
import alluxio.util.io.PathUtils;
import alluxio.wire.WorkerNetAddress;

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

import java.io.IOException;
import java.io.OutputStream;
import java.util.LinkedList;
import java.util.List;

import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Provides a streaming API to write a file. This class wraps the BlockOutStreams for each of the
 * blocks in the file and abstracts the switching between streams. The backing streams can write to
 * Alluxio space in the local machine or remote machines. If the {@link UnderStorageType} is
 * {@link UnderStorageType#SYNC_PERSIST}, another stream will write the data to the under storage
 * system.
 */
@PublicApi
@NotThreadSafe
public class FileOutStream extends AbstractOutStream {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  private final long mBlockSize;
  protected final AlluxioStorageType mAlluxioStorageType;
  private final UnderStorageType mUnderStorageType;
  private final FileSystemContext mContext;
  private final UnderFileSystemFileOutStream.Factory mUnderOutStreamFactory;
  private final OutputStream mUnderStorageOutputStream;
  private final long mNonce;
  /** Whether this stream should delegate operations to the ufs to a worker. */
  private final boolean mUfsDelegation;
  /** The client to a file system worker, null if mUfsDelegation is false. */
  private final FileSystemWorkerClient mFileSystemWorkerClient;
  /** The worker file id for the ufs file, null if mUfsDelegation is false. */
  private final Long mUfsFileId;

  private String mUfsPath;
  private FileWriteLocationPolicy mLocationPolicy;

  protected boolean mCanceled;
  protected boolean mClosed;
  private boolean mShouldCacheCurrentBlock;
  protected BufferedBlockOutStream mCurrentBlockOutStream;
  protected List<BufferedBlockOutStream> mPreviousBlockOutStreams;

  protected final AlluxioURI mUri;

  /**
   * Creates a new file output stream.
   *
   * @param path the file path
   * @param options the client options
   * @throws IOException if an I/O error occurs
   */
  public FileOutStream(AlluxioURI path, OutStreamOptions options) throws IOException {
    this(path, options, FileSystemContext.INSTANCE, UnderFileSystemFileOutStream.Factory.get());
  }

  /**
   * Creates a new file output stream.
   *
   * @param path the file path
   * @param options the client options
   * @param context the file system context
   * @param underOutStreamFactory a factory for creating any necessary under storage out streams
   * @throws IOException if an I/O error occurs
   */
  public FileOutStream(AlluxioURI path, OutStreamOptions options, FileSystemContext context,
      UnderFileSystemFileOutStream.Factory underOutStreamFactory) throws IOException {
    mUri = Preconditions.checkNotNull(path);
    mNonce = IdUtils.getRandomNonNegativeLong();
    mBlockSize = options.getBlockSizeBytes();
    mAlluxioStorageType = options.getAlluxioStorageType();
    mUnderStorageType = options.getUnderStorageType();
    mContext = context;
    mUnderOutStreamFactory = underOutStreamFactory;
    mPreviousBlockOutStreams = new LinkedList<>();
    mUfsDelegation = Configuration.getBoolean(PropertyKey.USER_UFS_DELEGATION_ENABLED);
    if (mUnderStorageType.isSyncPersist()) {
      if (mUfsDelegation) {
        mFileSystemWorkerClient = mContext.createWorkerClient();
        try {
          Permission perm = options.getPermission();
          mUfsFileId =
              mFileSystemWorkerClient.createUfsFile(new AlluxioURI(mUfsPath),
                  CreateUfsFileOptions.defaults().setPermission(perm));
        } catch (AlluxioException e) {
          mFileSystemWorkerClient.close();
          throw new IOException(e);
        }
        mUnderStorageOutputStream = mUnderOutStreamFactory
            .create(mFileSystemWorkerClient.getWorkerDataServerAddress(), mUfsFileId);<commit_msg>set ufs path since client still needs this.<commit_after>/*
 * 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.client.file;

import alluxio.AlluxioURI;
import alluxio.Configuration;
import alluxio.Constants;
import alluxio.PropertyKey;
import alluxio.annotation.PublicApi;
import alluxio.client.AbstractOutStream;
import alluxio.client.AlluxioStorageType;
import alluxio.client.UnderStorageType;
import alluxio.client.block.BufferedBlockOutStream;
import alluxio.client.file.options.CancelUfsFileOptions;
import alluxio.client.file.options.CompleteFileOptions;
import alluxio.client.file.options.CompleteUfsFileOptions;
import alluxio.client.file.options.CreateUfsFileOptions;
import alluxio.client.file.options.OutStreamOptions;
import alluxio.client.file.policy.FileWriteLocationPolicy;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.PreconditionMessage;
import alluxio.metrics.MetricsSystem;
import alluxio.security.authorization.Permission;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.CreateOptions;
import alluxio.util.IdUtils;
import alluxio.util.io.PathUtils;
import alluxio.wire.WorkerNetAddress;

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

import java.io.IOException;
import java.io.OutputStream;
import java.util.LinkedList;
import java.util.List;

import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Provides a streaming API to write a file. This class wraps the BlockOutStreams for each of the
 * blocks in the file and abstracts the switching between streams. The backing streams can write to
 * Alluxio space in the local machine or remote machines. If the {@link UnderStorageType} is
 * {@link UnderStorageType#SYNC_PERSIST}, another stream will write the data to the under storage
 * system.
 */
@PublicApi
@NotThreadSafe
public class FileOutStream extends AbstractOutStream {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  private final long mBlockSize;
  protected final AlluxioStorageType mAlluxioStorageType;
  private final UnderStorageType mUnderStorageType;
  private final FileSystemContext mContext;
  private final UnderFileSystemFileOutStream.Factory mUnderOutStreamFactory;
  private final OutputStream mUnderStorageOutputStream;
  private final long mNonce;
  /** Whether this stream should delegate operations to the ufs to a worker. */
  private final boolean mUfsDelegation;
  /** The client to a file system worker, null if mUfsDelegation is false. */
  private final FileSystemWorkerClient mFileSystemWorkerClient;
  /** The worker file id for the ufs file, null if mUfsDelegation is false. */
  private final Long mUfsFileId;

  private String mUfsPath;
  private FileWriteLocationPolicy mLocationPolicy;

  protected boolean mCanceled;
  protected boolean mClosed;
  private boolean mShouldCacheCurrentBlock;
  protected BufferedBlockOutStream mCurrentBlockOutStream;
  protected List<BufferedBlockOutStream> mPreviousBlockOutStreams;

  protected final AlluxioURI mUri;

  /**
   * Creates a new file output stream.
   *
   * @param path the file path
   * @param options the client options
   * @throws IOException if an I/O error occurs
   */
  public FileOutStream(AlluxioURI path, OutStreamOptions options) throws IOException {
    this(path, options, FileSystemContext.INSTANCE, UnderFileSystemFileOutStream.Factory.get());
  }

  /**
   * Creates a new file output stream.
   *
   * @param path the file path
   * @param options the client options
   * @param context the file system context
   * @param underOutStreamFactory a factory for creating any necessary under storage out streams
   * @throws IOException if an I/O error occurs
   */
  public FileOutStream(AlluxioURI path, OutStreamOptions options, FileSystemContext context,
      UnderFileSystemFileOutStream.Factory underOutStreamFactory) throws IOException {
    mUri = Preconditions.checkNotNull(path);
    mNonce = IdUtils.getRandomNonNegativeLong();
    mBlockSize = options.getBlockSizeBytes();
    mAlluxioStorageType = options.getAlluxioStorageType();
    mUnderStorageType = options.getUnderStorageType();
    mContext = context;
    mUnderOutStreamFactory = underOutStreamFactory;
    mPreviousBlockOutStreams = new LinkedList<>();
    mUfsDelegation = Configuration.getBoolean(PropertyKey.USER_UFS_DELEGATION_ENABLED);
    if (mUnderStorageType.isSyncPersist()) {
      // Get the ufs path from the master.
      FileSystemMasterClient client = mContext.acquireMasterClient();
      try {
        mUfsPath = client.getStatus(mUri).getUfsPath();
      } catch (AlluxioException e) {
        throw new IOException(e);
      } finally {
        mContext.releaseMasterClient(client);
      }
      if (mUfsDelegation) {
        mFileSystemWorkerClient = mContext.createWorkerClient();
        try {
          Permission perm = options.getPermission();
          mUfsFileId =
              mFileSystemWorkerClient.createUfsFile(new AlluxioURI(mUfsPath),
                  CreateUfsFileOptions.defaults().setPermission(perm));
        } catch (AlluxioException e) {
          mFileSystemWorkerClient.close();
          throw new IOException(e);
        }
        mUnderStorageOutputStream = mUnderOutStreamFactory
            .create(mFileSystemWorkerClient.getWorkerDataServerAddress(), mUfsFileId);
{'fraction_non_alphanumeric': 0.0694300518134715, 'fraction_numerical': 0.0010362694300518134, 'mean_word_length': 4.7049261083743845, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2768612', 'n_tokens_mistral': 3156, 'n_tokens_neox': 3143, 'n_words': 929}