Read the block with the given ID into the given byte buffer.
@param blockId  block id of the block to load
@param block    The block to write the bytes to
@return The original address of the block
@throws IOException
769541
void loadBlock(BlockId blockId, ByteBuffer block) throws IOException {
        synchronized (m_accessLock) {
            if (! m_blockPathMap.containsKey(blockId)) {
                throw new IllegalArgumentException("Request to load block that is not stored: " + blockId);
            }

            int origPosition = block.position();
            block.position(0);
            Path blockPath = m_blockPathMap.get(blockId);
            try (SeekableByteChannel channel = Files.newByteChannel(blockPath)) {
                channel.read(block);
            }
            finally {
                block.position(origPosition);
            }
        }
    }
1194603
private static int getBlockIdInFile(DistributedFileSystem srcFs, 
      Path srcPath, long blockId) throws IOException {
    FileStatus srcStat = srcFs.getFileStatus(srcPath);
    LocatedBlocks lbs =
        srcFs.getClient().getLocatedBlocks(srcPath.toUri().getPath(), 0, 
            srcStat.getLen());
    
    int i = 0;
    LOG.info("Look for block " + blockId + " in file " + srcPath);
    for (LocatedBlock lb: lbs.getLocatedBlocks()) {
      if (lb.getBlock().getBlockId() == blockId) {
        return i;
      }
      i++;
    }
    
    return -1;
  }
407567
def _writeBlock(block, blockID):
    ''''''
    with open("blockIDs.txt", "a") as fp:
        fp.write("blockID: " + str(blockID) + "\n")
        sentences = ""
        for sentence in block:
            sentences += sentence+","
        fp.write("block sentences: "+sentences[:-1]+"\n")
        fp.write("\n")
105033
void storeBlock(BlockId blockId, ByteBuffer block) throws IOException {
        synchronized (m_accessLock) {
            if (m_blockPathMap.containsKey(blockId)) {
                throw new IllegalArgumentException("Request to store block that is already stored: "
                                                    + blockId.toString());
            }

            int origPosition = block.position();
            block.position(0);
            Path blockPath = makeBlockPath(blockId);
            try (SeekableByteChannel channel = Files.newByteChannel(blockPath, OPEN_OPTIONS, PERMISSIONS)) {
                channel.write(block);
            }
            finally {
                block.position(origPosition);
            }

            m_blockPathMap.put(blockId, blockPath);
        }
    }
878296
private static int get(byte[] block, int offset, byte[] b, int off, int len) {
    System.arraycopy(block, offset, b, off, len);
    return len;
  }
