Merge pull request #13 from CortexFoundation/dev

task channel for batch

Author: ucwong

cmd/main.go

@@ -35,11 +35,10 @@ func main() {
 		//m.SwitchService(robot.SRV_RECORD)
 		go func() {
 			for {
-				mm.SwitchService(robot.SRV_RECORD)
+				//mm.SwitchService(robot.SRV_RECORD)
+				//time.Sleep(30 * time.Second)
+				//mm.SwitchService(robot.SRV_MODEL)
 				time.Sleep(30 * time.Second)
-				mm.SwitchService(robot.SRV_MODEL)
-				time.Sleep(30 * time.Second)
-				//m.SwitchService(robot.SRV_RECORD)
 			}
 		}()
 

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/CortexFoundation/merkletree v0.0.0-20230724124840-b6e80265a137
 	github.com/CortexFoundation/torrentfs v1.0.52-0.20230711131012-594cb55f6538
 	github.com/google/uuid v1.3.0
-	github.com/hashicorp/golang-lru v0.5.5-0.20221011183528-d4900dc688bf
+	github.com/hashicorp/golang-lru/v2 v2.0.5-0.20230705200825-f4cd393e7b29
 	github.com/ucwong/golang-kv v1.0.22-0.20230711090627-455356f3db9c
 	go.etcd.io/bbolt v1.3.7
 )
