seaweedfs/weed/mount/filehandle.go

126 lines
2.8 KiB
Go
Raw Normal View History

2022-02-14 06:50:44 +00:00
package mount
import (
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/util"
2022-12-05 07:33:05 +00:00
"golang.org/x/exp/slices"
"golang.org/x/sync/semaphore"
"math"
"sync"
2022-02-14 06:50:44 +00:00
)
type FileHandleId uint64
type FileHandle struct {
fh FileHandleId
counter int64
entry *LockedEntry
entryLock sync.Mutex
inode uint64
wfs *WFS
2022-02-14 06:50:44 +00:00
// cache file has been written to
2022-02-14 07:27:11 +00:00
dirtyMetadata bool
2022-02-14 06:50:44 +00:00
dirtyPages *PageWriter
entryViewCache []filer.VisibleInterval
2022-07-07 18:49:40 +00:00
reader *filer.ChunkReadAt
2022-02-14 06:50:44 +00:00
contentType string
handle uint64
orderedMutex *semaphore.Weighted
2022-02-14 06:50:44 +00:00
isDeleted bool
}
func newFileHandle(wfs *WFS, handleId FileHandleId, inode uint64, entry *filer_pb.Entry) *FileHandle {
fh := &FileHandle{
fh: handleId,
counter: 1,
inode: inode,
wfs: wfs,
orderedMutex: semaphore.NewWeighted(int64(math.MaxInt64)),
2022-02-14 06:50:44 +00:00
}
// dirtyPages: newContinuousDirtyPages(file, writeOnly),
fh.dirtyPages = newPageWriter(fh, wfs.option.ChunkSizeLimit)
if entry != nil {
entry.Attributes.FileSize = filer.FileSize(entry)
}
2022-12-05 07:33:05 +00:00
fh.entry = &LockedEntry{
Entry: entry,
}
2022-02-14 06:50:44 +00:00
return fh
}
func (fh *FileHandle) FullPath() util.FullPath {
fp, _ := fh.wfs.inodeToPath.GetPath(fh.inode)
return fp
2022-02-14 06:50:44 +00:00
}
func (fh *FileHandle) GetEntry() *filer_pb.Entry {
2022-12-05 07:33:05 +00:00
return fh.entry.GetEntry()
}
func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) {
2022-12-05 07:33:05 +00:00
fh.entry.SetEntry(entry)
}
func (fh *FileHandle) UpdateEntry(fn func(entry *filer_pb.Entry)) *filer_pb.Entry {
2022-12-05 07:33:05 +00:00
return fh.entry.UpdateEntry(fn)
}
func (fh *FileHandle) AddChunks(chunks []*filer_pb.FileChunk) {
fh.entryLock.Lock()
defer fh.entryLock.Unlock()
2022-02-14 06:50:44 +00:00
if fh.entry == nil {
return
}
2022-02-14 06:50:44 +00:00
// find the earliest incoming chunk
newChunks := chunks
earliestChunk := newChunks[0]
for i := 1; i < len(newChunks); i++ {
if lessThan(earliestChunk, newChunks[i]) {
earliestChunk = newChunks[i]
}
}
// pick out-of-order chunks from existing chunks
for _, chunk := range fh.entry.GetChunks() {
2022-02-14 06:50:44 +00:00
if lessThan(earliestChunk, chunk) {
chunks = append(chunks, chunk)
}
}
// sort incoming chunks
slices.SortFunc(chunks, func(a, b *filer_pb.FileChunk) bool {
return lessThan(a, b)
2022-02-14 06:50:44 +00:00
})
glog.V(4).Infof("%s existing %d chunks adds %d more", fh.FullPath(), len(fh.entry.GetChunks()), len(chunks))
2022-02-14 06:50:44 +00:00
2022-12-05 07:33:05 +00:00
fh.entry.AppendChunks(newChunks)
2022-02-14 06:50:44 +00:00
fh.entryViewCache = nil
}
2022-07-08 08:04:15 +00:00
func (fh *FileHandle) CloseReader() {
2022-07-07 18:49:40 +00:00
if fh.reader != nil {
_ = fh.reader.Close()
fh.reader = nil
2022-07-07 18:49:40 +00:00
}
}
2022-03-07 22:01:24 +00:00
func (fh *FileHandle) Release() {
fh.dirtyPages.Destroy()
2022-07-08 08:04:15 +00:00
fh.CloseReader()
2022-03-07 22:01:24 +00:00
}
2022-02-14 06:50:44 +00:00
func lessThan(a, b *filer_pb.FileChunk) bool {
if a.ModifiedTsNs == b.ModifiedTsNs {
2022-02-14 06:50:44 +00:00
return a.Fid.FileKey < b.Fid.FileKey
}
return a.ModifiedTsNs < b.ModifiedTsNs
2022-02-14 06:50:44 +00:00
}