fabric1.0學習筆記（2）

fabric1.0學習筆記（1）
fabric1.0學習筆記（2）

一、fabric1.0目錄結構（只列出了主要的檔案夾）

• bccsp 密碼學相關：加密簽名及證書等，將相關函式抽象成了一組介面
• bddtests 一種新的軟體開發模式：行為驅動開發 需求->開發
• common 公共庫：包括錯誤處理、日志處理、賬本存盤及各種工具
• core 核心庫：組件核心邏輯
• devenv 開發環境：用的是vagrant
• docs 檔案：檔案相關的內容
• events 事件監聽：事件監聽機制，例如確定某一筆交易已經包括到區塊中
• example 例子：里面包括了一些fabric網路的例子
• gossip gossip協議：最終一致性共識演算法，用于組織內部的區塊同步
• images 鏡像：用于docker鏡像打包
• msp 成員服務管理：member service provider證書管理
• orderer orderer模塊：orderer（排序）節點入口
• peer peer模塊：peer（記賬）節點入口
• proposals 提案：新功能提案
• protos 資料定義：包括了幾乎所有fabric的資料結構、資料服務的定義，

閱讀原始碼需最好用ide方便查看代碼之間的參考，這里選用的是goland，把fabric檔案放到gopath下，編譯器import首先尋找的是gopath

因為orderer節點相當于fabric網路中的一個中心節點，所以從orderer的代碼開始看
代碼注釋括號里的內容是個人理解可能認識的有問題

• orderer的main函式

func main() {

	kingpin.Version("0.0.1")
	switch kingpin.MustParse(app.Parse(os.Args[1:])) {

	// "start" command
	case start.FullCommand():
		logger.Infof("Starting %s", metadata.GetVersionInfo())
	
        '載入配置資訊'
		conf := config.Load()
		
		'初始化日志級別'
		initializeLoggingLevel(conf)
		
		'啟動profile服務（里面涉及的pprof是性能分析服務）'
		initializeProfilingService(conf)
		
		'初始化grpc服務，（讓系統監聽某個埠）'
		grpcServer := initializeGrpcServer(conf)
		
		'加載本地msp服務，需要LocalMspDir,BCCSP,LocalMspID三個引數'
		initializeLocalMsp(conf)
		
		'msp證書用于簽名者實體化(實體化了一個介面)'
		signer := localmsp.NewSigner()

		'初始化多鏈管理者，這里的manager需要著重看一下！！！'
		manager := initializeMultiChainManager(conf, signer)
		server := NewServer(manager, signer)
		ab.RegisterAtomicBroadcastServer(grpcServer.Server(), server)
		logger.Info("Beginning to serve requests")
		grpcServer.Start()
	// "version" command
	case version.FullCommand():
		fmt.Println(metadata.GetVersionInfo())
	}

}

• manager包

// Manager coordinates the creation and access of chains
type Manager interface {
	// GetChain retrieves the chain support for a chain (and whether it exists)
	'根據鏈的名稱獲取一個鏈物件（chainsupport是個介面，并沒有對鏈直接操作）'
	GetChain(chainID string) (ChainSupport, bool)

	// SystemChannelID returns the channel ID for the system channel
	'獲取系統鏈的ID（系統鏈是一個空鏈，用于引導產生新的鏈，因為交易都要屬于一個鏈，所以創建鏈時的交易就需要一個初始鏈即系統鏈）'
	SystemChannelID() string

	// NewChannelConfig returns a bare bones configuration ready for channel
	// creation request to be applied on top of it
	'生成或更新鏈的配置'
	NewChannelConfig(envConfigUpdate *cb.Envelope) (configtxapi.Manager, error)
}

'這是一個配置資源，每個鏈的配置都會打包在這個類里'
type configResources struct {
	configtxapi.Manager
}

