Fix dead lock (#5815)

* reduce locks to avoid dead lock

Flush->FlushData->uplloadPipeline.FluahAll
uploaderCount>0

goroutine 1 [sync.Cond.Wait, 71 minutes]:
sync.runtime_notifyListWait(0xc0007ae4d0, 0x0)
	/usr/local/go/src/runtime/sema.go:569 +0x159
sync.(*Cond).Wait(0xc001a59290?)
	/usr/local/go/src/sync/cond.go:70 +0x85
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*UploadPipeline).waitForCurrentWritersToComplete(0xc0002ee4d0)
	/github/workspace/weed/mount/page_writer/upload_pipeline_lock.go:58 +0x32
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*UploadPipeline).FlushAll(0xc0002ee4d0)
	/github/workspace/weed/mount/page_writer/upload_pipeline.go:151 +0x25
github.com/seaweedfs/seaweedfs/weed/mount.(*ChunkedDirtyPages).FlushData(0xc00087e840)
	/github/workspace/weed/mount/dirty_pages_chunked.go:54 +0x29
github.com/seaweedfs/seaweedfs/weed/mount.(*PageWriter).FlushData(...)
	/github/workspace/weed/mount/page_writer.go:50
github.com/seaweedfs/seaweedfs/weed/mount.(*WFS).doFlush(0xc0006ad600, 0xc00030d380, 0x0, 0x0)
	/github/workspace/weed/mount/weedfs_file_sync.go:101 +0x169
github.com/seaweedfs/seaweedfs/weed/mount.(*WFS).Flush(0xc0006ad600, 0xc001a594a8?, 0xc0004c1ca0)
	/github/workspace/weed/mount/weedfs_file_sync.go:59 +0x48
github.com/hanwen/go-fuse/v2/fuse.doFlush(0xc0000da870?, 0xc0004c1b08)

SaveContent -> MemChunk.RLock ->
	ChunkedDirtyPages.saveChunkedFileIntervalToStorage
	pages.fh.AddChunks([]*filer_pb.FileChunk{chunk})
		fh.entryLock.Lock()

sync.(*RWMutex).Lock(0x0?)
	/usr/local/go/src/sync/rwmutex.go:146 +0x31
github.com/seaweedfs/seaweedfs/weed/mount.(*FileHandle).AddChunks(0xc00030d380, {0xc00028bdc8, 0x1, 0x1})
	/github/workspace/weed/mount/filehandle.go:93 +0x45
github.com/seaweedfs/seaweedfs/weed/mount.(*ChunkedDirtyPages).saveChunkedFileIntervalToStorage(0xc00087e840, {0x2be7ac0, 0xc00018d9e0}, 0x0, 0x121, 0x17e3c624565ace45, 0x1?)
	/github/workspace/weed/mount/dirty_pages_chunked.go:80 +0x2d4
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*MemChunk).SaveContent(0xc0008d9130, 0xc0008093e0)
	/github/workspace/weed/mount/page_writer/page_chunk_mem.go:115 +0x112
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*UploadPipeline).moveToSealed.func1()
	/github/workspace/weed/mount/page_writer/upload_pipeline.go:187 +0x55
github.com/seaweedfs/seaweedfs/weed/util.(*LimitedConcurrentExecutor).Execute.func1()
	/github/workspace/weed/util/limited_executor.go:38 +0x62
created by github.com/seaweedfs/seaweedfs/weed/util.(*LimitedConcurrentExecutor).Execute in goroutine 1
	/github/workspace/weed/util/limited_executor.go:33 +0x97

On metadata update
fh.entryLock.Lock()
	fh.dirtyPages.Destroy()
		up.chunksLock.Lock => each sealed chunk.FreeReference => MemChunk.Lock

goroutine 134 [sync.RWMutex.Lock, 71 minutes]:
sync.runtime_SemacquireRWMutex(0xc0007c3558?, 0xea?, 0x3fb0800?)
	/usr/local/go/src/runtime/sema.go:87 +0x25
sync.(*RWMutex).Lock(0xc0007c35a8?)
	/usr/local/go/src/sync/rwmutex.go:151 +0x6a
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*MemChunk).FreeResource(0xc0008d9130)
	/github/workspace/weed/mount/page_writer/page_chunk_mem.go:38 +0x2a
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*SealedChunk).FreeReference(0xc00071cdb0, {0xc0006ba1a0, 0x20})
	/github/workspace/weed/mount/page_writer/upload_pipeline.go:38 +0xb7
