Hbase源码分析：server端RPC

server端rpc包括master和RegionServer。接下来主要梳理一下，master和regionserver中有关rpc创建，启动以及处理的过程。

1，server rpc的初始化过程

首先看一下上篇rpc概述中有关hbase rpc端的总体流程图。

由于HMaster继承自HRegionServer，master和region server中有关rpc的成员变量主要在HRegionServer中，主要包括（rpcServices和rpcClient）。当前主要讨论rpcServices，有关RpcClient会另外单独讨论。

Master和Region Server启动过程中有关rpc初始化和启动过程中的步骤如下：

1，在HRegionServer构造函数中调用createRpcService生成RSRpcServices对象。如果是master启动，HRegionServer是Hmaster的父类，该函数也会调用。

  protected RSRpcServices createRpcServices() throws IOException {
    return new RSRpcServices(this);
  }

2，在RSRpcServices的构造函数中，生成RpcServer对象。

    rpcServer = new RpcServer(rs, name, getServices(),
      bindAddress, // use final bindAddress for this server.
      rs.conf,
      rpcSchedulerFactory.create(rs.conf, this, rs));

在构造RpcServer对象的过程中，HMaster和HRegionServer分别实现了getService（）函数以使HMaster和HRegionServer响应不同的rpc服务。

3，在RpcServer的构造函数中，分别生成Listener，responder以及scheduler等几个重要的对象

    // Start the listener here and let it bind to the port
    listener = new Listener(name);
    this.port = listener.getAddress().getPort();

    this.metrics = new MetricsHBaseServer(name, new MetricsHBaseServerWrapperImpl(this));
    this.tcpNoDelay = conf.getBoolean("hbase.ipc.server.tcpnodelay", true);
    this.tcpKeepAlive = conf.getBoolean("hbase.ipc.server.tcpkeepalive", true);

    this.warnDelayedCalls = conf.getInt(WARN_DELAYED_CALLS, DEFAULT_WARN_DELAYED_CALLS);
    this.delayedCalls = new AtomicInteger(0);
    this.ipcUtil = new IPCUtil(conf);


    // Create the responder here
    responder = new Responder();
    this.authorize = conf.getBoolean(HADOOP_SECURITY_AUTHORIZATION, false);
    this.userProvider = UserProvider.instantiate(conf);
    this.isSecurityEnabled = userProvider.isHBaseSecurityEnabled();
    if (isSecurityEnabled) {
      HBaseSaslRpcServer.init(conf);
    }
    this.scheduler = scheduler;
    this.scheduler.init(new RpcSchedulerContext(this));

4，在Listener的构造函数中，还包含了readThreads个reader用来读取请求。

readers = new Reader[readThreads];
      readPool = Executors.newFixedThreadPool(readThreads,
        new ThreadFactoryBuilder().setNameFormat(
          "RpcServer.reader=%d,bindAddress=" + bindAddress.getHostName() +
          ",port=" + port).setDaemon(true).build());
      for (int i = 0; i < readThreads; ++i) {
        Reader reader = new Reader();
        readers[i] = reader;
        readPool.execute(reader);
      }

在以上这些对象构造完成以后，在HRegionServer的构造函数中会调用rpcServices.start()---》rpcServer.start(). 在rpcServer start函数中会分别启动responder，listener以及scheduler。

  public synchronized void start() {
    if (started) return;
    authTokenSecretMgr = createSecretManager();
    if (authTokenSecretMgr != null) {
      setSecretManager(authTokenSecretMgr);
      authTokenSecretMgr.start();
    }
    this.authManager = new ServiceAuthorizationManager();
    HBasePolicyProvider.init(conf, authManager);
    responder.start();
    listener.start();
    scheduler.start();
    started = true;
  }

2，server rpc的处理过程

rpcserver监控，读取，请求基于Reactor模式, 流程图如下（来自引用）。 

2.1 Listener

对于Listener，有一个acceptChannle的ServerSocketChannel，acceptChannle在selector注册了OP_ACCEPT事件，同时Listener中包含了readThreads的readers线程由线程池管理。Listener的主要处理流程在doRunLoop函数中：

private synchronized void doRunLoop() {
        while (running) {
          try {
            readSelector.select();
            while (adding) {
              this.wait(1000);
            }

            Iterator<SelectionKey> iter = readSelector.selectedKeys().iterator();
            while (iter.hasNext()) {
              SelectionKey key = iter.next();
              iter.remove();
              if (key.isValid()) {
                if (key.isReadable()) {
                  doRead(key);
                }
              }
            }
          } catch (InterruptedException e) {
            LOG.debug("Interrupted while sleeping");
            return;
          } catch (IOException ex) {
            LOG.info(getName() + ": IOException in Reader", ex);
          }
        }
      }

当没有请求的时候，线程阻塞在第4行的select处。当有请求来临时，在判断请求有效后，会读取该连接上的请求上的数据（具体逻辑在readAndProcess函数中）。读取数据以后，会处理数据，具体在processOneRpc函数中。

    private void processOneRpc(byte[] buf) throws IOException, InterruptedException {
      if (connectionHeaderRead) {
        processRequest(buf);
      } else {
        processConnectionHeader(buf);
        this.connectionHeaderRead = true;
        if (!authorizeConnection()) {
          // Throw FatalConnectionException wrapping ACE so client does right thing and closes
          // down the connection instead of trying to read non-existent retun.
          throw new AccessDeniedException("Connection from " + this + " for service " +
            connectionHeader.getServiceName() + " is unauthorized for user: " + user);
        }
      }
    }

