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。

但这样的事件机制在大数据处理时，可能会出现问题。