cerc-io/go-ethereum
13
0
Fork 0
Code Issues 9 Pull Requests Packages Projects Releases 66 Wiki Activity

intermediate chain event channel for metrics

Browse Source
This commit is contained in:
Roy Crihfield 2020-11-24 22:58:27 +08:00
parent 8c9d8cbc3f
commit ab841a9abe
1 changed files with 17 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
statediff/service.go
View File

@ -230,9 +230,25 @@ func (sds *Service) WriteLoop(chainEventCh chan core.ChainEvent) {
defer chainEventSub.Unsubscribe() defer chainEventSub.Unsubscribe()
errCh := chainEventSub.Err() errCh := chainEventSub.Err()
var wg sync.WaitGroup var wg sync.WaitGroup
// Process metrics for chain events, then forward to workers
chainEventFwd := make(chan core.ChainEvent, chainEventChanSize)
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case chainEvent := <-chainEventCh:
statediffMetrics.lastEventHeight.Update(int64(chainEvent.Block.Number().Uint64()))
statediffMetrics.writeLoopChannelLen.Update(int64(len(chainEventCh)))
chainEventFwd <- chainEvent
case <-sds.QuitChan:
return
}
}
}()
wg.Add(int(sds.numWorkers)) wg.Add(int(sds.numWorkers))
for worker := uint(0); worker < sds.numWorkers; worker++ { for worker := uint(0); worker < sds.numWorkers; worker++ {
params := workerParams{chainEventCh: chainEventCh, errCh: errCh, wg: &wg, id: worker} params := workerParams{chainEventCh: chainEventFwd, errCh: errCh, wg: &wg, id: worker}
go sds.writeLoopWorker(params) go sds.writeLoopWorker(params)
} }
wg.Wait() wg.Wait()
@ -244,10 +260,8 @@ func (sds *Service) writeLoopWorker(params workerParams) {
select { select {
//Notify chain event channel of events //Notify chain event channel of events
case chainEvent := <-params.chainEventCh: case chainEvent := <-params.chainEventCh:
statediffMetrics.writeLoopChannelLen.Update(int64(len(params.chainEventCh)))
log.Debug("WriteLoop(): chain event received", "event", chainEvent) log.Debug("WriteLoop(): chain event received", "event", chainEvent)
currentBlock := chainEvent.Block currentBlock := chainEvent.Block
statediffMetrics.lastEventHeight.Update(int64(currentBlock.Number().Uint64()))
parentBlock := sds.BlockCache.getParentBlock(currentBlock, sds.BlockChain) parentBlock := sds.BlockCache.getParentBlock(currentBlock, sds.BlockChain)
if parentBlock == nil { if parentBlock == nil {
log.Error("Parent block is nil, skipping this block", "block height", currentBlock.Number()) log.Error("Parent block is nil, skipping this block", "block height", currentBlock.Number())