github.com/seaweedfs/seaweedfs/weed/mount/page_writer.(*UploadPipeline).Shutdown(0xc0002ee4d0)
	/github/workspace/weed/mount/page_writer/upload_pipeline.go:220 +0x185
github.com/seaweedfs/seaweedfs/weed/mount.(*ChunkedDirtyPages).Destroy(0xc0008cea40?)
	/github/workspace/weed/mount/dirty_pages_chunked.go:87 +0x17
github.com/seaweedfs/seaweedfs/weed/mount.(*PageWriter).Destroy(...)
	/github/workspace/weed/mount/page_writer.go:78
github.com/seaweedfs/seaweedfs/weed/mount.NewSeaweedFileSystem.func3({0xc00069a6c0, 0x30}, 0x6?)
	/github/workspace/weed/mount/weedfs.go:119 +0x17a
github.com/seaweedfs/seaweedfs/weed/mount/meta_cache.NewMetaCache.func1({0xc00069a6c0?, 0xc00069a480?}, 0x4015b40?)
	/github/workspace/weed/mount/meta_cache/meta_cache.go:37 +0x1c
github.com/seaweedfs/seaweedfs/weed/mount/meta_cache.SubscribeMetaEvents.func1(0xc000661810)
	/github/workspace/weed/mount/meta_cache/meta_cache_subscribe.go:43 +0x570

* use locked entry everywhere

* modifiable remote entry

* skip locking after getting lock from fhLockTable

weed/mount/filehandle.go

@@ -66,8 +66,8 @@ func (fh *FileHandle) FullPath() util.FullPath {
 	return fp
 }
 
-func (fh *FileHandle) GetEntry() *filer_pb.Entry {
-	return fh.entry.GetEntry()
+func (fh *FileHandle) GetEntry() *LockedEntry {
+	return fh.entry
 }
 
 func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) {
@@ -90,13 +90,6 @@ func (fh *FileHandle) UpdateEntry(fn func(entry *filer_pb.Entry)) *filer_pb.Entr
 }
 
 func (fh *FileHandle) AddChunks(chunks []*filer_pb.FileChunk) {
-	fh.entryLock.Lock()
-	defer fh.entryLock.Unlock()
-
-	if fh.entry == nil {
-		return
-	}
-
 	fh.entry.AppendChunks(chunks)
 }
 
