properly lock file.entry object
fix https://github.com/chrislusf/seaweedfs/issues/1882
This commit is contained in:
@@ -35,15 +35,20 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// update old file to hardlink mode
|
oldEntry := oldFile.getEntry()
|
||||||
if len(oldFile.entry.HardLinkId) == 0 {
|
if oldEntry == nil {
|
||||||
oldFile.entry.HardLinkId = append(util.RandomBytes(16), HARD_LINK_MARKER)
|
return nil, fuse.EIO
|
||||||
oldFile.entry.HardLinkCounter = 1
|
|
||||||
}
|
}
|
||||||
oldFile.entry.HardLinkCounter++
|
|
||||||
|
// update old file to hardlink mode
|
||||||
|
if len(oldEntry.HardLinkId) == 0 {
|
||||||
|
oldEntry.HardLinkId = append(util.RandomBytes(16), HARD_LINK_MARKER)
|
||||||
|
oldEntry.HardLinkCounter = 1
|
||||||
|
}
|
||||||
|
oldEntry.HardLinkCounter++
|
||||||
updateOldEntryRequest := &filer_pb.UpdateEntryRequest{
|
updateOldEntryRequest := &filer_pb.UpdateEntryRequest{
|
||||||
Directory: oldFile.dir.FullPath(),
|
Directory: oldFile.dir.FullPath(),
|
||||||
Entry: oldFile.entry,
|
Entry: oldEntry,
|
||||||
Signatures: []int32{dir.wfs.signature},
|
Signatures: []int32{dir.wfs.signature},
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -53,11 +58,11 @@ func (dir *Dir) Link(ctx context.Context, req *fuse.LinkRequest, old fs.Node) (f
|
|||||||
Entry: &filer_pb.Entry{
|
Entry: &filer_pb.Entry{
|
||||||
Name: req.NewName,
|
Name: req.NewName,
|
||||||
IsDirectory: false,
|
IsDirectory: false,
|
||||||
Attributes: oldFile.entry.Attributes,
|
Attributes: oldEntry.Attributes,
|
||||||
Chunks: oldFile.entry.Chunks,
|
Chunks: oldEntry.Chunks,
|
||||||
Extended: oldFile.entry.Extended,
|
Extended: oldEntry.Extended,
|
||||||
HardLinkId: oldFile.entry.HardLinkId,
|
HardLinkId: oldEntry.HardLinkId,
|
||||||
HardLinkCounter: oldFile.entry.HardLinkCounter,
|
HardLinkCounter: oldEntry.HardLinkCounter,
|
||||||
},
|
},
|
||||||
Signatures: []int32{dir.wfs.signature},
|
Signatures: []int32{dir.wfs.signature},
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -30,7 +30,7 @@ func newDirtyPages(file *File) *ContinuousDirtyPages {
|
|||||||
|
|
||||||
func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) {
|
func (pages *ContinuousDirtyPages) AddPage(offset int64, data []byte) {
|
||||||
|
|
||||||
glog.V(4).Infof("%s AddPage [%d,%d) of %d bytes", pages.f.fullpath(), offset, offset+int64(len(data)), pages.f.entry.Attributes.FileSize)
|
glog.V(4).Infof("%s AddPage [%d,%d)", pages.f.fullpath(), offset, offset+int64(len(data)))
|
||||||
|
|
||||||
if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
|
if len(data) > int(pages.f.wfs.option.ChunkSizeLimit) {
|
||||||
// this is more than what buffer can hold.
|
// this is more than what buffer can hold.
|
||||||
@@ -69,7 +69,12 @@ func (pages *ContinuousDirtyPages) saveExistingLargestPageToStorage() (hasSavedD
|
|||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
fileSize := int64(pages.f.entry.Attributes.FileSize)
|
entry := pages.f.getEntry()
|
||||||
|
if entry == nil {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
fileSize := int64(entry.Attributes.FileSize)
|
||||||
|
|
||||||
chunkSize := min(maxList.Size(), fileSize-maxList.Offset())
|
chunkSize := min(maxList.Size(), fileSize-maxList.Offset())
|
||||||
if chunkSize == 0 {
|
if chunkSize == 0 {
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"os"
|
"os"
|
||||||
"sort"
|
"sort"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/seaweedfs/fuse"
|
"github.com/seaweedfs/fuse"
|
||||||
@@ -33,6 +34,7 @@ type File struct {
|
|||||||
dir *Dir
|
dir *Dir
|
||||||
wfs *WFS
|
wfs *WFS
|
||||||
entry *filer_pb.Entry
|
entry *filer_pb.Entry
|
||||||
|
entryLock sync.RWMutex
|
||||||
entryViewCache []filer.VisibleInterval
|
entryViewCache []filer.VisibleInterval
|
||||||
isOpen int
|
isOpen int
|
||||||
reader io.ReaderAt
|
reader io.ReaderAt
|
||||||
@@ -47,7 +49,7 @@ func (file *File) Attr(ctx context.Context, attr *fuse.Attr) (err error) {
|
|||||||
|
|
||||||
glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
|
glog.V(4).Infof("file Attr %s, open:%v existing:%v", file.fullpath(), file.isOpen, attr)
|
||||||
|
|
||||||
entry := file.entry
|
entry := file.getEntry()
|
||||||
if file.isOpen <= 0 || entry == nil {
|
if file.isOpen <= 0 || entry == nil {
|
||||||
if entry, err = file.maybeLoadEntry(ctx); err != nil {
|
if entry, err = file.maybeLoadEntry(ctx); err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -258,7 +260,7 @@ func (file *File) Forget() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
|
func (file *File) maybeLoadEntry(ctx context.Context) (entry *filer_pb.Entry, err error) {
|
||||||
entry = file.entry
|
entry = file.getEntry()
|
||||||
if file.isOpen > 0 {
|
if file.isOpen > 0 {
|
||||||
return entry, nil
|
return entry, nil
|
||||||
}
|
}
|
||||||
@@ -299,8 +301,13 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
entry := file.getEntry()
|
||||||
|
if entry == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// pick out-of-order chunks from existing chunks
|
// pick out-of-order chunks from existing chunks
|
||||||
for _, chunk := range file.entry.Chunks {
|
for _, chunk := range entry.Chunks {
|
||||||
if lessThan(earliestChunk, chunk) {
|
if lessThan(earliestChunk, chunk) {
|
||||||
chunks = append(chunks, chunk)
|
chunks = append(chunks, chunk)
|
||||||
}
|
}
|
||||||
@@ -318,18 +325,22 @@ func (file *File) addChunks(chunks []*filer_pb.FileChunk) {
|
|||||||
|
|
||||||
file.reader = nil
|
file.reader = nil
|
||||||
|
|
||||||
glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(file.entry.Chunks), len(chunks))
|
glog.V(4).Infof("%s existing %d chunks adds %d more", file.fullpath(), len(entry.Chunks), len(chunks))
|
||||||
|
|
||||||
file.entry.Chunks = append(file.entry.Chunks, newChunks...)
|
entry.Chunks = append(entry.Chunks, newChunks...)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (file *File) setEntry(entry *filer_pb.Entry) {
|
func (file *File) setEntry(entry *filer_pb.Entry) {
|
||||||
|
file.entryLock.Lock()
|
||||||
|
defer file.entryLock.Unlock()
|
||||||
file.entry = entry
|
file.entry = entry
|
||||||
file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
|
file.entryViewCache, _ = filer.NonOverlappingVisibleIntervals(file.wfs.LookupFn(), entry.Chunks)
|
||||||
file.reader = nil
|
file.reader = nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (file *File) clearEntry() {
|
func (file *File) clearEntry() {
|
||||||
|
file.entryLock.Lock()
|
||||||
|
defer file.entryLock.Unlock()
|
||||||
file.entry = nil
|
file.entry = nil
|
||||||
file.entryViewCache = nil
|
file.entryViewCache = nil
|
||||||
file.reader = nil
|
file.reader = nil
|
||||||
@@ -359,3 +370,9 @@ func (file *File) saveEntry(entry *filer_pb.Entry) error {
|
|||||||
return nil
|
return nil
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (file *File) getEntry() *filer_pb.Entry {
|
||||||
|
file.entryLock.RLock()
|
||||||
|
defer file.entryLock.RUnlock()
|
||||||
|
return file.entry
|
||||||
|
}
|
||||||
|
|||||||
@@ -40,8 +40,9 @@ func newFileHandle(file *File, uid, gid uint32) *FileHandle {
|
|||||||
Uid: uid,
|
Uid: uid,
|
||||||
Gid: gid,
|
Gid: gid,
|
||||||
}
|
}
|
||||||
if fh.f.entry != nil {
|
entry := fh.f.getEntry()
|
||||||
fh.f.entry.Attributes.FileSize = filer.FileSize(fh.f.entry)
|
if entry != nil {
|
||||||
|
entry.Attributes.FileSize = filer.FileSize(entry)
|
||||||
}
|
}
|
||||||
|
|
||||||
return fh
|
return fh
|
||||||
@@ -104,22 +105,27 @@ func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64) (maxSto
|
|||||||
|
|
||||||
func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
|
func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
|
||||||
|
|
||||||
fileSize := int64(filer.FileSize(fh.f.entry))
|
entry := fh.f.getEntry()
|
||||||
|
if entry == nil {
|
||||||
|
return 0, io.EOF
|
||||||
|
}
|
||||||
|
|
||||||
|
fileSize := int64(filer.FileSize(entry))
|
||||||
|
|
||||||
if fileSize == 0 {
|
if fileSize == 0 {
|
||||||
glog.V(1).Infof("empty fh %v", fh.f.fullpath())
|
glog.V(1).Infof("empty fh %v", fh.f.fullpath())
|
||||||
return 0, io.EOF
|
return 0, io.EOF
|
||||||
}
|
}
|
||||||
|
|
||||||
if offset+int64(len(buff)) <= int64(len(fh.f.entry.Content)) {
|
if offset+int64(len(buff)) <= int64(len(entry.Content)) {
|
||||||
totalRead := copy(buff, fh.f.entry.Content[offset:])
|
totalRead := copy(buff, entry.Content[offset:])
|
||||||
glog.V(4).Infof("file handle read cached %s [%d,%d] %d", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead)
|
glog.V(4).Infof("file handle read cached %s [%d,%d] %d", fh.f.fullpath(), offset, offset+int64(totalRead), totalRead)
|
||||||
return int64(totalRead), nil
|
return int64(totalRead), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
var chunkResolveErr error
|
var chunkResolveErr error
|
||||||
if fh.f.entryViewCache == nil {
|
if fh.f.entryViewCache == nil {
|
||||||
fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), fh.f.entry.Chunks)
|
fh.f.entryViewCache, chunkResolveErr = filer.NonOverlappingVisibleIntervals(fh.f.wfs.LookupFn(), entry.Chunks)
|
||||||
if chunkResolveErr != nil {
|
if chunkResolveErr != nil {
|
||||||
return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
|
return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
|
||||||
}
|
}
|
||||||
@@ -158,8 +164,13 @@ func (fh *FileHandle) Write(ctx context.Context, req *fuse.WriteRequest, resp *f
|
|||||||
copy(data, req.Data)
|
copy(data, req.Data)
|
||||||
}
|
}
|
||||||
|
|
||||||
fh.f.entry.Content = nil
|
entry := fh.f.getEntry()
|
||||||
fh.f.entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(fh.f.entry.Attributes.FileSize)))
|
if entry == nil {
|
||||||
|
return fuse.EIO
|
||||||
|
}
|
||||||
|
|
||||||
|
entry.Content = nil
|
||||||
|
entry.Attributes.FileSize = uint64(max(req.Offset+int64(len(data)), int64(entry.Attributes.FileSize)))
|
||||||
glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
|
glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data))
|
||||||
|
|
||||||
fh.dirtyPages.AddPage(req.Offset, data)
|
fh.dirtyPages.AddPage(req.Offset, data)
|
||||||
@@ -242,35 +253,40 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
|
|||||||
|
|
||||||
err := fh.f.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
err := fh.f.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
|
||||||
|
|
||||||
if fh.f.entry.Attributes != nil {
|
entry := fh.f.getEntry()
|
||||||
fh.f.entry.Attributes.Mime = fh.contentType
|
if entry == nil {
|
||||||
if fh.f.entry.Attributes.Uid == 0 {
|
return nil
|
||||||
fh.f.entry.Attributes.Uid = header.Uid
|
}
|
||||||
|
|
||||||
|
if entry.Attributes != nil {
|
||||||
|
entry.Attributes.Mime = fh.contentType
|
||||||
|
if entry.Attributes.Uid == 0 {
|
||||||
|
entry.Attributes.Uid = header.Uid
|
||||||
}
|
}
|
||||||
if fh.f.entry.Attributes.Gid == 0 {
|
if entry.Attributes.Gid == 0 {
|
||||||
fh.f.entry.Attributes.Gid = header.Gid
|
entry.Attributes.Gid = header.Gid
|
||||||
}
|
}
|
||||||
if fh.f.entry.Attributes.Crtime == 0 {
|
if entry.Attributes.Crtime == 0 {
|
||||||
fh.f.entry.Attributes.Crtime = time.Now().Unix()
|
entry.Attributes.Crtime = time.Now().Unix()
|
||||||
}
|
}
|
||||||
fh.f.entry.Attributes.Mtime = time.Now().Unix()
|
entry.Attributes.Mtime = time.Now().Unix()
|
||||||
fh.f.entry.Attributes.FileMode = uint32(os.FileMode(fh.f.entry.Attributes.FileMode) &^ fh.f.wfs.option.Umask)
|
entry.Attributes.FileMode = uint32(os.FileMode(entry.Attributes.FileMode) &^ fh.f.wfs.option.Umask)
|
||||||
fh.f.entry.Attributes.Collection = fh.dirtyPages.collection
|
entry.Attributes.Collection = fh.dirtyPages.collection
|
||||||
fh.f.entry.Attributes.Replication = fh.dirtyPages.replication
|
entry.Attributes.Replication = fh.dirtyPages.replication
|
||||||
}
|
}
|
||||||
|
|
||||||
request := &filer_pb.CreateEntryRequest{
|
request := &filer_pb.CreateEntryRequest{
|
||||||
Directory: fh.f.dir.FullPath(),
|
Directory: fh.f.dir.FullPath(),
|
||||||
Entry: fh.f.entry,
|
Entry: entry,
|
||||||
Signatures: []int32{fh.f.wfs.signature},
|
Signatures: []int32{fh.f.wfs.signature},
|
||||||
}
|
}
|
||||||
|
|
||||||
glog.V(4).Infof("%s set chunks: %v", fh.f.fullpath(), len(fh.f.entry.Chunks))
|
glog.V(4).Infof("%s set chunks: %v", fh.f.fullpath(), len(entry.Chunks))
|
||||||
for i, chunk := range fh.f.entry.Chunks {
|
for i, chunk := range entry.Chunks {
|
||||||
glog.V(4).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.GetFileIdString(), chunk.Offset, chunk.Offset+int64(chunk.Size))
|
glog.V(4).Infof("%s chunks %d: %v [%d,%d)", fh.f.fullpath(), i, chunk.GetFileIdString(), chunk.Offset, chunk.Offset+int64(chunk.Size))
|
||||||
}
|
}
|
||||||
|
|
||||||
manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(fh.f.entry.Chunks)
|
manifestChunks, nonManifestChunks := filer.SeparateManifestChunks(entry.Chunks)
|
||||||
|
|
||||||
chunks, _ := filer.CompactFileChunks(fh.f.wfs.LookupFn(), nonManifestChunks)
|
chunks, _ := filer.CompactFileChunks(fh.f.wfs.LookupFn(), nonManifestChunks)
|
||||||
chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
|
chunks, manifestErr := filer.MaybeManifestize(fh.f.wfs.saveDataAsChunk(fh.f.fullpath()), chunks)
|
||||||
@@ -278,7 +294,7 @@ func (fh *FileHandle) doFlush(ctx context.Context, header fuse.Header) error {
|
|||||||
// not good, but should be ok
|
// not good, but should be ok
|
||||||
glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
|
glog.V(0).Infof("MaybeManifestize: %v", manifestErr)
|
||||||
}
|
}
|
||||||
fh.f.entry.Chunks = append(chunks, manifestChunks...)
|
entry.Chunks = append(chunks, manifestChunks...)
|
||||||
|
|
||||||
fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
|
fh.f.wfs.mapPbIdFromLocalToFiler(request.Entry)
|
||||||
defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)
|
defer fh.f.wfs.mapPbIdFromFilerToLocal(request.Entry)
|
||||||
|
|||||||
@@ -124,8 +124,9 @@ func (c *FsCache) Move(oldPath util.FullPath, newPath util.FullPath) *FsNode {
|
|||||||
}
|
}
|
||||||
if f, ok := src.node.(*File); ok {
|
if f, ok := src.node.(*File); ok {
|
||||||
f.Name = target.name
|
f.Name = target.name
|
||||||
if f.entry != nil {
|
entry := f.getEntry()
|
||||||
f.entry.Name = f.Name
|
if entry != nil {
|
||||||
|
entry.Name = f.Name
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
parent.disconnectChild(target)
|
parent.disconnectChild(target)
|
||||||
|
|||||||
Reference in New Issue
Block a user