根据连接头是否已经读取，如果没有读取连接头信息，变通过ProcessConnectionHeader读取连接头信息。如果读取连接头信息以后，会解析请求，并且将请求构造成统一的结构CallRunner，最终这个CallRunnder会被添加到scheduler中任务队列中，根据不同的调度策略（FifoRpcScheduler和SimpleRpcScheduler）进行处理。ProcessRequest的核心代码如下：

      Call call = new Call(id, this.service, md, header, param, cellScanner, this, responder,
              totalRequestSize, traceInfo, RpcServer.getRemoteIp());
      scheduler.dispatch(new CallRunner(RpcServer.this, call));

2.2 Scheduler

Scheduler 默认使用了SimpleRpcScheduler。SimpleRpcScheduler包含三个不同的RpcExecutor（callExecutor，priorityExecutor，replicationExecutor）。对于大部分基于用户表的请求都是通过callExecutor来执行，callExecutor从之前添加的请求任务队列中获取请求，并且将请求交流对应的handler进行处理。具体逻辑在RpcExecutor的consumerLoop中，如下：

 protected void consumerLoop(final BlockingQueue<CallRunner> myQueue) {
    boolean interrupted = false;
    double handlerFailureThreshhold =
        conf == null ? 1.0 : conf.getDouble(HConstants.REGION_SERVER_HANDLER_ABORT_ON_ERROR_PERCENT,
          HConstants.DEFAULT_REGION_SERVER_HANDLER_ABORT_ON_ERROR_PERCENT);
    try {
      while (running) {
        try {
          MonitoredRPCHandler status = RpcServer.getStatus();
          CallRunner task = myQueue.take();
          task.setStatus(status);
          try {
            activeHandlerCount.incrementAndGet();
            task.run();
          } 

由于myQueue是阻塞队列，如果没有请求，那么scheduler将阻塞在第10行take处。否则将执行CallRunner中的run函数。而紧接着会调用rpcServer中的call函数。

        // make the call
        resultPair = this.rpcServer.call(call.service, call.md, call.param, call.cellScanner,
          call.timestamp, this.status);

而在rpcServer的call函数中，首先会根据请求调用本地的对应的实现函数，并且通过阻塞的方法调用，返回结果（result）。

PayloadCarryingRpcController controller = new PayloadCarryingRpcController(cellScanner);
      Message result = service.callBlockingMethod(md, controller, param);
...
return new Pair<Message, CellScanner>(result, controller.cellScanner());

并且在CallRunner的 run函数中，将结果通过调用setResponse函数生成返回结果，将结果通过调用sendResponseIfReady通过responder将结果返回给client端。

      // Set the response for undelayed calls and delayed calls with
      // undelayed responses.
      if (!call.isDelayed() || !call.isReturnValueDelayed()) {
        Message param = resultPair != null ? resultPair.getFirst() : null;
        CellScanner cells = resultPair != null ? resultPair.getSecond() : null;
        call.setResponse(param, cells, errorThrowable, error);
      }

2.3 responder

responder负责将结果写回给client端。responder的实现也是通过类似Listener的React模式。上面schedule调度执行完以后生成的结果，将通过doRespond函数加入到返回结果的相应队列里面。在这个函数里面，如果一次channlewrite能够完成操作，则直接完成该写结果请求。否则将该call的connection注册OP_WRITE到selector。

void doRespond(Call call) throws IOException {
      boolean added = false;

      // If there is already a write in progress, we don't wait. This allows to free the handlers
      //  immediately for other tasks.
      if (call.connection.responseQueue.isEmpty() && call.connection.responseWriteLock.tryLock()) {
        try {
          if (call.connection.responseQueue.isEmpty()) {
            // If we're alone, we can try to do a direct call to the socket. It's
            //  an optimisation to save on context switches and data transfer between cores..
            if (processResponse(call)) {
              return; // we're done.
            }
            // Too big to fit, putting ahead.
            call.connection.responseQueue.addFirst(call);
            added = true; // We will register to the selector later, outside of the lock.
          }
        } finally {
          call.connection.responseWriteLock.unlock();
        }
      }

      if (!added) {
        call.connection.responseQueue.addLast(call);
      }
      call.responder.registerForWrite(call.connection);

      // set the serve time when the response has to be sent later
      call.timestamp = System.currentTimeMillis();
    }
  }

在registerForWrite中会唤醒writeSelect，使得一旦有该连接上的请求数据过来，那么responder将通过doAsSyncWrite--》ProcessAllResponse处理请求，此时便和Listener的处理类似了。

          registerWrites();
          int keyCt = writeSelector.select(purgeTimeout);
          if (keyCt == 0) {
            continue;
          }

          Set<SelectionKey> keys = writeSelector.selectedKeys();
          Iterator<SelectionKey> iter = keys.iterator();
          while (iter.hasNext()) {
            SelectionKey key = iter.next();
            iter.remove();
            try {
              if (key.isValid() && key.isWritable()) {
                doAsyncWrite(key);
              }

3 小结

本文结合代码了解了rpcserver的listener，reader，scheduler以及responder处理rpc请求的过程。对server端处理rpc请求有了一个较为清晰的认识。接下来会对client端的rpc请求逻辑做一个梳理，加油！