diff --git a/weed/mount/dirty_pages_chunked.go b/weed/mount/dirty_pages_chunked.go index 63852c2d8c..26a8d2aee6 100644 --- a/weed/mount/dirty_pages_chunked.go +++ b/weed/mount/dirty_pages_chunked.go @@ -54,19 +54,17 @@ func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *ChunkedDirtyPages { return dirtyPages } -func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte, isSequential bool, tsNs int64) error { +func (pages *ChunkedDirtyPages) AddPage(offset int64, data []byte, isSequential bool, tsNs int64) (n int, err error) { // A failed chunk upload already dropped its data, so this handle can never // flush cleanly. Reject further writes so the writer fails fast instead of // pushing the rest of the file through a pipeline that cannot persist it. if err := pages.LastError(); err != nil { - return err + return 0, err } pages.hasWrites = true glog.V(4).Infof("%v memory AddPage [%d, %d)", pages.fh.fh, offset, offset+int64(len(data))) - _, err := pages.uploadPipeline.SaveDataAt(data, offset, isSequential, tsNs) - - return err + return pages.uploadPipeline.SaveDataAt(data, offset, isSequential, tsNs) } func (pages *ChunkedDirtyPages) HasWrites() bool { diff --git a/weed/mount/dirty_pages_chunked_test.go b/weed/mount/dirty_pages_chunked_test.go index b63715609a..a0077a0db1 100644 --- a/weed/mount/dirty_pages_chunked_test.go +++ b/weed/mount/dirty_pages_chunked_test.go @@ -32,7 +32,7 @@ func TestChunkedDirtyPagesFailWritesAfterUploadError(t *testing.T) { uploadErr := errors.New("assign volume failure: no writable volumes") pages.setLastError(uploadErr) - if err := pages.AddPage(0, []byte("x"), true, 1); !errors.Is(err, uploadErr) { + if _, err := pages.AddPage(0, []byte("x"), true, 1); !errors.Is(err, uploadErr) { t.Fatalf("AddPage after upload failure = %v, want sticky %v", err, uploadErr) } if err := pages.FlushData(); !errors.Is(err, uploadErr) { diff --git a/weed/mount/page_writer.go b/weed/mount/page_writer.go index 81f197c737..5f690a30bd 100644 --- a/weed/mount/page_writer.go +++ b/weed/mount/page_writer.go @@ -15,10 +15,6 @@ type PageWriter struct { randomWriter page_writer.DirtyPages } -var ( - _ = page_writer.DirtyPages(&PageWriter{}) -) - func newPageWriter(fh *FileHandle, chunkSize int64) *PageWriter { pw := &PageWriter{ fh: fh, @@ -29,23 +25,27 @@ func newPageWriter(fh *FileHandle, chunkSize int64) *PageWriter { return pw } -func (pw *PageWriter) AddPage(offset int64, data []byte, isSequential bool, tsNs int64) error { +// AddPage returns how many leading bytes of data it buffered; on error those +// bytes stay buffered. +func (pw *PageWriter) AddPage(offset int64, data []byte, isSequential bool, tsNs int64) (buffered int64, err error) { glog.V(4).Infof("%v AddPage [%d, %d)", pw.fh.fh, offset, offset+int64(len(data))) chunkIndex := offset / pw.chunkSize for i := chunkIndex; len(data) > 0; i++ { writeSize := min(int64(len(data)), (i+1)*pw.chunkSize-offset) - if err := pw.addToOneChunk(i, offset, data[:writeSize], isSequential, tsNs); err != nil { - return err + n, err := pw.addToOneChunk(i, offset, data[:writeSize], isSequential, tsNs) + buffered += int64(n) + if err != nil { + return buffered, err } offset += writeSize data = data[writeSize:] } - return nil + return buffered, nil } -func (pw *PageWriter) addToOneChunk(chunkIndex, offset int64, data []byte, isSequential bool, tsNs int64) error { +func (pw *PageWriter) addToOneChunk(chunkIndex, offset int64, data []byte, isSequential bool, tsNs int64) (int, error) { return pw.randomWriter.AddPage(offset, data, isSequential, tsNs) } diff --git a/weed/mount/page_writer/dirty_pages.go b/weed/mount/page_writer/dirty_pages.go index 33ec7adb83..7d94d3c0ab 100644 --- a/weed/mount/page_writer/dirty_pages.go +++ b/weed/mount/page_writer/dirty_pages.go @@ -1,7 +1,7 @@ package page_writer type DirtyPages interface { - AddPage(offset int64, data []byte, isSequential bool, tsNs int64) error + AddPage(offset int64, data []byte, isSequential bool, tsNs int64) (n int, err error) FlushData() error HasWrites() bool ReadDirtyDataAt(data []byte, startOffset int64, tsNs int64) (maxStop int64) diff --git a/weed/mount/page_writer/upload_pipeline.go b/weed/mount/page_writer/upload_pipeline.go index 1af97fa258..4762f1b1c8 100644 --- a/weed/mount/page_writer/upload_pipeline.go +++ b/weed/mount/page_writer/upload_pipeline.go @@ -2,6 +2,7 @@ package page_writer import ( "fmt" + "io" "math" "sync" "sync/atomic" @@ -144,12 +145,15 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool, tsN pageChunk = NewMemChunk(logicChunkIndex, up.ChunkSize) // fmt.Printf(" create mem chunk %d\n", logicChunkIndex) } else { - pageChunk = up.swapFile.NewSwapFileChunk(logicChunkIndex) + // Check the *SwapFileChunk before it becomes a PageChunk: a nil + // pointer inside the interface would not compare equal to nil. + swapChunk := up.swapFile.NewSwapFileChunk(logicChunkIndex) // fmt.Printf(" create file chunk %d\n", logicChunkIndex) - if pageChunk == nil { + if swapChunk == nil { up.accountant.Release(up.ChunkSize) return 0, fmt.Errorf("failed to create swap file chunk") } + pageChunk = swapChunk } up.writableChunks[logicChunkIndex] = pageChunk } @@ -161,6 +165,9 @@ func (up *UploadPipeline) SaveDataAt(p []byte, off int64, isSequential bool, tsN //} n = pageChunk.WriteDataAt(p, off, tsNs) up.maybeMoveToSealed(pageChunk, logicChunkIndex) + if n < len(p) { + err = fmt.Errorf("chunk %d stored %d of %d bytes: %w", logicChunkIndex, n, len(p), io.ErrShortWrite) + } return } diff --git a/weed/mount/page_writer/upload_pipeline_swap_test.go b/weed/mount/page_writer/upload_pipeline_swap_test.go new file mode 100644 index 0000000000..17588c60af --- /dev/null +++ b/weed/mount/page_writer/upload_pipeline_swap_test.go @@ -0,0 +1,56 @@ +package page_writer + +import ( + "os" + "path/filepath" + "testing" +) + +// When the swap file cannot be created, SaveDataAt must fail the write +// instead of keeping a nil swap chunk and writing through it, and give back +// the write budget it reserved for the chunk. +func TestSaveDataAtFailsWhenSwapFileCannotBeCreated(t *testing.T) { + for _, tc := range []struct { + name string + accountant *WriteBufferAccountant + }{ + {name: "no write budget"}, + {name: "write budget", accountant: NewWriteBufferAccountant(1 << 20)}, + } { + t.Run(tc.name, func(t *testing.T) { + notADir := filepath.Join(t.TempDir(), "file") + if err := os.WriteFile(notADir, nil, 0600); err != nil { + t.Fatal(err) + } + up := NewUploadPipeline(nil, 1024, nil, 16, filepath.Join(notADir, "swap"), tc.accountant) + + // A write that is not sequential goes to the swap file. + n, err := up.SaveDataAt([]byte("data"), 0, false, 0) + if err == nil { + t.Fatalf("SaveDataAt = %d, nil; want an error", n) + } + if _, found := up.writableChunks[0]; found { + t.Error("the failed write left a writable chunk behind") + } + if used := tc.accountant.Used(); used != 0 { + t.Errorf("write budget still holds %d bytes", used) + } + }) + } +} + +// A swap chunk whose file write fails stores nothing, so SaveDataAt must +// report the short write instead of claiming the bytes. +func TestSaveDataAtReportsFailedSwapWrite(t *testing.T) { + up := NewUploadPipeline(nil, 1024, nil, 16, t.TempDir(), nil) + defer up.Shutdown() + if n, err := up.SaveDataAt([]byte("data"), 0, false, 0); err != nil || n != 4 { + t.Fatalf("first write = %d, %v", n, err) + } + if err := up.swapFile.file.Close(); err != nil { + t.Fatal(err) + } + if n, err := up.SaveDataAt([]byte("more"), 4, false, 0); err == nil || n != 0 { + t.Fatalf("write through a closed swap file = %d, %v; want 0 and an error", n, err) + } +} diff --git a/weed/mount/weedfs_file_copy_range.go b/weed/mount/weedfs_file_copy_range.go index 0497d23b6d..963c85cec1 100644 --- a/weed/mount/weedfs_file_copy_range.go +++ b/weed/mount/weedfs_file_copy_range.go @@ -180,19 +180,23 @@ func (wfs *WFS) CopyFileRange(cancel <-chan struct{}, in *fuse.CopyFileRangeIn) // Perform the write fhOut.dirtyPages.writerPattern.MonitorWriteAt(offsetOut, int(numBytesRead)) - if err := fhOut.dirtyPages.AddPage( + buffered, err := fhOut.dirtyPages.AddPage( offsetOut, buff[:numBytesRead], fhOut.dirtyPages.writerPattern.IsSequentialMode(), - nowUnixNano); err != nil { + nowUnixNano) + if buffered > 0 { + fhOut.pageWrites.Add(1) + totalCopied += buffered + fhOut.dirtyContent = true + } + if err != nil { glog.Errorf("AddPage error: %v", err) - return 0, writeErrorToFuseStatus(err) + if totalCopied == 0 { + return 0, writeErrorToFuseStatus(err) + } + break } - fhOut.pageWrites.Add(1) - - // Accumulate for the next loop iteration - totalCopied += numBytesRead - fhOut.dirtyContent = true } if totalCopied == 0 { diff --git a/weed/mount/weedfs_file_copy_range_test.go b/weed/mount/weedfs_file_copy_range_test.go index 6ea5471c89..ac7cfd6bfc 100644 --- a/weed/mount/weedfs_file_copy_range_test.go +++ b/weed/mount/weedfs_file_copy_range_test.go @@ -1,8 +1,10 @@ package mount import ( + "bytes" "context" "errors" + "os" "path/filepath" "testing" @@ -408,6 +410,47 @@ func TestStreamingCopyMarksOnlyAcceptedContentDirty(t *testing.T) { } } +// Bytes the copy buffered before a page failed stay in the handle, so the +// copy reports them and grows the destination over them. +func TestStreamingCopyReportsPartlyBufferedCopy(t *testing.T) { + wfs := newCopyRangeTestWFS() + wfs.option.ConcurrentWriters = 1 + notADir := filepath.Join(t.TempDir(), "file") + if err := os.WriteFile(notADir, nil, 0600); err != nil { + t.Fatal(err) + } + wfs.option.uniqueCacheDirForWrite = filepath.Join(notADir, "swap") + srcInode := wfs.inodeToPath.Lookup(util.FullPath("/src.txt"), 1, false, false, 0, true) + dstInode := wfs.inodeToPath.Lookup(util.FullPath("/dst.txt"), 1, false, false, 0, true) + srcHandle, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, srcInode, &filer_pb.Entry{ + Name: "src.txt", + Attributes: &filer_pb.FuseAttributes{FileMode: 0100644, FileSize: 1024, Inode: srcInode}, + Content: bytes.Repeat([]byte{'s'}, 1024), + }, 0, 0) + dstHandle, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, dstInode, &filer_pb.Entry{ + Name: "dst.txt", + Attributes: &filer_pb.FuseAttributes{FileMode: 0100644, Inode: dstInode}, + }, 0, 0) + t.Cleanup(dstHandle.dirtyPages.Destroy) + + // The destination range crosses into a second chunk, which needs the swap file. + written, status := wfs.CopyFileRange(make(chan struct{}), &fuse.CopyFileRangeIn{ + FhIn: uint64(srcHandle.fh), + FhOut: uint64(dstHandle.fh), + OffOut: 512, + Len: 1024, + }) + if status != fuse.OK || written != 512 { + t.Fatalf("CopyFileRange = %d, %v; want a short copy of 512", written, status) + } + if size := dstHandle.GetEntry().GetEntry().GetAttributes().GetFileSize(); size != 1024 { + t.Fatalf("destination size = %d, want 1024", size) + } + if !dstHandle.dirtyContent || !dstHandle.dirtyMetadata { + t.Fatal("the buffered prefix must leave the destination dirty") + } +} + func TestServerSideCopySkipsFencedDestination(t *testing.T) { wfs := newCopyRangeTestWFS() wfs.option.WritebackCache = true diff --git a/weed/mount/weedfs_file_mkrm_test.go b/weed/mount/weedfs_file_mkrm_test.go index b9c855b69e..2542cdd746 100644 --- a/weed/mount/weedfs_file_mkrm_test.go +++ b/weed/mount/weedfs_file_mkrm_test.go @@ -270,7 +270,7 @@ func TestTruncateEntryClearsDirtyPagesForOpenHandle(t *testing.T) { fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, entry, 0, 0) fh.RememberPath(fullPath) - if err := fh.dirtyPages.AddPage(0, []byte("hello"), true, time.Now().UnixNano()); err != nil { + if _, err := fh.dirtyPages.AddPage(0, []byte("hello"), true, time.Now().UnixNano()); err != nil { t.Fatalf("AddPage: %v", err) } oldDirtyPages := fh.dirtyPages diff --git a/weed/mount/weedfs_file_write.go b/weed/mount/weedfs_file_write.go index a630b8984a..2ec8b79b4b 100644 --- a/weed/mount/weedfs_file_write.go +++ b/weed/mount/weedfs_file_write.go @@ -68,6 +68,21 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr return 0, fuse.OK } + // Size and times cover only the bytes buffered, or the next flush + // publishes a hole where the data should be. A partly buffered write + // is reported as a short write, or as an error to kernel writeback, + // which ignores the count. + offset := int64(in.Offset) + buffered, err := fh.dirtyPages.AddPage(offset, data, fh.dirtyPages.writerPattern.IsSequentialMode(), tsNs) + if err != nil { + glog.Errorf("AddPage error: %v", err) + if buffered == 0 { + return 0, writeErrorToFuseStatus(err) + } + data = data[:buffered] + } + fh.pageWrites.Add(1) + // Only an append past all locally owned bytes needs the stored EOF guard. if in.Offset > entry.Attributes.FileSize && wfs.isAppendWrite(in) { if !fh.pendingAppend || in.Offset < fh.pendingAppendOffset { @@ -76,7 +91,6 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr fh.pendingAppend = true } entry.Content = nil - offset := int64(in.Offset) oldFileSize := int64(entry.Attributes.FileSize) newFileSize := max(offset+int64(len(data)), oldFileSize) entry.Attributes.FileSize = uint64(newFileSize) @@ -96,14 +110,6 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr wfs.AddUncommittedBytes(newFileSize - oldFileSize) } - // glog.V(4).Infof("%v write [%d,%d) %d", fh.f.fullpath(), req.Offset, req.Offset+int64(len(req.Data)), len(req.Data)) - - if err := fh.dirtyPages.AddPage(offset, data, fh.dirtyPages.writerPattern.IsSequentialMode(), tsNs); err != nil { - glog.Errorf("AddPage error: %v", err) - return 0, writeErrorToFuseStatus(err) - } - fh.pageWrites.Add(1) - written = uint32(len(data)) if offset == 0 { @@ -127,5 +133,8 @@ func (wfs *WFS) Write(cancel <-chan struct{}, in *fuse.WriteIn, data []byte) (wr fh.mirrorFile.WriteAt(data, offset) } + if err != nil && in.WriteFlags&fuse.WRITE_CACHE != 0 { + return 0, writeErrorToFuseStatus(err) + } return written, fuse.OK } diff --git a/weed/mount/weedfs_file_write_test.go b/weed/mount/weedfs_file_write_test.go new file mode 100644 index 0000000000..320c3ecd7f --- /dev/null +++ b/weed/mount/weedfs_file_write_test.go @@ -0,0 +1,95 @@ +//go:build !freebsd + +package mount + +import ( + "bytes" + "context" + "os" + "path/filepath" + "syscall" + "testing" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/protobuf/proto" +) + +// A write the swap file cannot buffer fails, and must not grow the file: the +// next flush would publish that size with a hole where its data should be. A +// write that buffers only its first chunk is a short write of those bytes, or +// an error for kernel writeback, which would drop the rest silently. +func TestFailedWriteDoesNotGrowFile(t *testing.T) { + t.Run("short write", func(t *testing.T) { testFailedWriteDoesNotGrowFile(t, 0) }) + t.Run("kernel writeback", func(t *testing.T) { testFailedWriteDoesNotGrowFile(t, fuse.WRITE_CACHE) }) +} + +func testFailedWriteDoesNotGrowFile(t *testing.T, writeFlags uint32) { + base := attributeAppendEntry(1) + base.CommitTsNs = 100 + peer := &g6ConditionalPeer{entry: proto.Clone(base).(*filer_pb.Entry)} + peer.blobs = map[string][]byte{base.Chunks[0].GetFileIdString(): bytes.Repeat([]byte{'O'}, 256)} + + w := newInvalidateTestWFS(t) + notADir := filepath.Join(t.TempDir(), "file") + if err := os.WriteFile(notADir, nil, 0600); err != nil { + t.Fatal(err) + } + w.option.uniqueCacheDirForWrite = filepath.Join(notADir, "swap") + w.option.DefaultPermissions = true + w.posixLocks = NewPosixLockTable() + w.option.ConcurrentWriters = 1 + w.concurrentWriters = util.NewLimitedConcurrentExecutor(1) + startFakeFiler(t, w, peer) + w.inodeToPath.Lookup("/dir", 1, true, false, 0, false) + w.inodeToPath.Lookup("/dir/file", 1, false, false, 42, true) + w.inodeToPath.MarkChildrenCached("/dir") + if err := w.metaCache.InsertEntry(context.Background(), filer.FromPbEntry("/dir", proto.Clone(base).(*filer_pb.Entry)), 1); err != nil { + t.Fatal(err) + } + out := &fuse.OpenOut{} + if code := w.Open(nil, &fuse.OpenIn{InHeader: fuse.InHeader{NodeId: 42}, Flags: syscall.O_WRONLY}, out); code != fuse.OK { + t.Fatal(code) + } + fh := w.GetHandle(FileHandleId(out.Fh)) + g6AppendUploader(t, fh, 100, peer) + + write := func(offset uint64, size int, value byte) (uint32, fuse.Status) { + return w.Write(nil, &fuse.WriteIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh, Offset: offset, Size: uint32(size), WriteFlags: writeFlags}, bytes.Repeat([]byte{value}, size)) + } + if _, code := write(256, 256, 'A'); code != fuse.OK { + t.Fatalf("buffered write: %v", code) + } + // The only writable chunk slot is taken, so this write needs the swap file. + if _, code := write(4096, 256, 'B'); code == fuse.OK { + t.Fatal("write through an unusable swap directory succeeded") + } + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 512 { + t.Fatalf("file size after the failed write = %d, want 512", size) + } + // The first 24 bytes fit the existing chunk; the next chunk needs the swap file. + n, code := write(1000, 48, 'C') + if writeFlags == 0 && (code != fuse.OK || n != 24) { + t.Fatalf("boundary-crossing write = %d, %v; want a short write of 24", n, code) + } + if writeFlags != 0 && code == fuse.OK { + t.Fatalf("boundary-crossing writeback = %d, OK; want an error", n) + } + if size := fh.GetEntry().GetEntry().Attributes.FileSize; size != 1024 { + t.Fatalf("file size after the short write = %d, want 1024", size) + } + if code := w.Fsync(nil, &fuse.FsyncIn{InHeader: fuse.InHeader{NodeId: 42}, Fh: out.Fh}); code != fuse.OK { + t.Fatal(code) + } + + peer.mu.Lock() + defer peer.mu.Unlock() + want := append(bytes.Repeat([]byte{'O'}, 256), bytes.Repeat([]byte{'A'}, 256)...) + want = append(want, make([]byte, 1000-512)...) + want = append(want, bytes.Repeat([]byte{'C'}, 24)...) + if got := g6StoredBytes(t, peer); peer.entry.Attributes.FileSize != 1024 || !bytes.Equal(got, want) { + t.Fatalf("stored size %d, want 1024 with the original and buffered bytes", peer.entry.Attributes.FileSize) + } +} diff --git a/weed/mount/weedfs_flush_deadline_test.go b/weed/mount/weedfs_flush_deadline_test.go index 8272118fc4..bc0b0f0ad4 100644 --- a/weed/mount/weedfs_flush_deadline_test.go +++ b/weed/mount/weedfs_flush_deadline_test.go @@ -121,7 +121,7 @@ func newSlowDataFlush(t *testing.T, assignDelay, deadline time.Duration, stuckCo Chunks: stored, }, 0, 0) t.Cleanup(func() { wfs.ReleaseHandle(fh.fh) }) - if err := fh.dirtyPages.AddPage(offset, data, true, time.Now().UnixNano()); err != nil { + if _, err := fh.dirtyPages.AddPage(offset, data, true, time.Now().UnixNano()); err != nil { t.Fatalf("AddPage: %v", err) } fh.dirtyMetadata = true