<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.Configuration;
import alluxio.PropertyKey;
import alluxio.client.block.BlockMasterClient;
import alluxio.client.block.BlockMasterClientPool;
import alluxio.client.block.BlockWorkerClient;
import alluxio.client.block.BlockWorkerThriftClientPool;
import alluxio.client.block.RetryHandlingBlockWorkerClient;
import alluxio.client.netty.NettyClient;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.metrics.MetricsSystem;
import alluxio.network.connection.NettyChannelPool;
import alluxio.resource.CloseableResource;
import alluxio.util.IdUtils;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.wire.WorkerInfo;
import alluxio.wire.WorkerNetAddress;

import com.codahale.metrics.Gauge;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.util.internal.chmv8.ConcurrentHashMapV8;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import javax.security.auth.Subject;

/**
 * A shared context that isolates all operations within a {@link FileSystem}. Usually, one user
 * only needs one instance of {@link FileSystemContext}.
 *
 * <p>
 * NOTE: The context maintains a pool of file system master clients that is already thread-safe.
 * Synchronizing {@link FileSystemContext} methods could lead to deadlock: thread A attempts to
 * acquire a client when there are no clients left in the pool and blocks holding a lock on the
 * {@link FileSystemContext}, when thread B attempts to release a client it owns it is unable to do
 * so, because thread A holds the lock on {@link FileSystemContext}.
 */
@ThreadSafe
public final class FileSystemContext implements Closeable {
  public static final FileSystemContext INSTANCE = create(null);

  static {
    MetricsSystem.startSinks();
    Metrics.initializeGauges();
  }

  // Master client pools.
  private volatile FileSystemMasterClientPool mFileSystemMasterClientPool;
  private volatile BlockMasterClientPool mBlockMasterClientPool;

  // Block worker client pools.
  private final ConcurrentHashMapV8<InetSocketAddress, BlockWorkerThriftClientPool>
      mBlockWorkerClientPools = new ConcurrentHashMapV8<>();
  private final ConcurrentHashMapV8<InetSocketAddress, BlockWorkerThriftClientPool>
      mBlockWorkerHeartbeatClientPools = new ConcurrentHashMapV8<>();

  // The file system worker client pools.
  private final ConcurrentHashMapV8<InetSocketAddress, FileSystemWorkerThriftClientPool>
      mFileSystemWorkerClientPools = new ConcurrentHashMapV8<>();
  private final ConcurrentHashMapV8<InetSocketAddress, FileSystemWorkerThriftClientPool>
      mFileSystemWorkerClientHeartbeatPools = new ConcurrentHashMapV8<>();

  // The netty data server channel pools.
  private final ConcurrentHashMapV8<InetSocketAddress, NettyChannelPool>
      mNettyChannelPools = new ConcurrentHashMapV8<>();

  /** The shared master address associated with the {@link FileSystemContext}. */
  @GuardedBy("this")
  private InetSocketAddress mMasterAddress;

  /** A list of valid workers, if there is a local worker, only the local worker addresses. */
  @GuardedBy("this")
  private List<WorkerNetAddress> mWorkerAddresses;

  /**
   * Indicates whether there is any Alluxio worker running in the local machine. This is initialized
   * lazily.
   */
  @GuardedBy("this")
  private Boolean mHasLocalWorker;

  /** The parent user associated with the {@link FileSystemContext}. */
  private final Subject mParentSubject;

  /**
   * Creates a new file system context.
   *
   * @return the context
   */
  public static FileSystemContext create() {
    return create(null);
  }

  /**
   * Creates a file system context with a subject.
   *
   * @param subject the parent subject, set to null if not present
   * @return the context
   */
  public static FileSystemContext create(Subject subject) {
    FileSystemContext context = new FileSystemContext(subject);
    context.init();
    return context;
  }

  /**
   * Creates a file system context with a subject.
   *
   * @param subject the parent subject, set to null if not present
   */
  private FileSystemContext(Subject subject) {
    mParentSubject = subject;
  }

