Merge pull request #154 from anyproto/GO-5827-speedup-deletion

GO-5827 make cidEntriesByString parallel, rename del key, remove old recheck …

Author: cheggaaa

index/cidentry.go

@@ -2,6 +2,7 @@ package index
 
 import (
 	"context"
+	"sync"
 	"time"
 
 	"github.com/ipfs/go-cid"
@@ -12,6 +13,7 @@ import (
 
 type CidEntries struct {
 	entries []*cidEntry
+	mu      sync.Mutex
 }
 
 func (ce *CidEntries) Release() {
@@ -23,6 +25,12 @@ func (ce *CidEntries) Release() {
 	return
 }
 
+func (ce *CidEntries) Add(entry *cidEntry) {
+	ce.mu.Lock()
+	defer ce.mu.Unlock()
+	ce.entries = append(ce.entries, entry)
+}
+
 type cidEntry struct {
 	Cid     cid.Cid
 	release func()

index/cids.go

@@ -9,6 +9,7 @@ import (
 	"github.com/ipfs/go-cid"
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/anyproto/any-sync-filenode/index/indexproto"
 )
@@ -33,18 +34,43 @@ func (ri *redisIndex) CidEntries(ctx context.Context, cids []cid.Cid) (entries *
 	return entries, nil
 }
 
-func (ri *redisIndex) CidEntriesByString(ctx context.Context, cids []string) (entries *CidEntries, err error) {
-	entries = &CidEntries{}
-	var c cid.Cid
-	for _, cs := range cids {
-		c, err = cid.Decode(cs)
-		if err != nil {
-			return
-		}
-		if err = ri.getAndAddToEntries(ctx, entries, c); err != nil {
-			entries.Release()
-			return nil, err
+func (ri *redisIndex) CidEntriesByString(ctx context.Context, cids []string) (*CidEntries, error) {
+	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(10)
+
+	results := make([]*cidEntry, len(cids))
+
+	for i, cs := range cids {
+		i, cs := i, cs
+		g.Go(func() error {
+			// decode
+			c, err := cid.Decode(cs)
+			if err != nil {
+				return err
+			}
+			// fetch entry
+			entry, err := ri.acquireCidEntry(ctx, c)
+			if err != nil {
+				return err
+			}
+			// store in the correct slot
+			results[i] = entry
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		for _, result := range results {
+			if result != nil {
+				result.release()
+			}
 		}
+		return nil, err
+	}
+
+	entries := &CidEntries{}
+	for _, e := range results {
+		entries.Add(e)
 	}
 	return entries, nil
 }
@@ -66,22 +92,29 @@ func (ri *redisIndex) CidEntriesByBlocks(ctx context.Context, bs []blocks.Block)
 }
 
 func (ri *redisIndex) getAndAddToEntries(ctx context.Context, entries *CidEntries, c cid.Cid) (err error) {
-	_, release, err := ri.AcquireKey(ctx, CidKey(c))
+	entry, err := ri.acquireCidEntry(ctx, c)
+	if err != nil {
+		return err
+	}
+	entries.Add(entry)
+	return
+}
+
+func (ri *redisIndex) acquireCidEntry(ctx context.Context, c cid.Cid) (entry *cidEntry, err error) {
+	ok, release, err := ri.AcquireKey(ctx, CidKey(c))
 	if err != nil {
 		return
 	}
-	//temporarily ignore the exists check to make a deep check
-	/*if !ok {
+	if !ok {
 		release()
-		return ErrCidsNotExist
-	}*/
-	entry, err := ri.getCidEntry(ctx, c)
+		return nil, ErrCidsNotExist
+	}
+	entry, err = ri.getCidEntry(ctx, c)
 	if err != nil {
 		release()
-		return err
+		return nil, err
 	}
 	entry.release = release
-	entries.entries = append(entries.entries, entry)
 	return
 }
 
@@ -134,15 +167,7 @@ func (ri *redisIndex) getCidEntry(ctx context.Context, c cid.Cid) (entry *cidEnt
 	cidData, err := ri.cl.Get(ctx, ck).Result()
 	if err != nil {
 		if errors.Is(err, redis.Nil) {
-			// temporary additional check: try to load data from store and restore cid
-			var b blocks.Block
-			if b, err = ri.persistStore.Get(ctx, c); err != nil {
-				log.WarnCtx(ctx, "restore cid entry error", zap.String("cid", c.String()), zap.Error(err))
-				err = ErrCidsNotExist
-				return
-			}
-			log.InfoCtx(ctx, "restore cid entry", zap.String("cid", c.String()))
-			return ri.createCidEntry(ctx, b)
+			return nil, ErrCidsNotExist
 		}
 		return
 	}

index/cids_test.go

@@ -90,23 +90,6 @@ func TestRedisIndex_CidEntries(t *testing.T) {
 			assert.NotEmpty(t, e.Version)
 		}
 	})
-	t.Run("restore from store", func(t *testing.T) {
-		bs := testutil.NewRandBlocks(4)
-		fx := newFixture(t)
-		defer fx.Finish(t)
-
-		require.NoError(t, fx.BlocksAdd(ctx, bs[:3]))
-
-		cids := testutil.BlocksToKeys(bs)
-
-		fx.persistStore.EXPECT().Get(ctx, bs[3].Cid()).Return(bs[3], nil)
-
-		result, err := fx.CidEntries(ctx, cids)
-		defer result.Release()
-		require.NoError(t, err)
-		require.Len(t, result.entries, len(bs))
-		t.Log(result.entries[3])
-	})
 }
 
 func TestRedisIndex_CidExistsInSpace(t *testing.T) {

index/index.go

@@ -119,6 +119,7 @@ type redisIndex struct {
 	redsync      *redsync.Redsync
 	persistStore persistentStore
 	persistTtl   time.Duration
+	persistMu    sync.Mutex
 	ticker       periodicsync.PeriodicSync
 	defaultLimit uint64
 
@@ -301,7 +302,7 @@ func FileKey(fileId string) string {
 
 func DelKey(k Key) string {
 	hash := strconv.FormatUint(uint64(xxhash.ChecksumString32(k.GroupId)), 36)
-	return "d:" + k.SpaceId + ".{" + hash + "}"
+	return "del:" + k.SpaceId + ".{" + hash + "}"
 }
 
 const infoKey = "info"

index/loader.go

@@ -195,6 +195,8 @@ func (ri *redisIndex) updateKeyUsage(ctx context.Context, key string) (err error
 }
 
 func (ri *redisIndex) PersistKeys(ctx context.Context) {
+	ri.persistMu.Lock()
+	defer ri.persistMu.Unlock()
 	st := time.Now()
 	rand.Shuffle(len(partitions), func(i, j int) {
 		partitions[i], partitions[j] = partitions[j], partitions[i]