1212162
void getBlockChecksum(DataInputStream in,
      VersionAndOpcode versionAndOpcode) throws IOException {
    // header
    BlockChecksumHeader blockChecksumHeader =
        new BlockChecksumHeader(versionAndOpcode);
    blockChecksumHeader.readFields(in);
    final int namespaceId = blockChecksumHeader.getNamespaceId();
    final Block block = new Block(blockChecksumHeader.getBlockId(), 0,
            blockChecksumHeader.getGenStamp());

    DataOutputStream out = null;
    InputStream rawStreamIn = null;
    DataInputStream streamIn = null;

    ReplicaToRead ri = datanode.data.getReplicaToRead(namespaceId, block);
    if (ri == null) {
      throw new IOException("Unknown block");
    }

    updateCurrentThreadName("getting checksum for block " + block);
    try {
      int bytesPerCRC;
      int checksumSize;

      long crcPerBlock;
      MD5Hash md5;
      if (!ri.isInlineChecksum()) {
        rawStreamIn = BlockWithChecksumFileReader.getMetaDataInputStream(
            datanode.data, namespaceId, block);
        streamIn = new DataInputStream(new BufferedInputStream(rawStreamIn,
            BUFFER_SIZE));

        final BlockMetadataHeader header = BlockMetadataHeader
            .readHeader(streamIn);
        final DataChecksum checksum = header.getChecksum();
        bytesPerCRC = checksum.getBytesPerChecksum();
        checksumSize = checksum.getChecksumSize();
        crcPerBlock = (((BlockWithChecksumFileReader.MetaDataInputStream) rawStreamIn)
            .getLength() - BlockMetadataHeader.getHeaderSize()) / checksumSize;

       //compute block checksum
       md5 = MD5Hash.digest(streamIn);
      } else {
        bytesPerCRC = ri.getBytesPerChecksum();
        checksumSize = DataChecksum.getChecksumSizeByType(ri.getChecksumType());
        ReplicaToRead replica = datanode.data.getReplicaToRead(namespaceId, block);
        rawStreamIn = replica.getBlockInputStream(datanode, 0);
        streamIn = new DataInputStream(new BufferedInputStream(rawStreamIn,
            BUFFER_SIZE));

        long lengthLeft = ((FileInputStream) rawStreamIn).getChannel().size()
            - BlockInlineChecksumReader.getHeaderSize();
        if (lengthLeft == 0) {
          crcPerBlock = 0;
          md5 = MD5Hash.digest(new byte[0]);
        } else {
          crcPerBlock = (lengthLeft - 1) / (checksumSize + bytesPerCRC) + 1;
          MessageDigest digester = MD5Hash.getDigester();
          byte[] buffer = new byte[checksumSize];
          while (lengthLeft > 0) {
            if (lengthLeft >= bytesPerCRC + checksumSize) {
              streamIn.skip(bytesPerCRC);
              IOUtils.readFully(streamIn, buffer, 0, buffer.length);
              lengthLeft -= bytesPerCRC
                  + checksumSize;
            } else if (lengthLeft > checksumSize) {
              streamIn.skip(lengthLeft - checksumSize);
              IOUtils.readFully(streamIn, buffer, 0, buffer.length);
              lengthLeft = 0;
            } else {
              out = new DataOutputStream(
                  NetUtils.getOutputStream(s, datanode.socketWriteTimeout));
              out.writeShort(DataTransferProtocol.OP_STATUS_ERROR);
              out.flush();
              // report to name node the corruption.
              DataBlockScanner.reportBadBlocks(block, namespaceId, datanode);
              throw new IOException("File for namespace " + namespaceId
                  + " block " + block + " seems to be corrupted");
            }
            digester.update(buffer);
          }
          md5 = new MD5Hash(digester.digest(), checksumSize * crcPerBlock);
        }
      }

      if (LOG.isDebugEnabled()) {
        LOG.debug("block=" + block + ", bytesPerCRC=" + bytesPerCRC
            + ", crcPerBlock=" + crcPerBlock + ", md5=" + md5);
      }

      //write reply
      out = new DataOutputStream(
          NetUtils.getOutputStream(s, datanode.socketWriteTimeout));
      out.writeShort(DataTransferProtocol.OP_STATUS_SUCCESS);
      out.writeInt(bytesPerCRC);
      out.writeLong(crcPerBlock);
      md5.write(out);
      out.flush();
    } finally {
      IOUtils.closeStream(out);
      if (streamIn != null) {
        IOUtils.closeStream(streamIn);
      }
      if (rawStreamIn != null) {
        IOUtils.closeStream(rawStreamIn);
      }
    }
  }
429365
public BlockInfo getBlockInfo(final long blockId) throws IOException {
    return retryRPC(() -> {
      return GrpcUtils.fromProto(
          mClient.getBlockInfo(GetBlockInfoPRequest.newBuilder().setBlockId(blockId).build())
              .getBlockInfo());
    });
  }
482191
BlocksWithLocations getBlocks(DatanodeID datanode, long size)
    throws IOException {
    readLock();
    try {
      checkSuperuserPrivilege();

      DatanodeDescriptor node = getDatanode(datanode);
      if (node == null) {
        NameNode.stateChangeLog.warn("BLOCK* NameSystem.getBlocks: "
          + "Asking for blocks from an unrecorded node " + datanode.getName());
        throw new IllegalArgumentException(
          "Unexpected exception.  Got getBlocks message for datanode " +
            datanode.getName() + ", but there is no info for it");
      }

      int numBlocks = node.numBlocks();
      if (numBlocks == 0) {
        return new BlocksWithLocations(new BlockWithLocations[0]);
      }
      Iterator<BlockInfo> iter = node.getBlockIterator();
      int startBlock = r.nextInt(numBlocks); // starting from a random block
      // skip blocks
      for (int i = 0; i < startBlock; i++) {
        iter.next();
      }
      List<BlockWithLocations> results = new ArrayList<BlockWithLocations>();
      long totalSize = 0;
      while (totalSize < size && iter.hasNext()) {
        totalSize += addBlock(iter.next(), results);
      }
      if (totalSize < size) {
        iter = node.getBlockIterator(); // start from the beginning
        for (int i = 0; i < startBlock && totalSize < size; i++) {
          totalSize += addBlock(iter.next(), results);
        }
      }

      return new BlocksWithLocations(
        results.toArray(new BlockWithLocations[results.size()]));
    } finally {
      readUnlock();
    }
  }
