Compute block replication and block invalidation work
that can be scheduled on data-nodes.
The datanode will be informed of this work at the next heartbeat.

@return number of blocks scheduled for replication or removal.
193066
public int computeDatanodeWork() throws IOException {
    int workFound = 0;
    int blocksToProcess = 0;
    int nodesToProcess = 0;
    // blocks should not be replicated or removed if safe mode is on
    if (isInSafeMode()) {
      updateReplicationCounts(workFound);
      return workFound;
    }
    synchronized (heartbeats) {
      blocksToProcess = (int) (heartbeats.size()
        * ReplicationConfigKeys.replicationWorkMultiplier);
      nodesToProcess = (int) Math.ceil((double) heartbeats.size()
        * ReplicationConfigKeys.INVALIDATE_WORK_PCT_PER_ITERATION / 100);
    }

    workFound = computeReplicationWork(blocksToProcess);

    // Update FSNamesystemMetrics counters
    updateReplicationCounts(workFound);

    workFound += computeInvalidateWork(nodesToProcess);
    return workFound;
  }
450583
void removeBlock(Block block) {
    BlockInfo blockInfo = removeBlockFromMap(block);
    if (blockInfo == null)
      return;
    blockInfo.inode = null;
    for(int idx = blockInfo.numNodes()-1; idx >= 0; idx--) {
      DatanodeDescriptor dn = blockInfo.getDatanode(idx);
      dn.removeBlock(blockInfo); // remove from the list and wipe the location
    }
  }
107943
public short getBlockReplication(BlockInfo block) {
    if (storage.isSourceBlock(block)) {
      return getReplication();
    } else {
      if (storage.getStorageType() == StorageType.RAID_STORAGE) {
        return ((INodeRaidStorage)storage).getCodec().parityReplication;
      } else {
        throw new IllegalStateException("parity block " + block +
            " belongs to a non-raid file");
      }
    }
  }
29098
private boolean blockOnPipeLine() {
    if (this.isKilled()) {
      return true;
    }

    // For pipelining of jobs. Will watch other jobs.
    if (!this.pipelineJobs.isEmpty()) {
      String blockedList = "";
      final ArrayList<BlockingStatus> blockingStatus =
          new ArrayList<>();
      for (final String waitingJobId : this.pipelineJobs) {
        final Status status = this.watcher.peekStatus(waitingJobId);
        if (status != null && !Status.isStatusFinished(status)) {
          final BlockingStatus block = this.watcher.getBlockingStatus(waitingJobId);
          blockingStatus.add(block);
          blockedList += waitingJobId + ",";
        }
      }
      if (!blockingStatus.isEmpty()) {
        this.logger.info("Pipeline job " + this.jobId + " waiting on " + blockedList
            + " in execution " + this.watcher.getExecId());

        for (final BlockingStatus bStatus : blockingStatus) {
          this.logger.info("Waiting on pipelined job " + bStatus.getJobId());
          this.currentBlockStatus = bStatus;
          bStatus.blockOnFinishedStatus();
          if (this.isKilled()) {
            this.logger.info("Job was killed while waiting on pipeline. Quiting.");
            return true;
          } else {
            this.logger.info("Pipelined job " + bStatus.getJobId() + " finished.");
          }
        }
      }
    }

    this.currentBlockStatus = null;
    return false;
  }
