<commit_before>package tachyon.client;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import tachyon.Constants;
import tachyon.TachyonURI;
import tachyon.UnderFileSystem;
import tachyon.client.table.RawTable;
import tachyon.conf.CommonConf;
import tachyon.conf.UserConf;
import tachyon.master.MasterClient;
import tachyon.thrift.ClientBlockInfo;
import tachyon.thrift.ClientDependencyInfo;
import tachyon.thrift.ClientFileInfo;
import tachyon.thrift.ClientRawTableInfo;
import tachyon.thrift.ClientWorkerInfo;
import tachyon.util.CommonUtils;
import tachyon.worker.WorkerClient;

/**
 * Tachyon's user client API. It contains a MasterClient and several WorkerClients depending on how
 * many workers the client program is interacting with.
 */
public class TachyonFS extends AbstractTachyonFS {
  /**
   * Create a TachyonFS handler.
   * 
   * @param tachyonURI a Tachyon URI contains master address. e.g., tachyon://localhost:19998,
   *        tachyon://localhost:19998/ab/c.txt
   * @return the corresponding TachyonFS handler
   * @throws IOException
   */
  public static synchronized TachyonFS get(final TachyonURI tachyonURI) throws IOException {
    if (tachyonURI == null) {
      throw new IOException("Tachyon Uri cannot be null. Use " + Constants.HEADER + "host:port/ ,"
          + Constants.HEADER_FT + "host:port/");
    } else {
      String scheme = tachyonURI.getScheme();
      if (scheme == null || tachyonURI.getHost() == null || tachyonURI.getPort() == -1
          || (!Constants.SCHEME.equals(scheme) && !Constants.SCHEME_FT.equals(scheme))) {
        throw new IOException("Invalid Tachyon URI: " + tachyonURI + ". Use " + Constants.HEADER
            + "host:port/ ," + Constants.HEADER_FT + "host:port/");
      }
      return new TachyonFS(tachyonURI);
    }
  }

  /**
   * Create a TachyonFS handler.
   * 
   * @param masterHost master host details
   * @param masterPort port master listens on
   * @param zookeeperMode use zookeeper
   * 
   * @return the corresponding TachyonFS hanlder
   * @throws IOException
   */
  public static synchronized TachyonFS get(String masterHost, int masterPort, boolean zookeeperMode)
      throws IOException {
    return new TachyonFS(new InetSocketAddress(masterHost, masterPort), zookeeperMode);
  }

  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
  private final long mUserQuotaUnitBytes = UserConf.get().QUOTA_UNIT_BYTES;
  private final int mUserFailedSpaceRequestLimits = UserConf.get().FAILED_SPACE_REQUEST_LIMITS;

  // The RPC client talks to the system master.
  private MasterClient mMasterClient = null;
  // The Master address.
  private InetSocketAddress mMasterAddress = null;
  // The RPC client talks to the local worker if there is one.
  private WorkerClient mWorkerClient = null;
  // Whether use ZooKeeper or not
  private boolean mZookeeperMode;
  // Cached ClientFileInfo
  private Map<String, ClientFileInfo> mPathToClientFileInfo = new HashMap<String, ClientFileInfo>();
  private Map<Integer, ClientFileInfo> mIdToClientFileInfo = new HashMap<Integer, ClientFileInfo>();

  private UnderFileSystem mUnderFileSystem = null;

  // All Blocks has been locked.
  private Map<Long, Set<Integer>> mLockedBlockIds = new HashMap<Long, Set<Integer>>();

  // Each user facing block has a unique block lock id.
  private AtomicInteger mBlockLockId = new AtomicInteger(0);

  // Available memory space for this client.
  private Long mAvailableSpaceBytes;

  private TachyonFS(TachyonURI tachyonURI) throws IOException {
    this(new InetSocketAddress(tachyonURI.getHost(), tachyonURI.getPort()), tachyonURI.getScheme()
        .equals(Constants.SCHEME_FT));
  }

  private TachyonFS(InetSocketAddress masterAddress, boolean zookeeperMode) throws IOException {
    mMasterAddress = masterAddress;
    mZookeeperMode = zookeeperMode;
    mAvailableSpaceBytes = 0L;

    mMasterClient = new MasterClient(mMasterAddress, mZookeeperMode);
    mWorkerClient = new WorkerClient(mMasterClient);
  }

  /**
   * Update the latest block access time on the worker.
   * 
   * @param blockId the local block's id
   * @throws IOException
   */
  synchronized void accessLocalBlock(long blockId) throws IOException {
    if (mWorkerClient.isLocal()) {
      mWorkerClient.accessBlock(blockId);
    }
  }

