diff --git a/blockproducer/blocknode.go b/blockproducer/blocknode.go index 317ee655f..bc3e56c48 100644 --- a/blockproducer/blocknode.go +++ b/blockproducer/blocknode.go @@ -17,6 +17,8 @@ package blockproducer import ( + "sync/atomic" + "github.com/CovenantSQL/CovenantSQL/crypto/hash" "github.com/CovenantSQL/CovenantSQL/types" ) @@ -27,12 +29,13 @@ type blockNode struct { count uint32 height uint32 // Cached fields for quick reference - hash hash.Hash - block *types.BPBlock + hash hash.Hash + txCount int + block atomic.Value } -func newBlockNode(h uint32, b *types.BPBlock, p *blockNode) *blockNode { - return &blockNode{ +func newBlockNode(h uint32, b *types.BPBlock, p *blockNode) (node *blockNode) { + node = &blockNode{ parent: p, count: func() uint32 { @@ -43,17 +46,27 @@ func newBlockNode(h uint32, b *types.BPBlock, p *blockNode) *blockNode { }(), height: h, - hash: b.SignedHeader.DataHash, - block: b, + hash: b.SignedHeader.DataHash, + txCount: len(b.Transactions), } + node.block.Store(b) + return +} + +func (n *blockNode) load() *types.BPBlock { + return n.block.Load().(*types.BPBlock) +} + +func (n *blockNode) clear() { + n.block.Store((*types.BPBlock)(nil)) } -// fetchNodeList returns the block node list within range (from, n.count] from node head n. +// fetchNodeList returns the block node list within range [from, n.count] from node head n. func (n *blockNode) fetchNodeList(from uint32) (bl []*blockNode) { - if n.count <= from { + if n.count < from { return } - bl = make([]*blockNode, n.count-from) + bl = make([]*blockNode, n.count-from+1) var iter = n for i := len(bl) - 1; i >= 0; i-- { bl[i] = iter diff --git a/blockproducer/blocknode_test.go b/blockproducer/blocknode_test.go index 09299abdc..8c72f5111 100644 --- a/blockproducer/blocknode_test.go +++ b/blockproducer/blocknode_test.go @@ -115,11 +115,11 @@ func TestBlockNode(t *testing.T) { So(n0.count, ShouldEqual, 0) So(n1.count, ShouldEqual, n0.count+1) - So(n0.fetchNodeList(0), ShouldBeEmpty) So(n0.fetchNodeList(1), ShouldBeEmpty) So(n0.fetchNodeList(2), ShouldBeEmpty) - So(n3.fetchNodeList(0), ShouldResemble, []*blockNode{n1, n2, n3}) - So(n4p.fetchNodeList(2), ShouldResemble, []*blockNode{n3p, n4p}) + So(n0.fetchNodeList(3), ShouldBeEmpty) + So(n3.fetchNodeList(1), ShouldResemble, []*blockNode{n1, n2, n3}) + So(n4p.fetchNodeList(3), ShouldResemble, []*blockNode{n3p, n4p}) So(n0.ancestor(1), ShouldBeNil) So(n3.ancestor(3), ShouldEqual, n3) diff --git a/blockproducer/branch.go b/blockproducer/branch.go index fcba6e5dc..6182d3f7d 100644 --- a/blockproducer/branch.go +++ b/blockproducer/branch.go @@ -10,7 +10,8 @@ * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and * limitations under the License. + * See the License for the specific language governing permissions and + * limitations under the License. */ package blockproducer @@ -43,7 +44,7 @@ func newBranch( br *branch, err error, ) { var ( - list = headNode.fetchNodeList(baseNode.count) + list = headNode.fetchNodeList(baseNode.count + 1) inst = &branch{ head: headNode, preview: baseState.makeCopy(), @@ -57,11 +58,12 @@ func newBranch( } // Apply new blocks to view and pool for _, bn := range list { - if len(bn.block.Transactions) > conf.MaxTransactionsPerBlock { + if bn.txCount > conf.MaxTransactionsPerBlock { return nil, ErrTooManyTransactionsInBlock } - for _, v := range bn.block.Transactions { + var block = bn.load() + for _, v := range block.Transactions { var k = v.Hash() // Check in tx pool if _, ok := inst.unpacked[k]; ok { @@ -126,17 +128,18 @@ func (b *branch) addTx(tx pi.Transaction) { } func (b *branch) applyBlock(n *blockNode) (br *branch, err error) { - if !b.head.hash.IsEqual(n.block.ParentHash()) { + var block = n.load() + if !b.head.hash.IsEqual(block.ParentHash()) { err = ErrParentNotMatch return } var cpy = b.makeArena() - if len(n.block.Transactions) > conf.MaxTransactionsPerBlock { + if n.txCount > conf.MaxTransactionsPerBlock { return nil, ErrTooManyTransactionsInBlock } - for _, v := range n.block.Transactions { + for _, v := range block.Transactions { var k = v.Hash() // Check in tx pool if _, ok := cpy.unpacked[k]; ok { @@ -258,13 +261,13 @@ func (b *branch) sprint(from uint32) (buff string) { if i == 0 { var p = v.parent buff += fmt.Sprintf("* #%d:%d %s {%d}", - p.height, p.count, p.hash.Short(4), len(p.block.Transactions)) + p.height, p.count, p.hash.Short(4), p.txCount) } if d := v.height - v.parent.height; d > 1 { buff += fmt.Sprintf(" <-- (skip %d blocks)", d-1) } buff += fmt.Sprintf(" <-- #%d:%d %s {%d}", - v.height, v.count, v.hash.Short(4), len(v.block.Transactions)) + v.height, v.count, v.hash.Short(4), v.txCount) } return } diff --git a/blockproducer/chain.go b/blockproducer/chain.go index 0e05ea9d8..86789e2fc 100644 --- a/blockproducer/chain.go +++ b/blockproducer/chain.go @@ -25,10 +25,11 @@ import ( "sync" "time" + lru "github.com/hashicorp/golang-lru" + "github.com/pkg/errors" mw "github.com/zserge/metric" pi "github.com/CovenantSQL/CovenantSQL/blockproducer/interfaces" - "github.com/CovenantSQL/CovenantSQL/chainbus" "github.com/CovenantSQL/CovenantSQL/conf" "github.com/CovenantSQL/CovenantSQL/crypto" "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" @@ -40,26 +41,44 @@ import ( "github.com/CovenantSQL/CovenantSQL/types" "github.com/CovenantSQL/CovenantSQL/utils/log" xi "github.com/CovenantSQL/CovenantSQL/xenomint/interfaces" - "github.com/pkg/errors" ) +// Metric keys +const ( + mwKeyHeight = "service:bp:height" + mwKeyTxPooled = "service:bp:pooled" + mwKeyTxConfirmed = "service:bp:confirmed" +) + +func init() { + expvar.Publish(mwKeyTxPooled, mw.NewCounter("5m1m")) + expvar.Publish(mwKeyTxConfirmed, mw.NewCounter("5m1m")) +} + // Chain defines the main chain. type Chain struct { // Routine controlling components ctx context.Context cancel context.CancelFunc wg *sync.WaitGroup + // RPC components server *rpc.Server caller *rpc.Caller + // Other components - storage xi.Storage - chainBus chainbus.Bus + storage xi.Storage + // NOTE(leventeliu): this LRU object is only used for block cache control, + // do NOT read it in any case. + blockCache *lru.Cache + // Channels for incoming blocks and transactions pendingBlocks chan *types.BPBlock pendingAddTxReqs chan *types.AddTxReq + // The following fields are read-only in runtime address proto.AccountAddress + mode RunMode genesisTime time.Time period time.Duration tick time.Duration @@ -77,47 +96,32 @@ type Chain struct { headBranch *branch branches []*branch txPool map[hash.Hash]pi.Transaction - mode RunMode } // NewChain creates a new blockchain. func NewChain(cfg *Config) (c *Chain, err error) { - // Normally, NewChain() should only be called once in app. - // So, we just check expvar without a lock - if expvar.Get("height") == nil { - expvar.Publish("height", mw.NewGauge("5m1s")) - } return NewChainWithContext(context.Background(), cfg) } // NewChainWithContext creates a new blockchain with context. func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) { var ( - existed bool - ierr error - - cld context.Context - ccl context.CancelFunc - l = uint32(len(cfg.Peers.Servers)) - t float64 - m uint32 + ierr error st xi.Storage - irre *blockNode + cache *lru.Cache + lastIrre *blockNode heads []*blockNode immutable *metaState txPool map[hash.Hash]pi.Transaction - branches []*branch - br, head *branch - headIndex int + branches []*branch + headBranch *branch + headIndex int - pubKey *asymmetric.PublicKey addr proto.AccountAddress bpInfos []*blockProducerInfo localBPInfo *blockProducerInfo - - bus = chainbus.New() ) // Verify genesis block in config @@ -131,6 +135,7 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) } // Open storage + var existed bool if fi, err := os.Stat(cfg.DataFile); err == nil && fi.Mode().IsRegular() { existed = true } @@ -144,6 +149,21 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) } }() + // Create block cache + if cfg.BlockCacheSize > conf.MaxCachedBlock { + cfg.BlockCacheSize = conf.MaxCachedBlock + } + if cfg.BlockCacheSize <= 0 { + cfg.BlockCacheSize = 1 // Must provide a positive size + } + if cache, err = lru.NewWithEvict(cfg.BlockCacheSize, func(key interface{}, value interface{}) { + if node, ok := value.(*blockNode); ok && node != nil { + node.clear() + } + }); err != nil { + return + } + // Create initial state from genesis block and store if !existed { var init = newMetaState() @@ -162,25 +182,36 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) } } - // Load from database and rebuild branches - if irre, heads, immutable, txPool, ierr = loadDatabase(st); ierr != nil { + // Load from database + if lastIrre, heads, immutable, txPool, ierr = loadDatabase(st); ierr != nil { err = errors.Wrap(ierr, "failed to load data from storage") return } - if persistedGenesis := irre.ancestorByCount(0); persistedGenesis == nil || + + // Check genesis block + var irreBlocks = lastIrre.fetchNodeList(0) + if persistedGenesis := irreBlocks[0]; persistedGenesis == nil || !persistedGenesis.hash.IsEqual(cfg.Genesis.BlockHash()) { err = ErrGenesisHashNotMatch return } + + // Add blocks to LRU list + for _, v := range irreBlocks { + cache.Add(v.count, v) + } + + // Rebuild branches for _, v := range heads { log.WithFields(log.Fields{ - "irre_hash": irre.hash.Short(4), - "irre_count": irre.count, + "irre_hash": lastIrre.hash.Short(4), + "irre_count": lastIrre.count, "head_hash": v.hash.Short(4), "head_count": v.count, }).Debug("checking head") - if v.hasAncestor(irre) { - if br, ierr = newBranch(irre, v, immutable, txPool); ierr != nil { + if v.hasAncestor(lastIrre) { + var br *branch + if br, ierr = newBranch(lastIrre, v, immutable, txPool); ierr != nil { err = errors.Wrapf(ierr, "failed to rebuild branch with head %s", v.hash.Short(4)) return } @@ -192,15 +223,16 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) return } - // Set head branch + // Select head branch for i, v := range branches { - if head == nil || v.head.count > head.head.count { + if headBranch == nil || v.head.count > headBranch.head.count { headIndex = i - head = v + headBranch = v } } // Get accountAddress + var pubKey *asymmetric.PublicKey if pubKey, err = kms.GetLocalPublicKey(); err != nil { return } @@ -209,18 +241,26 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) } // Setup peer list - if localBPInfo, bpInfos, err = buildBlockProducerInfos(cfg.NodeID, cfg.Peers, cfg.Mode == APINodeMode); err != nil { + var ( + l = uint32(len(cfg.Peers.Servers)) + + threshold float64 + needConfirms uint32 + ) + if localBPInfo, bpInfos, err = buildBlockProducerInfos( + cfg.NodeID, cfg.Peers, cfg.Mode == APINodeMode, + ); err != nil { return } - if t = cfg.ConfirmThreshold; t <= 0.0 { - t = conf.DefaultConfirmThreshold + if threshold = cfg.ConfirmThreshold; threshold <= 0.0 { + threshold = conf.DefaultConfirmThreshold } - if m = uint32(math.Ceil(float64(l)*t + 1)); m > l { - m = l + if needConfirms = uint32(math.Ceil(float64(l)*threshold + 1)); needConfirms > l { + needConfirms = l } // create chain - cld, ccl = context.WithCancel(ctx) + var cld, ccl = context.WithCancel(ctx) c = &Chain{ ctx: cld, cancel: ccl, @@ -229,13 +269,14 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) server: cfg.Server, caller: rpc.NewCaller(), - storage: st, - chainBus: bus, + storage: st, + blockCache: cache, pendingBlocks: make(chan *types.BPBlock), pendingAddTxReqs: make(chan *types.AddTxReq), address: addr, + mode: cfg.Mode, genesisTime: cfg.Genesis.SignedHeader.Timestamp, period: cfg.Period, tick: cfg.Tick, @@ -243,18 +284,24 @@ func NewChainWithContext(ctx context.Context, cfg *Config) (c *Chain, err error) bpInfos: bpInfos, localBPInfo: localBPInfo, localNodeID: cfg.NodeID, - confirms: m, - nextHeight: head.head.height + 1, + confirms: needConfirms, + nextHeight: headBranch.head.height + 1, offset: time.Duration(0), // TODO(leventeliu): initialize offset + lastIrre: lastIrre, + immutable: immutable, + headIndex: headIndex, + headBranch: headBranch, + branches: branches, + txPool: txPool, + } - lastIrre: irre, - immutable: immutable, - headIndex: headIndex, - headBranch: head, - branches: branches, - txPool: txPool, - mode: cfg.Mode, + // NOTE(leventeliu): this implies that BP chain is a singleton, otherwise we will need + // independent metric key for each chain instance. + if expvar.Get(mwKeyHeight) == nil { + expvar.Publish(mwKeyHeight, mw.NewGauge(fmt.Sprintf("5m%.0fs", cfg.Period.Seconds()))) } + expvar.Get(mwKeyHeight).(mw.Metric).Add(float64(c.head().height)) + log.WithFields(log.Fields{ "local": c.getLocalBPInfo(), "period": c.period, @@ -353,7 +400,11 @@ func (c *Chain) advanceNextHeight(now time.Time, d time.Duration) { "elapsed_seconds": elapsed.Seconds(), }).Info("enclosing current height and advancing to next height") - defer c.increaseNextHeight() + defer func() { + c.increaseNextHeight() + expvar.Get(mwKeyHeight).(mw.Metric).Add(float64(c.head().height)) + }() + // Skip if it's not my turn if c.mode == APINodeMode || !c.isMyTurn() { return @@ -368,7 +419,6 @@ func (c *Chain) advanceNextHeight(now time.Time, d time.Duration) { }).Warn("too much time elapsed in the new period, skip this block") return } - expvar.Get("height").(mw.Metric).Add(float64(c.getNextHeight())) log.WithField("height", c.getNextHeight()).Info("producing a new block") if err := c.produceBlock(now); err != nil { log.WithField("now", now.Format(time.RFC3339Nano)).WithError(err).Errorln( @@ -500,7 +550,9 @@ func (c *Chain) processAddTxReq(addTxReq *types.AddTxReq) { // Add to tx pool if err = c.storeTx(tx); err != nil { le.WithError(err).Error("failed to add transaction") + return } + expvar.Get(mwKeyTxPooled).(mw.Metric).Add(1) } func (c *Chain) processTxs(ctx context.Context) { @@ -629,6 +681,7 @@ func (c *Chain) replaceAndSwitchToBranch( newIrres []*blockNode sps []storageProcedure up storageCallback + txCount int height = c.heightOfTime(newBlock.Timestamp()) resultTxPool = make(map[hash.Hash]pi.Transaction) @@ -641,14 +694,15 @@ func (c *Chain) replaceAndSwitchToBranch( // May have multiple new irreversible blocks here if peer list shrinks. May also have // no new irreversible block at all if peer list expands. lastIrre = newBranch.head.lastIrreversible(c.confirms) - newIrres = lastIrre.fetchNodeList(c.lastIrre.count) + newIrres = lastIrre.fetchNodeList(c.lastIrre.count + 1) // Apply irreversible blocks to create dirty map on immutable cache for k, v := range c.txPool { resultTxPool[k] = v } for _, b := range newIrres { - for _, tx := range b.block.Transactions { + txCount += b.txCount + for _, tx := range b.load().Transactions { if err := c.immutable.apply(tx); err != nil { log.WithError(err).Fatal("failed to apply block to immutable database") } @@ -679,7 +733,7 @@ func (c *Chain) replaceAndSwitchToBranch( sps = append(sps, addBlock(height, newBlock)) sps = append(sps, buildBlockIndex(height, newBlock)) for _, n := range newIrres { - sps = append(sps, deleteTxs(n.block.Transactions)) + sps = append(sps, deleteTxs(n.load().Transactions)) } if len(expiredTxs) > 0 { sps = append(sps, deleteTxs(expiredTxs)) @@ -725,7 +779,7 @@ func (c *Chain) replaceAndSwitchToBranch( // Clear transactions in each branch for _, b := range newIrres { for _, br := range c.branches { - br.clearPackedTxs(b.block.Transactions) + br.clearPackedTxs(b.load().Transactions) } } for _, br := range c.branches { @@ -733,12 +787,18 @@ func (c *Chain) replaceAndSwitchToBranch( } // Update txPool to result txPool (packed and expired transactions cleared!) c.txPool = resultTxPool + // Register new irreversible blocks to LRU cache list + for _, b := range newIrres { + c.blockCache.Add(b.count, b) + } } // Write to immutable database and update cache if err = store(c.storage, sps, up); err != nil { c.immutable.clean() + return } + expvar.Get(mwKeyTxConfirmed).(mw.Metric).Add(float64(txCount)) // TODO(leventeliu): trigger ChainBus.Publish. // ... return @@ -754,7 +814,7 @@ func (c *Chain) stat() { } else { buff += fmt.Sprintf("[%04d] ", i) } - buff += v.sprint(c.lastIrre.count) + buff += v.sprint(c.lastIrre.count + 1) log.WithFields(log.Fields{ "branch": buff, }).Info("runtime state") diff --git a/blockproducer/chain_io.go b/blockproducer/chain_io.go index 7e0755dc2..c93ef536f 100644 --- a/blockproducer/chain_io.go +++ b/blockproducer/chain_io.go @@ -49,8 +49,7 @@ func (c *Chain) fetchLastIrreversibleBlock() ( b *types.BPBlock, count uint32, height uint32, err error, ) { var node = c.lastIrreversibleBlock() - if node.block != nil { - b = node.block + if b = node.load(); b != nil { height = node.height count = node.count return @@ -71,8 +70,7 @@ func (c *Chain) fetchBlockByHeight(h uint32) (b *types.BPBlock, count uint32, er return } // OK, and block is cached - if node.block != nil { - b = node.block + if b = node.load(); b != nil { count = node.count return } @@ -91,8 +89,7 @@ func (c *Chain) fetchBlockByCount(count uint32) (b *types.BPBlock, height uint32 return } // OK, and block is cached - if node.block != nil { - b = node.block + if b = node.load(); b != nil { height = node.height return } diff --git a/blockproducer/chain_test.go b/blockproducer/chain_test.go index cc6c662a9..9461c93d0 100644 --- a/blockproducer/chain_test.go +++ b/blockproducer/chain_test.go @@ -382,11 +382,11 @@ func TestChain(t *testing.T) { So(err, ShouldBeNil) So(count, ShouldEqual, chain.lastIrre.count) So(height, ShouldEqual, chain.lastIrre.height) - So(bl, ShouldResemble, chain.lastIrre.block) + So(bl, ShouldResemble, chain.lastIrre.load()) // Try to use the no-cache version var node = chain.headBranch.head.ancestorByCount(5) - node.block = nil // Clear cached block + node.clear() bl, count, err = chain.fetchBlockByHeight(node.height) So(err, ShouldBeNil) So(count, ShouldEqual, node.count) @@ -396,8 +396,8 @@ func TestChain(t *testing.T) { So(height, ShouldEqual, node.height) So(bl.BlockHash(), ShouldResemble, &node.hash) - var irreBlock = chain.lastIrre.block - chain.lastIrre.block = nil // Clear cached block + var irreBlock = chain.lastIrre.load() + node.clear() bl, count, height, err = chain.fetchLastIrreversibleBlock() So(err, ShouldBeNil) So(bl, ShouldResemble, irreBlock) diff --git a/blockproducer/config.go b/blockproducer/config.go index d6f5befb7..9c167a50f 100644 --- a/blockproducer/config.go +++ b/blockproducer/config.go @@ -51,21 +51,6 @@ type Config struct { Period time.Duration Tick time.Duration -} -// NewConfig creates new config. -func NewConfig(genesis *types.BPBlock, dataFile string, - server *rpc.Server, peers *proto.Peers, - nodeID proto.NodeID, period time.Duration, tick time.Duration) *Config { - config := Config{ - Mode: BPMode, - Genesis: genesis, - DataFile: dataFile, - Server: server, - Peers: peers, - NodeID: nodeID, - Period: period, - Tick: tick, - } - return &config + BlockCacheSize int } diff --git a/blockproducer/rpc.go b/blockproducer/rpc.go index 1b021614b..61e46b723 100644 --- a/blockproducer/rpc.go +++ b/blockproducer/rpc.go @@ -138,13 +138,6 @@ func (s *ChainRPCService) QueryTxState( return } -// Sub is the RPC method to subscribe some event. -func (s *ChainRPCService) Sub(req *types.SubReq, resp *types.SubResp) (err error) { - return s.chain.chainBus.Subscribe(req.Topic, func(request interface{}, response interface{}) { - s.chain.caller.CallNode(req.NodeID.ToNodeID(), req.Callback, request, response) - }) -} - // WaitDatabaseCreation waits for database creation complete. func WaitDatabaseCreation( ctx context.Context, diff --git a/blockproducer/storage.go b/blockproducer/storage.go index 414bed504..23c1adaad 100644 --- a/blockproducer/storage.go +++ b/blockproducer/storage.go @@ -36,71 +36,71 @@ var ( ddls = [...]string{ // Chain state tables `CREATE TABLE IF NOT EXISTS "blocks" ( - "height" INT, - "hash" TEXT, - "parent" TEXT, - "encoded" BLOB, - UNIQUE ("hash") - );`, + "height" INT, + "hash" TEXT, + "parent" TEXT, + "encoded" BLOB, + UNIQUE ("hash") +);`, `CREATE TABLE IF NOT EXISTS "txPool" ( - "type" INT, - "hash" TEXT, - "encoded" BLOB, - UNIQUE ("hash") - );`, + "type" INT, + "hash" TEXT, + "encoded" BLOB, + UNIQUE ("hash") +);`, `CREATE TABLE IF NOT EXISTS "irreversible" ( - "id" INT, - "hash" TEXT, - UNIQUE ("id") - );`, + "id" INT, + "hash" TEXT, + UNIQUE ("id") +);`, // Meta state tables `CREATE TABLE IF NOT EXISTS "accounts" ( - "address" TEXT, - "encoded" BLOB, - UNIQUE ("address") - );`, + "address" TEXT, + "encoded" BLOB, + UNIQUE ("address") +);`, `CREATE TABLE IF NOT EXISTS "shardChain" ( - "address" TEXT, - "id" TEXT, - "encoded" BLOB, - UNIQUE ("address", "id") - );`, + "address" TEXT, + "id" TEXT, + "encoded" BLOB, + UNIQUE ("address", "id") +);`, `CREATE TABLE IF NOT EXISTS "provider" ( - "address" TEXT, - "encoded" BLOB, - UNIQUE ("address") - );`, + "address" TEXT, + "encoded" BLOB, + UNIQUE ("address") +);`, `CREATE TABLE IF NOT EXISTS "indexed_blocks" ( - "height" INTEGER PRIMARY KEY, - "hash" TEXT, - "timestamp" INTEGER, - "version" INTEGER, - "producer" TEXT, - "merkle_root" TEXT, - "parent" TEXT, - "tx_count" INTEGER - );`, + "height" INTEGER PRIMARY KEY, + "hash" TEXT, + "timestamp" INTEGER, + "version" INTEGER, + "producer" TEXT, + "merkle_root" TEXT, + "parent" TEXT, + "tx_count" INTEGER +);`, `CREATE INDEX IF NOT EXISTS "idx__indexed_blocks__hash" ON "indexed_blocks" ("hash");`, `CREATE INDEX IF NOT EXISTS "idx__indexed_blocks__timestamp" ON "indexed_blocks" ("timestamp" DESC);`, `CREATE TABLE IF NOT EXISTS "indexed_transactions" ( - "block_height" INTEGER, - "tx_index" INTEGER, - "hash" TEXT, - "block_hash" TEXT, - "timestamp" INTEGER, - "tx_type" INTEGER, - "address" TEXT, - "raw" TEXT, - PRIMARY KEY ("block_height", "tx_index") - );`, + "block_height" INTEGER, + "tx_index" INTEGER, + "hash" TEXT, + "block_hash" TEXT, + "timestamp" INTEGER, + "tx_type" INTEGER, + "address" TEXT, + "raw" TEXT, + PRIMARY KEY ("block_height", "tx_index") +);`, `CREATE INDEX IF NOT EXISTS "idx__indexed_transactions__hash" ON "indexed_transactions" ("hash");`, `CREATE INDEX IF NOT EXISTS "idx__indexed_transactions__block_hash" ON "indexed_transactions" ("block_hash");`, @@ -416,7 +416,7 @@ func loadTxPool(st xi.Storage) (txPool map[hash.Hash]pi.Transaction, err error) } func loadBlocks( - st xi.Storage, irreHash hash.Hash) (irre *blockNode, heads []*blockNode, err error, + st xi.Storage, irreHash hash.Hash) (lastIrre *blockNode, heads []*blockNode, err error, ) { var ( rows *sql.Rows @@ -494,7 +494,7 @@ func loadBlocks( headsIndex[bh] = bn } - if irre, ok = index[irreHash]; !ok { + if lastIrre, ok = index[irreHash]; !ok { err = errors.Wrapf(ErrParentNotFound, "irreversible block %s not found", ph.Short(4)) return } diff --git a/cmd/cqld/bootstrap.go b/cmd/cqld/bootstrap.go index 72387d2d0..80bf2d423 100644 --- a/cmd/cqld/bootstrap.go +++ b/cmd/cqld/bootstrap.go @@ -158,16 +158,17 @@ func runNode(nodeID proto.NodeID, listenAddr string) (err error) { // init main chain service log.Info("register main chain service rpc") - chainConfig := bp.NewConfig( - genesis, - conf.GConf.BP.ChainFileName, - server, - peers, - nodeID, - conf.GConf.BPPeriod, - conf.GConf.BPTick, - ) - chainConfig.Mode = mode + chainConfig := &bp.Config{ + Mode: mode, + Genesis: genesis, + DataFile: conf.GConf.BP.ChainFileName, + Server: server, + Peers: peers, + NodeID: nodeID, + Period: conf.GConf.BPPeriod, + Tick: conf.GConf.BPTick, + BlockCacheSize: 1000, + } chain, err := bp.NewChain(chainConfig) if err != nil { log.WithError(err).Error("init chain failed") diff --git a/conf/limits.go b/conf/limits.go index 36359a024..e662b791f 100644 --- a/conf/limits.go +++ b/conf/limits.go @@ -17,9 +17,6 @@ package conf const ( - // MaxTxBroadcastTTL defines the TTL limit of a AddTx request broadcasting within the - // block producers. - MaxTxBroadcastTTL = 1 // MaxPendingTxsPerAccount defines the limit of pending transactions of one account. MaxPendingTxsPerAccount = 1000 // MaxTransactionsPerBlock defines the limit of transactions per block. @@ -27,3 +24,11 @@ const ( // MaxRPCPoolPhysicalConnection defines max underlying physical connection for one node pair. MaxRPCPoolPhysicalConnection = 2 ) + +// These limits will not cause inconsistency within certain range. +const ( + // MaxTxBroadcastTTL defines the TTL limit of a AddTx request broadcasting within the + // block producers. + MaxTxBroadcastTTL = 1 + MaxCachedBlock = 1000 +) diff --git a/conf/parameters.go b/conf/parameters.go index 4fa2296ef..0025fda57 100644 --- a/conf/parameters.go +++ b/conf/parameters.go @@ -21,7 +21,7 @@ const ( DefaultConfirmThreshold = float64(2) / 3.0 ) -// This parameters will not cause inconsistency within certain range. +// These parameters will not cause inconsistency within certain range. const ( BPStartupRequiredReachableCount = 2 // NOTE: this includes myself )