Find out datanodes to store parity blocks and schedule raiding on one of them
@return whether we want to add the raid task back to the queue
193967
private boolean processRaidEncodingTask(RaidBlockInfo rbi) {
    ArrayList<ArrayList<DatanodeInfo>> stripeDatanodes =
        new ArrayList<ArrayList<DatanodeInfo>>();
    BlockInfo[] stripeBlocks = null;
    RaidCodec codec = null;
    StringBuilder sb = new StringBuilder();
    StringBuilder dnSb = new StringBuilder();
    readLock();
    try {
      INodeFile inode = blocksMap.getINode(rbi);
      if (inode == null) {
        return false; // file has been deleted already, nothing to do.
      }
      if (inode.getStorageType() != StorageType.RAID_STORAGE) {
        LOG.error("File for block " + rbi + " is not raidable");
        return false;
      }
      INodeRaidStorage storage = (INodeRaidStorage)inode.getStorage();
      codec = storage.getCodec();
      // Find out related blocks
      stripeBlocks = codec.getBlocksInOneStripe(
          inode.getBlocks(), rbi);
      // Find out datanode locations
      for (int i = 0; i < stripeBlocks.length; i++) {
        ArrayList<DatanodeInfo> liveNodes = 
            new ArrayList<DatanodeInfo>();
        stripeDatanodes.add(liveNodes);
        Collection<DatanodeDescriptor> nodesCorrupt = 
            corruptReplicas.getNodes(stripeBlocks[i]);
        StringBuilder dn = new StringBuilder();
        dn.append("[");
        for (Iterator<DatanodeDescriptor> it =
            blocksMap.nodeIterator(stripeBlocks[i]); 
            it.hasNext();) {
          DatanodeDescriptor node = it.next();
          if (nodesCorrupt != null && nodesCorrupt.contains(node)) {
            // do nothing
          } else {
            liveNodes.add(node);
            dn.append(node);
            dn.append(" ");
          }
        }
        dn.append("]");
        if (i >= codec.numParityBlocks && liveNodes.size() == 0) {
          LOG.error("File for block " + rbi +
              " is not raidable, not all source blocks have replicas");
          return true;
        }
        sb.append(stripeBlocks[i]);
        sb.append(" ");
        dnSb.append(dn.toString());
        dnSb.append(" ");
      }
    } finally {
      readUnlock();
    }
    DatanodeDescriptor scheduleNode = null; 
    int[] toRaidIdxs = new int[codec.numParityBlocks];
    // Randomly pick datanodes first
    for (int i = 0; i < codec.numParityBlocks; i++) {
      toRaidIdxs[i] = i;
      int size = stripeDatanodes.get(i).size();
      stripeDatanodes.get(i).clear();
      if (size < codec.parityReplication) {
        // If not enough replicas, assign targets left
        for (int j = 0; j < codec.parityReplication - size; j++) {
          stripeDatanodes.get(i).add(getRandomDatanode());
        }
        if (scheduleNode == null) {
          scheduleNode = (DatanodeDescriptor)stripeDatanodes.get(i).get(0);
        }
      }
    }
    // Assign raiding task to scheduledNode
    if (scheduleNode == null) {
      // All parity blocks are generated, no need to raid
      return false;
    }
    RaidTask rt = new RaidTask(codec, stripeBlocks, stripeDatanodes, toRaidIdxs);
    writeLock();
    try {
      scheduleNode.addRaidEncodingTask(rt);
      if (NameNode.stateChangeLog.isDebugEnabled()) {
        NameNode.stateChangeLog.debug("Stripe* NameSystem: Blocks " + sb.toString()
            + " Datanodes: " + dnSb.toString()
            + " are added to raidEncodingTasks of" + scheduleNode);
      }
      return false;
    } finally {
      writeUnlock();
    }
  }