  /**
   * Notify the worker that the checkpoint file of the file mFileId has been added.
   * 
   * @param fid the file id
   * @throws IOException
   */
  synchronized void addCheckpoint(int fid) throws IOException {
    mWorkerClient.addCheckpoint(mMasterClient.getUserId(), fid);
  }

  /**
   * Notify the worker to checkpoint the file asynchronously
   * 
   * @param fid the file id
   * @return true if succeed, false otherwise
   * @throws IOException
   */
  synchronized boolean asyncCheckpoint(int fid) throws IOException {
    return mWorkerClient.asyncCheckpoint(fid);
  }

  /**
   * Notify the worker the block is cached.
   * 
   * @param blockId the block id
   * @throws IOException
   */
  public synchronized void cacheBlock(long blockId) throws IOException {
    mWorkerClient.cacheBlock(blockId);
  }

  /**
   * Close the client. Close the connections to both to master and worker
   * 
   * @throws IOException
   */
  @Override
  public synchronized void close() throws IOException {
    if (mWorkerClient.isConnected()) {
      mWorkerClient.returnSpace(mMasterClient.getUserId(), mAvailableSpaceBytes);
      mWorkerClient.close();
    }

    if (mMasterClient != null) {
      mMasterClient.close();
    }
  }

  /**
   * The file is complete.
   * 
   * @param fid the file id
   * @throws IOException
   */
  synchronized void completeFile(int fid) throws IOException {
    mMasterClient.user_completeFile(fid);
  }

