hyperledger fabric 共識算法初始化

fabric的 hyperledger共識算法代碼全部都在consensus文件夾里，consensus文件夾里主要分為controller，executor，helper，noops，pbft，util文件模塊。 
其中consensus.Go 主要包含了算法插件內部對外部暴露的接口和hyperledger外部對算法內部暴露的接口。

• controller：共識算法模塊是可插拔的，在controller里面可以選擇具體使用哪種共識算法。目前hyperledger它提供了一個pbft算法和一個比較簡單的noops算法。
• executor：executor和helper是兩個相互依賴的模塊，主要提供了共識算法和外部銜接的一塊代碼。主要負責事件處理的轉接。
• helper:這里面主要包含了對外部接口的一個調用，比如執行處理transaction，stateupdate，持久化一些對象等。
• noops： noops也是一種比較簡單共識算法
• pbft： pbft算法，下面會簡單的介紹一下pbft算法的調用流程。
• util： 一些交互需要的工具包，最主要的一個實現的功能就是它的消息機制。

下面簡要介紹兩點，一點pbft算法代碼內部從頭到尾的一個調用流程，一點是pbft算法內部的事件機制和timeout代碼的一個簡要解析。

內部調用流程

在engine.go里面有獲取一個共識算法plugin；

它初始化一個consenter和一個helper，並互相把一個句柄賦值給了對方。這樣做的目的，就是為了可以讓外部調用內部，內部可以調用外部。

 

// GetEngine returns initialized peer.Engine
func GetEngine(coord peer.MessageHandlerCoordinator) (peer.Engine, error) {
	var err error
	engineOnce.Do(func() {
		engine = new(EngineImpl)
		engine.helper = NewHelper(coord)
		engine.consenter = controller.NewConsenter(engine.helper)
		engine.helper.setConsenter(engine.consenter)
		engine.peerEndpoint, err = coord.GetPeerEndpoint()
		engine.consensusFan = util.NewMessageFan()

		go func() {
			logger.Debug("Starting up message thread for consenter")

			// The channel never closes, so this should never break
			for msg := range engine.consensusFan.GetOutChannel() {
				engine.consenter.RecvMsg(msg.Msg, msg.Sender)
			}
		}()
	})
	return engine, err
}

 

 

其中：

首先看一下它是如何初始化一個共識模塊的：

controller.NewConsenter(engine.helper)

1. 調用controller獲取一個plugin，當選擇是pbft算法時，它會調用pbft.go 里的 GetPlugin(c consensus.Stack)

// NewConsenter constructs a Consenter object if not already present
func NewConsenter(stack consensus.Stack) consensus.Consenter {

	plugin := strings.ToLower(viper.GetString("peer.validator.consensus.plugin"))
	if plugin == "pbft" {
		logger.Infof("Creating consensus plugin %s", plugin)
		return pbft.GetPlugin(stack)
	}
	logger.Info("Creating default consensus plugin (noops)")
	return noops.GetNoops(stack)

}

　　pbft.GetPlugin(stack)：獲取一個pbft的plugin。

// GetPlugin returns the handle to the Consenter singleton
func GetPlugin(c consensus.Stack) consensus.Consenter {
	if pluginInstance == nil {
		pluginInstance = New(c)
	}
	return pluginInstance
}


// New creates a new Obc* instance that provides the Consenter interface.
// Internally, it uses an opaque pbft-core instance.
func New(stack consensus.Stack) consensus.Consenter {
	handle, _, _ := stack.GetNetworkHandles()
	id, _ := getValidatorID(handle)

	switch strings.ToLower(config.GetString("general.mode")) {
	case "batch":
		return newObcBatch(id, config, stack)
	default:
		panic(fmt.Errorf("Invalid PBFT mode: %s", config.GetString("general.mode")))
	}
}

　　

2. 用方法newObcBatch(id uint64, config *viper.Viper, stack consensus.Stack)初始化一個obcbatch對象。
   這個batch對象的作用就是用來做request緩存，提高transaction的執行效率，如果每來一個請求就去做一次共識，那代價會很高。緩存存儲在batchStore里。


3. 在newobcbatch時，會初始化得到一個pbftcore的一個實例，這個是算法的核心模塊。
   並此時會啟動一個batchTimer（這個batchTimer是一個計時器，當batchTimer timeout后會觸發一個sendbatch操作，這個只有primary節點才會去做）。
   當然此時會創建一個事件處理機制，這個事件處理機制是各個模塊溝通的一個bridge。
　　

    func newPbftCore(id uint64, config *viper.Viper, consumer innerStack, etf events.TimerFactory) *pbftCore 

