README
¶
Go package syncx provides generic helpers extending the standard sync package
Documentation
¶
Overview ¶
Package syncx provides generic helpers extending the standard sync package.
WaitGroup ¶
WaitGroup is a drop-in replacement for sync.WaitGroup that adds protection against the sync.WaitGroup.Add/sync.WaitGroup.Wait data race. Unlike sync.WaitGroup, WaitGroup owns its underlying sync.WaitGroup and provides WaitGroup.Add, which panics instead of causing a data race.
WaitGroup supports both the classic Add/Done/Wait pattern and the TryAdd/TryGo pattern for graceful shutdown.
TryWaitGroup ¶
TryWaitGroup wraps an external sync.WaitGroup and provides TryWaitGroup.TryAdd, which refuses to increment the counter after TryWaitGroup.Stop has been called. Use TryWaitGroup when the sync.WaitGroup is shared between multiple components.
TryWaitGroup solves a specific data-race problem that arises when sync.WaitGroup.Add is called concurrently with sync.WaitGroup.Wait. Standard Go idiom requires Add to happen before Wait starts; TryWaitGroup makes the concurrent case safe by providing TryAdd, which refuses to increment the counter after Stop has been called.
This pattern is useful when:
- External callers may invoke methods concurrently on a long-lived service, and you need to drain them during shutdown (a "reverse" worker pool).
- A scheduler or timer may spawn goroutines while shutdown is in progress.
- Goroutines are themselves created inside other goroutines, making it impossible to guarantee Add runs before Wait.
Alternatives ¶
Most graceful-shutdown needs are better served by golang.org/x/sync/errgroup with context cancellation:
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {
for {
select {
case <-ctx.Done():
return nil // shutdown requested
case job := <-jobs:
process(job)
}
}
})
// ... on signal:
cancel()
g.Wait()
TryWaitGroup is only needed when the code that adds work is outside your control — for example, when you expose public methods that can be called by an external framework, and you cannot restructure them to check a context before starting.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type TryWaitGroup ¶
type TryWaitGroup struct {
// contains filtered or unexported fields
}
TryWaitGroup provides an alternative to sync.WaitGroup.Add that can be used concurrently with sync.WaitGroup.Wait. Without it, calling Add with a positive delta once Wait might have started is a data race; TryAdd makes that case safe by refusing to add after Stop.
TryWaitGroup does not own the underlying WaitGroup — it wraps an external one. The caller is responsible for the overall lifecycle:
- Create a sync.WaitGroup and a TryWaitGroup via NewTryWaitGroup.
- Goroutines call TryWaitGroup.TryAdd before starting work and TryWaitGroup.Done when finished.
- On shutdown, call TryWaitGroup.Stop to prevent new work from registering, then call sync.WaitGroup.Wait on the underlying group to wait for already-registered work to complete.
Usage:
var wg sync.WaitGroup
twg := syncx.NewTryWaitGroup(&wg)
// In a goroutine that may race with shutdown:
func worker(twg *syncx.TryWaitGroup) {
if !twg.TryAdd(1) {
return // shutdown already in progress
}
defer twg.Done()
// do work ...
}
// On shutdown:
twg.Stop()
wg.Wait()
Example (Basic) ¶
This example shows the core pattern: workers that take varying time to start register via TryAdd, and the orchestrator uses Stop to prevent any further registrations before waiting on the underlying WaitGroup.
Workers with delays 40ms and 50ms arrive after Stop (called at ~33ms) and are rejected — TryAdd returns false and they return early.
package main
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/powerman/syncx"
)
func main() {
var wg sync.WaitGroup
twg := syncx.NewTryWaitGroup(&wg)
delays := []int{10, 20, 30, 40, 50}
var started atomic.Int64
for _, d := range delays {
go func() {
// Simulate variable-length setup before registration.
time.Sleep(time.Duration(d) * time.Millisecond)
if !twg.TryAdd(1) {
return // shutdown already in progress, bail out
}
defer twg.Done()
started.Add(1)
}()
}
// Give the first three workers time to register,
// but not the last two.
time.Sleep(33 * time.Millisecond)
// Stop accepting new work, then wait for active workers.
twg.Stop()
wg.Wait()
fmt.Println("started workers:", started.Load())
}
Output: started workers: 3
Example (GracefulShutdown) ¶
This example shows graceful shutdown of a service whose public methods are called concurrently by external code — a pattern where the callers are outside your control, so TryAdd is needed.
Without TryWaitGroup, calling wg.Add inside a method while another goroutine calls wg.Wait would be a data race. TryAdd makes this safe.
package main
import (
"fmt"
"slices"
"sync"
"time"
"github.com/powerman/syncx"
)
func main() {
var wg sync.WaitGroup
twg := syncx.NewTryWaitGroup(&wg)
var mu sync.Mutex
var processed []string
// A service that processes requests.
// Its Process method can be called from any goroutine.
process := func(msg string) {
if !twg.TryAdd(1) {
return // shutdown in progress, silently drop
}
defer twg.Done()
mu.Lock()
processed = append(processed, msg)
mu.Unlock()
}
// Start a few concurrent requests.
go process("request-a")
go process("request-b")
go process("request-c")
// Give them time to register.
time.Sleep(5 * time.Millisecond)
// Initiate graceful shutdown.
twg.Stop()
wg.Wait() // waits for a, b, c to finish
// Any new request after Stop is silently dropped.
process("request-d") // never added to processed
// Sort for deterministic output.
slices.Sort(processed)
for _, msg := range processed {
fmt.Println("processed:", msg)
}
}
Output: processed: request-a processed: request-b processed: request-c
func NewTryWaitGroup ¶
func NewTryWaitGroup(wg *sync.WaitGroup) *TryWaitGroup
NewTryWaitGroup creates and returns a TryWaitGroup related to wg.
Do not forget to call Stop on each TryWaitGroup related to wg before wg.Wait.
Panics if wg is nil — use NewWaitGroup if you need a self-owned WaitGroup.
func (*TryWaitGroup) Stop ¶
func (twg *TryWaitGroup) Stop()
Stop marks the TryWaitGroup as stopped. After Stop returns, any subsequent call to TryAdd with a positive delta will return false.
Stop must be called before the underlying sync.WaitGroup.Wait to avoid the data race it is designed to prevent. Stop is idempotent.
func (*TryWaitGroup) TryAdd ¶
func (twg *TryWaitGroup) TryAdd(delta int) bool
TryAdd works like sync.WaitGroup.Add except it returns false if called with a positive delta after Stop. This makes it safe to use TryAdd concurrently with sync.WaitGroup.Wait:
if !twg.TryAdd(1) {
return // shutdown in progress, bail out
}
defer twg.Done()
// ...
The positive-delta check means that Done (which calls Add(-1)) always passes through, even after Stop.
func (*TryWaitGroup) TryGo ¶
func (twg *TryWaitGroup) TryGo(fn func()) bool
TryGo starts fn in a new goroutine, registered via TryWaitGroup.TryAdd. It returns false if the group has been stopped, in which case fn is not called.
type WaitGroup ¶
type WaitGroup struct {
*TryWaitGroup
}
WaitGroup is a drop-in replacement for sync.WaitGroup that adds protection against the sync.WaitGroup.Add/sync.WaitGroup.Wait data race. Unlike TryWaitGroup, WaitGroup owns its underlying sync.WaitGroup.
A zero WaitGroup is not valid. Use NewWaitGroup to create one.
func NewWaitGroup ¶
func NewWaitGroup() *WaitGroup
NewWaitGroup creates and returns a new WaitGroup.
func (*WaitGroup) Add ¶
Add works like sync.WaitGroup.Add. It panics if called with a positive delta after WaitGroup.Wait has been called.
func (*WaitGroup) Wait ¶
func (wg *WaitGroup) Wait()
Wait blocks until the WaitGroup counter is zero. It calls [WaitGroup.Stop] before waiting to prevent new goroutines from registering. After Wait returns, any subsequent WaitGroup.Add with a positive delta panics.
Source Files