  /**
   * Create a user local temporary folder and return it
   * 
   * @return the local temporary folder for the user or null if unable to allocate one.
   * @throws IOException
   */
  synchronized File createAndGetUserLocalTempFolder() throws IOException {
    String userTempFolder = mWorkerClient.getUserTempFolder();

    if (StringUtils.isBlank(userTempFolder)) {
      LOG.error("Unable to get local temporary folder location for user.");
      return null;
    }

    File ret = new File(userTempFolder);
    if (!ret.exists()) {
      if (ret.mkdir()) {
        CommonUtils.changeLocalFileToFullPermission(ret.getAbsolutePath());
        LOG.info("Folder " + ret + " was created!");
      } else {
        LOG.error("Failed to create folder " + ret);
        return null;
      }
    }
<commit_msg>tweak TachyonFS logging.<commit_after>package tachyon.client;

import java.io.File;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import tachyon.Constants;
import tachyon.TachyonURI;
import tachyon.UnderFileSystem;
import tachyon.client.table.RawTable;
import tachyon.conf.CommonConf;
import tachyon.conf.UserConf;
import tachyon.master.MasterClient;
import tachyon.thrift.ClientBlockInfo;
import tachyon.thrift.ClientDependencyInfo;
import tachyon.thrift.ClientFileInfo;
import tachyon.thrift.ClientRawTableInfo;
import tachyon.thrift.ClientWorkerInfo;
import tachyon.util.CommonUtils;
import tachyon.worker.WorkerClient;

/**
 * Tachyon's user client API. It contains a MasterClient and several WorkerClients depending on how
 * many workers the client program is interacting with.
 */
public class TachyonFS extends AbstractTachyonFS {
  /**
   * Create a TachyonFS handler.
   * 
   * @param tachyonURI a Tachyon URI contains master address. e.g., tachyon://localhost:19998,
   *        tachyon://localhost:19998/ab/c.txt
   * @return the corresponding TachyonFS handler
   * @throws IOException
   */
  public static synchronized TachyonFS get(final TachyonURI tachyonURI) throws IOException {
    if (tachyonURI == null) {
      throw new IOException("Tachyon Uri cannot be null. Use " + Constants.HEADER + "host:port/ ,"
          + Constants.HEADER_FT + "host:port/");
    } else {
      String scheme = tachyonURI.getScheme();
      if (scheme == null || tachyonURI.getHost() == null || tachyonURI.getPort() == -1
          || (!Constants.SCHEME.equals(scheme) && !Constants.SCHEME_FT.equals(scheme))) {
        throw new IOException("Invalid Tachyon URI: " + tachyonURI + ". Use " + Constants.HEADER
            + "host:port/ ," + Constants.HEADER_FT + "host:port/");
      }
      return new TachyonFS(tachyonURI);
    }
  }

  /**
   * Create a TachyonFS handler.
   * 
   * @param masterHost master host details
   * @param masterPort port master listens on
   * @param zookeeperMode use zookeeper
   * 
   * @return the corresponding TachyonFS hanlder
   * @throws IOException
   */
  public static synchronized TachyonFS get(String masterHost, int masterPort, boolean zookeeperMode)
      throws IOException {
    return new TachyonFS(new InetSocketAddress(masterHost, masterPort), zookeeperMode);
  }

  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
  private final long mUserQuotaUnitBytes = UserConf.get().QUOTA_UNIT_BYTES;
  private final int mUserFailedSpaceRequestLimits = UserConf.get().FAILED_SPACE_REQUEST_LIMITS;

  // The RPC client talks to the system master.
  private MasterClient mMasterClient = null;
  // The Master address.
  private InetSocketAddress mMasterAddress = null;
  // The RPC client talks to the local worker if there is one.
  private WorkerClient mWorkerClient = null;
  // Whether use ZooKeeper or not
  private boolean mZookeeperMode;
  // Cached ClientFileInfo
  private Map<String, ClientFileInfo> mPathToClientFileInfo = new HashMap<String, ClientFileInfo>();
  private Map<Integer, ClientFileInfo> mIdToClientFileInfo = new HashMap<Integer, ClientFileInfo>();

  private UnderFileSystem mUnderFileSystem = null;

  // All Blocks has been locked.
  private Map<Long, Set<Integer>> mLockedBlockIds = new HashMap<Long, Set<Integer>>();

  // Each user facing block has a unique block lock id.
  private AtomicInteger mBlockLockId = new AtomicInteger(0);

  // Available memory space for this client.
  private Long mAvailableSpaceBytes;

  private TachyonFS(TachyonURI tachyonURI) throws IOException {
    this(new InetSocketAddress(tachyonURI.getHost(), tachyonURI.getPort()), tachyonURI.getScheme()
        .equals(Constants.SCHEME_FT));
  }

  private TachyonFS(InetSocketAddress masterAddress, boolean zookeeperMode) throws IOException {
    mMasterAddress = masterAddress;
    mZookeeperMode = zookeeperMode;
    mAvailableSpaceBytes = 0L;

    mMasterClient = new MasterClient(mMasterAddress, mZookeeperMode);
    mWorkerClient = new WorkerClient(mMasterClient);
  }

  /**
   * Update the latest block access time on the worker.
   * 
   * @param blockId the local block's id
   * @throws IOException
   */
  synchronized void accessLocalBlock(long blockId) throws IOException {
    if (mWorkerClient.isLocal()) {
      mWorkerClient.accessBlock(blockId);
    }
  }

  /**
   * Notify the worker that the checkpoint file of the file mFileId has been added.
   * 
   * @param fid the file id
   * @throws IOException
   */
  synchronized void addCheckpoint(int fid) throws IOException {
    mWorkerClient.addCheckpoint(mMasterClient.getUserId(), fid);
  }

  /**
   * Notify the worker to checkpoint the file asynchronously
   * 
   * @param fid the file id
   * @return true if succeed, false otherwise
   * @throws IOException
   */
  synchronized boolean asyncCheckpoint(int fid) throws IOException {
    return mWorkerClient.asyncCheckpoint(fid);
  }

  /**
   * Notify the worker the block is cached.
   * 
   * @param blockId the block id
   * @throws IOException
   */
  public synchronized void cacheBlock(long blockId) throws IOException {
    mWorkerClient.cacheBlock(blockId);
  }

  /**
   * Close the client. Close the connections to both to master and worker
   * 
   * @throws IOException
   */
  @Override
  public synchronized void close() throws IOException {
    if (mWorkerClient.isConnected()) {
      mWorkerClient.returnSpace(mMasterClient.getUserId(), mAvailableSpaceBytes);
      mWorkerClient.close();
    }

    if (mMasterClient != null) {
      mMasterClient.close();
    }
  }

  /**
   * The file is complete.
   * 
   * @param fid the file id
   * @throws IOException
   */
  synchronized void completeFile(int fid) throws IOException {
    mMasterClient.user_completeFile(fid);
  }

  /**
   * Create a user local temporary folder and return it
   * 
   * @return the local temporary folder for the user or null if unable to allocate one.
   * @throws IOException
   */
  synchronized File createAndGetUserLocalTempFolder() throws IOException {
    String userTempFolder = mWorkerClient.getUserTempFolder();

    if (StringUtils.isBlank(userTempFolder)) {
      LOG.error("Unable to get local temporary folder \"{}\" for user.", userTempFolder);
      return null;
    }

    File ret = new File(userTempFolder);
    if (!ret.exists()) {
      if (ret.mkdir()) {
        CommonUtils.changeLocalFileToFullPermission(ret.getAbsolutePath());
        LOG.info("Folder " + ret + " was created!");
      } else {
        LOG.error("Failed to create folder " + ret);
        return null;
      }
    }

{'fraction_non_alphanumeric': 0.09271427539894578, 'fraction_numerical': 0.002166221387825836, 'mean_word_length': 3.680635349780331, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5209751', 'n_tokens_mistral': 4225, 'n_tokens_neox': 4078, 'n_words': 1138}