前言

HDFS短路讀是性能優(yōu)化的一個(gè)重要特性，它利用操作系統(tǒng)的內(nèi)存映射mmap、Domain Socket和共享內(nèi)存，避開傳統(tǒng)的基于TCP的數(shù)據(jù)通信，極大提升了數(shù)據(jù)讀取效率。
整個(gè)短路讀的過(guò)程完全放棄傳統(tǒng)的基于TCP/IP的通信方式，基于Domain Socket進(jìn)行通信，基于mmap和內(nèi)存共享進(jìn)行數(shù)據(jù)同步和塊的高效率讀取，整個(gè)讀取過(guò)程涉及到了操作系統(tǒng)(Domain Socket、mmap等)、Java NIO(ByteBuffer, Channel)等知識(shí)，同時(shí)也和HDFS的緩存系統(tǒng)交叉在一起，因此涉及到的東西很多，本文篇幅較長(zhǎng)。
短路讀的數(shù)據(jù)讀取方式由于其在本地讀取的高效率，很多分布式存儲(chǔ)系統(tǒng)都會(huì)用到，而且我發(fā)現(xiàn)很多面試官也非常喜歡問HDFS短路讀的實(shí)現(xiàn)原理，因此對(duì)短路讀的深入理解很有價(jià)值。
我看了一些介紹短路讀的文章，發(fā)現(xiàn)大多數(shù)文章都只在上層介紹其基本流程，只要具體到實(shí)現(xiàn)和一些概念的深入理解（比如到底mmap是什么？到底Domain Socket通信是什么？Domain Socket跟mmap是一回事嗎？）就泛泛而談。因此，本文以較長(zhǎng)篇幅，力求在上層原理和底層實(shí)現(xiàn)都能讓讀者徹底理解短路讀。
在第一章，我們先拋開Hadoop的實(shí)現(xiàn)本身，介紹短路讀中用到的一些操作系統(tǒng)的概念和特性，這是后續(xù)理解短路讀的基礎(chǔ)，包括域套接字讀，mmap和共享內(nèi)存。
在第二章， 我們開始介紹短路讀的具體流程，連接的建立，元數(shù)據(jù)的同步，以及最后對(duì)塊的讀取。
在第三章，列舉了一些跟本文相關(guān)的有價(jià)值的網(wǎng)絡(luò)文章作為參考。

1. 知識(shí)準(zhǔn)備

1.1 關(guān)于域套接字(Domain Socket)

什么是Domain Socket

首先注意，Domain Socket通信并不是Hadoop中的概念，而是Unix操作系統(tǒng)提供的功能。Hadoop通過(guò)自己的libhadoop可以使用Unix Domain Socket接口，從而進(jìn)行Domain Socket通信。
我們都知道，Socket 原本是為網(wǎng)絡(luò)通信而設(shè)計(jì)的，但后來(lái)在 Socket 的框架上發(fā)展出一種 IPC 機(jī)制， 就是 Unix Domain Socket， 它還有另一個(gè)名字叫 IPC(inter-process communication, 進(jìn)程間通信)。

相比于基于TCP/IP協(xié)議棧進(jìn)行的網(wǎng)絡(luò)通信，使用 Unix Domain Socket 的好處顯而易見：不需要經(jīng)過(guò)網(wǎng)絡(luò)協(xié)議棧, 不需要打包拆包、計(jì)算校驗(yàn)和、維護(hù)序號(hào)和應(yīng)答等，只是將應(yīng)用層數(shù)據(jù)從一個(gè)進(jìn)程拷貝到另一個(gè)進(jìn)程。這是因?yàn)?，進(jìn)程間通信機(jī)制本質(zhì)上是可靠的通訊，而網(wǎng)絡(luò)協(xié)議是為不可靠的通訊設(shè)計(jì)的。

但是既然是通信，就涉及到尋址，即我們總得知道對(duì)方(Hadoop中將通信雙方叫做一個(gè)Peer)的URI把？TCP/IP的URI就是URL或者IP地址，而僅限于本機(jī)的進(jìn)程間通信依靠什么去確定一個(gè)URI呢？是文件，即，通信雙方都綁定到所在機(jī)器（別忘了大家在同一臺(tái)機(jī)器上）的某一個(gè)文件，實(shí)現(xiàn)進(jìn)程之間的全雙工通信。
除了通信方式意外，Domain Socket的一個(gè)重要特性是，通信雙方可以直接傳遞文件描述符。比如，服務(wù)器端打開了一個(gè)replica文件，將打開文件的描述符通過(guò)Domain Socket傳遞給客戶端，客戶端拿著這個(gè)文件描述符就可以讀取服務(wù)器端打開的這個(gè)文件。文件描述符的傳遞將整個(gè)文件的讀寫權(quán)限掌控在DataNode 這里，但是數(shù)據(jù)的讀寫又不經(jīng)過(guò)DataNode，這是ShortCircuit讀寫與普通TCP/IP通信的最重要的不同。

再重復(fù)以便在ShortCircuit的場(chǎng)景下，Domain Socket 通信的雙方為用戶的DFSClient和同處于一臺(tái)機(jī)器上的DataNode。

Domain Socket 通信在ShortCircuit Read中做了什么

答案：進(jìn)行replica讀取前的簡(jiǎn)單通信和信息同步。
我們知道，ShortCircuit Read的目的就是避免使用TCP/IP，讓整個(gè)塊讀取過(guò)程不建立TCP/IP，由于不需要網(wǎng)絡(luò)協(xié)議棧、計(jì)算校驗(yàn)和、序號(hào)維護(hù)和應(yīng)答等，因此減少DataNode的負(fù)載、網(wǎng)絡(luò)開銷，提高通信和數(shù)據(jù)傳輸效率。要完成整個(gè)ShortCircuit Read，在進(jìn)行塊傳輸以前和過(guò)程中，客戶端和DataNode前后肯定還需要進(jìn)行一些必要的信息溝通和同步，這些元信息的溝通和同步使用的是Linux Domain Socket，避免建立TCP/IP連接，這些必要的信息溝通和同步包括但不僅僅包括：

• 客戶端告訴DataNode, 我需要讀ID為blk_123123123的文件，我的姓名(client name)是。。。。
• DataNode告訴客戶端，我剛剛創(chuàng)建好了一塊兒內(nèi)存區(qū)域，我們用這一塊兒內(nèi)存區(qū)域來(lái)放每一個(gè)你要讀取的replica的一些元數(shù)據(jù)信息，比如，這個(gè)Block需要不需要做crc校驗(yàn)等等，我把這塊內(nèi)存區(qū)域的地址給你。
• DataNode準(zhǔn)備好了對(duì)應(yīng)的replica文件信息(就是文件描述符索引fd)，通過(guò)Domain Socket發(fā)送給客戶端，客戶端根據(jù)這個(gè)文件描述符，就可以繞過(guò)DataNode，直接讀取對(duì)應(yīng)的replica文件了。

DomainSocket 在Hadoop上的基本實(shí)現(xiàn)

這里，我們先拋開Short Circuit讀的實(shí)現(xiàn)，先徹底理解Domain Socket在Java這一層的實(shí)現(xiàn)。

1. 確定Domain Socket通信的文件路徑
   Domain Socket通信的雙方通過(guò)一個(gè)本地文件作為地址空間，在通信過(guò)程中雙方會(huì)基于這個(gè)文件打開相同的socket進(jìn)行通信。這HDFS中，通過(guò)以下配置項(xiàng)配置這個(gè)文件路徑：

  <property>
    <name>dfs.domain.socket.path</name>
    <value>/var/lib/hadoop-hdfs/dn._PORT</value>
  </property>

2. 服務(wù)端在Domain Socket上綁定(bind())和監(jiān)聽(listen())
   這個(gè)綁定和監(jiān)聽的概念與TCP/IP的通信的服務(wù)端操作類似，只不過(guò)底層實(shí)現(xiàn)是基于Domain Socket。

在Hadoop上，跟Domain Socket通信相關(guān)的操作都封裝在一個(gè)類似工具類的叫做DomainSocket的類中，供DomainSocket的客戶端和服務(wù)器端共同使用。它是java層對(duì)操作系統(tǒng)層的Domain Socket的封裝，因此DomainSocket這個(gè)工具類會(huì)負(fù)責(zé)幫我們1）類初始化的時(shí)候加載libhadoop 2)提供了針對(duì)DomainSocket的一些方法，比如服務(wù)器端的綁定監(jiān)聽，客戶端的連接，3) 根據(jù)打開的Domain Socket創(chuàng)建對(duì)應(yīng)的輸入輸出流，底層就通過(guò)調(diào)用native方法來(lái)調(diào)用libhadoop中的library，然后libhadoop會(huì)調(diào)用操作系統(tǒng)的Domain Socket的相關(guān)API實(shí)現(xiàn)Domain Socket 通信。

DataNode啟動(dòng)的時(shí)候，會(huì)創(chuàng)建DataXceiverServer，用來(lái)作為服務(wù)器，用于接收/發(fā)送數(shù)據(jù)塊。 創(chuàng)建它是為了偵聽來(lái)自客戶端或其他 DataNode 的請(qǐng)求。它封裝了一個(gè)PeerServer的實(shí)現(xiàn)。目前，傳統(tǒng)的PeerServer的實(shí)現(xiàn)是TcpPeerServer，用來(lái)接收TCP/IP的通信請(qǐng)求，后來(lái)有了ShortCircuit讀，就有了基于Domain Socket的實(shí)現(xiàn)DomainPeerServer。
DataXceiverServer運(yùn)行起來(lái)以后，會(huì)通過(guò)封裝的PeerServer實(shí)現(xiàn)開始監(jiān)聽請(qǐng)求，當(dāng)監(jiān)聽到請(qǐng)求到來(lái)，就通過(guò)維護(hù)的線程池來(lái)運(yùn)行對(duì)應(yīng)的請(qǐng)求處理。
在DataNode啟動(dòng)的時(shí)候，創(chuàng)建DataXceiverServer和DomainPeerServer，就是基于Domain Socket的文件路徑，進(jìn)行綁定(bind())和監(jiān)聽。

// -------------------- DomainPeer.java --------------------
@InterfaceAudience.Private
public class DomainPeerServer implements PeerServer {
  static final Logger LOG = LoggerFactory.getLogger(DomainPeerServer.class);
  private final DomainSocket sock;
  public DomainPeerServer(String path, int port) 
      throws IOException {
    this(DomainSocket.bindAndListen(DomainSocket.getEffectivePath(path, port)));
  }

// -------------------- DomainSocket.java --------------------
  /**
   * Create a new DomainSocket listening on the given path.
   *
   * @param path         The path to bind and listen on.
   * @return             The new DomainSocket.
   */
  public static DomainSocket bindAndListen(String path) throws IOException {
    int fd = bind0(path); // 綁定到我們配置的DomainSocket 路徑上
    return new DomainSocket(path, fd);
  }

