import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
/**
 * 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) {
    Peer peer = clientContext.getPeerCache().get(datanode, true);
    if (peer != null) {
      if (LOG.isTraceEnabled()) {
        LOG.trace("nextDomainPeer: reusing existing peer " + peer);
      }
      return new BlockReaderPeer(peer, true);
    }
  }
  DomainSocket sock = clientContext.getDomainSocketFactory().
      createSocket(pathInfo, conf.socketTimeout);
  if (sock == null) return null;
  return new BlockReaderPeer(new DomainPeer(sock), false);
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
public Slot allocSlot(DatanodeInfo datanode, DomainPeer peer,
    MutableBoolean usedPeer, ExtendedBlockId blockId,
    String clientName) throws IOException {
  lock.lock();
  try {
    if (closed) {
      LOG.trace(this + ": the DfsClientShmManager isclosed.");
      return null;
    }
    EndpointShmManager shmManager = datanodes.get(datanode);
    if (shmManager == null) {
      shmManager = new EndpointShmManager(datanode);
      datanodes.put(datanode, shmManager);
    }
    return shmManager.allocSlot(peer, usedPeer, clientName, blockId);
  } finally {
    lock.unlock();
  }
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
/**
 * 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) {
    Peer peer = clientContext.getPeerCache().get(datanode, true);
    if (peer != null) {
      LOG.trace("nextDomainPeer: reusing existing peer {}", peer);
      return new BlockReaderPeer(peer, true);
    }
  }
  DomainSocket sock = clientContext.getDomainSocketFactory().
      createSocket(pathInfo, conf.getSocketTimeout());
  if (sock == null) return null;
  return new BlockReaderPeer(new DomainPeer(sock), false);
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
/**
 * Get a RemoteBlockReader that communicates over a UNIX domain socket.
 *
 * @return The new BlockReader, or null if we failed to create the block
 * reader.
 *
 * @throws InvalidToken    If the block token was invalid.
 * Potentially other security-related execptions.
 */
