兄弟连区块链教程以太坊源码分析chain-indexer区块链索引二

Start方法。 这个方法在eth协议启动的时候被调用,这个方法接收两个参数,一个是当前的区块头,一个是事件订阅器,通过这个订阅器可以获取区块链的改变信息。
  eth.bloomIndexer.Start(eth.blockchain.CurrentHeader(), eth.blockchain.SubscribeChainEvent)
  // Start creates a goroutine to feed chain head events into the indexer for
  // cascading background processing. Children do not need to be started, they
  // are notified about new events by their parents.
  // 子链不需要被启动。 以为他们的父节点会通知他们。
  func (c *ChainIndexer) Start(currentHeader *types.Header, chainEventer func(ch chan<- ChainEvent) event.Subscription) {
    go c.eventLoop(currentHeader, chainEventer)
  }
  // eventLoop is a secondary - optional - event loop of the indexer which is only
  // started for the outermost indexer to push chain head events into a processing
  // queue.
  // eventLoop 循环只会在最外面的索引节点被调用。 所有的Child indexer不会被启动这个方法。
  func (c *ChainIndexer) eventLoop(currentHeader *types.Header, chainEventer func(ch chan<- ChainEvent) event.Subscription) {
    // Mark the chain indexer as active, requiring an additional teardown
    atomic.StoreUint32(&c.active, 1)
  
    events := make(chan ChainEvent, 10)
    sub := chainEventer(events)
    defer sub.Unsubscribe()
  
    // Fire the initial new head event to start any outstanding processing
    // 设置我们的其实的区块高度,用来触发之前未完成的操作。
    c.newHead(currentHeader.Number.Uint64(), false)
  
    var (
      prevHeader = currentHeader
      prevHash = currentHeader.Hash()
    )
    for {
      select {
      case errc := <-c.quit:
        // Chain indexer terminating, report no failure and abort
        errc <- nil
        return
  
      case ev, ok := <-events:
        // Received a new event, ensure it's not nil (closing) and update
        if !ok {
          errc := <-c.quit
          errc <- nil
          return
        }
        header := ev.Block.Header()
        if header.ParentHash != prevHash { //如果出现了分叉,那么我们首先
          //找到公共祖先, 从公共祖先之后的索引需要重建。
          c.newHead(FindCommonAncestor(c.chainDb, prevHeader, header).Number.Uint64(), true)
        }
        // 设置新的head
        c.newHead(header.Number.Uint64(), false)
  
        prevHeader, prevHash = header, header.Hash()
      }
    }
  }
newHead方法,通知indexer新的区块链头,或者是需要重建索引,newHead方法会触发
  
  // newHead notifies the indexer about new chain heads and/or reorgs.
  func (c *ChainIndexer) newHead(head uint64, reorg bool) {
    c.lock.Lock()
    defer c.lock.Unlock()
  
    // If a reorg happened, invalidate all sections until that point
    if reorg { // 需要重建索引 从head开始的所有section都需要重建。
      // Revert the known section number to the reorg point
      changed := head / c.sectionSize
      if changed < c.knownSections {
        c.knownSections = changed
      }
      // Revert the stored sections from the database to the reorg point
      // 将存储的部分从数据库恢复到索引重建点
      if changed < c.storedSections {
        c.setValidSections(changed)
      }
      // Update the new head number to te finalized section end and notify children
      // 生成新的head 并通知所有的子索引
      head = changed * c.sectionSize
  
      if head < c.cascadedHead {
        c.cascadedHead = head
        for _, child := range c.children {
          child.newHead(c.cascadedHead, true)
        }
      }
      return
    }
    // No reorg, calculate the number of newly known sections and update if high enough
    var sections uint64
    if head >= c.confirmsReq {
      sections = (head + 1 - c.confirmsReq) / c.sectionSize
      if sections > c.knownSections {
        c.knownSections = sections
  
        select {
        case c.update <- struct{}{}:
        default:
        }
      }
    }
  }
父子索引数据的关系
父Indexer负载事件的监听然后把结果通过newHead传递给子Indexer的updateLoop来处理。
setValidSections方法,写入当前已经存储的sections的数量。 如果传入的值小于已经存储的数量,那么从数据库里面删除对应的section
  // setValidSections writes the number of valid sections to the index database
  func (c *ChainIndexer) setValidSections(sections uint64) {
    // Set the current number of valid sections in the database
    var data [8]byte
    binary.BigEndian.PutUint64(data[:], sections)
    c.indexDb.Put([]byte("count"), data[:])
  
    // Remove any reorged sections, caching the valids in the mean time
    for c.storedSections > sections {
      c.storedSections--
      c.removeSectionHead(c.storedSections)
    }
    c.storedSections = sections // needed if new > old
  }
processSection
  
  // processSection processes an entire section by calling backend functions while
  // ensuring the continuity of the passed headers. Since the chain mutex is not
  // held while processing, the continuity can be broken by a long reorg, in which
  // case the function returns with an error.
  //processSection通过调用后端函数来处理整个部分,同时确保传递的头文件的连续性。 由于链接互斥锁在处理过程中没有保持,连续性可能会被重新打断,在这种情况下,函数返回一个错误。
  func (c *ChainIndexer) processSection(section uint64, lastHead common.Hash) (common.Hash, error) {
    c.log.Trace("Processing new chain section", "section", section)
  
    // Reset and partial processing
    c.backend.Reset(section)
  
    for number := section * c.sectionSize; number < (section+1)*c.sectionSize; number++ {
      hash := GetCanonicalHash(c.chainDb, number)
      if hash == (common.Hash{}) {
        return common.Hash{}, fmt.Errorf("canonical block #%d unknown", number)
      }
      header := GetHeader(c.chainDb, hash, number)
      if header == nil {
        return common.Hash{}, fmt.Errorf("block #%d [%x...] not found", number, hash[:4])
      } else if header.ParentHash != lastHead {
        return common.Hash{}, fmt.Errorf("chain reorged during section processing")
      }
      c.backend.Process(header)
      lastHead = header.Hash()
    }
    if err := c.backend.Commit(); err != nil {
      c.log.Error("Section commit failed", "error", err)
      return common.Hash{}, err
    }
    return lastHead, nil
  }

本文来自:Segmentfault

感谢作者:兄弟连区块链培训

查看原文:兄弟连区块链教程以太坊源码分析chain-indexer区块链索引二

• 编辑
• 预览