diff options
| author | Chris Lu <chrislusf@users.noreply.github.com> | 2024-07-24 23:46:40 -0700 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2024-07-24 23:46:40 -0700 |
| commit | 952afd810c20cba38c8e389c7cf972cdbca2931d (patch) | |
| tree | 3bc7e5288a5b3272f26aff20058d79ebc4cddc0f | |
| parent | 8b402fb4b956ac8b10db3453d8a205cda392b372 (diff) | |
| download | seaweedfs-952afd810c20cba38c8e389c7cf972cdbca2931d.tar.xz seaweedfs-952afd810c20cba38c8e389c7cf972cdbca2931d.zip | |
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
| -rw-r--r-- | weed/mount/filehandle.go | 14 | ||||
| -rw-r--r-- | weed/mount/filehandle_map.go | 2 | ||||
| -rw-r--r-- | weed/mount/filehandle_read.go | 14 | ||||
| -rw-r--r-- | weed/mount/locked_entry.go | 2 | ||||
| -rw-r--r-- | weed/mount/weedfs.go | 5 | ||||
| -rw-r--r-- | weed/mount/weedfs_dir_lookup.go | 2 | ||||
| -rw-r--r-- | weed/mount/weedfs_dir_read.go | 2 | ||||
| -rw-r--r-- | weed/mount/weedfs_file_lseek.go | 4 | ||||
| -rw-r--r-- | weed/mount/weedfs_file_sync.go | 7 |
9 files changed, 15 insertions, 37 deletions
diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 2e08432c0..f47d4a877 100644 --- a/weed/mount/filehandle.go +++ b/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() diff --git a/weed/mount/filehandle_map.go b/weed/mount/filehandle_map.go index f0051f061..bb78d0b14 100644 --- a/weed/mount/filehandle_map.go +++ b/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 diff --git a/weed/mount/filehandle_read.go b/weed/mount/filehandle_read.go index 7b2629c13..3c315b1c4 100644 --- a/weed/mount/filehandle_read.go +++ b/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 } diff --git a/weed/mount/locked_entry.go b/weed/mount/locked_entry.go index f3b4bf484..c5fbaee91 100644 --- a/weed/mount/locked_entry.go +++ b/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() diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index c5a1d2755..a9fbd9380 100644 --- a/weed/mount/weedfs.go +++ b/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) } } diff --git a/weed/mount/weedfs_dir_lookup.go b/weed/mount/weedfs_dir_lookup.go index e646b06a9..f3ba0cc85 100644 --- a/weed/mount/weedfs_dir_lookup.go +++ b/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) } diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index f140fd86f..02ed72431 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/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) } diff --git a/weed/mount/weedfs_file_lseek.go b/weed/mount/weedfs_file_lseek.go index 35157d993..0cf7ef43b 100644 --- a/weed/mount/weedfs_file_lseek.go +++ b/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( diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 762a9b8de..d857606bd 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/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, } |
