aboutsummaryrefslogtreecommitdiffhomepage
path: root/internal/fmsg/defer.go
diff options
context:
space:
mode:
Diffstat (limited to 'internal/fmsg/defer.go')
-rw-r--r--internal/fmsg/defer.go23
1 files changed, 23 insertions, 0 deletions
diff --git a/internal/fmsg/defer.go b/internal/fmsg/defer.go
index deef58f0..ed4a7735 100644
--- a/internal/fmsg/defer.go
+++ b/internal/fmsg/defer.go
@@ -8,6 +8,7 @@ import (
var (
wstate atomic.Bool
+ dropped atomic.Uint64
withhold = make(chan struct{}, 1)
msgbuf = make(chan dOp, 64) // these ops are tiny so a large buffer is allocated for withholding output
@@ -29,6 +30,25 @@ func dequeue() {
}()
}
+// queue submits ops to msgbuf but drops messages
+// when the buffer is full and dequeue is withholding
+func queue(op dOp) {
+ select {
+ case msgbuf <- op:
+ queueSync.Add(1)
+ default:
+ // send the op anyway if not withholding
+ // as dequeue will get to it eventually
+ if !wstate.Load() {
+ queueSync.Add(1)
+ msgbuf <- op
+ } else {
+ // increment dropped message count
+ dropped.Add(1)
+ }
+ }
+}
+
type dOp interface{ Do() }
func Exit(code int) {
@@ -47,6 +67,9 @@ func Resume() {
dequeueOnce.Do(dequeue)
if wstate.CompareAndSwap(true, false) {
withhold <- struct{}{}
+ if d := dropped.Swap(0); d != 0 {
+ Printf("dropped %d messages during withhold", d)
+ }
}
}