磁盘 → Page Cache → DataNode 用户态 Buffer → TCP Send Buffer → TCP Recv Buffer → 客户端用户态 Buffer

DataNode write()
  → tcp_sendmsg()          // 分段、拷贝到 sk_buff
  → tcp_transmit_skb()     // 构造 TCP 头、计算校验和（纯 CPU 算！）
  → ip_queue_xmit()        // IP 层路由
  → loopback_xmit()        // 发现目标在本地，直接丢入接收队列
  → netif_rx()             // 触发软中断（NET_RX_SOFTIRQ）
  → tcp_v4_rcv()           // TCP 入站处理
  → tcp_rcv_established()  // 重组、ACK、窗口
  → sock_def_readable()    // 通知接收 socket
Client read()

// BlockReaderFactory.java — build() 方法（简化）
public BlockReader build() throws IOException {
    // 第0级：外部插件（ReplicaAccessorBuilder）
    BlockReader reader = tryToCreateExternalBlockReader();
    if (reader != null) return reader;

    final ShortCircuitConf scConf = conf.getShortCircuitConf();
    try {
        if (scConf.isShortCircuitLocalReads() && allowShortCircuitLocalReads) {
            if (clientContext.getUseLegacyBlockReaderLocal()) {
                // 第一级短路：Legacy 路径（本文主角之一）
                reader = getLegacyBlockReaderLocal();
                if (reader != null) return reader;
            } else {
                // 第二级短路：新式 fd 传递路径（本文主角之二）
                reader = getBlockReaderLocal();
                if (reader != null) return reader;
            }
        }
        if (scConf.isDomainSocketDataTraffic()) {
            // 第三级：Domain Socket 数据流（仍走 DataNode，但无 TCP）
            reader = getRemoteBlockReaderFromDomain();
            if (reader != null) return reader;
        }
    } catch (IOException e) { /* 降级 */ }
    
    // 最终兜底：TCP 远程读取
    return getRemoteBlockReaderFromTcp();
}

// BlockReaderFactory.java — getLegacyBlockReaderLocal()
private BlockReader getLegacyBlockReaderLocal() throws IOException

private BlockReader getLegacyBlockReaderLocal() throws IOException {
    // 【守卫条件1】目标必须为本机
    if (!DFSUtilClient.isLocalAddress(inetSocketAddress)) {
        return null;
    }
    
    // 【守卫条件2】未被上次失败禁用
    if (clientContext.getDisableLegacyBlockReaderLocal()) {
        return null;
    }
    
    try {
        // 【核心】通过 RPC 获取块文件路径，然后直接打开
        return BlockReaderLocalLegacy.newBlockReader(conf,
            userGroupInformation, configuration, fileName, block, token,
            datanode, startOffset, length, storageType);
    } catch (RemoteException re) {
        ioe = re.unwrapRemoteException(
            InvalidToken.class, AccessControlException.class);
    }
    
    // 【失败处理】权限异常说明用户不在白名单，禁用 Legacy 路径
    if ((!(ioe instanceof AccessControlException)) && isSecurityException(ioe)) {
        throw ioe;  // 非权限异常，直接抛出（阻止降级）
    }
    LOG.warn("Disabling legacy local reads.", ioe);
    clientContext.setDisableLegacyBlockReaderLocal();  // 禁用
    return null;
}

