Multithreading queue

config/types.go

@@ -24,6 +24,7 @@ type ChainConfig struct {
 
 type Config struct {
 	QueueInterval    int64              `yaml:"queue_interval" mapstructure:"queue_interval"`
+	QueueThreads     int8               `yaml:"queue_threads" mapstructure:"queue_threads"`
 	ProofInterval    int64              `yaml:"proof_interval" mapstructure:"proof_interval"`
 	StrayManagerCfg  StrayManagerConfig `yaml:"stray_manager" mapstructure:"stray_manager"`
 	ChainCfg         ChainConfig        `yaml:"chain_config" mapstructure:"chain_config"`
@@ -39,6 +40,10 @@ func DefaultQueueInterval() int64 {
 	return 10
 }
 
+func DefaultQueueThreads() int8 {
+	return 5
+}
+
 func DefaultProofInterval() int64 {
 	return 120
 }
@@ -143,6 +148,7 @@ func DefaultChainConfig() ChainConfig {
 func DefaultConfig() *Config {
 	return &Config{
 		QueueInterval:    DefaultQueueInterval(),
+		QueueThreads:     DefaultQueueThreads(),
 		ProofInterval:    DefaultProofInterval(),
 		StrayManagerCfg:  DefaultStrayManagerConfig(),
 		ChainCfg:         DefaultChainConfig(),

core/app.go

@@ -249,7 +249,13 @@ func (a *App) Start() error {
 		return err
 	}
 
-	a.q = queue.NewQueue(a.wallet, cfg.QueueInterval)
+	refreshInterval := time.Second * time.Duration(cfg.QueueInterval)
+	a.q, err = queue.NewQueue(a.wallet, refreshInterval, cfg.QueueThreads)
+	if err != nil {
+		log.Error().Err(err).Msg("failed to initialize Queue module")
+		return err
+	}
+
 	go a.q.Listen()
 
 	prover := proofs.NewProver(a.wallet, a.q, a.fileSystem, cfg.ProofInterval, cfg.ProofThreads, int(params.ChunkSize))

proofs/proofs.go

@@ -32,7 +32,7 @@ const (
 )
 
 func GenerateMerkleProof(tree *merkletree.MerkleTree, index int, item []byte) (bool, *merkletree.Proof, error) {
-	log.Debug().Msg(fmt.Sprintf("Generating Merkle proof for %d", index))
+	log.Debug().Msg(fmt.Sprintf("Generating Merkle proof for index: %d", index))
 
 	h := sha256.New()
 	_, err := fmt.Fprintf(h, "%d%x", index, item)
@@ -56,15 +56,15 @@ func GenerateMerkleProof(tree *merkletree.MerkleTree, index int, item []byte) (b
 //
 // returns proof, item and error
 func GenProof(io FileSystem, merkle []byte, owner string, start int64, block int, chunkSize int, proofType int64) ([]byte, []byte, error) {
+	log.Debug().Msg(fmt.Sprintf("About to generate merkle proof for file: %x", merkle))
+
 	tree, chunk, err := io.GetFileTreeByChunk(merkle, owner, start, block, chunkSize, proofType)
 	if err != nil {
 		e := fmt.Errorf("cannot get chunk for %x at %d | %w", merkle, block, err)
 		log.Error().Err(e)
 		return nil, nil, e
 	}
 
-	log.Debug().Msg(fmt.Sprintf("About to generate merkle proof for %x", merkle))
-
 	valid, proof, err := GenerateMerkleProof(tree, block, chunk)
 	if err != nil {
 		return nil, nil, err
@@ -220,11 +220,12 @@ func (p *Prover) Start() {
 		}
 
 		time.Sleep(time.Millisecond * 1000) // pauses for one third of a second
+		// sleep until next proving cycle
 		if !p.processed.Add(time.Second * time.Duration(p.interval)).Before(time.Now()) {
 			continue
 		}
 
-		log.Debug().Msg("Starting proof cycle...")
+		log.Debug().Time("start at", p.processed).Msg("Starting proof cycle...")
 
 		abciInfo, err := p.wallet.Client.RPCClient.ABCIInfo(context.Background())
 		if err != nil {
@@ -241,15 +242,16 @@ func (p *Prover) Start() {
 
 				time.Sleep(time.Second * 5)
 			}
-			log.Debug().Msg(fmt.Sprintf("proving: %x", merkle))
+			log.Debug().Hex("merkle", merkle).Str("owner", owner).Int64("start", start).Msg("proving file")
 			filesProving.Inc()
 			p.Inc()
 			go p.wrapPostProof(merkle, owner, start, height, t)
 		})
 		if err != nil {
-			log.Error().Err(err)
+			log.Error().Err(err).Msg("something went wrong while processing files in this proving cycle")
 		}
 
+		log.Debug().Time("finish at", time.Now()).TimeDiff("duration", time.Now(), p.processed).Msg("End of proof cycle")
 		p.processed = time.Now()
 	}
 	log.Info().Msg("Prover module stopped")

queue/queue.go

@@ -1,29 +1,54 @@
 package queue
 
 import (
+	"errors"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/cosmos/cosmos-sdk/types"
-	walletTypes "github.com/desmos-labs/cosmos-go-wallet/types"
 	"github.com/desmos-labs/cosmos-go-wallet/wallet"
 	"github.com/rs/zerolog/log"
 )
 
-func (m *Message) Done() {
-	m.wg.Done()
-}
+// does it really need to sleep less than 0.1 second?
+const minSleepDuration = time.Millisecond * 500
+
+// There must be at least 1 thread
+// For every thread, TxWorker is assigned its own wallet with an offset
+// Each TxWorker uses its wallet to sign and broadcast messages
+// Queue sleeps for refreshInterval after Queue distributes messages to the workers
+func NewQueue(w *wallet.Wallet, refreshInterval time.Duration, thread int8) (*Queue, error) {
+	if thread < 1 {
+		return nil, errors.New("thread must be at least 1")
+	}
+
+	if refreshInterval < time.Second {
+		return nil, errors.New("refresh interval cannot be shorter than 1 second")
+	}
 
-func NewQueue(w *wallet.Wallet, interval int64) *Queue {
 	q := &Queue{
-		wallet:    w,
-		messages:  make([]*Message, 0),
-		processed: time.Now(),
-		running:   false,
-		interval:  interval,
+		wallet:          w,
+		txWorkers:       make([]*TxWorker, 0),
+		msgPool:         make([]*Message, 0),
+		refreshInterval: refreshInterval,
+		processed:       time.Now(),
+		running:         false,
+	}
+
+	for i := 0; i < int(thread); i++ {
+		offset := byte(i + 1)
+		w, err := w.CloneWalletOffset(offset)
+		if err != nil {
+			// should be debugged if this happens
+			return nil, errors.New("failed to create offset wallet for txWorker")
+		}
+
+		worker := NewTxWorker(int8(i), w)
+		q.txWorkers = append(q.txWorkers, worker)
 	}
-	return q
+
+	return q, nil
 }
 
 func (q *Queue) Add(msg types.Msg) (*Message, *sync.WaitGroup) {
@@ -37,71 +62,77 @@ func (q *Queue) Add(msg types.Msg) (*Message, *sync.WaitGroup) {
 		err: nil,
 	}
 
-	q.messages = append(q.messages, m) // adding the message to the end of the list
+	q.addLast(m)
 
 	return m, &wg
 }
 
-func (q *Queue) Stop() {
-	q.running = false
-}
-
 func (q *Queue) Listen() {
 	q.running = true
 	defer log.Info().Msg("Queue module stopped")
 
 	log.Info().Msg("Queue module started")
-	for q.running {
-		time.Sleep(time.Millisecond * 333)                                                // pauses for one third of a second
-		if !q.processed.Add(time.Second * time.Duration(q.interval)).Before(time.Now()) { // check every ten seconds
-			continue
-		}
 
-		lmsg := len(q.messages)
+	for q.running {
+		q.sleep()
 
-		if lmsg == 0 { // skipping this queue cycle if there is no messages to be pushed
+		if len(q.msgPool) == 0 {
 			continue
 		}
 
-		log.Info().Msg(fmt.Sprintf("Queue: %d messages waiting to be put on-chain...", lmsg))
-
-		// maxSize := 1024 * 1024 // 1mb
-		maxSize := 45
-
-		total := len(q.messages)
-		queueSize.Set(float64(total))
+		before := len(q.msgPool)
+		count := q.distributeMsg()
+		log.Info().Msg(fmt.Sprintf("assigned %d out of %d messages", count, before))
+	}
+}
 
-		if total > maxSize {
-			total = maxSize
-		}
+func (q *Queue) Stop() {
+	q.running = false
+}
 
-		log.Info().Msg(fmt.Sprintf("Queue: Posting %d messages to chain...", total))
+// fcfs
+func (q *Queue) popFront() *Message {
+	if len(q.msgPool) == 0 {
+		return nil
+	}
 
-		toProcess := q.messages[:total]
-		q.messages = q.messages[total:]
+	msg := q.msgPool[0]
+	q.msgPool = q.msgPool[1:]
+	return msg
+}
 
-		allMsgs := make([]types.Msg, len(toProcess))
+// assign message to free workers from msg pool
+// returns number of messages assigned
+func (q *Queue) distributeMsg() int {
+	if len(q.msgPool) == 0 {
+		return 0
+	}
 
-		for i, process := range toProcess {
-			allMsgs[i] = process.msg
+	count := 0
+	for _, w := range q.txWorkers {
+		if w.busy() {
+			continue
 		}
 
-		data := walletTypes.NewTransactionData(
-			allMsgs...,
-		).WithGasAuto().WithFeeAuto()
+		w.assign(q.popFront())
+		go w.broadCast()
+		count++
+	}
+	return count
+}
 
-		res, err := q.wallet.BroadcastTxCommit(data)
-		if err != nil {
-			log.Warn().Err(err).Msg("tx broadcast failed from queue")
-		}
+// sleep until next interval
+// if slept duration is too small consider increasing interval or threads
+func (q *Queue) sleep() (slept time.Duration) {
+	wakeUpTime := q.processed.Add(q.refreshInterval)
+	sleep := wakeUpTime.Sub(time.Now())
+	if sleep > minSleepDuration {
+		time.Sleep(sleep)
+	}
 
-		for i, process := range toProcess {
-			process.err = err
-			process.res = res
-			process.msgIndex = i
-			process.Done()
-		}
+	return sleep
+}
 
-		q.processed = time.Now()
-	}
+func (q *Queue) addLast(msg *Message) {
+	q.msgPool = append(q.msgPool, msg)
 }

queue/types.go

@@ -8,12 +8,19 @@ import (
 	"github.com/desmos-labs/cosmos-go-wallet/wallet"
 )
 
+type TxWorker struct {
+	id     int8
+	wallet *wallet.Wallet
+	msg    *Message
+}
+
 type Queue struct {
-	wallet    *wallet.Wallet
-	messages  []*Message
-	processed time.Time
-	running   bool
-	interval  int64
+	txWorkers       []*TxWorker
+	wallet          *wallet.Wallet
+	refreshInterval time.Duration
+	msgPool         []*Message
+	processed       time.Time
+	running         bool
 }
 
 type Message struct {

queue/worker.go

@@ -0,0 +1,84 @@
+package queue
+
+import (
+	walletTypes "github.com/desmos-labs/cosmos-go-wallet/types"
+	"github.com/desmos-labs/cosmos-go-wallet/wallet"
+
+	"github.com/cosmos/cosmos-sdk/types"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func NewTxWorker(id int8, bucketSize int16, msgBatchSize int8, retryAttempt int8, offsetWallet *wallet.Wallet) *TxWorker {
+	return &TxWorker{
+		id:            id,
+		wallet:        offsetWallet,
+		msgBucketSize: bucketSize,
+		msgBucket:     make([]*Message, 0, bucketSize),
+		msgBatchSize:  msgBatchSize,
+		retryAttempt:  retryAttempt,
+	}
+}
+
+func (t *TxWorker) Address() string {
+	return t.wallet.AccAddress()
+}
+
+func (t *TxWorker) available() int {
+	return int(t.msgBucketSize) - len(t.msgBucket)
+}
+
+func (t *TxWorker) assign(msgs []*Message) int {
+	if msgs == nil {
+		return 0
+	}
+
+	// take whatever it can to fill the bucket
+	// required for the sanity check
+	fillCount := min(int(t.msgBucketSize)-len(t.msgBucket), len(msgs))
+	m := msgs[:fillCount]
+
+	t.msgBucket = append(t.msgBucket, m...)
+	return fillCount
+}
+
+func (t *TxWorker) grabNextBatch() []*Message {
+	total := min(len(t.msgBucket), int(t.msgBatchSize))
+	msgs := t.msgBucket[:total]
+	t.msgBucket = t.msgBucket[total:]
+
+	return msgs
+}
+
+func (t *TxWorker) broadCast() {
+	batchMessage := t.grabNextBatch()
+
+	batch := make([]types.Msg, len(batchMessage))
+	for i, m := range batchMessage {
+		batch[i] = m.msg
+	}
+	data := walletTypes.NewTransactionData(batch...).WithGasAuto().WithFeeAuto()
+
+	var resp *types.TxResponse
+	var err error
+	for attempt := 0; attempt < int(t.retryAttempt); attempt++ {
+		resp, err = t.wallet.BroadcastTxCommit(data)
+		if err != nil {
+			// retry if network is not responding
+			if code := status.Code(err); code == codes.DeadlineExceeded {
+				err = nil
+			} else {
+				break
+			}
+		} else {
+			break
+		}
+
+	}
+	for _, m := range batchMessage {
+		m.wg.Done()
+		m.res = resp
+		m.err = err
+	}
+}