'獲取orderer節點的配置'
func (cr *configResources) SharedConfig() config.Orderer {
	oc, ok := cr.OrdererConfig()
	if !ok {
		logger.Panicf("[channel %s] has no orderer configuration", cr.ChainID())
	}
	return oc
}
'賬本資源類，包括了賬本的配置資源和賬本的讀寫物件，可看作對賬本進行操作的入口'
type ledgerResources struct {
	*configResources
	ledger ledger.ReadWriter
}
'manager實作類'
type multiLedger struct {
	chains          map[string]*chainSupport]   多個鏈的物件
	consenters      map[string]Consenter        目前支持的共識機制
	ledgerFactory   ledger.Factory              賬本工廠
	signer          crypto.LocalSigner          簽名物件（方法），一般是msp，msp實作signer的介面
	systemChannelID string                      系統鏈名稱
	systemChannel   *chainSupport				系統鏈物件
}
'獲取區塊里最新的配置交易解碼后列印出來，reader是用來操作賬本的'
func getConfigTx(reader ledger.Reader) *cb.Envelope {
	lastBlock := ledger.GetBlock(reader, reader.Height()-1)
	index, err := utils.GetLastConfigIndexFromBlock(lastBlock)
	if err != nil {
		logger.Panicf("Chain did not have appropriately encoded last config in its latest block: %s", err)
	}
	configBlock := ledger.GetBlock(reader, index)
	if configBlock == nil {
		logger.Panicf("Config block does not exist")
	}

	return utils.ExtractEnvelopeOrPanic(configBlock, 0)
}

// NewManagerImpl produces an instance of a Manager

'輸入賬本工廠、支持的共識機制、簽名物件回傳一個manager實體'
func NewManagerImpl(ledgerFactory ledger.Factory, consenters map[string]Consenter, signer crypto.LocalSigner) Manager {
	ml := &multiLedger{
		chains:        make(map[string]*chainSupport),
		ledgerFactory: ledgerFactory,
		consenters:    consenters,
		signer:        signer,
	}
	'讀取本地存盤鏈的ID'
	existingChains := ledgerFactory.ChainIDs()
	for _, chainID := range existingChains {
		//通過賬本工廠實體化一個賬本讀的物件，read ledger
		rl, err := ledgerFactory.GetOrCreate(chainID)
		if err != nil {
			logger.Panicf("Ledger factory reported chainID %s but could not retrieve it: %s", chainID, err)
		}
		'獲取鏈最新的配置交易'
		configTx := getConfigTx(rl)
		if configTx == nil {
			logger.Panic("Programming error, configTx should never be nil here")
		}
		'將配置交易和ledger物件系結起來'
		ledgerResources := ml.newLedgerResources(configTx)
		chainID := ledgerResources.ChainID()
		//判斷是否有創建其他鏈的權限
		if _, ok := ledgerResources.ConsortiumsConfig(); ok {
			if ml.systemChannelID != "" {
				logger.Panicf("There appear to be two system chains %s and %s", ml.systemChannelID, chainID)
			}
			chain := newChainSupport(createSystemChainFilters(ml, ledgerResources),
				ledgerResources,
				consenters,
				signer)
			logger.Infof("Starting with system channel %s and orderer type %s", chainID, chain.SharedConfig().ConsensusType())
			ml.chains[chainID] = chain
			ml.systemChannelID = chainID
			ml.systemChannel = chain
			// We delay starting this chain, as it might try to copy and replace the chains map via newChain before the map is fully built
			defer chain.start()//延遲啟動該鏈
		} else {
			logger.Debugf("Starting chain: %s", chainID)
			chain := newChainSupport(createStandardFilters(ledgerResources),
				ledgerResources,
				consenters,
				signer)
			ml.chains[chainID] = chain
			chain.start()
		}

	}
	'最后檢驗'
	if ml.systemChannelID == "" {
		logger.Panicf("No system chain found.  If bootstrapping, does your system channel contain a consortiums group definition?")
	}

	return ml
}
'獲取系統鏈（通道）名稱'
func (ml *multiLedger) SystemChannelID() string {
	return ml.systemChannelID
}
'(回傳orderer節點中的一條鏈，以鏈物件chainsupport形式存盤)'
// GetChain retrieves the chain support for a chain (and whether it exists)
func (ml *multiLedger) GetChain(chainID string) (ChainSupport, bool) {
	cs, ok := ml.chains[chainID]
	return cs, ok
}

