micro-optimize prefetch; simplify cold-get; add stats-updater i/f; al…

…ign fields * with refactoring Signed-off-by: Alex Aizman <[email protected]>
NVIDIA · Mar 8, 2025 · a250bfd · a250bfd
1 parent 3206a6b
commit a250bfd
Show file tree

Hide file tree

Showing 25 changed files with 155 additions and 124 deletions.
diff --git a/ais/htrun.go b/ais/htrun.go
@@ -99,6 +99,8 @@ func (h *htrun) String() string     { return h.si.String() }
 func (h *htrun) Bowner() meta.Bowner { return h.owner.bmd }
 func (h *htrun) Sowner() meta.Sowner { return h.owner.smap }
 
+func (h *htrun) StatsUpdater() cos.StatsUpdater { return h.statsT }
+
 func (h *htrun) PageMM() *memsys.MMSA { return h.gmm }
 func (h *htrun) ByteMM() *memsys.MMSA { return h.smm }
 

diff --git a/ais/target.go b/ais/target.go
@@ -322,9 +322,7 @@ func (t *target) Run() error {
 	t.htrun.init(config)
 	t.setusr1()
 
-	tstats := t.statsT.(*stats.Trunner)
-
-	core.Tinit(t, tstats, config, true /*run hk*/)
+	core.Tinit(t, config, true /*run hk*/)
 
 	fatalErr, writeErr := t.checkRestarted(config)
 	if fatalErr != nil {
@@ -355,6 +353,8 @@ func (t *target) Run() error {
 	}
 	t.owner.smap.put(smap)
 
+	tstats := t.statsT.(*stats.Trunner)
+
 	if daemon.cli.target.standby {
 		tstats.Standby(true)
 		t.regstate.disabled.Store(true)
@@ -411,8 +411,8 @@ func (t *target) Run() error {
 		go t.goresilver(marked.Interrupted)
 	}
 
-	dsort.Tinit(t.statsT, db, config)
-	dload.Init(t.statsT, db, &config.Client)
+	dsort.Tinit(db, config)
+	dload.Init(db, &config.Client)
 
 	err = t.htrun.run(config)
 

diff --git a/ais/tgtimpl.go b/ais/tgtimpl.go
@@ -142,39 +142,33 @@ func (t *target) CopyObject(lom *core.LOM, dm *bundle.DataMover, params *xs.CoiP
 	return size, err
 }
 
-// use `backend.GetObj`
-// (compare w/ `backend.GetObjReader` via ldp and blob download)
 func (t *target) GetCold(ctx context.Context, lom *core.LOM, xkind string, owt cmn.OWT) (ecode int, err error) {
-	// 1. lock
+	// lock
 	switch owt {
-	case cmn.OwtGetPrefetchLock:
-		// do nothing
-	case cmn.OwtGetTryLock, cmn.OwtGetLock:
-		if owt == cmn.OwtGetTryLock {
-			if !lom.TryLock(true) {
-				if cmn.Rom.FastV(4, cos.SmoduleAIS) {
-					nlog.Warningln(t.String(), lom.String(), owt.String(), "is busy")
-				}
-				return 0, cmn.ErrSkip // e.g. prefetch can skip it and keep on going
+	case cmn.OwtGetTryLock: // e.g., downloader
+		if !lom.TryLock(true) {
+			if cmn.Rom.FastV(4, cos.SmoduleAIS) {
+				nlog.Warningln(t.String(), lom.String(), owt.String(), "is busy")
 			}
-		} else {
-			lom.Lock(true)
+			return 0, cmn.ErrSkip // TODO: must be cmn.ErrBusy
 		}
+	case cmn.OwtGetLock: // regular usage
+		lom.Lock(true)
 	default:
-		// for cmn.OwtGet, see goi.getCold
+		// other cold-get use cases include:
+		// - cmn.OwtGet, goi.getCold
+		// - cmn.OwtGetPrefetchLock, xs/prefetch
 		debug.Assert(false, owt.String())
-		return http.StatusInternalServerError, errors.New("invalid " + owt.String())
+		return http.StatusInternalServerError, errors.New("cold-get: invalid " + owt.String())
 	}
 
-	// 2. GET remote object and store it
+	// cold GET
 	var (
 		started = mono.NanoTime()
 		backend = t.Backend(lom.Bck())
 	)
 	if ecode, err = backend.GetObj(ctx, lom, owt, nil /*origReq*/); err != nil {
-		if owt != cmn.OwtGetPrefetchLock {
-			lom.Unlock(true)
-		}
+		lom.Unlock(true)
 		if cmn.IsErrFailedTo(err) {
 			nlog.Warningln(err)
 		} else {
@@ -183,17 +177,11 @@ func (t *target) GetCold(ctx context.Context, lom *core.LOM, xkind string, owt c
 		return ecode, err
 	}
 
-	// 3. unlock
-	switch owt {
-	case cmn.OwtGetPrefetchLock:
-		// do nothing
-	case cmn.OwtGetTryLock, cmn.OwtGetLock:
-		lom.Unlock(true)
-	}
-
-	// 4. stats
+	// unlock and stats
+	lom.Unlock(true)
 	lat := mono.SinceNano(started)
 	t.coldstats(backend, lom.Bck().Cname(""), xkind, lom.Lsize(), lat)
+
 	return 0, nil
 }
 

diff --git a/ais/tgtobj.go b/ais/tgtobj.go
@@ -379,7 +379,7 @@ func (poi *putOI) fini() (ecode int, err error) {
 	case cmn.OwtGetPrefetchLock:
 		if !lom.TryLock(true) {
 			nlog.Warningln(poi.loghdr(), "is busy")
-			return 0, cmn.ErrSkip // e.g. prefetch can skip it and keep on going
+			return 0, cmn.ErrSkip // e.g. prefetch can skip it and keep on going // TODO: must be cmn.ErrBusy
 		}
 		defer lom.Unlock(true)
 	default:

diff --git a/ais/tgtobj_internal_test.go b/ais/tgtobj_internal_test.go
@@ -75,7 +75,7 @@ func TestMain(m *testing.M) {
 	t.htrun.init(config)
 
 	t.statsT = mock.NewStatsTracker()
-	core.Tinit(t, t.statsT, config, false)
+	core.Tinit(t, config, false)
 
 	bck := meta.NewBck(testBucket, apc.AIS, cmn.NsGlobal)
 	bmd := newBucketMD()

diff --git a/api/etl.go b/api/etl.go
@@ -18,12 +18,13 @@ import (
 )
 
 type ETLObjArgs struct {
-	// ETLName specifies the running ETL instance to be used in inline transform.
-	ETLName string
-
 	// TransformArgs holds the arguments to be used in ETL inline transform,
 	// which will be sent as `apc.QparamETLArgs` query parameter in the request.
+	// Optional, can be omitted (nil).
 	TransformArgs any
+
+	// ETLName specifies the running ETL instance to be used in inline transform.
+	ETLName string
 }
 
 // Initiate custom ETL workload by executing one of the documented `etl.InitMsg`

diff --git a/cmn/err.go b/cmn/err.go
@@ -447,7 +447,7 @@ func (e *ErrBusy) Error() string {
 	return fmt.Sprintf("%s %q is currently busy%s, please try again", e.whereOrType, e.what, s)
 }
 
-func isErrBusy(err error) bool {
+func IsErrBusy(err error) bool {
 	_, ok := err.(*ErrBusy)
 	return ok
 }
@@ -1286,7 +1286,7 @@ func WriteErr(w http.ResponseWriter, r *http.Request, err error, opts ...int /*[
 			status = http.StatusRequestedRangeNotSatisfiable
 		case isErrUnsupp(err), isErrNotImpl(err):
 			status = http.StatusNotImplemented
-		case isErrBusy(err):
+		case IsErrBusy(err):
 			status = http.StatusConflict
 		}
 	}

diff --git a/core/lcache.go b/core/lcache.go
@@ -307,7 +307,7 @@ func (lchk *lchk) evict(timeout time.Duration, now time.Time, pct int) {
 	var (
 		avail   = fs.GetAvail()
 		wg      = &sync.WaitGroup{}
-		evicted = g.tstats.Get(LcacheEvictedCount)
+		evicted = T.StatsUpdater().Get(LcacheEvictedCount)
 	)
 	lchk.total.Store(0)
 	for _, mi := range avail {
@@ -323,7 +323,7 @@ func (lchk *lchk) evict(timeout time.Duration, now time.Time, pct int) {
 		go evct.do()
 	}
 	wg.Wait()
-	evicted = g.tstats.Get(LcacheEvictedCount) - evicted
+	evicted = T.StatsUpdater().Get(LcacheEvictedCount) - evicted
 	nlog.Infoln("hk done:", lchk.total.Load(), evicted)
 }
 
@@ -368,7 +368,7 @@ func (evct *evct) f(hkey, value any) bool {
 	if lmd2 == md {
 		*md = lom0.md // zero out
 	}
-	g.tstats.Inc(LcacheEvictedCount)
+	T.StatsUpdater().Inc(LcacheEvictedCount)
 
 	// throttle
 	evct.evicted++
@@ -392,13 +392,13 @@ func _flushAtime(md *lmeta, atime time.Time, mdTime int64) {
 		return
 	}
 	if err = lom.flushAtime(atime); err != nil {
-		g.tstats.Inc(LcacheErrCount)
+		T.StatsUpdater().Inc(LcacheErrCount)
 		T.FSHC(err, lom.Mountpath(), lom.FQN)
 		return
 	}
 
 	// stats
-	g.tstats.Inc(LcacheFlushColdCount)
+	T.StatsUpdater().Inc(LcacheFlushColdCount)
 
 	if !md.isDirty() {
 		return
@@ -418,7 +418,7 @@ func _flushAtime(md *lmeta, atime time.Time, mdTime int64) {
 				continue
 			}
 			if err = fs.SetXattr(copyFQN, XattrLOM, buf); err != nil {
-				g.tstats.Inc(LcacheErrCount)
+				T.StatsUpdater().Inc(LcacheErrCount)
 				nlog.Errorln("set-xattr [", copyFQN, err, "]")
 				break
 			}

diff --git a/core/ldp.go b/core/ldp.go
@@ -178,7 +178,7 @@ func (lom *LOM) CheckRemoteMD(locked, sync bool, origReq *http.Request) (res CRM
 			ecode, err = 0, errDel
 		} else {
 			vlabs := map[string]string{"bucket": lom.Bck().Cname("")} // TODO -- FIXME: cannot import stats
-			g.tstats.IncWith(RemoteDeletedDelCount, vlabs)
+			T.StatsUpdater().IncWith(RemoteDeletedDelCount, vlabs)
 		}
 		debug.Assert(err != nil)
 		return CRMD{ErrCode: ecode, Err: err}

diff --git a/core/lom.go b/core/lom.go
@@ -68,7 +68,6 @@ type (
 
 type (
 	global struct {
-		tstats   cos.StatsUpdater // (stats.Trunner)
 		pmm      *memsys.MMSA
 		smm      *memsys.MMSA
 		locker   nameLocker
@@ -97,13 +96,12 @@ var (
 
 func Pinit() { bckLocker = newNameLocker() }
 
-func Tinit(t Target, tstats cos.StatsUpdater, config *cmn.Config, runHK bool) {
+func Tinit(t Target, config *cmn.Config, runHK bool) {
 	bckLocker = newNameLocker()
 	T = t
 	{
 		g.maxLmeta.Store(xattrMaxSize)
 		g.locker = newNameLocker()
-		g.tstats = tstats
 		g.pmm = t.PageMM()
 		g.smm = t.ByteMM()
 	}
@@ -516,7 +514,7 @@ func (lom *LOM) _collide(lmd *lmeta) {
 	if cmn.Rom.FastV(4, cos.SmoduleCore) || lom.digest&0xf == 5 {
 		nlog.InfoDepth(1, LcacheCollisionCount, lom.digest, "[", *lmd.uname, "]", *lom.md.uname, lom.Cname())
 	}
-	g.tstats.Inc(LcacheCollisionCount)
+	T.StatsUpdater().Inc(LcacheCollisionCount)
 }
 
 func (lom *LOM) Uncache() {

diff --git a/core/mock/target_mock.go b/core/mock/target_mock.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/NVIDIA/aistore/cmn"
+	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/core"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/fs"
@@ -31,21 +32,22 @@ var _ core.Target = (*TargetMock)(nil)
 
 func NewTarget(bo meta.Bowner) *TargetMock {
 	t := &TargetMock{BO: bo}
-	core.Tinit(t, NewStatsTracker(), nil /*config*/, false /*run HK*/)
+	core.Tinit(t, nil /*config*/, false /*run HK*/)
 	return t
 }
 
 func (t *TargetMock) Bowner() meta.Bowner { return t.BO }
 func (t *TargetMock) Sowner() meta.Sowner { return t.SO }
 
-func (*TargetMock) SID() string              { return mockID }
-func (*TargetMock) String() string           { return "tmock" }
-func (*TargetMock) Snode() *meta.Snode       { return &meta.Snode{DaeID: mockID} }
-func (*TargetMock) ClusterStarted() bool     { return true }
-func (*TargetMock) NodeStarted() bool        { return true }
-func (*TargetMock) DataClient() *http.Client { return http.DefaultClient }
-func (*TargetMock) PageMM() *memsys.MMSA     { return memsys.PageMM() }
-func (*TargetMock) ByteMM() *memsys.MMSA     { return memsys.ByteMM() }
+func (*TargetMock) SID() string                    { return mockID }
+func (*TargetMock) String() string                 { return "tmock" }
+func (*TargetMock) Snode() *meta.Snode             { return &meta.Snode{DaeID: mockID} }
+func (*TargetMock) ClusterStarted() bool           { return true }
+func (*TargetMock) NodeStarted() bool              { return true }
+func (*TargetMock) DataClient() *http.Client       { return http.DefaultClient }
+func (*TargetMock) StatsUpdater() cos.StatsUpdater { return NewStatsTracker() }
+func (*TargetMock) PageMM() *memsys.MMSA           { return memsys.PageMM() }
+func (*TargetMock) ByteMM() *memsys.MMSA           { return memsys.ByteMM() }
 
 func (*TargetMock) MaxUtilLoad() (int64, float64) { return 0, 0 }
 

diff --git a/core/node.go b/core/node.go
@@ -1,11 +1,12 @@
 // Package core provides core metadata and in-cluster API
 /*
- * Copyright (c) 2018-2024, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
  */
 package core
 
 import (
 	"github.com/NVIDIA/aistore/cmn"
+	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/memsys"
 )
@@ -30,6 +31,8 @@ type (
 		ClusterStarted() bool
 		NodeStarted() bool
 
+		StatsUpdater() cos.StatsUpdater
+
 		// Memory allocators
 		PageMM() *memsys.MMSA
 		ByteMM() *memsys.MMSA

diff --git a/ext/dload/dispatcher.go b/ext/dload/dispatcher.go
@@ -21,7 +21,6 @@ import (
 	"github.com/NVIDIA/aistore/core"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/fs"
-	"github.com/NVIDIA/aistore/stats"
 	"github.com/NVIDIA/aistore/xact/xreg"
 	"golang.org/x/sync/errgroup"
 )
@@ -46,9 +45,8 @@ type (
 	}
 
 	global struct {
-		tstats stats.Tracker
-		db     kvdb.Driver
-		store  *infoStore
+		db    kvdb.Driver
+		store *infoStore
 
 		// Downloader selects one of the two clients (below) by the destination URL.
 		// Certification check is disabled for now and does not depend on cluster settings.
@@ -61,16 +59,13 @@ type (
 
 var g global
 
-func Init(tstats stats.Tracker, db kvdb.Driver, clientConf *cmn.ClientConf) {
+func Init(db kvdb.Driver, clientConf *cmn.ClientConf) {
 	g.clientH, g.clientTLS = cmn.NewDefaultClients(clientConf.TimeoutLong.D())
 
 	if db == nil { // unit tests only
 		return
 	}
-	{
-		g.tstats = tstats
-		g.db = db
-	}
+	g.db = db
 	xreg.RegNonBckXact(&factory{})
 }
 

diff --git a/ext/dload/task.go b/ext/dload/task.go
@@ -98,7 +98,7 @@ func (task *singleTask) download(lom *core.LOM) {
 
 	vlabs := map[string]string{stats.VarlabBucket: lom.Bck().Cname("")}
 	lsize := task.currentSize.Load()
-	g.tstats.AddWith(
+	core.T.StatsUpdater().AddWith(
 		cos.NamedVal64{Name: stats.DloadSize, Value: lsize, VarLabs: vlabs},
 		cos.NamedVal64{Name: stats.DloadLatencyTotal, Value: int64(task.ended.Load().Sub(task.started.Load())), VarLabs: vlabs},
 	)
@@ -251,7 +251,7 @@ func (task *singleTask) wrapReader(r io.ReadCloser) io.ReadCloser {
 // Probably we need to extend the persistent database (db.go) so that it will contain
 // also information about specific tasks.
 func (task *singleTask) markFailed(statusMsg string) {
-	g.tstats.Inc(stats.ErrDloadCount)
+	core.T.StatsUpdater().Inc(stats.ErrDloadCount)
 	g.store.persistError(task.jobID(), task.obj.objName, statusMsg)
 	g.store.incErrorCnt(task.jobID())
 }

diff --git a/ext/dload/utils_test.go b/ext/dload/utils_test.go
@@ -56,7 +56,7 @@ func TestCompareObject(t *testing.T) {
 	// initialize http clients
 	clientConf.Timeout = 5 * cos.Duration(time.Second)
 	clientConf.TimeoutLong = 15 * cos.Duration(time.Second)
-	dload.Init(nil, nil, &clientConf)
+	dload.Init(nil, &clientConf)
 
 	// Modify local object to contain invalid (meta)data.
 	customMD := cos.StrKVs{