Hadoop3.2.1 【 HDFS 】源码分析 : BlockScanner & VolumeScanner 实现

曾经终败给现在 2022-12-23 12:25 214阅读 0赞

# 一. 前言 #

每个Datanode都会初始化一个数据块扫描器周期性地验证Datanode上存储的所有数据块的正确性， 并把发现的损坏数据块报告给Namenode。 BlockScanner类就是Datanode上数据块扫描器的实现。BlockScanner里面有多个VolumeScanner, 每个BlockScanner对应一个volume , 每个BlockScanner都是单独的线程负责扫描block任务…

# 二.BlockScanner #

BlockScanner中有一个属性scanners 用于存放BlockScanner所要监控的VolumeScanner对象. 然后VolumeScanner会启动线程来监控block状态.addVolumeScanner ,removeVolumeScanner,removeAllVolumeScanners,enableBlockPoolId,disableBlockPoolId等等其实都是对集合scanners的CRUD. 真正处理任务的是VolumeScanner.

/**
       * The DataNode that this scanner is associated with.
       */
      private final DataNode datanode;
    
      /**
       * Maps Storage IDs to VolumeScanner objects.
       */
      private final TreeMap<String, VolumeScanner> scanners = new TreeMap<String, VolumeScanner>();
    
      /**
       * The scanner configuration.
       */
      private Conf conf;

# 三.VolumeScanner #

VolumeScanner用于扫描单个volume , 每一个VolumeScanner 都启动自己的线程.  
它们都由DataNode的BlockScanner管理。

这里维护了一个队列blockIters 用于标识待处理的块, 由markSuspectBlock(ExtendedBlock block) 方法为添加的入口.  
run()方法执行的时候,会处理这个队列里的内容

/**
       * The block iterators associated with this VolumeScanner.<p/>
       *
       * Each block pool has its own BlockIterator.
       */
      private final List<BlockIterator> blockIters =   new LinkedList<BlockIterator>();

最重要的还是blockIters这个属性. 检查的时候会根据这个做遍历,然后循环验证每个block …

我直接把run()方法贴出来,就不细说了.因为就是处理blockIters 或者blockIters 中的block信息, 响应信息交给ScanResultHandler 进行处理.

@Override
      public void run() { 
        // 记录scanner的启动事件
        // Record the minute on which the scanner started.
        this.startMinute =  TimeUnit.MINUTES.convert(Time.monotonicNow(), TimeUnit.MILLISECONDS);
        this.curMinute = startMinute;
    
    
        try { 
    
    
          LOG.trace("{}: thread starting.", this);
          resultHandler.setup(this);
          try { 
            long timeout = 0;
            while (true) { 
              ExtendedBlock suspectBlock = null;
              // Take the lock to check if we should stop, and access the
              // suspect block list.
              synchronized (this) { 
                if (stopping) { 
                  break;
                }
                if (timeout > 0) { 
                  LOG.debug("{}: wait for {} milliseconds", this, timeout);
                  wait(timeout);
                  if (stopping) { 
                    break;
                  }
                }
                //处理  有问题的 block 数据
                suspectBlock = popNextSuspectBlock();
              }
    
              // 处理任务!!!!
              timeout = runLoop(suspectBlock);
            }
          } catch (InterruptedException e) { 
            // We are exiting because of an InterruptedException,
            // probably sent by VolumeScanner#shutdown.
            LOG.trace("{} exiting because of InterruptedException.", this);
          } catch (Throwable e) { 
            LOG.error("{} exiting because of exception ", this, e);
          }
          LOG.info("{} exiting.", this);
          // Save the current position of all block iterators and close them.
          for (BlockIterator iter : blockIters) { 
    
            saveBlockIterator(iter);
            IOUtils.cleanup(null, iter);
          }
    
    
        } finally { 
          // When the VolumeScanner exits, release the reference we were holding
          // on the volume.  This will allow the volume to be removed later.
          IOUtils.cleanup(null, ref);
        }
      }

这里核心的还是timeout = runLoop(suspectBlock); 方法

