fix(data): avoid write extent after store closed

Signed-off-by: NaturalSelect <huangzhibin1@oppo.com>

datanode/disk.go

@@ -49,7 +49,7 @@ const (
 	ExpiredPartitionExistTime = time.Hour * time.Duration(24*7)
 )
 
-const DefaultCurrentLoadDpLimit = 1
+const DefaultCurrentLoadDpLimit = 4
 
 const (
 	DecommissionDiskMark = "decommissionDiskMark"

proto/packet.go

@@ -257,7 +257,9 @@ const (
 	OpReadRepairExtentAgain uint8 = 0xEF
 
 	// io speed limit
-	OpLimitedIoErr uint8 = 0xB1
+	OpLimitedIoErr          uint8 = 0xB1
+	OpReadRepairExtentAgain uint8 = 0xEF
+	OpStoreClosed           uint8 = 0xB2
 )
 
 const (

repl/packet.go

@@ -128,6 +128,8 @@ func (p *FollowerPacket) identificationErrorResultCode(errLog string, errMsg str
 		p.ResultCode = proto.OpTryOtherAddr
 	} else if strings.Contains(errMsg, raft.ErrStopped.Error()) {
 		p.ResultCode = proto.OpTryOtherAddr
+	} else if strings.Contains(errMsg, storage.ErrStoreAlreadyClosed.Error()) {
+		p.ResultCode = proto.OpStoreClosed
 	} else if strings.Contains(errLog, ActionReceiveFromFollower) || strings.Contains(errLog, ActionSendToFollowers) ||
 		strings.Contains(errLog, ConnIsNullErr) {
 		p.ResultCode = proto.OpIntraGroupNetErr
@@ -452,6 +454,8 @@ func (p *Packet) identificationErrorResultCode(errLog string, errMsg string) {
 		// log.LogDebugf("action[identificationErrorResultCode] not change ver erro code, (%v)", string(debug.Stack()))
 	} else if strings.Contains(errMsg, storage.NoDiskReadRepairExtentTokenError.Error()) {
 		p.ResultCode = proto.OpReadRepairExtentAgain
+	} else if strings.Contains(errMsg, storage.ErrStoreAlreadyClosed.Error()) {
+		p.ResultCode = proto.OpStoreClosed
 	} else {
 		log.LogErrorf("action[identificationErrorResultCode] error %v, errmsg %v", errLog, errMsg)
 		p.ResultCode = proto.OpIntraGroupNetErr

storage/extent_store.go

@@ -70,6 +70,10 @@ const (
 	StaleExtStoreTimeFormat   = "20060102150405.000000000"
 )
 
+var (
+	ErrStoreAlreadyClosed = errors.New("extent store already closed")
+)
+
 var (
 	RegexpExtentFile, _ = regexp.Compile(`^(\d)+$`)
 	SnapShotFilePool    = &sync.Pool{New: func() interface{} {
@@ -178,6 +182,7 @@ type ExtentStore struct {
 	extentLockMap                     map[uint64]proto.GcFlag
 	elMutex                           sync.RWMutex
 	extentLock                        bool
+	stopMutex                         sync.RWMutex
 	stopC                             chan interface{}
 
 	ApplyId      uint64
@@ -370,6 +375,14 @@ func (s *ExtentStore) startFlushCache() {
 
 // Create creates an extent.
 func (s *ExtentStore) Create(extentID uint64) (err error) {
+	s.stopMutex.RLock()
+	defer s.stopMutex.RUnlock()
+	if s.closed {
+		err = ErrStoreAlreadyClosed
+		log.LogErrorf("[Create] store(%v) failed to create extent(%v), err(%v)", s.dataPath, extentID, err)
+		return
+	}
+
 	var e *Extent
 	name := path.Join(s.dataPath, strconv.Itoa(int(extentID)))
 	if s.HasExtent(extentID) {
@@ -678,6 +691,14 @@ func (s *ExtentStore) initBaseFileID(allowDelay bool, loadTimeout time.Duration)
 
 // Write writes the given extent to the disk.
 func (s *ExtentStore) Write(extentID uint64, offset, size int64, data []byte, crc uint32, writeType int, isSync bool, isHole bool, isRepair, isBackupWrite bool) (status uint8, err error) {
+	s.stopMutex.RLock()
+	defer s.stopMutex.RUnlock()
+	if s.closed {
+		err = ErrStoreAlreadyClosed
+		log.LogErrorf("[Write] store(%v) failed to write extent(%v), err(%v)", s.dataPath, extentID, err)
+		return
+	}
+
 	var (
 		e  *Extent
 		ei *ExtentInfo
@@ -874,6 +895,14 @@ func (s *ExtentStore) GetGcFlag(extId uint64) proto.GcFlag {
 
 // MarkDelete marks the given extent as deleted.
 func (s *ExtentStore) MarkDelete(extentID uint64, offset, size int64) (err error) {
+	s.stopMutex.RLock()
+	defer s.stopMutex.RUnlock()
+	if s.closed {
+		err = ErrStoreAlreadyClosed
+		log.LogErrorf("[MarkDelete] store(%v) failed to mark delete extent(%v), err(%v)", s.dataPath, extentID, err)
+		return
+	}
+
 	var ei *ExtentInfo
 	s.eiMutex.RLock()
 	ei = s.extentInfoMap[extentID]
@@ -985,8 +1014,10 @@ func (s *ExtentStore) Close() {
 			vFp.Close()
 		}
 	}
-	s.closed = true
 
+	s.stopMutex.Lock()
+	defer s.stopMutex.Unlock()
+	s.closed = true
 	if err := s.writeReadDirHint(); err != nil {
 		log.LogErrorf("[Close] store(%v) failed to write extent hint, err(%v)", s.dataPath, err)
 	}