288454
private void writeBlock(DataInputStream in,
      VersionAndOpcode versionAndOpcode) throws IOException {
    DatanodeInfo srcDataNode = null;
    if (LOG.isDebugEnabled()) {
      LOG.debug("writeBlock receive buf size " + s.getReceiveBufferSize() +
                " tcp no delay " + s.getTcpNoDelay());
    }
    //
    // Read in the header
    //
    long startTime = System.currentTimeMillis();
    
    WriteBlockHeader headerToReceive = new WriteBlockHeader(
        versionAndOpcode);
    headerToReceive.readFields(in);

    WriteOptions options = headerToReceive.getWritePipelineInfo()
        .getWriteOptions();
    boolean ioprioEnabled = !options.isIoprioDisabled();
    if (ioprioEnabled) {
      NativeIO.ioprioSetIfPossible(options.getIoprioClass(),
          options.getIoprioData());
    }

    int namespaceid = headerToReceive.getNamespaceId();
    Block block = new Block(headerToReceive.getBlockId(), 
        dataXceiverServer.estimateBlockSize, headerToReceive.getGenStamp());
    if (LOG.isInfoEnabled()) {
      if (remoteAddress == null) {
        getAddresses();
      }
      LOG.info("Receiving block " + block + 
               " src: " + remoteAddress +
               " dest: " + localAddress);
    }
    int pipelineSize = headerToReceive.getPipelineDepth(); // num of datanodes in entire pipeline
    boolean isRecovery = headerToReceive.isRecoveryFlag(); // is this part of recovery?
    String client = headerToReceive.getWritePipelineInfo().getClientName(); // working on behalf of this client
    boolean hasSrcDataNode = headerToReceive.getWritePipelineInfo()
        .hasSrcDataNode(); // is src node info present
    if (hasSrcDataNode) {
      srcDataNode = headerToReceive.getWritePipelineInfo().getSrcDataNode();
    }
    int numTargets = headerToReceive.getWritePipelineInfo().getNumTargets();
    DatanodeInfo targets[] = headerToReceive.getWritePipelineInfo().getNodes();
    int fadvise = headerToReceive.getWritePipelineInfo().getWriteOptions()
        .getFadvise();

    DataOutputStream mirrorOut = null;  // stream to next target
    DataInputStream mirrorIn = null;    // reply from next target
    DataOutputStream replyOut = null;   // stream to prev target
    Socket mirrorSock = null;           // socket to next target
    BlockReceiver blockReceiver = null; // responsible for data handling
    String mirrorNode = null;           // the name:port of next target
    String firstBadLink = "";           // first datanode that failed in connection setup

    updateCurrentThreadName("receiving block " + block + " client=" + client);
    InjectionHandler.processEvent(InjectionEvent.DATANODE_WRITE_BLOCK);
    try {
      boolean ifProfileEnabled = headerToReceive.getWritePipelineInfo()
          .getWriteOptions().ifProfileEnabled();
      boolean isSecondary = (targets.length + 1 != pipelineSize);
      // open a block receiver and check if the block does not exist
      blockReceiver = new BlockReceiver(namespaceid, block, block, in, 
          s.getRemoteSocketAddress().toString(),
          s.getLocalSocketAddress().toString(),
          isRecovery, client, srcDataNode, datanode, isSecondary, fadvise,
          ifProfileEnabled, versionAndOpcode.getDataTransferVersion() >=
            DataTransferProtocol.PACKET_INCLUDE_VERSION_VERSION,
            options.getSyncFileRange());

      // get a connection back to the previous target
      replyOut = new DataOutputStream(new BufferedOutputStream(
                     NetUtils.getOutputStream(s, datanode.socketWriteTimeout),
                     SMALL_BUFFER_SIZE));

      //
      // Open network conn to backup machine, if 
      // appropriate
      //
      if (targets.length > 0) {
        InetSocketAddress mirrorTarget = null;
        // Connect to backup machine
        mirrorNode = targets[0].getName();
        mirrorTarget = NetUtils.createSocketAddr(mirrorNode);
        mirrorSock = datanode.newSocket();
        try {
          int timeoutValue = datanode.socketTimeout +
                             (datanode.socketReadExtentionTimeout * numTargets);
          int writeTimeout = datanode.socketWriteTimeout + 
                             (datanode.socketWriteExtentionTimeout * numTargets);
          NetUtils.connect(mirrorSock, mirrorTarget, timeoutValue);
          mirrorSock.setSoTimeout(timeoutValue);
          mirrorSock.setSendBufferSize(DEFAULT_DATA_SOCKET_SIZE);
          mirrorOut = new DataOutputStream(
             new BufferedOutputStream(
                         NetUtils.getOutputStream(mirrorSock, writeTimeout),
                         SMALL_BUFFER_SIZE));
          mirrorIn = new DataInputStream(NetUtils.getInputStream(mirrorSock));

          // Write header: Copied from DFSClient.java!
          WriteBlockHeader headerToSend = new WriteBlockHeader(
              versionAndOpcode.getDataTransferVersion(), namespaceid,
              block.getBlockId(), block.getGenerationStamp(), pipelineSize,
              isRecovery, hasSrcDataNode, srcDataNode, targets.length - 1, targets,
              client);
          headerToSend.getWritePipelineInfo().setWriteOptions(options);
          headerToSend.writeVersionAndOpCode(mirrorOut);
          headerToSend.write(mirrorOut);
          blockReceiver.writeChecksumHeader(mirrorOut);
          mirrorOut.flush();

          // read connect ack (only for clients, not for replication req)
          if (client.length() != 0) {
            firstBadLink = Text.readString(mirrorIn);
            if (LOG.isDebugEnabled() || firstBadLink.length() > 0) {
              LOG.info("Datanode " + targets.length +
                       " got response for connect ack " +
                       " from downstream datanode with firstbadlink as " +
                       firstBadLink);
            }
          }

        } catch (IOException e) {
          if (client.length() != 0) {
            Text.writeString(replyOut, mirrorNode);
            replyOut.flush();
          }
          IOUtils.closeStream(mirrorOut);
          mirrorOut = null;
          IOUtils.closeStream(mirrorIn);
          mirrorIn = null;
          IOUtils.closeSocket(mirrorSock);
          mirrorSock = null;
          if (client.length() > 0) {
            throw e;
          } else {
            LOG.info(datanode.getDatanodeInfo() + ":Exception transfering block " +
                     block + " to mirror " + mirrorNode +
                     ". continuing without the mirror.\n" +
                     StringUtils.stringifyException(e));
          }
        }
      }

      // send connect ack back to source (only for clients)
      if (client.length() != 0) {
        if (LOG.isDebugEnabled() || firstBadLink.length() > 0) {
          LOG.info("Datanode " + targets.length +
                   " forwarding connect ack to upstream firstbadlink is " +
                   firstBadLink);
        }
        Text.writeString(replyOut, firstBadLink);
        replyOut.flush();
      }

      // receive the block and mirror to the next target
      String mirrorAddr = (mirrorSock == null) ? null : mirrorNode;
      long totalReceiveSize = blockReceiver.receiveBlock(mirrorOut, mirrorIn, replyOut,
                                 mirrorAddr, null, targets.length);

      // if this write is for a replication request (and not
      // from a client), then confirm block. For client-writes,
      // the block is finalized in the PacketResponder.
      if (client.length() == 0) {
        datanode.notifyNamenodeReceivedBlock(namespaceid, block, null);
        LOG.info("Received block " + block + 
                 " src: " + remoteAddress +
                 " dest: " + localAddress +
                 " of size " + block.getNumBytes());
      } else {
        // Log the fact that the block has been received by this datanode and
        // has been written to the local disk on this datanode.
        LOG.info("Received Block " + block +
            " src: " + remoteAddress +
            " dest: " + localAddress +
            " of size " + block.getNumBytes() +
            " and written to local disk");
      }

      if (datanode.blockScanner != null) {
        datanode.blockScanner.addBlock(namespaceid, block);
      }
      
      long writeDuration = System.currentTimeMillis() - startTime;
      datanode.myMetrics.bytesWrittenLatency.inc(writeDuration);
      if (totalReceiveSize > KB_RIGHT_SHIFT_MIN) {
        datanode.myMetrics.bytesWrittenRate.inc((int) (totalReceiveSize >> KB_RIGHT_SHIFT_BITS),
                              writeDuration);
      }

    } catch (IOException ioe) {
      LOG.info("WriteBlock: Got exception while receiving block: " + block + 
                " from " + s.getInetAddress() + ", client: " + client + ":\n" + 
                StringUtils.stringifyException(ioe) );
      throw ioe;
    } finally {
      // close all opened streams
      IOUtils.closeStream(mirrorOut);
      IOUtils.closeStream(mirrorIn);
      IOUtils.closeStream(replyOut);
      IOUtils.closeSocket(mirrorSock);
      IOUtils.closeStream(blockReceiver);
    }
  }
