<commit_before>/*
 * Licensed to the University of California, Berkeley under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package tachyon.client;

import java.io.InputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import tachyon.Constants;
import tachyon.conf.TachyonConf;
import tachyon.thrift.ClientBlockInfo;
import tachyon.thrift.NetAddress;
import tachyon.underfs.UnderFileSystem;
import tachyon.util.network.NetworkAddressUtils;

/**
 * BlockInStream for remote block.
 */
public class RemoteBlockInStream extends BlockInStream {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  /** The block info of the block we are reading */
  private ClientBlockInfo mBlockInfo;
  /**
   * An input stream for the checkpointed copy of the block. If we are ever unable to read part of
   * the block from the workers, we use this checkpoint stream
   */
  private InputStream mCheckpointInputStream = null;

  /**
   * The position in the checkpointed file that the open input stream is on, relative to the block.
   * That is, if the checkpoint input stream is set to read the 0th byte of the current block,
   * mCheckpointPos will be 0, even if we are n blocks into the input stream, so the actual file
   * position may be something else.
   */
  private long mCheckpointPos = -1;

  /**
   * The position in the block we are currently at, relative to the block. The position relative to
   * the file would be mBlockInfo.offset + mBlockPos.
   */
  private long mBlockPos = 0;

  /**
   * A byte buffer for the current chunk of the block we are reading from
   */
  private ByteBuffer mCurrentBuffer = null;

  /** We keep track of the position relative to the block that the current buffer starts at. */
  private long mBufferStartPos;

  /**
   * true if we are re-caching the file. The re-caching gets canceled if we do anything other than a
   * straight read through the file. That means, any skipping or seeking around will cancel the
   * re-cache.
   */
  private boolean mRecache;

  /**
   * True initially, will be false after a cache miss, meaning no worker had this block in memory.
   * Afterward, all reads will go directly to the under filesystem.
   */
  private boolean mAttemptReadFromWorkers = true;

  /**
   * If we are re-caching the file, we write it to a block out stream as we read it.
   */
  private BlockOutStream mBlockOutStream = null;

  /**
   * The under filesystem configuration that we use to set up the checkpoint input stream
   */
  private Object mUFSConf = null;

  /**
   * The bytes read from remote.
   */
  private long mBytesReadRemote = 0;

  /**
   * The maximum number of tries to read a remote block. Since the stored ClientBlockInfo might not
   * be accurate when executing a remote read, we refresh it and retry reading a certain number of
   * times before giving up.
   */
  private static final int MAX_REMOTE_READ_ATTEMPTS = 2;

  /** A reference to the current reader so we can clear it after reading is finished. */
  private RemoteBlockReader mCurrentReader = null;