// BlockReaderLocalLegacy.java — newBlockReader() 核心片段
static BlockReaderLocalLegacy newBlockReader(...) throws IOException {
    // 【步骤1】从本地缓存获取块路径（LRU，最多 10000 条）
    LocalDatanodeInfo localDatanodeInfo = 
        getLocalDatanodeInfo(node.getIpcPort());
    BlockLocalPathInfo pathinfo = localDatanodeInfo.getBlockLocalPathInfo(blk);
    
    if (pathinfo == null) {
        // 【步骤2】缓存未命中，发起 RPC 调用
        // 调用 DataNode 的 ClientDatanodeProtocol.getBlockLocalPathInfo()
        pathinfo = getBlockPathInfo(userGroupInformation, blk, node,
            configuration, conf.getSocketTimeout(), token,
            conf.isConnectToDnViaHostname(), storageType);
    }
    
    // 【步骤3】直接打开本地文件（这是"短路"的本质）
    File blkfile = new File(pathinfo.getBlockPath());
    dataIn = new FileInputStream(blkfile);  // 直接 open()
    
    if (!skipChecksumCheck) {
        File metafile = new File(pathinfo.getMetaPath());
        checksumIn = new FileInputStream(metafile);
        // 解析 checksum 头
        final DataChecksum checksum = BlockMetadataHeader.readDataChecksum(...);
    }
    
    return new BlockReaderLocalLegacy(scConf, file, blk,
        startOffset, checksum, verifyChecksum, dataIn, ..., checksumIn);
}

// BlockReaderFactory.java — getBlockReaderLocal()
private BlockReader getBlockReaderLocal() throws IOException

private BlockReader getBlockReaderLocal() throws IOException {
    // 【步骤1】获取 Domain Socket 路径状态（带缓存）
    if (pathInfo == null) {
        pathInfo = clientContext.getDomainSocketFactory()
            .getPathInfo(inetSocketAddress, conf.getShortCircuitConf());
    }
    
    // 【步骤2】检查路径是否可用
    if (!pathInfo.getPathState().getUsableForShortCircuit()) {
        return null;
    }
    
    // 【步骤3】从 ShortCircuitCache 获取或创建 ShortCircuitReplica
    // （这是整个流程的灵魂）
    ShortCircuitCache cache = clientContext.getShortCircuitCache();
    ExtendedBlockId key = new ExtendedBlockId(
        block.getBlockId(), block.getBlockPoolId());
    
    // 【关键】fetchOrCreate() 是线程安全的 Fetch-Or-Create 模式
    // 如果缓存中没有，会回调 this.createShortCircuitReplicaInfo()
    // （this 实现了 ShortCircuitReplicaCreator 接口）
    ShortCircuitReplicaInfo info = cache.fetchOrCreate(key, this);
    
    if (info.getInvalidTokenException() != null) {
        throw info.getInvalidTokenException();
    }
    if (info.getReplica() == null) {
        return null;  // 创建失败，返回 null 触发降级
    }
    
    // 【步骤4】使用 Builder 模式构造 BlockReaderLocal
    return new BlockReaderLocal.Builder(conf.getShortCircuitConf())
        .setFilename(fileName)
        .setBlock(block)
        .setStartOffset(startOffset)
        .setShortCircuitReplica(info.getReplica())  // 注入 replica（含 data/meta fd）
        .setVerifyChecksum(verifyChecksum)
        .setCachingStrategy(cachingStrategy)
        .setStorageType(storageType)
        .build();
}

// ShortCircuitCache.java — fetchOrCreate()
public ShortCircuitReplicaInfo fetchOrCreate(ExtendedBlockId key,
    ShortCircuitReplicaCreator creator) {
    
    lock.lock();
    try {
        // 【快路径】缓存命中
        Waitable<ShortCircuitReplicaInfo> waitable = replicaInfoMap.get(key);
        if (waitable != null) {
            info = fetch(key, waitable);  // 等待或直接获取
            if (info != null) return info;
        }
        
        // 【慢路径】缓存未命中：放入 Waitable，让其他线程等待
        newWaitable = new Waitable<>(lock.newCondition());
        replicaInfoMap.put(key, newWaitable);
    } finally { lock.unlock(); }
    
    // 【释放锁后】回调 creator.createShortCircuitReplicaInfo()
    // 即 BlockReaderFactory.createShortCircuitReplicaInfo()
    return create(key, creator, newWaitable);
}

