perf + fix: zero-alloc fan-out, send-on-closed race, logger cleanup#4
Merged
Conversation
deliver's fan-out read each subscriber's per-topic channel via channels.Load then select-sent; a concurrent unsubscribe could close(ch) in that window, panicking the publisher goroutine and crashing the whole process. Switch Subscriber.channels from sync.Map to a mutex-guarded map[string]chan T, and have deliver hold subscriber.mutex while reading the channel and sending. This serializes send against unsubscribe's close: channels[topic] absent means already unsubscribed so deliver skips; otherwise the send completes before close can run. resetTimer runs outside the lock to avoid re-entrant deadlock with its own subscriber.mutex take. Lock order is unchanged: deliver releases subscription.rwMutex before taking subscriber.mutex, so no subscription->subscriber ordering is introduced. Verified with go test -race -count=3 ./pubsub/... plus the new TestConcurrentPublishWithDynamicSubscribe stress test (4 publishers + 8 dynamic subscribe/close goroutines). Co-Authored-By: glm-5.2
deliver snapshotted the subscriber set on every Publish with a
fresh make([]*Subscriber[T], 0, len(subscribers)) — at N=10000
subscribers that is ~80KB allocated per Publish, hammering the
GC under concurrent publishers.
Pool the snapshot slice per subscription. Pool element is
*[]*Subscriber[T] (pointer, not slice) so only an 8-byte pointer
boxes into the Get/Put interface{} — storing the slice directly
would escape the whole slice header. After fan-out, Reset to [:0]
and Put; the backing array's capacity naturally grows to the
subscription's steady-state subscriber count. sync.Pool is
concurrency-safe and per-P cached, so concurrent delivers get
independent slice instances.
Measured (count=5 median): UltraLarge 10k subs 2.67ms->2.03ms
(-24%), HighLoadParallel 113us->84us (-26%), both 1 alloc/op ->
0. Single-subscriber +5ns (mutex) is acceptable for a
fire-and-forget fan-out library.
TestDeliverZeroAlloc (//go:build !race; the race detector skews
alloc counts) guards the invariant.
Co-Authored-By: glm-5.2
Topics, tryRemoveSubscription, isEmpty, addSubscriber, removeSubscriber (broker.go) and Subscribe, unsubscribe (subscriber.go) each called logger.Enabled(LevelDebug) twice — once on acquire, once inside the release defer. Fold both into a single `debug :=` local, matching the pattern already used in createOrLoadSubscription and deliver. Pure consistency cleanup; no behavior change (the Enabled gate is inlined to a fast false when debug is disabled). TestDebugLevelLoggerExercisesHotPath still exercises the gated bodies. Co-Authored-By: glm-5.2
README: update the benchmark table to median-of-3 numbers post PR #4 (fan-out: UltraLarge -24%, HighLoadParallel -26%, all hot paths zero-allocation). Add a note that the non-zero B/op on the big fan-out benchmarks comes from the harness's drain goroutines, not library code. CLAUDE.md: - Architecture: Subscriber.channels is now a subscriber.mutex- guarded map[string]chan T (was sync.Map); the per-subscription errCh lives on each *Subscription. deliver snapshots the subscriber set via a sync.Pool-recycled slice and sends under subscriber.mutex. - Locking pattern: document the new deliver<->subscriber.mutex contract (send vs unsubscribe close are mutually exclusive) and why resetTimer runs outside the lock. - Gotchas: add the send-on-closed-channel race history and its regression test TestConcurrentPublishWithDynamicSubscribe. - Code layout: list concurrent_test.go and perf_test.go. Co-Authored-By: glm-5.2
CI runs a gofmt gate (gofmt -l must be empty). The channels map field comment and the NewSubscriber struct-literal keys were not aligned to the column gofmt expects after channels: was added. Verified: gofmt -l pubsub/ cmd/ is now empty. Co-Authored-By: glm-5.2
CLAUDE.md claimed "There is no Makefile, linter config, or CI file" and that go test/vet was the whole story — all three were wrong: Makefile, .golangci.yml, and .github/workflows/test.yml all exist. This inaccuracy is what led to committing un-gofmt'd code and turning CI red on this very PR. Replace with an accurate list of the gates CI enforces (go mod tidy, gofmt, vet, race test, golangci-lint v2.12.2 with only-new-issues), and note that infertypeargs is excluded in .golangci.yml so it won't fail CI. Co-Authored-By: glm-5.2
Address code-review feedback: - broker.go: nil out the *Subscriber pointers in the pooled snapshot slice before Put. sync.Pool is cleared by the runtime on GC, and the next Get's [:0] only resets length (not elements), so without this the Pool's backing array keeps every snapshotted Subscriber reachable until the next GC — delaying reclamation of unsubscribed subscribers and inflating memory peaks for large fan-out. - perf_test.go: debug.SetGCPercent(-1) around the measurement so a GC clearing the Pool mid-run can't make the zero-alloc assertion flaky. - concurrent_test.go: count Subscribe errors instead of silently returning; a systemic Subscribe failure would otherwise let the stress test pass vacuously. - subscriber.go: check channels[topic] before make-ing the chan so re-Subscribe doesn't allocate a channel it immediately discards. Verified: go test -race -count=3 green; TestDeliverZeroAlloc green. Co-Authored-By: glm-5.2
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Three logically separated commits hardening and speeding up the fan-out hot path. Split so the bug fix can be cherry-picked / reverted independently of the perf work.
1.
fix(deliver): send-on-closed-channel race (815c24a) — pre-existing bugdeliver's fan-out read each subscriber's per-topic channel viachannels.Loadthenselect-sent. A concurrentunsubscribecouldclose(ch)in that window, panicking the publisher goroutine and crashing the whole process. The new safety-net testTestConcurrentPublishWithDynamicSubscribereproduces it deterministically under-race.Fix:
Subscriber.channelsmoves fromsync.Mapto asubscriber.mutex-guardedmap[string]chan T, anddeliverholdssubscriber.mutexwhile reading the channel + sending. This serializes send againstunsubscribe's close (an absentchannels[topic]means already unsubscribed → skip).resetTimerruns outside the lock to avoid re-entrant deadlock. Lock order is unchanged (no new deadlock surface —deliverreleasessubscription.rwMutexbefore takingsubscriber.mutex).2.
perf(deliver): zero-allocation snapshot viasync.Pool(44d9403)deliverallocated a fresh snapshot slice on every Publish (~80KB at 10k subscribers, GC-thrashing under concurrent publishers). Pooled per-subscriptionwith*[]*Subscriber[T]elements — a pointer (not the slice) so only 8 bytes box into theGet/Putinterface{}; the backing array grows naturally to steady-state subscriber count.3.
chore: fold duplicatedlogger.Enabledcalls (3feb554)Topics,tryRemoveSubscription,isEmpty,addSubscriber,removeSubscriber,Subscribe,unsubscribeeach calledlogger.Enabled(LevelDebug)twice (acquire + release). Folded into onedebug :=local, matching the pattern already increateOrLoadSubscription/deliver. Pure consistency; no behavior change.Benchmarks (count=5 median, Intel Core Ultra 5 125H)
Multi-subscriber fan-out — the core fire-and-forget use case — wins big; single-subscriber pays ~5ns for the mutex.
mutex(≈15ns) beatssync.Map.Load(≈30ns) + type assertion once N grows.Test plan
go test -race -count=3 ./pubsub/...green — each commit independentlyTestConcurrentPublishWithDynamicSubscribe(4 publishers + 8 dynamic subscribe/close goroutines) race-clean — the regression test for the race fixTestDeliverZeroAlloc(//go:build !race— the race detector skews alloc counts) guards the 0-alloc invariantgo vetclean🤖 Generated with Claude Code