663899
public Deferred<ArrayList<Object>> checkNecessaryTablesExist() {
    final ArrayList<Deferred<Object>> checks =
      new ArrayList<Deferred<Object>>(2);
    checks.add(client.ensureTableExists(
        config.getString("tsd.storage.hbase.data_table")));
    checks.add(client.ensureTableExists(
        config.getString("tsd.storage.hbase.uid_table")));
    if (config.enable_tree_processing()) {
      checks.add(client.ensureTableExists(
          config.getString("tsd.storage.hbase.tree_table")));
    }
    if (config.enable_realtime_ts() || config.enable_realtime_uid() ||
        config.enable_tsuid_incrementing()) {
      checks.add(client.ensureTableExists(
          config.getString("tsd.storage.hbase.meta_table")));
    }
    return Deferred.group(checks);
  }
492401
@GuardedBy("workerInfo")
  private void processWorkerAddedBlocks(MasterWorkerInfo workerInfo,
      Map<String, List<Long>> addedBlockIds) {
    for (Map.Entry<String, List<Long>> entry : addedBlockIds.entrySet()) {
      for (long blockId : entry.getValue()) {
        try (LockResource lr = lockBlock(blockId)) {
          Optional<BlockMeta> block = mBlockStore.getBlock(blockId);
          if (block.isPresent()) {
            workerInfo.addBlock(blockId);
            mBlockStore.addLocation(blockId, BlockLocation.newBuilder()
                .setWorkerId(workerInfo.getId())
                .setTier(entry.getKey())
                .build());
            mLostBlocks.remove(blockId);
          } else {
            LOG.warn("Invalid block: {} from worker {}.", blockId,
                workerInfo.getWorkerAddress().getHost());
          }
        }
      }
    }
  }
565400
private boolean removeFromSplitterQueue(Block<S, L> block) {
        ElementReference ref = block.getSplitterQueueReference();
        if (ref == null) {
            return false;
        }

        splitters.remove(ref);
        block.setSplitterQueueReference(null);

        return true;
    }
744666
private boolean switchAndCheckAbort() {
    if (Thread.holdsLock(cache.lock)) {
      return switchCheckAndAbortLocked();
    }
    return cache.executeWithGlobalLock(new Job<Boolean>() {
      @Override
      public Boolean call() {
        return switchCheckAndAbortLocked();
      }
    });
  }