private BlockReader getRemoteBlockReaderFromDomain() throws IOException {
  if (pathInfo == null) {
    pathInfo = clientContext.getDomainSocketFactory().
                    getPathInfo(inetSocketAddress, conf);
  }
  if (!pathInfo.getPathState().getUsableForDataTransfer()) {
    PerformanceAdvisory.LOG.debug(this + ": not trying to create a " +
        "remote block reader because the UNIX domain socket at " +
        pathInfo + " is not usable.");
    return null;
  }
  if (LOG.isTraceEnabled()) {
    LOG.trace(this + ": trying to create a remote block reader from the " +
        "UNIX domain socket at " + pathInfo.getPath());
  }

  while (true) {
    BlockReaderPeer curPeer = nextDomainPeer();
    if (curPeer == null) break;
    if (curPeer.fromCache) remainingCacheTries--;
    DomainPeer peer = (DomainPeer)curPeer.peer;
    BlockReader blockReader = null;
    try {
      blockReader = getRemoteBlockReader(peer);
      return blockReader;
    } catch (IOException ioe) {
      IOUtils.cleanup(LOG, peer);
      if (isSecurityException(ioe)) {
        if (LOG.isTraceEnabled()) {
          LOG.trace(this + ": got security exception while constructing " +
              "a remote block reader from the unix domain socket at " +
              pathInfo.getPath(), ioe);
        }
        throw ioe;
      }
      if (curPeer.fromCache) {
        // Handle an I/O error we got when using a cached peer.  These are
        // considered less serious, because the underlying socket may be stale.
        if (LOG.isDebugEnabled()) {
          LOG.debug("Closed potentially stale domain peer " + peer, ioe);
        }
      } else {
        // Handle an I/O error we got when using a newly created domain peer.
        // We temporarily disable the domain socket path for a few minutes in
        // this case, to prevent wasting more time on it.
        LOG.warn("I/O error constructing remote block reader.  Disabling " +
            "domain socket " + peer.getDomainSocket(), ioe);
        clientContext.getDomainSocketFactory()
            .disableDomainSocketPath(pathInfo.getPath());
        return null;
      }
    } finally {
      if (blockReader == null) {
        IOUtils.cleanup(LOG, peer);
      }
    }
  }
  return null;
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
/**
 * Ask the DataNode for a new shared memory segment.  This function must be
 * called with the manager lock held.  We will release the lock while
 * communicating with the DataNode.
 *
 * @param clientName    The current client name.
 * @param peer          The peer to use to talk to the DataNode.
 *
 * @return              Null if the DataNode does not support shared memory
 *                        segments, or experienced an error creating the
 *                        shm.  The shared memory segment itself on success.
 * @throws IOException  If there was an error communicating over the socket.
 *                        We will not throw an IOException unless the socket
 *                        itself (or the network) is the problem.
 */
private DfsClientShm requestNewShm(String clientName, DomainPeer peer)
    throws IOException {
  final DataOutputStream out = 
      new DataOutputStream(
          new BufferedOutputStream(peer.getOutputStream()));
  new Sender(out).requestShortCircuitShm(clientName);
  ShortCircuitShmResponseProto resp = 
      ShortCircuitShmResponseProto.parseFrom(
          PBHelper.vintPrefixed(peer.getInputStream()));
  String error = resp.hasError() ? resp.getError() : "(unknown)";
  switch (resp.getStatus()) {
  case SUCCESS:
    DomainSocket sock = peer.getDomainSocket();
    byte buf[] = new byte[1];
    FileInputStream fis[] = new FileInputStream[1];
    if (sock.recvFileInputStreams(fis, buf, 0, buf.length) < 0) {
      throw new EOFException("got EOF while trying to transfer the " +
          "file descriptor for the shared memory segment.");
    }
    if (fis[0] == null) {
      throw new IOException("the datanode " + datanode + " failed to " +
          "pass a file descriptor for the shared memory segment.");
    }
    try {
      DfsClientShm shm = 
          new DfsClientShm(PBHelper.convert(resp.getId()),
              fis[0], this, peer);
      if (LOG.isTraceEnabled()) {
        LOG.trace(this + ": createNewShm: created " + shm);
      }
      return shm;
    } finally {
      IOUtils.cleanup(LOG,  fis[0]);
    }
  case ERROR_UNSUPPORTED:
    // The DataNode just does not support short-circuit shared memory
    // access, and we should stop asking.
    LOG.info(this + ": datanode does not support short-circuit " +
        "shared memory access: " + error);
    disabled = true;
    return null;
  default:
    // The datanode experienced some kind of unexpected error when trying to
    // create the short-circuit shared memory segment.
    LOG.warn(this + ": error requesting short-circuit shared memory " +
        "access: " + error);
    return null;
  }
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
DfsClientShm(ShmId shmId, FileInputStream stream, EndpointShmManager manager,
    DomainPeer peer) throws IOException {
  super(shmId, stream);
  this.manager = manager;
  this.peer = peer;
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
public DomainPeer getPeer() {
  return peer;
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
private static DomainPeer getDomainPeerToDn(Configuration conf)
    throws IOException {
  DomainSocket sock =
      DomainSocket.connect(conf.get(DFS_DOMAIN_SOCKET_PATH_KEY));
  return new DomainPeer(sock);
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
/**
 * Get a RemoteBlockReader that communicates over a UNIX domain socket.
 *
 * @return The new BlockReader, or null if we failed to create the block
 * reader.
 *
 * @throws InvalidToken    If the block token was invalid.
 * Potentially other security-related execptions.
 */
private BlockReader getRemoteBlockReaderFromDomain() throws IOException {
  if (pathInfo == null) {
    pathInfo = clientContext.getDomainSocketFactory()
        .getPathInfo(inetSocketAddress, conf.getShortCircuitConf());
  }
  if (!pathInfo.getPathState().getUsableForDataTransfer()) {
    PerformanceAdvisory.LOG.debug("{}: not trying to create a " +
        "remote block reader because the UNIX domain socket at {}" +
         " is not usable.", this, pathInfo);
    return null;
  }
  LOG.trace("{}: trying to create a remote block reader from the UNIX domain "
      + "socket at {}", this, pathInfo.getPath());

  while (true) {
    BlockReaderPeer curPeer = nextDomainPeer();
    if (curPeer == null) break;
    if (curPeer.fromCache) remainingCacheTries--;
    DomainPeer peer = (DomainPeer)curPeer.peer;
    BlockReader blockReader = null;
    try {
      blockReader = getRemoteBlockReader(peer);
      return blockReader;
    } catch (IOException ioe) {
      IOUtilsClient.cleanup(LOG, peer);
      if (isSecurityException(ioe)) {
        LOG.trace("{}: got security exception while constructing a remote "
                + " block reader from the unix domain socket at {}",
            this, pathInfo.getPath(), ioe);
        throw ioe;
      }
      if (curPeer.fromCache) {
        // Handle an I/O error we got when using a cached peer.  These are
        // considered less serious because the underlying socket may be stale.
        LOG.debug("Closed potentially stale domain peer {}", peer, ioe);
      } else {
        // Handle an I/O error we got when using a newly created domain peer.
        // We temporarily disable the domain socket path for a few minutes in
        // this case, to prevent wasting more time on it.
        LOG.warn("I/O error constructing remote block reader.  Disabling " +
            "domain socket " + peer.getDomainSocket(), ioe);
        clientContext.getDomainSocketFactory()
            .disableDomainSocketPath(pathInfo.getPath());
        return null;
      }
    } finally {
      if (blockReader == null) {
        IOUtilsClient.cleanup(LOG, peer);
      }
    }
  }
  return null;
}
 

import org.apache.hadoop.hdfs.net.DomainPeer; //导入依赖的package包/类
/**
 * Ask the DataNode for a new shared memory segment.  This function must be
 * called with the manager lock held.  We will release the lock while
 * communicating with the DataNode.
 *
 * @param clientName    The current client name.
 * @param peer          The peer to use to talk to the DataNode.
 *
 * @return              Null if the DataNode does not support shared memory
 *                        segments, or experienced an error creating the
 *                        shm.  The shared memory segment itself on success.
 * @throws IOException  If there was an error communicating over the socket.
 *                        We will not throw an IOException unless the socket
 *                        itself (or the network) is the problem.
 */
private DfsClientShm requestNewShm(String clientName, DomainPeer peer)
    throws IOException {
  final DataOutputStream out =
      new DataOutputStream(
          new BufferedOutputStream(peer.getOutputStream()));
  new Sender(out).requestShortCircuitShm(clientName);
  ShortCircuitShmResponseProto resp =
      ShortCircuitShmResponseProto.parseFrom(
        PBHelperClient.vintPrefixed(peer.getInputStream()));
  String error = resp.hasError() ? resp.getError() : "(unknown)";
  switch (resp.getStatus()) {
  case SUCCESS:
    DomainSocket sock = peer.getDomainSocket();
    byte buf[] = new byte[1];
    FileInputStream fis[] = new FileInputStream[1];
    if (sock.recvFileInputStreams(fis, buf, 0, buf.length) < 0) {
      throw new EOFException("got EOF while trying to transfer the " +
          "file descriptor for the shared memory segment.");
    }
    if (fis[0] == null) {
      throw new IOException("the datanode " + datanode + " failed to " +
          "pass a file descriptor for the shared memory segment.");
    }
    try {
      DfsClientShm shm =
          new DfsClientShm(PBHelperClient.convert(resp.getId()),
              fis[0], this, peer);
      LOG.trace("{}: createNewShm: created {}", this, shm);
      return shm;
    } finally {
      try {
        fis[0].close();
      } catch (Throwable e) {
        LOG.debug("Exception in closing " + fis[0], e);
      }
    }
  case ERROR_UNSUPPORTED:
    // The DataNode just does not support short-circuit shared memory
    // access, and we should stop asking.
    LOG.info(this + ": datanode does not support short-circuit " +
        "shared memory access: " + error);
    disabled = true;
    return null;
  default:
    // The datanode experienced some kind of unexpected error when trying to
    // create the short-circuit shared memory segment.
    LOG.warn(this + ": error requesting short-circuit shared memory " +
        "access: " + error);
    return null;
  }
}