1200429
LocatedBlocksWithMetaInfo updateDatanodeInfo(LocatedBlocks locatedBlocks)
      throws IOException {
    if (locatedBlocks.getLocatedBlocks().size() == 0)
      return new LocatedBlocksWithMetaInfo(locatedBlocks.getFileLength(),
          locatedBlocks.getLocatedBlocks(), false,
          DataTransferProtocol.DATA_TRANSFER_VERSION, getNamespaceId(),
          this.nameNode.getClientProtocolMethodsFingerprint());
    List<LocatedBlock> newBlocks = new ArrayList<LocatedBlock>();

    readLock();
    try {
      for (LocatedBlock locBlock: locatedBlocks.getLocatedBlocks()) {
        Block block = locBlock.getBlock();
        int numNodes = blocksMap.numNodes(block);
        int numCorruptNodes = countNodes(block).corruptReplicas();
        int numCorruptReplicas = corruptReplicas.numCorruptReplicas(block); 

        if (numCorruptNodes != numCorruptReplicas) {
          LOG.warn("Inconsistent number of corrupt replicas for " + 
                   block + "blockMap has " + numCorruptNodes + 
                   " but corrupt replicas map has " + numCorruptReplicas);
        }

        boolean blockCorrupt = numCorruptNodes == numNodes;
        int numMachineSet = blockCorrupt ? numNodes : (numNodes - numCorruptNodes);
        DatanodeDescriptor[] machineSet = new DatanodeDescriptor[numMachineSet];

        if (numMachineSet > 0) {
          numNodes = 0;
          for(Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(block); 
              it.hasNext();) {
            DatanodeDescriptor dn = it.next();
            boolean replicaCorrupt = corruptReplicas.isReplicaCorrupt(block, dn);
            if (blockCorrupt || (!blockCorrupt && !replicaCorrupt))
              machineSet[numNodes++] = dn;
          }
        }

        // We need to make a copy of the block object before releasing the lock
        // to prevent the state of block is changed after that and before the
        // object is serialized to clients, to avoid potential inconsistency.
        // Further optimization is possible to avoid some object copy. Since it
        // is so far not a critical path. We leave a safe approach here.
        //
        Block blockCopy  = null;
        if (block != null) {
          blockCopy = new Block(block);
        }
        LocatedBlock newBlock = new LocatedBlock(blockCopy, machineSet, 0,
            blockCorrupt);
  newBlocks.add(newBlock);
      }
    } finally {
      readUnlock();
    }

    return new LocatedBlocksWithMetaInfo(locatedBlocks.getFileLength(),
        newBlocks, false, DataTransferProtocol.DATA_TRANSFER_VERSION,
        getNamespaceId(), this.nameNode.getClientProtocolMethodsFingerprint());
  }
714063
static public int getBlockCrc(DataNode datanode, ReplicaToRead ri,
      int namespaceId, Block block) throws IOException {

    InputStream rawStreamIn = null;
    DataInputStream streamIn = null;
    int blockCrc = 0;

    try {
      int bytesPerCRC;
      int checksumSize;

      bytesPerCRC = ri.getBytesPerChecksum();
      int checksumType = ri.getChecksumType();
      if (checksumType != DataChecksum.CHECKSUM_CRC32) {
        throw new IOException("File Checksum now is only supported for CRC32");
      }
      DataChecksum dataChecksum = DataChecksum.newDataChecksum(checksumType,
          bytesPerCRC);
      checksumSize = dataChecksum.getChecksumSize();
      
      rawStreamIn = ri.getBlockInputStream(datanode, 0);
      streamIn = new DataInputStream(new BufferedInputStream(rawStreamIn,
          FSConstants.BUFFER_SIZE));
      IOUtils.skipFully(streamIn, BlockInlineChecksumReader.getHeaderSize());

      long lengthLeft = ((FileInputStream) rawStreamIn).getChannel().size()
          - BlockInlineChecksumReader.getHeaderSize();
      if (lengthLeft == 0) {
        blockCrc = (int) dataChecksum.getValue();
      } else {
        byte[] buffer = new byte[checksumSize];
        boolean firstChecksum = true;

        while (lengthLeft > 0) {
          long dataByteLengh;

          if (lengthLeft >= bytesPerCRC + checksumSize) {
            lengthLeft -= bytesPerCRC + checksumSize;
            dataByteLengh = bytesPerCRC;
          } else if (lengthLeft > checksumSize) {
            dataByteLengh = lengthLeft - checksumSize;
            lengthLeft = 0;
          } else {
            // report to name node the corruption.
            DataBlockScanner.reportBadBlocks(block, namespaceId, datanode);
            throw new IOException("File for namespace " + namespaceId
                + " block " + block + " seems to be corrupted");
          }

          IOUtils.skipFully(streamIn, dataByteLengh);
          IOUtils.readFully(streamIn, buffer, 0, buffer.length);
          int intChecksum = DataChecksum.getIntFromBytes(buffer, 0);
          if (firstChecksum) {
            blockCrc = intChecksum;
            firstChecksum = false;
          } else {
            blockCrc = CrcConcat.concatCrc(blockCrc, intChecksum,
                (int) dataByteLengh);
          }
        }
      }

      if (LOG.isDebugEnabled()) {
        LOG.debug("block=" + block + ", bytesPerCRC=" + bytesPerCRC
            + ", crc=" + blockCrc);
      }
      return blockCrc;
    } finally {
      IOUtils.closeStream(streamIn);
      IOUtils.closeStream(rawStreamIn);
    }

  }
