Introduce single power table cache (#1007)

* Vendor singleflight

Signed-off-by: Jakub Sztandera <[email protected]>

* Update singleflight to Generics

Signed-off-by: Jakub Sztandera <[email protected]>

* Introduce centralised power table cache

Signed-off-by: Jakub Sztandera <[email protected]>

* Include full BSD License

Signed-off-by: Jakub Sztandera <[email protected]>

* use standard singleflight package

Signed-off-by: Jakub Sztandera <[email protected]>

---------

Signed-off-by: Jakub Sztandera <[email protected]>

Author: Kubuxu

ec/caching.go

@@ -0,0 +1,68 @@
+package ec
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/filecoin-project/go-f3/gpbft"
+	lru "github.com/hashicorp/golang-lru/v2"
+	"golang.org/x/sync/singleflight"
+)
+
+type PowerCachingECWrapper struct {
+	Backend
+
+	cache *lru.Cache[string, gpbft.PowerEntries]
+
+	smaphore chan struct{}
+	dedup    singleflight.Group
+}
+
+func NewPowerCachingECWrapper(backend Backend, concurrency int, cacheSize int) *PowerCachingECWrapper {
+	cache, err := lru.New[string, gpbft.PowerEntries](cacheSize)
+	if err != nil {
+		panic(err)
+	}
+	smaphore := make(chan struct{}, concurrency)
+
+	return &PowerCachingECWrapper{
+		Backend:  backend,
+		cache:    cache,
+		smaphore: smaphore,
+	}
+}
+
+func (p *PowerCachingECWrapper) GetPowerTable(ctx context.Context, tsk gpbft.TipSetKey) (gpbft.PowerEntries, error) {
+	entry, ok := p.cache.Get(string(tsk))
+	if ok {
+		return entry, nil
+	}
+
+	ch := p.dedup.DoChan(string(tsk),
+		// break context cancellation chain as the dedup group might start with short context and then get called with longer one
+		func() (any, error) { return p.executeGetPowerTable(context.WithoutCancel(ctx), tsk) })
+
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case res := <-ch:
+		if res.Err != nil {
+			return nil, fmt.Errorf("getting power table: %w", res.Err)
+		}
+		return res.Val.(gpbft.PowerEntries), nil
+	}
+}
+
+func (p *PowerCachingECWrapper) executeGetPowerTable(ctx context.Context, tsk gpbft.TipSetKey) (gpbft.PowerEntries, error) {
+	// take semaphore
+	p.smaphore <- struct{}{}
+	defer func() { <-p.smaphore }()
+
+	res, err := p.Backend.GetPowerTable(ctx, tsk)
+	if err != nil {
+		return nil, fmt.Errorf("getting power table: %w", err)
+	}
+
+	p.cache.Add(string(tsk), res)
+	return res, nil
+}

f3.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"path/filepath"
+	"runtime"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -62,16 +63,19 @@ type F3 struct {
 // New creates and setups f3 with libp2p
 // The context is used for initialization not runtime.
 func New(_ctx context.Context, manifest manifest.Manifest, ds datastore.Datastore, h host.Host,
-	ps *pubsub.PubSub, verif gpbft.Verifier, ec ec.Backend, diskPath string) (*F3, error) {
+	ps *pubsub.PubSub, verif gpbft.Verifier, ecBackend ec.Backend, diskPath string) (*F3, error) {
 	runningCtx, cancel := context.WithCancel(context.WithoutCancel(_ctx))
+
+	// concurrency is limited to half of the number of CPUs, and cache size is set to 256 which is more than 2x max ECChain size
+	ecBackend = ec.NewPowerCachingECWrapper(ecBackend, max(runtime.NumCPU()/2, 8), 256)
 	return &F3{
 		verifier:         verif,
 		mfst:             manifest,
 		diskPath:         diskPath,
 		outboundMessages: make(chan *gpbft.MessageBuilder, 128),
 		host:             h,
 		ds:               ds,
-		ec:               ec,
+		ec:               ecBackend,
 		pubsub:           ps,
 		clock:            clock.GetClock(runningCtx),
 		runningCtx:       runningCtx,

go.mod

@@ -32,7 +32,7 @@ require (
 	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.36.0
 	golang.org/x/exp v0.0.0-20241009180824-f66d83c29e7c
-	golang.org/x/sync v0.12.0
+	golang.org/x/sync v0.15.0
 	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028
 )
 

go.sum

@@ -616,8 +616,8 @@ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.12.0 h1:MHc5BpPuC30uJk597Ri8TV3CNZcTLu6B6z4lJy+g6Jw=
-golang.org/x/sync v0.12.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sync v0.15.0 h1:KWH3jNZsfyT6xfAfKiz6MRNmd46ByHDYaZ7KSkCtdW8=
+golang.org/x/sync v0.15.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
 golang.org/x/sys v0.0.0-20180810173357-98c5dad5d1a0/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=