'新建賬本資源，即新建一個賬本并回傳新賬本的資源類作為賬本的入口'
func (ml *multiLedger) newLedgerResources(configTx *cb.Envelope) *ledgerResources {
	initializer := configtx.NewInitializer()
	configManager, err := configtx.NewManagerImpl(configTx, initializer, nil)
	if err != nil {
		logger.Panicf("Error creating configtx manager and handlers: %s", err)
	}
		'（這里的configManager是單條鏈上的manager，本檔案是orderer上的manager是multiManager）'
	chainID := configManager.ChainID()

	ledger, err := ml.ledgerFactory.GetOrCreate(chainID)
	if err != nil {
		logger.Panicf("Error getting ledger for %s", chainID)
	}

	return &ledgerResources{
		configResources: &configResources{Manager: configManager},
		ledger:          ledger,
	}
}

'創建新鏈（通道），會把配置資訊寫入新的賬本中，向multiLedger的鏈的map中增加新鏈的內容'
func (ml *multiLedger) newChain(configtx *cb.Envelope) {
	ledgerResources := ml.newLedgerResources(configtx)
	ledgerResources.ledger.Append(ledger.CreateNextBlock(ledgerResources.ledger, []*cb.Envelope{configtx}))      配置資訊寫入新鏈的賬本

	// Copy the map to allow concurrent reads from broadcast/deliver while the new chainSupport is
	newChains := make(map[string]*chainSupport)
	for key, value := range ml.chains {
		newChains[key] = value
	}

	cs := newChainSupport(createStandardFilters(ledgerResources), ledgerResources, ml.consenters, ml.signer)
	chainID := ledgerResources.ChainID()

	logger.Infof("Created and starting new chain %s", chainID)

	newChains[string(chainID)] = cs
	cs.start()

	ml.chains = newChains
}
'回傳當前orderer節點上的鏈數（通道數）'
func (ml *multiLedger) channelsCount() int {
	return len(ml.chains)
}

