feat: Batch StartAPI

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 Ankur Anand
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1 +1,182 @@
 ## taskwheel
+
+[![Go Reference](https://pkg.go.dev/badge/github.com/ankur-anand/taskwheel.svg)](https://pkg.go.dev/github.com/ankur-anand/taskwheel)
+[![Go Report Card](https://goreportcard.com/badge/github.com/ankur-anand/taskwheel)](https://goreportcard.com/report/github.com/ankur-anand/taskwheel)
+
+A high-performance, generic **Hierarchical Timing Wheel** implementation in Go for efficient timer management at scale.
+
+### Installation
+
+```bash
+go get github.com/ankur-anand/taskwheel
+```
+
+### Quick Start
+
+```go
+package main
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/ankur-anand/taskwheel"
+)
+
+func main() {
+	// Create hierarchical timing wheel
+	intervals := []time.Duration{10 * time.Millisecond, 1 * time.Second}
+	slots := []int{100, 60}
+	wheel := taskwheel.NewHierarchicalTimingWheel[string](intervals, slots)
+
+	// Start the wheel
+	stop := wheel.Start(10*time.Millisecond, func(timer *taskwheel.Timer[string]) {
+		fmt.Printf("Timer fired: %s\n", timer.Value)
+	})
+	defer stop()
+
+	// Schedule timers
+	wheel.AfterTimeout("task1", "Process payment", 100*time.Millisecond)
+	wheel.AfterTimeout("task2", "Send email", 500*time.Millisecond)
+
+	time.Sleep(1 * time.Second)
+}
+```
+
+### High-Throughput Usage (10,000+ timers/sec)
+
+For production systems with high timer volumes, use `StartBatch()` with a worker pool:
+
+```go
+package main
+
+import (
+	"fmt"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/ankur-anand/taskwheel"
+)
+
+func main() {
+	intervals := []time.Duration{10 * time.Millisecond, 100 * time.Millisecond, time.Second}
+	slots := []int{10, 100, 60}
+	wheel := taskwheel.NewHierarchicalTimingWheel[string](intervals, slots)
+
+	// Create worker pool
+	workerPool := make(chan *taskwheel.Timer[string], 1000)
+	var wg sync.WaitGroup
+
+	numWorkers := runtime.NumCPU() * 2
+	for i := 0; i < numWorkers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for timer := range workerPool {
+				// process timer
+				processTask(timer)
+			}
+		}()
+	}
+
+	// start with batch callback
+	stop := wheel.StartBatch(10*time.Millisecond, func(timers []*taskwheel.Timer[string]) {
+		for _, t := range timers {
+			workerPool <- t
+		}
+	})
+
+	// schedule timers
+	for i := 0; i < 10000; i++ {
+		wheel.AfterTimeout(
+			taskwheel.TimerID(fmt.Sprintf("task-%d", i)),
+			fmt.Sprintf("Task %d", i),
+			time.Duration(i)*time.Millisecond,
+		)
+	}
+
+	time.Sleep(15 * time.Second)
+	stop()
+	close(workerPool)
+	wg.Wait()
+}
+
+func processTask(timer *taskwheel.Timer[string]) {
+	// business logic here
+}
+```
+
+### Performance Comparison
+
+```bash
+goos: darwin
+goarch: arm64
+pkg: github.com/ankur-anand/taskwheel
+cpu: Apple M2 Pro
+BenchmarkNativeTimers/1K_timers_100ms-10         	      10	 102039642 ns/op	  191393 B/op	    2091 allocs/op
+BenchmarkNativeTimers/10K_timers_100ms-10        	       9	 114114778 ns/op	 1820984 B/op	   20260 allocs/op
+BenchmarkNativeTimers/100K_timers_1s-10          	       1	1090769709 ns/op	39694704 B/op	  240698 allocs/op
+BenchmarkTimingWheelAfterTimeout/1K_timers_100ms-10         	      10	 110101579 ns/op	  315608 B/op	    1056 allocs/op
+BenchmarkTimingWheelAfterTimeout/10K_timers_100ms-10        	       9	 111119176 ns/op	 2857496 B/op	   10181 allocs/op
+BenchmarkTimingWheelAfterTimeout/100K_timers_100ms-10       	       9	 122693727 ns/op	26476164 B/op	  101094 allocs/op
+BenchmarkMemoryComparison/Native_10K_timers-10              	      10	 111232592 ns/op	 1429328 B/op	   20003 allocs/op
+BenchmarkMemoryComparison/TimingWheel_10K_timers-10         	      10	 110203346 ns/op	 2857505 B/op	   10181 allocs/op
+```
+
+| Workload              | Metric     | NativeTimers | TimingWheel | Difference |
+|-----------------------|------------|--------------|-------------|------------|
+| **1K timers (100ms)** | Time/op    | 102 ms       | 110 ms      | +8% slower |
+|                       | Mem/op     | 191 KB       | 316 KB      | +65% more  |
+|                       | Allocs/op  | 2.1 K        | 1.1 K       | -50% fewer |
+| **10K timers (100ms)**| Time/op    | 114 ms       | 111 ms      | -3% faster |
+|                       | Mem/op     | 1.8 MB       | 2.9 MB      | +57% more  |
+|                       | Allocs/op  | 20.3 K       | 10.2 K      | -50% fewer |
+| **100K timers (1s)**  | Time/op    | 1.09 s       | 0.12 s      | -89% faster |
+|                       | Mem/op     | 39.7 MB      | 26.5 MB     | -33% less  |
+|                       | Allocs/op  | 240.7 K      | 101.1 K     | -58% fewer |
+
+## Advanced Usage
+
+### Custom Payload Types
+
+```go
+type Task struct {
+    UserID   string
+    Action   string
+    Priority int
+}
+
+wheel := taskwheel.NewHierarchicalTimingWheel[Task](intervals, slots)
+wheel.AfterTimeout("task1", Task{
+    UserID:   "user123",
+    Action:   "send_email",
+    Priority: 1,
+}, 5*time.Second)
+```
+
+### Priority-Based Processing
+
+```go
+stop := wheel.StartBatch(10*time.Millisecond, func(timers []*taskwheel.Timer[Task]) {
+    // Sort by priority
+    sort.Slice(timers, func(i, j int) bool {
+        return timers[i].Value.Priority > timers[j].Value.Priority
+    })
+
+    for _, t := range timers {
+        processTask(t)
+    }
+})
+```
+
+## License
+
+MIT License - see [LICENSE](LICENSE) file for details
+
+## Credits
+
+Inspired by:
+- [Kafka's Hierarchical Timing Wheels](https://www.confluent.io/blog/apache-kafka-purgatory-hierarchical-timing-wheels/)
+- ["Hashed and Hierarchical Timing Wheels" paper](http://www.cs.columbia.edu/~nahum/w6998/papers/ton97-timing-wheels.pdf)
+

hierarchical_timewheel.go

@@ -174,6 +174,7 @@ func (htw *HierarchicalTimingWheel[T]) calcPlacement(timeout time.Duration, now
 
 // Start begins ticking the hierarchical timing wheel at the specified interval in a new goroutine.
 // If wheel is paused Tick() is a no-op and no timers will be fired until resumed.
+// The callback is executed synchronously, care should be taken to not block the onTimer Callback.
 func (htw *HierarchicalTimingWheel[T]) Start(tickInterval time.Duration, onTimer func(*Timer[T])) (stop func()) {
 	stopCh := make(chan struct{})
 	go func() {
@@ -184,7 +185,31 @@ func (htw *HierarchicalTimingWheel[T]) Start(tickInterval time.Duration, onTimer
 			case <-ticker.C:
 				due := htw.Tick()
 				for _, t := range due {
-					go onTimer(t)
+					onTimer(t)
+				}
+			case <-stopCh:
+				return
+			}
+		}
+	}()
+	return func() { close(stopCh) }
+}
+
+// StartBatch begins ticking the hierarchical timing wheel at the specified interval in a new goroutine.
+// If wheel is paused Tick() is a no-op and no timers will be fired until resumed.
+// All timers that are due are passed as a batch to a onTimerBatch callback.
+// The callback is executed synchronously, care should be taken to not block the onTimerBatch Callback.
+func (htw *HierarchicalTimingWheel[T]) StartBatch(tickInterval time.Duration, onTimerBatch func([]*Timer[T])) (stop func()) {
+	stopCh := make(chan struct{})
+	go func() {
+		ticker := time.NewTicker(tickInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				due := htw.Tick()
+				if len(due) > 0 {
+					onTimerBatch(due)
 				}
 			case <-stopCh:
 				return

hierarchical_timewheel_test.go

@@ -2,6 +2,7 @@ package taskwheel
 
 import (
 	"fmt"
+	"reflect"
 	"sync"
 	"testing"
 	"time"
@@ -26,6 +27,35 @@ func ExampleHierarchicalTimingWheel() {
 	// Timer long fired
 }
 
+func ExampleHierarchicalTimingWheel_StartBatch() {
+	intervals := []time.Duration{10 * time.Millisecond, 1 * time.Second}
+	slots := []int{100, 60}
+	wheel := NewHierarchicalTimingWheel[string](intervals, slots)
+
+	stop := wheel.StartBatch(10*time.Millisecond, func(timers []*Timer[string]) {
+		fmt.Printf("Batch of %d timers fired\n", len(timers))
+
+		for _, timer := range timers {
+			fmt.Printf("- Timer %s fired\n", timer.Value)
+		}
+
+		// or use can use pool or any other pattern
+		//for _, timer := range timers {
+		//	workerpool <- timer
+		//}
+	})
+
+	defer stop()
+
+	_, _ = wheel.AfterTimeout("a", "short", 45*time.Millisecond)
+	_, _ = wheel.AfterTimeout("b", "long", 50*time.Millisecond)
+	time.Sleep(3 * time.Second)
+	// Output:
+	// Batch of 2 timers fired
+	// - Timer short fired
+	// - Timer long fired
+}
+
 func TestHierarchicalTimingWheel_FiresTimersCorrectly(t *testing.T) {
 	intervals := []time.Duration{10 * time.Millisecond, 1 * time.Second, 1 * time.Minute}
 	slots := []int{100, 60, 60}
@@ -316,3 +346,44 @@ func TestHierarchicalTimingWheel_Len(t *testing.T) {
 		t.Fatalf("Expected 0 timers after reset")
 	}
 }
+
+func TestHierarchicalTimingWheel_StartBatch(t *testing.T) {
+	intervals := []time.Duration{10 * time.Millisecond, 1 * time.Second}
+	slots := []int{100, 60}
+	wheel := NewHierarchicalTimingWheel[string](intervals, slots)
+
+	var mu sync.Mutex
+	var batches []int
+	var allFired []string
+
+	stop := wheel.StartBatch(10*time.Millisecond, func(timers []*Timer[string]) {
+		mu.Lock()
+		batches = append(batches, len(timers))
+		for _, t := range timers {
+			allFired = append(allFired, t.Value)
+		}
+		mu.Unlock()
+	})
+
+	defer stop()
+	_, _ = wheel.AfterTimeout("a", "A", 48*time.Millisecond)
+	_, _ = wheel.AfterTimeout("b", "B", 45*time.Millisecond)
+	_, _ = wheel.AfterTimeout("c", "C", 100*time.Millisecond)
+
+	time.Sleep(300 * time.Millisecond)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	if len(batches) != 2 {
+		t.Fatalf("Expected 2 timer batchess, got %d", len(batches))
+	}
+
+	if len(allFired) != 3 {
+		t.Fatalf("Expected 3 timer, got %d", len(allFired))
+	}
+
+	if !reflect.DeepEqual(allFired, []string{"A", "B", "C"}) {
+		t.Fatalf("Expected all fired timer, got %v", allFired)
+	}
+}