@@ -37,7 +37,6 @@ require (
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
 	github.com/google/flatbuffers v23.5.26+incompatible // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
-	github.com/hashicorp/golang-lru/v2 v2.0.4 // indirect
 	github.com/holiman/uint256 v1.2.3 // indirect
 	github.com/klauspost/compress v1.16.7 // indirect
 	github.com/kr/pretty v0.3.1 // indirect

go.sum

@@ -109,10 +109,8 @@ github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
 github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/hashicorp/go-uuid v1.0.3 h1:2gKiV6YVmrJ1i2CKKa9obLvRieoRGviZFL26PcT/Co8=
-github.com/hashicorp/golang-lru v0.5.5-0.20221011183528-d4900dc688bf h1:BQyif+/dqmbIGXyGhe5bDx/3grIchislVu5pK7j/bMQ=
-github.com/hashicorp/golang-lru v0.5.5-0.20221011183528-d4900dc688bf/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
-github.com/hashicorp/golang-lru/v2 v2.0.4 h1:7GHuZcgid37q8o5i3QI9KMT4nCWQQ3Kx3Ov6bb9MfK0=
-github.com/hashicorp/golang-lru/v2 v2.0.4/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
+github.com/hashicorp/golang-lru/v2 v2.0.5-0.20230705200825-f4cd393e7b29 h1:ms53+xNey/RHASahOxI1AdzSm5snVNH8xQGTDc1xV2I=
+github.com/hashicorp/golang-lru/v2 v2.0.5-0.20230705200825-f4cd393e7b29/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/holiman/uint256 v1.2.3 h1:K8UWO1HUJpRMXBxbmaY1Y8IAMZC/RsKB+ArEnnK4l5o=
 github.com/holiman/uint256 v1.2.3/go.mod h1:SC8Ryt4n+UBbPbIBKaG9zbbDlp4jOru9xFZmPzLUTxw=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=

monitor.go

@@ -30,7 +30,7 @@ import (
 	"github.com/CortexFoundation/robot/backend"
 	"github.com/CortexFoundation/torrentfs/params"
 	"github.com/CortexFoundation/torrentfs/types"
-	lru "github.com/hashicorp/golang-lru"
+	lru "github.com/hashicorp/golang-lru/v2"
 	"github.com/ucwong/golang-kv"
 	"math"
 	"math/big"
@@ -74,10 +74,11 @@ type Monitor struct {
 	wg            sync.WaitGroup
 	rpcWg         sync.WaitGroup
 
-	//taskCh      chan *types.Block
+	taskCh chan *types.Block
+	errCh  chan error
 	//newTaskHook func(*types.Block)
-	blockCache *lru.Cache
-	sizeCache  *lru.Cache
+	blockCache *lru.Cache[uint64, string]
+	sizeCache  *lru.Cache[string, uint64]
 	ckp        *params.TrustedCheckpoint
 	start      mclock.AbsTime
 
@@ -124,10 +125,12 @@ func New(flag *params.Config, cache, compress, listen bool, callback chan any) (
 		srvCh:  make(chan int),
 		//exitSyncCh: make(chan any),
 		scope: uint64(math.Min(float64(runtime.NumCPU()), float64(8))),
-		//taskCh:        make(chan *types.Block, batch),
+		//taskCh: make(chan *types.Block, batch),
+		//taskCh:        make(chan *types.Block, 1),
 		//start: mclock.Now(),
 	}
-
+	m.errCh = make(chan error, m.scope)
+	m.taskCh = make(chan *types.Block, m.scope)
 	// TODO https://github.com/ucwong/golang-kv
 	if fs_, err := backend.NewChainDB(flag); err != nil {
 		log.Error("file storage failed", "err", err)
@@ -140,8 +143,8 @@ func New(flag *params.Config, cache, compress, listen bool, callback chan any) (
 	m.startNumber.Store(0)
 
 	m.terminated.Store(false)
-	m.blockCache, _ = lru.New(delay)
-	m.sizeCache, _ = lru.New(batch)
+	m.blockCache, _ = lru.New[uint64, string](delay)
+	m.sizeCache, _ = lru.New[string, uint64](batch)
 	m.listen = listen
 	m.callback = callback
 
@@ -173,10 +176,6 @@ func New(flag *params.Config, cache, compress, listen bool, callback chan any) (
 	return m, nil
 }
 
-//func (m *Monitor) DB() *backend.ChainDB {
-//	return m.fs
-//}
-
 func (m *Monitor) CurrentNumber() uint64 {
 	return m.currentNumber.Load()
 }
@@ -288,27 +287,29 @@ func (m *Monitor) indexInit() error {
 	return nil
 }
 
-/*func (m *Monitor) taskLoop() {
+func (m *Monitor) taskLoop() {
+	log.Info("Task channel started")
 	defer m.wg.Done()
 	for {
 		select {
 		case task := <-m.taskCh:
-			if m.newTaskHook != nil {
-				m.newTaskHook(task)
-			}
+			//if m.newTaskHook != nil {
+			//	m.newTaskHook(task)
+			//}
 
-			if err := m.solve(task); err != nil {
-				log.Warn("Block solved failed, try again", "err", err, "num", task.Number)
-			}
+			/*if err := m.solve(task); err != nil {
+				m.errCh <- err
+				log.Warn("Block solved failed, try again", "err", err, "num", task.Number, "last", m.lastNumber.Load())
+			} else {
+				m.errCh <- nil
+			}*/
+			m.errCh <- m.solve(task)
 		case <-m.exitCh:
-			if cap(m.taskCh) > 0 {
-				continue
-			}
 			log.Info("Monitor task channel closed")
 			return
 		}
 	}
-}*/
+}
 
 // SetConnection method builds connection to remote or local communicator.
 func (m *Monitor) buildConnection(ipcpath string, rpcuri string) (*rpc.Client, error) {
@@ -379,8 +380,8 @@ func (m *Monitor) rpcBatchBlockByNumber(from, to uint64) (result []*types.Block,
 }
 
 func (m *Monitor) getRemainingSize(address string) (uint64, error) {
-	if size, suc := m.sizeCache.Get(address); suc && size.(uint64) == 0 {
-		return size.(uint64), nil
+	if size, suc := m.sizeCache.Get(address); suc && size == 0 {
+		return size, nil
 	}
 	var remainingSize hexutil.Uint64
 	rpcUploadMeter.Mark(1)
@@ -520,13 +521,6 @@ func (m *Monitor) parseBlockTorrentInfo(b *types.Block) (bool, error) {
 
 func (m *Monitor) exit() {
 	m.closeOnce.Do(func() {
-		/*if m.exitSyncCh != nil {
-			close(m.exitSyncCh)
-			m.exitSyncCh = nil
-		} else {
-			log.Warn("Listener sync has already been stopped")
-		}*/
-
 		if m.exitCh != nil {
 			close(m.exitCh)
 			m.wg.Wait()
@@ -540,7 +534,6 @@ func (m *Monitor) exit() {
 func (m *Monitor) Stop() error {
 	m.lock.Lock()
 	defer m.lock.Unlock()
-	//m.closeOnce.Do(func() {
 	if m.terminated.Swap(true) {
 		return nil
 	}
@@ -567,7 +560,6 @@ func (m *Monitor) Stop() error {
 	}
 	log.Info("Fs listener synchronizing closed")
 	return nil
-	//})
 }
 
 // Start ... start ListenOn on the rpc port of a blockchain full node
@@ -622,8 +614,8 @@ func (m *Monitor) run() error {
 	//if err := m.loadHistory(); err != nil {
 	//	return err
 	//}
-	//m.wg.Add(1)
-	//go m.taskLoop()
+	m.wg.Add(1)
+	go m.taskLoop()
 	//m.wg.Add(1)
 	//go m.listenLatestBlock()
 	m.wg.Add(1)
@@ -685,7 +677,6 @@ func (m *Monitor) syncLatestBlock() {
 							continue
 						}
 						m.fs.Flush()
-						//go m.exit()
 						elapsed := time.Duration(mclock.Now()) - time.Duration(m.start)
 						log.Debug("Finish sync, listener will be paused", "current", m.currentNumber.Load(), "elapsed", common.PrettyDuration(elapsed), "progress", progress, "end", end, "last", m.lastNumber.Load())
 						//return
@@ -789,8 +780,8 @@ func (m *Monitor) syncLastBlock() uint64 {
 			maxNumber = i - 1
 			break
 		}
-		if maxNumber > minNumber && (i-minNumber)%128 == 0 {
-			log.Debug("Running", "min", minNumber, "max", maxNumber, "cur", currentNumber, "last", m.lastNumber.Load(), "batch", batch, "i", i, "srv", m.srv.Load(), "size", maxNumber-minNumber, "progress", float64(i-minNumber)/float64(maxNumber-minNumber))
+		if maxNumber > minNumber && i%2048 == 0 {
+			log.Info("Running", "min", minNumber, "max", maxNumber, "cur", currentNumber, "last", m.lastNumber.Load(), "batch", batch, "i", i, "srv", m.srv.Load(), "size", maxNumber-minNumber, "progress", float64(i)/float64(currentNumber))
 		}
 		if m.ckp != nil && m.skip(i) {
 			//m.lastNumber = i - 1
@@ -808,25 +799,24 @@ func (m *Monitor) syncLastBlock() uint64 {
 
 			// batch blocks operation according service category
 			for _, rpcBlock := range blocks {
-				if err := m.solve(rpcBlock); err != nil {
-					log.Error("solve err", "err", err)
+				m.taskCh <- rpcBlock
+			}
+
+			for n := 0; n < len(blocks); n++ {
+				select {
+				case err := <-m.errCh:
+					if err != nil {
+						m.lastNumber.Store(i - 1)
+						log.Error("solve err", "err", err, "last", m.lastNumber.Load(), "i", i, "scope", m.scope, "min", minNumber, "max", maxNumber, "cur", currentNumber)
+						return 0
+					}
+				case <-m.exitCh:
 					m.lastNumber.Store(i - 1)
+					log.Info("Task checker quit")
 					return 0
 				}
-				i++
-				/*if len(m.taskCh) < cap(m.taskCh) {
-					m.taskCh <- rpcBlock
-					i++
-				} else {
-					m.lastNumber = i - 1
-					if maxNumber-minNumber > delay/2 {
-						elapsed := time.Duration(mclock.Now()) - time.Duration(start)
-						elapsedA := time.Duration(mclock.Now()) - time.Duration(m.start)
-						log.Warn("Chain segment frozen", "from", minNumber, "to", i, "range", uint64(i-minNumber), "current", uint64(m.currentNumber), "progress", float64(i)/float64(m.currentNumber), "last", m.lastNumber, "elapsed", common.PrettyDuration(elapsed), "bps", float64(i-minNumber)*1000*1000*1000/float64(elapsed), "bps_a", float64(maxNumber)*1000*1000*1000/float64(elapsedA), "cap", len(m.taskCh))
-					}
-					return 0
-				}*/
 			}
+			i += uint64(len(blocks))
 		} else {
 			rpcBlock, rpcErr := m.rpcBlockByNumber(i)
 			if rpcErr != nil {
@@ -840,24 +830,10 @@ func (m *Monitor) syncLastBlock() uint64 {
 				return 0
 			}
 			i++
-			/*if len(m.taskCh) < cap(m.taskCh) {
-				m.taskCh <- rpcBlock
-				i++
-			} else {
-				m.lastNumber = i - 1
-				if maxNumber-minNumber > delay/2 {
-					elapsed := time.Duration(mclock.Now()) - time.Duration(start)
-					elapsedA := time.Duration(mclock.Now()) - time.Duration(m.start)
-					log.Warn("Chain segment frozen", "from", minNumber, "to", i, "range", uint64(i-minNumber), "current", uint64(m.currentNumber), "progress", float64(i)/float64(m.currentNumber), "last", m.lastNumber, "elapsed", common.PrettyDuration(elapsed), "bps", float64(i-minNumber)*1000*1000*1000/float64(elapsed), "bps_a", float64(maxNumber)*1000*1000*1000/float64(elapsedA), "cap", len(m.taskCh))
-				}
-				return 0
-			}*/
 		}
 	}
 	log.Debug("Last number changed", "min", minNumber, "max", maxNumber, "cur", currentNumber, "last", m.lastNumber.Load(), "batch", batch)
 	m.lastNumber.Store(maxNumber)
-	//m.storeLastNumber(maxNumber)
-	//if maxNumber-minNumber > batch-1 {
 	if maxNumber-minNumber > delay {
 		elapsedA := time.Duration(mclock.Now()) - time.Duration(m.start)
 		log.Debug("Chain segment frozen", "from", minNumber, "to", maxNumber, "range", uint64(maxNumber-minNumber), "current", uint64(m.CurrentNumber()), "progress", float64(maxNumber)/float64(m.CurrentNumber()), "last", m.lastNumber.Load(), "bps", float64(maxNumber)*1000*1000*1000/float64(elapsedA), "elapsed", common.PrettyDuration(elapsedA))
@@ -951,7 +927,9 @@ func (m *Monitor) forExchangeService(block *types.Block) error {
 }
 
 func (m *Monitor) forRecordService(block *types.Block) error {
-	log.Debug("Block record", "num", block.Number, "hash", block.Hash, "txs", len(block.Txs), "last", m.lastNumber.Load())
+	if block.Number%4096 == 0 {
+		log.Info("Block record", "num", block.Number, "hash", block.Hash, "txs", len(block.Txs), "last", m.lastNumber.Load())
+	}
 	if len(block.Txs) > 0 {
 		for _, t := range block.Txs {
 			x := new(big.Float).Quo(new(big.Float).SetInt(t.Amount), new(big.Float).SetInt(big.NewInt(params1.Cortex)))