From 19584d0939c1e124cb8f6c8b1cb0ea980097ee7b Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 30 Sep 2026 22:57:58 +0800 Subject: [PATCH] fix: harden high-fanout delivery under load --- .../IM-单机60%容量压力测试结果-20260929.md | 49 ++++++++++ loadtest/im-pressure/agent.go | 27 +++++- loadtest/im-pressure/agent_metrics.go | 21 ++++ loadtest/im-pressure/callback_queue.go | 70 ++++++++++++++ loadtest/im-pressure/callback_queue_test.go | 47 +++++++++ loadtest/im-pressure/model.go | 95 +++++++++++-------- loadtest/im-pressure/recorder.go | 84 ++++++++++++---- loadtest/im-pressure/wsclient.go | 23 +++-- server/db/mysql/adapter.go | 21 ++++ server/db/mysql/invalid_connection_test.go | 55 +++++++++++ server/db/mysql/metrics_driver.go | 10 ++ server/db/mysql/tenant_adapter.go | 45 ++++++--- server/hdl_websock.go | 24 ++++- server/metrics/metrics.go | 10 +- 14 files changed, 491 insertions(+), 90 deletions(-) create mode 100644 loadtest/im-pressure/callback_queue.go create mode 100644 loadtest/im-pressure/callback_queue_test.go create mode 100644 server/db/mysql/invalid_connection_test.go diff --git a/docs/test/IM-单机60%容量压力测试结果-20260929.md b/docs/test/IM-单机60%容量压力测试结果-20260929.md index c972cb2..35e063e 100644 --- a/docs/test/IM-单机60%容量压力测试结果-20260929.md +++ b/docs/test/IM-单机60%容量压力测试结果-20260929.md @@ -347,3 +347,52 @@ Run ID：`s900m-pushfix-r2-0930`，精确群成员 900 人，A/B 各 450 条 Web 3. 压测客户端为每条连接新增有序数据回调队列。Socket 读循环不再同步执行消息去重、顺序和延迟统计，避免统计锁竞争阻塞 ACK、Ping 和 Pong 读取。 发版后仍使用同一精确 900 人群复测。起压前先确认三个 Pod 的 heartbeat gauge 均为 15/90；验收时对比 Pong 增量、timeout 增量、agent 断连和入服/落库数。若仍然同时批量 timeout，下一步必须取得 Ingress/LB TCP 指标或改用同地域压测机，不再继续放大应用心跳窗口。 + +## 18. 90 秒心跳发版后精确 900 人复测 + +Run ID：`s900m-heart-r1-0930`。三个 Pod 发版后均为 Ready、零重启，指标确认有效心跳配置均为 `ping_period=15`、`pong_wait=90`。使用与上一轮相同的精确 900 人群，A/B 各 450 条连接，完整运行至历史回查和渐进关闭结束。 + +| 指标 | 本轮 | 改造前 R2 | +|---|---:|---:| +| 发送尝试 | 900 | 900 | +| 收到 ACK | 854 | 590 | +| ACK 超时 / socket 错误 | 45 / 0 | 310 条结果未知，其中 124 条未入服 | +| 明确失败 | 1 条 `500 internal error` | 0 | +| 服务端实际接收并落库 | 899 / 900 | 776 / 900 | +| 断连 | 11 | 410 | +| ACK P95 | 11475.912ms | 17075.685ms | +| ACK 台账实时投递 | 765336 / 768600 | 381424 / 531000 | +| ACK 台账实时到达率 | 99.5753% | 71.8313% | +| 重复 / 乱序 | 0 / 0 | 0 / 0 | +| 历史回查 | 899 / 899 | 776 / 776 | +| MySQL 精确命中 | 899 / 899 | 776 / 776 | +| MySQL 缺失 / 错配 / 重复 / 额外 | 0 / 0 / 0 / 0 | 0 / 0 / 0 / 0 | + +心跳容错和压测端读循环解耦有明显效果：断连从 410 次降到 11 次，降幅约 97.3%；未入服数从 124 降到 1。负载期三 Pod 累计成功发送约 14554 次 Ping，已观察到约 14472 次 Pong。11 次断连仍在 90 秒窗口附近发生，客户端全部自动重连。当前心跳仅在 Pong 回调时续期；下一步应将有效的客户端 Ping 和业务入站帧同样视为连接活动，而不是继续扩大 Pong 窗口。 + +45 条 ACK 超时消息全部被对端 WS 观察，且 899 条实际入服消息全部历史可查、MySQL 精确命中，证明已入服消息仍为零持久化丢失。唯一未入服消息已定位到 `10:07:58` 群主 Pod 的 MySQL `invalid connection`：同时出现群禁言策略查询失败和消息保存失败，客户端收到 `500 internal error`。需要对失效 MySQL 连接进行识别、丢弃并安全重试。 + +服务端分段指标中，899 条成功消息的 permission check P95 不超过 25ms、save P95 不超过 50ms、Kafka produce P95 不超过 25ms、realtime fanout P95 不超过 5ms、push prepare 和 push queue wait P95 不超过 10ms、push dispatch P95 不超过 25ms，无任何内部队列 full，Kafka lag 为 0。但 Topic queue wait P95 为 5s，最后 124 条在连接 timeout 与 presence 处理同期等待 0.1–10s，使 ACK P95 仍未过门禁。 + +压测 agent 峰值 CPU 约为 28.8%/22.8% 单核，RSS 约 121MiB，FD 各 460，进程资源未饱和。但异步数据回调队列尚未暴露当前深度和等待时间，因此本轮 98 秒投递 P95 包含网络、socket 读取和压测统计队列三段时间，不能单独作为服务端投递 SLA。下一轮前需增加压测回调队列深度/等待指标，并减少 recorder 在同一 `client_mid` 上的锁竞争。 + +新增产物： + +- `loadtest/im-pressure/results/s900m-heartbeat-r1-0930/report.json` +- `loadtest/im-pressure/results/s900m-heartbeat-r1-0930/db-verification.json` +- `loadtest/im-pressure/results/s900m-heartbeat-r1-0930/server-metrics/` + +## 19. 900 人复测后第二轮修复 + +针对第 18 节剩余的 11 次 timeout、1 条 MySQL `invalid connection` 和压测端回调排队不可观测问题，已完成以下修复： + +1. 服务端收到有效客户端 Ping 时立即续期读截止时间并回 Pong；成功读取业务帧时也续期读截止时间。连接活跃判定不再只依赖服务端 Ping 对应的 Pong，新增 `im_ws_heartbeat_total{event="ping_received"}` 观测客户端心跳。 +2. MySQL 只读查询遇到 `ErrInvalidConn` 时向 `database/sql` 返回 `ErrBadConn`，使连接池丢弃坏连接并在新连接上重试。个人消息仅在存在 `client_msg_id` 唯一键保护时重试一次；群消息由 `message_id` 和 `client_msg_id` 唯一键保护，整个保存操作可安全重试一次。新增 `im_db_invalid_connection_retry_total{operation,outcome}`。 +3. recorder 的群消息接收位图改为原子 CAS。同一 `client_mid` 只在首次创建 observation 时使用 shard 锁，后续数百个收件人不再串行争用同一把写锁。 +4. agent 新增回调队列当前深度、处理数、峰值、入队/完成数、累计等待和最大等待指标，并将快照写入 `summary.json` 的 `callback_queue`。生成实时到达率前会等待排队及正在处理的回调全部完成，避免统计尚未落入 recorder 时产生假丢失。 + +下一轮仍使用精确 900 人群复测。除原有正确性门禁外，重点检查： + +- `im_ws_heartbeat_total{event="ping_received"}` 有持续增量，服务端 timeout 和客户端断连进一步下降。 +- `im_db_invalid_connection_retry_total` 若出现，结果应为 `discard` 或 `recovered`，客户端不再收到对应 500。 +- `callback_queue.current` 和 `callback_queue.processing` 在结算时均为 0；结合 `maximum`、`wait_seconds_total / completed` 和 `wait_seconds_max` 区分压测端排队与服务端投递延迟。 diff --git a/loadtest/im-pressure/agent.go b/loadtest/im-pressure/agent.go index e556cea..6306fc2 100644 --- a/loadtest/im-pressure/agent.go +++ b/loadtest/im-pressure/agent.go @@ -26,6 +26,7 @@ type agentRun struct { clients map[int]*WSClient clientsMu sync.RWMutex recorder *deliveryRecorder + callbackQueue *callbackQueueMetrics ack *DurationHistogram attempted atomic.Int64 ready atomic.Int64 @@ -125,7 +126,8 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo } coord := newCoordinatorClient(coordinatorURL) placeholderRunID := "pending" - run := &agentRun{cfg: cfg, machine: machine, clients: make(map[int]*WSClient), ack: NewDurationHistogram()} + run := &agentRun{cfg: cfg, machine: machine, clients: make(map[int]*WSClient), ack: NewDurationHistogram(), + callbackQueue: &callbackQueueMetrics{}} run.recorder = newDeliveryRecorder(placeholderRunID, machine, len(accounts)) stopMetrics, err := startAgentMetrics(metricsListen, run) if err != nil { @@ -206,6 +208,9 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo if err = waitContext(ctx, drain); err != nil { return err } + if err = waitForCallbackQueue(ctx, run.callbackQueue, 30*time.Second); err != nil { + return err + } observedLocal := run.recorder.unackedObservedMessages(ledger, start.Scenario) if err = coord.submitObserved(ctx, machine, observedLocal); err != nil { return err @@ -223,6 +228,7 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo PublishRejected: run.explicitRejected(), PublishOutcomeUnknown: run.outcomeUnknown(), PublishErrors: run.publishErrorCounts(), AckLatency: run.ack.Snapshot(), AcceptedLocal: local, + CallbackQueue: run.callbackQueue.snapshot(), } run.mu.Lock() base.Errors = append(base.Errors, run.errors...) @@ -270,6 +276,23 @@ func runAgent(ctx context.Context, cfg *Config, manifest *Manifest, machine, coo return nil } +func waitForCallbackQueue(ctx context.Context, queue *callbackQueueMetrics, timeout time.Duration) error { + deadline := time.NewTimer(timeout) + defer deadline.Stop() + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for !queue.isDrained() { + select { + case <-ctx.Done(): + return ctx.Err() + case <-deadline.C: + return fmt.Errorf("data callback queue did not drain within %s: current=%d processing=%d", + timeout, queue.current.Load(), queue.processing.Load()) + case <-ticker.C: + } + } + return nil +} type historyQuery struct { client *WSClient topic string @@ -667,7 +690,7 @@ func (r *agentRun) maintainConnections(ctx context.Context, accounts []AccountRe func (r *agentRun) connectAccountWithRetry(ctx context.Context, account AccountRecord, groupTopic, apiKey, password string) (*WSClient, error) { var lastErr error for attempt := 1; attempt <= 6; attempt++ { - client, err := connectWSClient(ctx, r.cfg, account, apiKey, password, r.recorder.onData, r.recorder.onDisconnect) + client, err := connectWSClient(ctx, r.cfg, account, apiKey, password, r.recorder.onData, r.recorder.onDisconnect, r.callbackQueue) if err == nil { if err = client.Subscribe(ctx, account.P2PTopic); err == nil { err = client.Subscribe(ctx, groupTopic) diff --git a/loadtest/im-pressure/agent_metrics.go b/loadtest/im-pressure/agent_metrics.go index c620cf8..9dd69de 100644 --- a/loadtest/im-pressure/agent_metrics.go +++ b/loadtest/im-pressure/agent_metrics.go @@ -63,6 +63,27 @@ func startAgentMetrics(address string, run *agentRun) (func(), error) { _, _, _, disconnects := run.recorder.metricCounts() return float64(disconnects) }), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_callback_queue_current", Help: "Data callbacks currently waiting to start.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.current.Load()) + }), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_callback_queue_max", Help: "Peak data callbacks waiting to start.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.maximum.Load()) + }), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_callback_processing", Help: "Data callbacks currently processing.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.processing.Load()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_callback_queue_enqueued_total", Help: "Data callbacks enqueued.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.enqueued.Load()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_callback_queue_completed_total", Help: "Data callbacks which completed processing.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.completed.Load()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{Name: "im_load_agent_callback_queue_wait_seconds_total", Help: "Cumulative time spent waiting to start data callbacks.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.waitNanos.Load()) / float64(time.Second) + }), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: "im_load_agent_callback_queue_wait_seconds_max", Help: "Maximum time spent waiting to start a data callback.", ConstLabels: labels}, func() float64 { + return float64(run.callbackQueue.maxWait.Load()) / float64(time.Second) + }), ) mux := http.NewServeMux() mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{})) diff --git a/loadtest/im-pressure/callback_queue.go b/loadtest/im-pressure/callback_queue.go new file mode 100644 index 0000000..71b4613 --- /dev/null +++ b/loadtest/im-pressure/callback_queue.go @@ -0,0 +1,70 @@ +package main + +import ( + "sync/atomic" + "time" +) + +type callbackQueueMetrics struct { + current atomic.Int64 + processing atomic.Int64 + maximum atomic.Int64 + enqueued atomic.Int64 + completed atomic.Int64 + waitNanos atomic.Int64 + maxWait atomic.Int64 +} + +func (m *callbackQueueMetrics) enqueue() { + if m == nil { + return + } + current := m.current.Add(1) + m.enqueued.Add(1) + updateAtomicMax(&m.maximum, current) +} + +func (m *callbackQueueMetrics) startCallback(queuedAt time.Time) { + if m == nil { + return + } + wait := time.Since(queuedAt).Nanoseconds() + if wait < 0 { + wait = 0 + } + m.current.Add(-1) + m.processing.Add(1) + m.waitNanos.Add(wait) + updateAtomicMax(&m.maxWait, wait) +} + +func (m *callbackQueueMetrics) finishCallback() { + if m == nil { + return + } + m.processing.Add(-1) + m.completed.Add(1) +} + +func (m *callbackQueueMetrics) snapshot() CallbackQueueSnapshot { + if m == nil { + return CallbackQueueSnapshot{} + } + return CallbackQueueSnapshot{ + Current: m.current.Load(), Processing: m.processing.Load(), Maximum: m.maximum.Load(), Enqueued: m.enqueued.Load(), + Completed: m.completed.Load(), WaitSecondsTotal: float64(m.waitNanos.Load()) / float64(time.Second), + WaitSecondsMax: float64(m.maxWait.Load()) / float64(time.Second), + } +} + +func (m *callbackQueueMetrics) isDrained() bool { + return m == nil || (m.current.Load() == 0 && m.processing.Load() == 0) +} + +func updateAtomicMax(target *atomic.Int64, value int64) { + for current := target.Load(); value > current; current = target.Load() { + if target.CompareAndSwap(current, value) { + return + } + } +} diff --git a/loadtest/im-pressure/callback_queue_test.go b/loadtest/im-pressure/callback_queue_test.go new file mode 100644 index 0000000..e7015d6 --- /dev/null +++ b/loadtest/im-pressure/callback_queue_test.go @@ -0,0 +1,47 @@ +package main + +import ( + "context" + "testing" + "time" +) + +func TestCallbackQueueMetricsTrackWaitingAndProcessing(t *testing.T) { + metrics := &callbackQueueMetrics{} + metrics.enqueue() + metrics.enqueue() + metrics.startCallback(time.Now().Add(-10 * time.Millisecond)) + + snapshot := metrics.snapshot() + if snapshot.Current != 1 || snapshot.Processing != 1 || snapshot.Maximum != 2 || snapshot.Enqueued != 2 { + t.Fatalf("unexpected active queue snapshot: %+v", snapshot) + } + if snapshot.WaitSecondsTotal <= 0 || snapshot.WaitSecondsMax <= 0 || metrics.isDrained() { + t.Fatalf("wait or drain state not tracked: %+v", snapshot) + } + + metrics.finishCallback() + metrics.startCallback(time.Now()) + metrics.finishCallback() + if snapshot = metrics.snapshot(); snapshot.Current != 0 || snapshot.Processing != 0 || snapshot.Completed != 2 || !metrics.isDrained() { + t.Fatalf("unexpected drained queue snapshot: %+v", snapshot) + } +} + +func TestWaitForCallbackQueueWaitsForProcessingCallback(t *testing.T) { + metrics := &callbackQueueMetrics{} + metrics.enqueue() + metrics.startCallback(time.Now()) + go func() { + time.Sleep(20 * time.Millisecond) + metrics.finishCallback() + }() + + started := time.Now() + if err := waitForCallbackQueue(context.Background(), metrics, time.Second); err != nil { + t.Fatal(err) + } + if time.Since(started) < 15*time.Millisecond { + t.Fatal("wait returned before the callback finished") + } +} diff --git a/loadtest/im-pressure/model.go b/loadtest/im-pressure/model.go index ba14388..a354870 100644 --- a/loadtest/im-pressure/model.go +++ b/loadtest/im-pressure/model.go @@ -385,48 +385,59 @@ type AcceptedMessage struct { } type AgentSummary struct { - RunID string `json:"run_id"` - Scenario string `json:"scenario"` - Machine string `json:"machine"` - StartedAt time.Time `json:"started_at"` - LoadEndedAt time.Time `json:"load_ended_at"` - FinishedAt time.Time `json:"finished_at"` - ExpectedConnections int `json:"expected_connections"` - ReadyConnections int `json:"ready_connections"` - PublishAttempted int64 `json:"publish_attempted"` - PublishAccepted int64 `json:"publish_accepted"` - PublishRejected int64 `json:"publish_rejected"` - PublishOutcomeUnknown int64 `json:"publish_outcome_unknown"` - PublishErrors map[string]int64 `json:"publish_errors,omitempty"` - ExpectedPeer int64 `json:"expected_peer"` - ReceivedPeer int64 `json:"received_peer"` - ExpectedEcho int64 `json:"expected_echo"` - ReceivedEcho int64 `json:"received_echo"` - ExpectedGroup int64 `json:"expected_group"` - ReceivedGroup int64 `json:"received_group"` - RealtimeLost int64 `json:"realtime_lost"` - RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` - Duplicates int64 `json:"duplicates"` - OutOfOrder int64 `json:"out_of_order"` - OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` - Disconnects int64 `json:"disconnects"` - DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` - DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` - UnexpectedMessages int64 `json:"unexpected_messages"` - HistoryChecked int64 `json:"history_checked"` - HistoryFound int64 `json:"history_found"` - HistoryUnackedChecked int64 `json:"history_unacked_observed_checked"` - HistoryUnackedFound int64 `json:"history_unacked_observed_found"` - PersistenceLost int64 `json:"persistence_lost"` - AckLatency HistogramSnapshot `json:"ack_latency"` - DeliveryLatency HistogramSnapshot `json:"delivery_latency"` - DeliveryLatencyLocal HistogramSnapshot `json:"delivery_latency_local"` - DeliveryLatencyCross HistogramSnapshot `json:"delivery_latency_cross_machine"` - CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` - Errors []string `json:"errors,omitempty"` - Missing []MissingDelivery `json:"missing,omitempty"` - PersistenceMissing []string `json:"persistence_missing,omitempty"` - AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` + RunID string `json:"run_id"` + Scenario string `json:"scenario"` + Machine string `json:"machine"` + StartedAt time.Time `json:"started_at"` + LoadEndedAt time.Time `json:"load_ended_at"` + FinishedAt time.Time `json:"finished_at"` + ExpectedConnections int `json:"expected_connections"` + ReadyConnections int `json:"ready_connections"` + PublishAttempted int64 `json:"publish_attempted"` + PublishAccepted int64 `json:"publish_accepted"` + PublishRejected int64 `json:"publish_rejected"` + PublishOutcomeUnknown int64 `json:"publish_outcome_unknown"` + PublishErrors map[string]int64 `json:"publish_errors,omitempty"` + ExpectedPeer int64 `json:"expected_peer"` + ReceivedPeer int64 `json:"received_peer"` + ExpectedEcho int64 `json:"expected_echo"` + ReceivedEcho int64 `json:"received_echo"` + ExpectedGroup int64 `json:"expected_group"` + ReceivedGroup int64 `json:"received_group"` + RealtimeLost int64 `json:"realtime_lost"` + RealtimeLostStored int64 `json:"realtime_lost_but_persisted"` + Duplicates int64 `json:"duplicates"` + OutOfOrder int64 `json:"out_of_order"` + OrderingViolations []OrderingViolation `json:"ordering_violations,omitempty"` + Disconnects int64 `json:"disconnects"` + DisconnectReasons map[string]int64 `json:"disconnect_reasons,omitempty"` + DisconnectSamples []DisconnectSample `json:"disconnect_samples,omitempty"` + UnexpectedMessages int64 `json:"unexpected_messages"` + HistoryChecked int64 `json:"history_checked"` + HistoryFound int64 `json:"history_found"` + HistoryUnackedChecked int64 `json:"history_unacked_observed_checked"` + HistoryUnackedFound int64 `json:"history_unacked_observed_found"` + PersistenceLost int64 `json:"persistence_lost"` + AckLatency HistogramSnapshot `json:"ack_latency"` + DeliveryLatency HistogramSnapshot `json:"delivery_latency"` + DeliveryLatencyLocal HistogramSnapshot `json:"delivery_latency_local"` + DeliveryLatencyCross HistogramSnapshot `json:"delivery_latency_cross_machine"` + CrossClockSkewSamples int64 `json:"cross_machine_negative_latency_samples"` + CallbackQueue CallbackQueueSnapshot `json:"callback_queue"` + Errors []string `json:"errors,omitempty"` + Missing []MissingDelivery `json:"missing,omitempty"` + PersistenceMissing []string `json:"persistence_missing,omitempty"` + AcceptedLocal []AcceptedMessage `json:"accepted_local,omitempty"` +} + +type CallbackQueueSnapshot struct { + Current int64 `json:"current"` + Processing int64 `json:"processing"` + Maximum int64 `json:"maximum"` + Enqueued int64 `json:"enqueued"` + Completed int64 `json:"completed"` + WaitSecondsTotal float64 `json:"wait_seconds_total"` + WaitSecondsMax float64 `json:"wait_seconds_max"` } type DisconnectSample struct { diff --git a/loadtest/im-pressure/recorder.go b/loadtest/im-pressure/recorder.go index 2465d98..5b9e5f1 100644 --- a/loadtest/im-pressure/recorder.go +++ b/loadtest/im-pressure/recorder.go @@ -17,6 +17,16 @@ import ( const observationShardCount = 64 type observation struct { + topic string + seq int64 + senderMachine string + senderIndex int + scenario string + sentAt int64 + bits []atomic.Uint64 +} + +type observationView struct { topic string seq int64 senderMachine string @@ -121,23 +131,26 @@ func (r *deliveryRecorder) onData(account AccountRecord, data dataPacket) { return } shard := r.shard(mid) - shard.mu.Lock() + shard.mu.RLock() obs := shard.observed[mid] + shard.mu.RUnlock() if obs == nil { - obs = &observation{topic: data.Topic, seq: data.Seq, senderMachine: content.SenderMachine, - senderIndex: content.SenderIndex, scenario: content.Scenario, sentAt: content.SentAt, - bits: make([]uint64, (r.localUsers+63)/64)} - shard.observed[mid] = obs + shard.mu.Lock() + obs = shard.observed[mid] + if obs == nil { + obs = &observation{topic: data.Topic, seq: data.Seq, senderMachine: content.SenderMachine, + senderIndex: content.SenderIndex, scenario: content.Scenario, sentAt: content.SentAt, + bits: make([]atomic.Uint64, (r.localUsers+63)/64)} + shard.observed[mid] = obs + } + shard.mu.Unlock() } word, bit := index/64, uint(index%64) mask := uint64(1) << bit - if obs.bits[word]&mask != 0 { - shard.mu.Unlock() + if !markObservationBit(obs.bits, word, mask) { r.duplicates.Add(1) return } - obs.bits[word] |= mask - shard.mu.Unlock() r.received.Add(1) sequence := &r.sequences[index] sequence.mu.Lock() @@ -235,7 +248,7 @@ func (r *deliveryRecorder) unackedObservedMessages(ledger []AcceptedMessage, sce } message := AcceptedMessage{RunID: runID, Scenario: observedScenario, Kind: kind, ClientMid: mid, Topic: obs.topic, Seq: obs.seq, SenderMachine: obs.senderMachine, - SenderIndex: obs.senderIndex, PeerMachine: r.machine, PeerIndex: firstObservedIndex(obs.bits)} + SenderIndex: obs.senderIndex, PeerMachine: r.machine, PeerIndex: firstObservedAtomicIndex(obs.bits)} if obs.sentAt > 0 { message.SentAt = time.Unix(0, obs.sentAt).UTC() } @@ -256,6 +269,27 @@ func firstObservedIndex(words []uint64) int { return 0 } +func firstObservedAtomicIndex(words []atomic.Uint64) int { + for wordIndex := range words { + if word := words[wordIndex].Load(); word != 0 { + return wordIndex*64 + bits.TrailingZeros64(word) + 1 + } + } + return 0 +} + +func markObservationBit(words []atomic.Uint64, word int, mask uint64) bool { + for { + current := words[word].Load() + if current&mask != 0 { + return false + } + if words[word].CompareAndSwap(current, current|mask) { + return true + } + } +} + func (r *deliveryRecorder) onDisconnect(account AccountRecord, err error) { category := disconnectCategory(err) r.disconnectMu.Lock() @@ -370,7 +404,7 @@ func (r *deliveryRecorder) summarize(base AgentSummary, ledger []AcceptedMessage shard.mu.RLock() for mid, obs := range shard.observed { if _, ok := accepted[mid]; !ok { - base.UnexpectedMessages += bitsetCount(obs.bits) + base.UnexpectedMessages += atomicBitsetCount(obs.bits) } } shard.mu.RUnlock() @@ -419,17 +453,21 @@ func (r *deliveryRecorder) shard(mid string) *observationShard { return &r.shards[hash%uint32(len(r.shards))] } -func (r *deliveryRecorder) observationSnapshot(mid string) *observation { +func (r *deliveryRecorder) observationSnapshot(mid string) *observationView { shard := r.shard(mid) shard.mu.RLock() obs := shard.observed[mid] - if obs != nil { - copyOfObservation := *obs - copyOfObservation.bits = append([]uint64(nil), obs.bits...) - obs = ©OfObservation - } shard.mu.RUnlock() - return obs + if obs == nil { + return nil + } + view := &observationView{topic: obs.topic, seq: obs.seq, senderMachine: obs.senderMachine, + senderIndex: obs.senderIndex, scenario: obs.scenario, sentAt: obs.sentAt, + bits: make([]uint64, len(obs.bits))} + for i := range obs.bits { + view.bits[i] = obs.bits[i].Load() + } + return view } func cloneCountMap(source map[string]int64) map[string]int64 { @@ -443,7 +481,7 @@ func cloneCountMap(source map[string]int64) map[string]int64 { return clone } -func observationHas(obs *observation, receiverIndex int) bool { +func observationHas(obs *observationView, receiverIndex int) bool { if obs == nil || receiverIndex <= 0 { return false } @@ -452,6 +490,14 @@ func observationHas(obs *observation, receiverIndex int) bool { return word < len(obs.bits) && obs.bits[word]&(uint64(1)<