meta/tkv: skip updating nlink when the transactions conflict a lot

pkg/meta/tkv.go

@@ -52,7 +52,7 @@ type kvtxn interface {
 
 type tkvClient interface {
 	name() string
-	txn(f func(*kvTxn) error) error
+	txn(f func(*kvTxn) error, retry int) error
 	scan(prefix []byte, handler func(key, value []byte)) error
 	reset(prefix []byte) error
 	close() error
@@ -62,6 +62,7 @@ type tkvClient interface {
 
 type kvTxn struct {
 	kvtxn
+	retry int
 }
 
 func (tx *kvTxn) deleteKeys(prefix []byte) {
@@ -332,7 +333,7 @@ func (m *kvMeta) get(key []byte) ([]byte, error) {
 	err := m.client.txn(func(tx *kvTxn) error {
 		value = tx.get(key)
 		return nil
-	})
+	}, 0)
 	return value, err
 }
 
@@ -344,7 +345,7 @@ func (m *kvMeta) scanKeys(prefix []byte) ([][]byte, error) {
 			return true
 		})
 		return nil
-	})
+	}, 0)
 	return keys, err
 }
 
@@ -363,7 +364,7 @@ func (m *kvMeta) scanValues(prefix []byte, limit int, filter func(k, v []byte) b
 			return limit < 0 || c < limit
 		})
 		return nil
-	})
+	}, 0)
 	return values, err
 }
 
@@ -685,7 +686,7 @@ func (m *kvMeta) txn(f func(tx *kvTxn) error, inodes ...Ino) error {
 	}
 	var lastErr error
 	for i := 0; i < 50; i++ {
-		err := m.client.txn(f)
+		err := m.client.txn(f, i)
 		if eno, ok := err.(syscall.Errno); ok && eno == 0 {
 			err = nil
 		}
@@ -1104,9 +1105,13 @@ func (m *kvMeta) doMknod(ctx Context, parent Ino, name string, _type uint8, mode
 		if parent != TrashInode {
 			if _type == TypeDirectory {
 				pattr.Nlink++
-				updateParent = true
+				if tx.retry < 10 {
+					updateParent = true
+				} else {
+					logger.Warnf("Skip updating nlink of directory %d to reduce conflict", parent)
+				}
 			}
-			if updateParent || now.Sub(time.Unix(pattr.Mtime, int64(pattr.Mtimensec))) >= minUpdateTime {
+			if updateParent || now.Sub(time.Unix(pattr.Mtime, int64(pattr.Mtimensec))) >= minUpdateTime*time.Duration(tx.retry+1) {
 				pattr.Mtime = now.Unix()
 				pattr.Mtimensec = uint32(now.Nanosecond())
 				pattr.Ctime = now.Unix()
@@ -1660,7 +1665,7 @@ func (m *kvMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry
 			err := m.client.txn(func(tx *kvTxn) error {
 				rs = tx.gets(keys...)
 				return nil
-			})
+			}, 0)
 			if err != nil {
 				return err
 			}
@@ -2110,7 +2115,7 @@ func (m *kvMeta) doCleanupDelayedSlices(edge int64) (int, error) {
 					return c < batch
 				})
 			return nil
-		}); err != nil {
+		}, 0); err != nil {
 			logger.Warnf("Scan delayed slices: %s", err)
 			return count, err
 		}
@@ -2259,7 +2264,7 @@ func (m *kvMeta) compactChunk(inode Ino, indx uint32, force bool) {
 				if s.id > 0 && m.client.txn(func(tx *kvTxn) error {
 					refs = tx.incrBy(m.sliceKey(s.id, s.size), 0)
 					return nil
-				}) == nil && refs < 0 {
+				}, 0) == nil && refs < 0 {
 					m.deleteSlice(s.id, s.size)
 				}
 			}
@@ -2594,7 +2599,7 @@ func (m *kvMeta) dumpEntry(inode Ino, e *DumpedEntry) error {
 			e.Symlink = string(l)
 		}
 		return nil
