use weight in context in more throttlers

Author: 1pkg

README.MD

@@ -73,7 +73,7 @@ func WithPriority(ctx context.Context, priority uint8) context.Context
 func WithWeight(ctx context.Context, weight int64) context.Context
 // WithKey adds the provided key to the provided context
 // to add additional call identifier to context.
-// Resulted context is used by: `pattern` and `generator` throtttlers.
+// Resulted context is used by: `before`, `after`, `timed`, `adaptive`, `semaphore` and `cellrate` throtttlers.
 func WithKey(ctx context.Context, key string) context.Context
 // WithMessage adds the provided message to the provided context
 // to add additional message that need to be used to context.
@@ -141,21 +141,21 @@ You can find list of returning error types for all existing throttlers in thrott
 | context | `func NewThrottlerContext() Throttler` | Always throttless on *done* context.<br> - could return `ErrorInternal`; |
 | panic | `func NewThrottlerPanic() Throttler` | Always panics with `ErrorInternal`. |
 | each | `func NewThrottlerEach(threshold uint64) Throttler` | Throttles each periodic *i-th* call defined by the specified threshold.<br> - could return `ErrorThreshold`; |
-| before | `func NewThrottlerBefore(threshold uint64) Throttler` | Throttles each call below the *i-th* call defined by the specified threshold.<br> - could return `ErrorThreshold`; |
-| after | `func NewThrottlerAfter(threshold uint64) Throttler` | Throttles each call after the *i-th* call defined by the specified threshold.<br> - could return `ErrorThreshold`; |
+| before | `func NewThrottlerBefore(threshold uint64) Throttler` | Throttles each call below the *i-th* call defined by the specified threshold.<br>Use `WithWeight` to override context call qunatity, 1 by default.<br> - could return `ErrorThreshold`; |
+| after | `func NewThrottlerAfter(threshold uint64) Throttler` | Throttles each call after the *i-th* call defined by the specified threshold.<br>Use `WithWeight` to override context call qunatity, 1 by default.<br> - could return `ErrorThreshold`; |
 | past | `func NewThrottlerPast(threshold time.Time) Throttler` | Throttles each call befor timestamp defined by the specified UTC time threshold.<br> - could return `ErrorThreshold`; |
 | future | `func NewThrottlerFuture(threshold time.Time) Throttler` | Throttles each call after timestamp defined by the specified UTC time threshold.<br> - could return `ErrorThreshold`; |
 | chance | `func NewThrottlerChance(threshold float64) Throttler` | Throttles each call with the chance *p* defined by the specified threshold.<br> Chance value is normalized to *[0.0, 1.0]* range.<br> Implementation uses secure `crypto/rand` as PRNG function.<br> - could return `ErrorThreshold`; |
 | running | `func NewThrottlerRunning(threshold uint64) Throttler` | Throttles each call which exeeds the running quota *acquired - release* *q* defined by the specified threshold.<br> - could return `ErrorThreshold`; |
 | buffered | `func NewThrottlerBuffered(threshold uint64) Throttler` | Waits on call which exeeds the running quota *acquired - release* *q* defined by the specified threshold until the running quota is available again. |
 | priority | `func NewThrottlerPriority(threshold uint64, levels uint8) Throttler` | Waits on call which exeeds the running quota *acquired - release* *q* defined by the specified threshold until the running quota is available again.<br> Running quota is not equally distributed between *n* levels of priority defined by the specified levels.<br> Use `func WithPriority(ctx context.Context, priority uint8) context.Context` to override context call priority, *1* by default. |