// BlockReaderFactory.java — createShortCircuitReplicaInfo()
@Override
public ShortCircuitReplicaInfo createShortCircuitReplicaInfo() {
    // 【步骤1】获取或创建 Domain Socket 连接
    DomainPeer peer = nextDomainPeer();
    
    // 【步骤2】在共享内存中分配 slot（与 DataNode 同步状态）
    ShortCircuitReplica.ShareMemorySlot slot = cache.allocShmSlot(...);
    
    // 【步骤3】向 DataNode 发送短路读取请求
    // 底层：Sender.requestShortCircuitFds()
    requestShortCircuitFds(peer, slot, ...);
    
    // 【步骤4】接收 DataNode 通过 SCM_RIGHTS 传递的 FD
    // 底层：DomainSocket.recvFileInputStreams()
    FileInputStream[] fis = peer.getDomainSocket().recvFileInputStreams(...);
    
    // 【步骤5】封装为 ShortCircuitReplica
    ShortCircuitReplica replica = new ShortCircuitReplica(
        key, fis[0], fis[1], slot, ...);  // fis[0]=data, fis[1]=meta
    
    return new ShortCircuitReplicaInfo(replica);
}

// DataXceiver.java — requestShortCircuitFds()（DataNode 侧）
public void requestShortCircuitFds(...) throws IOException {
    // 【步骤1】权限校验（验证 Block Token）
    checkAccess(...);
    
    // 【步骤2】在共享内存中注册 slot
    datanode.getShortCircuitRegistry().registerSlot(...);
    
    // 【步骤3】打开块文件（DataNode 有权限，因为它是文件 owner）
    FileInputStream[] fis = datanode.requestShortCircuitFdsForRead(blk, token, maxVersion);
    
    // 【步骤4】通过 SCM_RIGHTS 发送 FD
    // 内核处理：为客户端进程分配新 FD，指向同一个 struct file
    DomainSocket sock = peer.getDomainSocket();
    sock.sendFileDescriptors(
        new FileDescriptor[]{fis[0].getFD(), fis[1].getFD()},
        buf, 0, buf.length);
}

DataNode 调用 bind("/var/run/hdfs/dn.sock")：

  VFS 层：
    → 在路径上创建一个 inode
    → inode->i_mode  = S_IFSOCK  （socket 类型，不是普通文件！）
    → inode->i_fop   = &socket_file_ops

  内核内存：
    → 分配 struct socket + struct sock（AF_UNIX 协议族）
    → 服务端 fd 指向这个 struct socket
    → 把路径字符串和 struct socket 的对应关系记录到 Unix socket 哈希表

  ls -l 输出：
    srwx------ 1 hdfs hdfs ... /var/run/hdfs/dn.sock
    ↑ s 开头 = socket 文件类型

客户端 connect("/var/run/hdfs/dn.sock")：

  1. VFS 解析路径 → 找到 S_IFSOCK 类型的 inode
  2. 通过 inode 找到 DN 绑定的 struct socket
  3. 分配客户端自己的 struct socket
  4. 内核将两端的 struct socket 配对：
     → 服务端 struct unix_sock->peer 指向客户端
     → 客户端 struct unix_sock->peer 指向服务端
     → 两端 sk_receive_queue 互相关联

  结果：
    客户端 fd → struct socket ←→ struct socket ← DN accept fd
                 ↑                     ↑
            完全在内核内存中，不走网卡

正确的方向（HDFS 实际实现）：

DataNode 启动：
  ├─ socket() → bind(/var/run/hdfs/dn.sock) → listen()
  └─ accept() 阻塞等待

客户端发起短路读取：
  ├─ step1: connect(/var/run/hdfs/dn.sock)   ← 主动连接 DN
  ├─ step2: allocShmSlot()                     ← 分配共享内存
  ├─ step3: sendmsg() 发送读取请求（普通数据）
  └─ DN 收到请求后：
      ├─ checkAccess(token)                    ← 权限校验
      ├─ openBlockFile()                       ← 打开块文件，拿到 FD
      └─ sendmsg(SCM_RIGHTS)                  ← 把 FD 发回客户端

