1 Star 0 Fork 0

妥協/fabric

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
consensus.go 4.28 KB
一键复制 编辑 原始数据 按行查看 历史
Yoav Tock 提交于 2019-04-03 16:49 . FAB-15044 Mig-v1 cleanup #3
/*
Copyright IBM Corp. All Rights Reserved.
SPDX-License-Identifier: Apache-2.0
*/
package solo
import (
"fmt"
"time"
"github.com/hyperledger/fabric/common/flogging"
"github.com/hyperledger/fabric/orderer/consensus"
cb "github.com/hyperledger/fabric/protos/common"
)
var logger = flogging.MustGetLogger("orderer.consensus.solo")
type consenter struct{}
type chain struct {
support consensus.ConsenterSupport
sendChan chan *message
exitChan chan struct{}
}
type message struct {
configSeq uint64
normalMsg *cb.Envelope
configMsg *cb.Envelope
}
// New creates a new consenter for the solo consensus scheme.
// The solo consensus scheme is very simple, and allows only one consenter for a given chain (this process).
// It accepts messages being delivered via Order/Configure, orders them, and then uses the blockcutter to form the messages
// into blocks before writing to the given ledger
func New() consensus.Consenter {
return &consenter{}
}
func (solo *consenter) HandleChain(support consensus.ConsenterSupport, metadata *cb.Metadata) (consensus.Chain, error) {
return newChain(support), nil
}
func newChain(support consensus.ConsenterSupport) *chain {
return &chain{
support: support,
sendChan: make(chan *message),
exitChan: make(chan struct{}),
}
}
func (ch *chain) Start() {
go ch.main()
}
func (ch *chain) Halt() {
select {
case <-ch.exitChan:
// Allow multiple halts without panic
default:
close(ch.exitChan)
}
}
func (ch *chain) WaitReady() error {
return nil
}
// Order accepts normal messages for ordering
func (ch *chain) Order(env *cb.Envelope, configSeq uint64) error {
select {
case ch.sendChan <- &message{
configSeq: configSeq,
normalMsg: env,
}:
return nil
case <-ch.exitChan:
return fmt.Errorf("Exiting")
}
}
// Configure accepts configuration update messages for ordering
func (ch *chain) Configure(config *cb.Envelope, configSeq uint64) error {
select {
case ch.sendChan <- &message{
configSeq: configSeq,
configMsg: config,
}:
return nil
case <-ch.exitChan:
return fmt.Errorf("Exiting")
}
}
// Errored only closes on exit
func (ch *chain) Errored() <-chan struct{} {
return ch.exitChan
}
func (ch *chain) main() {
var timer <-chan time.Time
var err error
for {
seq := ch.support.Sequence()
err = nil
select {
case msg := <-ch.sendChan:
if msg.configMsg == nil {
// NormalMsg
if msg.configSeq < seq {
_, err = ch.support.ProcessNormalMsg(msg.normalMsg)
if err != nil {
logger.Warningf("Discarding bad normal message: %s", err)
continue
}
}
batches, pending := ch.support.BlockCutter().Ordered(msg.normalMsg)
for _, batch := range batches {
block := ch.support.CreateNextBlock(batch)
ch.support.WriteBlock(block, nil)
}
switch {
case timer != nil && !pending:
// Timer is already running but there are no messages pending, stop the timer
timer = nil
case timer == nil && pending:
// Timer is not already running and there are messages pending, so start it
timer = time.After(ch.support.SharedConfig().BatchTimeout())
logger.Debugf("Just began %s batch timer", ch.support.SharedConfig().BatchTimeout().String())
default:
// Do nothing when:
// 1. Timer is already running and there are messages pending
// 2. Timer is not set and there are no messages pending
}
} else {
// ConfigMsg
if msg.configSeq < seq {
msg.configMsg, _, err = ch.support.ProcessConfigMsg(msg.configMsg)
if err != nil {
logger.Warningf("Discarding bad config message: %s", err)
continue
}
}
batch := ch.support.BlockCutter().Cut()
if batch != nil {
block := ch.support.CreateNextBlock(batch)
ch.support.WriteBlock(block, nil)
}
block := ch.support.CreateNextBlock([]*cb.Envelope{msg.configMsg})
ch.support.WriteConfigBlock(block, nil)
timer = nil
}
case <-timer:
//clear the timer
timer = nil
batch := ch.support.BlockCutter().Cut()
if len(batch) == 0 {
logger.Warningf("Batch timer expired with no pending requests, this might indicate a bug")
continue
}
logger.Debugf("Batch timer expired, creating block")
block := ch.support.CreateNextBlock(batch)
ch.support.WriteBlock(block, nil)
case <-ch.exitChan:
logger.Debugf("Exiting")
return
}
}
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/liurenhao/fabric.git
git@gitee.com:liurenhao/fabric.git
liurenhao
fabric
fabric
v1.4.5

搜索帮助