  /**
   * Initializes the context. Only called in the factory methods and reset.
   */
  private void init() {
    String masterHostName =
        Preconditions.checkNotNull(Configuration.get(PropertyKey.MASTER_HOSTNAME));
    int masterPort = Configuration.getInt(PropertyKey.MASTER_RPC_PORT);
    mMasterAddress = new InetSocketAddress(masterHostName, masterPort);

    mFileSystemMasterClientPool = new FileSystemMasterClientPool(mParentSubject, mMasterAddress);
    mBlockMasterClientPool = new BlockMasterClientPool(mParentSubject, mMasterAddress);
  }

  /**
   * Closes all the resources associated with the context. Make sure all the resources are released
   * back to this context before calling this close. After closing the context, all the resources
   * that acquired from this context might fail. Only call this when you are done with using
   * the {@link FileSystem} associated with this {@link FileSystemContext}.
   */
  public void close() {
    mFileSystemMasterClientPool.close();
    mFileSystemMasterClientPool = null;
    mBlockMasterClientPool.close();
    mBlockMasterClientPool = null;

    for (BlockWorkerThriftClientPool pool : mBlockWorkerClientPools.values()) {
      pool.close();
    }
    mBlockWorkerClientPools.clear();
    for (FileSystemWorkerThriftClientPool pool : mFileSystemWorkerClientPools.values()) {
      pool.close();
    }
    mFileSystemWorkerClientPools.clear();
    for (NettyChannelPool pool : mNettyChannelPools.values()) {
      pool.close();
    }
    mNettyChannelPools.clear();<commit_msg>small fix<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.Configuration;
import alluxio.PropertyKey;
import alluxio.client.block.BlockMasterClient;
import alluxio.client.block.BlockMasterClientPool;
import alluxio.client.block.BlockWorkerClient;
import alluxio.client.block.BlockWorkerThriftClientPool;
import alluxio.client.block.RetryHandlingBlockWorkerClient;
import alluxio.client.netty.NettyClient;
import alluxio.exception.AlluxioException;
import alluxio.exception.ExceptionMessage;
import alluxio.metrics.MetricsSystem;
import alluxio.network.connection.NettyChannelPool;
import alluxio.resource.CloseableResource;
import alluxio.util.IdUtils;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.wire.WorkerInfo;
import alluxio.wire.WorkerNetAddress;

import com.codahale.metrics.Gauge;
import com.google.common.base.Preconditions;
import com.google.common.base.Throwables;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.util.internal.chmv8.ConcurrentHashMapV8;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import javax.security.auth.Subject;

/**
 * A shared context that isolates all operations within a {@link FileSystem}. Usually, one user
 * only needs one instance of {@link FileSystemContext}.
 *
 * <p>
 * NOTE: The context maintains a pool of file system master clients that is already thread-safe.
 * Synchronizing {@link FileSystemContext} methods could lead to deadlock: thread A attempts to
 * acquire a client when there are no clients left in the pool and blocks holding a lock on the
 * {@link FileSystemContext}, when thread B attempts to release a client it owns it is unable to do
 * so, because thread A holds the lock on {@link FileSystemContext}.
 */
@ThreadSafe
public final class FileSystemContext implements Closeable {
  public static final FileSystemContext INSTANCE = create(null);

  static {
    MetricsSystem.startSinks();
    Metrics.initializeGauges();
  }

  // Master client pools.
  private volatile FileSystemMasterClientPool mFileSystemMasterClientPool;
  private volatile BlockMasterClientPool mBlockMasterClientPool;

  // Block worker client pools.
  private final ConcurrentHashMapV8<InetSocketAddress, BlockWorkerThriftClientPool>
      mBlockWorkerClientPools = new ConcurrentHashMapV8<>();
  private final ConcurrentHashMapV8<InetSocketAddress, BlockWorkerThriftClientPool>
      mBlockWorkerHeartbeatClientPools = new ConcurrentHashMapV8<>();

  // The file system worker client pools.
  private final ConcurrentHashMapV8<InetSocketAddress, FileSystemWorkerThriftClientPool>
      mFileSystemWorkerClientPools = new ConcurrentHashMapV8<>();
  private final ConcurrentHashMapV8<InetSocketAddress, FileSystemWorkerThriftClientPool>
      mFileSystemWorkerClientHeartbeatPools = new ConcurrentHashMapV8<>();

  // The netty data server channel pools.
  private final ConcurrentHashMapV8<InetSocketAddress, NettyChannelPool>
      mNettyChannelPools = new ConcurrentHashMapV8<>();

  /** The shared master address associated with the {@link FileSystemContext}. */
  @GuardedBy("this")
  private InetSocketAddress mMasterAddress;

  /** A list of valid workers, if there is a local worker, only the local worker addresses. */
  @GuardedBy("this")
  private List<WorkerNetAddress> mWorkerAddresses;

  /**
   * Indicates whether there is any Alluxio worker running in the local machine. This is initialized
   * lazily.
   */
  @GuardedBy("this")
  private Boolean mHasLocalWorker;

  /** The parent user associated with the {@link FileSystemContext}. */
  private final Subject mParentSubject;

  /**
   * Creates a new file system context.
   *
   * @return the context
   */
  public static FileSystemContext create() {
    return create(null);
  }

  /**
   * Creates a file system context with a subject.
   *
   * @param subject the parent subject, set to null if not present
   * @return the context
   */
  public static FileSystemContext create(Subject subject) {
    FileSystemContext context = new FileSystemContext(subject);
    context.init();
    return context;
  }

  /**
   * Creates a file system context with a subject.
   *
   * @param subject the parent subject, set to null if not present
   */
  private FileSystemContext(Subject subject) {
    mParentSubject = subject;
  }

  /**
   * Initializes the context. Only called in the factory methods and reset.
   */
  private void init() {
    String masterHostName =
        Preconditions.checkNotNull(Configuration.get(PropertyKey.MASTER_HOSTNAME));
    int masterPort = Configuration.getInt(PropertyKey.MASTER_RPC_PORT);
    mMasterAddress = new InetSocketAddress(masterHostName, masterPort);

    mFileSystemMasterClientPool = new FileSystemMasterClientPool(mParentSubject, mMasterAddress);
    mBlockMasterClientPool = new BlockMasterClientPool(mParentSubject, mMasterAddress);
  }

  /**
   * Closes all the resources associated with the context. Make sure all the resources are released
   * back to this context before calling this close. After closing the context, all the resources
   * that acquired from this context might fail. Only call this when you are done with using
   * the {@link FileSystem} associated with this {@link FileSystemContext}.
   */
  public void close() {
    mFileSystemMasterClientPool.close();
    mFileSystemMasterClientPool = null;
    mBlockMasterClientPool.close();
    mBlockMasterClientPool = null;

    for (BlockWorkerThriftClientPool pool : mBlockWorkerClientPools.values()) {
      pool.close();
    }
    mBlockWorkerClientPools.clear();

    for (BlockWorkerThriftClientPool pool : mBlockWorkerHeartbeatClientPools.values()) {
      pool.close();
    }
    mBlockWorkerHeartbeatClientPools.clear();

    for (FileSystemWorkerThriftClientPool pool : mFileSystemWorkerClientPools.values()) {
      pool.close();
    }
    mFileSystemWorkerClientPools.clear();

    for (FileSystemWorkerThriftClientPool pool : mFileSystemWorkerClientHeartbeatPools.values()) {
      pool.close();
    }
    mFileSystemWorkerClientHeartbeatPools.clear();

    for (NettyChannelPool pool : mNettyChannelPools.values()) {
      pool.close();
    }
    mNettyChannelPools.clear();
{'fraction_non_alphanumeric': 0.07799689785065367, 'fraction_numerical': 0.002363542359110717, 'mean_word_length': 4.627597672485453, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, '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': '6721120', 'n_tokens_mistral': 3675, 'n_tokens_neox': 3638, 'n_words': 1140}