-| timed | `func NewThrottlerTimed(threshold uint64, interval time.Duration, quantum time.Duration) Throttler` | Throttles each call which exeeds the running quota *acquired - release* *q* defined by the specified threshold in the specified interval.<br> Periodically each specified interval the running quota number is reseted.<br> If quantum is set then quantum will be used instead of interval to provide the running quota delta updates.<br> - could return `ErrorThreshold`; |
+| timed | `func NewThrottlerTimed(threshold uint64, interval time.Duration, quantum time.Duration) Throttler` | Throttles each call which exeeds the running quota *acquired - release* *q* defined by the specified threshold in the specified interval.<br> Periodically each specified interval the running quota number is reseted.<br> If quantum is set then quantum will be used instead of interval to provide the running quota delta updates.<br>Use `WithWeight` to override context call qunatity, 1 by default.<br> - could return `ErrorThreshold`; |
 | latency | `func NewThrottlerLatency(threshold time.Duration, retention time.Duration) Throttler` | Throttles each call after the call latency *l* defined by the specified threshold was exeeded once.<br> If retention is set then throttler state will be reseted after retention duration.<br> Use `func WithTimestamp(ctx context.Context, ts time.Time) context.Context` to specify running duration between throttler *acquire* and *release*.<br> - could return `ErrorThreshold`; |
 | percentile | `func NewThrottlerPercentile(threshold time.Duration, capacity uint8, percentile float64, retention time.Duration) Throttler` | Throttles each call after the call latency *l* defined by the specified threshold was exeeded once considering the specified percentile.<br> Percentile values are kept in bounded buffer with capacity *c* defined by the specified capacity. <br> If retention is set then throttler state will be reseted after retention duration.<br> Use `func WithTimestamp(ctx context.Context, ts time.Time) context.Context` to specify running duration between throttler *acquire* and *release*.<br> - could return `ErrorThreshold`; |
 | monitor | `func NewThrottlerMonitor(mnt Monitor, threshold Stats) Throttler` | Throttles call if any of the stats returned by provided monitor exceeds any of the stats defined by the specified threshold or if any internal error occurred.<br> Builtin `Monitor` implementations come with stats caching by default.<br> Use builtin `NewMonitorSystem` to create go system monitor instance.<br> - could return `ErrorInternal`;<br> - could return `ErrorThreshold`; |
 | metric | `func NewThrottlerMetric(mtc Metric) Throttler` | Throttles call if boolean metric defined by the specified boolean metric is reached or if any internal error occurred.<br> Builtin `Metric` implementations come with boolean metric caching by default.<br> Use builtin `NewMetricPrometheus` to create Prometheus metric instance.<br> - could return `ErrorInternal`;<br> - could return `ErrorThreshold`; |
 | enqueuer | `func NewThrottlerEnqueue(enq Enqueuer) Throttler` | Always enqueues message to the specified queue throttles only if any internal error occurred.<br> Use `func WithMessage(ctx context.Context, message interface{}) context.Context` to specify context message for enqueued message and `func WithMarshaler(ctx context.Context, mrsh Marshaler) context.Context` to specify context message marshaler.<br> Builtin `Enqueuer` implementations come with connection reuse and retries by default.<br> Use builtin `func NewEnqueuerRabbit(url string, queue string, retries uint64) Enqueuer` to create RabbitMQ enqueuer instance or `func NewEnqueuerKafka(net string, url string, topic string, retries uint64) Enqueuer` to create Kafka enqueuer instance.<br> - could return `ErrorInternal`; |
-| adaptive | `func NewThrottlerAdaptive(threshold uint64, interval time.Duration, quantum time.Duration, step uint64, thr Throttler) Throttler` | Throttles each call which exeeds the running quota *acquired - release* *q* defined by the specified threshold in the specified interval.<br> Periodically each specified interval the running quota number is reseted.<br> If quantum is set then quantum will be used instead of interval to provide the running quota delta updates.<br> Provided adapted throttler adjusts the running quota of adapter throttler by changing the value by *d* defined by the specified step, it subtracts *d^2* from the running quota if adapted throttler throttles or adds *d* to the running quota if it doesn't.<br> - could return `ErrorThreshold`; |
+| adaptive | `func NewThrottlerAdaptive(threshold uint64, interval time.Duration, quantum time.Duration, step uint64, thr Throttler) Throttler` | Throttles each call which exeeds the running quota *acquired - release* *q* defined by the specified threshold in the specified interval.<br> Periodically each specified interval the running quota number is reseted.<br> If quantum is set then quantum will be used instead of interval to provide the running quota delta updates.<br> Provided adapted throttler adjusts the running quota of adapter throttler by changing the value by *d* defined by the specified step, it subtracts *d^2* from the running quota if adapted throttler throttles or adds *d* to the running quota if it doesn't.<br>Use `WithWeight` to override context call qunatity, 1 by default.<br> - could return `ErrorThreshold`; |
 | pattern | `func NewThrottlerPattern(patterns ...Pattern) Throttler` | Throttles if matching throttler from provided patterns throttles.<br> Use `func WithKey(ctx context.Context, key string) context.Context` to specify key for regexp pattern throttler matching.<br> `Pattern` defines a pair of regexp and related throttler.<br> - could return `ErrorInternal`;<br> - could return any underlying throttler error; |
 | ring | `func NewThrottlerRing(thrs ...Throttler) Throttler` | Throttles if the *i-th* call throttler from provided list throttle.<br> - could return `ErrorInternal`;<br> - could return any underlying throttler error; |
 | all | `func NewThrottlerAll(thrs ...Throttler) Throttler` | Throttles call if all provided throttlers throttle.<br> - could return `ErrorInternal`; |

atomic.go

@@ -5,6 +5,13 @@ import (
 	"sync/atomic"
 )
 
