HBase中scan的ReadType探究

背景知识

Linux层面

linux对于文件的读取，提供了不同的函数，引用资料如下：

当对同一文件句柄（在Windows下）或是文件描述符（在Linux下）进行随机读写操作时，会存在文件指针的定位与读/写俩个步骤，但由于这不是一个原子操作，就可能产生如下问题：进程A对某文件先定位到 f1 处，然后被中断，然后进程B对同一文件定位到 f2 处，然后被中断，进程A再次执行，从文件的当前指针处开始读或是写，于是这便产生了不是期望的结果了。（这里要注意，对同一文件的俩次打开，得到的将是俩个不同的句柄或是描述符，所以不用担心这种情况会出问题）

解决办法：

　　在Linux下，pread函数就好像是专门为上面的问题服务的，它本身就是原子性的操作，定位文件指针与读操作一气呵成，而且读操作并不改变文件指针。

总体来说，常用的有seek()+read() 和 pread()这2种方式，优劣如下：
seek()+read()非线程安全，但由于利用了文件描述符所保存的文件指针，不需要每次读取时都去定位，因此读取效率较高，应用层面多线程访问时则需要做同步；
pread()是原子操作，线程安全，但由于每次都需要定位文件指针，所以读取效率较低；

Hdfs层面

hdfs基于linux的不同函数，提供了不同的实现，对应issue如下(https://issues.apache.org/jira/browse/HADOOP-519)：

HDFS File API should be extended to include positional read

HDFS Input streams should support positional read. Positional read (such as the pread syscall on linux) allows reading for a specified offset without affecting the current file offset. Since the underlying file state is not touched, pread can be used efficiently in multi-threaded programs.

Here is how I plan to implement it.

Provide PositionedReadable interface, with the following methods:

int read(long position, byte[] buffer, int offset, int length);
void readFully(long position, byte[] buffer, int offset, int length);
void readFully(long position, byte[] buffer);

Abstract class FSInputStream would provide default implementation of the above methods using getPos(), seek() and read() methods. The default implementation is inefficient in multi-threaded programs since it locks the object while seeking, reading, and restoring to old state.

DFSClient.DFSInputStream, which extends FSInputStream will provide an efficient non-synchronized implementation for above calls.

In addition, FSDataInputStream, which is a wrapper around FSInputStream, will provide wrapper methods for above read methods as well.

HBase中的应用

HBase中，定义了2种不同的ReadType：PREAD和STREAM，分别代表pread()和seek()+read()：

  @InterfaceAudience.Public
  public enum ReadType {
    DEFAULT, STREAM, PREAD
  }

读取hfile需要通过scanner，而创建StoreFileScanner的时候，会根据ReadType进入到不同的流程：

for (int i = 0, n = files.size(); i < n; i++) {
        HStoreFile sf = sortedFiles.remove();
        StoreFileScanner scanner;
        if (usePread) {
          scanner = sf.getPreadScanner(cacheBlocks, readPt, i, canOptimizeForNonNullColumn);
        } else {
          scanner = sf.getStreamScanner(canUseDrop, cacheBlocks, isCompaction, readPt, i,
              canOptimizeForNonNullColumn);
        }
        scanners.add(scanner);
      }

其中，getPreadScanner会直接返回共享的reader对象，即底层共享同一个inputStream：

  /**
   * Get a scanner which uses pread.
   * <p>
   * Must be called after initReader.
   */
  public StoreFileScanner getPreadScanner(boolean cacheBlocks, long readPt, long scannerOrder,
      boolean canOptimizeForNonNullColumn) {
    return getReader().getStoreFileScanner(cacheBlocks, true, false, readPt, scannerOrder,
      canOptimizeForNonNullColumn);
  }
  
    /**
   * @return Current reader. Must call initReader first else returns null.
   * @see #initReader()
   */
  public StoreFileReader getReader() {
    return this.reader;
  }

而getStreamScanner会创建一个新的reader，在fileInfo.open方法中，会打开一个新的inputStream，然后读取hfile中相关元数据信息，如果启用了preFetchOnOpen也会触发读取数据块：

  /**
   * Get a scanner which uses streaming read.
   * <p>
   * Must be called after initReader.
   */
  public StoreFileScanner getStreamScanner(boolean canUseDropBehind, boolean cacheBlocks,
      boolean isCompaction, long readPt, long scannerOrder, boolean canOptimizeForNonNullColumn)
      throws IOException {
    return createStreamReader(canUseDropBehind).getStoreFileScanner(cacheBlocks, false,
      isCompaction, readPt, scannerOrder, canOptimizeForNonNullColumn);
  }
  
  private StoreFileReader createStreamReader(boolean canUseDropBehind) throws IOException {
    initReader();
    StoreFileReader reader = fileInfo.open(this.fs, this.cacheConf, canUseDropBehind, -1L,
      primaryReplica, refCount, false);
    reader.copyFields(this.reader);
    return reader;
  }

  /**
   * Open a Reader for the StoreFile
   * @param fs The current file system to use.
   * @param cacheConf The cache configuration and block cache reference.
   * @return The StoreFile.Reader for the file
   */
  public StoreFileReader open(FileSystem fs, CacheConfig cacheConf, boolean canUseDropBehind,
      long readahead, boolean isPrimaryReplicaStoreFile, AtomicInteger refCount, boolean shared)
      throws IOException {
    FSDataInputStreamWrapper in;
    FileStatus status;

    final boolean doDropBehind = canUseDropBehind && cacheConf.shouldDropBehindCompaction();
    if (this.link != null) {
      // HFileLink
      in = new FSDataInputStreamWrapper(fs, this.link, doDropBehind, readahead);
      status = this.link.getFileStatus(fs);
    } else if (this.reference != null) {
      // HFile Reference
      Path referencePath = getReferredToFile(this.getPath());
      in = new FSDataInputStreamWrapper(fs, referencePath, doDropBehind, readahead);
      status = fs.getFileStatus(referencePath);
    } else {
      in = new FSDataInputStreamWrapper(fs, this.getPath(), doDropBehind, readahead);
      status = fs.getFileStatus(initialPath);
    }
    long length = status.getLen();
    hdfsBlocksDistribution = computeHDFSBlocksDistribution(fs);

    StoreFileReader reader = null;
    if (this.coprocessorHost != null) {
      reader = this.coprocessorHost.preStoreFileReaderOpen(fs, this.getPath(), in, length,
        cacheConf, reference);
    }
    if (reader == null) {
      if (this.reference != null) {
        reader = new HalfStoreFileReader(fs, this.getPath(), in, length, cacheConf, reference,
            isPrimaryReplicaStoreFile, refCount, shared, conf);
      } else {
        reader = new StoreFileReader(fs, status.getPath(), in, length, cacheConf,
            isPrimaryReplicaStoreFile, refCount, shared, conf);
      }
    }
    if (this.coprocessorHost != null) {
      reader = this.coprocessorHost.postStoreFileReaderOpen(fs, this.getPath(), in, length,
        cacheConf, reference, reader);
    }
    return reader;
  }

这里有2个疑问

1：共享reader是哪里来的

在open region或者由于flush和bulkload产生新的hfile时，都会open hfile，此时会创建一个reader进行元数据读取，此reader即为共享reader，其shared属性被设置为true；

StoreFileReader.java

  // indicate that whether this StoreFileReader is shared, i.e., used for pread. If not, we will
  // close the internal reader when readCompleted is called.
  @VisibleForTesting
  final boolean shared;

2：pread和stream分别什么时候使用

默认情况下，get请求使用pread,compact scan使用stream；
对于user scan，则有以下规则：

• 如果客户端明确指定readType，则直接使用
• 如果客户端未指定，则服务端会以pread方式开始，读取超过4个blocksize大小数据时，切换为stream方式，该阈值通过hbase.storescanner.pread.max.bytes进行配置
• 如果不希望服务端进行上述切换，而固定使用pread，可将hbase.storescanner.use.pread配置为true

另外，在读取完成，关闭scanner时，会调用readCompleted方法，该方法会判断shared来决定是否关闭所使用的reader：

StoreFileScanner.java

  @Override
  public void close() {
    if (closed) return;
    cur = null;
    this.hfs.close();
    if (this.reader != null) {
      this.reader.readCompleted();
    }
    closed = true;
  }
  
StoreFileReader.java

  /**
   * Indicate that the scanner has finished reading with this reader. We need to decrement the ref
   * count, and also, if this is not the common pread reader, we should close it.
   */
  void readCompleted() {
    refCount.decrementAndGet();
    if (!shared) {
      try {
        reader.close(false);
      } catch (IOException e) {
        LOG.warn("failed to close stream reader", e);
      }
    }
  }

问题和优化

以上为2.0版本中的代码，其中有个很明显的问题，就是很多scan都重复执行了fileInfo.open方法，而该方法包含的逻辑过多，造成了很多不必要的读取，影响了scan性能且浪费系统资源，社区较新的版本对此处进行了优化，相关的issue为https://issues.apache.org/jira/browse/HBASE-22888；