mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-21 22:12:36 +00:00
mount: count only wire silence against the rename timeout
Applying a rename event can outlast the silence bound on a healthy filer — a directory rename migrates an open child's DLM lock under another writer. Stop the mux's response timer for the local work and re-arm it before the next wait; the fallback stream's watchdog defers while the handler runs instead of cancelling a live stream.
This commit is contained in:
@@ -6,6 +6,7 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
@@ -33,10 +34,21 @@ func (wfs *WFS) doRename(ctx context.Context, request *filer_pb.StreamRenameEntr
|
||||
// abandons just this rename. Bound the silence between events the way
|
||||
// the mutation mux does: a filer that accepts the stream then stalls
|
||||
// must not hold the failover walk hostage, but a long rename that
|
||||
// keeps sending events may take as long as it needs.
|
||||
// keeps sending events may take as long as it needs. Local event
|
||||
// work is not wire silence — a directory rename can spend longer
|
||||
// than the bound migrating a child's DLM lock — so the timer defers
|
||||
// while the handler runs instead of cancelling a healthy stream.
|
||||
streamCtx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
silence := time.AfterFunc(wfs.filerRPCWait(), cancel)
|
||||
var inHandler atomic.Bool
|
||||
var silence *time.Timer
|
||||
silence = time.AfterFunc(wfs.filerRPCWait(), func() {
|
||||
if inHandler.Load() {
|
||||
silence.Reset(wfs.filerRPCWait())
|
||||
return
|
||||
}
|
||||
cancel()
|
||||
})
|
||||
defer silence.Stop()
|
||||
|
||||
stream, streamErr := client.StreamRenameEntry(streamCtx, request)
|
||||
@@ -54,8 +66,10 @@ func (wfs *WFS) doRename(ctx context.Context, request *filer_pb.StreamRenameEntr
|
||||
}
|
||||
return fmt.Errorf("dir Rename %s => %s receive: %v", oldPath, newPath, recvErr)
|
||||
}
|
||||
silence.Reset(wfs.filerRPCWait())
|
||||
if err := wfs.handleRenameResponse(ctx, resp, newPath, newPathLock); err != nil {
|
||||
inHandler.Store(true)
|
||||
err := wfs.handleRenameResponse(ctx, resp, newPath, newPathLock)
|
||||
inHandler.Store(false)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -209,7 +209,16 @@ func (m *streamMutateMux) Rename(ctx context.Context, req *filer_pb.StreamRename
|
||||
if !ok {
|
||||
return fmt.Errorf("rename recv: %w: stream closed", ErrStreamTransport)
|
||||
}
|
||||
respTimer.Reset(m.wfs.filerRPCWait())
|
||||
// Applying an event can legitimately outlast the silence bound —
|
||||
// a directory rename migrates an open child's DLM lock under
|
||||
// another writer. The timer bounds wire silence only, so stop it
|
||||
// for the local work and re-arm it before the next wait.
|
||||
if !respTimer.Stop() {
|
||||
select {
|
||||
case <-respTimer.C:
|
||||
default:
|
||||
}
|
||||
}
|
||||
if r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_RenameResponse); ok {
|
||||
if r.RenameResponse != nil && r.RenameResponse.EventNotification != nil {
|
||||
if err := onEvent(r.RenameResponse); err != nil {
|
||||
@@ -226,6 +235,7 @@ func (m *streamMutateMux) Rename(ctx context.Context, req *filer_pb.StreamRename
|
||||
}
|
||||
return nil
|
||||
}
|
||||
respTimer.Reset(m.wfs.filerRPCWait())
|
||||
case <-respTimer.C:
|
||||
m.teardownStream(gen)
|
||||
return fmt.Errorf("rename recv: %w: timed out", ErrStreamTransport)
|
||||
|
||||
Reference in New Issue
Block a user