[config change] add cluster config (tco, arch) sections; tcb burst

alex-aizman · alex-aizman · commit fc3d8f31665f · 2025-04-18T11:57:14.000-04:00
* add `TCOConf` and `ArchConf` (new)
* add TCB (channel size) burst
* all three sections (`TCBConf`, `TCOConf`, `ArchConf`) will now have same knobs
  - embed the same struct
  - honor configured compression - ie., utilize transport to compress in
    flight
  - share the same config validation
* burst (chan size): 256 minimum w/ validated upper limit 10,000
* max parallelism += 4

Signed-off-by: Alex Aizman &lt;alex.aizman@gmail.com&gt;
diff --git a/cmd/aisinit/config.go b/cmd/aisinit/config.go
@@ -37,8 +37,11 @@ var (
 	}
 
 	defaultTCB = aiscmn.TCBConf{
-		Compression: aisapc.CompressNever,
-		SbundleMult: 2,
+		XactConf: aiscmn.XactConf{
+			Compression: aisapc.CompressNever,
+			SbundleMult: 2,
+			Burst:       512,
+		},
 	}
 
 	defaultDisk = aiscmn.DiskConf{
diff --git a/cmn/config.go b/cmn/config.go
@@ -102,7 +102,9 @@ type (
 		Proxy       ProxyConf       `json:"proxy" allow:"cluster"`
 		Auth        AuthConf        `json:"auth"`
 		Cksum       CksumConf       `json:"checksum"`
-		TCB         TCBConf         `json:"tcb"` // transform (offline) or copy src bucket => dst bucket
+		TCB         TCBConf         `json:"tcb"`
+		TCO         TCOConf         `json:"tco"`
+		Arch        ArchConf        `json:"arch"`
 		Tracing     TracingConf     `json:"tracing"`
 		Keepalive   KeepaliveConf   `json:"keepalivetracker"`
 		Rebalance   RebalanceConf   `json:"rebalance" allow:"cluster"`
@@ -655,15 +657,30 @@ type (
 		MinPctFree     *int          `json:"min_pct_free,omitempty"`
 	}
 
-	TCBConf struct {
+	// generic xaction --
+	XactConf struct {
 		Compression string `json:"compression"`       // enum { CompressAlways, ... } in api/apc/compression.go
 		SbundleMult int    `json:"bundle_multiplier"` // stream-bundle multiplier: num streams to destination
+		Burst       int    `json:"burst_buffer"`      // xaction channel (buffer) size
 	}
-	TCBConfToSet struct {
+	XactConfToSet struct {
 		Compression *string `json:"compression,omitempty"`
 		SbundleMult *int    `json:"bundle_multiplier,omitempty"`
+		Burst       *int    `json:"burst_buffer,omitempty"`
 	}
 
+	// bucket-to-bucket copy/transform
+	TCBConf      struct{ XactConf }
+	TCBConfToSet struct{ XactConfToSet }
+
+	// multi-object copy/transform
+	TCOConf      struct{ XactConf }
+	TCOConfToSet struct{ XactConfToSet }
+
+	// multi-object archive (multiple objects => shard)
+	ArchConf      struct{ XactConf }
+	ArchConfToSet struct{ XactConfToSet }
+
 	WritePolicyConf struct {
 		Data apc.WritePolicy `json:"data"`
 		MD   apc.WritePolicy `json:"md"`
@@ -1779,18 +1796,21 @@ func (c *TransportConf) Validate() (err error) {
 	return nil
 }
 
-/////////////
-// TCBConf //
-/////////////
+//////////////
+// XactConf //
+//////////////
 
-func (c *TCBConf) Validate() error {
-	if c.SbundleMult < 0 || c.SbundleMult > 16 {
-		return fmt.Errorf("invalid tcb.bundle_multiplier: %v (expected range [0, 16])", c.SbundleMult)
-	}
+func (c *XactConf) Validate() error {
 	if !apc.IsValidCompression(c.Compression) {
-		return fmt.Errorf("invalid tcb.compression: %q (expecting one of: %v)",
+		return fmt.Errorf("invalid compression: %q (expecting one of: %v)",
 			c.Compression, apc.SupportedCompression)
 	}
+	if c.SbundleMult < 0 || c.SbundleMult > 16 {
+		return fmt.Errorf("invalid bundle_multiplier: %v (expected range [0, 16])", c.SbundleMult)
+	}
+	if c.Burst < 0 || c.Burst > 10_000 {
+		return fmt.Errorf("invalid burst_buffer: %v (expected range [0, 10000])", c.Burst)
+	}
 	return nil
 }
 
diff --git a/xact/xs/archive.go b/xact/xs/archive.go
@@ -28,6 +28,7 @@ import (
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/fs"
 	"github.com/NVIDIA/aistore/transport"
+	"github.com/NVIDIA/aistore/transport/bundle"
 	"github.com/NVIDIA/aistore/xact"
 	"github.com/NVIDIA/aistore/xact/xreg"
 )
@@ -114,18 +115,31 @@ func (p *archFactory) Start() (err error) {
 	if err != nil {
 		return err
 	}
-	//
+
 	// new x-archive
-	//
-	r := &XactArch{streamingX: streamingX{p: &p.streamingF, config: cmn.GCO.Get()}}
-	r.pending.m = make(map[string]*archwi, maxNumInParallel)
+	var (
+		config = cmn.GCO.Get()
+		burst  = max(256, config.Arch.Burst)
+		r      = &XactArch{streamingX: streamingX{p: &p.streamingF, config: config}}
+	)
+	r.pending.m = make(map[string]*archwi, burst)
+
 	avail := fs.GetAvail()
 	r.joggers.m = make(map[string]*jogger, len(avail))
 	r.smap = core.T.Sowner().Get()
 	p.xctn = r
-	r.DemandBase.Init(p.UUID(), p.kind, "" /*delay ctlmsg until Do()*/, p.Bck /*from*/, xact.IdleDefault)
 
-	if err := p.newDM(p.Args.UUID /*trname*/, r.recv, r.config, r.smap, cmn.OwtPut, 0 /*pdu*/); err != nil {
+	// delay ctlmsg until DoMsg()
+	r.DemandBase.Init(p.UUID(), p.kind, "" /*ctlmsg*/, p.Bck /*from*/, xact.IdleDefault)
+
+	dmxtra := bundle.Extra{
+		RecvAck:     nil, // no ACKs
+		Config:      r.config,
+		Compression: r.config.Arch.Compression,
+		Multiplier:  r.config.Arch.SbundleMult,
+		SizePDU:     0,
+	}
+	if err := p.newDM(p.Args.UUID /*trname*/, r.recv, r.smap, dmxtra, cmn.OwtPut); err != nil {
 		return err
 	}
 
@@ -163,6 +177,7 @@ func (r *XactArch) BeginMsg(msg *cmn.ArchiveBckMsg, archlom *core.LOM) (err erro
 
 	// bind a new/existing jogger to this archwi based on archlom's mountpath
 	var (
+		burst  = max(256, r.config.Arch.Burst)
 		mpath  = archlom.Mountpath()
 		exists bool
 	)
@@ -171,7 +186,7 @@ func (r *XactArch) BeginMsg(msg *cmn.ArchiveBckMsg, archlom *core.LOM) (err erro
 		r.joggers.m[mpath.Path] = &jogger{
 			r:      r,
 			mpath:  mpath,
-			workCh: make(chan *archtask, maxNumInParallel*2), // TODO -- FIXME: unify
+			workCh: make(chan *archtask, burst),
 		}
 		wi.j = r.joggers.m[mpath.Path]
 		r.joggers.wg.Add(1)
@@ -253,7 +268,7 @@ func (r *XactArch) DoMsg(msg *cmn.ArchiveBckMsg) {
 	// lrpWorkersNone since we need a single writer to serialize adding files
 	// into an eventual `archlom`
 	lrit := &lrit{}
-	err := lrit.init(r, &msg.ListRange, r.Bck(), nwpNone)
+	err := lrit.init(r, &msg.ListRange, r.Bck(), nwpNone, r.config.Arch.Burst)
 	if err != nil {
 		r.Abort(err)
 		r.DecPending()
diff --git a/xact/xs/evict.go b/xact/xs/evict.go
@@ -76,7 +76,7 @@ func newEvictDelete(xargs *xreg.Args, kind string, bck *meta.Bck, msg *apc.ListR
 
 	// default num-workers hardcoded
 	// (currently, always num mountpaths)
-	if err := r.lrit.init(r, msg, bck, nwpDflt); err != nil {
+	if err := r.lrit.init(r, msg, bck, nwpDflt, 0 /*burst*/); err != nil {
 		return nil, err
 	}
 
diff --git a/xact/xs/lrit.go b/xact/xs/lrit.go
@@ -91,7 +91,7 @@ type (
 // lrit //
 //////////
 
-func (r *lrit) init(xctn lrxact, msg *apc.ListRange, bck *meta.Bck, numWorkers int) error {
+func (r *lrit) init(xctn lrxact, msg *apc.ListRange, bck *meta.Bck, numWorkers, confBurst int) error {
 	l := len(fs.GetAvail())
 	if l == 0 {
 		xctn.Abort(cmn.ErrNoMountpaths)
@@ -138,18 +138,18 @@ func (r *lrit) init(xctn lrxact, msg *apc.ListRange, bck *meta.Bck, numWorkers i
 		return err
 	}
 
-	r._iniNwp(numWorkers)
+	r._iniNwp(numWorkers, confBurst)
 	return nil
 }
 
-func (r *lrit) _iniNwp(numWorkers int) {
+func (r *lrit) _iniNwp(numWorkers, confBurst int) {
 	r.nwp.workers = make([]*lrworker, 0, numWorkers)
 	for range numWorkers {
 		r.nwp.workers = append(r.nwp.workers, &lrworker{r})
 	}
 
 	// [burst] work channel capacity: up to 4 pending work items per
-	r.nwp.workCh = make(chan lrpair, numWorkers*nwpBurst)
+	r.nwp.workCh = make(chan lrpair, max(numWorkers*nwpBurst, confBurst))
 	nlog.Infoln(r.parent.Name(), "workers:", numWorkers)
 }
 
diff --git a/xact/xs/prefetch.go b/xact/xs/prefetch.go
@@ -105,7 +105,7 @@ func (*prfFactory) WhenPrevIsRunning(xreg.Renewable) (xreg.WPR, error) {
 func newPrefetch(xargs *xreg.Args, kind string, bck *meta.Bck, msg *apc.PrefetchMsg) (r *prefetch, err error) {
 	r = &prefetch{config: cmn.GCO.Get(), msg: msg}
 
-	err = r.lrit.init(r, &msg.ListRange, bck, msg.NumWorkers)
+	err = r.lrit.init(r, &msg.ListRange, bck, msg.NumWorkers, 0 /*burst*/)
 	if err != nil {
 		return nil, err
 	}
diff --git a/xact/xs/streaming.go b/xact/xs/streaming.go
@@ -31,9 +31,6 @@ const (
 	waitRegRecv   = 4 * time.Second
 	waitUnregRecv = 2 * waitRegRecv
 	waitUnregMax  = 2 * waitUnregRecv
-
-	// TODO -- FIXME: grow 2x upon chanFull
-	maxNumInParallel = 256
 )
 
 type (
@@ -106,16 +103,14 @@ func (p *streamingF) genBEID(fromBck, toBck *meta.Bck) (string, error) {
 	return "", err
 }
 
-func (p *streamingF) newDM(trname string, recv transport.RecvObj, config *cmn.Config, smap *meta.Smap, owt cmn.OWT, sizePDU int32) error {
+func (p *streamingF) newDM(trname string, recv transport.RecvObj, smap *meta.Smap, dmxtra bundle.Extra, owt cmn.OWT) error {
 	if err := core.InMaintOrDecomm(smap, core.T.Snode(), p.xctn); err != nil {
 		return err
 	}
 	if smap.CountActiveTs() <= 1 {
 		return nil
 	}
 
-	// consider adding config.X.Compression, config.X.SbundleMult (currently, always 1), etc.
-	dmxtra := bundle.Extra{Config: config, Multiplier: 1, SizePDU: sizePDU}
 	p.dm = bundle.NewDM(trname, recv, owt, dmxtra)
 
 	err := p.dm.RegRecv()
diff --git a/xact/xs/tcb.go b/xact/xs/tcb.go
@@ -165,7 +165,7 @@ func (r *XactTCB) _iniNwp(numWorkers int) {
 	for range numWorkers {
 		r.numwp.workers = append(r.numwp.workers, tcbworker{r})
 	}
-	r.numwp.workCh = make(chan core.LIF, numWorkers*nwpBurst)
+	r.numwp.workCh = make(chan core.LIF, max(numWorkers*nwpBurst, r.Config.TCB.Burst))
 	r.numwp.stopCh = cos.NewStopCh()
 	nlog.Infoln(r.Name(), "workers:", numWorkers)
 }
@@ -333,6 +333,7 @@ func (r *XactTCB) Run(wg *sync.WaitGroup) {
 		r.dm.UnregRecv()
 	}
 	if r.args.Msg.Sync {
+		// TODO -- FIXME: revisit stopCh and related
 		r.prune.wait()
 	}
 
diff --git a/xact/xs/tcobjs.go b/xact/xs/tcobjs.go
@@ -26,6 +26,7 @@ import (
 	"github.com/NVIDIA/aistore/memsys"
 	"github.com/NVIDIA/aistore/stats"
 	"github.com/NVIDIA/aistore/transport"
+	"github.com/NVIDIA/aistore/transport/bundle"
 	"github.com/NVIDIA/aistore/xact"
 	"github.com/NVIDIA/aistore/xact/xreg"
 )
@@ -100,9 +101,16 @@ func (p *tcoFactory) Start() error {
 	p.Args.UUID = PrefixTcoID + uuid
 
 	// new x-tco
-	workCh := make(chan *cmn.TCOMsg, maxNumInParallel)
-	r := &XactTCO{streamingX: streamingX{p: &p.streamingF, config: cmn.GCO.Get()}, args: p.args, workCh: workCh}
-	r.pend.m = make(map[string]*tcowi, maxNumInParallel)
+	var (
+		config = cmn.GCO.Get()
+		burst  = max(256, config.TCO.Burst)
+		r      = &XactTCO{
+			streamingX: streamingX{p: &p.streamingF, config: config},
+			args:       p.args,
+			workCh:     make(chan *cmn.TCOMsg, burst),
+		}
+	)
+	r.pend.m = make(map[string]*tcowi, burst)
 	r.owt = cmn.OwtCopy
 
 	if p.kind == apc.ActETLObjects {
@@ -131,7 +139,14 @@ func (p *tcoFactory) Start() error {
 
 	// TODO: sentinels require DM; no-DM still requires sentinels
 	if useDM := !r.args.DisableDM; useDM && nat > 1 {
-		if err := p.newDM(p.Args.UUID /*trname*/, r.recv, r.config, smap, r.owt, sizePDU); err != nil {
+		dmxtra := bundle.Extra{
+			RecvAck:     nil, // no ACKs
+			Config:      r.config,
+			Compression: r.config.TCO.Compression,
+			Multiplier:  r.config.TCO.SbundleMult,
+			SizePDU:     sizePDU,
+		}
+		if err := p.newDM(p.Args.UUID /*trname*/, r.recv, smap, dmxtra, r.owt); err != nil {
 			return err
 		}
 	}
@@ -231,7 +246,7 @@ func (r *XactTCO) doMsg(msg *cmn.TCOMsg) (stop bool) {
 	wi.pend.n.Store(int64(r.sntl.nat - 1)) // must dec down to zero
 
 	lrit := &lrit{}
-	if err := lrit.init(r, &msg.ListRange, r.Bck(), msg.NumWorkers); err != nil {
+	if err := lrit.init(r, &msg.ListRange, r.Bck(), msg.NumWorkers, r.config.TCO.Burst); err != nil {
 		r.AddErr(err)
 		return !msg.ContinueOnError // stop?
 	}
@@ -248,6 +263,7 @@ func (r *XactTCO) doMsg(msg *cmn.TCOMsg) (stop bool) {
 	}
 	// run
 	if msg.Sync && lrit.lrp != lrpList {
+		// TODO -- FIXME: revisit stopCh and related
 		wg = &sync.WaitGroup{}
 		wg.Add(1)
 		go func(pt *cos.ParsedTemplate, wg *sync.WaitGroup) {
@@ -256,11 +272,11 @@ func (r *XactTCO) doMsg(msg *cmn.TCOMsg) (stop bool) {
 		}(lrit.pt.Clone(), wg)
 	}
 	err := lrit.run(wi, smap, true /*prealloc buf*/)
+
+	lrit.wait()
 	if wg != nil {
 		wg.Wait()
 	}
-
-	lrit.wait()
 	if r.IsAborted() {
 		return true // stop
 	}
@@ -437,7 +453,7 @@ func (r *XactTCO) prune(pruneit *lrit, smap *meta.Smap, pt *cos.ParsedTemplate)
 	var syncit lrit
 	debug.Assert(pruneit.lrp == lrpRange)
 
-	err := syncit.init(pruneit.parent, pruneit.msg, rp.bckTo, nwpDflt) // TODO -- FIXME: stopCh
+	err := syncit.init(pruneit.parent, pruneit.msg, rp.bckTo, nwpDflt, r.config.TCO.Burst)
 	debug.AssertNoErr(err)
 	syncit.pt = pt
 	syncwi := &syncwi{&rp} // reusing only prune.do (and not init/run/wait)
diff --git a/xact/xs/xutils.go b/xact/xs/xutils.go
@@ -66,10 +66,11 @@ func (rate *tcrate) acquire() {
 //
 
 const (
-	nwpBurst = 4  // burst multiplier (channel size)
-	nwpNone  = -1 // no workers at all - iterated LOMs get executed by the (iterating) goroutine
-	nwpMin   = 2  // throttled
-	nwpDflt  = 0  // (number of mountpaths)
+	nwpBurst = 32 // work channel burst multiplier (channel size = burst * num-workers)
+
+	nwpNone = -1 // no workers at all - iterated LOMs get executed by the (iterating) goroutine
+	nwpMin  = 2  // throttled
+	nwpDflt = 0  // (number of mountpaths)
 )
 
 // strict rules
@@ -82,8 +83,8 @@ func throttleNwp(xname string, n int) (int, error) {
 		return nwpNone, nil
 	}
 
-	// 2. do not exceed num available cores
-	n = min(sys.MaxParallelism(), n)
+	// 2. number of available cores(*)
+	n = min(sys.MaxParallelism()+4, n)
 
 	// 3. factor in memory pressure
 	var (

Original file line number	Diff line number	Diff line change
`@@ -37,8 +37,11 @@ var (`
`37`	`37`	`}`
`38`	`38`
`39`	`39`	`defaultTCB = aiscmn.TCBConf{`
`40`		`- Compression: aisapc.CompressNever,`
`41`		`- SbundleMult: 2,`
	`40`	`+ XactConf: aiscmn.XactConf{`
	`41`	`+ Compression: aisapc.CompressNever,`
	`42`	`+ SbundleMult: 2,`
	`43`	`+ Burst: 512,`
	`44`	`+ },`
`42`	`45`	`}`
`43`	`46`
`44`	`47`	`defaultDisk = aiscmn.DiskConf{`
Original file line number	Diff line number	Diff line change
`@@ -76,7 +76,7 @@ func newEvictDelete(xargs xreg.Args, kind string, bck meta.Bck, msg *apc.ListR`
`76`	`76`
`77`	`77`	`// default num-workers hardcoded`
`78`	`78`	`// (currently, always num mountpaths)`
`79`		`- if err := r.lrit.init(r, msg, bck, nwpDflt); err != nil {`
	`79`	`+ if err := r.lrit.init(r, msg, bck, nwpDflt, 0 /burst/); err != nil {`
`80`	`80`	`return nil, err`
`81`	`81`	`}`
`82`	`82`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ func (*prfFactory) WhenPrevIsRunning(xreg.Renewable) (xreg.WPR, error) {`
`105`	`105`	`func newPrefetch(xargs xreg.Args, kind string, bck meta.Bck, msg apc.PrefetchMsg) (r prefetch, err error) {`
`106`	`106`	`r = &prefetch{config: cmn.GCO.Get(), msg: msg}`
`107`	`107`
`108`		`- err = r.lrit.init(r, &msg.ListRange, bck, msg.NumWorkers)`
	`108`	`+ err = r.lrit.init(r, &msg.ListRange, bck, msg.NumWorkers, 0 /burst/)`
`109`	`109`	`if err != nil {`
`110`	`110`	`return nil, err`
`111`	`111`	`}`
Original file line number	Diff line number	Diff line change
`@@ -31,9 +31,6 @@ const (`
`31`	`31`	`waitRegRecv = 4 * time.Second`
`32`	`32`	`waitUnregRecv = 2 * waitRegRecv`
`33`	`33`	`waitUnregMax = 2 * waitUnregRecv`
`34`		`-`
`35`		`- // TODO -- FIXME: grow 2x upon chanFull`
`36`		`- maxNumInParallel = 256`
`37`	`34`	`)`
`38`	`35`
`39`	`36`	`type (`
`@@ -106,16 +103,14 @@ func (p streamingF) genBEID(fromBck, toBck meta.Bck) (string, error) {`
`106`	`103`	`return "", err`
`107`	`104`	`}`
`108`	`105`
`109`		`-func (p streamingF) newDM(trname string, recv transport.RecvObj, config cmn.Config, smap *meta.Smap, owt cmn.OWT, sizePDU int32) error {`
	`106`	`+func (p streamingF) newDM(trname string, recv transport.RecvObj, smap meta.Smap, dmxtra bundle.Extra, owt cmn.OWT) error {`
`110`	`107`	`if err := core.InMaintOrDecomm(smap, core.T.Snode(), p.xctn); err != nil {`
`111`	`108`	`return err`
`112`	`109`	`}`
`113`	`110`	`if smap.CountActiveTs() <= 1 {`
`114`	`111`	`return nil`
`115`	`112`	`}`
`116`	`113`
`117`		`- // consider adding config.X.Compression, config.X.SbundleMult (currently, always 1), etc.`
`118`		`- dmxtra := bundle.Extra{Config: config, Multiplier: 1, SizePDU: sizePDU}`
`119`	`114`	`p.dm = bundle.NewDM(trname, recv, owt, dmxtra)`
`120`	`115`
`121`	`116`	`err := p.dm.RegRecv()`
Original file line number	Diff line number	Diff line change
`@@ -165,7 +165,7 @@ func (r *XactTCB) _iniNwp(numWorkers int) {`
`165`	`165`	`for range numWorkers {`
`166`	`166`	`r.numwp.workers = append(r.numwp.workers, tcbworker{r})`
`167`	`167`	`}`
`168`		`- r.numwp.workCh = make(chan core.LIF, numWorkers*nwpBurst)`
	`168`	`+ r.numwp.workCh = make(chan core.LIF, max(numWorkers*nwpBurst, r.Config.TCB.Burst))`
`169`	`169`	`r.numwp.stopCh = cos.NewStopCh()`
`170`	`170`	`nlog.Infoln(r.Name(), "workers:", numWorkers)`
`171`	`171`	`}`
`@@ -333,6 +333,7 @@ func (r XactTCB) Run(wg sync.WaitGroup) {`
`333`	`333`	`r.dm.UnregRecv()`
`334`	`334`	`}`
`335`	`335`	`if r.args.Msg.Sync {`
	`336`	`+ // TODO -- FIXME: revisit stopCh and related`
`336`	`337`	`r.prune.wait()`
`337`	`338`	`}`
`338`	`339`