Documentation
¶
Overview ¶
Package leakybucket provides leaky bucket rate limiting for Go applications.
A leaky bucket enforces a constant smooth output rate by "leaking" requests at a fixed interval, unlike token bucket which allows controlled bursts. This makes it ideal for traffic shaping and ensuring predictable resource consumption.
Basic usage:
limiter, err := leakybucket.NewSafe(5, 10) // 5 requests/sec leak rate, capacity 10
if err != nil {
log.Fatal(err)
}
if limiter.Allow() {
// Process request
}
Key Characteristics:
The leaky bucket algorithm provides smooth traffic flow by:
- Accepting requests up to capacity when space is available
- "Leaking" requests at a constant rate regardless of input patterns
- Buffering burst traffic and processing it at a steady rate
- Preventing downstream systems from being overwhelmed
Comparison with Token Bucket:
// Token Bucket: Allows bursts, starts with full tokens
tokenLimiter, err := bucket.NewSafe(5, 10) // Allows immediate burst of 10
if err != nil {
log.Fatal(err)
}
// Leaky Bucket: Smooth flow, starts empty
leakyLimiter, err2 := leakybucket.NewSafe(5, 10) // Builds up to capacity gradually
if err2 != nil {
log.Fatal(err2)
}
Use Cases:
Leaky bucket is ideal for:
- Video streaming and media processing
- Network traffic shaping
- Database write operations
- Any scenario requiring predictable resource usage
Token bucket is better for:
- Interactive web applications
- API rate limiting with burst tolerance
- Variable load handling
Configuration Options:
config := leakybucket.Config{
LeakRate: 10, // Requests per second
Capacity: 20, // Maximum requests to buffer
InitialLevel: 5, // Start with some requests in bucket
Clock: clock, // Custom time source (for testing)
}
limiter, err := leakybucket.NewWithConfigSafe(config)
if err != nil {
log.Fatal(err)
}
Advanced Features:
The limiter supports context-aware operations:
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := limiter.Wait(ctx); err != nil {
// Handle timeout or cancellation
}
Reservation pattern for advance booking:
reservation := limiter.Reserve()
if reservation.OK() {
delay := reservation.Delay()
// Wait for delay or cancel reservation
reservation.Cancel()
}
Dynamic configuration changes:
limiter.SetLeakRate(20) // Change processing rate limiter.SetCapacity(50) // Change buffer size
State inspection:
level := limiter.Level() // Current fill level available := limiter.Available() // Available space rate := limiter.LeakRate() // Current leak rate capacity := limiter.Capacity() // Maximum capacity
Thread Safety:
All operations are safe for concurrent use. The limiter uses mutex-based synchronization to protect internal state while maintaining good performance for high-throughput scenarios.
Example ¶
Example demonstrates basic usage of the leaky bucket rate limiter
package main
import (
"fmt"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Create a leaky bucket that leaks 5 requests per second with capacity of 10
limiter, err := leakybucket.NewSafe(5, 10)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
// Check if a request is allowed (non-blocking)
if limiter.Allow() {
fmt.Println("Request allowed")
} else {
fmt.Println("Request denied")
}
}
Output: Request allowed
Example (Comparison) ¶
Example_comparison demonstrates differences from token bucket
package main
import (
"fmt"
"github.com/1mb-dev/goflow/pkg/ratelimit/bucket"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Both limiters allow same sustained rate but different burst behavior
tokenBucket, err := bucket.NewSafe(5, 10) // 5 tokens/sec, burst of 10
if err != nil {
panic(fmt.Sprintf("Failed to create token bucket: %v", err))
}
leakyBucket, err := leakybucket.NewSafe(5, 10) // 5 leaks/sec, capacity of 10
if err != nil {
panic(fmt.Sprintf("Failed to create leaky bucket: %v", err))
}
fmt.Println("=== Token Bucket (allows bursts) ===")
fmt.Printf("Initial tokens: %.0f\n", tokenBucket.Tokens())
// Token bucket starts full - allows immediate burst
burstCount := 0
for i := 0; i < 15; i++ { // Try more than capacity
if tokenBucket.Allow() {
burstCount++
}
}
fmt.Printf("Burst requests allowed: %d\n", burstCount)
fmt.Println("\n=== Leaky Bucket (smooth flow) ===")
fmt.Printf("Initial level: %.0f\n", leakyBucket.Level())
// Leaky bucket starts empty - builds up gradually
allowedCount := 0
for i := 0; i < 15; i++ { // Try more than capacity
if leakyBucket.Allow() {
allowedCount++
}
}
fmt.Printf("Requests allowed before full: %d\n", allowedCount)
fmt.Printf("Current level: %.0f\n", leakyBucket.Level())
}
Output: === Token Bucket (allows bursts) === Initial tokens: 10 Burst requests allowed: 10 === Leaky Bucket (smooth flow) === Initial level: 0 Requests allowed before full: 10 Current level: 10
Example (Configuration) ¶
Example_configuration demonstrates advanced configuration
package main
import (
"fmt"
"time"
"github.com/1mb-dev/goflow/pkg/ratelimit/bucket"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Create with specific configuration
config := leakybucket.Config{
LeakRate: bucket.Every(200 * time.Millisecond), // 1 request every 200ms = 5/sec
Capacity: 8,
InitialLevel: 3, // Start with 3 requests in bucket
}
limiter, err := leakybucket.NewWithConfigSafe(config)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
fmt.Printf("Initial level: %.0f\n", limiter.Level())
fmt.Printf("Leak rate: %.1f/sec\n", limiter.LeakRate())
fmt.Printf("Capacity: %d\n", limiter.Capacity())
fmt.Printf("Available space: %.0f\n", limiter.Available())
}
Output: Initial level: 3 Leak rate: 5.0/sec Capacity: 8 Available space: 5
Example (DynamicConfiguration) ¶
Example_dynamicConfiguration demonstrates changing limits at runtime
package main
import (
"fmt"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
limiter, err := leakybucket.NewSafe(5, 10)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
fmt.Printf("Original rate: %.0f/sec, capacity: %d\n", limiter.LeakRate(), limiter.Capacity())
// Increase the leak rate for faster processing
limiter.SetLeakRate(15)
fmt.Printf("Updated rate: %.0f/sec, capacity: %d\n", limiter.LeakRate(), limiter.Capacity())
// Increase capacity for larger bursts
limiter.SetCapacity(20)
fmt.Printf("Final rate: %.0f/sec, capacity: %d\n", limiter.LeakRate(), limiter.Capacity())
}
Output: Original rate: 5/sec, capacity: 10 Updated rate: 15/sec, capacity: 10 Final rate: 15/sec, capacity: 20
Example (MultipleRequests) ¶
Example_multipleRequests demonstrates handling multiple requests at once
package main
import (
"fmt"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Create a leaky bucket (10 requests per second, capacity 20)
limiter, err := leakybucket.NewSafe(10, 20)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
// Try to handle 5 requests at once
if limiter.AllowN(5) {
fmt.Println("Batch operation allowed (5 requests)")
}
// Check current state
fmt.Printf("Current level: %.0f\n", limiter.Level())
fmt.Printf("Available space: %.0f\n", limiter.Available())
}
Output: Batch operation allowed (5 requests) Current level: 5 Available space: 15
Example (Reservation) ¶
Example_reservation demonstrates the reservation pattern
package main
import (
"fmt"
"time"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Create a leaky bucket (3 requests per second, capacity 5)
limiter, err := leakybucket.NewSafe(3, 5)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
// Fill bucket to capacity
for i := 0; i < 5; i++ {
limiter.Allow()
}
// Make a reservation for the next request
reservation := limiter.Reserve()
if reservation.OK() {
delay := reservation.Delay()
fmt.Printf("Need to wait %v before next request\n", delay.Round(time.Millisecond*100))
// Cancel the reservation if we don't want to wait
reservation.Cancel()
fmt.Println("Reservation canceled")
}
}
Output: Need to wait 300ms before next request Reservation canceled
Example (TrafficShaping) ¶
Example_trafficShaping demonstrates smooth traffic flow characteristics
package main
import (
"fmt"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Create a leaky bucket for smooth traffic shaping (2 requests/sec, capacity 4)
limiter, err := leakybucket.NewSafe(2, 4)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
fmt.Printf("Initial level: %.0f/%.0f\n", limiter.Level(), float64(limiter.Capacity()))
// Send burst of requests
for i := 0; i < 4; i++ {
limiter.Allow()
}
fmt.Printf("After burst: %.0f/%.0f\n", limiter.Level(), float64(limiter.Capacity()))
fmt.Printf("Available space: %.0f\n", limiter.Available())
}
Output: Initial level: 0/4 After burst: 4/4 Available space: 0
Example (Wait) ¶
Example_wait demonstrates blocking until space is available
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/1mb-dev/goflow/pkg/ratelimit/leakybucket"
)
func main() {
// Create a slow leaky bucket (1 request per second, capacity 1)
limiter, err := leakybucket.NewSafe(1, 1)
if err != nil {
panic(fmt.Sprintf("Failed to create limiter: %v", err))
}
ctx := context.Background()
// First request succeeds immediately
if err := limiter.Wait(ctx); err != nil {
log.Fatal(err)
}
fmt.Println("First request processed")
// Second request would need to wait, but we'll use a timeout
ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
defer cancel()
if err := limiter.Wait(ctx); err != nil {
fmt.Printf("Second request failed: %v\n", err)
}
}
Output: First request processed Second request failed: context deadline exceeded
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
// LeakRate is the rate at which requests leak from the bucket (requests per second).
LeakRate bucket.Limit
// Capacity is the maximum number of requests the bucket can hold.
Capacity int
// Clock provides the current time. If nil, SystemClock is used.
Clock bucket.Clock
// InitialLevel is the initial fill level of the bucket.
// If negative, starts empty (0).
InitialLevel int
}
Config holds configuration options for creating a new Limiter.
type Limiter ¶
type Limiter interface {
// Allow reports whether an event may happen now. It does not block.
Allow() bool
// AllowN reports whether n events may happen now. It does not block.
AllowN(n int) bool
// Wait blocks until an event can happen. It returns an error
// if the context is canceled or the deadline is exceeded.
Wait(ctx context.Context) error
// WaitN blocks until n events can happen. It returns an error
// if the context is canceled or the deadline is exceeded.
WaitN(ctx context.Context, n int) error
// Reserve returns a Reservation that indicates how long the caller
// must wait before an event can happen.
Reserve() *Reservation
// ReserveN returns a Reservation that indicates how long the caller
// must wait before n events can happen.
ReserveN(n int) *Reservation
// SetLeakRate changes the leak rate. It preserves the current capacity.
SetLeakRate(rate bucket.Limit)
// SetCapacity changes the bucket capacity. It preserves the current leak rate.
SetCapacity(capacity int)
// LeakRate returns the current leak rate.
LeakRate() bucket.Limit
// Capacity returns the current bucket capacity.
Capacity() int
// Level returns the current fill level of the bucket.
Level() float64
// Available returns the available space in the bucket.
Available() float64
}
Limiter controls the rate at which events are allowed to happen using a leaky bucket algorithm. Unlike token bucket, it enforces a constant smooth output rate by "leaking" requests at a fixed interval.
func New
deprecated
New creates a new leaky bucket rate limiter with the specified leak rate and capacity. The limiter starts empty.
Deprecated: Use NewSafe instead. This function panics on invalid input and will be removed in v2.0.0. NewSafe provides the same functionality with proper error handling.
func NewSafe ¶
NewSafe creates a new leaky bucket rate limiter with validation that returns an error instead of panicking. This is the recommended way to create leaky bucket limiters for production use.
func NewWithConfig
deprecated
NewWithConfig creates a new leaky bucket rate limiter with the specified configuration.
Deprecated: Use NewWithConfigSafe instead. This function panics on invalid input and will be removed in v2.0.0. NewWithConfigSafe provides the same functionality with proper error handling.
func NewWithConfigSafe ¶
NewWithConfigSafe creates a new leaky bucket rate limiter with validation that returns an error instead of panicking. This is the recommended way to create leaky bucket limiters for production use.
type Reservation ¶
type Reservation struct {
// contains filtered or unexported fields
}
Reservation holds information about an event that should happen in the future. It can be used to cancel the reservation or check when the event should occur.
func (*Reservation) Cancel ¶
func (r *Reservation) Cancel()
Cancel cancels the reservation. This removes the reserved level from the bucket if the reservation was valid.
func (*Reservation) Delay ¶
func (r *Reservation) Delay() time.Duration
Delay returns the time until the reservation should act. If the reservation is not OK, Delay returns zero.
func (*Reservation) DelayFrom ¶
func (r *Reservation) DelayFrom(now time.Time) time.Duration
DelayFrom returns the time until the reservation should act, measured from the given time.
func (*Reservation) OK ¶
func (r *Reservation) OK() bool
OK returns whether the reservation is valid.
Source Files