4. 在初始化pbftcore時，即newPbftCore中，在把所用配置讀進的同時，創建了三個timer ：

    instance.newViewTimer = etf.CreateTimer()

    instance.vcResendTimer = etf.CreateTimer()

    instance.nullRequestTimer = etf.CreateTimer()

• newViewTimer對應於viewChangeTimerEvent{}，當這個timer在一定時間沒有close時，就會觸發一個viewchange事件。
• vcResendTimer對應viewChangeResendTimerEvent，發出viewchange過時時會觸發一個將viewchange從新發送。
• nullRequestTimer對應nullRequestEvent，如果主節點長期沒有發送preprepare消息，也就是分配了seq的reqBatch。它timeout就認為主節點掛掉了然后發送viewchange消息。

 

 

算法內部的事件機制

         要想完全看懂它的調用流程，就必須理解它的事件流。說起來它的事件流真是復雜，我覺得可以不必要這么復雜。

它有兩個事件流，一個是在helper里，一個是在batch里。

首先分析一下它的事件流工具，它整個代碼都在util包里的events.go。

首先看一下它提供的接口：

 

type Manager interface {
    Inject(Event)   // A temporary interface to allow the event manager thread to skip the queue
    Queue() chan<- Event // Get a write-only reference to the queue, to submit events
    SetReceiver(Receiver) // Set the target to route events to
    Start()              // Starts the Manager thread TODO, these thread management things should probably go away
    Halt()                // Stops the Manager thread
}

　　SetReceiver(Receiver)是一個很重要的接口，決定了這個事件機制的靈活性。考驗我表達能力的時候到了～～，一個事件機制必定有一個輸入和一個輸出，這個SetReceiver(Receiver) interface 方法就決定了事件流的去向。下面是receiver的interface，凡是事件的接受者都必須實現ProcessEvent(e Event) Event方法。batch里面實現了此方法。

 

type Receiver interface {
    // ProcessEvent delivers an event to the Receiver, if it returns non-nil, the return is the next processed event
    ProcessEvent(e Event) Event
}

　　那對應的輸出，Queue() chan<- Event ，它返回一個event channel，你所有的消息盡管往里面發。接收者取決於receiver。

 

 

func SendEvent(receiver Receiver, event Event) {
    next := event
    for {
        // If an event returns something non-nil, then process it as a new event
        next = receiver.ProcessEvent(next)
        if next == nil {
            break
        }
    }
}

　　

這段代碼是把事件傳給receiver處理。舉個batch事件流機制的例子。

在external.go里面實現了接收外邊request請求的接口。在obcbatch初始化會對其創建並且把event manager復制給externalEventReceiver。因此所有接收到這個manager的消息都會進入到batch里面。

 

 

// RecvMsg is called by the stack when a new message is received
func (eer *externalEventReceiver) RecvMsg(ocMsg *pb.Message, senderHandle *pb.PeerID) error {
    eer.manager.Queue() <- batchMessageEvent{
        msg:    ocMsg,
        sender: senderHandle,
    }
    return nil
}

　　當接收到一個request時，將batchMessageEvent放到事件流，之后

func (em *managerImpl) eventLoop() {
    for {
        select {
        case next := <-em.events:
            em.Inject(next)
        case <-em.exit:
            logger.Debug("eventLoop told to exit")
            return
        }
    }
}

　　

這個死循環接收到的event 進行了em.Inject(next),並執行

 

func SendEvent(receiver Receiver, event Event) {
    next := event
    for {
        // If an event returns something non-nil, then process it as a new event
        next = receiver.ProcessEvent(next)
        if next == nil {
            break
        }
    }
}

　　

之后在obcbatch ProcessEvent里執行了這個操作

case batchMessageEvent:
        ocMsg := et
        return op.processMessage(ocMsg.msg, ocMsg.sender)

　　

這是消息往里拋的過程，同理，消息往外拋，就是算法內部把event拋給外部executor的event manager。

Timer機制

timer機制和event機制有很大關聯，time out后，會把事先創建的event塞到eventmanager里的事件流里。

 

type Timer interface {
    SoftReset(duration time.Duration, event Event) // start a new countdown, only if one is not already started
    Reset(duration time.Duration, event Event)     // start a new countdown, clear any pending events
    Stop()                                         // stop the countdown, clear any pending events
    Halt()                                         // Stops the Timer thread
}

　　

設置time out主要是SoftReset和reset方法。因此在初始化，會把Manager傳給Timer。

但這樣的事件機制在大數據處理時，可能會出現問題。