mirror of
https://github.com/esengine/DeepSeek-Reasonix.git
synced 2026-11-07 02:36:01 -10:00
fix(lease): separate run and yielded-claim lifetimes
Problem: Aborting provisional admission can expire an unrelated live user wait or retake paths deliberately yielded to a user. A pending wait is also lost when the last writer or an earlier restoration finishes afterward. Root cause: Physical release reset run state, and one retained extent represented both active and parked ownership. Restoration dropped its temporary protection before reholding the original caller. Expired queued grants and retention callbacks could contribute resources to the next run. Fix: Separate active and parked retention, advance generations at the final participating run end, and reconcile outstanding yields whenever writing holds drain. Keep restoration protection through rehold, fence late admission and retention, and clean expired grants without dropping current writers. Actual in-flight writes stay protected after their run ends. Verification: The submitted owner compiles and fails all 14 selected test/subtest failure events across the new public-API schedules. Corrected full workspacelease and agent race pass, as do actual boot race, all 238 packages in disjoint required/remaining partitions, required builtin/boot, vet, lint and check. Local evidence is macOS only; hosted checks and review apply to the new head. Refs #14025 Refs #14133
This commit is contained in:
1 parent
affb71f9dd
commit
bc330925c0
6 files changed
+377
-45
No files matched your search
@@ -26,6 +26,10 @@ cannot acquire B even though this call never entered a hook or Execute.
|
||||
- The existing workspacelease Owner keeps a retained extent and a set of
|
||||
identified provisional calls. An unset extent differs from the existing
|
||||
empty-path representation of the whole workspace.
|
||||
- Active retention and retention parked for a user are separate. Aborting a
|
||||
call cannot retake parked files or expire an unrelated same-run user wait.
|
||||
- Run generations end when the last participating run ends, independently of
|
||||
physical claim release. A late expired grant cleans its provisional contribution.
|
||||
- PreparePaths admits the entire normalized set through the existing registrar.
|
||||
Retain runs before possibly writing PreToolUse hooks and concrete execution;
|
||||
success, failure and partial execution then preserve legacy run retention.
|
||||
@@ -40,6 +44,10 @@ cannot acquire B even though this call never entered a hook or Execute.
|
||||
- A completed callback cannot affect a new run. Suspended calls keep their own
|
||||
identities; resume restores the current union without promoting provisional
|
||||
paths to retained paths or resurrecting an already aborted call.
|
||||
- The restoration's temporary hold remains live until the original caller is
|
||||
reheld, preventing a release gap during that transfer.
|
||||
- A pending user wait is reconciled when a writer or restoration finishes.
|
||||
With no writing call left, retained paths are parked instead of kept active.
|
||||
- An expired restoration returns ErrHoldEnded. Repeated restoration/cleanup returns the
|
||||
same outcome, including failures, instead of falsely reporting success.
|
||||
- Registrar waits close on acquisition or refusal. Completed cleanup clears
|
||||
|
||||
@@ -286,6 +286,13 @@ func (o *Owner) EndRun() {
|
||||
o.mu.Lock()
|
||||
if o.activeRuns > 0 {
|
||||
o.activeRuns--
|
||||
if o.activeRuns == 0 {
|
||||
o.holds.generation++
|
||||
o.holds.retained = leaseExtent{}
|
||||
o.holds.yielders = 0
|
||||
o.holds.parked, o.holds.parkedSet = nil, false
|
||||
o.holds.parkedRetained = leaseExtent{}
|
||||
}
|
||||
}
|
||||
release := o.releaseIfIdleLocked()
|
||||
if o.activeRuns == 0 && o.holds.inFlight == 0 && !o.acquired && !o.acquiring {
|
||||
@@ -326,7 +333,7 @@ func (o *Owner) acquire(ctx context.Context, paths []string, h *callHold) error
|
||||
for {
|
||||
o.mu.Lock()
|
||||
o.askedLocked(time.Now())
|
||||
if h != nil && h.done {
|
||||
if h != nil && (h.done || h.generation != o.holds.generation) {
|
||||
o.mu.Unlock()
|
||||
return ErrHoldEnded
|
||||
}
|
||||
@@ -359,6 +366,7 @@ func (o *Owner) acquire(ctx context.Context, paths []string, h *callHold) error
|
||||
o.mu.Lock()
|
||||
o.acquiring = false
|
||||
o.waiting = false
|
||||
var stale *callHold
|
||||
if err == nil {
|
||||
o.acquired = true
|
||||
o.scope.paths = claim
|
||||
@@ -366,8 +374,10 @@ func (o *Owner) acquire(ctx context.Context, paths []string, h *callHold) error
|
||||
o.acquiredAt = time.Now()
|
||||
o.releaseSystem = release
|
||||
}
|
||||
if h != nil && h.done {
|
||||
if h != nil && (h.done || h.generation != o.holds.generation) {
|
||||
err = ErrHoldEnded
|
||||
stale = &callHold{owner: o, generation: o.holds.generation}
|
||||
o.registerCallLocked(stale, requested)
|
||||
} else {
|
||||
o.registerCallLocked(h, requested)
|
||||
}
|
||||
@@ -378,15 +388,24 @@ func (o *Owner) acquire(ctx context.Context, paths []string, h *callHold) error
|
||||
if releaseIfIdle != nil {
|
||||
releaseIfIdle()
|
||||
}
|
||||
if stale != nil {
|
||||
err = errors.Join(err, (&PreparedCall{hold: stale}).Close(context.WithoutCancel(ctx)))
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
func (o *Owner) releaseIfIdleLocked() func() {
|
||||
if !o.acquired || o.acquiring || o.activeRuns != 0 || o.holds.inFlight != 0 {
|
||||
if !o.acquired || o.acquiring || o.holds.inFlight != 0 {
|
||||
return nil
|
||||
}
|
||||
return o.releaseLocked(true)
|
||||
if o.activeRuns == 0 {
|
||||
return o.releaseLocked(true)
|
||||
}
|
||||
if o.holds.yielders > 0 {
|
||||
return o.parkLocked()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// releaseLocked drops the claim. A yield (final false) is a pause inside one
|
||||
@@ -400,7 +419,9 @@ func (o *Owner) releaseLocked(final bool) func() {
|
||||
o.stats.Held += time.Since(o.acquiredAt)
|
||||
return release
|
||||
}
|
||||
o.resetCallsLocked()
|
||||
if o.activeRuns == 0 {
|
||||
o.resetCallsLocked()
|
||||
}
|
||||
closed, report := o.closeStatsLocked(time.Now())
|
||||
notice := o.onRelease
|
||||
return func() {
|
||||
|
||||
@@ -56,13 +56,19 @@ func (o *Owner) PreparePaths(ctx context.Context, paths []string) (context.Conte
|
||||
if o == nil {
|
||||
return ctx, nil, nil
|
||||
}
|
||||
h := &callHold{owner: o}
|
||||
h := o.newCall(false)
|
||||
if err := o.acquire(ctx, paths, h); err != nil {
|
||||
return ctx, nil, err
|
||||
}
|
||||
return context.WithValue(ctx, callKey{}, h), &PreparedCall{hold: h}, nil
|
||||
}
|
||||
|
||||
func (o *Owner) newCall(retained bool) *callHold {
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
return &callHold{owner: o, retained: retained, generation: o.holds.generation}
|
||||
}
|
||||
|
||||
func (o *Owner) registerCallLocked(h *callHold, paths []string) {
|
||||
if h != nil && h.done {
|
||||
return
|
||||
@@ -94,16 +100,35 @@ func (p *PreparedCall) Retain() {
|
||||
o := h.owner
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if _, present := o.holds.calls[h]; present && !h.done && !h.retained {
|
||||
if _, present := o.holds.calls[h]; present && !h.done && !h.retained && h.generation == o.holds.generation {
|
||||
h.retained = true
|
||||
o.holds.retained.add(h.paths)
|
||||
if h.live {
|
||||
o.holds.retained.add(h.paths)
|
||||
} else {
|
||||
o.holds.parkedRetained.add(h.paths)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (o *Owner) callExtentLocked() leaseExtent {
|
||||
extent := o.holds.retained
|
||||
for h := range o.holds.calls {
|
||||
extent.add(h.paths)
|
||||
if h.live {
|
||||
extent.add(h.paths)
|
||||
}
|
||||
}
|
||||
return extent
|
||||
}
|
||||
|
||||
func (o *Owner) restoreExtentLocked() leaseExtent {
|
||||
extent := o.holds.retained
|
||||
if o.holds.parkedRetained.set {
|
||||
extent.add(o.holds.parkedRetained.paths)
|
||||
}
|
||||
for h := range o.holds.calls {
|
||||
if !h.done && h.generation == o.holds.generation {
|
||||
extent.add(h.paths)
|
||||
}
|
||||
}
|
||||
return extent
|
||||
}
|
||||
@@ -114,9 +139,9 @@ func (o *Owner) resetCallsLocked() {
|
||||
h.done = true
|
||||
}
|
||||
clear(o.holds.calls)
|
||||
o.holds.generation++
|
||||
o.holds.yielders = 0
|
||||
o.holds.parked, o.holds.parkedSet = nil, false
|
||||
o.holds.parkedRetained = leaseExtent{}
|
||||
}
|
||||
|
||||
// Close removes an unretained extent through the existing claim registrar.
|
||||
@@ -166,13 +191,17 @@ func (p *PreparedCall) close(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
if !extent.set || o.activeRuns == 0 && o.holds.inFlight == 0 {
|
||||
release := o.releaseLocked(true)
|
||||
release := o.releaseLocked(o.activeRuns == 0 || !o.holds.parkedSet)
|
||||
o.mu.Unlock()
|
||||
release()
|
||||
return nil
|
||||
}
|
||||
if h.retained || coversPaths(extent.paths, o.scope.paths) {
|
||||
if coversPaths(extent.paths, o.scope.paths) {
|
||||
release := o.releaseIfIdleLocked()
|
||||
o.mu.Unlock()
|
||||
if release != nil {
|
||||
release()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
o.acquiring, o.acquireDone = true, make(chan struct{})
|
||||
|
||||
@@ -572,3 +572,254 @@ func TestPreparedRegistrarWaitClosesForCleanupAndRefusedUpgrade(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreparedAbortPreservesAnUnrelatedLiveUserWait(t *testing.T) {
|
||||
a, b, _ := yieldPair(t)
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
_, p := prepareCall(t, a, []string{"pending"})
|
||||
resume := a.Yield(t.Context())
|
||||
if err := p.Close(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
requireClaim(t, b, []string{"pending"}, false)
|
||||
if _, err := resume(t.Context()); err != nil {
|
||||
t.Fatalf("aborting another call expired a live user wait: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreparedAbortDoesNotRetakePathsParkedForAUser(t *testing.T) {
|
||||
for _, whole := range []bool{false, true} {
|
||||
t.Run(map[bool]string{false: "named", true: "whole"}[whole], func(t *testing.T) {
|
||||
a, b, _ := yieldPair(t)
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
if err := a.AcquirePaths(t.Context(), []string{"parked"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resume := a.Yield(t.Context())
|
||||
if !whole {
|
||||
requireClaim(t, b, []string{"parked"}, false)
|
||||
}
|
||||
paths := []string{"pending"}
|
||||
if whole {
|
||||
paths = nil
|
||||
}
|
||||
_, p := prepareCall(t, a, paths)
|
||||
if err := p.Close(t.Context()); err != nil {
|
||||
t.Fatalf("cleanup tried to retake a parked write: %v", err)
|
||||
}
|
||||
if whole {
|
||||
requireClaim(t, b, []string{"parked"}, false)
|
||||
}
|
||||
requireClaim(t, b, []string{"pending"}, false)
|
||||
b.EndRun()
|
||||
if _, err := resume(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.BeginRun()
|
||||
requireClaim(t, b, []string{"parked"}, true)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreparedQueuedInAnEndedRunCannotPinTheNextRun(t *testing.T) {
|
||||
withWaitGrace(t, time.Nanosecond)
|
||||
root, locks := t.TempDir(), t.TempDir()
|
||||
waiting := make(chan struct{})
|
||||
var once sync.Once
|
||||
a, err := New(root, locks, func(w Wait) {
|
||||
if w.Outcome == WaitBegan {
|
||||
once.Do(func() { close(waiting) })
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, err := New(root, locks, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
defer a.EndRun()
|
||||
defer b.EndRun()
|
||||
if err := b.AcquirePaths(t.Context(), []string{"held"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, p, err := a.PreparePaths(ctx, []string{"held"})
|
||||
if p != nil {
|
||||
_ = p.Close(context.Background())
|
||||
}
|
||||
done <- err
|
||||
}()
|
||||
select {
|
||||
case <-waiting:
|
||||
case <-ctx.Done():
|
||||
t.Fatal("preparation never queued")
|
||||
}
|
||||
a.EndRun()
|
||||
a.BeginRun()
|
||||
b.EndRun()
|
||||
if err := <-done; !errors.Is(err, ErrHoldEnded) {
|
||||
t.Fatalf("old preparation entered a new run: %v", err)
|
||||
}
|
||||
b.BeginRun()
|
||||
requireClaim(t, b, []string{"held"}, false)
|
||||
}
|
||||
|
||||
func TestPreparedAbortPreservesAParkedWholeExtent(t *testing.T) {
|
||||
a, b, _ := yieldPair(t)
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
if err := a.AcquireWrite(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resume := a.Yield(t.Context())
|
||||
_, p := prepareCall(t, a, []string{"pending"})
|
||||
if err := p.Close(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
requireClaim(t, b, []string{"outside_pending"}, false)
|
||||
requireClaim(t, b, []string{"pending"}, false)
|
||||
b.EndRun()
|
||||
if _, err := resume(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.BeginRun()
|
||||
requireClaim(t, b, []string{"outside_pending"}, true)
|
||||
}
|
||||
|
||||
func TestPreparedStandaloneCallCanYieldWithoutARun(t *testing.T) {
|
||||
a, b, _ := yieldPair(t)
|
||||
b.BeginRun()
|
||||
ctx, p := prepareCall(t, a, []string{"pending"})
|
||||
resume := a.Yield(ctx)
|
||||
requireClaim(t, b, []string{"pending"}, false)
|
||||
b.EndRun()
|
||||
if _, err := resume(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := p.Close(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.BeginRun()
|
||||
requireClaim(t, b, []string{"pending"}, false)
|
||||
}
|
||||
|
||||
func TestPendingUserWaitParksRetentionWhenTheLastWriterEnds(t *testing.T) {
|
||||
for _, legacy := range []bool{false, true} {
|
||||
t.Run(map[bool]string{false: "prepared", true: "legacy"}[legacy], func(t *testing.T) {
|
||||
a, b, _ := yieldPair(t)
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
var end func()
|
||||
if legacy {
|
||||
_, closeHold, err := a.HoldPaths(t.Context(), []string{"written"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
end = closeHold
|
||||
} else {
|
||||
_, p := prepareCall(t, a, []string{"written"})
|
||||
p.Retain()
|
||||
end = func() {
|
||||
if err := p.Close(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
resume := a.Yield(t.Context())
|
||||
end()
|
||||
requireClaim(t, b, []string{"written"}, false)
|
||||
b.EndRun()
|
||||
if _, err := resume(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.BeginRun()
|
||||
requireClaim(t, b, []string{"written"}, true)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUserWaitArrivingDuringRestorationStaysYielded(t *testing.T) {
|
||||
withWaitGrace(t, time.Nanosecond)
|
||||
root, locks := t.TempDir(), t.TempDir()
|
||||
waiting := make(chan struct{})
|
||||
var once sync.Once
|
||||
a, err := New(root, locks, func(w Wait) {
|
||||
if w.Outcome == WaitBegan {
|
||||
once.Do(func() { close(waiting) })
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, err := New(root, locks, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
defer a.EndRun()
|
||||
defer b.EndRun()
|
||||
if err := a.AcquirePaths(t.Context(), []string{"written"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
first := a.Yield(t.Context())
|
||||
if err := b.AcquirePaths(t.Context(), []string{"written"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { _, err := first(ctx); done <- err }()
|
||||
select {
|
||||
case <-waiting:
|
||||
case <-ctx.Done():
|
||||
t.Fatal("restoration never queued")
|
||||
}
|
||||
second := a.Yield(t.Context())
|
||||
b.EndRun()
|
||||
if err := <-done; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.BeginRun()
|
||||
requireClaim(t, b, []string{"written"}, false)
|
||||
b.EndRun()
|
||||
if _, err := second(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEndedRunRetentionCannotPinAPathInTheNextRun(t *testing.T) {
|
||||
for _, retainedBeforeEnd := range []bool{false, true} {
|
||||
t.Run(map[bool]string{false: "late_retain", true: "retained_before_end"}[retainedBeforeEnd], func(t *testing.T) {
|
||||
a, b, _ := yieldPair(t)
|
||||
a.BeginRun()
|
||||
b.BeginRun()
|
||||
_, old := prepareCall(t, a, []string{"old"})
|
||||
if retainedBeforeEnd {
|
||||
old.Retain()
|
||||
}
|
||||
a.EndRun()
|
||||
a.BeginRun()
|
||||
_, current := prepareCall(t, a, []string{"current"})
|
||||
if !retainedBeforeEnd {
|
||||
old.Retain()
|
||||
}
|
||||
if err := old.Close(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
requireClaim(t, b, []string{"old"}, false)
|
||||
requireClaim(t, b, []string{"current"}, true)
|
||||
if err := current.Close(t.Context()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -10,25 +10,27 @@ import (
|
||||
// holdState is what the claim's lifetime depends on besides runs: calls that
|
||||
// are executing under it, and sessions waiting on a person who took it off.
|
||||
type holdState struct {
|
||||
generation uint64
|
||||
retained leaseExtent
|
||||
calls map[*callHold]struct{}
|
||||
inFlight int
|
||||
yielders int
|
||||
parked []string
|
||||
parkedSet bool
|
||||
// parkedMark is the claim record when the parked extent was given up.
|
||||
generation uint64
|
||||
retained leaseExtent
|
||||
parkedRetained leaseExtent
|
||||
calls map[*callHold]struct{}
|
||||
inFlight int
|
||||
yielders int
|
||||
parked []string
|
||||
parkedSet bool
|
||||
// parkedMark is the claim record at the first outstanding yield request.
|
||||
parkedMark claimMark
|
||||
}
|
||||
|
||||
// callHold is one writing call's share of inFlight. It stays reachable from the
|
||||
// call's context so a question that call asks can suspend it.
|
||||
type callHold struct {
|
||||
owner *Owner
|
||||
paths []string
|
||||
live bool
|
||||
done bool
|
||||
retained bool
|
||||
owner *Owner
|
||||
paths []string
|
||||
live bool
|
||||
done bool
|
||||
retained bool
|
||||
generation uint64
|
||||
}
|
||||
|
||||
type callKey struct{}
|
||||
@@ -40,7 +42,7 @@ func (o *Owner) HoldPaths(ctx context.Context, paths []string) (context.Context,
|
||||
if o == nil {
|
||||
return ctx, func() {}, nil
|
||||
}
|
||||
h := &callHold{owner: o, retained: true}
|
||||
h := o.newCall(true)
|
||||
err := o.acquire(ctx, paths, h)
|
||||
if err != nil {
|
||||
return ctx, func() {}, err
|
||||
@@ -85,7 +87,14 @@ func (o *Owner) Yield(ctx context.Context) Resume {
|
||||
}
|
||||
mark := o.claimMark()
|
||||
o.mu.Lock()
|
||||
if o.activeRuns == 0 && !o.acquired && !o.acquiring && o.holds.inFlight == 0 {
|
||||
o.mu.Unlock()
|
||||
return noResume
|
||||
}
|
||||
generation := o.holds.generation
|
||||
if o.holds.yielders == 0 {
|
||||
o.holds.parkedMark = mark
|
||||
}
|
||||
o.holds.yielders++
|
||||
var suspended *callHold
|
||||
if h, _ := ctx.Value(callKey{}).(*callHold); h != nil && h.owner == o && h.live {
|
||||
@@ -95,14 +104,7 @@ func (o *Owner) Yield(ctx context.Context) Resume {
|
||||
}
|
||||
var release func()
|
||||
if o.acquired && !o.acquiring && o.holds.inFlight == 0 {
|
||||
held := slices.Clone(o.scope.paths)
|
||||
if o.holds.parkedSet {
|
||||
held = mergePaths(o.holds.parked, held)
|
||||
} else {
|
||||
o.holds.parkedMark = mark
|
||||
}
|
||||
o.holds.parked, o.holds.parkedSet = held, true
|
||||
release = o.releaseLocked(false)
|
||||
release = o.parkLocked()
|
||||
}
|
||||
o.mu.Unlock()
|
||||
if release != nil {
|
||||
@@ -117,9 +119,22 @@ func (o *Owner) Yield(ctx context.Context) Resume {
|
||||
}
|
||||
}
|
||||
|
||||
func (o *Owner) parkLocked() func() {
|
||||
held := slices.Clone(o.scope.paths)
|
||||
if o.holds.parkedSet {
|
||||
held = mergePaths(o.holds.parked, held)
|
||||
}
|
||||
o.holds.parked, o.holds.parkedSet = held, true
|
||||
if o.holds.retained.set {
|
||||
o.holds.parkedRetained.add(o.holds.retained.paths)
|
||||
o.holds.retained = leaseExtent{}
|
||||
}
|
||||
return o.releaseLocked(false)
|
||||
}
|
||||
|
||||
func (o *Owner) resume(ctx context.Context, suspended *callHold, generation uint64) (Contention, error) {
|
||||
o.mu.Lock()
|
||||
if generation != o.holds.generation {
|
||||
if generation != o.holds.generation || o.holds.yielders == 0 {
|
||||
o.mu.Unlock()
|
||||
return Contention{}, ErrHoldEnded
|
||||
}
|
||||
@@ -131,27 +146,35 @@ func (o *Owner) resume(ctx context.Context, suspended *callHold, generation uint
|
||||
}
|
||||
o.mu.Unlock()
|
||||
var contention Contention
|
||||
var restored *PreparedCall
|
||||
if last {
|
||||
o.mu.Lock()
|
||||
extent := o.callExtentLocked()
|
||||
extent := o.restoreExtentLocked()
|
||||
o.mu.Unlock()
|
||||
if extent.set {
|
||||
restore := &callHold{owner: o}
|
||||
restore := &callHold{owner: o, generation: generation}
|
||||
if err := o.acquire(ctx, extent.paths, restore); err != nil {
|
||||
return Contention{}, err
|
||||
}
|
||||
if err := (&PreparedCall{hold: restore}).Close(ctx); err != nil {
|
||||
return Contention{}, err
|
||||
restored = &PreparedCall{hold: restore}
|
||||
o.mu.Lock()
|
||||
valid := generation == o.holds.generation
|
||||
if valid && o.holds.parkedRetained.set {
|
||||
o.holds.retained.add(o.holds.parkedRetained.paths)
|
||||
o.holds.parkedRetained = leaseExtent{}
|
||||
}
|
||||
o.mu.Unlock()
|
||||
if !valid {
|
||||
return Contention{}, errors.Join(ErrHoldEnded, restored.Close(ctx))
|
||||
}
|
||||
}
|
||||
contention = o.contendersSince(mark, parked)
|
||||
}
|
||||
var err error
|
||||
if suspended != nil {
|
||||
if err := o.rehold(ctx, suspended); err != nil {
|
||||
return contention, err
|
||||
}
|
||||
err = o.rehold(ctx, suspended)
|
||||
}
|
||||
return contention, nil
|
||||
return contention, errors.Join(err, restored.Close(ctx))
|
||||
}
|
||||
|
||||
// rehold puts a suspended call back under the claim it gave up, waiting for it
|
||||
|
||||
@@ -1,2 +1,2 @@
|
||||
- Fixed: a write refused before hooks or execution releases its newly requested paths without dropping earlier writes or concurrent calls. Refs #14025, #14133.
|
||||
- 修复:hooks 和执行前被拒绝的写入会释放本次新增路径,同时保留此前写入和并行调用的有效占用。关联 #14025、#14133。
|
||||
- Fixed: refused writes release provisional paths while preserving earlier writes, concurrent calls and live user waits; cleanup does not retake paths yielded to a user. Refs #14025, #14133.
|
||||
- 修复:被拒绝的写入会释放临时路径,保留此前写入、并行调用和有效的用户等待;清理不会重新占住为等待用户而让出的路径。关联 #14025、#14133。
|
||||
Reference in new issue
Block a user