Multithreading queue

config/types.go

@@ -23,8 +23,8 @@ type ChainConfig struct {
 }
 
 type Config struct {
-	QueueInterval    int64              `yaml:"queue_interval" mapstructure:"queue_interval"`
 	ProofInterval    int64              `yaml:"proof_interval" mapstructure:"proof_interval"`
+	QueueConfig      QueueConfig        `yaml:"queue_config" mapstructure:"queue_config"`
 	StrayManagerCfg  StrayManagerConfig `yaml:"stray_manager" mapstructure:"stray_manager"`
 	ChainCfg         ChainConfig        `yaml:"chain_config" mapstructure:"chain_config"`
 	Ip               string             `yaml:"domain" mapstructure:"domain"`
@@ -35,10 +35,6 @@ type Config struct {
 	BlockStoreConfig BlockStoreConfig   `yaml:"block_store_config" mapstructure:"block_store_config"`
 }
 
-func DefaultQueueInterval() int64 {
-	return 10
-}
-
 func DefaultProofInterval() int64 {
 	return 120
 }
@@ -140,10 +136,32 @@ func DefaultChainConfig() ChainConfig {
 	}
 }
 
+type QueueConfig struct {
+	// seconds
+	QueueInterval int64 `yaml:"queue_interval" mapstructure:"queue_interval"`
+	QueueThreads  int8  `yaml:"queue_threads" mapstructure:"queue_threads"`
+	// resend tx if network isn't responding
+	MaxRetryAttempt int8 `yaml:"max_retry_attempt" mapstructure:"max_retry_attempt"`
+	// group individual messages into one tx
+	TxBatchSize int8 `yaml:"tx_batch_size" mapstructure:"tx_batch_size"`
+	// worker's own message pool, set this value > TxBatchSize
+	WorkerQueueSize int16 `yaml:"worker_queue_size" mapstructure:"worker_queue_size"`
+}
+
+func DefaultQueueConfig() QueueConfig {
+	return QueueConfig{
+		QueueInterval:   10,
+		QueueThreads:    5,
+		MaxRetryAttempt: 100,
+		TxBatchSize:     45,
+		WorkerQueueSize: 100,
+	}
+}
+
 func DefaultConfig() *Config {
 	return &Config{
-		QueueInterval:    DefaultQueueInterval(),
 		ProofInterval:    DefaultProofInterval(),
+		QueueConfig:      DefaultQueueConfig(),
 		StrayManagerCfg:  DefaultStrayManagerConfig(),
 		ChainCfg:         DefaultChainConfig(),
 		Ip:               DefaultIP(),
@@ -156,8 +174,7 @@ func DefaultConfig() *Config {
 }
 
 func (c Config) MarshalZerologObject(e *zerolog.Event) {
-	e.Int64("QueueInterval", c.QueueInterval).
-		Int64("ProofInterval", c.ProofInterval).
+	e.Int64("ProofInterval", c.ProofInterval).
 		Int64("StrayCheckInterval", c.StrayManagerCfg.CheckInterval).
 		Int64("StrayRefreshInterval", c.StrayManagerCfg.RefreshInterval).
 		Int("StrayHandCount", c.StrayManagerCfg.HandCount).
@@ -176,7 +193,7 @@ func (c Config) MarshalZerologObject(e *zerolog.Event) {
 }
 
 func init() {
-	viper.SetDefault("QueueInterval", DefaultQueueInterval())
+	viper.SetDefault("QueueConfig", DefaultQueueConfig())
 	viper.SetDefault("ProofInterval", DefaultProofInterval())
 	viper.SetDefault("StrayManagerCfg", DefaultStrayManagerConfig())
 	viper.SetDefault("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")

proofs/types.go

@@ -12,7 +12,7 @@ import (
 type Prover struct {
 	running        bool
 	wallet         *wallet.Wallet
-	q              *queue.Queue
+	q              queue.Queue
 	processed      time.Time
 	interval       int64
 	io             FileSystem

queue/pool.go

@@ -0,0 +1,182 @@
+package queue
+
+import (
+	"context"
+	"errors"
+	"reflect"
+	"slices"
+	"sync"
+
+	"github.com/JackalLabs/sequoia/config"
+	"github.com/cosmos/cosmos-sdk/types"
+
+	walletTypes "github.com/desmos-labs/cosmos-go-wallet/types"
+	"github.com/desmos-labs/cosmos-go-wallet/wallet"
+
+	storageTypes "github.com/jackalLabs/canine-chain/v4/x/storage/types"
+
+	"github.com/rs/zerolog/log"
+)
+
+var _ Queue = &Pool{}
+
+type Pool struct {
+	workers        []*worker
+	workerChannels []chan *Message // worker id should correspond to index of this
+	wallet         *wallet.Wallet
+}
+
+func NewPool(wallet *wallet.Wallet, config config.QueueConfig) (*Pool, error) {
+	workerWallets, err := initAuthClaimers(wallet, config.QueueThreads)
+	if err != nil {
+		return nil, errors.Join(errors.New("failed to initialize auth claimers"), err)
+	}
+
+	workers, workerChannels := createWorkers(workerWallets, config.MaxRetryAttempt)
+	if workers == nil {
+		panic("no workers created")
+	}
+	if workerChannels == nil {
+		panic("no worker channels created")
+	}
+	if len(workerChannels) != len(workers) {
+		panic("size of workers does not match size of worker channels")
+	}
+
+	pool := &Pool{
+		wallet:         wallet,
+		workers:        workers,
+		workerChannels: workerChannels,
+	}
+
+	return pool, nil
+}
+
+func (p *Pool) Stop() {
+	for _, c := range p.workerChannels {
+		close(c)
+	}
+}
+
+func (p *Pool) Listen() {
+	for _, w := range p.workers {
+		go w.start()
+	}
+}
+
+func (p *Pool) Add(msg types.Msg) (*Message, *sync.WaitGroup) {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	m := &Message{
+		msg:      msg,
+		wg:       &wg,
+		err:      nil,
+		res:      nil,
+		msgIndex: -1, // no longer relevant(?)
+	}
+
+	_ = p.sendToAny(m)
+
+	return m, &wg
+}
+
+func (p *Pool) sendToAny(msg *Message) (workerId int) {
+	set := make([]reflect.SelectCase, 0, len(p.workerChannels))
+	for _, ch := range p.workerChannels {
+		set = append(set, reflect.SelectCase{
+			Dir:  reflect.SelectSend,
+			Chan: reflect.ValueOf(ch),
+			Send: reflect.ValueOf(msg),
+		})
+	}
+
+	// blocks until a worker is free
+	to, _, _ := reflect.Select(set)
+	return to
+}
+
+func createWorkers(workerWallets []*wallet.Wallet, maxRetryAttempt int8) ([]*worker, []chan *Message) {
+	wChannels := make([]chan *Message, 0, len(workerWallets))
+	for _ = range len(workerWallets) {
+		wChannels = append(wChannels, make(chan *Message))
+	}
+
+	workers := make([]*worker, 0, len(workerWallets))
+	for i, w := range workerWallets {
+		worker := newWorker(int8(i), w, maxRetryAttempt, wChannels[i])
+		workers = append(workers, worker)
+	}
+
+	return workers, wChannels
+}
+
+func initAuthClaimers(wallet *wallet.Wallet, count int8) (workerWallets []*wallet.Wallet, err error) {
+	query := &storageTypes.QueryProvider{
+		Address: wallet.AccAddress(),
+	}
+
+	cl := storageTypes.NewQueryClient(wallet.Client.GRPCConn)
+	res, err := cl.Provider(context.Background(), query)
+	if err != nil {
+		return nil, errors.Join(errors.New("unable to query provider auth claimers"), err)
+	}
+	claimers := res.Provider.AuthClaimers
+
+	for i := range count {
+		workerWallet := newOffsetWallet(wallet, int(i))
+		if !slices.Contains(claimers, workerWallet.AccAddress()) {
+			err := addClaimer(wallet, workerWallet)
+			if err != nil {
+				return nil, errors.Join(errors.New("failed to add claimer on chain"), err)
+			}
+		}
+		workerWallets = append(workerWallets, workerWallet)
+	}
+
+	return workerWallets, nil
+}
+
+func addClaimer(main *wallet.Wallet, claimer *wallet.Wallet) error {
+	msg := storageTypes.NewMsgAddClaimer(main.AccAddress(), claimer.AccAddress())
+	txData := walletTypes.NewTransactionData(msg).WithFeeAuto().WithGasAuto()
+
+	res, err := main.BroadcastTxCommit(txData)
+	if err != nil {
+		return errors.Join(errors.New("unable to broadcast MsgAddClaimer"), err)
+	}
+
+	log.Info().Msg(res.TxHash)
+
+	return nil
+}
+
+func newOffsetWallet(main *wallet.Wallet, index int) *wallet.Wallet {
+	w, err := main.CloneWalletOffset(byte(index + 1))
+	if err != nil {
+		panic(err)
+	}
+	return w
+}
+
+func newBuffer(in <-chan *Message) <-chan *Message {
+	// From: https://blogtitle.github.io/go-advanced-concurrency-patterns-part-4-unlimited-buffer-channels/
+	var buf []*Message
+	out := make(chan *Message)
+
+	go func() {
+		defer close(out)
+		for msg := range in {
+			select {
+			case out <- msg:
+			default:
+				buf = append(buf, msg)
+			}
+		}
+
+		for _, v := range buf {
+			out <- v
+		}
+	}()
+
+	return out
+}