  //  綁定到對(duì)應(yīng)的Socket Path上去
  private static native int bind0(String path) throws IOException;

完成了綁定(bind())和被動(dòng)監(jiān)聽(listen())，會(huì)返回一個(gè)int類型的文件描述符表的索引值，這個(gè)索引值就如同一個(gè)Linux操作系統(tǒng)的文件描述符表的索引，代表了打開這個(gè)文件的文件描述符。

DataNode啟動(dòng)的時(shí)候完成了對(duì)指定的Socket Path的綁定和被動(dòng)監(jiān)聽，那么就開始進(jìn)入阻塞等待請(qǐng)求的監(jiān)聽(accept())階段（流程跟TCP/IP的Client-Server連接方式是一樣的），一旦有客戶端請(qǐng)求過(guò)來(lái)，就跟客戶端建立通信：

// -------------------- DataXceiverServer.java --------------------
 @Override
  public void run() {
    while (datanode.shouldRun && !datanode.shutdownForUpgrade) {
      try {
        peer = peerServer.accept();// 阻塞等待這個(gè)socket上的連接請(qǐng)求，可能來(lái)自本機(jī)上的一個(gè)或者多個(gè)客戶端
        // 阻塞結(jié)束，說(shuō)明接受了連接請(qǐng)求
        .......
        // 開始使用獨(dú)立線程處理請(qǐng)求
        new Daemon(datanode.threadGroup,
            DataXceiver.create(peer, datanode, this))
            .start();

// -------------------- DomainSocket.java --------------------
  /**
   * Accept a new UNIX domain connection.
   *
   * This method can only be used on sockets that were bound with bind().
   *
   * @return                The new connection.
   * @throws IOException    If there was an I/O error performing the accept--
   *                        such as the socket being closed from under us.
   *                        Particularly when the accept is timed out, it throws
   *                        SocketTimeoutException.
   */
  public DomainSocket accept() throws IOException {
    refCount.reference();
    boolean exc = true;
    try {
      // 基于剛剛已經(jīng)完成綁定和監(jiān)聽的fd，開始進(jìn)入accept狀態(tài)，接收客戶端的連接請(qǐng)求
      DomainSocket ret = new DomainSocket(path, accept0(fd));
      exc = false;
      return ret;
    } finally {
      unreference(exc);
    }
  
  // 基于剛剛已經(jīng)完成綁定和監(jiān)聽的fd，開始進(jìn)入accept狀態(tài)，接收客戶端的連接請(qǐng)求
  private static native int accept0(int fd) throws IOException;

當(dāng)accept()成功，就會(huì)基于accept()調(diào)用返回的文件描述符（一個(gè)整數(shù)，這個(gè)整數(shù)此時(shí)代表了和當(dāng)前客戶端的一對(duì)一連接），創(chuàng)建跟當(dāng)前客戶端對(duì)應(yīng)的java.io.InputStream(實(shí)現(xiàn)類為DomainSocket.DomainInputStream)和java.io.OutputStream（DomainSocket.DomainOutputStream）。有了基于Domain Socket的InputStream和OutputStream的具體實(shí)現(xiàn)，那么后續(xù)的發(fā)送和接收數(shù)據(jù)就只需要基于構(gòu)建的InputStream和OutputStream進(jìn)行了，不需要關(guān)心這個(gè)InputStream和OutputStream背后是具體實(shí)現(xiàn)TCP/IP還是DomainSocket了。
基于Domain Socket創(chuàng)建的InputStream和OutputStream進(jìn)行數(shù)據(jù)通信的代碼如下：

// -------------------- DomainSocket.DomainInputStream --------------------
  /**
   * Input stream for UNIX domain sockets.
   */
  public class DomainInputStream extends InputStream {
    .....
    @Override
    public int read() throws IOException {
      ....
      try {
        byte b[] = new byte[1];
        // 調(diào)用native方法，基于已經(jīng)建立的DomainSocket連接，將fd(文件描述符對(duì)應(yīng)的打開的文件)對(duì)應(yīng)的socket中的信息讀入到byte數(shù)組中，實(shí)現(xiàn)通信的接收操作
        int ret = DomainSocket.readArray0(DomainSocket.this.fd, b, 0, 1);
        ......
    }

// -------------------- DomainSocket.DomainOutputStream --------------------
  /**
   * Output stream for UNIX domain sockets.
   */
  @InterfaceAudience.LimitedPrivate("HDFS")
  public class DomainOutputStream extends OutputStream {
    ....
    @Override
    public void write(int val) throws IOException {
      ......
        byte b[] = new byte[1];
        b[0] = (byte)val;
        // 基于已經(jīng)建立的DomainSocket連接，將b中的數(shù)據(jù)寫入到fd(文件描述符對(duì)應(yīng)的打開的文件)對(duì)應(yīng)的socket中，實(shí)現(xiàn)通信的發(fā)送操作
        DomainSocket.writeArray0(DomainSocket.this.fd, b, 0, 1);
        exc = false;
      } ....
    }

3. 客戶端基于DomainSocket連接建立
   傳統(tǒng)的TCP/IP的連接建立，是通過(guò)連接到服務(wù)器端的IP地址和端口實(shí)現(xiàn)通信。Domain Socket的客戶端的連接，是通過(guò)連接到上面配置的文件路徑來(lái)進(jìn)行通信。

// -------------------- BlockReaderFactory.java --------------------
  public DomainSocket createSocket(PathInfo info, int socketTimeout) {
    ....
    try {
      // 通過(guò)dfs.domain.socket.path來(lái)建立連接
      sock = DomainSocket.connect(info.getPath());
      sock.setAttribute(DomainSocket.RECEIVE_TIMEOUT, socketTimeout);
      ......

// -------------------- DomainSocket.java --------------------
  /**
   * Create a new DomainSocket connected to the given path.
   *
   * @param path              The path to connect to.
   * @throws IOException      If there was an I/O error performing the connect.
   *
   * @return                  The new DomainSocket.
   */
  public static DomainSocket connect(String path) throws IOException {
    int fd = connect0(path);
    return new DomainSocket(path, fd);
  }
  // 建立DomainSocket 連接的native方法，這個(gè)native方法的具體調(diào)用定義在libhadoop的本地代碼庫(kù)中
  private static native int connect0(String path) throws IOException;

可以看到，在客戶端， 其實(shí)就是通過(guò)提供的一個(gè)本地文件路徑，同Server端建立連接。連接建立以后，會(huì)返回一個(gè)int類型的文件描述符表的索引值，剛才提到過(guò)，這個(gè)索引值就如同一個(gè)Linux操作系統(tǒng)的文件描述符表的索引，代表了打開這個(gè)文件的文件描述符。

為了向DataNode發(fā)送數(shù)據(jù)，Client把剛剛基于DomainSocket構(gòu)建好的OutputStream進(jìn)行適當(dāng)封裝，基于這個(gè)封裝好的OutputStream把數(shù)據(jù)發(fā)送出去。為了讀取DataNode返回的數(shù)據(jù)，Client會(huì)基于DomainSocket 構(gòu)建好的InputStream，讀取本地的DataNode從DomainSocket發(fā)送過(guò)來(lái)的數(shù)據(jù)：

----------------------- DfsClientShmManager.java --------------------------
// 通過(guò)OutputStream來(lái)接收數(shù)據(jù)
// peer.getOutputStream()就是我們剛剛基于DomainSocket構(gòu)建好的一個(gè)OutputStream的實(shí)現(xiàn)      
final DataOutputStream out =
          new DataOutputStream(
              new BufferedOutputStream(peer.getOutputStream())); // 基于DomainSocket與本地的DataNode進(jìn)行通信
      new Sender(out).requestShortCircuitShm(clientName);

      // 通過(guò)InputStream來(lái)接收數(shù)據(jù)
      ShortCircuitShmResponseProto resp =
          ShortCircuitShmResponseProto.parseFrom(
            PBHelperClient.vintPrefixed(peer.getInputStream())); // 接收本地DataNode基于DomainSocket發(fā)送過(guò)來(lái)的細(xì)膩

1.2 關(guān)于內(nèi)存映射(MMAP)

跟Domain Socket一樣，內(nèi)存映射 (mmap) 并不是Hadoop中的概念，而是Unix操作系統(tǒng)提供的功能。Hadoop通過(guò)自己的libhadoop可以使用mmap相關(guān)的接口調(diào)用，從而使用操作系統(tǒng)的mmap功能。

什么是MMAP

跟Domain Socket一樣，內(nèi)存映射 (mmap()) 也是操作系統(tǒng)提供的一項(xiàng)功能，它將輔助存儲(chǔ)上的文件內(nèi)容映射到程序的內(nèi)存地址空間。 然后，程序通過(guò)指針訪問頁(yè)面，就好像文件完全駐留在內(nèi)存中一樣。 僅當(dāng)程序引用頁(yè)面時(shí)，操作系統(tǒng)才會(huì)透明地加載頁(yè)面，并在內(nèi)存填滿時(shí)自動(dòng)逐出頁(yè)面。
這里關(guān)鍵是操作系統(tǒng)提供的 mmap()/munmap()函數(shù)，mmap()可以把磁盤文件的一部分直接映射到內(nèi)存，這樣文件中的位置直接就有對(duì)應(yīng)的地址，對(duì)文件的讀寫可以直接用指針來(lái)做而不需要read/write函數(shù)。 munmap()就是解除映射。示意圖如下：

圖-1 ： mmap示意圖

總結(jié)來(lái)說(shuō)，常規(guī)文件操作為了提高讀寫效率和保護(hù)磁盤，使用了頁(yè)緩存機(jī)制。這樣造成讀文件時(shí)需要先將文件頁(yè)從磁盤拷貝到頁(yè)緩存中，由于頁(yè)緩存處在內(nèi)核空間，不能被用戶進(jìn)程直接尋址，所以還需要將頁(yè)緩存中數(shù)據(jù)頁(yè)再次拷貝到內(nèi)存對(duì)應(yīng)的用戶空間中。這樣，通過(guò)了兩次數(shù)據(jù)拷貝，才能完成文件讀取。寫操作也是一樣，待寫入的buffer在內(nèi)核空間不能直接訪問，必須要先拷貝至內(nèi)核空間，再寫回磁盤中（延遲寫回），也是需要兩次數(shù)據(jù)拷貝。

相比之下，而使用mmap()操作文件中，創(chuàng)建新的虛擬內(nèi)存區(qū)域，以及建立文件磁盤地址和虛擬內(nèi)存區(qū)域映射這兩步，沒有任何文件拷貝操作。而之后訪問數(shù)據(jù)時(shí)發(fā)現(xiàn)內(nèi)存中并無(wú)數(shù)據(jù)而發(fā)起的缺頁(yè)異常過(guò)程，可以通過(guò)已經(jīng)建立好的映射關(guān)系，只使用一次數(shù)據(jù)拷貝，就從磁盤中將數(shù)據(jù)傳入內(nèi)存的用戶空間中，供進(jìn)程使用。即少了一次數(shù)據(jù)拷貝，因此效率比常規(guī)文件讀取要高。因此，mmap()一般具有以下優(yōu)點(diǎn)：

1. 相比于普通文件讀寫，減少了數(shù)據(jù)拷貝次數(shù)，用一次內(nèi)存讀寫取代了平常的I/O讀寫方式，讀寫效率更高
2. 減少了復(fù)雜的用戶態(tài)和內(nèi)核臺(tái)的切換
3. 有助于實(shí)現(xiàn)高效的數(shù)據(jù)傳輸，一定程度上緩解了內(nèi)存不足的問題

MMAP在ShortCircuit中的作用是什么

基于上文介紹的mmap()的文件讀取方式，在短路讀的場(chǎng)景中，主要有兩個(gè)地方用到了短路讀：

1. 通過(guò)短路讀直接讀取塊文件，而不再使用傳統(tǒng)讀寫方式。顯然，如上文所述，這種使用短路讀方式更高效。
2. 通過(guò)mmap的方式實(shí)現(xiàn)下文要講的共享內(nèi)存。即，客戶端和NodeNode同時(shí)mmap到一個(gè)文件，由于mmap會(huì)在內(nèi)存中建立文件映射，因此兩個(gè)進(jìn)程操作這個(gè)文件，就相當(dāng)于在同時(shí)操作同一內(nèi)存區(qū)域，從而實(shí)現(xiàn)內(nèi)存共享。

1.3 關(guān)于共享內(nèi)存(Shared Memory)

什么是共享內(nèi)存

共享內(nèi)存是進(jìn)程間通信的方法，即在同時(shí)運(yùn)行的程序之間交換數(shù)據(jù)的方法。一個(gè)進(jìn)程將在RAM中創(chuàng)建一個(gè)其他進(jìn)程可以訪問的區(qū)域。由于兩個(gè)進(jìn)程可以像訪問自身內(nèi)存一樣訪問共享內(nèi)存區(qū)域，因此是一種非?？焖俚耐ㄐ欧绞?。但是它的擴(kuò)展性較差，例如通信必須在同一臺(tái)機(jī)器上運(yùn)行。而且必須要避免如果共享內(nèi)存的進(jìn)程在不同的CPU上運(yùn)行，并且底層架構(gòu)不是緩存一致的。

傳統(tǒng)的不共享內(nèi)存的文件讀取方式和共享內(nèi)存的文件讀取如下面兩張圖所示：
不共享內(nèi)存的讀取方式如下圖2-1：

圖-2-1 不共享內(nèi)存的多進(jìn)程打開同一文件的方式

共享內(nèi)存的讀取方式如下圖2-2：

圖-2-2 共享內(nèi)存的多進(jìn)程打開同一文件的方式

共享內(nèi)存在Hadoop短路讀中的使用

在Hadoop中，共享內(nèi)存主要用在兩個(gè)地方：

1. 存放元數(shù)據(jù)信息，下文會(huì)講到，主要是客戶端和DataNode進(jìn)行replica相關(guān)的元數(shù)據(jù)信息的共享。
2. 通過(guò)共享內(nèi)存的方式讀取replica。下文會(huì)詳細(xì)介紹，通過(guò)共享內(nèi)存的方式，那些已經(jīng)被DataNode的緩存系統(tǒng)緩存到內(nèi)存中的replica，客戶端可以直接使用了，無(wú)需再?gòu)拇疟P讀取replica。

共享內(nèi)存和mmap的區(qū)別和聯(lián)系

上面說(shuō)過(guò)，共享內(nèi)存是進(jìn)程間通信的一種方式，允許兩個(gè)或多個(gè)進(jìn)程共享一給定的內(nèi)存區(qū)域，因此數(shù)據(jù)不需要在內(nèi)存中或者從內(nèi)存到磁盤的重復(fù)拷貝，所以是很快的一種進(jìn)程間通信機(jī)制。共享內(nèi)存可以通過(guò)systemV共享內(nèi)存機(jī)制實(shí)現(xiàn)（本文不做介紹），也可以通過(guò)mmap()映射普通文件 （hadoop采用的共享內(nèi)存方式）機(jī)制實(shí)現(xiàn)。
但是，Hadoop中的共享內(nèi)存是基于mmap實(shí)現(xiàn)的：
我們?cè)谏衔闹薪榻B了DataNode在建立共享內(nèi)存的時(shí)候，會(huì)先為該客戶端創(chuàng)建一個(gè)本地磁盤文件，然后通過(guò)mmap()的方式打開文件，隨后把打開文件的FileInputStream發(fā)送給客戶端，客戶端再以同樣的方式通過(guò)mmap()打開文件，這樣，客戶端和DataNode進(jìn)程都對(duì)同一文件以R/W的方式進(jìn)行mmap()，因此它們其實(shí)在操作該文件在內(nèi)存中的同一塊區(qū)域，因此實(shí)現(xiàn)了共享內(nèi)存。

2. ShortCircuit Read(短路讀)的基本流程

我們都知道，客戶端在從一個(gè)DataNode讀取block的時(shí)候，還有一段跟NameNode的溝通過(guò)程，實(shí)現(xiàn)從文件到replica、以及replca的定位的過(guò)程。這個(gè)過(guò)程不是本文討論的范圍。本文不介紹前期跟NameNode的交互過(guò)程，而是直接假設(shè)客戶端已經(jīng)從NameNode獲取了Block的全部信息(block id, DataNode地址等)，開始和DataNode建立連接，讀取塊數(shù)據(jù)。
短路讀的整個(gè)流程簡(jiǎn)化起來(lái)如圖-3所示。

1. 建立Socket連接：客戶端和DataNode先通過(guò)Domain Socket建立通信。但是這種通信僅僅用于少量API的調(diào)用，replica的讀取流不會(huì)使用Domain Socket
2. 建立和協(xié)商共享內(nèi)存區(qū)域：客戶端通過(guò)Domain Socket向DataNode申請(qǐng)共享內(nèi)存，DataNode創(chuàng)建共享內(nèi)存，Client在共享內(nèi)存中創(chuàng)建Slot。注意，這段共享內(nèi)存和其中的Slot是存放replica的元數(shù)據(jù)信息，不存儲(chǔ)replica數(shù)據(jù)本身。
3. 申請(qǐng)塊文件描述符： 客戶端通過(guò)Domain Socket向DataNode申請(qǐng)replica的輸入流，DataNode將replica的輸入流
4. 讀取塊文件：客戶端將輸入流轉(zhuǎn)換為mmap()的讀取方式，或者直接通過(guò)傳統(tǒng)方式讀取replica

圖-3 ： 短路讀基本流程

在短路讀完成以后，整個(gè)內(nèi)存的數(shù)據(jù)結(jié)構(gòu)如圖-4所示。客戶端和DataNode的共享內(nèi)存用來(lái)存放Slot，每個(gè) Slot保存了一個(gè)replica的元數(shù)據(jù)信息(元數(shù)據(jù)信息的具體內(nèi)容下文介紹)。

這里必須區(qū)分JVM中的Slot和共享內(nèi)存的Slot?？蛻舳撕?code>DataNode的JVM中的Slot是Java對(duì)象，對(duì)象中放的是共享內(nèi)存中的Slot的地址，即JVM中的Slot對(duì)象是共享內(nèi)存中的Slot的引用，它指向了共享內(nèi)存的Slot。共享內(nèi)存中的Slot的大小為64B，一個(gè)Segment大小為8KB，因此一個(gè)共享Segment可以存放128個(gè)Slot，對(duì)應(yīng)128個(gè)replica。

圖-4： 短路讀的內(nèi)存示意圖

圖-5展示了客戶端和DataNode在進(jìn)行塊讀取之前建立短路讀和通過(guò)域套接字進(jìn)行通信的基本過(guò)程。主要做了以下幾件事：

• 客戶端是否選擇短路讀：對(duì)于每一次讀取，都會(huì)優(yōu)先考慮短路讀，短路讀無(wú)法實(shí)現(xiàn)則進(jìn)行域套接字讀，如果還是無(wú)法實(shí)現(xiàn)，則進(jìn)行傳統(tǒng)的TCP/IP讀。
• 服務(wù)器端啟動(dòng)短路讀的監(jiān)聽服務(wù)：DataNode啟動(dòng)的時(shí)候，如果配置為需要進(jìn)行短路讀或域套接字讀，則會(huì)為之啟動(dòng)一個(gè)獨(dú)立的DataXceiverServer，專門負(fù)責(zé)短路讀和域套接字讀的連接和通信操作。
• 客戶端通過(guò)Domain Socket和服務(wù)器端DataNode建立Domain Socket 連接
• 基于Domain Socket連接建立共享內(nèi)存（Shared Memory），用來(lái)存放后來(lái)的塊的元數(shù)據(jù)信息

圖-4： 客戶端建立短路讀連接的基本流程

當(dāng)上圖的流程都成功走完，才開始進(jìn)行塊的讀取。我們?cè)谙挛臅?huì)詳細(xì)介紹。

1.1 客戶端的讀取策略選擇 – 怎樣才會(huì)選擇短路讀

目前，客戶端的三種讀取策略如下（選取的優(yōu)先級(jí)從高到低）：

1. 短路讀(Short Circuit Read， 優(yōu)先級(jí)最高)
   通過(guò)域套接字Domain Socket與DataNode進(jìn)行前期通信，然后通過(guò)Linux mmap()和共享內(nèi)存，繞開DataNode進(jìn)行塊的讀取。這是本文詳細(xì)討論的內(nèi)容。
2. 域套接字讀(Domain Socket Read)
   通過(guò)域套接字與DataNode通信，包括信息溝通和replica的讀取。DataNode全程參與replica讀取，replica的讀取流會(huì)經(jīng)過(guò)DataNode。
3. 網(wǎng)絡(luò)讀(TCP Read，優(yōu)先級(jí)最低)
   通過(guò)TCP/IP協(xié)議與DataNode通信，包括信息溝通和replica的讀取。DataNode全程參與replica讀取，replica的讀取流會(huì)經(jīng)過(guò)DataNode。

選取策略為：

• 如果dfs.client.read.shortcircuit=true，并且當(dāng)前這個(gè)block的條件允許短路讀(這個(gè)block不是under construction的狀態(tài))， 那么會(huì)嘗試建立短路讀連接。顯然，在建立短路讀連接和進(jìn)行讀block以前的準(zhǔn)備工作有任何的失敗，都不會(huì)進(jìn)行短路讀。
• 短路讀構(gòu)建失敗，如果客戶端配置dfs.client.domain.socket.data.traffic為true，那么會(huì)退化到嘗試使用Domain Socket讀。顯然，在建立域套接字讀的前期準(zhǔn)備過(guò)程中發(fā)生任何失敗，都不會(huì)進(jìn)行域套接字讀。
• 以上嘗試都失敗，退化到傳統(tǒng)TCP/IP網(wǎng)絡(luò)讀

Domain Socket讀(域套接字讀)和Short Circuit讀（短路讀）的區(qū)別和聯(lián)系
域套接字讀和短路讀，都使用套做系統(tǒng)的域套接字（DomainSocket）和DataNode通信，這意味著兩種讀方式都只能和同一機(jī)器的DataNode進(jìn)行通信。 但是，從Hadoop分類來(lái)講，域套接字讀卻和TCP/IP的讀劃分在了一起，被認(rèn)為是Remote Read，而只有ShortCircuit Read才被認(rèn)為是 local read， 這是因?yàn)椋?br> 域套接字讀和TCP/IP的唯一區(qū)別只是Socket建立的方式不同，前者和DataNode的連接封裝在DomainPeer中，后者和DataNode的通信封裝在BasicInetPeer/NioInetPeer中，一旦Socket通信建立，那么域套接字讀讀和TCP/IP 讀的邏輯完全一致，都是直接和DataNode進(jìn)行通信，都不使用mmap、共享內(nèi)存等特性，因此它們都被hadoop劃分為遠(yuǎn)程讀。而短路讀的流程則完全不同，雖然也會(huì)通過(guò)Domain Socket的方式和本地DataNode建立連接，但是后續(xù)的讀塊流程則完全發(fā)生了不同，不在經(jīng)過(guò)DataNode，使用了mmap和共享內(nèi)存等特性，這個(gè)我們會(huì)在后面的介紹中看到。由于域套接字讀和短路讀都是基于Unix Domain Socket建立通信，因此域套接字讀和短路讀在DataNode這一端都被封裝在同一個(gè)DataXceiverServer中，而TCP/IP讀由于使用的是網(wǎng)絡(luò)通信，因此其單獨(dú)封裝在另外一個(gè)DataXceiverServer中。

1.2 DataNode啟動(dòng)時(shí)監(jiān)聽短路讀DomainSocket

DataNode這一端，通過(guò)PeerServer封裝了基于Socket的數(shù)據(jù)讀寫的基本操作，比如，綁定，監(jiān)聽，關(guān)閉等等。目前支持了兩種PeerServer的實(shí)現(xiàn)，TCPPeerServer 和 DomainPeerServer, 顧名思義，分別用來(lái)支持基于TCP/IP的數(shù)據(jù)通信(TCP/IP網(wǎng)絡(luò)讀)和基于Domain Socket的數(shù)據(jù)通信(短路讀和域套接字讀)。負(fù)責(zé)基于PeerServer的實(shí)現(xiàn)來(lái)運(yùn)行的，是DataXceiver，而DataXceiver被DataXCeiverServer調(diào)度。類關(guān)系如圖-5所示：

圖-5: DataNode端的數(shù)據(jù)通信相關(guān)的類關(guān)系圖

下面的代碼顯示了DataNode啟動(dòng)的時(shí)候?yàn)?code>TCP/IP和為短路讀以及域套接字讀啟動(dòng)啟動(dòng)DataXceiverServer的過(guò)程：

------------------------------DataNode.java--------------------------
private void initDataXceiver() throws IOException {
    // find free port or use privileged port provided
    TcpPeerServer tcpPeerServer;
    
    streamingAddr = tcpPeerServer.getStreamingAddr();
    this.threadGroup = new ThreadGroup("dataXceiverServer");
    xserver = new DataXceiverServer(tcpPeerServer, getConf(), this); // 創(chuàng)建基于`TCP/IP`的TcpPeerServer的DataXceiverServer
    this.dataXceiverServer = new Daemon(threadGroup, xserver);
    this.threadGroup.setDaemon(true); // auto destroy when empty
    // 如果配置了shortCircuit讀，那么會(huì)創(chuàng)建一個(gè)單獨(dú)的Daemon， 封裝了一個(gè)DataXceiver,這個(gè)DataXceiver專門
    // 處理shortCircuit
    if (getConf().getBoolean(
        HdfsClientConfigKeys.Read.ShortCircuit.KEY,
        HdfsClientConfigKeys.Read.ShortCircuit.DEFAULT) ||
        getConf().getBoolean(
            HdfsClientConfigKeys.DFS_CLIENT_DOMAIN_SOCKET_DATA_TRAFFIC,
            HdfsClientConfigKeys
              .DFS_CLIENT_DOMAIN_SOCKET_DATA_TRAFFIC_DEFAULT)) {
      DomainPeerServer domainPeerServer = // 基于DomainSocket的PeerServer實(shí)現(xiàn),這個(gè)PeerServer既用于ShortCircuit讀，也用于DomainSocket讀
                getDomainPeerServer(getConf(), streamingAddr.getPort());
      if (domainPeerServer != null) { // 創(chuàng)建基于DomainSocket的DomainPeerServer的DataXceiverServer
        this.localDataXceiverServer = new Daemon(threadGroup,
            new DataXceiverServer(domainPeerServer, getConf(), this));
      }

可以看到，DataNode啟動(dòng)的時(shí)候，會(huì)首先無(wú)條件啟動(dòng)基于TCP/IP的DataXceiverServer，以接收以TCP/IP為信道的網(wǎng)絡(luò)請(qǐng)求，這個(gè)不可以disable。同時(shí)，根據(jù)系統(tǒng)配置，選擇是否啟動(dòng)基于Domain Socket的DataXceiverServer，啟動(dòng)的條件是：如果用戶配置為enable 短路讀（dfs.client.read.shortcircuit）或者 enable 了域套接字讀(dfs.client.domain.socket.data.traffic)，那么就會(huì)啟動(dòng)一個(gè)使用Linux Domain Socket作為socket實(shí)現(xiàn)的DataXceiverServer。

------------------------------DataXceiverServer.java--------------------------
/**
 * Server used for receiving/sending a block of data. This is created to listen
 * for requests from clients or other DataNodes. This small server does not use
 * the Hadoop IPC mechanism.
 */
class DataXceiverServer implements Runnable {
 ......
  @Override
  public void run() {
    Peer peer = null;
    while (datanode.shouldRun && !datanode.shutdownForUpgrade) {
      try {
        peer = peerServer.accept();// 阻塞等待這個(gè)socket上的連接請(qǐng)求，可能來(lái)自本機(jī)上的一個(gè)或者多個(gè)客戶端
        // 阻塞結(jié)束，說(shuō)明接受了連接請(qǐng)求
        .....
                // 構(gòu)造一個(gè)DataXceiver，使用線程池中的獨(dú)立線程處理請(qǐng)求
        new Daemon(datanode.threadGroup,
            DataXceiver.create(peer, datanode, this))
            .start();

------------------------------DataXceiver.java--------------------------
/**
 * Thread for processing incoming/outgoing data stream.
 */
class DataXceiver extends Receiver implements Runnable {
  .......
  
  /**
   * Read/write data from/to the DataXceiverServer.
   */
  @Override
  public void run() {
    try {
      dataXceiverServer.addPeer(peer, Thread.currentThread(), this);
      InputStream input = socketIn;
      try {
        IOStreamPair saslStreams = datanode.saslServer.receive(peer, socketOut,
          socketIn, datanode.getXferAddress().getPort(),
          datanode.getDatanodeId());
        input = new BufferedInputStream(saslStreams.in,
            smallBufferSize);
          op = readOp(); // 從InputStream中提取出操作類型
        ......
        processOp(op); // 根據(jù)操作類型，交給不同的方法去處理

從上面的實(shí)現(xiàn)可以看到，DataXceiverServer啟動(dòng)以后，會(huì)通過(guò)accept()不斷阻塞等待新的請(qǐng)求，一旦監(jiān)聽到了新的請(qǐng)求，就會(huì)啟動(dòng)一個(gè)DataXceiver來(lái)負(fù)責(zé)處理這個(gè)請(qǐng)求，每一個(gè)DataXceiver都是一個(gè)Runnable，使用線程池對(duì)所有的DataXceiver進(jìn)行處理。
DataXceiver的請(qǐng)求處理邏輯就是首先會(huì)從Socket的InputStream中讀取操作類型，這些操作類型如下：

------------------------------------Op.java----------------------------------------
public enum Op {
  WRITE_BLOCK((byte)80), // 寫數(shù)據(jù)塊
  READ_BLOCK((byte)81), // 讀數(shù)據(jù)塊
  READ_METADATA((byte)82), // 讀數(shù)據(jù)塊元數(shù)據(jù)

  COPY_BLOCK((byte)84), // 拷貝數(shù)據(jù)塊
  BLOCK_CHECKSUM((byte)85), // 計(jì)算數(shù)據(jù)塊的crc校驗(yàn)和
  TREPLACE_BLOCK((byte)83),RANSFER_BLOCK((byte)86),
  REQUEST_SHORT_CIRCUIT_FDS((byte)87), // 申請(qǐng)短路讀的塊文件描述符
  RELEASE_SHORT_CIRCUIT_FDS((byte)88), // 釋放短路讀的塊文件描述符
  REQUEST_SHORT_CIRCUIT_SHM((byte)89), // 申請(qǐng)短路讀的共享內(nèi)存塊
  BLOCK_GROUP_CHECKSUM((byte)90), // 請(qǐng)求BlockGroup的checksum(用在Stripped Block，即糾刪碼中)
  CUSTOM((byte)127);

然后根據(jù)op的類型，交給不同的方法去處理：

------------------------------------Receiver.java----------------------------------------
 /** Process op by the corresponding method. */
  protected final void processOp(Op op) throws IOException {
    switch(op) {
    case READ_BLOCK:
      opReadBlock();
      break;
    ......
    case REQUEST_SHORT_CIRCUIT_FDS:
      opRequestShortCircuitFds(in);
      break;
    case RELEASE_SHORT_CIRCUIT_FDS:
      opReleaseShortCircuitFds(in);
      break;
    case REQUEST_SHORT_CIRCUIT_SHM:
      opRequestShortCircuitShm(in);
      break;
    default:
      throw new IOException("Unknown op " + op + " in data stream");
    }
  }

顯然，跟短路讀相關(guān)的就三種操作，REQUEST_SHORT_CIRCUIT_FDS，RELEASE_SHORT_CIRCUIT_FDS和REQUEST_SHORT_CIRCUIT_SHM。后面會(huì)詳細(xì)講解。

1.3 客戶端申請(qǐng)共享內(nèi)存并在共享內(nèi)存中分配Slot

上文介紹了，DataNode啟動(dòng)的時(shí)候就已經(jīng)在配置好的在Domain Socket上的監(jiān)聽?？蛻舳藰?gòu)造時(shí)，會(huì)進(jìn)行以下工作：

• 與DataNode基于DomainSocket 建立連接
• 申請(qǐng)共享內(nèi)存
• 在共享內(nèi)存中分配Slot，并通過(guò)Slot與DataNode進(jìn)行replica相關(guān)信息的溝通同步

下文詳細(xì)解釋。

1.3.1 與DataNode 建立Socket 通信

在分配slot之前，dfs客戶端會(huì)先通過(guò)Domain Socket與DataNode建立連接。
客戶端和DataNode之間的連接都保存在PeerCache中，包括基于Domain Socket的Peer和基于常規(guī)的TCP的Peer。建立連接時(shí)，會(huì)以DataNodeID和連接類型(Domain Socket還是 TCP？)先從PeerCache獲取。如果緩存中沒有，則創(chuàng)建Domain Socket客戶端并建立連接，并將創(chuàng)建好的Peer放到cache中以便重用(比如該客戶端可能會(huì)從這個(gè)本機(jī)的DataNode上讀取其他數(shù)據(jù)塊，那么可以不用再建立新的Domain Socket連接)。同時(shí)，dfs.client.cached.conn.retry約束了該客戶端最多使用緩存中的peer的次數(shù)為3次。

--------------------------------------PeerCache.java------------------------------------------
public class PeerCache {
  private static class Key {
    final DatanodeID dnID; // datanode  id
    final boolean isDomain; // 是否是 domain socket 的peer，目前有兩種peer，基于domain socket的peer和 基于Tcp的peer

------------------------------------DomainSocketFactory.java----------------------------------------

  /**
   * Get the next DomainPeer-- either from the cache or by creating it.
   *
   * @return the next DomainPeer, or null if we could not construct one.
   */
  private BlockReaderPeer nextDomainPeer() {
    if (remainingCacheTries > 0) { // 最多使用3次peer。由于是一個(gè)client都會(huì)創(chuàng)建一個(gè)BlockReaderFactory,因此，一個(gè)client只能使用3次cache，超過(guò)了3次則必須重新創(chuàng)建socket
      // 如果緩存中有，并且緩存次數(shù)沒有超過(guò)限制，那么從緩存中拿對(duì)應(yīng)的Domain Socket Peer
      Peer peer = clientContext.getPeerCache().get(datanode, true);
      if (peer != null) {
        return new BlockReaderPeer(peer, true);// 這個(gè)BlockReaderPeer是來(lái)自cache
    // 否則，創(chuàng)建一個(gè)新的Domain Socket Peer
    DomainSocket sock = clientContext.getDomainSocketFactory().
        createSocket(pathInfo, conf.getSocketTimeout());
    return new BlockReaderPeer(new DomainPeer(sock), false); // 這個(gè)BlockReaderPeer不是來(lái)自cache
  }

  public DomainSocket createSocket(PathInfo info, int socketTimeout) {
    ...
      // 通過(guò)dfs.domain.socket.path來(lái)建立連接
      sock = DomainSocket.connect(info.getPath());
      .....

下面的代碼中，客戶端通過(guò)調(diào)用一些native方法，調(diào)用到libhadoop中對(duì)應(yīng)方法的library，并最終會(huì)調(diào)用操作系統(tǒng)的Domain Socket相關(guān)api，來(lái)建立socket連接：

// -------------------- DomainSocket.java --------------------
  /**
   * Create a new DomainSocket connected to the given path.
   *
   * @param path              The path to connect to.
   * @throws IOException      If there was an I/O error performing the connect.
   *
   * @return                  The new DomainSocket.
   */
  public static DomainSocket connect(String path) throws IOException {
    int fd = connect0(path);
    return new DomainSocket(path, fd);
  }
  // 建立DomainSocket 連接的native方法，這個(gè)native方法的具體調(diào)用定義在libhadoop的本地代碼庫(kù)中
  private static native int connect0(String path) throws IOException;

1.3.2 客戶端嘗試在Segment中分配Slot，Slot不夠用則申請(qǐng)新的segment

客戶端與DataNode的連接建立完成，開始進(jìn)入分配共享內(nèi)存和Slot的階段，即OP.REQUEST_SHORT_CIRCUIT_SHM的調(diào)用階段。
上文提到過(guò)，共享內(nèi)存中的一個(gè)Slot代表了客戶端需要讀取的一個(gè)replica的元數(shù)據(jù)信息，客戶端和服務(wù)器端的JVM中都有對(duì)應(yīng)的Slot對(duì)象，指向了共享內(nèi)存中的這一塊64B的Slot區(qū)域，都對(duì)這塊區(qū)域進(jìn)行必要的讀寫。

客戶端會(huì)為每個(gè)DataNode構(gòu)建一個(gè)DfsClientShmManager，專門負(fù)責(zé)與這個(gè)DataNode進(jìn)行共享內(nèi)存和分配slot相關(guān)的通信，在DataNode這一端的對(duì)應(yīng)類為ShortCircuitRegistry，負(fù)責(zé)與該客戶端進(jìn)行與共享內(nèi)存相關(guān)的工作。

基本流程如下圖-6所示。

圖-6 Segment申請(qǐng)和Slot分配的基本流程

從上圖可以看到，Segment的創(chuàng)建和Slot的分配流程是：

• 當(dāng)需要讀取一個(gè)replica，客戶端會(huì)查看本地是否還有剩余的分配槽位，有的話直接分配，沒有的話則向DataNode申請(qǐng)新的Segment
• DataNode收到segment申請(qǐng)以后，會(huì)基于mmap()的方式創(chuàng)建一段共享內(nèi)存，然后把共享內(nèi)存的文件打開描述符(fd)返回給客戶端
• 客戶端拿到了fd, 就可以同樣通過(guò)mmap()的方式打開，相當(dāng)于和DataNode共享了內(nèi)存，客戶端在共享內(nèi)存中為replica分配Slot，并通過(guò)Slot和DataNode同步信息。

DfsClientShmManager通過(guò)allocSlot()方法為replica在共享內(nèi)存中創(chuàng)建Slot。有剩余Slot那么就直接分配，沒有剩余Slot就需要先申請(qǐng)共享內(nèi)存，申請(qǐng)到了共享內(nèi)存，再在共享內(nèi)存中創(chuàng)建Slot。

-------------------------------DfsClientShmManager.java-------------------------------
public class DfsClientShmManager implements Closeable {
  .....
  class EndpointShmManager {
    private final DatanodeInfo datanode;
    private final TreeMap<ShmId, DfsClientShm> full = new TreeMap<>();// 已經(jīng)滿了的共享內(nèi)存段
    private final TreeMap<ShmId, DfsClientShm> notFull = new TreeMap<>(); // 還有空閑slot的共享內(nèi)存段
 /**
 分配slot，可能從已有的segment中分配，如果segment沒有空閑slot，那么先申請(qǐng)新的segment，然后分配slot
 */
 Slot allocSlot(DomainPeer peer, MutableBoolean usedPeer,
        String clientName, ExtendedBlockId blockId) throws IOException {
      while (true) { // 不斷循環(huán)等待slot的分配，有可能通過(guò)已有slot的釋放，有可能通過(guò)新的共享內(nèi)存分配slot，成功則退出
        // Try to use an existing slot.
        Slot slot = allocSlotFromExistingShm(blockId); // 當(dāng)前的共享內(nèi)存還有空閑slot可供分配，則直接拿來(lái)分配就行了，無(wú)需申請(qǐng)新的共享內(nèi)存
        if (slot != null) { //客戶端針對(duì)這個(gè)block的slot已經(jīng)存在，則直接返回slot
          return slot;
        }
        // 沒有空閑slot，需要先申請(qǐng)共享內(nèi)存，再在共享內(nèi)存中分配slot
        ......
          try {
            // 沒有空閑slot，那么需要申請(qǐng)一片新的共享內(nèi)存，然后再放置這一塊slot
            shm = requestNewShm(clientName, peer); //申請(qǐng)了一個(gè)新的segment成功，剩下的交給下一輪循環(huán)，從這個(gè)segment中取選擇slot
            if (shm == null) continue;
            // See #{DfsClientShmManager#domainSocketWatcher} for details
            // about why we do this before retaking the manager lock.
            domainSocketWatcher.add(peer.getDomainSocket(), shm);
            // The DomainPeer is now our responsibility, and should not be
            // closed by the caller.
            usedPeer.setValue(true); // 這個(gè)DomainPeer已經(jīng)使用過(guò)了(這個(gè)DomainPeer已經(jīng)用來(lái)申請(qǐng)
   }

allocSlotFromExistingShm()會(huì)嘗試從已有的Segment中分配一個(gè)可用的Slot：

-------------------------------DfsClientShmManager.java------------------------------- 
   /**
    嘗試在當(dāng)前已有的共享內(nèi)存段中分配一個(gè)slot
   */
    private Slot allocSlotFromExistingShm(ExtendedBlockId blockId) {
      if (notFull.isEmpty()) { //從當(dāng)前的DfsClientShmManager中沒有找到任何一個(gè)還有slot位置的DfsClientShm
        return null;
      }
      //找到一個(gè)還有空閑slot的SharedMemory segment
      Entry<ShmId, DfsClientShm> entry = notFull.firstEntry();
      DfsClientShm shm = entry.getValue();
      ShmId shmId = shm.getShmId(); // 這個(gè)共享內(nèi)存塊的id，是由DataNode在創(chuàng)建這個(gè)共享內(nèi)存的時(shí)候分配的
      Slot slot = shm.allocAndRegisterSlot(blockId); // 找到空閑的slot
      .....
      return slot;
    }
 
  /**
  找到一個(gè)可用的slot，構(gòu)造成為Slot對(duì)象，這個(gè)Slot對(duì)象專門維護(hù)了針對(duì)這個(gè)Block的短路讀的元數(shù)據(jù)信息
 */
  synchronized public final Slot allocAndRegisterSlot(
      ExtendedBlockId blockId) {
    // allocatedSlots為標(biāo)記slot可用性的bit數(shù)組，從該數(shù)組中找到第一個(gè)空閑slot的位置
    int idx = allocatedSlots.nextClearBit(0);
    allocatedSlots.set(idx, true); // 標(biāo)記這個(gè)slot為已占用
    // 創(chuàng)建了一個(gè)新的slot， 根據(jù)這個(gè)slot的索引值確定這個(gè)slot相對(duì)于這個(gè)segment的地址偏移
    Slot slot = new Slot(calculateSlotAddress(idx), blockId);
    slot.clear();
    slot.makeValid();
    slots[idx] = slot;
    return slot;
  }

如果無(wú)法從當(dāng)前的segment中獲取一個(gè)空閑slot(所有的slot都已經(jīng)被占用)，那么就嘗試向DataNode申請(qǐng)一個(gè)新的共享內(nèi)存，然后再分配Slot。requestNewShm()就是向DataNode申請(qǐng)共享內(nèi)存。注意，客戶端每次是申請(qǐng)一個(gè) 8KB大小的segment，而不是申請(qǐng)一個(gè)Slot。Segment的申請(qǐng)是客戶端發(fā)起，Segment創(chuàng)建是DataNode操作，而在Segment中分配Slot又是客戶端做的。

/**
向DataNode申請(qǐng)一段新的共享內(nèi)存段，因?yàn)楫?dāng)前已經(jīng)分配的共享內(nèi)存已經(jīng)沒有可用的slot了
*/
private DfsClientShm requestNewShm(String clientName, DomainPeer peer)
        throws IOException {
      final DataOutputStream out =
          new DataOutputStream(
              new BufferedOutputStream(peer.getOutputStream())); // 基于DomainSocket與本地的DataNode進(jìn)行通信
      new Sender(out).requestShortCircuitShm(clientName);
      ShortCircuitShmResponseProto resp =
          ShortCircuitShmResponseProto.parseFrom(
            PBHelperClient.vintPrefixed(peer.getInputStream())); // 接收本地DataNode基于DomainSocket發(fā)送過(guò)來(lái)的回復(fù)
      switch (resp.getStatus()) {
      case SUCCESS: // DataNode為該client成功創(chuàng)建了對(duì)應(yīng)的Segment
        DomainSocket sock = peer.getDomainSocket();
        byte buf[] = new byte[1];
        FileInputStream[] fis = new FileInputStream[1];
        // 在 DataNode端，DataNode負(fù)責(zé)創(chuàng)建好對(duì)應(yīng)的SharedMemory,大小為8192B, 此時(shí)，
        // 客戶端和DataNode端共享了相同的fd, 這個(gè)fd指向了對(duì)應(yīng)的shared memory segment，同時(shí)這個(gè)segment是
        // DN通過(guò)mmap的方式創(chuàng)建的，對(duì)應(yīng)了內(nèi)存中的一塊區(qū)域
        // shm = new RegisteredShm(clientName, shmId, fis, this);
        try {
          DfsClientShm shm =
              new DfsClientShm(PBHelperClient.convert(resp.getId()),
                  fis[0], this, peer);
          LOG.trace("{}: createNewShm: created {}", this, shm);
          return shm;
          ...
    }

下面的代碼是在JVM中的Slot的定義。注意，這是在堆內(nèi)存中的Slot定義，它存放了在共享內(nèi)存中的Slot的地址，而不是共享內(nèi)存的Slot本身。

---------------------------------Slot.java----------------------------------
  public class Slot {
    // slot是否可用的標(biāo)記位
    private static final long VALID_FLAG =          1L<<63;
    // slot的錨定位
    private static final long ANCHORABLE_FLAG =     1L<<62;
    private final long slotAddress;// slot地址，其實(shí)就是該slot在共享內(nèi)存中的地址偏移
    private final ExtendedBlockId blockId; //該slot對(duì)應(yīng)的block

參考圖-4： 短路讀的內(nèi)存示意圖， 可以看到，在共享內(nèi)存中，一個(gè)Slot占用64字節(jié)，第一個(gè)字節(jié)的前32位用來(lái)存放跟Slot相關(guān)的標(biāo)記位，目前只有兩個(gè)標(biāo)記位，后32位用來(lái)存放錨定數(shù)量，剩余字節(jié)為保留字節(jié)。兩個(gè)標(biāo)記為是：

• VALID_FLAG：表示當(dāng)前Slot是否有效。當(dāng)DFSClient 在其共享內(nèi)存區(qū)域之一內(nèi)分配新槽時(shí)，它會(huì)設(shè)置此標(biāo)志。當(dāng)與此槽關(guān)聯(lián)的副本不再有效時(shí)，DataNode 會(huì)清除此標(biāo)志。 當(dāng)客戶端認(rèn)為DataNode不再使用此槽進(jìn)行通信時(shí)，客戶端本身也會(huì)清除此標(biāo)志。

• ANCHORABLE_FLAG:表示槽位對(duì)應(yīng)的副本可以被錨定。DataNode的緩存系統(tǒng)在將該replica緩存的時(shí)候，會(huì)調(diào)用操作系統(tǒng)的mlock()接口，將該block在內(nèi)存中鎖定(錨定，anchor)，然后將該標(biāo)識(shí)位置位。標(biāo)志位置位，對(duì)于客戶端意味著這個(gè)block的讀取已經(jīng)不需要再校驗(yàn)checksum了，因?yàn)镈ataNode進(jìn)行replica的緩存的時(shí)候都校驗(yàn)過(guò)crc了，這時(shí)候客戶端可以通過(guò)零拷貝和免校驗(yàn)讀取(后面會(huì)介紹)該副本?？蛻舳俗x取這樣的副本的時(shí)候，會(huì)將對(duì)應(yīng)的錨定計(jì)數(shù)器+1，釋放的時(shí)候減去1。DataNode只有當(dāng)該副本的錨定數(shù)位0的時(shí)候才會(huì)將該副本從內(nèi)存緩存中刪除，表明沒有客戶端還在使用這塊緩存區(qū)域。

注意，共享內(nèi)存的分配是DataNode進(jìn)行的，但是在共享內(nèi)存區(qū)域中分配Slot是客戶端進(jìn)行的，分配完Slot，客戶端會(huì)把Slot的信息寫入共享內(nèi)存區(qū)域，顯然，這些信息就對(duì)DataNode可見了。

在DataNode端，在收到OP.REQUEST_SHORT_CIRCUIT_SHM請(qǐng)求以后，會(huì)開始創(chuàng)建共享內(nèi)存。下文詳細(xì)介紹。

1.3.3 DataNode通過(guò)mmap創(chuàng)建共享內(nèi)存

在上一節(jié)的圖中展示了DataNode處理共享內(nèi)存的請(qǐng)求和創(chuàng)建共享內(nèi)存的基本流程。

當(dāng)DataNode收到了1.3.2中客戶端OP.REQUEST_SHORT_CIRCUIT_SHM請(qǐng)求以后，會(huì)創(chuàng)建共享內(nèi)存，大小為8KB。由于一個(gè)Slot占用64B，因此一個(gè)Segment一共可以分配8192B / 64B = 128個(gè)Slot。當(dāng)客戶端用盡了Slot，會(huì)再次過(guò)來(lái)申請(qǐng)Segment。

創(chuàng)建共享內(nèi)存是基于mmap實(shí)現(xiàn)的：

• 先創(chuàng)建一個(gè)磁盤文件并打開，拿到文件的文件描述符。這個(gè)磁盤文件與當(dāng)前請(qǐng)求共享內(nèi)存的客戶端一一對(duì)應(yīng)，即DataNode會(huì)給每一個(gè)客戶端創(chuàng)建一個(gè)獨(dú)立的磁盤文件。

• 調(diào)用POXIS.mmap()，即通過(guò)mmap()的方式，將該文件映射為內(nèi)存映射文件，返回值為該內(nèi)存映射文件的地址(baseAddress)。調(diào)用mmap()的時(shí)候，使用R&W的方式，因此無(wú)論是客戶端還是DataNode，都有對(duì)該內(nèi)存區(qū)域的讀寫權(quán)限

• 將剛剛打開的文件的描述符(是調(diào)用mmap()以前的那個(gè)文件描述符)，通過(guò)Domain Socket返回給客戶端

• 客戶端拿到文件描述符以后，通過(guò)同樣的mmap()方式，打開文件。由于客戶端和DataNode基于同一文件的同一描述符通過(guò)mmap()打開文件，內(nèi)存中只有一份拷貝，因此實(shí)現(xiàn)了內(nèi)存共享。
  可以看到，DataNode創(chuàng)建共享內(nèi)存，是在磁盤上創(chuàng)建一個(gè)文件，再把這個(gè)文件通過(guò)mmap()映射到內(nèi)存。因此，這段共享內(nèi)存，是通過(guò)mmap的映射實(shí)現(xiàn)的共享內(nèi)存。

• DataNode服務(wù)端收到客戶端的創(chuàng)建共享內(nèi)存請(qǐng)求

--------------------------DataXceiver.java--------------------------
// 服務(wù)器端處理客戶端的創(chuàng)建共享內(nèi)存請(qǐng)求
@Override
  public void requestShortCircuitShm(String clientName) throws IOException {
      try {
        shmInfo = datanode.shortCircuitRegistry.
            createNewMemorySegment(clientName, sock);共享內(nèi)存創(chuàng)建完畢，返回共享內(nèi)存信息
        // After calling #{ShortCircuitRegistry#createNewMemorySegment}, the
        // socket is managed by the DomainSocketWatcher, not the DataXceiver.
        releaseSocket();
      } 
      ......
      sendShmSuccessResponse(sock, shmInfo); // 把共享內(nèi)存區(qū)域的信息通過(guò)Domain Socket發(fā)送給客戶端
      success = true;

下面的代碼顯示DataNode打開一個(gè)大小為8KB的磁盤文件(文件路徑通過(guò))，獲取該文件的InputStream

// 打開文件，用來(lái)進(jìn)行mmap操作。這里的共享文件路徑是通過(guò)dfs.datanode.shared.file.descriptor.paths配置的，為了提高可用性，我們可以配置多個(gè)文件，DataNode會(huì)選取其中一個(gè)可用的文件，作為基于mmap的共享內(nèi)存背后的文件
-------------------------- SharedFileDescriptorFactory.java --------------------------
  public FileInputStream createDescriptor(String info, int length)
      throws IOException {
    return new FileInputStream( // 在創(chuàng)建shm的時(shí)候，這個(gè)info就是clientname, 因此各個(gè)client的filename 是不同的
            // 這里實(shí)際上是創(chuàng)建了一個(gè)文件
        createDescriptor0(prefix + info, path, length));
  }
  /**
   * Create a file with O_EXCL, and then resize it to the desired size.
   */
  private static native FileDescriptor createDescriptor0(String prefix,
      String path, int length) throws IOException;
}

下面的代碼顯示DataNode將剛剛創(chuàng)建和打開的文件的InputStream通過(guò)mmap的方式映射到內(nèi)存，作為共享內(nèi)存區(qū)域：

// DataNode將打開的文件通過(guò)POXIS.mmap()的native 調(diào)用，在內(nèi)存中創(chuàng)建對(duì)應(yīng)的映射區(qū)
-------------------------- ShortCircuitShm.java --------------------------
  public ShortCircuitShm(ShmId shmId, FileInputStream stream)
        throws IOException {
    // 下面的代碼在客戶端和服務(wù)器端同時(shí)執(zhí)行
    this.shmId = shmId;
    this.mmappedLength = getUsableLength(stream);
    //客戶端和服務(wù)器端都把這個(gè)文件直接映射到自己的進(jìn)程內(nèi)存空間中，避免在讀寫文件過(guò)程中還需要通過(guò)內(nèi)核態(tài)，避免數(shù)據(jù)從文件到內(nèi)核、內(nèi)核到用戶態(tài)的多次拷貝
    // 可以看到，調(diào)用mmap的時(shí)候，對(duì)應(yīng)的權(quán)限是RW,因?yàn)闊o(wú)論是DataNode還是客戶端，都將對(duì)這段內(nèi)存進(jìn)行讀和寫，實(shí)現(xiàn)狀態(tài)同步
    this.baseAddress = POSIX.mmap(stream.getFD(),
        POSIX.MMAP_PROT_READ | POSIX.MMAP_PROT_WRITE, true, mmappedLength);
    this.slots = new Slot[mmappedLength / BYTES_PER_SLOT]; // 默認(rèn)情況下，一個(gè)segment可以存放8192  / 64 個(gè)slot
    this.allocatedSlots = new BitSet(slots.length);//默認(rèn)情況下，所有slot都沒有分配

1.3.4 DataNode將共享內(nèi)存的文件描述符返回給客戶端

服務(wù)器端把在1.3.3中打開文件的InputStream通過(guò)Domain Socket的方式發(fā)送給客戶端。
這是Domain Socket的一個(gè)重要特性，它支持文件描述符的傳遞： 即進(jìn)程A 打開一個(gè)文件，然后將文件描述符通過(guò)Domain Socket發(fā)送給同一機(jī)器的另外一個(gè)進(jìn)程B，實(shí)現(xiàn)文件共享。多個(gè)進(jìn)程基于同一文件的文件打開描述符進(jìn)行mmap內(nèi)存映射，內(nèi)存只會(huì)存在一份文件內(nèi)容拷貝，就是共享內(nèi)存。

----------------------------NewShmInfo.java --------------------------------------
  public static class NewShmInfo implements Closeable {
    private final ShmId shmId;
    private final FileInputStream stream;

    NewShmInfo(ShmId shmId, FileInputStream stream) {
      this.shmId = shmId;
      this.stream = stream; // 基于文件創(chuàng)建的讀數(shù)據(jù)流
    }

------------------------------DataXceiver.java--------------------------------------
 /**
 DataNode直接將基于mmap創(chuàng)建好的共享內(nèi)存的FileInputStream的FD發(fā)送給客戶端，實(shí)現(xiàn)內(nèi)存共享
**/
  private void sendShmSuccessResponse(DomainSocket sock, NewShmInfo shmInfo)
      throws IOException {
    // Send the file descriptor for the shared memory segment.
    byte buf[] = new byte[] { (byte)0 };
    FileDescriptor shmFdArray[] =
        new FileDescriptor[] {shmInfo.getFileStream().getFD()};
    sock.sendFileDescriptors(shmFdArray, buf, 0, buf.length);
  }

1.3.5 客戶端打開文件，寫入Slot數(shù)據(jù)

收到DataNode發(fā)送過(guò)來(lái)的FD以后，客戶端通過(guò)與DataNode一樣的方式，即通過(guò)調(diào)用POXIS.map()，通過(guò)mmap()的方式，將該文件映射為內(nèi)存映射文件，并獲取了該內(nèi)存映射的地址(baseAddress)。由于DataNode和服務(wù)器端使用的是同一個(gè)文件的fd進(jìn)行的POXIS.map()，因此彼此對(duì)應(yīng)了內(nèi)存中的同一塊映射區(qū)域，從而實(shí)現(xiàn)內(nèi)存共享。
如下的類圖-7所示，在客戶端，使用DfsClientShm對(duì)象來(lái)控制和管理一塊共享內(nèi)存區(qū)域，在服務(wù)器端，與之對(duì)等的對(duì)象為ShortCircuitRegistry，它們都是 ShortCircuitShm的子類，一些對(duì)共享內(nèi)存的共同操作封裝在ShortCircuitShm中，比如，將文件描述符通過(guò)mmap()的方式轉(zhuǎn)換為共享內(nèi)存等。

圖-7 客戶端和服務(wù)器端共享內(nèi)存的管理類

至此，客戶端和服務(wù)器端都有了對(duì)同一片內(nèi)存內(nèi)存映射的讀寫權(quán)限，從而實(shí)現(xiàn)了共享內(nèi)存。這個(gè)共享內(nèi)存是基于mmap，即DataNode先打開這個(gè)文件并獲取InputStream, 然后客戶端和服務(wù)器端都同時(shí)基于這個(gè)InputStream, 通過(guò)mmap()的方式獲取到了這片內(nèi)存區(qū)域的操作權(quán)限，并且相互的操作對(duì)對(duì)方可見。

// DataNode將打開的文件通過(guò)POXIS.mmap()的native 調(diào)用，在內(nèi)存中創(chuàng)建對(duì)應(yīng)的映射區(qū)
-------------------------- ShortCircuitShm.java --------------------------
  public ShortCircuitShm(ShmId shmId, FileInputStream stream)
        throws IOException {
    // 下面的代碼在客戶端和服務(wù)器端同時(shí)執(zhí)行
    this.shmId = shmId;
    this.mmappedLength = getUsableLength(stream);
    //客戶端和服務(wù)器端都把這個(gè)文件直接映射到自己的進(jìn)程內(nèi)存空間中，避免在讀寫文件過(guò)程中還需要通過(guò)內(nèi)核態(tài)，避免數(shù)據(jù)從文件到內(nèi)核、內(nèi)核到用戶態(tài)的多次拷貝
    // 可以看到，調(diào)用mmap的時(shí)候，對(duì)應(yīng)的權(quán)限是RW,因?yàn)闊o(wú)論是DataNode還是客戶端，都將對(duì)這段內(nèi)存進(jìn)行讀和寫，實(shí)現(xiàn)狀態(tài)同步
    this.baseAddress = POSIX.mmap(stream.getFD(),
        POSIX.MMAP_PROT_READ | POSIX.MMAP_PROT_WRITE, true, mmappedLength);
    this.slots = new Slot[mmappedLength / BYTES_PER_SLOT]; // 默認(rèn)情況下，一個(gè)segment可以存放8192  / 64 個(gè)slot
    this.allocatedSlots = new BitSet(slots.length);//默認(rèn)情況下，所有slot都沒有分配

1.4 讀取Replica

到這里為止，客戶端已經(jīng)為該Replica準(zhǔn)備好了Slot。讀取replica的準(zhǔn)備工作完成，可以進(jìn)行replica的讀取了。
下文將詳細(xì)講解，客戶端會(huì)向DataNode通過(guò)Domain Socket的方式獲取replica的讀取句柄InputStream，然后脫離開DataNode讀取replica的全部過(guò)程。

1.4.1 向DataNode請(qǐng)求replica的文件描述符

下圖-8展示了在共享內(nèi)存和Slot構(gòu)建結(jié)束以后，客戶端向DataNode申請(qǐng)對(duì)應(yīng)的replica的InputStream的過(guò)程。

圖-8 申請(qǐng)副本的文件描述符的基本流程

1. 客戶端通過(guò)Linux Domain Socket 向遠(yuǎn)程的 DataNode申請(qǐng)讀取某個(gè)replica。上文提到過(guò)，這個(gè)操作的名稱是OP.REQUEST_SHORT_CIRCUIT_FDS，即申請(qǐng)短路讀的文件描述符(FD)。 收到請(qǐng)求以后，DataNode會(huì)把這個(gè)replica文件和這個(gè)replica的meta文件打開(meta中主要存放了replica的checksum，如果需要對(duì)replica計(jì)算校驗(yàn)和，就將收到的replica的校驗(yàn)和跟meta文件記錄的校驗(yàn)和進(jìn)行比對(duì))，將這兩個(gè)文件的InputStream的FileDescription通過(guò)Domain Socket 返回給客戶端。 上文講過(guò)，Domain Socket的一個(gè)巨大優(yōu)勢(shì)是直接支持進(jìn)程間進(jìn)行文件描述符的傳遞。
   下面的代碼顯示了DataNode打開自己管理的replica文件，然后獲取文件的InputStream：

----------------------------------LocalReplica.java-----------------------------------
  /**
   * 打開Block文件，獲取文件的InputStream
   */
  private FileInputStream getDataInputStream(File f, long seekOffset)
      throws IOException {
    FileInputStream fis;
    .....
    else {
      try {
        // 通過(guò)普通的方式（RandomAccessFile + seek）打開Block文件
        fis = fileIoProvider.openAndSeek(getVolume(), f, seekOffset);
      }     }
    return fis;
  }

    ------------------------------ ShortCircuitReplica.java ------------------------------------
  /**
  DataNode打開本地的replica文件
*/
  public static FileDescriptor openAndSeek(File file, long offset)
      throws IOException {
    RandomAccessFile raf = null;
    try {
      // 以只讀的方式打開Block文件
      raf = new RandomAccessFile(file, "r");
      if (offset > 0) {
        raf.seek(offset);
      }
      return raf.getFD();
    } ......
  }

DataXceiver處理客戶端的REQUEST_SHORT_CIRCUIT_FDS請(qǐng)求，然后把replica和replica meta的FD返回給客戶端：

-------------------------------------------DataXceiver.java---------------------------------------------
 public void requestShortCircuitFds(final ExtendedBlock blk,
      final Token<BlockTokenIdentifier> token,
      SlotId slotId, int maxVersion, boolean supportsReceiptVerification)
        throws IOException {
        ....
        if (slotId != null) {
          // class FsDatasetImpl implements FsDatasetSpi<FsVolumeImpl>
          boolean isCached = datanode.data.
              isCached(blk.getBlockPoolId(), blk.getBlockId());
          // 在registerslot的時(shí)候，需要判定是否對(duì)slot的archorable置位。如果在緩存中，那么是archorable的，因?yàn)?/span>
          // 無(wú)論是內(nèi)存緩存還是pmem緩存，在數(shù)據(jù)存入進(jìn)去的時(shí)候，都已經(jīng)做了crc校驗(yàn)
          datanode.shortCircuitRegistry.registerSlot(
              ExtendedBlockId.fromExtendedBlock(blk), slotId, isCached);
          registeredSlotId = slotId;
        }
        // 假如block是在緩存中，那么這個(gè)fis是否是指向緩存的？
        fis = datanode.requestShortCircuitFdsForRead(blk, token, maxVersion);
      bld.build().writeDelimitedTo(socketOut);
      if (fis != null) {
        FileDescriptor fds[] = new FileDescriptor[fis.length];
        DomainSocket sock = peer.getDomainSocket();
        sock.sendFileDescriptors(fds, buf, 0, buf.length);// 將replica和replica meta的文件描述符通過(guò)domain socket返回給客戶端

由于這個(gè)文件是DataNode打開而不是客戶端打開的，所以DataNode 完全自行控制打開文件以后的權(quán)限。在這里，客戶端只需要讀replica，因此DataNode只需要以Readonly的方式打開就行。可以看到，直到這一步，整個(gè)文件打開都與mmap()無(wú)關(guān)，后文會(huì)講，是否通過(guò)mmap()的方式通過(guò)零拷貝讀取replica，是客戶端自行決定，最壞情況是直接使用這個(gè)InputStream進(jìn)行傳統(tǒng)的replica讀取。

1.4.2 打開該replica

下圖顯示了客戶端拿到了DataNode打開的副本文件的描述符以后的基本處理流程，主要是：

1. 客戶端拿到副本文件的文件描述符以后，不是直接進(jìn)行讀取，而是嘗試進(jìn)行最高效的零拷貝讀取（下面會(huì)詳細(xì)介紹）。在零拷貝讀取中，如果我們確認(rèn)數(shù)據(jù)已經(jīng)被DataNode緩存在內(nèi)存中，那么這些數(shù)據(jù)不僅零拷貝，而且無(wú)需計(jì)算checksum。
2. 如果零拷貝讀取無(wú)法實(shí)現(xiàn)，就基于這個(gè)文件描述符進(jìn)行直接的、常規(guī)的塊文件讀取。

可以看到，無(wú)論是否成功進(jìn)行零拷貝讀取，副本塊的讀取都不再經(jīng)過(guò)DataNode了，這就是零拷貝讀取和基于TCP網(wǎng)絡(luò)通信以及Domain Socket通信的最重大不同：塊文件的真正讀取不經(jīng)過(guò)DataNode的進(jìn)程。

圖-9 嘗試進(jìn)行零拷貝讀取的基本流程

Client端拿到InputStream以后，實(shí)際上已經(jīng)完全可以直接讀取到該replica了。但是，為了提高讀取效率，客戶端會(huì)首先嘗試進(jìn)行高效的零拷貝讀取(zero copy read)，如果零拷貝讀取失敗，才會(huì)回退到基于常規(guī)的InputStream的讀寫，即從文件讀取到操作系統(tǒng)內(nèi)核，然后從操作系統(tǒng)內(nèi)核緩存拷貝到用戶進(jìn)程空間 。

---------------------------------------DFSInputStream.java------------------------------------------
 @Override
  public synchronized ByteBuffer read(ByteBufferPool bufferPool,
      int maxLength, EnumSet<ReadOption> opts)
          throws IOException, UnsupportedOperationException {
    ...
    // 如果dfs.client.mmap.enabled=true，那么會(huì)先嘗試進(jìn)行零拷貝讀取
    if (dfsClient.getConf().getShortCircuitConf().isShortCircuitMmapEnabled()) {
      buffer = tryReadZeroCopy(maxLength, opts);
    }
    if (buffer != null) {
      return buffer;
    }
    // 無(wú)法進(jìn)行zerocopy的copy, 回退到基于常規(guī)讀寫的模式，但是同樣不用再經(jīng)過(guò)DataNode了
    buffer = ByteBufferUtil.fallbackRead(this, bufferPool, maxLength);
    ....
    return buffer;
  }

這里的零拷貝讀取對(duì)理解短路讀的高效非常重要，因此著重解釋：
什么是零拷貝：
零拷貝讀取，使用了內(nèi)存映射文件mmap的讀取方式。在本文開頭講mmap的時(shí)候就講過(guò)，內(nèi)存映射文件和標(biāo)準(zhǔn)IO操作最大的不同是并不需要將數(shù)據(jù)讀取到操作系統(tǒng)的內(nèi)核緩沖區(qū)，而是直接將進(jìn)程私有地址空間中的一部分區(qū)域與文件對(duì)象建立起映射關(guān)系，就好像直接從內(nèi)存中讀寫文件一樣，減少了IO的拷貝次數(shù)，提高了文件的讀寫速度。
java提供了三種內(nèi)存映射模式，即:只讀(readonly)、讀寫(read_write)、專用(private)。

• 對(duì)于只讀模式來(lái)說(shuō)，如果程序試圖進(jìn)行寫操作，則會(huì)拋出ReadOnlyBufferException異常；
• 對(duì)于讀寫模式來(lái)說(shuō)，如果程序通過(guò)內(nèi)存映射文件的方式寫或者修改文件內(nèi)容，則修改內(nèi)容會(huì)立刻反映到磁盤文件中，如果另一個(gè)進(jìn)程共享了同一個(gè)映射文件，也會(huì)立即看到變化；
• 專用模式采用的是操作系統(tǒng)的"寫時(shí)拷貝"原則，即在沒有發(fā)生寫操作的情況下，多個(gè)進(jìn)程之間都是共享文件的同一塊物理內(nèi)存的(進(jìn)程各自的虛擬地址指向同一片物理地址)，一旦某個(gè)進(jìn)程進(jìn)行寫操作，就會(huì)把受影響的文件數(shù)據(jù)單獨(dú)拷貝一份到進(jìn)程的私有緩沖區(qū)中，不會(huì)反映到物理文件中。
  在tryReadZeroCopy()方法中，使用的是只讀模式，多個(gè)進(jìn)程通過(guò)共享同一塊內(nèi)存映射區(qū)域，來(lái)實(shí)現(xiàn)零拷貝讀取，文件只需要往內(nèi)存中l(wèi)oad一次。
  tryReadZeroCopy()會(huì)通過(guò)mmap()的方式，打開DataNode傳過(guò)來(lái)的replica文件的描述符。那么問題來(lái)了，為什么通過(guò)mmap()的方式打開描述符，就成了零拷貝呢？不也需要把磁盤文件讀取到內(nèi)存嗎？
  其實(shí)，問題的關(guān)鍵在于HDFS的緩存。即，客戶端讀取的這個(gè)replica可能已經(jīng)在DataNode端被緩存起來(lái)了。了解DataNode緩存的就知道，DataNode緩存數(shù)據(jù)的時(shí)候，就是通過(guò)mmap()和mlock()的方式將replica映射到內(nèi)存。因此，當(dāng)客戶端通過(guò)該文件的FileDescriptor通過(guò)mmap()的方式打開文件，已經(jīng)不需要再?gòu)拇疟P讀取文件，而是直接映射到了DataNode 的緩存中，因此是零拷貝。

零拷貝帶來(lái)的好處：

• 數(shù)據(jù)讀取效率和節(jié)省內(nèi)存空間 如果數(shù)據(jù)已經(jīng)在DataNode端緩存起來(lái)，那么通過(guò)零拷貝方式，用戶的客戶端進(jìn)程無(wú)需再次讀取數(shù)據(jù)到內(nèi)存，而是直接引用這塊緩存數(shù)據(jù)，既提高了數(shù)據(jù)讀取效率，也節(jié)省了內(nèi)存(同一塊數(shù)據(jù)無(wú)需重復(fù)存取)
• 在特定情況下不用再進(jìn)行checksum校驗(yàn)，因?yàn)榱憧截惓晒?，說(shuō)明數(shù)據(jù)已經(jīng)被DataNode通過(guò)mmap()的方式緩存到內(nèi)存了，DataNode在緩存時(shí)肯定已經(jīng)做過(guò)checksum了。checksum是cpu密集型操作。

HDFS的緩存的調(diào)用發(fā)生在CachingTask的Runnable中，我們以內(nèi)存緩存為例（HDFS還支持持久內(nèi)存緩存Pmem， 本文不討論）：

----------------------------------------CachingTask.java----------------------------------------------
 private class CachingTask implements Runnable {
    /**
      DataNode的CachingTask會(huì)通過(guò)mmap()的方式將replica加載到緩存中，同時(shí)通過(guò)mlock()將這段內(nèi)存鎖定在緩存中。
    */
    @Override
    public void run() {
      ...
          blockIn = (FileInputStream)dataset.getBlockInputStream(extBlk, 0);
          metaIn = DatanodeUtil.getMetaDataInputStream(extBlk, dataset);
        .....

        // 通過(guò)cacheLoader，將塊load到緩存。對(duì)于基于內(nèi)存的緩存，cacheLoader的實(shí)現(xiàn)是MemoryMappableBlockLoader
          mappableBlock = cacheLoader.load(length, blockIn, metaIn,
              blockFileName, key);

----------------------------------------MemoryMappableBlockLoader.java----------------------------------------------
/**
   * Load the block. 
   * mmap and mlock the block, and then verify its checksum.
   */
  @Override
  MappableBlock load(long length, FileInputStream blockIn,
      FileInputStream metaIn, String blockFileName, ExtendedBlockId key)
      throws IOException {
      // 通過(guò)mmap的方式將文件映射到內(nèi)存
      mmap = blockChannel.map(FileChannel.MapMode.READ_ONLY, 0, length);
      NativeIO.POSIX.getCacheManipulator().mlock(blockFileName, mmap, length);//調(diào)用操作系統(tǒng)mlock接口，將
      verifyChecksum(length, metaIn, blockChannel, blockFileName);// 計(jì)算checksum
      mappableBlock = new MemoryMappedBlock(mmap, length）

我們看到，DataNode在進(jìn)行緩存操作的時(shí)候，除了mmap()，還調(diào)用mlock()的系統(tǒng)調(diào)用。mlock()的作用是將指定地址范圍內(nèi)的頁(yè)被鎖定在物理內(nèi)存中。 在此進(jìn)程或其他進(jìn)程中引用鎖定頁(yè)不會(huì)導(dǎo)致需要 I/O 操作的頁(yè)錯(cuò)誤。

在客戶端，在嘗試進(jìn)行零拷貝讀取的時(shí)候，會(huì)嘗試調(diào)用createNoChecksumContext()來(lái)創(chuàng)建無(wú)需進(jìn)行checksum的上下文。createNoChecksumContext()只有在下面的情況會(huì)返回true，即無(wú)需進(jìn)行checksum的前提是，這段數(shù)據(jù)已經(jīng)被DataNode緩存到了內(nèi)存，因?yàn)?，根?jù)DataNode的緩存機(jī)制，緩存到內(nèi)存的數(shù)據(jù)都必須進(jìn)行了checksum都校驗(yàn) ，因此，客戶端使用這段內(nèi)存無(wú)需再進(jìn)行checksum了，況且checksum是cpu密集型的操作。
下面的代碼可以看到createNoChecksumContext()的判斷標(biāo)準(zhǔn)和創(chuàng)建客戶端的mmap的過(guò)程：

-------------------------------------------BlockReaderLocal.java-------------------------------------------------
  public ClientMmap getClientMmap(EnumSet<ReadOption> opts) {
    boolean anchor = verifyChecksum &&
        !opts.contains(ReadOption.SKIP_CHECKSUMS);
    if (anchor) { // 需要進(jìn)行數(shù)據(jù)校驗(yàn)
      if (!createNoChecksumContext()) { // 創(chuàng)建nochecksum context失敗
        return null;
      }
    }
      clientMmap = replica.getOrCreateClientMmap(anchor);
  }
  // 創(chuàng)建Nochecksum 的上下文，如果成功，那么說(shuō)明正在讀取DataNode緩存中的數(shù)據(jù)，因此無(wú)需再進(jìn)行checksum，因?yàn)镈N在進(jìn)行緩存的時(shí)候已經(jīng)做了
  private boolean createNoChecksumContext() {
    // 客戶端通過(guò)配置的方式配置為跳過(guò)checksum, 或者是臨時(shí)存儲(chǔ)， 或者，如果verifyChecksum為true，同時(shí)storagetype不是臨時(shí)存儲(chǔ)，
    // 那么添加checksum anchor計(jì)數(shù)。如果計(jì)數(shù)成功，就返回true，計(jì)數(shù)失敗返回false。
    // 能否anchor，是在DataNode端通過(guò)ShortCircuitRegistry.registerSlot() -》 slot.makeAnchorable()來(lái)判斷的
    return !verifyChecksum ||
        // Checksums are not stored for replicas on transient storage.  We do
        // not anchor, because we do not intend for client activity to block
        // eviction from transient storage on the DataNode side.
        (storageType != null && storageType.isTransient()) ||
        replica.addNoChecksumAnchor();
  }

----------------------------------------------ShortCircuitReplica.java-----------------------------------------------
  /**
  客戶端嘗試通過(guò)mmap的方式實(shí)現(xiàn)零拷貝
  */
  MappedByteBuffer loadMmapInternal() {
    try {
      // 將DataNode返回的讀取block的dataStream轉(zhuǎn)換成內(nèi)存的mmap
      FileChannel channel = dataStream.getChannel();
      MappedByteBuffer mmap = channel.map(MapMode.READ_ONLY, 0,
          Math.min(Integer.MAX_VALUE, channel.size()));
      return mmap;
  }

NoChecksumAnchor的作用：
NoChecksumAnchor是這個(gè)replica對(duì)應(yīng)的slot中的一個(gè)計(jì)數(shù)器，通過(guò)Unsafe去增加或者減少以進(jìn)行計(jì)數(shù)。它主要是統(tǒng)計(jì)當(dāng)前有多少客戶端正在通過(guò)無(wú)需校驗(yàn)的方式讀取緩存中的replica。
剛剛說(shuō)到，對(duì)于已經(jīng)被DataNode緩存的數(shù)據(jù)，客戶端通過(guò)mmap 的方式讀取，無(wú)需再進(jìn)行checksum。replica在DataNode上既然的緩存，那肯定有緩存的淘汰策略。如果一個(gè)緩存的replica正在被客戶端通過(guò)mmap nochecksum 的方式讀取，我們肯定不希望這個(gè)replica在緩存中被淘汰，否則會(huì)使得客戶端的讀操作失敗。那么，如果知道當(dāng)前緩存的replica是否有客戶端正在通過(guò)mmap的方式讀取呢？這是通過(guò)Slot中的noCheckSumAnchor來(lái)完成的：每當(dāng)客戶端通過(guò)mmap的方式讀取一個(gè)block，如果創(chuàng)建nonchecksum context成功，那么就將noChecksumAnchor的計(jì)數(shù)器加1，讀完以后減去1。這樣，DataNode只會(huì)在這段緩存的nochecksumAnchor的計(jì)數(shù)為0的時(shí)候，才會(huì)從物理內(nèi)存中解鎖這段緩存，即進(jìn)行munlock()操作，代表緩存可以被swap out 了。

可以看到，將普通文件讀寫的InputStream轉(zhuǎn)換成MappedByteBuffer的過(guò)程都被Java NIO進(jìn)行了很好的封裝，使用很簡(jiǎn)單：

• 從文件的InputStream中獲取文件讀寫的Channel
• 通過(guò)map()將InputStream 映射到MappedByteBuffer(注意，mmap調(diào)用完成以后，并沒有真正將文件數(shù)據(jù)拷貝到MappedByteBuffer中，真生的數(shù)據(jù)拷貝發(fā)生在客戶端真的開始從MappedByteBuffer讀取數(shù)據(jù)的時(shí)候)
• 通過(guò)mlock()的方式將這段內(nèi)存區(qū)域鎖定

這里涉及到對(duì)ByteBuffer的理解、對(duì)Channel的理解，但是不在本文介紹的范圍內(nèi)。我們只需要知道，MappedByteBuffer是一種·ByteBuffer·的實(shí)現(xiàn)，因此ByteBuffer的整個(gè)操作邏輯對(duì)MappedByteBuffer也適用。但是，MappedByteBuffer是一種特殊的ByteBuffer， 它的內(nèi)容是某個(gè)文件的一部分(Segment), 并且文件的內(nèi)容是存放在Direct Memory中的（不在jvm heap中）。

3. 結(jié)束

可以看到，整個(gè)短路讀的方式涉及到了很多特殊的操作系統(tǒng)層面的知識(shí)，因此傳統(tǒng)的做Java開發(fā)的，如果僅僅從Hadoop代碼層面無(wú)法對(duì)短路讀有準(zhǔn)確的理解，所以本文開始的時(shí)候介紹了短路讀所涉及到的一些操作系統(tǒng)知識(shí)。
短路讀的整個(gè)流程講清楚也非常不容易，因?yàn)榱鞒潭啵ㄐ欧绞蕉?，又是Domain Socket通信，又是共享內(nèi)存通信，又是文件描述符傳遞。
本文力求詳盡，但是如果想要徹底理解，還需要讀者順著本文的思路真實(shí)地去看代碼。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-718669.html

4.參考文章

• 認(rèn)真分析mmap：是什么 為什么 怎么用
• HDFS短路讀詳解
• 文件操作之FileChannel和mmap
• Java NIO 中的 Channel 詳解
• 詳解Java NIO之Channel(通道)
• 讀HDFS書筆記—5.2 文件讀操作與輸入流(5.2.2)—下
• mmap和共享內(nèi)存
• 關(guān)于共享內(nèi)存shm和內(nèi)存映射mmap的區(qū)別是什么？
• 共享內(nèi)存之——mmap內(nèi)存映射

到了這里，關(guān)于HDFS 短路讀的實(shí)現(xiàn)(全網(wǎng)最全面深入講解)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！