假设客户端做服务端（不可行）：
  → DN 需要 connect() 到客户端的 socket
  → 客户端怎么告诉 DN "我的 socket 路径在哪"？
  → 任意进程都可以伪装成客户端...

正确设计（DN 做服务端）：
  → /var/run/hdfs/dn.sock 固定路径，权限 0600
  → 客户端 connect() 时，DN 通过 SO_PEERCRED 获取客户端 UID
  → 结合 block token 校验，确保只有合法客户端能拿到 FD

UDS 连接建立后：

  客户端 ←─────── 全双工 ──────→ DataNode
            (同一条连接，双向同时可用)

  客户端可以：sendmsg(SCM_RIGHTS) → DN 收
  DN 也可以：sendmsg(SCM_RIGHTS) → 客户端收

应用层：
  sendmsg(sockfd, &msg, 0)       ← 一次系统调用，从用户态陷入内核态
    │
    │  msg.msg_control 包含：
    │  struct cmsghdr {
    │      cmsg_type = SCM_RIGHTS;   ← 常量标志，告诉内核"我要传 FD"
    │      cmsg_data = {block_fd};   ← 要传递的 FD 值
    │  }
    │
    ▼ 内核态
  __sys_sendmsg() → ___sys_sendmsg()
    └─ scm_send()
       └─ case SCM_RIGHTS:
          scm_fp_copy(cmsg_data)
            1. fget(fd) → 获取 struct file
            2. 在接收方 FD 表中分配新条目，指向同一 struct file
            3. 新 FD 值写回接收方的 cmsg_data

// BlockReaderLocal.java — read(ByteBuffer buf)
@Override
public synchronized int read(ByteBuffer buf) throws IOException {
    boolean canSkipChecksum = createNoChecksumContext();
    try {
        if (canSkipChecksum && zeroReadaheadRequested) {
            // 【路径 A】直接操作 FileChannel，无 bounce buffer
            nRead = readWithoutBounceBuffer(buf);
        } else {
            // 【路径 B，默认】通过 bounce buffer 预读 + 可选 checksum 验证
            nRead = readWithBounceBuffer(buf, canSkipChecksum);
        }
        return nRead;
    } finally {
        if (canSkipChecksum) releaseNoChecksumContext();
    }
}

// BlockReaderLocal 构造函数（源码第 257-265 行）
int maxReadaheadChunks = (bytesPerChecksum == 0) ? 0 :
    ((Math.min(builder.bufferSize, builder.maxReadahead) +
        bytesPerChecksum - 1) / bytesPerChecksum);

if (maxReadaheadChunks == 0) {
    this.zeroReadaheadRequested = true;    // ← 唯一触发条件
    maxReadaheadChunks = 1;
} else {
    this.zeroReadaheadRequested = false;   // ← 正常 checksum 类型，永远 false
}

// Builder 构造函数（第 89 行）
this.verifyChecksum = !conf.isSkipShortCircuitChecksums();
//                              ↑ dfs.client.read.shortcircuit.skip.checksum

// BlockReaderLocal.java 第 68 行
private static final DirectBufferPool bufferPool = new DirectBufferPool();

// 第 224 行
private ByteBuffer dataBuf;  // ← 这就是 bounce buffer

// 第 284-286 行：懒初始化
private synchronized void createDataBufIfNeeded() {
    if (dataBuf == null) {
        dataBuf = bufferPool.getBuffer(
            maxAllocatedChunks * bytesPerChecksum);
    }
}

// DirectBufferPool.java — 底层是 allocateDirect
public ByteBuffer getBuffer(int size) {
    // ...先从池里尝试复用...
    return ByteBuffer.allocateDirect(size);  // ← Direct ByteBuffer!
}

// DFSInputStream.java — read(pool, maxLen, opts) 完整实现
// 这才是真实代码，下面两个方法共同完成零拷贝读

