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

Commit 524a219

Browse files
Nahemah1022Nahemah1022
committed
etl: respect init_timeout in 2PC transaction GC for ETL initialization
* previously, the txn GC reclaimed any begin-phase transaction after a flat `2 * max_host_busy` timeout, ignoring the ETL's `init_timeout` * ETL init's begin phase creates the K8s pod and blocks on readiness; a slow package install may exceed timeout and get GC-aborted mid-init * this patch adds `commitTimeout()` to the txn interface: default stays `2 * max_host_busy`, `txnETLInit` returns `init_timeout` plus slack Signed-off-by: Tony Chen <a122774007@gmail.com>
1 parent 15760de commit 524a219

3 files changed

Lines changed: 182 additions & 8 deletions

File tree

‎ais/tgtobj_internal_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,8 @@ func TestMain(m *testing.M) {
8787
t.owner.bmd.putPersist(bmd, nil)
8888
fs.CreateBucket(bck.Bucket(), false /*nilbmd*/)
8989

90+
t.owner.smap.put(newSmap()) // Note: required by ais/txn_internal_test.go
91+
9092
m.Run()
9193
}
9294

‎ais/txn.go‎

Lines changed: 28 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -26,17 +26,19 @@ import (
2626

2727
// GC
2828
const (
29-
gcTxnsNumKeep = 16
30-
gcTxnsTimeotMult = 10
29+
gcTxnsNumKeep = 16 // in-flight txns above 4x this count tighten the GC sweep cadence
30+
gcTxnsTimeoutMult = 10 // commit -> done deadline = N * max_host_busy
3131

32-
TxnTimeoutMult = 2
32+
txnTimeoutMult = 2 // begin -> start-commit deadline (and warn/slack) = N * max_host_busy
3333
)
3434

3535
type (
3636
txn interface {
3737
// accessors
3838
uuid() string
3939
started(phase string, tm ...time.Time) time.Time
40+
// max duration of the begin -> start-commit (2PC) window before the GC reclaims the txn
41+
commitTimeout(config *cmn.Config) time.Duration
4042
isDone() (done bool, err error)
4143
set(nlps []core.NLP)
4244
// triggers
@@ -157,7 +159,9 @@ var (
157159
_ txn = (*txnTCB)(nil)
158160
_ txn = (*txnTCObjs)(nil)
159161
_ txn = (*txnECEncode)(nil)
162+
_ txn = (*txnArchMultiObj)(nil)
160163
_ txn = (*txnPromote)(nil)
164+
_ txn = (*txnCreateNBI)(nil)
161165
_ txn = (*txnETLInit)(nil)
162166
)
163167

@@ -397,15 +401,15 @@ func checkTimeout(txn txn, now time.Time, config *cmn.Config) (err, warn error)
397401
elapsed := now.Sub(txn.started(apc.Begin2PC))
398402
if commitTimestamp := txn.started(apc.Commit2PC); !commitTimestamp.IsZero() {
399403
elapsed = now.Sub(commitTimestamp)
400-
if elapsed > gcTxnsTimeotMult*config.Timeout.MaxHostBusy.D() {
404+
if elapsed > gcTxnsTimeoutMult*config.Timeout.MaxHostBusy.D() {
401405
err = fmt.Errorf("gc %s: [commit - done] timeout", txn)
402-
} else if elapsed >= TxnTimeoutMult*config.Timeout.MaxHostBusy.D() {
406+
} else if elapsed >= txnTimeoutMult*config.Timeout.MaxHostBusy.D() {
403407
err = fmt.Errorf("gc %s: commit is taking too long", txn)
404408
}
405409
} else {
406-
if elapsed > TxnTimeoutMult*config.Timeout.MaxHostBusy.D() {
410+
if elapsed > txn.commitTimeout(config) {
407411
err = fmt.Errorf("gc %s: [begin - start-commit] timeout", txn)
408-
} else if elapsed >= TxnTimeoutMult*cmn.Rom.MaxKeepalive() {
412+
} else if elapsed >= txnTimeoutMult*cmn.Rom.MaxKeepalive() {
409413
warn = fmt.Errorf("gc %s: commit message is taking too long", txn)
410414
}
411415
}
@@ -433,7 +437,12 @@ func (txn *txnBase) started(phase string, tm ...time.Time) (ts time.Time) {
433437
default:
434438
debug.Assert(false)
435439
}
436-
return
440+
return ts
441+
}
442+
443+
// default: the begin -> start-commit window is bounded by max_host_busy
444+
func (*txnBase) commitTimeout(config *cmn.Config) time.Duration {
445+
return txnTimeoutMult * config.Timeout.MaxHostBusy.D()
437446
}
438447

439448
func (txn *txnBase) isDone() (done bool, err error) {
@@ -684,6 +693,17 @@ func (txn *txnETLInit) abort(err error) {
684693
etl.StopByXid(txn.uuid(), err) // only stop the ETL created from this transaction
685694
}
686695

696+
// ETL init's begin phase creates the K8s pod and blocks on its readiness, which is
697+
// bounded by the user-specified init_timeout (not max_host_busy). Allow init_timeout
698+
// plus the usual slack so the GC won't reclaim an init still making progress.
699+
func (txn *txnETLInit) commitTimeout(config *cmn.Config) time.Duration {
700+
initTimeout, _ := txn.msg.Timeouts()
701+
if initTimeout.D() <= 0 {
702+
return txn.txnBase.commitTimeout(config)
703+
}
704+
return initTimeout.D() + txnTimeoutMult*config.Timeout.MaxHostBusy.D()
705+
}
706+
687707
func (txn *txnETLInit) String() string { return txn.msg.String() }
688708

689709
// no-op: no bucket/resources needs to be locked for initializing ETL

‎ais/txn_internal_test.go‎

Lines changed: 152 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,152 @@
1+
// Package ais provides AIStore's proxy and target nodes.
2+
/*
3+
* Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
4+
*/
5+
package ais
6+
7+
import (
8+
"strings"
9+
"testing"
10+
"time"
11+
12+
"github.com/NVIDIA/aistore/api/apc"
13+
"github.com/NVIDIA/aistore/cmn"
14+
"github.com/NVIDIA/aistore/cmn/cos"
15+
"github.com/NVIDIA/aistore/core/meta"
16+
"github.com/NVIDIA/aistore/ext/etl"
17+
)
18+
19+
const testMaxHostBusy = 20 * time.Second
20+
21+
func testTxnConfig() *cmn.Config {
22+
config := &cmn.Config{}
23+
config.Timeout.MaxHostBusy = cos.Duration(testMaxHostBusy)
24+
return config
25+
}
26+
27+
// returns the txn interface (not the concrete type) so callers exercise it exactly
28+
// as production does - notably started(), which is interface-dispatched everywhere.
29+
func newETLInitTxn(initTimeout time.Duration) txn {
30+
msg := &etl.InitSpecMsg{InitMsgBase: etl.InitMsgBase{InitTimeout: cos.Duration(initTimeout)}}
31+
c := &txnSrv{t: t, uuid: "test-etl-init", msg: &actMsgExt{ActMsg: apc.ActMsg{Action: apc.ActETLInline}}}
32+
return newTxnETLInit(c, msg)
33+
}
34+
35+
// a representative non-ETL txn: inherits txnBase.commitTimeout unchanged.
36+
func newCreateBucketTxn() txn {
37+
bck := meta.NewBck(testBucket, apc.AIS, cmn.NsGlobal)
38+
c := &txnSrv{t: t, bck: bck, uuid: "test-create-bck", msg: &actMsgExt{ActMsg: apc.ActMsg{Action: apc.ActCreateBck}}}
39+
return newTxnCreateBucket(c)
40+
}
41+
42+
// commitTimeout: ETL respects init_timeout (+ slack); everything else is generic.
43+
func TestCommitTimeout(t *testing.T) {
44+
config := testTxnConfig()
45+
slack := txnTimeoutMult * testMaxHostBusy // begin->commit slack added on top of init_timeout
46+
generic := txnTimeoutMult * testMaxHostBusy
47+
48+
tests := []struct {
49+
name string
50+
txn txn
51+
expect time.Duration
52+
}{
53+
{"non-etl-create-bucket", newCreateBucketTxn(), generic},
54+
{"etl-5m", newETLInitTxn(5 * time.Minute), 5*time.Minute + slack},
55+
{"etl-default-45s", newETLInitTxn(etl.DefaultInitTimeout), etl.DefaultInitTimeout + slack},
56+
{"etl-zero-falls-back", newETLInitTxn(0), generic}, // guard: never below generic
57+
}
58+
for _, tc := range tests {
59+
t.Run(tc.name, func(t *testing.T) {
60+
if got := tc.txn.commitTimeout(config); got != tc.expect {
61+
t.Fatalf("commitTimeout = %v, want %v", got, tc.expect)
62+
}
63+
})
64+
}
65+
}
66+
67+
// checkTimeout (begin phase): the actual GC decision. This is the regression guard:
68+
// pre-fix, the deadline was a flat 2*max_host_busy (40s) for ALL txns, so a slow ETL
69+
// init was reaped mid-bootstrap. Post-fix, ETL init survives until init_timeout+slack.
70+
func TestCheckTimeoutBeginPhaseETL(t *testing.T) {
71+
config := testTxnConfig()
72+
begin := time.Now()
73+
slack := txnTimeoutMult * testMaxHostBusy
74+
75+
tests := []struct {
76+
name string
77+
initTimeout time.Duration
78+
elapsed time.Duration
79+
wantReaped bool
80+
}{
81+
// 5m init_timeout, still booting at 2m. Old code reaped at 40s.
82+
{"etl-5m-at-2m-survives", 5 * time.Minute, 2 * time.Minute, false},
83+
// boundary: just inside vs just past init_timeout+slack.
84+
{"etl-5m-just-inside", 5 * time.Minute, 5*time.Minute + slack - time.Second, false},
85+
{"etl-5m-just-past", 5 * time.Minute, 5*time.Minute + slack + time.Second, true},
86+
// even a default (45s) ETL init must outlive the old 40s flat deadline.
87+
{"etl-default-at-41s-survives", etl.DefaultInitTimeout, 41 * time.Second, false},
88+
{"etl-default-past-deadline", etl.DefaultInitTimeout, etl.DefaultInitTimeout + slack + time.Second, true},
89+
}
90+
for _, tc := range tests {
91+
t.Run(tc.name, func(t *testing.T) {
92+
txn := newETLInitTxn(tc.initTimeout)
93+
txn.started(apc.Begin2PC, begin)
94+
err, _ := checkTimeout(txn, begin.Add(tc.elapsed), config)
95+
if tc.wantReaped {
96+
if err == nil {
97+
t.Fatalf("expected GC reap at elapsed=%v (init_timeout=%v), got nil", tc.elapsed, tc.initTimeout)
98+
}
99+
if !strings.Contains(err.Error(), "[begin - start-commit] timeout") {
100+
t.Fatalf("unexpected reap error: %v", err)
101+
}
102+
} else if err != nil {
103+
t.Fatalf("ETL init must NOT be reaped at elapsed=%v (init_timeout=%v): %v", tc.elapsed, tc.initTimeout, err)
104+
}
105+
})
106+
}
107+
}
108+
109+
// checkTimeout (begin phase): non-ETL transactions keep the generic deadline (unchanged
110+
// behavior). Uses an ETL txn with init_timeout=0 as the stand-in for the default path,
111+
// since commitTimeout then returns exactly 2*max_host_busy like every txnBase-backed type.
112+
func TestCheckTimeoutBeginPhaseGeneric(t *testing.T) {
113+
config := testTxnConfig()
114+
begin := time.Now()
115+
deadline := txnTimeoutMult * testMaxHostBusy // 40s
116+
117+
txn := newETLInitTxn(0) // init_timeout=0 -> generic commitTimeout (== txnBase default)
118+
txn.started(apc.Begin2PC, begin)
119+
120+
if err, _ := checkTimeout(txn, begin.Add(deadline-time.Second), config); err != nil {
121+
t.Fatalf("generic txn within %v deadline must survive: %v", deadline, err)
122+
}
123+
if err, _ := checkTimeout(txn, begin.Add(deadline+time.Second), config); err == nil {
124+
t.Fatalf("generic txn past %v deadline must be reaped", deadline)
125+
}
126+
}
127+
128+
// checkTimeout (commit phase): the post-commit branch is type-agnostic and must remain
129+
// governed by max_host_busy regardless of the begin-phase change above.
130+
func TestCheckTimeoutCommitPhaseUnaffected(t *testing.T) {
131+
config := testTxnConfig()
132+
begin := time.Now()
133+
134+
txn := newETLInitTxn(5 * time.Minute) // long init_timeout must NOT leak into commit phase
135+
txn.started(apc.Begin2PC, begin)
136+
txn.started(apc.Commit2PC, begin)
137+
138+
// under 2*max_host_busy after commit -> not reaped
139+
if err, _ := checkTimeout(txn, begin.Add(txnTimeoutMult*testMaxHostBusy-time.Second), config); err != nil {
140+
t.Fatalf("commit-phase txn within deadline must survive: %v", err)
141+
}
142+
// just past 2*max_host_busy -> flagged even though init_timeout=5m (proves no leak)
143+
err, _ := checkTimeout(txn, begin.Add(txnTimeoutMult*testMaxHostBusy+time.Second), config)
144+
if err == nil || !strings.Contains(err.Error(), "commit is taking too long") {
145+
t.Fatalf("commit-phase must use max_host_busy, not init_timeout; got: %v", err)
146+
}
147+
// past gcTxnsTimeoutMult*max_host_busy -> terminal commit-phase reap
148+
err, _ = checkTimeout(txn, begin.Add(gcTxnsTimeoutMult*testMaxHostBusy+time.Second), config)
149+
if err == nil || !strings.Contains(err.Error(), "[commit - done] timeout") {
150+
t.Fatalf("expected [commit - done] timeout, got: %v", err)
151+
}
152+
}

0 commit comments

Comments
 (0)