/**
       * Run an iteration of the VolumeScanner loop.
       *
       * @param suspectBlock   A suspect block which we should scan, or null to
       *                       scan the next regularly scheduled block.
       *
       * @return     The number of milliseconds to delay before running the loop
       *               again, or 0 to re-run the loop immediately.
       */
      private long runLoop(ExtendedBlock suspectBlock) { 
        long bytesScanned = -1;
        boolean scanError = false;
        ExtendedBlock block = null;
        try { 
          long monotonicMs = Time.monotonicNow();
          expireOldScannedBytesRecords(monotonicMs);
    
          if (!calculateShouldScan(volume.getStorageID(), conf.targetBytesPerSec,
              scannedBytesSum, startMinute, curMinute)) { 
            // If neededBytesPerSec is too low, then wait few seconds for some old
            // scannedBytes records to expire.
            return 30000L;
          }
    
          // Find a usable block pool to scan.
          if (suspectBlock != null) { 
            block = suspectBlock;
          } else { 
            if ((curBlockIter == null) || curBlockIter.atEnd()) { 
              long timeout = findNextUsableBlockIter();
              if (timeout > 0) { 
                LOG.trace("{}: no block pools are ready to scan yet. Waiting " +
                    "{} ms.", this, timeout);
                synchronized (stats) { 
                  stats.nextBlockPoolScanStartMs = Time.monotonicNow() + timeout;
                }
                return timeout;
              }
              synchronized (stats) { 
                stats.scansSinceRestart++;
                stats.blocksScannedInCurrentPeriod = 0;
                stats.nextBlockPoolScanStartMs = -1;
              }
              return 0L;
            }
            try { 
              block = curBlockIter.nextBlock();
            } catch (IOException e) { 
              // There was an error listing the next block in the volume.  This is a
              // serious issue.
              LOG.warn("{}: nextBlock error on {}", this, curBlockIter);
              // On the next loop iteration, curBlockIter#eof will be set to true, and
              // we will pick a different block iterator.
              return 0L;
            }
            if (block == null) { 
              // The BlockIterator is at EOF.
              LOG.info("{}: finished scanning block pool {}",
                  this, curBlockIter.getBlockPoolId());
              saveBlockIterator(curBlockIter);
              return 0;
            }
          }
          if (curBlockIter != null) { 
            long saveDelta = monotonicMs - curBlockIter.getLastSavedMs();
            if (saveDelta >= conf.cursorSaveMs) { 
              LOG.debug("{}: saving block iterator {} after {} ms.",
                  this, curBlockIter, saveDelta);
              saveBlockIterator(curBlockIter);
            }
          }
          bytesScanned = scanBlock(block, conf.targetBytesPerSec);
          if (bytesScanned >= 0) { 
            scannedBytesSum += bytesScanned;
            scannedBytes[(int)(curMinute % MINUTES_PER_HOUR)] += bytesScanned;
          } else { 
            scanError = true;
          }
          return 0L;
        } finally { 
          synchronized (stats) { 
            stats.bytesScannedInPastHour = scannedBytesSum;
            if (bytesScanned > 0) { 
              stats.blocksScannedInCurrentPeriod++;
              stats.blocksScannedSinceRestart++;
            }
            if (scanError) { 
              stats.scanErrorsSinceRestart++;
            }
            if (block != null) { 
              stats.lastBlockScanned = block;
            }
            if (curBlockIter == null) { 
              stats.eof = true;
              stats.blockPoolPeriodEndsMs = -1;
            } else { 
              stats.eof = curBlockIter.atEnd();
              stats.blockPoolPeriodEndsMs =
                  curBlockIter.getIterStartMs() + conf.scanPeriodMs;
            }
          }
        }
      }

扫描block数据 :

bytesScanned = scanBlock(block, conf.targetBytesPerSec);

/**
       * Scan a block.
       *
       * @param cblock               The block to scan.
       * @param bytesPerSec          The bytes per second to scan at.
       *
       * @return                     The length of the block that was scanned, or
       *                               -1 if the block could not be scanned.
       */
      private long scanBlock(ExtendedBlock cblock, long bytesPerSec) { 
        // 'cblock' has a valid blockId and block pool id, but we don't yet know the
        // genstamp the block is supposed to have.  Ask the FsDatasetImpl for this
        // information.
        ExtendedBlock block = null;
        try { 
          
          Block b = volume.getDataset().getStoredBlock(
              cblock.getBlockPoolId(), cblock.getBlockId());
          if (b == null) { 
            LOG.info("Replica {} was not found in the VolumeMap for volume {}",
                cblock, volume);
          } else { 
            block = new ExtendedBlock(cblock.getBlockPoolId(), b);
          }
        } catch (FileNotFoundException e) { 
          LOG.info("FileNotFoundException while finding block {} on volume {}",
              cblock, volume);
        } catch (IOException e) { 
          LOG.warn("I/O error while finding block {} on volume {}",
                cblock, volume);
        }
        if (block == null) { 
          return -1; // block not found.
        }
        LOG.debug("start scanning block {}", block);
        BlockSender blockSender = null;
        try { 
    
          //构造BlockSender对象， 将数据块发送到空的输出流中
          blockSender = new BlockSender(block, 0, -1,
              false, true, true, datanode, null,
              CachingStrategy.newDropBehind());
    
          // 限流
          throttler.setBandwidth(bytesPerSec);
          long bytesRead = blockSender.sendBlock(nullStream, null, throttler);
    
          /// 数据块验证正确， 调用updateScanStatus()方法更新数据块扫描状态
          resultHandler.handle(block, null);
          metrics.incrBlocksVerified();
          return bytesRead;
        } catch (IOException e) { 
          //如果检查数据块失败....
          resultHandler.handle(block, e);
        } finally { 
          IOUtils.cleanup(null, blockSender);
        }
        metrics.incrBlockVerificationFailures();
        return -1;
      }

扫描block的时候,如果能正常发送的话,那block属于正常, 如果不能的话, resultHandler.handle(block, e); 进行处理:

public void handle(ExtendedBlock block, IOException e) { 
          FsVolumeSpi volume = scanner.volume;
          if (e == null) { 
            LOG.trace("Successfully scanned {} on {}", block, volume);
            return;
          }
          // If the block does not exist anymore, then it's not an error.
          if (!volume.getDataset().contains(block)) { 
            LOG.debug("Volume {}: block {} is no longer in the dataset.",
                volume, block);
            return;
          }
          // If the block exists, the exception may due to a race with write:
          // The BlockSender got an old block path in rbw. BlockReceiver removed
          // the rbw block from rbw to finalized but BlockSender tried to open the
          // file before BlockReceiver updated the VolumeMap. The state of the
          // block can be changed again now, so ignore this error here. If there
          // is a block really deleted by mistake, DirectoryScan should catch it.
          if (e instanceof FileNotFoundException ) { 
            LOG.info("Volume {}: verification failed for {} because of " +
                    "FileNotFoundException. This may be due to a race with write.",
                volume, block);
            return;
          }
          LOG.warn("Reporting bad {} on {}", block, volume);
          scanner.datanode.handleBadBlock(block, e, true);
        }