597642
boolean processParityFile(Path parityPath, Decoder decoder, 
      Context context)
  throws IOException, InterruptedException {
    LOG.info("Processing parity file " + parityPath);
    
    Progressable progress = context;
    if (progress == null) {
      progress = RaidUtils.NULL_PROGRESSABLE;
    }
    DistributedFileSystem parityFs = getDFS(parityPath);
    Path srcPath = RaidUtils.sourcePathFromParityPath(parityPath, parityFs);
    if (srcPath == null) {
      LOG.warn("Could not get regular file corresponding to parity file " +  
          parityPath + ", ignoring...");
      return false;
    }
    
    DistributedFileSystem srcFs = getDFS(srcPath);
    FileStatus parityStat = parityFs.getFileStatus(parityPath);
    long blockSize = parityStat.getBlockSize();
    FileStatus srcStat = srcFs.getFileStatus(srcPath);

    // Check timestamp.
    if (srcStat.getModificationTime() != parityStat.getModificationTime()) {
      LOG.warn("Mismatching timestamp for " + srcPath + " and " + parityPath + 
          ", ignoring...");
      return false;
    }

    String uriPath = parityPath.toUri().getPath();
    int numBlocksReconstructed = 0;
    List<LocatedBlockWithMetaInfo> lostBlocks = 
      lostBlocksInFile(parityFs, uriPath, parityStat);
    if (lostBlocks.size() == 0) {
      LOG.warn("Couldn't find any lost blocks in parity file " + parityPath + 
          ", ignoring...");
      return false;
    }
    List<Block> blocksLostChecksum = new ArrayList<Block>();
    List<Block> blocksLostStripe = new ArrayList<Block>();
    
    for (LocatedBlockWithMetaInfo lb: lostBlocks) {
      Block lostBlock = lb.getBlock();
      long lostBlockOffset = lb.getStartOffset();
      LOG.info("Found lost block " + lostBlock +
          ", offset " + lostBlockOffset);
      
      Long oldCRC = decoder.retrieveChecksum(lostBlock,
          parityPath, lostBlockOffset, parityFs, context);
      if (abortReconstruction(oldCRC, decoder)) {
        blocksLostChecksum.add(lostBlock);
        continue;
      }
      StripeInfo si = decoder.retrieveStripe(lostBlock,
          srcPath, lostBlockOffset, srcFs, context, false);
      if (si == null && decoder.stripeStore != null) {
        blocksLostStripe.add(lostBlock);
        continue;
      }
      
      File localBlockFile =
        File.createTempFile(lostBlock.getBlockName(), ".tmp");
      localBlockFile.deleteOnExit();

      try {
        CRC32 crc = decoder.recoverParityBlockToFile(srcFs, srcStat, parityFs,
            parityPath, blockSize, lostBlockOffset, localBlockFile, si,
            context);
        LOG.info("Recovered crc: " + ((crc == null)?null: crc.getValue()) +
            " expected crc:" + oldCRC);
        if (crc != null && oldCRC != null &&
            crc.getValue() != oldCRC) {
          // checksum doesn't match, it's dangerous to send it
          IOException ioe = new IOException("Block " + lostBlock.toString()
              + " new checksum " + crc.getValue()
              + " doesn't match the old one " + oldCRC);
          LogUtils.logRaidReconstructionMetrics(LOGRESULTS.FAILURE, 0,
              decoder.codec, parityPath, lostBlockOffset,
              LOGTYPES.OFFLINE_RECONSTRUCTION_CHECKSUM_VERIFICATION,
              parityFs, ioe, context);
          throw ioe;
        }
        // Now that we have recovered the parity file block locally, send it.
        computeMetadataAndSendReconstructedBlock(
            localBlockFile, 
            lostBlock, blockSize, lb.getLocations(),
            lb.getDataProtocolVersion(), lb.getNamespaceID(),
            progress);

        numBlocksReconstructed++;
      } finally {
        localBlockFile.delete();
      }
      progress.progress();
    }
    
    LOG.info("Reconstructed " + numBlocksReconstructed + " blocks in " + parityPath);
    checkLostBlocks(blocksLostChecksum, blocksLostStripe, parityPath, decoder.codec);
    return true;
  }
424609
private HashMap<DatanodeDescriptor, Integer> directoryDataNodeUsage(INodeDirectory dir,
      int threshold) {
    HashMap<DatanodeDescriptor, Integer> dataNodeUsage = new HashMap<DatanodeDescriptor, Integer>();

    List<INode> children;
    nameSystem.readLock();
    try {
      if (dir.getChildrenRaw() == null) {
        return dataNodeUsage;
      }
      children = new ArrayList<INode>(dir.getChildrenRaw());
      Collections.shuffle(children);

      for (INode node : children) {
        if (!(node instanceof INodeFile)) { // The condition is always false.
          continue;
        }

        INodeFile file = (INodeFile) node;
        BlockInfo[] blocks = file.getBlocks();

        for (BlockInfo block : blocks) {
          if (threshold == 0) {
            return dataNodeUsage;
          }

          int replication = block.numNodes();
          for (int i = 0; i < replication; i++) {
            DatanodeDescriptor datanode = block.getDatanode(i);
            Integer currentUsage = dataNodeUsage.get(datanode);
            dataNodeUsage.put(datanode, currentUsage == null ? 1 : currentUsage + 1);
          }
          threshold--;
        }
      }
    } finally {
      nameSystem.readUnlock();
    }
    return dataNodeUsage;
  }
