Hyperledger Fabric peer block 的交付流程详解

// /var/hyperledger/production/ledgersData 下有这些东西, 这里面的工作跟此目录有关
// chains/index goleveldb: 保存了所有通道的最新状态信息
// fileLock goleveldb: 用于锁程序的,文章最末尾有介绍
// historyLeveldb goleveldb: 保存历史交易的
// ledgerProvider goleveldb: 保存的是 chain ids, 也就是通道id
// pvtdataStore goleveldb: 存储私有数据库
// stateLeveldb goleveldb: 世界状态数据库, 可以替换为 couchdb

// 这两个我不确定没有细追
// configHistory 看名字应该是保存了 config block 相关的东西.
// bookkeeper 不知道.
ledgermgmt.Initialize(&ledgermgmt.Initializer{
    CustomTxProcessors:            ConfigTxProcessors,
    PlatformRegistry:              pr,
    DeployedChaincodeInfoProvider: deployedCCInfoProvider,
    MembershipInfoProvider:        membershipProvider,
    MetricsProvider:               metricsProvider,
})

// 就是通过 /var/hyperledger/production/ledgersData/ledgerProvider 获取的 id 列表
ledgerIds, err := ledgermgmt.GetLedgerIDs()
...
for _, cid := range ledgerIds {

    // 打开对应通道的账本, 例如打开 block 的最新文件, 加载最新的账本信息等
    if ledger, err = ledgermgmt.OpenLedger(cid); err != nil {
        ...
    }

    // 获取最新的交易块, 通过交易块获取最新的配置块的 number, 在通过 number 查询到最新的配置块
    // 有兴趣可以往下追一追看看
    if cb, err = getCurrConfigBlockFromLedger(ledger); err != nil {
        ...
    }

    // 创建通道实例, 传了 通道id, 账本实例, 最新的配置块等.
    if err = createChain(cid, ledger, cb, ccp, sccp, pm); err != nil {
        ...
    }

    InitChain(cid)
}

func createChain(
    cid string,
    ledger ledger.PeerLedger,
    cb *common.Block,
    ccp ccprovider.ChaincodeProvider,
    sccp sysccprovider.SystemChaincodeProvider,
    pm txvalidator.PluginMapper,
) error {
    // 初始化 gossip 服务用到的配置, vscc, 等相关资源初始化, 有兴趣的可以看看这个函数,
    // 这里就不做介绍了
    ...
    // 初始化 gossip service 方法
    service.GetGossipService().InitializeChannel(bundle.ConfigtxValidator().ChainID(), oac, service.Support{
        Validator:            validator,
        Committer:            c,
        Store:                store,
        Cs:                   simpleCollectionStore,
        IdDeserializeFactory: csStoreSupport,
        CapabilityProvider:   cp,
    })
    ...
    return nil
}

func (g *gossipServiceImpl) InitializeChannel(chainID string, oac OrdererAddressConfig, support Support) {
    ...
    // Delivery service might be nil only if it was not able to get connected
    // to the ordering service
    if g.deliveryService[chainID] != nil {
        // 根据 core.yaml 或者 环境变量设置的两个个值在下方有不同的启动方式
        leaderElection := viper.GetBool("peer.gossip.useLeaderElection")
        isStaticOrgLeader := viper.GetBool("peer.gossip.orgLeader")

        ...

        if leaderElection {
            logger.Debug("Delivery uses dynamic leader election mechanism, channel", chainID)
            g.leaderElection[chainID] = g.newLeaderElectionComponent(chainID, g.onStatusChangeFactory(chainID,
                support.Committer), g.metrics.ElectionMetrics)
        } else if isStaticOrgLeader {
            logger.Debug("This peer is configured to connect to ordering service for blocks delivery, channel", chainID)
            g.deliveryService[chainID].StartDeliverForChannel(chainID, support.Committer, func() {})
        } else {
            logger.Debug("This peer is not configured to connect to ordering service for blocks delivery, channel", chainID)
        }
    } else {
        logger.Warning("Delivery client is down won't be able to pull blocks for chain", chainID)
    }

}

func (b *blocksProviderImpl) DeliverBlocks() {
    ...
    for !b.isDone() {
        ...
        switch t := msg.Type.(type) {
        case *orderer.DeliverResponse_Status:
            ...
        case *orderer.DeliverResponse_Block:
            ...
            if err := b.mcs.VerifyBlock(gossipcommon.ChainID(b.chainID), blockNum, marshaledBlock); err != nil {
                logger.Errorf("[%s] Error verifying block with sequence number %d, due to %s", b.chainID, blockNum, err)
                continue
            }

            ...
            if err := b.gossip.AddPayload(b.chainID, payload); err != nil {
                logger.Warningf("Block [%d] received from ordering service wasn't added to payload buffer: %v", blockNum, err)
            }
            ...
        default:
            logger.Warningf("[%s] Received unknown: %v", b.chainID, t)
            return
        }
    }
}

func (s *GossipStateProviderImpl) commitBlock(block *common.Block, pvtData util.PvtDataCollections) error {
    ...
    // 将 block 和 pvtdata 交给 ledger 进行存储
    if err := s.ledger.StoreBlock(block, pvtData); err != nil {
        logger.Errorf("Got error while committing(%+v)", errors.WithStack(err))
        return err
    }
    ...
}

func (c *coordinator) StoreBlock(block *common.Block, privateDataSets util.PvtDataCollections) error {
    ...

    validationStart := time.Now()
    err := c.Validator.Validate(block)
    c.reportValidationDuration(time.Since(validationStart))
    if err != nil {
        logger.Errorf("Validation failed: %+v", err)
        return err
    }

    ...

    // commit block and private data
    commitStart := time.Now()
    err = c.CommitWithPvtData(blockAndPvtData, &ledger.CommitOptions{})
    c.reportCommitDuration(time.Since(commitStart))
    if err != nil {
        return errors.Wrap(err, "commit failed")
    }

    ...

    return nil
}