+func atomicBSingAdd(number *uint64, delta int64) (res uint64) {
+	if delta > 0 {
+		return atomicBAdd(number, uint64(delta))
+	}
+	return atomicBSub(number, uint64(-delta))
+}
+
 func atomicBAdd(number *uint64, delta uint64) (res uint64) {
 	prev := atomic.LoadUint64(number)
 	if res = atomic.AddUint64(number, delta); res < prev {

context.go

@@ -50,14 +50,14 @@ func ctxPriority(ctx context.Context, limit uint8) uint8 {
 
 // WithWeight adds the provided weight to the provided context
 // to differ `Acquire` weight levels.
-// Resulted context is used by: `semaphore` and `cellrate` throtttlers.
+// Resulted context is used by: `before`, `after`, `timed`, `adaptive`, `semaphore` and `cellrate` throtttlers.
 func WithWeight(ctx context.Context, weight int64) context.Context {
 	return context.WithValue(ctx, ghctxweight, weight)
 }
 
 func ctxWeight(ctx context.Context) int64 {
 	if val := ctx.Value(ghctxweight); val != nil {
-		if weight, ok := val.(int64); ok && weight > 0 {
+		if weight, ok := val.(int64); ok {
 			return weight
 		}
 	}

throttlers.go

@@ -219,13 +219,14 @@ type tbefore struct {
 
 // NewThrottlerBefore creates new throttler instance that
 // throttles each call below the i-th call defined by the specified threshold.
+// Use `WithWeight` to override context call qunatity, 1 by default.
 // - could return `ErrorThreshold`;
 func NewThrottlerBefore(threshold uint64) Throttler {
 	return &tbefore{threshold: threshold}
 }
 
-func (thr *tbefore) Acquire(context.Context) error {
-	if current := atomicBIncr(&thr.current); current <= thr.threshold {
+func (thr *tbefore) Acquire(ctx context.Context) error {
+	if current := atomicBSingAdd(&thr.current, ctxWeight(ctx)); current <= thr.threshold {
 		return ErrorThreshold{
 			Throttler: "before",
 			Threshold: strpair{current: current, threshold: thr.threshold},
@@ -245,13 +246,14 @@ type tafter struct {
 
 // NewThrottlerAfter creates new throttler instance that
 // throttles each call after the i-th call defined by the specified threshold.
+// Use `WithWeight` to override context call qunatity, 1 by default.
 // - could return `ErrorThreshold`;
 func NewThrottlerAfter(threshold uint64) Throttler {
 	return &tafter{threshold: threshold}
 }
 
-func (thr *tafter) Acquire(context.Context) error {
-	if current := atomicBIncr(&thr.current); current > thr.threshold {
+func (thr *tafter) Acquire(ctx context.Context) error {
+	if current := atomicBSingAdd(&thr.current, ctxWeight(ctx)); current > thr.threshold {
 		return ErrorThreshold{
 			Throttler: "after",
 			Threshold: strpair{current: current, threshold: thr.threshold},
@@ -453,6 +455,7 @@ type ttimed struct {
 // q defined by the specified threshold in the specified interval.
 // Periodically each specified interval the running quota number is reseted.
 // If quantum is set then quantum will be used instead of interval to provide the running quota delta updates.
+// Use `WithWeight` to override context call qunatity, 1 by default.
 // - could return `ErrorThreshold`;
 func NewThrottlerTimed(threshold uint64, interval time.Duration, quantum time.Duration) Throttler {
 	tafter := NewThrottlerAfter(threshold).(*tafter)
@@ -725,6 +728,7 @@ type tadaptive struct {
 // Provided adapted throttler adjusts the running quota of adapter throttler by changing the value by d
 // defined by the specified step, it subtracts *d^2* from the running quota
 // if adapted throttler throttles or adds *d* to the running quota if it doesn't.
+// Use `WithWeight` to override context call qunatity, 1 by default.
 // - could return `ErrorThreshold`;
 func NewThrottlerAdaptive(
 	threshold uint64,

throttlers_test.go

@@ -258,19 +258,27 @@ func TestThrottlers(t *testing.T) {
 		},
 		"Throttler before should throttle before threshold": {
 			tms: 6,
-			thr: NewThrottlerBefore(3),
+			thr: NewThrottlerBefore(4),
+			ctxs: []context.Context{
+				context.TODO(),
+				context.TODO(),
+				context.TODO(),
+				WithWeight(context.TODO(), 2),
+				context.TODO(),
+				context.TODO(),
+			},
 			errs: []error{
 				ErrorThreshold{
 					Throttler: "before",
-					Threshold: strpair{current: 1, threshold: 3},
+					Threshold: strpair{current: 1, threshold: 4},
 				},
 				ErrorThreshold{
 					Throttler: "before",
-					Threshold: strpair{current: 2, threshold: 3},
+					Threshold: strpair{current: 2, threshold: 4},
 				},
 				ErrorThreshold{
 					Throttler: "before",
-					Threshold: strpair{current: 3, threshold: 3},
+					Threshold: strpair{current: 3, threshold: 4},
 				},
 				nil,
 				nil,
@@ -280,6 +288,13 @@ func TestThrottlers(t *testing.T) {
 		"Throttler after should throttle after threshold": {
 			tms: 6,
 			thr: NewThrottlerAfter(3),
+			ctxs: []context.Context{
+				WithWeight(context.TODO(), 3),
+				WithWeight(context.TODO(), -1),
+				context.TODO(),
+				context.TODO(),
+				context.TODO(),
+			},
 			errs: []error{
 				nil,
 				nil,
@@ -388,9 +403,9 @@ func TestThrottlers(t *testing.T) {
 			tms: 3,
 			thr: NewThrottlerRunning(1),
 			acts: []Runnable{
-				delayed(ms1_0, nope),
-				delayed(ms1_0, nope),
-				delayed(ms1_0, nope),
+				delayed(ms2_0, nope),
+				delayed(ms2_0, nope),
+				delayed(ms2_0, nope),
 			},
 			errs: []error{
 				nil,