[actpool] persistent storage for actions

actpool/blobstore.go → actpool/actionstore.go

@@ -40,9 +40,9 @@ const (
 )
 
 type (
-	blobStore struct {
+	actionStore struct {
 		lifecycle.Readiness
-		config blobStoreConfig // Configuration for the blob store
+		config actionStoreConfig // Configuration for the blob store
 
 		store  billy.Database // Persistent data store for the tx
 		stored uint64         // Useful data size of all transactions on disk
@@ -53,7 +53,7 @@ type (
 		encode encodeAction // Encoder for the tx
 		decode decodeAction // Decoder for the tx
 	}
-	blobStoreConfig struct {
+	actionStoreConfig struct {
 		Datadir string `yaml:"datadir"` // Data directory containing the currently executable blobs
 		Datacap uint64 `yaml:"datacap"` // Soft-cap of database storage (hard cap is larger due to overhead)
 	}
@@ -68,24 +68,24 @@ var (
 	errStoreNotOpen = fmt.Errorf("blob store is not open")
 )
 
-var defaultBlobStoreConfig = blobStoreConfig{
-	Datadir: "blobpool",
+var defaultActionStoreConfig = actionStoreConfig{
+	Datadir: "actionstore",
 	Datacap: 10 * 1024 * 1024 * 1024,
 }
 
-func newBlobStore(cfg blobStoreConfig, encode encodeAction, decode decodeAction) (*blobStore, error) {
+func newBlobStore(cfg actionStoreConfig, encode encodeAction, decode decodeAction) (*actionStore, error) {
 	if len(cfg.Datadir) == 0 {
 		return nil, errors.New("datadir is empty")
 	}
-	return &blobStore{
+	return &actionStore{
 		config: cfg,
 		lookup: make(map[hash.Hash256]uint64),
 		encode: encode,
 		decode: decode,
 	}, nil
 }
 
-func (s *blobStore) Open(onData onAction) error {
+func (s *actionStore) Open(onData onAction) error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -131,7 +131,7 @@ func (s *blobStore) Open(onData onAction) error {
 	return s.TurnOn()
 }
 
-func (s *blobStore) Close() error {
+func (s *actionStore) Close() error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
@@ -141,7 +141,7 @@ func (s *blobStore) Close() error {
 	return s.store.Close()
 }
 
-func (s *blobStore) Get(hash hash.Hash256) (*action.SealedEnvelope, error) {
+func (s *actionStore) Get(hash hash.Hash256) (*action.SealedEnvelope, error) {
 	if !s.IsReady() {
 		return nil, errors.Wrap(errStoreNotOpen, "")
 	}
@@ -159,7 +159,7 @@ func (s *blobStore) Get(hash hash.Hash256) (*action.SealedEnvelope, error) {
 	return s.decode(blob)
 }
 
-func (s *blobStore) Put(act *action.SealedEnvelope) error {
+func (s *actionStore) Put(act *action.SealedEnvelope) error {
 	if !s.IsReady() {
 		return errors.Wrap(errStoreNotOpen, "")
 	}
@@ -189,7 +189,7 @@ func (s *blobStore) Put(act *action.SealedEnvelope) error {
 	return nil
 }
 
-func (s *blobStore) Delete(hash hash.Hash256) error {
+func (s *actionStore) Delete(hash hash.Hash256) error {
 	if !s.IsReady() {
 		return errors.Wrap(errStoreNotOpen, "")
 	}
@@ -208,7 +208,7 @@ func (s *blobStore) Delete(hash hash.Hash256) error {
 }
 
 // Range iterates over all stored with hashes
-func (s *blobStore) Range(fn func(hash.Hash256) bool) {
+func (s *actionStore) Range(fn func(hash.Hash256) bool) {
 	if !s.IsReady() {
 		return
 	}
@@ -222,7 +222,7 @@ func (s *blobStore) Range(fn func(hash.Hash256) bool) {
 	}
 }
 
-func (s *blobStore) drop() {
+func (s *actionStore) drop() {
 	h, ok := s.evict()
 	if !ok {
 		log.L().Debug("no worst action found")
@@ -241,7 +241,7 @@ func (s *blobStore) drop() {
 }
 
 // TODO: implement a proper eviction policy
-func (s *blobStore) evict() (hash.Hash256, bool) {
+func (s *actionStore) evict() (hash.Hash256, bool) {
 	for h := range s.lookup {
 		return h, true
 	}

actpool/blobstore_test.go → actpool/actionstore_test.go

@@ -21,7 +21,7 @@ import (
 
 func TestBlobStore(t *testing.T) {
 	r := require.New(t)
-	cfg := blobStoreConfig{
+	cfg := actionStoreConfig{
 		Datadir: t.TempDir(),
 		Datacap: 10 * 1024 * 1024,
 	}
@@ -83,7 +83,7 @@ func BenchmarkDatabase(b *testing.B) {
 	}
 
 	b.Run("billy-put", func(b *testing.B) {
-		cfg := blobStoreConfig{
+		cfg := actionStoreConfig{
 			Datadir: b.TempDir(),
 			Datacap: 1024,
 		}

actpool/actpool.go

@@ -11,6 +11,7 @@ import (
 	"sort"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
@@ -28,6 +29,7 @@ import (
 	"github.com/iotexproject/iotex-core/blockchain/genesis"
 	"github.com/iotexproject/iotex-core/pkg/log"
 	"github.com/iotexproject/iotex-core/pkg/prometheustimer"
+	"github.com/iotexproject/iotex-core/pkg/routine"
 	"github.com/iotexproject/iotex-core/pkg/tracer"
 )
 
@@ -103,6 +105,9 @@ type actPool struct {
 	senderBlackList          map[string]bool
 	jobQueue                 []chan workerJob
 	worker                   []*queueWorker
+
+	store     *actionStore           // store is the persistent cache for actpool
+	storeSync *routine.RecurringTask // storeSync is the recurring task to sync actions from store to memory
 }
 
 // NewActPool constructs a new actpool
@@ -142,7 +147,7 @@ func NewActPool(g genesis.Genesis, sf protocol.StateReader, cfg Config, opts ...
 		return nil, err
 	}
 	ap.timerFactory = timerFactory
-
+	// TODO: move to Start
 	for i := 0; i < _numWorker; i++ {
 		ap.jobQueue[i] = make(chan workerJob, ap.cfg.WorkerBufferSize)
 		ap.worker[i] = newQueueWorker(ap, ap.jobQueue[i])
@@ -153,6 +158,33 @@ func NewActPool(g genesis.Genesis, sf protocol.StateReader, cfg Config, opts ...
 	return ap, nil
 }
 
+func (ap *actPool) Start(ctx context.Context) error {
+	// open action store and load all actions
+	if ap.store != nil {
+		err := ap.store.Open(func(selp *action.SealedEnvelope) error {
+			return ap.add(ctx, selp)
+		})
+		if err != nil {
+			return err
+		}
+		return ap.storeSync.Start(ctx)
+	}
+	return nil
+}
+
+func (ap *actPool) Stop(ctx context.Context) error {
+	for i := 0; i < _numWorker; i++ {
+		if err := ap.worker[i].Stop(); err != nil {
+			return err
+		}
+	}
+	if ap.store != nil {
+		ap.storeSync.Stop(ctx)
+		return ap.store.Close()
+	}
+	return nil
+}
+
 func (ap *actPool) AddActionEnvelopeValidators(fs ...action.SealedEnvelopeValidator) {
 	ap.actionEnvelopeValidators = append(ap.actionEnvelopeValidators, fs...)
 }
@@ -217,6 +249,10 @@ func (ap *actPool) PendingActionMap() map[string][]*action.SealedEnvelope {
 }
 
 func (ap *actPool) Add(ctx context.Context, act *action.SealedEnvelope) error {
+	return ap.add(ctx, act)
+}
+
+func (ap *actPool) add(ctx context.Context, act *action.SealedEnvelope) error {
 	ctx, span := tracer.NewSpan(ap.context(ctx), "actPool.Add")
 	defer span.End()
 	ctx = ap.context(ctx)
@@ -345,6 +381,16 @@ func (ap *actPool) GetUnconfirmedActs(addrStr string) []*action.SealedEnvelope {
 func (ap *actPool) GetActionByHash(hash hash.Hash256) (*action.SealedEnvelope, error) {
 	act, ok := ap.allActions.Get(hash)
 	if !ok {
+		if ap.store != nil {
+			act, err := ap.store.Get(hash)
+			switch errors.Cause(err) {
+			case nil:
+				return act, nil
+			case errBlobNotFound:
+			default:
+				return nil, err
+			}
+		}
 		return nil, errors.Wrapf(action.ErrNotFound, "action hash %x does not exist in pool", hash)
 	}
 	return act.(*action.SealedEnvelope), nil
@@ -412,6 +458,21 @@ func (ap *actPool) validate(ctx context.Context, selp *action.SealedEnvelope) er
 	return nil
 }
 
+func (ap *actPool) removeReplacedActs(acts []*action.SealedEnvelope) {
+	for _, act := range acts {
+		hash, err := act.Hash()
+		if err != nil {
+			log.L().Debug("Skipping action due to hash error", zap.Error(err))
+			continue
+		}
+		log.L().Debug("Removed replaced action.", log.Hex("hash", hash[:]))
+		ap.allActions.Delete(hash)
+		intrinsicGas, _ := act.IntrinsicGas()
+		atomic.AddUint64(&ap.gasInPool, ^uint64(intrinsicGas-1))
+		ap.accountDesActs.delete(act)
+	}
+}
+
 func (ap *actPool) removeInvalidActs(acts []*action.SealedEnvelope) {
 	for _, act := range acts {
 		hash, err := act.Hash()
@@ -424,6 +485,11 @@ func (ap *actPool) removeInvalidActs(acts []*action.SealedEnvelope) {
 		intrinsicGas, _ := act.IntrinsicGas()
 		atomic.AddUint64(&ap.gasInPool, ^uint64(intrinsicGas-1))
 		ap.accountDesActs.delete(act)
+		if ap.store != nil {
+			if err = ap.store.Delete(hash); err != nil {
+				log.L().Warn("Failed to delete action from store", zap.Error(err), log.Hex("hash", hash[:]))
+			}
+		}
 	}
 }
 
@@ -461,6 +527,28 @@ func (ap *actPool) allocatedWorker(senderAddr address.Address) int {
 	return int(lastByte) % _numWorker
 }
 
+func (ap *actPool) syncFromStore() {
+	if ap.store == nil {
+		return
+	}
+	ap.store.Range(func(h hash.Hash256) bool {
+		if _, exist := ap.allActions.Get(h); !exist {
+			act, err := ap.store.Get(h)
+			if err != nil {
+				log.L().Warn("Failed to get action from store", zap.Error(err))
+				return true
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			if err := ap.add(ctx, act); err != nil {
+				log.L().Warn("Failed to add action to pool", zap.Error(err))
+				return true
+			}
+		}
+		return true
+	})
+}
+
 type destinationMap struct {
 	mu   sync.Mutex
 	acts map[string]map[hash.Hash256]*action.SealedEnvelope

actpool/config.go

@@ -37,6 +37,8 @@ type Config struct {
 	MinGasPriceStr string `yaml:"minGasPrice"`
 	// BlackList lists the account address that are banned from initiating actions
 	BlackList []string `yaml:"blackList"`
+	// Store defines the config for persistent cache
+	Store *StoreConfig `yaml:"store"`
 }
 
 // MinGasPrice returns the minimal gas price threshold
@@ -47,3 +49,14 @@ func (ap Config) MinGasPrice() *big.Int {
 	}
 	return mgp
 }
+
+// StoreConfig is the configuration for the blob store
+type StoreConfig struct {
+	Store        actionStoreConfig `yaml:"store"`
+	ReadInterval time.Duration     `yaml:"readInterval"` // Interval to read from store to actpool memory
+}
+
+var defaultStoreConfig = StoreConfig{
+	Store:        defaultActionStoreConfig,
+	ReadInterval: 10 * time.Minute,
+}

actpool/options.go

@@ -6,9 +6,12 @@
 package actpool
 
 import (
+	"errors"
 	"time"
 
 	"github.com/facebookgo/clock"
+
+	"github.com/iotexproject/iotex-core/pkg/routine"
 )
 
 // ActQueueOption is the option for actQueue.
@@ -33,3 +36,21 @@ func WithTimeOut(ttl time.Duration) interface{ ActQueueOption } {
 }
 
 func (o *ttlOption) SetActQueueOption(aq *actQueue) { aq.ttl = o.ttl }
+
+// WithStore is the option to set store encode and decode functions.
+func WithStore(cfg StoreConfig, encode encodeAction, decode decodeAction) func(*actPool) error {
+	return func(a *actPool) error {
+		if encode == nil || decode == nil {
+			return errors.New("encode and decode functions must be provided")
+		}
+		store, err := newBlobStore(cfg.Store, encode, decode)
+		if err != nil {
+			return err
+		}
+		a.store = store
+		a.storeSync = routine.NewRecurringTask(func() {
+			a.syncFromStore()
+		}, cfg.ReadInterval)
+		return nil
+	}
+}

actpool/queueworker.go

@@ -105,6 +105,12 @@ func (worker *queueWorker) Handle(job workerJob) error {
 	}
 
 	worker.ap.allActions.Set(actHash, act)
+	isBlobTx := false // TODO: only store blob tx
+	if worker.ap.store != nil && isBlobTx {
+		if err := worker.ap.store.Put(act); err != nil {
+			log.L().Warn("failed to store action", zap.Error(err), log.Hex("hash", actHash[:]))
+		}
+	}
 
 	if desAddress, ok := act.Destination(); ok && !strings.EqualFold(sender, desAddress) {
 		if err := worker.ap.accountDesActs.addAction(act); err != nil {
@@ -123,7 +129,7 @@ func (worker *queueWorker) Handle(job workerJob) error {
 			log.L().Warn("UNEXPECTED ERROR: action pool is full, but no action to drop")
 			return nil
 		}
-		worker.ap.removeInvalidActs([]*action.SealedEnvelope{actToReplace})
+		worker.ap.removeReplacedActs([]*action.SealedEnvelope{actToReplace})
 		if actToReplace.SenderAddress().String() == sender && actToReplace.Nonce() == nonce {
 			err = action.ErrTxPoolOverflow
 			_actpoolMtc.WithLabelValues("overMaxNumActsPerPool").Inc()