864016
public static int simulateSerialization(List<WayDataBlock> blocks) {
        int sum = 0;
        for (WayDataBlock wayDataBlock : blocks) {
            sum += mSimulateSerialization(wayDataBlock.getOuterWay());
            if (wayDataBlock.getInnerWays() != null) {
                for (List<Integer> list : wayDataBlock.getInnerWays()) {
                    sum += mSimulateSerialization(list);
                }
            }
        }
        return sum;
    }
403050
int numNodes(Block b) {
    BlockInfo info = blocks.get(b);
    return info == null ? 0 : info.numNodes();
  }
472260
private void removeBlocks(List<BlockInfo> blocks) {
    if (blocks == null) {
      return;
    }
    for (BlockInfo b : blocks) {
      removeFromExcessReplicateMap(b);
      neededReplications.remove(b, -1);
      corruptReplicas.removeFromCorruptReplicasMap(b);
      if (pendingReplications != null) {
        int replicas = pendingReplications.getNumReplicas(b);
        for (int i = 0; i < replicas; i++) {
          pendingReplications.remove(b);
        }    
      }    
      addToInvalidates(b, false);
      blocksMap.removeBlock(b);
    }
  }
789813
private int endCheckBlock()
    {
        Node n;
        int rv = END_DROPS_OFF;

        // check each statment and if the statement can continue onto the next
        // one, then check the next statement
        for (n=first; ((rv & END_DROPS_OFF) != 0) && n != null; n = n.next)
        {
            rv &= ~END_DROPS_OFF;
            rv |= n.endCheck();
        }
        return rv;
    }
571225
boolean removeNode(Block b, DatanodeDescriptor node) {
    BlockInfo info = blocks.get(b);
    if (info == null)
      return false;

    // remove block from the data-node list and the node from the block info
    boolean removed = node.removeBlock(info);

    if (info.getDatanode(0) == null     // no datanodes left
              && info.inode == null) {  // does not belong to a file
      removeBlockFromMap(b); // remove block from the map
    }
    return removed;
  }
1136865
public Double blockerRecall()
    {
        double n = 0;
        for (int i=0; i<pairs.length; i++) {
            if (correctPair(i)) {
                n++;
            }
        }
        return new Double(n/numCorrectPairs);
    }
