EC streams: synchronize opening cluster-wide; throttle more aggressiv…

…ely when OOM * all ec xactions: synchronize via `GoRunW` * throttle vs OOM * prev. commits: 5eb4677, aaa0bfa Signed-off-by: Alex Aizman <[email protected]>
NVIDIA · Nov 1, 2024 · 441d2dd · 441d2dd
1 parent 485dc87
commit 441d2dd
Show file tree

Hide file tree

Showing 6 changed files with 26 additions and 18 deletions.
diff --git a/ec/bencodex.go b/ec/bencodex.go
@@ -135,10 +135,9 @@ func newXactBckEncode(bck *meta.Bck, uuid string, checkAndRecover bool) (r *Xact
 	return r, nil
 }
 
-func (r *XactBckEncode) Run(wg *sync.WaitGroup) {
-	wg.Done()
-
+func (r *XactBckEncode) Run(gowg *sync.WaitGroup) {
 	ECM.incActive(r)
+	gowg.Done()
 
 	opts := &mpather.JgroupOpts{
 		CTs:      []string{fs.ObjectType},

diff --git a/ec/getx.go b/ec/getx.go
@@ -68,7 +68,8 @@ func (p *getFactory) Start() error {
 	xec := ECM.NewGetXact(p.Bck.Bucket())
 	xec.DemandBase.Init(cos.GenUUID(), p.Kind(), p.Bck, 0 /*use default*/)
 	p.xctn = xec
-	go xec.Run(nil)
+
+	xact.GoRunW(xec)
 	return nil
 }
 func (*getFactory) Kind() string     { return apc.ActECGet }
@@ -184,7 +185,7 @@ func (r *XactGet) dispatchReq(req *request, lom *core.LOM) error {
 	return nil
 }
 
-func (r *XactGet) Run(*sync.WaitGroup) {
+func (r *XactGet) Run(gowg *sync.WaitGroup) {
 	nlog.Infoln(r.Name())
 	for _, jog := range r.getJoggers {
 		go jog.run()
@@ -194,6 +195,7 @@ func (r *XactGet) Run(*sync.WaitGroup) {
 	defer ticker.Stop()
 
 	ECM.incActive(r)
+	gowg.Done()
 
 	// as of now all requests are equal. Some may get throttling later
 	for {

diff --git a/ec/putjogger.go b/ec/putjogger.go
@@ -53,6 +53,7 @@ type (
 		stopCh cos.StopCh    // jogger management channel: to stop it
 
 		ntotal int64 // (throttle to prevent OOM)
+		micro  bool  // (throttle tuneup)
 		toDisk bool  // use files or SGL (NOTE: toDisk == false may cause OOM)
 	}
 )
@@ -147,11 +148,12 @@ func (c *putJogger) _do(req *request, lom *core.LOM) {
 		c.parent.AddErr(err, 0)
 	} else if !c.toDisk { // throttle
 		c.ntotal++
-		if fs.IsMiniThrottle(c.ntotal) {
-			if pressure := g.pmm.Pressure(); pressure >= memsys.PressureExtreme {
+		if (c.micro && fs.IsMicroThrottle(c.ntotal)) || fs.IsMiniThrottle(c.ntotal) {
+			if pressure := g.pmm.Pressure(); pressure >= memsys.PressureHigh {
 				time.Sleep(fs.Throttle100ms)
-			} else if pressure == memsys.PressureHigh {
-				time.Sleep(fs.Throttle10ms)
+				if !c.micro && pressure >= memsys.PressureExtreme {
+					c.micro = true
+				}
 			}
 		}
 	}

diff --git a/ec/putx.go b/ec/putx.go
@@ -68,7 +68,8 @@ func (p *putFactory) Start() error {
 	xec := ECM.NewPutXact(p.Bck.Bucket())
 	xec.DemandBase.Init(cos.GenUUID(), p.Kind(), p.Bck, 0 /*use default*/)
 	p.xctn = xec
-	go xec.Run(nil)
+
+	xact.GoRunW(xec)
 	return nil
 }
 
@@ -153,7 +154,7 @@ func (r *XactPut) dispatchRequest(req *request, lom *core.LOM) error {
 	return nil
 }
 
-func (r *XactPut) Run(*sync.WaitGroup) {
+func (r *XactPut) Run(gowg *sync.WaitGroup) {
 	nlog.Infoln(r.Name())
 
 	var wg sync.WaitGroup
@@ -163,6 +164,7 @@ func (r *XactPut) Run(*sync.WaitGroup) {
 	}
 
 	ECM.incActive(r)
+	gowg.Done()
 
 	ticker := time.NewTicker(r.config.Periodic.StatsTime.D())
 	r.mainLoop(ticker)
@@ -172,7 +174,6 @@ func (r *XactPut) Run(*sync.WaitGroup) {
 	r.Finish()
 }
 
-// all requests are equal, throttle TODO
 func (r *XactPut) mainLoop(ticker *time.Ticker) {
 	for {
 		select {

diff --git a/ec/respondx.go b/ec/respondx.go
@@ -63,7 +63,8 @@ func (p *rspFactory) Start() error {
 	xec := ECM.NewRespondXact(p.Bck.Bucket())
 	xec.DemandBase.Init(cos.GenUUID(), p.Kind(), p.Bck, 0 /*use default*/)
 	p.xctn = xec
-	go xec.Run(nil)
+
+	xact.GoRunW(xec)
 	return nil
 }
 
@@ -77,10 +78,11 @@ func newRespondXact(bck *cmn.Bck, mgr *Manager) *XactRespond {
 	return xctn
 }
 
-func (r *XactRespond) Run(*sync.WaitGroup) {
+func (r *XactRespond) Run(gowg *sync.WaitGroup) {
 	nlog.Infoln(r.Name())
 
 	ECM.incActive(r)
+	gowg.Done()
 
 	ticker := time.NewTicker(r.config.Periodic.StatsTime.D())
 	defer ticker.Stop()

diff --git a/fs/throttle.go b/fs/throttle.go
@@ -12,9 +12,11 @@ const (
 	Throttle10ms  = 10 * time.Millisecond
 	Throttle100ms = 100 * time.Millisecond
 
-	throttleBatch = 0x1f      // a.k.a. unit or period
-	throMiniBatch = 0x1f >> 1 // ditto
+	throttleBatch  = 0x1f // a.k.a. unit or period
+	throMiniBatch  = 0x1f >> 1
+	throMicroBatch = 0x1f >> 2
 )
 
-func IsThrottle(n int64) bool     { return n&throttleBatch == throttleBatch }
-func IsMiniThrottle(n int64) bool { return n&throMiniBatch == throMiniBatch }
+func IsThrottle(n int64) bool      { return n&throttleBatch == throttleBatch }
+func IsMiniThrottle(n int64) bool  { return n&throMiniBatch == throMiniBatch }
+func IsMicroThrottle(n int64) bool { return n&throMicroBatch == throMicroBatch }