1223057
private Set<AllocationID> computeAllPriorAllocationIdsIfRequiredByScheduling() {
		// This is a temporary optimization to avoid computing all previous allocations if not required
		// This can go away when we progress with the implementation of the Scheduler.
		if (slotProvider instanceof Scheduler && ((Scheduler) slotProvider).requiresPreviousExecutionGraphAllocations()) {
			return computeAllPriorAllocationIds();
		} else {
			return Collections.emptySet();
		}
	}
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");
      }
    }
  }
824190
boolean add(DatanodeDescriptor node) {
    hostmapLock.writeLock().lock();
    try {
      if (node==null || contains(node)) {
        return false;
      }
      
      String host = node.getHost();
      DatanodeDescriptor[] nodes = map.get(host);
      DatanodeDescriptor[] newNodes;
      if (nodes==null) {
        newNodes = new DatanodeDescriptor[1];
        newNodes[0]=node;
      } else { // rare case: more than one datanode on the host
        newNodes = new DatanodeDescriptor[nodes.length+1];
        System.arraycopy(nodes, 0, newNodes, 0, nodes.length);
        newNodes[nodes.length] = node;
      }
      map.put(host, newNodes);
      return true;
    } finally {
      hostmapLock.writeLock().unlock();
    }
  }
473190
private List<TaskLockPosse> findLockPossesForTask(final Task task)
  {
    giant.lock();

    try {
      // Scan through all locks for this datasource
      final NavigableMap<DateTime, SortedMap<Interval, List<TaskLockPosse>>> dsRunning = running.get(task.getDataSource());
      if (dsRunning == null) {
        return ImmutableList.of();
      } else {
        return dsRunning.values().stream()
                        .flatMap(map -> map.values().stream())
                        .flatMap(Collection::stream)
                        .filter(taskLockPosse -> taskLockPosse.containsTask(task))
                        .collect(Collectors.toList());
      }
    }
    finally {
      giant.unlock();
    }
  }
