lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-11-24 08:46:54 +08:00
Code Issues Actions 17 Packages Projects Releases Wiki Activity

Context cancellation during reading range reading large files (#7093)

Browse Source 
* context cancellation during reading range reading large files

* address comments

* cancellation for fuse read

* fix cancellation

* pass in context for each function to avoid racing condition

* Update reader_at_test.go

* remove dead code

* Update weed/filer/reader_at.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/filer/filechunk_group.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/filer/filechunk_group.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* address comments

* Update weed/mount/weedfs_file_read.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/mount/weedfs_file_lseek.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/mount/weedfs_file_read.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/filer/reader_at.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update weed/mount/weedfs_file_lseek.go

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* test cancellation

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
Chris Lu
2025-08-06 10:09:26 -07:00
committed by GitHub
parent e446234e9c
commit 4af182f880
14 changed files with 231 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
weed/mount/filehandle.go
View File

@@ -1,12 +1,13 @@
package mount
import (
"os"
"sync"
"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"
"os"
"sync"
)
type FileHandleId uint64

6
weed/mount/filehandle_read.go
View File

@@ -23,6 +23,10 @@ func (fh *FileHandle) readFromDirtyPages(buff []byte, startOffset int64, tsNs in
}
func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, int64, error) {
return fh.readFromChunksWithContext(context.Background(), buff, offset)
}
func (fh *FileHandle) readFromChunksWithContext(ctx context.Context, buff []byte, offset int64) (int64, int64, error) {
fh.entryLock.RLock()
defer fh.entryLock.RUnlock()
@@ -60,7 +64,7 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, int64, e
return int64(totalRead), 0, nil
}
totalRead, ts, err := fh.entryChunkGroup.ReadDataAt(fileSize, buff, offset)
totalRead, ts, err := fh.entryChunkGroup.ReadDataAt(ctx, fileSize, buff, offset)
if err != nil && err != io.EOF {
glog.Errorf("file handle read %s: %v", fileFullPath, err)

15
weed/mount/weedfs_file_lseek.go
View File

@@ -1,9 +1,11 @@
package mount
import (
"github.com/seaweedfs/seaweedfs/weed/util"
"context"
"syscall"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/hanwen/go-fuse/v2/fuse"
"github.com/seaweedfs/seaweedfs/weed/filer"
@@ -54,8 +56,17 @@ func (wfs *WFS) Lseek(cancel <-chan struct{}, in *fuse.LseekIn, out *fuse.LseekO
return ENXIO
}
// Create a context that will be cancelled when the cancel channel receives a signal
ctx, cancelFunc := context.WithCancel(context.Background())
go func() {
select {
case <-cancel:
cancelFunc()
}
}()
// search chunks for the offset
found, offset := fh.entryChunkGroup.SearchChunks(offset, fileSize, in.Whence)
found, offset := fh.entryChunkGroup.SearchChunks(ctx, offset, fileSize, in.Whence)
if found {
out.Offset = uint64(offset)
return fuse.OK

34
weed/mount/weedfs_file_read.go
View File

@@ -2,10 +2,12 @@ package mount
import (
"bytes"
"context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/util"
"io"
"github.com/seaweedfs/seaweedfs/weed/util"
"github.com/hanwen/go-fuse/v2/fuse"
"github.com/seaweedfs/seaweedfs/weed/glog"
@@ -45,8 +47,17 @@ func (wfs *WFS) Read(cancel <-chan struct{}, in *fuse.ReadIn, buff []byte) (fuse
fhActiveLock := fh.wfs.fhLockTable.AcquireLock("Read", fh.fh, util.SharedLock)
defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock)
// Create a context that will be cancelled when the cancel channel receives a signal
ctx, cancelFunc := context.WithCancel(context.Background())
go func() {
select {
case <-cancel:
cancelFunc()
}
}()
offset := int64(in.Offset)
totalRead, err := readDataByFileHandle(buff, fh, offset)
totalRead, err := readDataByFileHandleWithContext(ctx, buff, fh, offset)
if err != nil {
glog.Warningf("file handle read %s %d: %v", fh.FullPath(), totalRead, err)
return nil, fuse.EIO
@@ -59,7 +70,7 @@ func (wfs *WFS) Read(cancel <-chan struct{}, in *fuse.ReadIn, buff []byte) (fuse
if bytes.Compare(mirrorData, buff[:totalRead]) != 0 {
againBuff := make([]byte, len(buff))
againRead, _ := readDataByFileHandle(againBuff, fh, offset)
againRead, _ := readDataByFileHandleWithContext(ctx, againBuff, fh, offset)
againCorrect := bytes.Compare(mirrorData, againBuff[:againRead]) == 0
againSame := bytes.Compare(buff[:totalRead], againBuff[:againRead]) == 0
@@ -88,3 +99,20 @@ func readDataByFileHandle(buff []byte, fhIn *FileHandle, offset int64) (int64, e
}
return n, err
}
func readDataByFileHandleWithContext(ctx context.Context, buff []byte, fhIn *FileHandle, offset int64) (int64, error) {
// read data from source file
size := len(buff)
fhIn.lockForRead(offset, size)
defer fhIn.unlockForRead(offset, size)
n, tsNs, err := fhIn.readFromChunksWithContext(ctx, buff, offset)
if err == nil || err == io.EOF {
maxStop := fhIn.readFromDirtyPages(buff, offset, tsNs)
n = max(maxStop-offset, n)
}
if err == io.EOF {
err = nil
}
return n, err
}