1176875
@Override
  public BlockLocationsResponse getBlockLocations(GetBlockLocationsRequest req) throws IOException {
    Server.setOrignalCaller(req.getOriginalCaller());
    return new BlockLocationsResponse(getBlockLocationsInternal(req.getSrc(), req.getOffset(),
        req.getLength(), BlockMetaInfoType.NONE));
  }
461113
private synchronized void readBlockSizeInfo() throws IOException {
    if (!transferBlockSize) {
      return;
    }
    blkLenInfoUpdated = true;
    isBlockFinalized = in.readBoolean();
    updatedBlockLength = in.readLong();
    if (dataTransferVersion >= DataTransferProtocol.READ_PROFILING_VERSION) {
      readDataNodeProfilingData();
    }
    
    if (LOG.isDebugEnabled()) {
      LOG.debug("ifBlockComplete? " + isBlockFinalized + " block size: "
          + updatedBlockLength);
    }      
  }
174116
def deserialize_block(value):
        
        # Block id strings are stored under batch/txn ids for reference.
        # Only Blocks, not ids or Nones, should be returned by _get_block.
        block = Block()
        block.ParseFromString(value)
        return BlockWrapper(
            block=block)
250932
protected final void copyBitcoinHeaderTo(final Block block) {
        block.nonce = nonce;
        block.prevBlockHash = prevBlockHash;
        block.merkleRoot = getMerkleRoot();
        block.version = version;
        block.time = time;
        block.difficultyTarget = difficultyTarget;
        block.transactions = null;
        block.hash = getHash();
    }