1152599
int computeReplicationWorkForBlocks(List<List<BlockInfo>> blocksToReplicate) {
    int requiredReplication, numEffectiveReplicas, priority;
    List<DatanodeDescriptor> containingNodes;
    DatanodeDescriptor srcNode;
    INodeFile fileINode = null;

    int scheduledWork = 0;
    List<ReplicationWork> work = new LinkedList<ReplicationWork>();

    writeLock();
    try {
      synchronized (neededReplications) {
        for (priority = 0; priority < blocksToReplicate.size(); priority++) {
          for (BlockInfo block : blocksToReplicate.get(priority)) {
            // block should belong to a file
            BlockInfo storedBlock = blocksMap.getBlockInfo(block);
            fileINode = (storedBlock == null) ? null : storedBlock.getINode();
            // abandoned block not belong to a file
            if (fileINode == null) {
              neededReplications.remove(block, priority); // remove from neededReplications
              replIndex[priority]--;
              continue;
            }
            requiredReplication = fileINode.getBlockReplication(storedBlock);

            // get a source data-node
            containingNodes = new ArrayList<DatanodeDescriptor>();
            NumberReplicas numReplicas = new NumberReplicas();
            srcNode = chooseSourceDatanode(block, containingNodes, numReplicas);
            if (srcNode == null) // block can not be replicated from any node
            {
              continue;
            }

            // do not schedule more if enough replicas is already pending
            numEffectiveReplicas = numReplicas.liveReplicas() +
              pendingReplications.getNumReplicas(block);
            if (numEffectiveReplicas >= requiredReplication) {
              neededReplications.remove(block, priority); // remove from neededReplications
              replIndex[priority]--;
              continue;
            }
            work.add(new ReplicationWork(block, fileINode, requiredReplication
                - numEffectiveReplicas, srcNode, containingNodes, priority));
          }
        }
      }
    } finally {
      writeUnlock();
    }

    // choose replication targets: NOT HODING THE GLOBAL LOCK
    for(ReplicationWork rw : work){
      DatanodeDescriptor targets[] = chooseTarget(rw);
      rw.targets = targets;
    }

    writeLock();
    try {
      for(ReplicationWork rw : work){
        DatanodeDescriptor[] targets = rw.targets;
        if(targets == null || targets.length == 0){
          rw.targets = null;
          continue;
        }
        if (targets.length == 1 && 
            !clusterMap.isOnSameRack(rw.srcNode, targets[0])) {
          // in case src & target are not on the same rack
          // see if there is an alternate valid source in the same rack 
          // as the target that we can use instead.
          for (DatanodeDescriptor node : rw.containingNodes) {
            if (node != rw.srcNode && 
                clusterMap.isOnSameRack(node, targets[0]) &&
                isGoodReplica(node, rw.block)) {
              rw.srcNode = node;
              break;
            }
          }
        }
        synchronized (neededReplications) {
          BlockInfo block = rw.block;
          priority = rw.priority;
          // Recheck since global lock was released
          // block should belong to a file
          BlockInfo storedBlock = blocksMap.getBlockInfo(block);
          fileINode = (storedBlock == null) ? null : storedBlock.getINode();
          // abandoned block not belong to a file
          if (fileINode == null ) {
            neededReplications.remove(block, priority); // remove from neededReplications
            rw.targets = null;
            replIndex[priority]--;
            continue;
          }
          requiredReplication = fileINode.getBlockReplication(storedBlock);

          // do not schedule more if enough replicas is already pending
          NumberReplicas numReplicas = countNodes(block);
          numEffectiveReplicas = numReplicas.liveReplicas() +
            pendingReplications.getNumReplicas(block);
          if (numEffectiveReplicas >= requiredReplication) {
            neededReplications.remove(block, priority); // remove from neededReplications
            replIndex[priority]--;
            rw.targets = null;
            continue;
          }

          // Add block to the to be replicated list
          rw.srcNode.addBlockToBeReplicated(block, targets);
          
          scheduledWork++;

          for (DatanodeDescriptor dn : targets) {
            dn.incBlocksScheduled();
          }

          // Move the block-replication into a "pending" state.
          // The reason we use 'pending' is so we can retry
          // replications that fail after an appropriate amount of time.
          pendingReplications.add(block, targets.length);
          if (NameNode.stateChangeLog.isDebugEnabled()) {
            NameNode.stateChangeLog.debug(
              "BLOCK* block " + block
                + " is moved from neededReplications to pendingReplications");
          }

          // remove from neededReplications
          if (numEffectiveReplicas + targets.length >= requiredReplication) {
            neededReplications.remove(block, priority); // remove from neededReplications
            replIndex[priority]--;
          }
        }
      }
    } finally {
      writeUnlock();
    }

    // update metrics
    updateReplicationMetrics(work);
    
    // print debug information
    if(NameNode.stateChangeLog.isInfoEnabled()){
      // log which blocks have been scheduled for replication
      for(ReplicationWork rw : work){
        // report scheduled blocks
        DatanodeDescriptor[] targets = rw.targets;
        if (targets != null && targets.length != 0) {
          StringBuffer targetList = new StringBuffer("datanode(s)");
          for (int k = 0; k < targets.length; k++) {
            targetList.append(' ');
            targetList.append(targets[k].getName());
          }
          NameNode.stateChangeLog.info(
            "BLOCK* ask "
              + rw.srcNode.getName() + " to replicate "
              + rw.block + " to " + targetList);
        }
      }
    }

    // log once per call
    if (NameNode.stateChangeLog.isDebugEnabled()) {
      NameNode.stateChangeLog.debug("BLOCK* neededReplications = "
          + neededReplications.size() + " pendingReplications = "
          + pendingReplications.size());
    }
    return scheduledWork;
  }
