Skip to content

Commit 66246e2

Browse files
freemans13freemans13
authored
Idle batcher 0% CPU (#46)
* Idle batcher 0% CPU * replace struct context with a close channel * explicitly handle error in hash()
1 parent 1b84f11 commit 66246e2

File tree

2 files changed

+152
-39
lines changed

2 files changed

+152
-39
lines changed

‎batcher.go‎

Lines changed: 89 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,14 @@
1010
// - Background processing option to avoid blocking the caller
1111
// - Manual trigger capability for immediate batch processing
1212
// - Thread-safe concurrent operations
13+
// - Zero idle CPU usage: Lazy timer activation only when items are batched
14+
// - Graceful shutdown support with Close() for clean goroutine lifecycle
15+
//
16+
// Performance characteristics:
17+
// - Idle state: 0% CPU usage (no timers running when batch is empty)
18+
// - Active state: Full performance with low-latency batching (configurable timeout)
19+
// - Peak performance: Maintains throughput regardless of load
20+
// - Memory efficient: Optional slice pooling to reduce GC pressure
1321
//
1422
// The package is structured to provide two main components:
1523
// - Basic Batcher: Simple batching with size and timeout-based triggers
@@ -22,15 +30,17 @@
2230
// db.BulkInsert(batch)
2331
// }, true)
2432
// batcher.Put(&User{Name: "John"})
33+
// defer batcher.Close() // Graceful shutdown
2534
//
2635
// Important notes:
27-
// - The batcher runs a background goroutine that continues indefinitely
36+
// - The batcher runs a background goroutine managed via context
2837
// - Items are passed by pointer to avoid unnecessary copying
2938
// - The processing function is called synchronously or asynchronously based on the background flag
3039
// - Batches are processed when size is reached, timeout expires, or Trigger() is called
40+
// - Call Close() for graceful shutdown to process remaining items and prevent goroutine leaks
3141
//
3242
// This package is part of the go-batcher library and provides efficient batch processing
33-
// capabilities for high-throughput applications.
43+
// capabilities for high-throughput applications with minimal resource consumption.
3444
package batcher
3545

3646
import (
@@ -73,6 +83,7 @@ type Batcher[T any] struct {
7383
background bool
7484
usePool bool
7585
pool *sync.Pool
86+
done chan struct{}
7687
}
7788

7889
// New creates a new Batcher instance with the specified configuration.
@@ -110,6 +121,7 @@ func New[T any](size int, timeout time.Duration, fn func(batch []*T), background
110121
triggerCh: make(chan struct{}),
111122
background: background,
112123
usePool: false,
124+
done: make(chan struct{}),
113125
}
114126

115127
go b.worker()
@@ -147,6 +159,7 @@ func NewWithPool[T any](size int, timeout time.Duration, fn func(batch []*T), ba
147159
return &slice
148160
},
149161
},
162+
done: make(chan struct{}),
150163
}
151164

152165
go b.worker()
@@ -211,6 +224,36 @@ func (b *Batcher[T]) Trigger() {
211224
b.triggerCh <- struct{}{}
212225
}
213226