// ===== 方法一：对外入口 =====
@Override
public synchronized ByteBuffer read(ByteBufferPool bufferPool,
        int maxLength, EnumSet<ReadOption> opts)
            throws IOException, UnsupportedOperationException {

    // 确保当前有可用的 blockReader
    if ((blockReader == null) || (blockEnd == -1)) {
        if (pos >= getFileLength())  return null;
        if ((!seekToBlockSource(pos)) || (blockReader == null))
            throw new IOException(...);
    }

    ByteBuffer buffer = null;

    // ① 先尝试 mmap 零拷贝（受 dfs.client.mmap.enabled 控制）
    if (dfsClient.getConf()
            .getShortCircuitConf().isShortCircuitMmapEnabled()) {
        buffer = tryReadZeroCopy(maxLength, opts);  // ← 真正调 getClientMmap() 的地方
    }

    // ② mmap 成功，直接返回
    if (buffer != null) {
        return buffer;  // ← 零拷贝路径，buffer 直接映射 PageCache
    }

    // ③ mmap 不可用或不支持，退回普通短路读（bounce buffer，2次拷贝）
    buffer = ByteBufferUtil.fallbackRead(this, bufferPool, maxLength);
    if (buffer != null) {
        getExtendedReadBuffers().put(buffer, bufferPool);
    }
    return buffer;
}

// ===== 方法二：零拷贝的实际执行者 =====
private synchronized ByteBuffer tryReadZeroCopy(
        int maxLength, EnumSet<ReadOption> opts) throws IOException {

    final long curPos   = pos;
    final long blockPos = curPos
            - currentLocatedBlock.getStartOffset();  // ← block 内偏移

    // 截断到 block 末尾，防止越界
    long length63 = (curPos + maxLength <= blockEnd + 1)
            ? maxLength
            : (blockEnd - curPos + 1);
    if (length63 <= 0) return null;

    // MappedByteBuffer 上限 2GB（31-bit），超出的部分本次不 mmap
    int length;
    if (blockPos + length63 <= Integer.MAX_VALUE)
        length = (int) length63;
    else
        return null;

    // ★ 真正发起 mmap 的地方
    final ClientMmap clientMmap = blockReader.getClientMmap(opts);
    if (clientMmap == null)  return null;  // 走普通短路读（bounce buffer，2次拷贝）
    ...
}

// BlockReaderLocal.java 第 677 行
@Override
public ClientMmap getClientMmap(EnumSet<ReadOption> opts) {

    // 需要校验 checksum 且没传 SKIP_CHECKSUMS 时，必须先 anchoring
    boolean anchor = verifyChecksum
            && !opts.contains(ReadOption.SKIP_CHECKSUMS);

    if (anchor) {
        if (!createNoChecksumContext()) {
            return null;  // 无法 anchoring（block 未 mlock），放弃 mmap
        }
    }

    ClientMmap clientMmap = null;
    try {
        clientMmap = replica.getOrCreateClientMmap(anchor);
        //            ↑ 最终走到 FileChannel.map(POSIX_MMAP, offset, length)
        //              → MappedByteBuffer（真正的零拷贝！）
    } finally {
        if (clientMmap == null && anchor) {
            releaseNoChecksumContext();  // 失败回滚
        }
    }
    return clientMmap;
}

// 唯一生效位置：DFSInputStream.read(pool, maxLen, opts)
if (dfsClient.getConf()
        .getShortCircuitConf().isShortCircuitMmapEnabled()) {
    buffer = tryReadZeroCopy(maxLength, opts);
}

① FileChannel.map() 执行后
   用户页表项全部为空（Present=0），未分配物理页框
   此时 "映射" 只是一个 VMA（虚拟内存区域）记录

② 用户第一次读 MappedByteBuffer.get()
   CPU 查页表 → Present=0 → 触发 Page Fault（缺页中断）
       │
       ▼
   内核缺页处理程序：
     1. 分配一块物理页框（4KB）
     2. 通过 DMA 从磁盘读 Block 数据到该页框
     3. 填写用户页表：PTE.paddr = 页框地址, Present=1
     4. 同时填写内核 PageCache 映射
     5. iret 返回用户态，CPU 重试刚才的指令
     6. 这次页表命中 → 数据已在内存 → 零拷贝读到！