810413
public long execute(Consumer<StatusContext> block) {
        long numUnitsProcessed = 0L;


        // Keep track of average time used per unig (average time per unit)
        MovingAverage averageTPU = new MovingAverage();

        // Average imposed delay
        MovingAverage averageDelay = new MovingAverage();

        StatusContext statusContext = new StatusContext();

        while (numUnitsProcessed < maxUnitsProcessedInSequence) {

            statusContext.reset();

            synchronized (semaphore) {
                if (shutdownRequests.size() > 0) {
                    String info = "Shutting down processing due to request: " + shutdownRequests.remove(0);
                    out.println(info);
                    inform(numUnitsProcessed, averageTPU, averageDelay);

                    return numUnitsProcessed;
                }
            }

            long startTime = System.currentTimeMillis();
            block.accept(statusContext);
            long endTime = System.currentTimeMillis();

            if (statusContext.isFinished() || (!statusContext.getStatus() && --numberOfErrorsAccepted < 0)) {
                return numUnitsProcessed;
            }

            averageTPU.update(endTime - startTime);
            ++numUnitsProcessed;

            // Possibly delay processing so that we do not process more than 'maxUnitsProcessedPerMinute' - on average
            final double slotTime = (60000) / maxUnitsProcessedPerMinute;   // milliseconds
            double averageProcessingTime = averageTPU.getAverage(); // milliseconds
            long waitTime = (long) (slotTime - Math.ceil(averageProcessingTime));  // milliseconds

            if (waitTime > 0) {
                averageDelay.update(waitTime);
                synchronized (semaphore) {
                    try {
                        semaphore.wait(waitTime);
                    } catch (InterruptedException ignore) {}
                }
            }
            synchronized (semaphore) {
                if (shutdownRequests.size() > 0) {
                    String info = "Shutting down processing due to request: " + shutdownRequests.remove(0);
                    out.println(info);
                    inform(numUnitsProcessed, averageTPU, averageDelay);

                    return numUnitsProcessed;
                }
            }

            if ((numUnitsProcessed % INFORM_EACH_N_UNITS) == 0) {
                // Inform each N:th time through
                inform(numUnitsProcessed, averageTPU, averageDelay);
            }
        }
        if ((numUnitsProcessed % INFORM_EACH_N_UNITS) != 0) {
            // Only inform if we have not just done that :)
            inform(numUnitsProcessed, averageTPU, averageDelay);
        }
        return numUnitsProcessed;
    }
882448
List<List<BlockInfo>> chooseUnderReplicatedBlocks(int blocksToProcess) {
    // initialize data structure for the return value
    List<List<BlockInfo>> blocksToReplicate =
      new ArrayList<List<BlockInfo>>(UnderReplicatedBlocks.LEVEL);
    for (int i = 0; i < UnderReplicatedBlocks.LEVEL; i++) {
      blocksToReplicate.add(new ArrayList<BlockInfo>());
    }

    writeLock();
    try {
      synchronized (neededReplications) {
        if (neededReplications.size() == 0) {
          return blocksToReplicate;
        }

        for (int priority = 0; priority<UnderReplicatedBlocks.LEVEL; priority++) {
        // Go through all blocks that need replications of priority
        BlockIterator neededReplicationsIterator = neededReplications.iterator(priority);
        int numBlocks = neededReplications.size(priority);
        if (replIndex[priority] > numBlocks) {
          replIndex[priority] = 0;
        }
        // skip to the first unprocessed block, which is at replIndex
        for (int i = 0; i < replIndex[priority] && neededReplicationsIterator.hasNext(); i++) {
          neededReplicationsIterator.next();
        }
        // # of blocks to process for this priority
        int blocksToProcessIter = getQuotaForThisPriority(blocksToProcess,
            numBlocks, neededReplications.getSize(priority+1));
        blocksToProcess -= blocksToProcessIter;

        for (int blkCnt = 0; blkCnt < blocksToProcessIter; blkCnt++, replIndex[priority]++) {
          if (!neededReplicationsIterator.hasNext()) {
            // start from the beginning
            replIndex[priority] = 0;
            neededReplicationsIterator = neededReplications.iterator(priority);
            assert neededReplicationsIterator.hasNext() :
              "neededReplications should not be empty.";
          }

          BlockInfo block = neededReplicationsIterator.next();
          blocksToReplicate.get(priority).add(block);
        } // end for
        }
      } // end try
      return blocksToReplicate;
    } finally {
      writeUnlock();
    }
  }
