diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 722df908b1..2672a49151 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -1,6 +1,7 @@ package mount import ( + "bytes" "os" "sync" "sync/atomic" @@ -222,23 +223,28 @@ func (fh *FileHandle) refreshChunkView() { }) } -// mergePeerContent folds a peer's content change into a dirty handle, so its -// own writes stay on top of the peer's instead of being discarded. Buffered -// writes are not chunks yet, so a merge that would shrink the file waits for -// the flush. +// mergePeerContent folds a peer's change into a dirty handle, so its own +// writes stay on top of the peer's instead of being discarded. func (fh *FileHandle) mergePeerContent(peer, base *filer_pb.Entry) bool { var local *filer_pb.Entry fh.UpdateEntry(func(e *filer_pb.Entry) { local = proto.Clone(e).(*filer_pb.Entry) }) merged, ok := mergeContentEntry(peer, base, local, local, fh.truncatedSince(base)) - ok = ok && merged.Attributes.FileSize >= filer.FileSize(local) + ok = ok && mergeKeepsBuffered(merged, local) if ok { fh.rebaseEntry(merged, local) } return ok } +// mergeKeepsBuffered reports whether merged leaves room for writes still +// buffered over local. They are not chunks yet, so a merge that shrinks the +// file or brings in inline content (read ahead of chunks) waits for the flush. +func mergeKeepsBuffered(merged, local *filer_pb.Entry) bool { + return merged.Attributes.FileSize >= filer.FileSize(local) && bytes.Equal(merged.Content, local.Content) +} + type truncateMark struct { base *filer_pb.Entry size int64 diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index 21efd000d4..b77f877f52 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -324,8 +324,7 @@ func (wfs *WFS) flushMetadataToFiler(ctx context.Context, fh *FileHandle, dir, n // When the stored entry moved on, the handle's own changes since its base are // replayed onto it and published against it, so a peer's acknowledged chunks // survive. merged is that entry in local ids; nil when the request went as is. -// With writes still buffered, a merge that would shrink the file is refused: -// the buffered range is not in local's chunks yet. +// With writes still buffered, the merge must keep room for them. func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request *filer_pb.CreateEntryRequest, local *filer_pb.Entry, buffered bool) (resp *filer_pb.CreateEntryResponse, merged *filer_pb.Entry, err error) { base := fh.baseEntry.Load() guard := &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS} @@ -358,7 +357,7 @@ func (wfs *WFS) createContentEntry(ctx context.Context, fh *FileHandle, request return resp, merged, err } next, ok := mergeContentEntry(stored, base, local, sent, fh.truncatedSince(base)) - if !ok || buffered && next.Attributes.FileSize < filer.FileSize(local) { + if !ok || buffered && !mergeKeepsBuffered(next, local) { return resp, nil, err } glog.V(1).Infof("content flush %s/%s: entry changed since the handle's base, merging", request.Directory, request.Entry.Name) diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index a7cfefdf36..78b9bb1569 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -2325,3 +2325,42 @@ func TestPeerChmodReachesDirtyHandle(t *testing.T) { }) } } + +// Inline content from a peer would be read ahead of the chunk a buffered +// write becomes, so it waits for the flush instead of merging. +func TestPeerInlineContentWaitsForBufferedWrite(t *testing.T) { + wfs := newInvalidateTestWFS(t) + + empty := func() *filer_pb.Entry { + e := flushConditionEntry() + e.Attributes.FileSize = 0 + return e + } + inode := wfs.inodeToPath.Lookup(util.FullPath("/dir/file"), time.Now().Unix(), false, false, 0, false) + fh, _, _, _ := wfs.fhMap.AcquireFileHandle(wfs, inode, empty(), 0, 0) + fh.UpdateEntry(func(e *filer_pb.Entry) { e.Attributes.FileSize = 4 }) + fh.dirtyMetadata = true + + peer := empty() + peer.Content, peer.Attributes.FileSize = []byte("peer"), 4 + resp := &filer_pb.SubscribeMetadataResponse{ + Directory: "/dir", + TsNs: 1500, + EventNotification: &filer_pb.EventNotification{ + OldEntry: empty(), + NewEntry: peer, + NewParentPath: "/dir", + }, + } + if err := wfs.metaCache.ApplyMetadataResponse(context.Background(), resp, meta_cache.SubscriberMetadataResponseApplyOptions); err != nil { + t.Fatalf("apply: %v", err) + } + wfs.metaCache.WaitForEntryInvalidations() + + if content := fh.GetEntry().GetEntry().Content; len(content) != 0 { + t.Fatalf("handle took the peer's inline content %q over a buffered write", content) + } + if base := fh.baseEntry.Load(); len(base.GetContent()) != 0 { + t.Fatalf("base %v moved to the peer's entry", base) + } +}