rate-limited backend: complete transition

alex-aizman · alex-aizman · commit b01194590874 · 2025-03-16T13:23:45.000-04:00
* _proactive_ and _reactive_ rate-limiting - while taking into account num-nodes but not num-workers - TODO: document * rewrite and refactor ais/rlbackend - common 'retry(closure)' * support all 4 (four) backend APIs * remove backend rate-limiting logic from all batch jobs - leave only _proactive_ rate limiting - consolidate and simplify * part four, prev. commit: 8ee8b44 Signed-off-by: Alex Aizman <alex.aizman@gmail.com>
diff --git a/ais/proxy.go b/ais/proxy.go
@@ -3437,8 +3437,7 @@ func (p *proxy) ratelimit(bck *meta.Bck, verb string, smap *smapX) error {
 	if ok {
 		brl = v.(*cos.BurstRateLim)
 	} else {
-		// ignore sleep time - only relevant for clients
-		brl, _ = bck.NewFrontendRateLim(smap.CountActivePs())
+		brl = bck.NewFrontendRateLim(smap.CountActivePs())
 		rl.Store(uhash, brl)
 	}
 	if !brl.TryAcquire() {
diff --git a/ais/rlbackend.go b/ais/rlbackend.go
@@ -12,13 +12,18 @@ import (
 
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
+	"github.com/NVIDIA/aistore/cmn/debug"
 	"github.com/NVIDIA/aistore/core"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/stats"
 	"github.com/NVIDIA/aistore/xact"
 )
 
-// NOTE: backend.GetObjReader is handled elsewhere - by xactions
+// rate-limit 4 backend APIs:
+// - PutObj()
+// - DeleteObj()
+// - GetObj()
+// - GetObjReader()
 
 type (
 	rlbackend struct {
@@ -27,96 +32,114 @@ type (
 	}
 )
 
-func (bp *rlbackend) GetObj(ctx context.Context, lom *core.LOM, owt cmn.OWT, origReq *http.Request) (ecode int, err error) {
-	ecode, err = bp.Backend.GetObj(ctx, lom, owt, origReq)
-	if err == nil {
-		return ecode, nil
+func (bp *rlbackend) GetObj(ctx context.Context, lom *core.LOM, owt cmn.OWT, origReq *http.Request) (int, error) {
+	// proactive
+	arl := bp.acquire(lom.Bck(), http.MethodGet)
+	ecode, err := bp.Backend.GetObj(ctx, lom, owt, origReq)
+	if err == nil || arl == nil || !cmn.IsErrTooManyRequests(err) {
+		return ecode, err
 	}
 
-	var tot time.Duration
-	for tot < cos.DfltRateMaxWait {
-		sleep := bp.retry(err, lom.Bck(), http.MethodGet)
-		if sleep == 0 {
-			break
-		}
-		tot += sleep
-		ecode, err = bp.Backend.GetObj(ctx, lom, owt, origReq)
-		if err == nil {
-			break
-		}
+	cb := func() (int, error) {
+		return bp.Backend.GetObj(ctx, lom, owt, origReq)
 	}
+	tot, code, e := bp.retry(ctx, arl, cb)
+
 	bp.stats(ctx, lom.Bck(), stats.GetRateRetryLatencyTotal, tot)
-	return ecode, err
+	return code, e
 }
 
-func (bp *rlbackend) PutObj(ctx context.Context, r io.ReadCloser, lom *core.LOM, origReq *http.Request) (ecode int, err error) {
-	ecode, err = bp.Backend.PutObj(ctx, r, lom, origReq)
-	if err == nil {
-		return ecode, nil
+func (bp *rlbackend) GetObjReader(ctx context.Context, lom *core.LOM, offset, length int64) (res core.GetReaderResult) {
+	// proactive
+	arl := bp.acquire(lom.Bck(), http.MethodGet)
+	res = bp.Backend.GetObjReader(ctx, lom, offset, length)
+	if res.Err == nil || arl == nil || !cmn.IsErrTooManyRequests(res.Err) {
+		return res
 	}
 
-	var tot time.Duration
-	for tot < cos.DfltRateMaxWait {
-		sleep := bp.retry(err, lom.Bck(), http.MethodPut)
-		if sleep == 0 {
-			break
-		}
-		tot += sleep
-		ecode, err = bp.Backend.PutObj(ctx, r, lom, origReq)
-		if err == nil {
-			break
-		}
+	cb := func() (int, error) {
+		res = bp.Backend.GetObjReader(ctx, lom, offset, length)
+		return res.ErrCode, res.Err
+	}
+	tot, code, e := bp.retry(ctx, arl, cb)
+	debug.Assertf(res.ErrCode == code && res.Err == e, "(%d, %v) vs (%d, %v)", res.ErrCode, res.Err, code, e)
+
+	bp.stats(ctx, lom.Bck(), stats.GetRateRetryLatencyTotal, tot)
+	return res
+}
+
+func (bp *rlbackend) PutObj(ctx context.Context, r io.ReadCloser, lom *core.LOM, origReq *http.Request) (int, error) {
+	// proactive
+	arl := bp.acquire(lom.Bck(), http.MethodPut)
+	ecode, err := bp.Backend.PutObj(ctx, r, lom, origReq)
+	if err == nil || arl == nil || !cmn.IsErrTooManyRequests(err) {
+		return ecode, err
+	}
+
+	cb := func() (int, error) {
+		return bp.Backend.PutObj(ctx, r, lom, origReq)
 	}
+	tot, code, e := bp.retry(ctx, arl, cb)
+
 	bp.stats(ctx, lom.Bck(), stats.PutRateRetryLatencyTotal, tot)
-	return ecode, err
+	return code, e
 }
 
-func (bp *rlbackend) DeleteObj(ctx context.Context, lom *core.LOM) (ecode int, err error) {
-	ecode, err = bp.Backend.DeleteObj(ctx, lom)
-	if err == nil {
-		return ecode, nil
+func (bp *rlbackend) DeleteObj(ctx context.Context, lom *core.LOM) (int, error) {
+	// proactive
+	arl := bp.acquire(lom.Bck(), http.MethodDelete)
+	ecode, err := bp.Backend.DeleteObj(ctx, lom)
+	if err == nil || arl == nil || !cmn.IsErrTooManyRequests(err) {
+		return ecode, err
 	}
 
-	var tot time.Duration
-	for tot < cos.DfltRateMaxWait {
-		sleep := bp.retry(err, lom.Bck(), http.MethodDelete)
-		if sleep == 0 {
-			break
-		}
-		tot += sleep
-		ecode, err = bp.Backend.DeleteObj(ctx, lom)
-		if err == nil {
-			break
-		}
+	cb := func() (int, error) {
+		return bp.Backend.DeleteObj(ctx, lom)
 	}
+	tot, code, e := bp.retry(ctx, arl, cb)
+
 	bp.stats(ctx, lom.Bck(), stats.DeleteRateRetryLatencyTotal, tot)
-	return ecode, err
+	return code, e
 }
 
 //
 // apply (compare with ais/prate_limit)
 //
 
-func (bp *rlbackend) retry(err error, bck *meta.Bck, verb string) time.Duration {
-	if !cmn.IsErrTooManyRequests(err) || !bck.Props.RateLimit.Backend.Enabled {
-		return 0 // not retrying
+func (bp *rlbackend) acquire(bck *meta.Bck, verb string) (arl *cos.AdaptRateLim) {
+	if !bck.Props.RateLimit.Backend.Enabled {
+		return nil
 	}
-
 	var (
-		arl   *cos.AdaptRateLim
 		uhash = bck.HashUname(verb) // see also (*RateLimitConf)verbs
 		v, ok = bp.t.ratelim.Load(uhash)
 	)
 	if ok {
 		arl = v.(*cos.AdaptRateLim)
 	} else {
 		smap := bp.t.owner.smap.get()
-		arl, _ = bck.NewBackendRateLim(smap.CountActiveTs())
+		arl = bck.NewBackendRateLim(smap.CountActiveTs())
 		bp.t.ratelim.Store(uhash, arl)
 	}
 
-	// onerr
-	return arl.OnErr()
+	arl.RetryAcquire(time.Second)
+	return arl
+}
+
+func (*rlbackend) retry(ctx context.Context, arl *cos.AdaptRateLim, cb func() (int, error)) (tot time.Duration, ecode int, err error) {
+	for tot < cos.DfltRateMaxWait {
+		// reactive
+		sleep := arl.OnErr()
+		tot += sleep
+		if err = ctx.Err(); err != nil {
+			break
+		}
+		ecode, err = cb()
+		if err == nil || !cmn.IsErrTooManyRequests(err) {
+			break
+		}
+	}
+	return tot, ecode, err
 }
 
 func (bp *rlbackend) stats(ctx context.Context, bck *meta.Bck, metric string, total time.Duration) {
diff --git a/cmn/cos/rate_limit.go b/cmn/cos/rate_limit.go
@@ -342,3 +342,13 @@ func (brl *BurstRateLim) TryAcquire() bool {
 func (brl *BurstRateLim) recompute(factor float64) {
 	brl.minBtwn = time.Duration(float64(brl.minBtwn) * factor)
 }
+
+// with exponential backoff
+func (brl *BurstRateLim) RetryAcquire(sleep time.Duration) {
+	for ; sleep < DfltRateMaxWait; sleep += sleep >> 1 {
+		if brl.TryAcquire() {
+			return
+		}
+		time.Sleep(sleep)
+	}
+}
diff --git a/core/meta/bck.go b/core/meta/bck.go
@@ -10,7 +10,6 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
@@ -278,29 +277,24 @@ func (b *Bck) MaxPageSize() int64 {
 // rate limits: frontend, backend with respect to `nat` (number active targets)
 //
 
-func (b *Bck) NewFrontendRateLim(na int) (*cos.BurstRateLim, time.Duration) {
+func (b *Bck) NewFrontendRateLim(na int) *cos.BurstRateLim {
 	conf := b.Props.RateLimit.Frontend
 	if !conf.Enabled {
-		return nil, 0
+		return nil
 	}
 	maxTokens := cos.DivRound(conf.MaxTokens, na)
 	brl, err := cos.NewBurstRateLim(maxTokens, conf.Size, conf.Interval.D())
 	if err != nil {
 		nlog.ErrorDepth(1, err)
 		debug.AssertNoErr(err)
 	}
-	return brl, divRound(conf.Interval.D(), maxTokens)
+	return brl
 }
 
-func divRound(ival time.Duration, maxTokens int) time.Duration {
-	a, b := int64(ival), int64(maxTokens)
-	return time.Duration((a + b/2) / b)
-}
-
-func (b *Bck) NewBackendRateLim(nat int) (*cos.AdaptRateLim, time.Duration) {
+func (b *Bck) NewBackendRateLim(nat int) *cos.AdaptRateLim {
 	conf := b.Props.RateLimit.Backend
 	if !conf.Enabled {
-		return nil, 0
+		return nil
 	}
 	if b.IsCloud() && conf.NumRetries < 3 {
 		nlog.Warningf("%s: rate_limit.backend.num_retries set to %d is, which is dangerously low", b.Cname(""), conf.NumRetries)
@@ -314,5 +308,5 @@ func (b *Bck) NewBackendRateLim(nat int) (*cos.AdaptRateLim, time.Duration) {
 		nlog.Errorln(err)
 		debug.AssertNoErr(err)
 	}
-	return arl, divRound(conf.Interval.D(), maxTokens)
+	return arl
 }
diff --git a/xact/xs/prefetch.go b/xact/xs/prefetch.go
@@ -55,11 +55,8 @@ type (
 		config *cmn.Config
 		msg    *apc.PrefetchMsg
 		vlabs  map[string]string
-		rate   struct {
-			arl   *cos.AdaptRateLim
-			sleep time.Duration
-		}
-		pebl pebl
+		brl    *cos.BurstRateLim
+		pebl   pebl
 		lrit
 		xact.Base
 		latestVer bool
@@ -108,10 +105,9 @@ func newPrefetch(xargs *xreg.Args, kind string, bck *meta.Bck, msg *apc.Prefetch
 	r.InitBase(xargs.UUID, kind, msg.Str(r.lrp == lrpPrefix), bck)
 	r.latestVer = bck.VersionConf().ValidateWarmGet || msg.LatestVer
 
-	// TODO: support RateLimitConf.Verbs, here and elsewhere
 	smap := core.T.Sowner().Get()
 	nat := smap.CountActiveTs()
-	r.rate.arl, r.rate.sleep = bck.NewBackendRateLim(nat)
+	r.brl = bck.NewFrontendRateLim(nat) // TODO: support RateLimitConf.Verbs - here and elsewhere; `nat` vs num-workers
 
 	r.bp = core.T.Backend(bck)
 	r.vlabs = map[string]string{
@@ -178,8 +174,9 @@ func (r *prefetch) do(lom *core.LOM, lrit *lrit) {
 		goto eret
 	}
 
-	if r.rate.arl != nil {
-		r.rate.arl.RetryAcquire(r.rate.sleep)
+	// apply frontend rate-limit, if any
+	if r.brl != nil {
+		r.brl.RetryAcquire(time.Second)
 	}
 	if r.msg.BlobThreshold > 0 && size >= r.msg.BlobThreshold && !r.pebl.busy() {
 		err = r.blobdl(lom, oa)
diff --git a/xact/xs/tcb.go b/xact/xs/tcb.go
@@ -44,7 +44,7 @@ type (
 		bp    core.Backend // backend(source bucket)
 		p     *tcbFactory
 		dm    *bundle.DataMover
-		rate  *tcrate
+		rate  tcrate
 		vlabs map[string]string
 		prune prune
 		nam   string
@@ -192,7 +192,7 @@ func newTCB(p *tcbFactory, slab *memsys.Slab, config *cmn.Config, smap *meta.Sma
 		r.str = r.Base.String() + "<=" + args.BckFrom.Cname(msg.Prefix)
 	}
 
-	r.rate = newrate(args.BckFrom, args.BckTo, nat)
+	r.rate.init(args.BckFrom, args.BckTo, nat)
 
 	if msg.Sync {
 		debug.Assert(msg.Prepend == "", msg.Prepend) // validated (cli, P)
@@ -295,16 +295,14 @@ func (r *XactTCB) do(lom *core.LOM, buf []byte) error {
 		nlog.Infoln(r.Base.Name()+":", lom.Cname(), "=>", args.BckTo.Cname(toName))
 	}
 
-	if r.rate != nil {
-		if r.rate.src.rl != nil {
-			r.rate.src.rl.RetryAcquire(r.rate.src.sleep) // with exponential backoff
-		}
-		if r.rate.dst.rl != nil {
-			r.rate.dst.rl.RetryAcquire(r.rate.dst.sleep) // ditto
-		}
+	// apply frontend rate-limit, if any
+	if r.rate.src != nil {
+		r.rate.src.RetryAcquire(time.Second)
+	}
+	if r.rate.dst != nil {
+		r.rate.dst.RetryAcquire(time.Second)
 	}
 
-retry:
 	a := AllocCOI()
 	{
 		a.DP = args.DP
@@ -343,12 +341,6 @@ retry:
 		// do nothing
 	case cos.IsErrOOS(res.Err):
 		r.Abort(res.Err)
-	case cmn.IsErrTooManyRequests(res.Err):
-		if r.rate != nil {
-			r.rate.onmanyreq(r.vlabs)
-			goto retry
-		}
-		fallthrough
 	default:
 		r.AddErr(res.Err, 5, cos.SmoduleXs)
 	}
diff --git a/xact/xs/tcobjs.go b/xact/xs/tcobjs.go
diff --git a/xact/xs/xutils.go b/xact/xs/xutils.go
