README
¶
concurrency-limiter
About
concurrency-limiter allows you to limit the number of goroutines accessing a resource with support for timeouts , dynamic priority of goroutines and context cancellation of goroutines.
The library now supports two admission modes:
- strict admission via
Wait/Run - soft admission via
WaitOrBypass/RunOrBypass
Installation
To install concurrency-limiter:
go get github.com/vivek-ng/concurrency-limiter
Then import concurrency-limiter to use it
import(
"time"
github.com/vivek-ng/concurrency-limiter/limiter
)
nl := limiter.New(3)
ctx := context.Background()
if err := nl.Wait(ctx); err != nil {
return
}
// Perform actions .........
nl.Finish()
Usage
Below are some examples of using this library. To run real examples , please check the examples folder.
Limiter
import(
"time"
github.com/vivek-ng/concurrency-limiter/limiter
)
func main() {
nl := limiter.New(3)
var wg sync.WaitGroup
wg.Add(15)
for i := 0; i < 15; i++ {
go func(index int) {
defer wg.Done()
ctx := context.Background()
if err := nl.Wait(ctx); err != nil {
return
}
// in real life , this can be DB operations , message publish to queue ........
fmt.Println("executing action...: ", "index: ", index, "current number of goroutines: ", nl.Count())
nl.Finish()
}(i)
}
wg.Wait()
}
Priority Limiter
import(
"time"
github.com/vivek-ng/concurrency-limiter/priority
)
func main() {
pr := priority.NewLimiter(1)
var wg sync.WaitGroup
wg.Add(15)
for i := 0; i < 15; i++ {
go func(index int) {
defer wg.Done()
ctx := context.Background()
if index%2 == 1 {
if err := pr.Wait(ctx, priority.High); err != nil {
return
}
} else {
if err := pr.Wait(ctx, priority.Low); err != nil {
return
}
}
// in real life , this can be DB operations , message publish to queue ........
fmt.Println("executing action...: ", "index: ", index, "current number of goroutines: ", pr.Count())
pr.Finish()
}(i)
}
wg.Wait()
}
Examples
Limiter
nl := limiter.New(3)
ctx := context.Background()
if err := nl.Wait(ctx); err != nil {
return
}
// Perform actions .........
nl.Finish()
In the above example , there can be a maximum of 3 goroutines accessing a resource concurrently. The other goroutines are added to the waiting list and are removed and given a
chance to access the resource in the FIFO order. If the context is cancelled , the goroutine is removed from the waitlist and Wait returns the context error.
Limiter with Timeout
nl := limiter.New(3,
WithTimeoutDuration(10 * time.Millisecond),
)
ctx := context.Background()
if err := nl.Wait(ctx); err != nil {
return
}
// Perform actions .........
nl.Finish()
In the above example , the goroutines will wait for a maximum of 10 milliseconds. Goroutines will be removed from the waitlist after 10 ms and Wait will return limiter.ErrTimeout.
Limiter with Soft Timeout
nl := limiter.New(3,
WithTimeoutDuration(10 * time.Millisecond),
)
ctx := context.Background()
result, err := nl.WaitOrBypass(ctx)
if err != nil {
return
}
if result == limiter.AdmissionAcquired {
defer nl.Finish()
}
// Perform actions .........
In the above example , the goroutine waits for up to 10 milliseconds to acquire real capacity. If capacity is still unavailable after the timeout, WaitOrBypass returns limiter.AdmissionBypassed and the caller can still proceed without consuming limiter capacity.
Priority Limiter
nl := priority.NewLimiter(3)
ctx := context.Background()
if err := nl.Wait(ctx , priority.High); err != nil {
return
}
// Perform actions .........
nl.Finish()
In Priority Limiter , goroutines with higher priority will be given preference to be removed from the waitlist. For instance in the above example , the goroutine will be given the maximum preference because it is of high priority. In the case of tie between the priorities , the goroutines will be removed from the waitlist in the FIFO order.
Priority Limiter with Dynamic priority
nl := priority.NewLimiter(3,
WithDynamicPriorityDuration(5 * time.Millisecond),
)
ctx := context.Background()
if err := nl.Wait(ctx , priority.Low); err != nil {
return
}
// Perform actions .........
nl.Finish()
In Dynamic Priority Limiter , the goroutines with lower priority will get their priority increased periodically by the time period specified. For instance in the above example , the goroutine will get it's priority increased every 5 ms. This will ensure that goroutines with lower priority do not suffer from starvation. It's highly recommended to use Dynamic Priority Limiter to avoid starving low priority goroutines.
Priority Limiter with Timeout
nl := priority.NewLimiter(3,
WithTimeoutDuration(30 * time.Millisecond),
WithDynamicPriorityDuration(5 * time.Millisecond),
)
ctx := context.Background()
if err := nl.Wait(ctx , priority.Low); err != nil {
return
}
// Perform actions .........
nl.Finish()
This is similar to the timeouts in the normal limiter. In the above example , goroutines will wait a maximum of 30 milliseconds and Wait will return limiter.ErrTimeout if capacity is not acquired in time. The low priority goroutines will get their
priority increased every 5 ms.
Priority Limiter with Soft Timeout
nl := priority.NewLimiter(3,
WithTimeoutDuration(30 * time.Millisecond),
WithDynamicPriorityDuration(5 * time.Millisecond),
)
ctx := context.Background()
result, err := nl.WaitOrBypass(ctx , priority.Low)
if err != nil {
return
}
if result == limiter.AdmissionAcquired {
defer nl.Finish()
}
// Perform actions .........
This uses the same soft-admission model for the priority limiter. The goroutine waits up to the configured timeout while still participating in the priority queue. If capacity is not acquired in time, the call returns limiter.AdmissionBypassed and the caller may continue outside the limiter.
Runnable Function
nl := priority.NewLimiter(3)
ctx := context.Background()
nl.Run(ctx , priority.Low , func()error {
return sendMetrics()
})
Runnable function will allow you to wrap your function and execute them with concurrency limit. This function is a wrapper on top of the Wait() and Finish() functions.
If Wait fails because the context is cancelled or the timeout expires, Run returns that error and does not execute the callback.
Runnable Function with Soft Timeout
nl := priority.NewLimiter(3,
WithTimeoutDuration(30 * time.Millisecond),
)
ctx := context.Background()
result, err := nl.RunOrBypass(ctx , priority.Low , func()error {
return sendMetrics()
})
if err != nil {
return
}
_ = result
RunOrBypass executes the callback after either real admission or timeout bypass. It returns limiter.AdmissionAcquired when the limiter granted capacity and limiter.AdmissionBypassed when the callback ran outside the limiter after the timeout.
The older WithTimeout(int) and WithDynamicPriority(int) helpers are still supported for compatibility and continue to interpret their arguments as milliseconds.
Contribution
Please feel free to open up issues , create PRs for bugs/features. All contributions are welcome :)
Documentation
¶
Index ¶
- Variables
- func WithTimeout(timeout int) func(*Limiter)
- func WithTimeoutDuration(timeout time.Duration) func(*Limiter)
- type AdmissionResult
- type Limiter
- func (l *Limiter) Count() int
- func (l *Limiter) Finish()
- func (l *Limiter) Run(ctx context.Context, callback func() error) error
- func (l *Limiter) RunOrBypass(ctx context.Context, callback func() error) (AdmissionResult, error)
- func (l *Limiter) Wait(ctx context.Context) error
- func (l *Limiter) WaitOrBypass(ctx context.Context) (AdmissionResult, error)
- type Option
Constants ¶
This section is empty.
Variables ¶
var ErrTimeout = errors.New("limiter: timed out waiting for capacity")
Functions ¶
func WithTimeout ¶ added in v0.5.0
WithTimeout configures timeout in milliseconds. Deprecated: use WithTimeoutDuration.
func WithTimeoutDuration ¶ added in v1.3.0
WithTimeoutDuration configures timeout as a time.Duration. Goroutines are removed from the waitlist after the timeout and Wait returns ErrTimeout.
Types ¶
type AdmissionResult ¶ added in v1.1.0
type AdmissionResult int
AdmissionResult describes how a caller proceeded through the limiter.
const ( // AdmissionAcquired means the caller acquired real limiter capacity. AdmissionAcquired AdmissionResult = iota + 1 // AdmissionBypassed means the caller proceeded without consuming limiter capacity. AdmissionBypassed )
type Limiter ¶
type Limiter struct {
// Deprecated: configure via New. Runtime behavior uses an internal snapshot.
Limit int
// Deprecated: configure via WithTimeoutDuration. Runtime behavior uses an internal snapshot.
Timeout *int
// contains filtered or unexported fields
}
Limiter stores the configuration need for concurrency limiter....
func New ¶ added in v0.4.2
New creates an instance of *Limiter. Configure the Limiter with the options specified. Example: limiter.New(4, WithTimeoutDuration(5*time.Millisecond))
func (*Limiter) Count ¶ added in v0.8.0
Count returns the current number of concurrent gouroutines executing...
func (*Limiter) Finish ¶
func (l *Limiter) Finish()
Finish will remove the goroutine from the waiting list and sends a signal to the waiting goroutine to access the resource
func (*Limiter) RunOrBypass ¶ added in v1.1.0
RunOrBypass executes the callback after real acquisition or bypass after timeout. Finish is only called when capacity was actually acquired.
func (*Limiter) Wait ¶
Wait waits until capacity is available or the context/timeout expires. It returns nil only when the caller successfully acquires capacity.
func (*Limiter) WaitOrBypass ¶ added in v1.1.0
func (l *Limiter) WaitOrBypass(ctx context.Context) (AdmissionResult, error)
WaitOrBypass waits until capacity is available, or bypasses the limiter after the configured timeout. It returns AdmissionBypassed only when a timeout occurs before capacity is acquired.
Source Files
Directories