994863
public void write(DataOutput out) throws IOException {
    out.writeLong(blockId);
    out.writeLong(blockGenStamp);
    super.write(out);
  }
214724
public static BlockReader newBlockReader( int dataTransferVersion,
                                     int namespaceId,
                                     Socket sock, String file, long blockId,
                                     long genStamp,
                                     long startOffset, long len,
                                     int bufferSize, boolean verifyChecksum)
                                     throws IOException {
    return newBlockReader(dataTransferVersion, namespaceId,
                          sock, file, blockId, genStamp,
                          startOffset,
                          len, bufferSize, verifyChecksum, "",
                          Long.MAX_VALUE, -1, false, null,
                          new ReadOptions());
  }
1271317
synchronized void deleteBlock(Block block) {
    if (!isInitialized()) {
      return;
    }
    BlockScanInfo info = blockMap.get(block);
    if ( info != null ) {
      delBlockInfo(info);
    }
  }
841822
protected final boolean readBlock(final long block, final byte[] buffer) throws IOException {
        final int bytesRead = readData(block * this.blockSize, buffer, 0, this.blockSize);

        if (bytesRead <= 0) {
            return false;
        }

        if (this.blockSize != bytesRead) {
            throw new IOException("Could not deserialize a complete block");
        }

        return true;
    }