588260
public LocatedBlock recoverBlock(int namespaceId, Block block,
      boolean keepLength, DatanodeID[] datanodeids, boolean closeFile,
      long deadline) throws IOException {
    int errorCount = 0;

    // Number of "replicasBeingWritten" in 0.21 parlance - these are replicas
    // on DNs that are still alive from when the write was happening
    int rbwCount = 0;
    // Number of "replicasWaitingRecovery" in 0.21 parlance - these replicas
    // have survived a DN restart, and thus might be truncated (eg if the
    // DN died because of a machine power failure, and when the ext3 journal
    // replayed, it truncated the file
    int rwrCount = 0;

    List<BlockRecord> blockRecords = new ArrayList<BlockRecord>();
    List<InterDatanodeProtocol> datanodeProxies = new ArrayList<InterDatanodeProtocol>();
    // check generation stamps
    for (DatanodeID id : datanodeids) {
      try {
        InterDatanodeProtocol datanode;
        if (dnr!= null && dnr.equals(id)) {
          LOG.info("Skipping IDNPP creation for local id " + id
              + " when recovering " + block);
          datanode = dn;
        } else {
          LOG.info("Creating IDNPP for non-local id " + id + " (dnReg="
              + dnr + ") when recovering "
              + block);
          datanode = DataNode.createInterDataNodeProtocolProxy(id, conf,
              socketTimeout);
          datanodeProxies.add(datanode);
        }
        BlockSyncer.throwIfAfterTime(deadline);
        BlockRecoveryInfo info = datanode
            .startBlockRecovery(namespaceId, block);
        if (info == null) {
          LOG.info("No block metadata found for block " + block
              + " on datanode " + id);
          continue;
        }
        if (info.getBlock().getGenerationStamp() < block.getGenerationStamp()) {
          LOG.info("Only old generation stamp "
              + info.getBlock().getGenerationStamp() + " found on datanode "
              + id + " (needed block=" + block + ")");
          continue;
        }
        blockRecords.add(new BlockRecord(id, datanode, info));

        if (info.wasRecoveredOnStartup()) {
          rwrCount++;
        } else {
          rbwCount++;
        }
      } catch (BlockRecoveryTimeoutException e) {
        throw e;
      } catch (IOException e) {
        ++errorCount;
        InterDatanodeProtocol.LOG.warn(
            "Failed to getBlockMetaDataInfo for block (=" + block
                + ") from datanode (=" + id + ")", e);
      }
    }

    // If we *only* have replicas from post-DN-restart, then we should
    // include them in determining length. Otherwise they might cause us
    // to truncate too short.
    boolean shouldRecoverRwrs = (rbwCount == 0);

    List<BlockRecord> syncList = new ArrayList<BlockRecord>();
    long minlength = Long.MAX_VALUE;

    for (BlockRecord record : blockRecords) {
      BlockRecoveryInfo info = record.info;
      assert (info != null && info.getBlock().getGenerationStamp() >= block
          .getGenerationStamp());
      if (!shouldRecoverRwrs && info.wasRecoveredOnStartup()) {
        LOG.info("Not recovering replica " + record
            + " since it was recovered on "
            + "startup and we have better replicas");
        continue;
      }
      if (keepLength) {
        if (info.getBlock().getNumBytes() == block.getNumBytes()) {
          syncList.add(record);
        }
      } else {
        syncList.add(record);
        if (info.getBlock().getNumBytes() < minlength) {
          minlength = info.getBlock().getNumBytes();
        }
      }
    }

    if (syncList.isEmpty() && errorCount > 0) {
      DataNode.stopAllProxies(datanodeProxies);
      throw new IOException("All datanodes failed: block=" + block
          + ", datanodeids=" + Arrays.asList(datanodeids));
    }
    if (!keepLength) {
      block.setNumBytes(minlength);
    }
    return blockSyncer.syncBlock(block, syncList, closeFile, datanodeProxies,
        deadline);
  }
473775
@Override
	public long blocksAccessed() {
		double rpb = (double) (Buffer.BUFFER_SIZE / RecordPage.slotSize(p.schema()));
		return (int) Math.ceil(p.recordsOutput() / rpb);
	}
570837
public int getNumNonExceptionSucessors(BasicBlock block) {
        int numNonExceptionSuccessors = block.getNumNonExceptionSuccessors();
        if (numNonExceptionSuccessors < 0) {
            numNonExceptionSuccessors = 0;
            for (Iterator<Edge> i = outgoingEdgeIterator(block); i.hasNext();) {
                Edge edge = i.next();
                if (!edge.isExceptionEdge()) {
                    numNonExceptionSuccessors++;
                }
            }
            block.setNumNonExceptionSuccessors(numNonExceptionSuccessors);
        }
        return numNonExceptionSuccessors;
    }