Allocate nextTrash inode in mknod to avoid generating too many txn conflicts

pkg/meta/base.go

@@ -2324,14 +2324,8 @@ func (m *baseMeta) checkTrash(parent Ino, trash *Ino) syscall.Errno {
 
 	st := m.en.doLookup(Background, TrashInode, name, trash, nil)
 	if st == syscall.ENOENT {
-		next, err := m.en.incrCounter("nextTrash", 1)
-		if err == nil {
-			*trash = TrashInode + Ino(next)
-			attr := Attr{Typ: TypeDirectory, Nlink: 2, Length: 4 << 10, Parent: TrashInode, Full: true}
-			st = m.en.doMknod(Background, TrashInode, name, TypeDirectory, 0555, 0, "", trash, &attr)
-		} else {
-			st = errno(err)
-		}
+		attr := Attr{Typ: TypeDirectory, Nlink: 2, Length: 4 << 10, Parent: TrashInode, Full: true}
+		st = m.en.doMknod(Background, TrashInode, name, TypeDirectory, 0555, 0, "", trash, &attr)
 	}
 
 	m.Lock()

pkg/meta/redis.go

@@ -603,6 +603,10 @@ func (m *redisMeta) usedSpaceKey() string {
 	return m.prefix + usedSpace
 }
 
+func (m *redisMeta) nextTrashKey() string {
+	return m.prefix + "nextTrash"
+}
+
 func (m *redisMeta) dirDataLengthKey() string {
 	return m.prefix + "dirDataLength"
 }
@@ -1248,6 +1252,12 @@ func (m *redisMeta) doMknod(ctx Context, parent Ino, name string, _type uint8, m
 				*inode = foundIno
 			}
 			return syscall.EEXIST
+		} else if parent == TrashInode {
+			if next, err := tx.Incr(ctx, m.nextTrashKey()).Result(); err != nil { // Some inode will be wasted if conflict happens
+				return err
+			} else {
+				*inode = TrashInode + Ino(next)
+			}
 		}
 
 		mode &= 07777

pkg/meta/sql.go

@@ -701,26 +701,30 @@ func (m *dbMeta) getCounter(name string) (v int64, err error) {
 	return
 }
 
-func (m *dbMeta) incrCounter(name string, value int64) (int64, error) {
-	var v int64
-	err := m.txn(func(s *xorm.Session) error {
-		var c = counter{Name: name}
-		ok, err := s.ForUpdate().Get(&c)
-		if err != nil {
-			return err
-		}
-		v = c.Value + value
-		if value > 0 {
-			c.Value = v
-			if ok {
-				_, err = s.Cols("value").Update(&c, &counter{Name: name})
-			} else {
-				err = mustInsert(s, &c)
-			}
-		}
+func (m *dbMeta) incrCounter(name string, value int64) (v int64, err error) {
+	err = m.txn(func(s *xorm.Session) error {
+		v, err = m.incrSessionCounter(s, name, value)
 		return err
 	})
-	return v, err
+	return
+}
+
+func (m *dbMeta) incrSessionCounter(s *xorm.Session, name string, value int64) (v int64, err error) {
+	var c = counter{Name: name}
+	ok, err := s.ForUpdate().Get(&c)
+	if err != nil {
+		return
+	}
+	v = c.Value + value
+	if value > 0 {
+		c.Value = v
+		if ok {
+			_, err = s.Cols("value").Update(&c, &counter{Name: name})
+		} else {
+			err = mustInsert(s, &c)
+		}
+	}
+	return
 }
 
 func (m *dbMeta) setIfSmall(name string, value, diff int64) (bool, error) {
@@ -1317,6 +1321,12 @@ func (m *dbMeta) doMknod(ctx Context, parent Ino, name string, _type uint8, mode
 				*inode = foundIno
 			}
 			return syscall.EEXIST
+		} else if parent == TrashInode {
+			if next, err := m.incrSessionCounter(s, "nextTrash", 1); err != nil {
+				return err
+			} else {
+				*inode = TrashInode + Ino(next)
+			}
 		}
 
 		n := node{Inode: *inode}

pkg/meta/tkv.go

@@ -1147,6 +1147,10 @@ func (m *kvMeta) doMknod(ctx Context, parent Ino, name string, _type uint8, mode
 				*inode = foundIno
 			}
 			return syscall.EEXIST
+		} else if parent == TrashInode { // user's inode is allocated by prefetch, trash inode is allocated on demand
+			key := m.counterKey("nextTrash")
+			next := tx.incrBy(key, 1)
+			*inode = TrashInode + Ino(next)
 		}
 
 		mode &= 07777