fix(sdk): client update extents from meta and drop extents in cache leadto ek conflict.

Signed-off-by: baihailong <baihailong@oppo.com>

sdk/data/stream/extent_cache.go

@@ -121,7 +121,7 @@ func (cache *ExtentCache) update(gen, size uint64, force bool, eks []proto.Exten
 	cache.root.Clear(false)
 	for _, ek := range eks {
 		extent := ek
-		log.LogDebugf("action[update] update cache replace or insert ek [%v]", ek.String())
+		log.LogDebugf("action[update] update cache ino(%v) replace or insert ek [%v]", cache.inode, ek.String())
 		cache.root.ReplaceOrInsert(&extent)
 	}
 }

sdk/data/stream/stream_reader.go

@@ -54,6 +54,7 @@ type Streamer struct {
 	pendingCache         chan bcacheKey
 	verSeq               uint64
 	needUpdateVer        int32
+	extentsLock          sync.Mutex
 }
 
 type bcacheKey struct {
@@ -91,6 +92,8 @@ func (s *Streamer) String() string {
 
 // TODO should we call it RefreshExtents instead?
 func (s *Streamer) GetExtents() error {
+	s.extentsLock.Lock()
+	defer s.extentsLock.Unlock()
 	if s.client.disableMetaCache || !s.needBCache {
 		return s.extents.RefreshForce(s.inode, s.client.getExtents)
 	}
@@ -99,6 +102,8 @@ func (s *Streamer) GetExtents() error {
 }
 
 func (s *Streamer) GetExtentsForce() error {
+	s.extentsLock.Lock()
+	defer s.extentsLock.Unlock()
 	return s.extents.RefreshForce(s.inode, s.client.getExtents)
 }
 

sdk/data/stream/stream_writer.go

@@ -299,6 +299,8 @@ func (s *Streamer) handleRequest(request interface{}) {
 }
 
 func (s *Streamer) write(data []byte, offset, size, flags int, checkFunc func() error) (total int, err error) {
+	s.extentsLock.Lock()
+	defer s.extentsLock.Unlock()
 	var (
 		direct     bool
 		retryTimes int8