🌐 US-Proxy
class="logged-out env-production page-responsive" style="word-wrap: break-word;" >
Skip to content

Commit b558031

Browse files
gaikwadabhishekalex-aizman
authored andcommitted
core/xactions: Stop/Abort terminal semantics (fix)
* for xactions that _idle_ - DemandBase.Stop is now idempotent, protected by its own atomic * common Base.SetStopping is now boolean - returns its CAS result * get-batch (x-moss): - use the latter to eliminate racing vs its own concurrent Abort - call DemandBase.Stop prior to waiting/quiescing Signed-off-by: Alex Aizman <alex.aizman@gmail.com>
1 parent a699cf7 commit b558031

3 files changed

Lines changed: 22 additions & 7 deletions

File tree

‎xact/base.go‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -98,8 +98,12 @@ func (xctn *Base) Bck() *meta.Bck { return &xctn.bck }
9898
// return true if 'stopping' OR 'finished'
9999
func (xctn *Base) IsDone() bool { return xctn.eutime.Load() != 0 }
100100

101-
// mark the xaction as stopping (and finishing soon); EndTime remains zero until Finish()
102-
func (xctn *Base) SetStopping() { xctn.eutime.CAS(0, cos.MSB64) } // i.e., not running vis-à-vis xreg
101+
// mark the xaction as stopping (and finishing soon);
102+
// callers that perform terminal cleanup can use the return value to decide
103+
// whether they own that cleanup path
104+
func (xctn *Base) SetStopping() bool {
105+
return xctn.eutime.CAS(0, cos.MSB64)
106+
}
103107

104108
func (xctn *Base) IsRunning() (yes bool) {
105109
yes = xctn.sutime.Load() != 0 && !xctn.IsDone() && !xctn.IsAborted()

‎xact/demand.go‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,8 @@ type (
4343
d atomic.Int64 // duration hk idle
4444
last atomic.Int64 // mono.NanoTime
4545
}
46-
pending atomic.Int64
46+
pending atomic.Int64
47+
unregged atomic.Bool // hk.Unreg() at most once per instance; NOTE: same-UUID renewals reuse HK names
4748
}
4849
)
4950

@@ -76,7 +77,9 @@ func (r *DemandBase) Init(uuid, kind string, bck *meta.Bck, idleDur time.Duratio
7677
r.InitBase(uuid, kind, bck)
7778

7879
r.idle.last.Store(mono.NanoTime())
80+
7981
hk.Reg(r.hkName+hk.NameSuffix, r.hkcb, 0 /*time.Duration*/)
82+
r.unregged.Store(false)
8083
}
8184

8285
// (e.g. usage: listed last page)
@@ -124,14 +127,19 @@ func (r *DemandBase) SubPending(n int) {
124127
}
125128

126129
func (r *DemandBase) Stop() {
130+
if !r.unregged.CAS(false, true) {
131+
return
132+
}
127133
r.SetStopping()
128134
hk.Unreg(r.hkName + hk.NameSuffix)
129135
if r.ticks != nil {
130136
r.ticks.Close()
131-
r.ticks = nil
132137
}
133138
}
134139

140+
// Only abort the Base part, and note:
141+
// for demand xactions, Stop() is a cleanup-sequence sensitive barrier: it unregisters
142+
// the idle HK callback and can allow same-UUID renewal to register the same HK name.
135143
func (r *DemandBase) Abort(err error) (ok bool) {
136144
if err == nil && !r.IsIdle() {
137145
err = cmn.NewErrAborted(r.Name(), "aborting non-idle", nil)

‎xact/xs/moss.go‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -378,6 +378,9 @@ func (r *XactMoss) Abort(err error) bool {
378378

379379
hk.UnregIf(r.hkName(), func(int64) time.Duration { return 0 })
380380

381+
// unreg idle callback prior to quiesce
382+
r.DemandBase.Stop()
383+
381384
nlog.Infoln(r.Name(), "aborting:", err)
382385

383386
r.activeWG.Wait()
@@ -389,8 +392,6 @@ func (r *XactMoss) Abort(err error) bool {
389392

390393
r.abortedWIDs.Clear()
391394

392-
r.DemandBase.Stop()
393-
394395
r.bewarmStop()
395396
return true
396397
}
@@ -426,7 +427,9 @@ func (r *XactMoss) fini(int64) (d time.Duration) {
426427
case r.Pending() > 0:
427428
return mossIdleTime
428429
default:
429-
r.SetStopping()
430+
if !r.SetStopping() {
431+
return hk.UnregInterval
432+
}
430433

431434
hk.UnregIf(r.hkName(), func(int64) time.Duration { return 0 })
432435
// stop receiving

0 commit comments

Comments
 (0)