Asynchronous rate-limiting

pipeline.go

package main

import (
	"sync/atomic"
	"time"
	"fmt"
)

var (
	counter int32 = 0
	workCount = 900
	rate = 300
	interval = time.Millisecond * 1000
	sleepInterval = time.Millisecond * 500
	reportInterval = time.Second * 1
	iterationsPerReport = int(reportInterval / interval)
)

func main(){
	limiter := make(chan time.Time)
	requester := make(chan int)

	go func(){
		for {
			<-requester
			<-limiter

			go func(){
				time.Sleep(sleepInterval)
				atomic.AddInt32(&counter, 1)
			}()
		}
	}()

	go func(){
		for i := 0; i < workCount; i++ {
			time.Sleep(1 * time.Millisecond)
			requester <- i
		}
	}()

	go func(){
		for t := range time.Tick(interval / time.Duration(rate)){
			limiter <- t
		}
	}()

	// report progress
	ticker := time.NewTicker(reportInterval)
	iterations := 0

	for {
		last := int(atomic.LoadInt32(&counter))

		<- ticker.C
		iterations++

		now := int(atomic.LoadInt32(&counter))
		diff := now - last

		// break for the concurrency visualizer
		if diff == 0 && iterations > 2 {
			return
		}

		validity := (last <= (iterations * iterationsPerReport * rate)) && (diff <= rate)


		fmt.Printf("#%v || counter: %v || validity: %v || diff: %v\n", iterations, now, validity, diff)
	}
}