574926
@Override
  public LocatedBlock syncBlock(
      Block block,
      List<BlockRecord> syncList,
    boolean closeFile, List<InterDatanodeProtocol> datanodeProxies,
    long deadline
  ) 
    throws IOException {
    if (LOG.isDebugEnabled()) {
      LOG.debug("block=" + block + ", (length=" + block.getNumBytes()
          + "), syncList=" + syncList + ", closeFile=" + closeFile);
    }

    //syncList.isEmpty() that all datanodes do not have the block
    //so the block can be deleted.
    if (syncList.isEmpty()) {
      bsp.commitBlockSynchronization(block, 0, 0, closeFile, true,
          DatanodeID.EMPTY_ARRAY);
      return null;
    }

    List<DatanodeID> successList = new ArrayList<DatanodeID>();

    throwIfAfterTime(deadline);
    long generationstamp = -1;
    try {
      generationstamp = bsp.nextGenerationStamp(block, closeFile);
    } catch (RemoteException e) {
      if (e.unwrapRemoteException() instanceof BlockAlreadyCommittedException) {
        throw new BlockAlreadyCommittedException(e);
      } else {
        throw e;
      }
    }
    Block newblock = new Block(block.getBlockId(), block.getNumBytes(), generationstamp);

    for(BlockRecord r : syncList) {
      try {
        throwIfAfterTime(deadline);
        LOG.info("Updating block " + r + " to " + newblock);
        r.datanode.updateBlock(namespaceId, r.info.getBlock(), newblock, closeFile);
        successList.add(r.id);
      } catch (BlockRecoveryTimeoutException e) {
        throw e;
      } catch (IOException e) {
        InterDatanodeProtocol.LOG.warn("Failed to updateBlock (newblock="
            + newblock + ", datanode=" + r.id + ")", e);
      }
    }

    LOG.info("Updated blocks on syncList for block " + block + " to " + newblock);

    DataNode.stopAllProxies(datanodeProxies);

    if (!successList.isEmpty()) {
      DatanodeID[] nlist = successList.toArray(new DatanodeID[successList.size()]);

      throwIfAfterTime(deadline);
      bsp.commitBlockSynchronization(block,
          newblock.getGenerationStamp(), newblock.getNumBytes(), closeFile, false,
          nlist);
      DatanodeInfo[] info = new DatanodeInfo[nlist.length];
      for (int i = 0; i < nlist.length; i++) {
        info[i] = new DatanodeInfo(nlist[i]);
      }
      return new LocatedBlock(newblock, info); // success
    }

    //failed
    StringBuilder b = new StringBuilder();
    for(BlockRecord r : syncList) {
      b.append("\n  " + r.id);
    }
    throw new IOException("Cannot recover " + block + ", none of these "
        + syncList.size() + " datanodes success {" + b + "\n}");
  }
1140490
protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
        Runnable task = pollTaskFrom(taskQueue);
        if (task == null) {
            return false;
        }
        for (;;) {
            safeExecute(task);
            task = pollTaskFrom(taskQueue);
            if (task == null) {
                return true;
            }
        }
    }
73223
private Block[] allocateParityBlocks(int numParityBlocks) throws IOException {
    Block[] blocks = new Block[numParityBlocks];
    for (int i = 0; i < numParityBlocks; i++) {
      Block b = new Block(generateBlockId(), 0, 0);
      while (isValidBlock(b)) {
        b.setBlockId(generateBlockId());
      }
      b.setGenerationStamp(getGenerationStamp());
      blocks[i] = b;
    }
    return blocks;
  }
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;
  }
23312
BlockInfo addINodeForLoading(Block b, INodeFile iNode) {
    // allocate new block when loading the image
    // for hardlinked files, we need to check if the blocks are already there
    BlockInfo info = checkBlockInfo(b, iNode.getReplication(),
        iNode.isHardlinkFile());
    info.inode = iNode;
    return info;
  }