'生成新鏈的配置，主要是一些檢查性的作業 （新建通道組態檔，也用到了單鏈上的manager，應該是有區別）'
func (ml *multiLedger) NewChannelConfig(envConfigUpdate *cb.Envelope) (configtxapi.Manager, error) {
	configUpdatePayload, err := utils.UnmarshalPayload(envConfigUpdate.Payload)
	if err != nil {
		return nil, fmt.Errorf("Failing initial channel config creation because of payload unmarshaling error: %s", err)
	}

	configUpdateEnv, err := configtx.UnmarshalConfigUpdateEnvelope(configUpdatePayload.Data)
	if err != nil {
		return nil, fmt.Errorf("Failing initial channel config creation because of config update envelope unmarshaling error: %s", err)
	}

	if configUpdatePayload.Header == nil {
		return nil, fmt.Errorf("Failed initial channel config creation because config update header was missing")
	}
	channelHeader, err := utils.UnmarshalChannelHeader(configUpdatePayload.Header.ChannelHeader)

	configUpdate, err := configtx.UnmarshalConfigUpdate(configUpdateEnv.ConfigUpdate)
	if err != nil {
		return nil, fmt.Errorf("Failing initial channel config creation because of config update unmarshaling error: %s", err)
	}

	if configUpdate.ChannelId != channelHeader.ChannelId {
		return nil, fmt.Errorf("Failing initial channel config creation: mismatched channel IDs: '%s' != '%s'", configUpdate.ChannelId, channelHeader.ChannelId)
	}

	if configUpdate.WriteSet == nil {
		return nil, fmt.Errorf("Config update has an empty writeset")
	}

	if configUpdate.WriteSet.Groups == nil || configUpdate.WriteSet.Groups[config.ApplicationGroupKey] == nil {
		return nil, fmt.Errorf("Config update has missing application group")
	}

	if uv := configUpdate.WriteSet.Groups[config.ApplicationGroupKey].Version; uv != 1 {
		return nil, fmt.Errorf("Config update for channel creation does not set application group version to 1, was %d", uv)
	}

	consortiumConfigValue, ok := configUpdate.WriteSet.Values[config.ConsortiumKey]
	if !ok {
		return nil, fmt.Errorf("Consortium config value missing")
	}

	consortium := &cb.Consortium{}
	err = proto.Unmarshal(consortiumConfigValue.Value, consortium)
	if err != nil {
		return nil, fmt.Errorf("Error reading unmarshaling consortium name: %s", err)
	}

	applicationGroup := cb.NewConfigGroup()
	consortiumsConfig, ok := ml.systemChannel.ConsortiumsConfig()
	if !ok {
		return nil, fmt.Errorf("The ordering system channel does not appear to support creating channels")
	}

	consortiumConf, ok := consortiumsConfig.Consortiums()[consortium.Name]
	if !ok {
		return nil, fmt.Errorf("Unknown consortium name: %s", consortium.Name)
	}

	applicationGroup.Policies[config.ChannelCreationPolicyKey] = &cb.ConfigPolicy{
		Policy: consortiumConf.ChannelCreationPolicy(),
	}
	applicationGroup.ModPolicy = config.ChannelCreationPolicyKey

	// Get the current system channel config
	systemChannelGroup := ml.systemChannel.ConfigEnvelope().Config.ChannelGroup

	// If the consortium group has no members, allow the source request to have no members.  However,
	// if the consortium group has any members, there must be at least one member in the source request
	if len(systemChannelGroup.Groups[config.ConsortiumsGroupKey].Groups[consortium.Name].Groups) > 0 &&
		len(configUpdate.WriteSet.Groups[config.ApplicationGroupKey].Groups) == 0 {
		return nil, fmt.Errorf("Proposed configuration has no application group members, but consortium contains members")
	}

	// If the consortium has no members, allow the source request to contain arbitrary members
	// Otherwise, require that the supplied members are a subset of the consortium members
	if len(systemChannelGroup.Groups[config.ConsortiumsGroupKey].Groups[consortium.Name].Groups) > 0 {
		for orgName := range configUpdate.WriteSet.Groups[config.ApplicationGroupKey].Groups {
			consortiumGroup, ok := systemChannelGroup.Groups[config.ConsortiumsGroupKey].Groups[consortium.Name].Groups[orgName]
			if !ok {
				return nil, fmt.Errorf("Attempted to include a member which is not in the consortium")
			}
			applicationGroup.Groups[orgName] = consortiumGroup
		}
	}

	channelGroup := cb.NewConfigGroup()

	// Copy the system channel Channel level config to the new config
	for key, value := range systemChannelGroup.Values {
		channelGroup.Values[key] = value
		if key == config.ConsortiumKey {
			// Do not set the consortium name, we do this later
			continue
		}
	}

	for key, policy := range systemChannelGroup.Policies {
		channelGroup.Policies[key] = policy
	}

	// Set the new config orderer group to the system channel orderer group and the application group to the new application group
	channelGroup.Groups[config.OrdererGroupKey] = systemChannelGroup.Groups[config.OrdererGroupKey]
	channelGroup.Groups[config.ApplicationGroupKey] = applicationGroup
	channelGroup.Values[config.ConsortiumKey] = config.TemplateConsortium(consortium.Name).Values[config.ConsortiumKey]

	templateConfig, _ := utils.CreateSignedEnvelope(cb.HeaderType_CONFIG, configUpdate.ChannelId, ml.signer, &cb.ConfigEnvelope{
		Config: &cb.Config{
			ChannelGroup: channelGroup,
		},
	}, msgVersion, epoch)

	initializer := configtx.NewInitializer()

	// This is a very hacky way to disable the sanity check logging in the policy manager
	// for the template configuration, but it is the least invasive near a release
	pm, ok := initializer.PolicyManager().(*policies.ManagerImpl)
	if ok {
		pm.SuppressSanityLogMessages = true
		defer func() {
			pm.SuppressSanityLogMessages = false
		}()
	}

	return configtx.NewManagerImpl(templateConfig, initializer, nil)
}