③ 后续读同一页 → MMU 硬件直接转译
   整个过程：用户态执行一条 mov 指令读取内存
   没有 read() 系统调用，没有内核态切换，没有拷贝！

1. dfs.client.read.shortcircuit=true
2. dfs.client.mmap.enabled=true（默认）
3. BlockReader 是 BlockReaderLocal（短路读已生效）
4. Replica 支持 mmap（数据在本地文件系统）
5. 用户通过 DFSInputStream.read(pool, maxLen, opts) 调用
   → 普通 FSDataInputStream.read(byte[]) / read(ByteBuffer) 不走 mmap！

DFSInputStream
    │
    ├── read(byte[] buf, int off, int len)      
    │     │
    │     └── ByteArrayStrategy
    │           │
    │           └── readWithStrategy()           ← 统一调度引擎
    │                 │
    │                 ├── blockSeekTo()
    │                 │     └── BlockReaderFactory.build()
    │                 │           └── getBlockReaderLocal()  ✅ 新版短路
    │                 │
    │                 └── BlockReaderLocal.read(byte[], int, int) 
    │                       │
    │                       └── readWithBounceBuffer()  ← 2 次 CPU 拷贝
    │
    ├── read(ByteBuffer buf)                    
    │     │
    │     └── ByteBufferStrategy
    │           │
    │           └── readWithStrategy()             ← 同一个引擎！
    │                 │
    │                 ├── blockSeekTo()
    │                 │     └── BlockReaderFactory.build()
    │                 │           └── getBlockReaderLocal()  ✅ 新版短路
    │                 │
    │                 └── BlockReaderLocal.read(ByteBuffer)
    │                       │
    │                       └── readWithBounceBuffer()  ← 2 次 CPU 拷贝
    │
    └── read(ByteBufferPool pool, int maxLen,
             EnumSet<ReadOption> opts)          
          │
          └── seekToBlockSource()
                │
                └── BlockReaderFactory.build()
                      └── getBlockReaderLocal()  ✅ 新版短路
          │
          ├── tryReadZeroCopy()
          │     └── getClientMmap()              ← 0 次拷贝 ✅
          │
          └── fallback: ByteBufferUtil.fallbackRead()
                └── DFSInputStream.read(ByteBuffer)  ← 退回 2 次拷贝

<!-- 启用短路本地读取 -->
<property>
    <name>dfs.client.read.shortcircuit</name>
    <value>true</value>
</property>

<!-- 使用新版 fd 传递机制（默认即 false，显式写出更清晰） -->
<property>
    <name>dfs.client.use.legacy.blockreader.local</name>
    <value>false</value>
</property>

<!-- Domain Socket 路径模板 -->
<property>
    <name>dfs.domain.socket.path</name>
    <value>/var/lib/hadoop-hdfs/dn._PORT</value>
</property>

<!-- 跳过 checksum（当 checksum 已由底层存储保证时，可选） -->
<property>
    <name>dfs.client.read.shortcircuit.skip.checksum</name>
    <value>false</value>
</property>

<!-- ShortCircuitCache 配置 -->
<property>
    <name>dfs.client.read.shortcircuit.streams.cache.size</name>
    <value>4096</value>  <!-- 默认 256，大内存机器可调大 -->
</property>

<!-- mmap 零拷贝相关 -->
<property>
    <name>dfs.client.mmap.enabled</name>
    <value>true</value>  <!-- 总开关：是否尝试 mmap 零拷贝，默认已开启 -->
</property>
<property>
    <name>dfs.client.mmap.cache.size</name>
    <value>256</value>  <!-- mmap 映射 LRU 缓存条目数 -->
</property>
<property>
    <name>dfs.client.mmap.cache.timeout.ms</name>
    <value>60000</value>  <!-- mmap 缓存过期时间，默认 1 分钟 -->
</property>