  /**
   * @param file the file the block belongs to
   * @param readType the InStream's read type
   * @param blockIndex the index of the block in the file
   * @param tachyonConf the TachyonConf instance for this file output stream.
   * @throws IOException
   */
  RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, TachyonConf tachyonConf)
      throws IOException {
    this(file, readType, blockIndex, null, tachyonConf);
  }

  /**
   * @param file the file the block belongs to
   * @param readType the InStream's read type
   * @param blockIndex the index of the block in the file
   * @param ufsConf the under file system configuration
   * @throws IOException
   */
  RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf,
      TachyonConf tachyonConf) throws IOException {
    super(file, readType, blockIndex, tachyonConf);

    if (!mFile.isComplete()) {
      throw new IOException("File " + mFile.getPath() + " is not ready to read");
    }

    mBlockInfo = mFile.getClientBlockInfo(mBlockIndex);

    mRecache = readType.isCache();

    mUFSConf = ufsConf;
  }

  /**
   * Only called by {@link getDummyStream}.
   * An alternative to construct a RemoteBlockInstream, bypassing all the checks.
   * The returned RemoteBlockInStream can be used to call {#link #readRemoteByteBuffer}.
   *
   * @param file, any, could be null
   * @param readType, any type
   * @param blockIndex, any index
   * @param ufsConf, any ufs configuration
   * @param tachyonConf, any
   * @param addFlag, add another field so that this constructor differentiates
   */
  private RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf,
      TachyonConf tachyonConf, boolean addFlag) {
    super(file, readType, blockIndex, tachyonConf);
  }

  /**
   * Return a dummy RemoteBlockInStream object.
   * The object can be used to perform near-stateless read using {@link #readRemoteByteBuffer}.
   * (The only state kept is a handler for the underlying reader, so we can close the reader
   * when we close the dummy stream.)
   *
   * @return a dummy RemoteBlockInStream object.
   */
  // See {tachyon.client.TachyonFile#readRemoteByteBuffer(ClientBlockInfo)} for usage
  public static RemoteBlockInStream getDummyStream() {
    return new RemoteBlockInStream(new TachyonFile(null, -1, null),
        ReadType.NO_CACHE, -1, null, null, true);
  }

  /**
   * Cancels the re-caching attempt
   *
   * @throws IOException
   */
  private void cancelRecache() throws IOException {
    if (mRecache) {
      mRecache = false;
      if (mBlockOutStream != null) {
        mBlockOutStream.cancel();
      }
    }
  }

  @Override
  public void close() throws IOException {
    if (mClosed) {
      return;
    }
    if (mRecache && mBlockOutStream != null) {
      // We only finish re-caching if we've gotten to the end of the file
      if (mBlockPos == mBlockInfo.length) {<commit_msg>Move comment into javadoc for format consistency<commit_after>/*
 * Licensed to the University of California, Berkeley under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package tachyon.client;

import java.io.InputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import tachyon.Constants;
import tachyon.conf.TachyonConf;
import tachyon.thrift.ClientBlockInfo;
import tachyon.thrift.NetAddress;
import tachyon.underfs.UnderFileSystem;
import tachyon.util.network.NetworkAddressUtils;

/**
 * BlockInStream for remote block.
 */
public class RemoteBlockInStream extends BlockInStream {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);

  /** The block info of the block we are reading */
  private ClientBlockInfo mBlockInfo;
  /**
   * An input stream for the checkpointed copy of the block. If we are ever unable to read part of
   * the block from the workers, we use this checkpoint stream
   */
  private InputStream mCheckpointInputStream = null;

  /**
   * The position in the checkpointed file that the open input stream is on, relative to the block.
   * That is, if the checkpoint input stream is set to read the 0th byte of the current block,
   * mCheckpointPos will be 0, even if we are n blocks into the input stream, so the actual file
   * position may be something else.
   */
  private long mCheckpointPos = -1;

  /**
   * The position in the block we are currently at, relative to the block. The position relative to
   * the file would be mBlockInfo.offset + mBlockPos.
   */
  private long mBlockPos = 0;

  /**
   * A byte buffer for the current chunk of the block we are reading from
   */
  private ByteBuffer mCurrentBuffer = null;

  /** We keep track of the position relative to the block that the current buffer starts at. */
  private long mBufferStartPos;

  /**
   * true if we are re-caching the file. The re-caching gets canceled if we do anything other than a
   * straight read through the file. That means, any skipping or seeking around will cancel the
   * re-cache.
   */
  private boolean mRecache;

  /**
   * True initially, will be false after a cache miss, meaning no worker had this block in memory.
   * Afterward, all reads will go directly to the under filesystem.
   */
  private boolean mAttemptReadFromWorkers = true;

  /**
   * If we are re-caching the file, we write it to a block out stream as we read it.
   */
  private BlockOutStream mBlockOutStream = null;

  /**
   * The under filesystem configuration that we use to set up the checkpoint input stream
   */
  private Object mUFSConf = null;

  /**
   * The bytes read from remote.
   */
  private long mBytesReadRemote = 0;

  /**
   * The maximum number of tries to read a remote block. Since the stored ClientBlockInfo might not
   * be accurate when executing a remote read, we refresh it and retry reading a certain number of
   * times before giving up.
   */
  private static final int MAX_REMOTE_READ_ATTEMPTS = 2;

  /** A reference to the current reader so we can clear it after reading is finished. */
  private RemoteBlockReader mCurrentReader = null;

  /**
   * @param file the file the block belongs to
   * @param readType the InStream's read type
   * @param blockIndex the index of the block in the file
   * @param tachyonConf the TachyonConf instance for this file output stream.
   * @throws IOException
   */
  RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, TachyonConf tachyonConf)
      throws IOException {
    this(file, readType, blockIndex, null, tachyonConf);
  }

  /**
   * @param file the file the block belongs to
   * @param readType the InStream's read type
   * @param blockIndex the index of the block in the file
   * @param ufsConf the under file system configuration
   * @throws IOException
   */
  RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf,
      TachyonConf tachyonConf) throws IOException {
    super(file, readType, blockIndex, tachyonConf);

    if (!mFile.isComplete()) {
      throw new IOException("File " + mFile.getPath() + " is not ready to read");
    }

    mBlockInfo = mFile.getClientBlockInfo(mBlockIndex);

    mRecache = readType.isCache();

    mUFSConf = ufsConf;
  }

  /**
   * Only called by {@link getDummyStream}.
   * An alternative to construct a RemoteBlockInstream, bypassing all the checks.
   * The returned RemoteBlockInStream can be used to call {#link #readRemoteByteBuffer}.
   *
   * @param file, any, could be null
   * @param readType, any type
   * @param blockIndex, any index
   * @param ufsConf, any ufs configuration
   * @param tachyonConf, any
   * @param addFlag, add another field so that this constructor differentiates
   */
  private RemoteBlockInStream(TachyonFile file, ReadType readType, int blockIndex, Object ufsConf,
      TachyonConf tachyonConf, boolean addFlag) {
    super(file, readType, blockIndex, tachyonConf);
  }

  /**
   * Return a dummy RemoteBlockInStream object.
   * The object can be used to perform near-stateless read using {@link #readRemoteByteBuffer}.
   * (The only state kept is a handler for the underlying reader, so we can close the reader
   * when we close the dummy stream.)
   * 
   * <p>
   * See {@link tachyon.client.TachyonFile#readRemoteByteBuffer(ClientBlockInfo)} for usage.
   * 
   * @return a dummy RemoteBlockInStream object.
   */
  public static RemoteBlockInStream getDummyStream() {
    return new RemoteBlockInStream(new TachyonFile(null, -1, null),
        ReadType.NO_CACHE, -1, null, null, true);
  }

  /**
   * Cancels the re-caching attempt
   *
   * @throws IOException
   */
  private void cancelRecache() throws IOException {
    if (mRecache) {
      mRecache = false;
      if (mBlockOutStream != null) {
        mBlockOutStream.cancel();
      }
    }
  }

  @Override
  public void close() throws IOException {
    if (mClosed) {
      return;
    }
    if (mRecache && mBlockOutStream != null) {
      // We only finish re-caching if we've gotten to the end of the file
      if (mBlockPos == mBlockInfo.length) {
{'fraction_non_alphanumeric': 0.07317602779386219, 'fraction_numerical': 0.002026635784597568, 'mean_word_length': 3.432787937119025, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, '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': '5397130', 'n_tokens_mistral': 3965, 'n_tokens_neox': 3818, 'n_words': 1630}