-	})
+	}, 0)
 }
 
 func (m *kvMeta) dumpDir(inode Ino, tree *DumpedEntry, bw *bufio.Writer, depth int, showProgress func(totalIncr, currentIncr int64)) error {

pkg/meta/tkv_badger.go

@@ -147,7 +147,7 @@ func (c *badgerClient) shouldRetry(err error) bool {
 	return err == badger.ErrConflict
 }
 
-func (c *badgerClient) txn(f func(*kvTxn) error) (err error) {
+func (c *badgerClient) txn(f func(*kvTxn) error, retry int) (err error) {
 	t := c.client.NewTransaction(true)
 	defer t.Discard()
 	defer func() {
@@ -161,7 +161,7 @@ func (c *badgerClient) txn(f func(*kvTxn) error) (err error) {
 		}
 	}()
 	tx := &badgerTxn{t, c.client}
-	err = f(&kvTxn{tx})
+	err = f(&kvTxn{tx, retry})
 	if err != nil {
 		return err
 	}

pkg/meta/tkv_etcd.go

@@ -213,7 +213,7 @@ func (c *etcdClient) shouldRetry(err error) bool {
 	return errors.Is(err, conflicted)
 }
 
-func (c *etcdClient) txn(f func(*kvTxn) error) (err error) {
+func (c *etcdClient) txn(f func(*kvTxn) error, retry int) (err error) {
 	ctx := context.Background()
 	tx := &etcdTxn{
 		ctx,
@@ -231,7 +231,7 @@ func (c *etcdClient) txn(f func(*kvTxn) error) (err error) {
 			}
 		}
 	}()
-	err = f(&kvTxn{tx})
+	err = f(&kvTxn{tx, retry})
 	if err != nil {
 		return err
 	}

pkg/meta/tkv_fdb.go

@@ -59,9 +59,9 @@ func (c *fdbClient) name() string {
 	return "fdb"
 }
 
-func (c *fdbClient) txn(f func(*kvTxn) error) error {
+func (c *fdbClient) txn(f func(*kvTxn) error, retry int) error {
 	_, err := c.client.Transact(func(t fdb.Transaction) (interface{}, error) {
-		e := f(&kvTxn{&fdbTxn{t}})
+		e := f(&kvTxn{&fdbTxn{t}, retry})
 		return nil, e
 	})
 	return err

pkg/meta/tkv_mem.go

@@ -199,13 +199,13 @@ func (c *memKV) set(key string, value []byte) {
 	}
 }
 
-func (c *memKV) txn(f func(*kvTxn) error) error {
+func (c *memKV) txn(f func(*kvTxn) error, retry int) error {
 	tx := &memTxn{
 		store:    c,
 		observed: make(map[string]int),
 		buffer:   make(map[string][]byte),
 	}
-	if err := f(&kvTxn{tx}); err != nil {
+	if err := f(&kvTxn{tx, retry}); err != nil {
 		return err
 	}
 
@@ -264,7 +264,7 @@ func (c *memKV) reset(prefix []byte) error {
 		return c.scan(prefix, func(key, value []byte) {
 			kt.delete(key)
 		})
-	})
+	}, 0)
 }
 
 func (c *memKV) close() error {

pkg/meta/tkv_prefix.go

@@ -76,10 +76,10 @@ type prefixClient struct {
 	prefix []byte
 }
 
-func (c *prefixClient) txn(f func(*kvTxn) error) error {
+func (c *prefixClient) txn(f func(*kvTxn) error, retry int) error {
 	return c.tkvClient.txn(func(tx *kvTxn) error {
-		return f(&kvTxn{&prefixTxn{tx, c.prefix}})
-	})
+		return f(&kvTxn{&prefixTxn{tx, c.prefix}, retry})
+	}, retry)
 }
 
 func (c *prefixClient) scan(prefix []byte, handler func(key, value []byte)) error {

pkg/meta/tkv_test.go

@@ -64,7 +64,7 @@ func testTKV(t *testing.T, c tkvClient) {
 		if err := c.txn(func(kt *kvTxn) error {
 			f(kt)
 			return nil
-		}); err != nil {
+		}, 0); err != nil {
 			t.Fatal(err)
 		}
 	}
@@ -197,21 +197,21 @@ func testTKV(t *testing.T, c tkvClient) {
 	c.txn(func(tx *kvTxn) error {
 		count = tx.incrBy([]byte("counter"), -1)
 		return nil
-	})
+	}, 0)
 	if count != -1 {
 		t.Fatalf("counter should be -1, but got %d", count)
 	}
 	c.txn(func(tx *kvTxn) error {
 		count = tx.incrBy([]byte("counter"), 0)
 		return nil
-	})
+	}, 0)
 	if count != -1 {
 		t.Fatalf("counter should be -1, but got %d", count)
 	}
 	c.txn(func(tx *kvTxn) error {
 		count = tx.incrBy([]byte("counter"), 2)
 		return nil
-	})
+	}, 0)
 	if count != 1 {
 		t.Fatalf("counter should be 1, but got %d", count)
 	}

pkg/meta/tkv_tikv.go

@@ -181,7 +181,7 @@ func (c *tikvClient) shouldRetry(err error) bool {
 	return strings.Contains(err.Error(), "write conflict") || strings.Contains(err.Error(), "TxnLockNotFound")
 }
 
-func (c *tikvClient) txn(f func(*kvTxn) error) (err error) {
+func (c *tikvClient) txn(f func(*kvTxn) error, retry int) (err error) {
 	tx, err := c.client.Begin()
 	if err != nil {
 		return err
@@ -196,7 +196,7 @@ func (c *tikvClient) txn(f func(*kvTxn) error) (err error) {
 			}
 		}
 	}()
-	if err = f(&kvTxn{&tikvTxn{tx}}); err != nil {
+	if err = f(&kvTxn{&tikvTxn{tx}, retry}); err != nil {
 		return err
 	}
 	if !tx.IsReadOnly() {