alluxio-master-2.x/core/server/worker/src/main/java/alluxio/worker/grpc/BlockReadRequestContext.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.worker.grpc;

import alluxio.wire.BlockReadRequest;
import alluxio.worker.block.io.BlockReader;

import com.codahale.metrics.Counter;
import com.codahale.metrics.Meter;

import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * Context of {@link BlockReadRequest}.
 */
@NotThreadSafe
public final class BlockReadRequestContext {
  /**
   * Set to true if the data reader is active. The following invariants must be maintained:
   * 1. If true, there will be at least one more message (data, eof or error) to be sent to gRPC.
   * 2. If false, there will be no more message sent to gRPC until it is set to true again.
   */
  private boolean mDataReaderActive;
  /**
   * The next pos to queue to the read buffer.
   */
  private long mPosToQueue;

  private long mPosReceived;
  /**
   * mEof, mCancel and mError are the notifications processed by the data reader thread. They can
   * be set by either the gRPC I/O thread or the data reader thread. mError overrides mCancel
   * and mEof, mEof overrides mCancel.
   *
   * These notifications determine 3 ways to complete a read request.
   * 1. mEof: The read request is fulfilled. All the data requested by the client or all the data in
   *    the block/file has been read. The data reader replies a SUCCESS response when processing
   *    mEof.
   * 2. mCancel: The read request is cancelled by the client. A cancel request is ignored if mEof
   *    is set. The data reader replies a CANCEL response when processing mCancel.
   *    Note: The client can send a cancel request after the server has sent a SUCCESS response. But
   *    it is not possible for the client to send a CANCEL request after the channel has been
   *    released. So it is impossible for a CANCEL request from one read request to cancel
   *    another read request.
   * 3. mError: mError is set whenever an error occurs. It can be from an exception when reading
   *    data, or writing data to stream or the client closes the channel etc. An ERROR response
   *    is optionally sent to the client when data reader thread process mError. The channel
   *    is closed after this error response is sent.
   *
   * Note: it is guaranteed that only one of SUCCESS and CANCEL responses is sent at most once
   * because the data reader thread won't be restarted as long as mCancel or mEof is set except
   * when error happens (mError overrides mCancel and mEof).
   */
  private boolean mEof;
  private boolean mCancel;
  private Error mError;
  private Counter mCounter;
  private Meter mMeter;
  /** This is set when the SUCCESS or CANCEL response is sent. This is only for sanity check. */
  private volatile boolean mDone;
  private BlockReader mBlockReader;
  /** The requests of this context. */
  private final BlockReadRequest mRequest;

  /**
   * @param request the read request
   */
  public BlockReadRequestContext(alluxio.grpc.ReadRequest request) {
    mRequest = BlockReadRequest.from(request);
    mPosToQueue = 0;
    mDataReaderActive = false;
    mEof = false;
    mCancel = false;
    mError = null;
    mDone = false;
  }

  /**
   * @return request received from channel
   */
  public BlockReadRequest getRequest() {
    return mRequest;
  }

  /**
   * @return whether the data reader is active
   */
  @GuardedBy("BlockReadHandler#mLock")
  public boolean isDataReaderActive() {
    return mDataReaderActive;
  }

  /**
   * @return the next position to queue to the read buffer
   */
  @GuardedBy("BlockReadHandler#mLock")
  public long getPosToQueue() {
    return mPosToQueue;
  }

  /**
   * @return the position before which data are received by the client
   */
  @GuardedBy("BlockReadHandler#mLock")
  public long getPosReceived() {
    return mPosReceived;
  }

  /**
   * @return true when the data reader replies a SUCCESS response, false otherwise
   */
  @GuardedBy("BlockReadHandler#mLock")
  public boolean isEof() {
    return mEof;
  }

  /**
   * @return true when a CANCEL request is received by the client, false otherwise
   */
  @GuardedBy("BlockReadHandler#mLock")
  public boolean isCancel() {
    return mCancel;
  }

  /**
   * @return the error during this read request
   */
  @GuardedBy("BlockReadHandler#mLock")
  @Nullable
  public Error getError() {
    return mError;
  }

  /**
   * @return true when the SUCCESS or CANCEL response is sent, false otherwise
   */
  public boolean isDoneUnsafe() {
    return mDone;
  }

  /**
   * @return metrics counter associated with this request
   */
  @Nullable
  public Counter getCounter() {
    return mCounter;
  }

  /**
   * @return metrics meter associated with this request
   */
  @Nullable
  public Meter getMeter() {
    return mMeter;
  }

  /**
   * @param packetReaderActive packet reader state to set
   */
  @GuardedBy("BlockReadHandler#mLock")
  public void setDataReaderActive(boolean packetReaderActive) {
    mDataReaderActive = packetReaderActive;
  }

  /**
   * @param posToQueue the next position to queue to the netty buffer to set
   */
  @GuardedBy("BlockReadHandler#mLock")
  public void setPosToQueue(long posToQueue) {
    mPosToQueue = posToQueue;
  }

  /**
   * @param posReceived the position before which data are received by the client
   */
  @GuardedBy("BlockReadHandler#mLock")
  public void setPosReceived(long posReceived) {
    mPosReceived = posReceived;
  }

  /**
   * @param eof whether SUCCESS response is replied
   */
  @GuardedBy("BlockReadHandler#mLock")
  public void setEof(boolean eof) {
    mEof = eof;
  }

  /**
   * @param cancel whether the CANCEL request is received
   */
  @GuardedBy("BlockReadHandler#mLock")
  public void setCancel(boolean cancel) {
    mCancel = cancel;
  }

  /**
   * @param error the error
   */
  @GuardedBy("BlockReadHandler#mLock")
  public void setError(Error error) {
    mError = error;
  }

  /**
   * @param done whether the SUCCESS or CANCEL response is sent
   */
  public void setDoneUnsafe(boolean done) {
    mDone = done;
  }

  /**
   * @param counter counter to set
   */
  public void setCounter(Counter counter) {
    mCounter = counter;
  }

  /**
   * @param meter meter to set
   */
  public void setMeter(Meter meter) {
    mMeter = meter;
  }

  /**
   * @return block reader
   */
  @Nullable
  public BlockReader getBlockReader() {
    return mBlockReader;
  }

  /**
   * @param blockReader block reader to set
   */
  public void setBlockReader(BlockReader blockReader) {
    mBlockReader = blockReader;
  }
}