227+
// Close gracefully shuts down the batcher, allowing pending items to be processed.
228+
//
229+
// This method signals the worker goroutine to stop accepting new items and process
230+
// any remaining items in the queue before exiting. It provides a clean shutdown
231+
// mechanism that prevents goroutine leaks and ensures all queued items are flushed.
232+
//
233+
// Parameters:
234+
// - None
235+
//
236+
// Returns:
237+
// - Nothing
238+
//
239+
// Side Effects:
240+
// - Cancels the internal context, signaling the worker to begin shutdown
241+
// - The worker will process all items currently in the channel
242+
// - The worker will flush any partial batch before exiting
243+
// - The internal channel is closed after draining, preventing further Put() calls
244+
//
245+
// Notes:
246+
// - This method returns immediately without waiting for shutdown to complete
247+
// - It's safe to call Close() multiple times (subsequent calls have no effect)
248+
// - Items already in the channel will be processed during shutdown
249+
//
250+
// IMPORTANT: Do not call Put() after Close() has been called. The channel is closed
251+
// during shutdown, and any Put() calls after Close() will panic with "send on closed channel".
252+
// Users must ensure proper synchronization to prevent Put() calls after Close().
253+
func (b *Batcher[T]) Close() {
254+
close(b.done)
255+
}
256+
214257
// worker is the core processing loop that manages batch aggregation and processing.
215258
//
216259
// This function runs as a background goroutine and continuously monitors three conditions
@@ -246,37 +289,61 @@ func (b *Batcher[T]) Trigger() {
246289
// - Reuses timers to reduce allocations and GC pressure
247290
// - When usePool=true, manages slice lifecycle through sync.Pool for memory efficiency
248291
func (b *Batcher[T]) worker() { //nolint:gocognit,gocyclo // Worker function handles multiple channels and conditions
249-
// Create a reusable timer for optimization
250-
timer := time.NewTimer(b.timeout)
251-
defer timer.Stop()
292+
var timer *time.Timer
293+
var timerCh <-chan time.Time // nil channel blocks forever, enabling lazy timer activation
252294

253-
for {
254-
// Reset the timer for this batch cycle
255-
if !timer.Stop() {
256-
select {
257-
case <-timer.C:
258-
default:
259-
}
295+
defer func() {
296+
if timer != nil {
297+
timer.Stop()
260298
}
261-
timer.Reset(b.timeout)
299+
}()
262300

263-
for {
264-
select {
265-
case item := <-b.ch:
301+
for {
302+
select {
303+
case <-b.done:
304+
// Shutdown: drain channel and process remaining items
305+
close(b.ch)
306+
for item := range b.ch {
266307
b.batch = append(b.batch, item)
308+
}
309+
// Flush final batch if any
310+
if len(b.batch) > 0 {
311+
b.fn(b.batch)
312+
}
313+
return
267314

268-
if len(b.batch) == b.size {
269-
goto saveBatch
270-
}
315+
case item := <-b.ch:
316+
b.batch = append(b.batch, item)
271317

272-
case <-timer.C:
273-
goto saveBatch
318+
// Start timer on first item (lazy timer activation)
319+
if len(b.batch) == 1 {
320+
timer = time.NewTimer(b.timeout)
321+
timerCh = timer.C
322+
}
274323

275-
case <-b.triggerCh:
324+
// Flush if size limit reached
325+
if len(b.batch) == b.size {
326+
if timer != nil {
327+
timer.Stop()
328+
timerCh = nil
329+
}
276330
goto saveBatch
277331
}
332+
333+
case <-timerCh: // Only fires when timerCh != nil (batch has items)
334+
timerCh = nil // Disable timer after firing
335+
goto saveBatch
336+
337+
case <-b.triggerCh:
338+
if timer != nil {
339+
timer.Stop()
340+
timerCh = nil
341+
}
342+
goto saveBatch
278343
}
279344

345+
continue
346+
280347
saveBatch:
281348
if len(b.batch) > 0 { //nolint:nestif // Necessary complexity for handling pooling and background modes
282349
batch := b.batch

‎batcher_deduplication.go‎

Lines changed: 63 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -74,65 +74,98 @@ func (bf *BloomFilter) Reset() {
7474
}
7575

7676
// hash generates multiple hash values for the given key.
77+
// Note: hash.Hash.Write() never returns an error for FNV hash, but we check defensively.
7778
func (bf *BloomFilter) hash(key interface{}) []uint64 { //nolint:gocyclo // Type switch for performance
7879
h := fnv.New64a()
7980
// Convert key to bytes for hashing - fast paths for common types
8081
switch k := key.(type) {
8182
case string:
82-
_, _ = h.Write([]byte(k))
83+
if _, err := h.Write([]byte(k)); err != nil {
84+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
85+
}
8386
case int:
8487
var buf [8]byte
8588
binary.BigEndian.PutUint64(buf[:], uint64(k)) //nolint:gosec // Safe conversion for hashing
86-
_, _ = h.Write(buf[:])
89+
if _, err := h.Write(buf[:]); err != nil {
90+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
91+
}
8792
case int8:
88-
_, _ = h.Write([]byte{byte(k)})
93+
if _, err := h.Write([]byte{byte(k)}); err != nil {
94+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
95+
}
8996
case int16:
9097
var buf [2]byte
9198
binary.BigEndian.PutUint16(buf[:], uint16(k)) //nolint:gosec // Safe conversion for hashing
92-
_, _ = h.Write(buf[:])
99+
if _, err := h.Write(buf[:]); err != nil {
100+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
101+
}
93102
case int32:
94103
var buf [4]byte
95104
binary.BigEndian.PutUint32(buf[:], uint32(k)) //nolint:gosec // Safe conversion for hashing
96-
_, _ = h.Write(buf[:])
105+
if _, err := h.Write(buf[:]); err != nil {
106+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
107+
}
97108
case int64:
98109
var buf [8]byte
99110
binary.BigEndian.PutUint64(buf[:], uint64(k)) //nolint:gosec // Safe conversion for hashing
100-
_, _ = h.Write(buf[:])
111+
if _, err := h.Write(buf[:]); err != nil {
112+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
113+
}
101114
case uint:
102115
var buf [8]byte
103116
binary.BigEndian.PutUint64(buf[:], uint64(k))
104-
_, _ = h.Write(buf[:])
117+
if _, err := h.Write(buf[:]); err != nil {
118+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
119+
}
105120
case uint8:
106-
_, _ = h.Write([]byte{k})
121+
if _, err := h.Write([]byte{k}); err != nil {
122+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
123+
}
107124
case uint16:
108125
var buf [2]byte
109126
binary.BigEndian.PutUint16(buf[:], k)
110-
_, _ = h.Write(buf[:])
127+
if _, err := h.Write(buf[:]); err != nil {
128+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
129+
}
111130
case uint32:
112131
var buf [4]byte
113132
binary.BigEndian.PutUint32(buf[:], k)
114-
_, _ = h.Write(buf[:])
133+
if _, err := h.Write(buf[:]); err != nil {
134+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
135+
}
115136
case uint64:
116137
var buf [8]byte
117138
binary.BigEndian.PutUint64(buf[:], k)
118-
_, _ = h.Write(buf[:])
139+
if _, err := h.Write(buf[:]); err != nil {
140+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
141+
}
119142
case float32:
120143
var buf [4]byte
121144
binary.BigEndian.PutUint32(buf[:], math.Float32bits(k))
122-
_, _ = h.Write(buf[:])
145+
if _, err := h.Write(buf[:]); err != nil {
146+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
147+
}
123148
case float64:
124149
var buf [8]byte
125150
binary.BigEndian.PutUint64(buf[:], math.Float64bits(k))
126-
_, _ = h.Write(buf[:])
151+
if _, err := h.Write(buf[:]); err != nil {
152+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
153+
}
127154
case bool:
128155
if k {
129-
_, _ = h.Write([]byte{1})
156+
if _, err := h.Write([]byte{1}); err != nil {
157+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
158+
}
130159
} else {
131-
_, _ = h.Write([]byte{0})
160+
if _, err := h.Write([]byte{0}); err != nil {
161+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
162+
}
132163
}
133164
default:
134165
// For other types (structs, arrays, etc), use fmt.Fprintf for generic conversion
135-
_, _ = fmt.Fprintf(h, "%v", key)
166+
if _, err := fmt.Fprintf(h, "%v", key); err != nil {
167+
panic(fmt.Sprintf("unexpected hash.Write error: %v", err))
168+
}
136169
}
137170
hash1 := h.Sum64()
138171
// Generate additional hashes using double hashing
@@ -685,6 +718,7 @@ func NewWithDeduplication[T comparable](size int, timeout time.Duration, fn func
685718
triggerCh: make(chan struct{}),
686719
background: background,
687720
usePool: false,
721+
done: make(chan struct{}),
688722
},
689723
deduplicationWindow: deduplicationWindow,
690724
// Create an optimized time-partitioned map with bloom filter
@@ -710,6 +744,7 @@ func NewWithDeduplicationAndPool[T comparable](size int, timeout time.Duration,
710744
triggerCh: make(chan struct{}),
711745
background: background,
712746
usePool: true,
747+
done: make(chan struct{}),
713748
pool: &sync.Pool{
714749
New: func() interface{} {
715750
slice := make([]*T, 0, size)
@@ -727,8 +762,19 @@ func NewWithDeduplicationAndPool[T comparable](size int, timeout time.Duration,
727762
return b
728763
}
729764

730-
// Close properly shuts down the deduplication map resources.
765+
// Close properly shuts down the batcher and deduplication map resources.
766+
//
767+
// This method performs a graceful shutdown by:
768+
// 1. Stopping the background worker goroutine via the parent Batcher.Close()
769+
// 2. Processing any remaining items in the queue
770+
// 3. Closing the deduplication map and stopping its cleanup ticker
771+
//
772+
// It is safe to call Close() multiple times (subsequent calls have no effect).
773+
//
774+
// IMPORTANT: Do not call Put() after Close() has been called, as this will
775+
// result in a panic due to sending on a closed channel.
731776
func (b *BatcherWithDedup[T]) Close() {
777+
b.Batcher.Close()
732778
b.deduplicationMap.Close()
733779
}
734780

0 commit comments

Comments
 (0)