@@ -105,9 +98,6 @@ func (fh *FileHandle) ReleaseHandle() {
 	fhActiveLock := fh.wfs.fhLockTable.AcquireLock("ReleaseHandle", fh.fh, util.ExclusiveLock)
 	defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
 
-	fh.entryLock.Lock()
-	defer fh.entryLock.Unlock()
-
 	fh.dirtyPages.Destroy()
 	if IsDebugFileReadWrite {
 		fh.mirrorFile.Close()

weed/mount/filehandle_map.go

@@ -50,7 +50,7 @@ func (i *FileHandleToInode) AcquireFileHandle(wfs *WFS, inode uint64, entry *fil
 	} else {
 		fh.counter++
 	}
-	if fh.GetEntry() != entry {
+	if fh.GetEntry().GetEntry() != entry {
 		fh.SetEntry(entry)
 	}
 	return fh

weed/mount/filehandle_read.go

@@ -29,23 +29,19 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, int64, e
 	fileFullPath := fh.FullPath()
 
 	entry := fh.GetEntry()
-	if entry == nil {
-		return 0, 0, io.EOF
-	}
 
 	if entry.IsInRemoteOnly() {
 		glog.V(4).Infof("download remote entry %s", fileFullPath)
-		newEntry, err := fh.downloadRemoteEntry(entry)
+		err := fh.downloadRemoteEntry(entry)
 		if err != nil {
 			glog.V(1).Infof("download remote entry %s: %v", fileFullPath, err)
 			return 0, 0, err
 		}
-		entry = newEntry
 	}
 
 	fileSize := int64(entry.Attributes.FileSize)
 	if fileSize == 0 {
-		fileSize = int64(filer.FileSize(entry))
+		fileSize = int64(filer.FileSize(entry.GetEntry()))
 	}
 
 	if fileSize == 0 {
@@ -70,7 +66,7 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, int64, e
 	return int64(totalRead), ts, err
 }
 
-func (fh *FileHandle) downloadRemoteEntry(entry *filer_pb.Entry) (*filer_pb.Entry, error) {
+func (fh *FileHandle) downloadRemoteEntry(entry *LockedEntry) (error) {
 
 	fileFullPath := fh.FullPath()
 	dir, _ := fileFullPath.DirAndName()
@@ -88,12 +84,12 @@ func (fh *FileHandle) downloadRemoteEntry(entry *filer_pb.Entry) (*filer_pb.Entr
 			return fmt.Errorf("CacheRemoteObjectToLocalCluster file %s: %v", fileFullPath, err)
 		}
 
-		entry = resp.Entry
+		entry.SetEntry(resp.Entry)
 
 		fh.wfs.metaCache.InsertEntry(context.Background(), filer.FromPbEntry(request.Directory, resp.Entry))
 
 		return nil
 	})
 
-	return entry, err
+	return err
 }

weed/mount/locked_entry.go

@@ -16,6 +16,8 @@ func (le *LockedEntry) GetEntry() *filer_pb.Entry {
 	return le.Entry
 }
 
+// SetEntry sets the entry of the LockedEntry
+// entry should never be nil
 func (le *LockedEntry) SetEntry(entry *filer_pb.Entry) {
 	le.Lock()
 	defer le.Unlock()

weed/mount/weedfs.go

@@ -112,9 +112,6 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 					fhActiveLock := fh.wfs.fhLockTable.AcquireLock("invalidateFunc", fh.fh, util.ExclusiveLock)
 					defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
 
-					fh.entryLock.Lock()
-					defer fh.entryLock.Unlock()
-
 					// Recreate dirty pages
 					fh.dirtyPages.Destroy()
 					fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
@@ -122,7 +119,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 					// Update handle entry
 					newentry, status := wfs.maybeLoadEntry(filePath)
 					if status == fuse.OK {
-						if fh.GetEntry() != newentry {
+						if fh.GetEntry().GetEntry() != newentry {
 							fh.SetEntry(newentry)
 						}
 					}

weed/mount/weedfs_dir_lookup.go

@@ -59,7 +59,7 @@ func (wfs *WFS) Lookup(cancel <-chan struct{}, header *fuse.InHeader, name strin
 
 	if fh, found := wfs.fhmap.FindFileHandle(inode); found {
 		fh.entryLock.RLock()
-		if entry := fh.GetEntry(); entry != nil {
+		if entry := fh.GetEntry().GetEntry(); entry != nil {
 			glog.V(4).Infof("lookup opened file %s size %d", dirPath.Child(localEntry.Name()), filer.FileSize(entry))
 			localEntry = filer.FromPbEntry(string(dirPath), entry)
 		}

weed/mount/weedfs_dir_read.go

@@ -173,7 +173,7 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPl
 			}
 			if fh, found := wfs.fhmap.FindFileHandle(inode); found {
 				glog.V(4).Infof("readdir opened file %s", dirPath.Child(dirEntry.Name))
-				entry = filer.FromPbEntry(string(dirPath), fh.GetEntry())
+				entry = filer.FromPbEntry(string(dirPath), fh.GetEntry().GetEntry())
 			}
 			wfs.outputFilerEntry(entryOut, inode, entry)
 		}

weed/mount/weedfs_file_lseek.go

@@ -38,10 +38,8 @@ func (wfs *WFS) Lseek(cancel <-chan struct{}, in *fuse.LseekIn, out *fuse.LseekO
 	// lock the file until the proper offset was calculated
 	fhActiveLock := fh.wfs.fhLockTable.AcquireLock("Lseek", fh.fh, util.SharedLock)
 	defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
-	fh.entryLock.RLock()
-	defer fh.entryLock.RUnlock()
 
-	fileSize := int64(filer.FileSize(fh.GetEntry()))
+	fileSize := int64(filer.FileSize(fh.GetEntry().GetEntry()))
 	offset := max(int64(in.Offset), 0)
 
 	glog.V(4).Infof(

weed/mount/weedfs_file_sync.go

@@ -116,13 +116,8 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32) fuse.Status {
 	defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
 
 	err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
-		fh.entryLock.Lock()
-		defer fh.entryLock.Unlock()
 
 		entry := fh.GetEntry()
-		if entry == nil {
-			return nil
-		}
 		entry.Name = name // this flush may be just after a rename operation
 
 		if entry.Attributes != nil {
@@ -141,7 +136,7 @@ func (wfs *WFS) doFlush(fh *FileHandle, uid, gid uint32) fuse.Status {
 
 		request := &filer_pb.CreateEntryRequest{
 			Directory:                string(dir),
-			Entry:                    entry,
+			Entry:                    entry.GetEntry(),
 			Signatures:               []int32{wfs.signature},
 			SkipCheckParentDirectory: true,
 		}