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 5 months ago |
|
package mount
import ( "context" "github.com/hanwen/go-fuse/v2/fuse" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" "github.com/seaweedfs/seaweedfs/weed/util" "math" "sync" )
type DirectoryHandleId uint64
const ( directoryStreamBaseOffset = 2 // . & ..
)
type DirectoryHandle struct { isFinished bool entryStream []*filer.Entry entryStreamOffset uint64 }
func (dh *DirectoryHandle) reset() { *dh = DirectoryHandle{ isFinished: false, entryStream: []*filer.Entry{}, entryStreamOffset: directoryStreamBaseOffset, } }
type DirectoryHandleToInode struct { // shares the file handle id sequencer with FileHandleToInode{nextFh}
sync.Mutex dir2inode map[DirectoryHandleId]*DirectoryHandle }
func NewDirectoryHandleToInode() *DirectoryHandleToInode { return &DirectoryHandleToInode{ dir2inode: make(map[DirectoryHandleId]*DirectoryHandle), } }
func (wfs *WFS) AcquireDirectoryHandle() (DirectoryHandleId, *DirectoryHandle) { fh := FileHandleId(util.RandomUint64())
wfs.dhmap.Lock() defer wfs.dhmap.Unlock() dh := new(DirectoryHandle) dh.reset() wfs.dhmap.dir2inode[DirectoryHandleId(fh)] = dh return DirectoryHandleId(fh), dh }
func (wfs *WFS) GetDirectoryHandle(dhid DirectoryHandleId) *DirectoryHandle { wfs.dhmap.Lock() defer wfs.dhmap.Unlock() if dh, found := wfs.dhmap.dir2inode[dhid]; found { return dh } dh := new(DirectoryHandle) dh.reset() wfs.dhmap.dir2inode[dhid] = dh return dh }
func (wfs *WFS) ReleaseDirectoryHandle(dhid DirectoryHandleId) { wfs.dhmap.Lock() defer wfs.dhmap.Unlock() delete(wfs.dhmap.dir2inode, dhid) }
// Directory handling
/** Open directory * * Unless the 'default_permissions' mount option is given, * this method should check if opendir is permitted for this * directory. Optionally opendir may also return an arbitrary * filehandle in the fuse_file_info structure, which will be * passed to readdir, releasedir and fsyncdir. */ func (wfs *WFS) OpenDir(cancel <-chan struct{}, input *fuse.OpenIn, out *fuse.OpenOut) (code fuse.Status) { if !wfs.inodeToPath.HasInode(input.NodeId) { return fuse.ENOENT } dhid, _ := wfs.AcquireDirectoryHandle() out.Fh = uint64(dhid) return fuse.OK }
/** Release directory * * If the directory has been removed after the call to opendir, the * path parameter will be NULL. */ func (wfs *WFS) ReleaseDir(input *fuse.ReleaseIn) { wfs.ReleaseDirectoryHandle(DirectoryHandleId(input.Fh)) }
/** Synchronize directory contents * * If the directory has been removed after the call to opendir, the * path parameter will be NULL. * * If the datasync parameter is non-zero, then only the user data * should be flushed, not the meta data */ func (wfs *WFS) FsyncDir(cancel <-chan struct{}, input *fuse.FsyncIn) (code fuse.Status) { return fuse.OK }
/** Read directory * * The filesystem may choose between two modes of operation: * * 1) The readdir implementation ignores the offset parameter, and * passes zero to the filler function's offset. The filler * function will not return '1' (unless an error happens), so the * whole directory is read in a single readdir operation. * * 2) The readdir implementation keeps track of the offsets of the * directory entries. It uses the offset parameter and always * passes non-zero offset to the filler function. When the buffer * is full (or an error happens) the filler function will return * '1'. */ func (wfs *WFS) ReadDir(cancel <-chan struct{}, input *fuse.ReadIn, out *fuse.DirEntryList) (code fuse.Status) { return wfs.doReadDirectory(input, out, false) }
func (wfs *WFS) ReadDirPlus(cancel <-chan struct{}, input *fuse.ReadIn, out *fuse.DirEntryList) (code fuse.Status) { return wfs.doReadDirectory(input, out, true) }
func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out *fuse.DirEntryList, isPlusMode bool) fuse.Status { dh := wfs.GetDirectoryHandle(DirectoryHandleId(input.Fh)) if input.Offset == 0 { dh.reset() } else if dh.isFinished && input.Offset >= dh.entryStreamOffset { entryCurrentIndex := input.Offset - dh.entryStreamOffset if uint64(len(dh.entryStream)) <= entryCurrentIndex { return fuse.OK } }
isEarlyTerminated := false dirPath, code := wfs.inodeToPath.GetPath(input.NodeId) if code != fuse.OK { return code }
var dirEntry fuse.DirEntry processEachEntryFn := func(entry *filer.Entry) bool { dirEntry.Name = entry.Name() dirEntry.Mode = toSyscallMode(entry.Mode) inode := wfs.inodeToPath.Lookup(dirPath.Child(dirEntry.Name), entry.Crtime.Unix(), entry.IsDirectory(), len(entry.HardLinkId) > 0, entry.Inode, isPlusMode) dirEntry.Ino = inode if !isPlusMode { if !out.AddDirEntry(dirEntry) { isEarlyTerminated = true return false } } else { entryOut := out.AddDirLookupEntry(dirEntry) if entryOut == nil { isEarlyTerminated = true return false } 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().GetEntry()) } wfs.outputFilerEntry(entryOut, inode, entry) } return true }
if input.Offset < directoryStreamBaseOffset { if !isPlusMode { if input.Offset == 0 { out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "."}) } out.AddDirEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".."}) } else { if input.Offset == 0 { out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: "."}) } out.AddDirLookupEntry(fuse.DirEntry{Mode: fuse.S_IFDIR, Name: ".."}) } input.Offset = directoryStreamBaseOffset }
var lastEntryName string if input.Offset >= dh.entryStreamOffset { if input.Offset > dh.entryStreamOffset { entryPreviousIndex := (input.Offset - dh.entryStreamOffset) - 1 if uint64(len(dh.entryStream)) > entryPreviousIndex { lastEntryName = dh.entryStream[entryPreviousIndex].Name() dh.entryStream = dh.entryStream[entryPreviousIndex:] dh.entryStreamOffset = input.Offset - 1 } } entryCurrentIndex := input.Offset - dh.entryStreamOffset for uint64(len(dh.entryStream)) > entryCurrentIndex { entry := dh.entryStream[entryCurrentIndex] if processEachEntryFn(entry) { lastEntryName = entry.Name() entryCurrentIndex++ } else { // early terminated
return fuse.OK } } }
var err error if err = meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath); err != nil { glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) return fuse.EIO } listErr := wfs.metaCache.ListDirectoryEntries(context.Background(), dirPath, lastEntryName, false, int64(math.MaxInt32), func(entry *filer.Entry) bool { dh.entryStream = append(dh.entryStream, entry) return processEachEntryFn(entry) }) if listErr != nil { glog.Errorf("list meta cache: %v", listErr) return fuse.EIO }
if !isEarlyTerminated { dh.isFinished = true }
return fuse.OK }
|