diff --git a/docs/test/IM-group2k-r10-双机群聊压测分析与改进-20261004.md b/docs/test/IM-group2k-r10-双机群聊压测分析与改进-20261004.md new file mode 100644 index 0000000..d6b8d72 --- /dev/null +++ b/docs/test/IM-group2k-r10-双机群聊压测分析与改进-20261004.md @@ -0,0 +1,159 @@ +# GROUP-2000 R10 双机群聊压测分析与改进 + +## 1. 测试结论 + +`group2k-r10-20261004-01` 的 A/B 同步有效，可以作为 2000 人群聊的真实双机容量结论。本轮持久化链路基本正常，但实时扇出已明显超过当前服务容量，整体判定为 **失败**。 + +- A/B 各建立 1000 个 WS 连接，共 2000/2000 就绪；两机实际开始时间相差约 84ms。 +- 2000 条发送尝试中，1984 条收到 202 ACK，16 条因 socket 中断而结果未知，无显式 403/502/503 拒绝。 +- 数据库最终找到 1999 条，无重复、无字段不一致；仅 `g-a-0098-02` 未落库。 +- 按 1985 条已在服务端确认的消息计算，期望实时投递 3,970,000 次，实际收到 1,979,444 次，到达率 49.86%，实时缺失率 50.14%。 +- 1,990,556 次实时缺失全部可从历史消息查回，说明主要问题是实时投递容量，不是 Kafka/MySQL 丢数据。 +- 负载期间共 152 次 WS 断连：147 次 connection reset、4 次 timeout、1 次 closed connection。 +- ACK P95 为 71.18s，本地实时投递 P95 为 329.68s，已不是普通抖动，而是持续性排队。 + +JPush 空设备查询放大、MySQL 连接频繁回收和 A/B 起跑错位等 R9 问题在本轮已解决。当前的主要容量瓶颈是大群每条消息的单 Topic 有序处理、跨 Pod multiplex 转发和各 Pod WS 逐连接序列化/写出组合成的实时扇出链路。 + +## 2. 四层对账 + +| 层级 | 结果 | 结论 | +| --- | ---: | --- | +| 发送尝试 | 2000 | A/B 各 1000 | +| 即时收到 202 | 1733 | 251 条在超时后迟到 | +| 最终收到 202 | 1984 | 发送成功率 99.2% | +| 发送结果未知 | 16 | 全部为 socket 中断，无业务拒绝 | +| 历史回查 | 1985/1985 | 服务端可确认消息全部可查 | +| 数据库实际落库 | 1999/2000 | 14 条“结果未知”也已落库 | +| 真正未落库 | 1 | A 机 `g-a-0098-02` | +| 实时投递 | 1,979,444/3,970,000 | 到达率 49.86% | +| 重复 / 乱序 | 0 / 0 | 已收到数据的正确性正常 | + +16 条结果未知中，14 条 B 机消息已在数据库找到；A 机 `g-a-0102-02` 被实时观测到且已落库；只有 `g-a-0098-02` 在连接断开前未被服务端处理。因此不能把“客户端未收到 ACK”直接等同于“消息未落库”。 + +## 3. A/B 表现 + +| 指标 | 机器 A | 机器 B | +| --- | ---: | ---: | +| 就绪连接 | 1000 | 1000 | +| 发送尝试 | 1000 | 1000 | +| 最终观测 ACK | 998 | 986 | +| 结果未知 | 2 | 14 | +| 收到群消息 | 996,563 | 982,881 | +| 断连 | 7 | 145 | + +B 机的断连明显更多，但 reset 分布在多个公网源 IP，没有单一坏 IP 的证据。机器差异会影响结果，但 A 机同样出现大量投递缺失和长尾，不能将本轮失败只归因于 B 机网络。 + +## 4. 指标与时间线 + +- 负载窗口约为 09:21:43–09:26:32。 +- A 机断连集中在 09:23:20–09:25:24；B 机采样到的 reset 集中在 09:24:20–09:25:23。 +- 服务端 `/metrics` 在约 09:23:01–09:25:37 多次超过 10s 采集超时；233 次 im-server 采样中 14 次失败，17 次超过 5s。 +- Kafka consumer lag 和 oldest age 在收敛阶段均为 0。 +- `im_kafka_message_persist_total{kind="group",result="batch_ok"}` 合计 1999，与数据库实际条数一致。 +- `im_push_dispatch_queue_enqueue_total{result="ok"}` 为 1999，未观测到 push queue full/error；推送平均处理约 22ms。 +- Topic group queue wait 未超过 1s，Cluster P2M/route 计数未观测到 full/error。 +- 三个 Pod 的 ACK enqueue/write 指标都在 1s 内。注意该计时从服务端 WS 读循环完成解析后开始，因此客户端几十秒至数分钟的 ACK 延迟，主要发生在服务端 dispatch 之前的读取/调度阶段，或写入 socket 后的网络/客户端接收阶段。 + +`/metrics` 与 WS 同时出现长时间停顿，说明不只是某个业务队列满，而是 Pod 的 CPU/调度/网络写路径在大规模扇出下被长时间占用。本轮无法通过集群 API 读取 Pod Event 和完整日志，因此不将 CPU throttling、Ingress 和节点网卡中的任何一项单独定为唯一根因。 + +## 5. 代码链路分析 + +### 5.1 Topic 主节点 + +`Topic.broadcastToSessions` 遍历主 Topic 上的本地 session 和远程 Pod multiplex session，为每个目标复制、整理消息并入队。一个 2000 人群的 2000 条消息会生成约 400 万次终端投递。 + +当 multiplex 队列暂时满时，Topic goroutine 会等待最多 250ms；一个主 Topic 又需要保持消息顺序，所以无法通过对单条消息无界并发来解决。 + +### 5.2 跨 Pod multiplex + +每个远程 Pod 使用一个 multiplex session。它的 drain 为保证 Topic 顺序，对每条 `Cluster.TopicProxy` RPC 同步等待远程返回。这保证了顺序，但对单条 RPC 往返、gob 编解码和远程 proxy Topic 处理的抖动很敏感。 + +### 5.3 Pod 内 WS 扇出 + +proxy Topic 收到一条群消息后，仍然遍历本 Pod 上的所有订阅 session，为每个 WS 连接复制消息并入队。每个 WS 只有单个 write loop 串行做序列化和 socket write。当扇出速度超过写出速度，session 队列积压，随后出现连接 reset、客户端重连以及更大的订阅恢复压力。 + +## 6. 已排除和未排除项 + +### 已排除为本轮主因 + +- Kafka 积压或持久化丢失：最终 lag=0，1999 条 batch_ok 与 DB 一致。 +- MySQL 批量落库能力：无重复、无字段不一致，只有 1 条在服务端未接收。 +- JPush 设备查询放大：1999 次异步入队成功，无队列满，平均耗时低。 +- A/B 起跑错位：本轮最终屏障通过，两机实际开始差距小于 100ms。 +- 显式限流或内部队列拒绝：没有 429/502/503，未观测到 Cluster、Topic、push 队列 full/error。 + +### 仍需补齐证据 + +- 每个 Topic 单条消息的完整处理耗时和扇出目标数。 +- multiplex 队列等待、RPC 耗时、远程 proxy 接收到入队完成的分段耗时。 +- WS send queue 深度、队列等待 P95/P99、full/purge 和按 Pod 的发生时间。 +- Pod CPU usage/throttling、Go scheduler latency、GC pause、goroutine 数、网卡吞吐/丢包/retransmit。 +- Ingress/LB 上游 reset、upstream response time、连接数和每 Pod 连接分布。 + +## 7. 改进方案 + +### P0：先定位每段长尾 + +1. 增加 Topic 处理总耗时、单条扇出目标数和 `broadcastToSessions` 耗时直方图，区分 master/proxy。 +2. 增加 multiplex 入队等待、队列深度、`Cluster.TopicProxy` RPC 耗时和远程处理耗时。 +3. 补齐 WS send queue P95/P99/max、full、purged，并区分 data/ctrl。ACK 必须能从大量 data 帧中被单独观测。 +4. 增加 WS 读取到 dispatch 前的时间戳，避免现有 ACK 指标忽略服务端读循环前的停顿。 +5. 恢复压测期间的 Pod/Event/CPU 可观测性，指标采集超时不应导致整段监控空白。 + +### P1：降低实时扇出成本 + +1. 将 master 到每个 proxy Pod 的多条消息改为有界批量 RPC，批内保持 SeqId 顺序，减少 2000 次同步 RPC 往返、gob 编码和调度开销。 +2. 为每个 Topic-Pod 建立单写者有界队列，允许不同 Pod 并行处理，同一 Pod 内严格按 SeqId 投递；不对单 Topic 做无界 goroutine 并发。 +3. 研究一次序列化、多 session 复用字节的可行性，避免同一条群消息在每个连接重复做等价 JSON 序列化。用户相关 Topic/权限字段必须先完成安全性分类，不可盲目全局复用。 +4. ctrl/ACK 与普通 data 帧分级调度，保证发送者在大群扇出积压时仍能快速获得明确结果。需保持单连接协议语义，不得导致客户端对 data/ctrl 的依赖关系错乱。 +5. 对慢消费 WS 定义明确策略：队列达阈值后记录可追踪原因、主动断开并依靠历史补拉，避免一批慢连接长时间拖住整个 Topic。 + +## 8. 实施和复测顺序 + +1. 先完成 P0 指标，用相同 2000 人、2000 条参数复现一轮，定量确认 Topic、multiplex RPC、WS queue 和 socket write 各占多少。 +2. 实施 multiplex 批量化和有界单写者，先做单元测试、顺序测试、队列满/节点断开/重连测试。 +3. 用 500、1000、1500、2000 人阶梯复测，每档至少两轮，不直接只跑最高档。 +4. 最后重复 R10 同参数双机测试，必须同时通过实时、历史、DB 和基础设施门禁。 + +## 9. 下一轮通过门禁 + +- 2000/2000 WS 就绪，A/B 实际开始差小于 1s。 +- 2000 条发送全部获得明确 ACK，无 outcome unknown，无 5xx。 +- 数据库精确找到 2000 条，缺失、重复、字段不一致均为 0，Kafka lag 最终归零。 +- 实时投递到达率不低于 99.99%，重复和乱序为 0。 +- ACK P95 小于 2s，投递 P95 小于 5s；本轮已明确不以 500ms 作为阻断门禁。 +- 断连、Topic/multiplex/WS queue full、Pod restart、OOM、liveness failure 均为 0。 +- 负载窗口内 `/metrics` 可持续采集，无连续 10s 超时；CPU throttling、GC、网络 retransmit 和 Ingress reset 有完整记录。 + +## 10. 产物 + +- 结果目录：`loadtest/im-pressure/results/group2k-r10-20261004-01/` +- 汇总报告：`report.json` +- 数据库核验：`db-verification.json` +- A/B 明细：`agent-a-summary.json`、`agent-b-summary.json` +- coordinator 台账：`coordinator.json` +- 服务指标：`metrics.jsonl` +- 代码版本：`c46209f` + +## 11. 本轮已实施改造 + +R10 分析后已完成以下代码改造，待发布后进行 R11 复测： + +- 普通非频道群消息在每个 Pod 仅执行一次 JSON 序列化，同 Pod 的所有 WS 连接复用同一份不可变字节。权限过滤、`SkipSid`、频道匿名、P2P 别名和非 WS 协议仍保持原路径。 +- 增加 Topic 完整处理耗时、广播耗时和按 websocket/multiplex/other 分类的扇出目标数。 +- 增加 master-to-proxy multiplex 入队深度和排队时间，以及远程 proxy Topic 接收后的队列等待时间。 +- 增加 WS 完成读帧后的 JSON 解析和本地 dispatch 耗时。 +- 新快路仍在 WS write loop 记录原有 `OutgoingMessageSize` 统计，不会因预序列化丢失原有监控口径。 + +本次没有直接引入新的批量 `net/rpc` 方法。StatefulSet 滚动发布时会短暂出现新旧 Pod 混合，新 RPC 端点与旧 Pod 不兼容，会产生发布窗口风险。R11 先用新增指标确认 multiplex RPC 在优化后是否仍为主瓶颈；若需要批量化，应采用“先发布双协议接收端，再开启批量发送端”的两阶段发布。 + +新增指标： + +- `im_topic_process_duration_seconds` +- `im_topic_broadcast_duration_seconds` +- `im_topic_broadcast_targets` +- `im_topic_shared_serialization_total` +- `im_cluster_m2p_queue_wait_duration_seconds` +- `im_cluster_m2p_queue_depth` +- `im_cluster_proxy_queue_wait_duration_seconds` +- `im_ws_read_dispatch_duration_seconds` diff --git a/server/cluster.go b/server/cluster.go index 2247540..7a20314 100644 --- a/server/cluster.go +++ b/server/cluster.go @@ -284,6 +284,9 @@ type ClusterResp struct { // Original request type. OrigReqType ProxyReqType + + // Set after RPC decoding on the receiving node; not serialized by gob. + proxyQueuedAt time.Time } // ClusterPing is used to detect node restarts. @@ -743,6 +746,7 @@ func (Cluster) TopicProxy(msg *ClusterResp, unused *bool) error { // Find appropriate topic, send the message to it. if t := globals.hub.topicGet(types.TopicKey{TenantID: msg.TenantID, Topic: msg.RcptTo}); t != nil { msg.SrvMsg.uid = types.ParseUserId(msg.SrvMsg.AsUser) + msg.proxyQueuedAt = time.Now() select { case t.proxy <- msg: immetrics.ObserveInternalQueueEnqueue("cluster_topic_proxy", len(t.proxy), cap(t.proxy), true) @@ -1420,6 +1424,8 @@ func (sess *Session) clusterWriteLoop(forTopic string) { return } srvMsg := msg.(*ServerComMessage) + immetrics.ObserveClusterM2PDequeue(sess.clnode.name, srvMsg.clusterQueuedAt) + srvMsg.clusterQueuedAt = time.Time{} response := &ClusterResp{TenantID: srvMsg.TenantID, SrvMsg: srvMsg} if srvMsg.sess == nil { response.OrigSid = "*" diff --git a/server/datamodel.go b/server/datamodel.go index 63d1950..ebace6c 100644 --- a/server/datamodel.go +++ b/server/datamodel.go @@ -881,6 +881,8 @@ type ServerComMessage struct { PreserveTopicName bool `json:"-"` // User id affected by this message. uid types.Uid + // Timestamp when this message entered a master-to-proxy multiplex queue. + clusterQueuedAt time.Time } // Deep-shallow copy of ServerComMessage. Deep copy of service fields, @@ -899,6 +901,7 @@ func (src *ServerComMessage) copy() *ServerComMessage { SkipSid: src.SkipSid, PreserveTopicName: src.PreserveTopicName, uid: src.uid, + clusterQueuedAt: src.clusterQueuedAt, } dst.Ctrl = src.Ctrl.copy() diff --git a/server/hdl_websock.go b/server/hdl_websock.go index 3f878dc..3c69ca6 100644 --- a/server/hdl_websock.go +++ b/server/hdl_websock.go @@ -105,7 +105,9 @@ func (sess *Session) readLoop() { atomic.StoreInt64(&sess.lastAction, types.TimeNow().UnixNano()) _ = sess.ws.SetReadDeadline(time.Now().Add(pongWait)) statsInc("IncomingMessagesWebsockTotal", 1) + dispatchStarted := time.Now() sess.dispatchRaw(raw) + immetrics.ObserveWSReadDispatch(dispatchStarted) } } @@ -160,6 +162,11 @@ func (sess *Session) writeLoop() { if isAcceptedCtrl(v) { immetrics.ObserveCtrlACKDelivery("write", v.Timestamp) } + case sharedServerMessage: + statsAddHistSample("OutgoingMessageSize", float64(len(v))) + if !sess.sendMessage([]byte(v)) { + return + } default: // serialized message if !sess.sendMessage(v) { return diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index d21dd55..87aaa80 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -190,6 +190,45 @@ var ( Help: "Time a client publish waited in a topic queue before processing by topic kind.", Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}, }, []string{"pod", "kind"}) + TopicProcessDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_topic_process_duration_seconds", + Help: "End-to-end topic event processing duration by pod, topic kind, role, and operation.", + Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}, + }, []string{"pod", "kind", "role", "operation"}) + TopicBroadcastDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_topic_broadcast_duration_seconds", + Help: "Time spent preparing and enqueueing one topic broadcast by pod, topic kind, and role.", + Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, []string{"pod", "kind", "role"}) + TopicBroadcastTargets = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_topic_broadcast_targets", + Help: "Number of local WebSocket, multiplex, or other targets considered by one topic broadcast.", + Buckets: []float64{0, 1, 10, 50, 100, 250, 500, 750, 1000, 1500, 2000, 3000}, + }, []string{"pod", "kind", "role", "target_type"}) + TopicSharedSerializationTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "im_topic_shared_serialization_total", + Help: "Shared WebSocket serialization events by pod, topic kind, and result.", + }, []string{"pod", "kind", "result"}) + ClusterM2PQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_cluster_m2p_queue_wait_duration_seconds", + Help: "Time a master-to-proxy response waited in the multiplex outbound queue by pod and peer.", + Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10}, + }, []string{"pod", "peer"}) + ClusterM2PQueueDepth = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_cluster_m2p_queue_depth", + Help: "Master-to-proxy multiplex outbound queue depth observed after enqueue by pod and peer.", + Buckets: []float64{0, 1, 4, 8, 16, 32, 64, 96, 128, 160}, + }, []string{"pod", "peer"}) + ClusterProxyQueueWaitDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_cluster_proxy_queue_wait_duration_seconds", + Help: "Time a cluster response waited in the proxy topic queue by pod and message type.", + Buckets: []float64{0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60}, + }, []string{"pod", "message_type"}) + WSReadDispatchDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "im_ws_read_dispatch_duration_seconds", + Help: "Time from a completed WebSocket frame read through JSON decoding and local dispatch.", + Buckets: []float64{0.0001, 0.0005, 0.001, 0.003, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5}, + }, []string{"pod"}) PushDispatchQueueEnqueueTotal = prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "im_push_dispatch_queue_enqueue_total", Help: "Asynchronous push dispatch queue enqueue attempts by pod and result.", @@ -242,6 +281,8 @@ func init() { ClusterP2MQueueCapacity, ClusterP2MEnqueueTotal, ClusterRPCInflight, ClusterRPCDuration, ClusterRouteTotal, ClusterProxyResponseTotal, InternalQueueEnqueueTotal, InternalQueueDepth, InternalQueueCapacity, InternalQueueCurrent, MessagePersistDuration, MessagePublishStageDuration, TopicQueueWaitDuration, + TopicProcessDuration, TopicBroadcastDuration, TopicBroadcastTargets, TopicSharedSerializationTotal, + ClusterM2PQueueWaitDuration, ClusterM2PQueueDepth, ClusterProxyQueueWaitDuration, WSReadDispatchDuration, PushDispatchQueueEnqueueTotal, PushDispatchQueueDepth, PushDispatchQueueWaitDuration, PushDispatchDuration, PushRecipientCount, KafkaMessagePersistTotal) @@ -279,6 +320,47 @@ func ObserveTopicQueueWait(kind string, queuedAt time.Time) { TopicQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(kind)).Observe(time.Since(queuedAt).Seconds()) } +func ObserveTopicProcess(kind, role, operation string, started time.Time) { + TopicProcessDuration.WithLabelValues(metricPodName(), normalizeLabel(kind), normalizeLabel(role), + normalizeLabel(operation)).Observe(time.Since(started).Seconds()) +} + +func ObserveTopicBroadcast(kind, role string, started time.Time, websocketTargets, multiplexTargets, otherTargets int) { + pod := metricPodName() + kind = normalizeLabel(kind) + role = normalizeLabel(role) + TopicBroadcastDuration.WithLabelValues(pod, kind, role).Observe(time.Since(started).Seconds()) + TopicBroadcastTargets.WithLabelValues(pod, kind, role, "websocket").Observe(float64(websocketTargets)) + TopicBroadcastTargets.WithLabelValues(pod, kind, role, "multiplex").Observe(float64(multiplexTargets)) + TopicBroadcastTargets.WithLabelValues(pod, kind, role, "other").Observe(float64(otherTargets)) +} + +func IncTopicSharedSerialization(kind, result string) { + TopicSharedSerializationTotal.WithLabelValues(metricPodName(), normalizeLabel(kind), normalizeLabel(result)).Inc() +} + +func ObserveClusterM2PEnqueue(peer string, depth int) { + ClusterM2PQueueDepth.WithLabelValues(metricPodName(), normalizeLabel(peer)).Observe(float64(depth)) +} + +func ObserveClusterM2PDequeue(peer string, queuedAt time.Time) { + if queuedAt.IsZero() { + return + } + ClusterM2PQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(peer)).Observe(time.Since(queuedAt).Seconds()) +} + +func ObserveClusterProxyQueueWait(messageType string, queuedAt time.Time) { + if queuedAt.IsZero() { + return + } + ClusterProxyQueueWaitDuration.WithLabelValues(metricPodName(), normalizeLabel(messageType)).Observe(time.Since(queuedAt).Seconds()) +} + +func ObserveWSReadDispatch(started time.Time) { + WSReadDispatchDuration.WithLabelValues(metricPodName()).Observe(time.Since(started).Seconds()) +} + func ObservePushDispatchEnqueue(depth int, accepted bool) { result := "ok" if !accepted { diff --git a/server/monitoring_http_test.go b/server/monitoring_http_test.go index 2996a80..9dab5a2 100644 --- a/server/monitoring_http_test.go +++ b/server/monitoring_http_test.go @@ -98,6 +98,13 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { immetrics.ObserveWSSendQueueDequeue(time.Now(), false) immetrics.ObserveCtrlACKDelivery("write", time.Now()) immetrics.ObserveTopicQueueWait("group", time.Now()) + immetrics.ObserveTopicProcess("group", "master", "publish", time.Now()) + immetrics.ObserveTopicBroadcast("group", "master", time.Now(), 1, 0, 0) + immetrics.IncTopicSharedSerialization("group", "serialized") + immetrics.ObserveClusterM2PEnqueue("peer", 1) + immetrics.ObserveClusterM2PDequeue("peer", time.Now()) + immetrics.ObserveClusterProxyQueueWait("data", time.Now()) + immetrics.ObserveWSReadDispatch(time.Now()) immetrics.ObservePushDispatchEnqueue(1, true) immetrics.ObservePushDispatch(time.Now(), 1, time.Now(), nil) wrt := httptest.NewRecorder() @@ -117,7 +124,11 @@ func TestServeMetricsIncludesWSConnections(t *testing.T) { "im_ctrl_ack_delivery_duration_seconds", "im_cluster_p2m_enqueue_total", "im_cluster_rpc_duration_seconds", "im_cluster_route_total", "im_internal_queue_enqueue_total", "im_internal_queue_depth", "im_internal_queue_capacity", "im_internal_queue_current", - "im_topic_queue_wait_duration_seconds", "im_push_dispatch_queue_enqueue_total", + "im_topic_queue_wait_duration_seconds", "im_topic_process_duration_seconds", + "im_topic_broadcast_duration_seconds", "im_topic_broadcast_targets", "im_topic_shared_serialization_total", + "im_cluster_m2p_queue_wait_duration_seconds", "im_cluster_m2p_queue_depth", + "im_cluster_proxy_queue_wait_duration_seconds", "im_ws_read_dispatch_duration_seconds", + "im_push_dispatch_queue_enqueue_total", "im_push_dispatch_queue_depth", "im_push_dispatch_queue_wait_duration_seconds", "im_push_dispatch_duration_seconds", "im_push_recipient_count", "im_kafka_message_persist_total", "im_kafka_consumer_lag", "im_kafka_dead_letters_total"} { diff --git a/server/session.go b/server/session.go index 28d2ccc..298f138 100644 --- a/server/session.go +++ b/server/session.go @@ -195,6 +195,9 @@ type Subscription struct { supd chan<- *sessionUpdate } +// sharedServerMessage is an immutable JSON payload reused by multiple WebSocket sessions. +type sharedServerMessage []byte + func (s *Session) enqueueOutbound(payload any) bool { if s.isMultiplex() { return s.enqueueMultiplexOutbound(payload, multiplexEnqueueWait) @@ -234,8 +237,16 @@ func (s *Session) enqueueMultiplexOutbound(payload any, wait time.Duration) bool s.clusterQueueMu.Unlock() return true } + if msg, ok := payload.(*ServerComMessage); ok { + msg.clusterQueuedAt = time.Now() + } select { case s.send <- payload: + peer := "unknown" + if s.clnode != nil { + peer = s.clnode.name + } + immetrics.ObserveClusterM2PEnqueue(peer, len(s.send)) s.clusterQueueMu.Unlock() return true default: @@ -525,6 +536,17 @@ func (s *Session) queueOutBytes(data []byte) bool { return true } +func (s *Session) queueOutSharedBytes(data []byte) bool { + if s == nil || atomic.LoadInt32(&s.terminating) > 0 { + return true + } + if !s.enqueueOutbound(sharedServerMessage(data)) { + logs.Err.Println("s.queueOutSharedBytes: session's send queue full", s.sid) + return false + } + return true +} + func (s *Session) maybeScheduleClusterWriteLoop() { if s.multi != nil { s.multi.scheduleClusterWriteLoop() diff --git a/server/topic.go b/server/topic.go index b394c5b..9a4ae0e 100644 --- a/server/topic.go +++ b/server/topic.go @@ -650,12 +650,20 @@ func (t *Topic) runLocal(hub *Hub) { // handleClientMsg is the top-level handler of messages received by the topic from sessions. func (t *Topic) handleClientMsg(msg *ClientComMessage) { + started := time.Now() kind := "other" if t.cat == types.TopicCatP2P { kind = "p2p" } else if t.cat == types.TopicCatGrp { kind = "group" } + operation := "other" + if msg.Pub != nil { + operation = "publish" + } else if msg.Note != nil { + operation = "note" + } + defer immetrics.ObserveTopicProcess(kind, t.metricRole(), operation, started) immetrics.ObserveTopicQueueWait(kind, msg.queuedAt) msg.queuedAt = time.Time{} if msg.TenantID.IsZero() { @@ -1958,9 +1966,18 @@ func (t *Topic) handlePresence(msg *ServerComMessage) { // broadcastToSessions writes message to attached sessions. func (t *Topic) broadcastToSessions(msg *ServerComMessage) { + started := time.Now() + kind := t.metricKind() + role := t.metricRole() + websocketTargets, multiplexTargets, otherTargets := 0, 0, 0 + defer func() { + immetrics.ObserveTopicBroadcast(kind, role, started, websocketTargets, multiplexTargets, otherTargets) + }() msg.TenantID = t.tenantID // List of sessions to be dropped. var dropSessions []*Session + var sharedPayload []byte + sharedPayloadReady := false // Broadcast the message. Only {data}, {pres}, {info} are broadcastable. // {meta} and {ctrl} are sent to the session only for sess, pssd := range t.sessions { @@ -2027,7 +2044,41 @@ func (t *Topic) broadcastToSessions(msg *ServerComMessage) { } } - // Make a copy of msg since messages sent to sessions differ. + if sess.proto == WEBSOCK { + websocketTargets++ + } else if sess.isMultiplex() { + multiplexTargets++ + } else { + otherTargets++ + } + + // Ordinary group messages are identical for every local WebSocket subscriber. + // Serialize once and reuse immutable bytes to avoid O(recipients) JSON work. + if t.canShareSerializedBroadcast(msg, sess, pssd) { + if !sharedPayloadReady { + var err error + sharedPayload, err = json.Marshal(msg) + if err != nil { + immetrics.IncTopicSharedSerialization(kind, "error") + logs.Warn.Printf("topic[%s]: shared broadcast serialization failed: %v", t.name, err) + sharedPayload = nil + } else { + immetrics.IncTopicSharedSerialization(kind, "serialized") + } + sharedPayloadReady = true + } + if sharedPayload != nil { + if sess.queueOutSharedBytes(sharedPayload) { + continue + } + immetrics.IncTopicSharedSerialization(kind, "full") + logs.Warn.Printf("topic[%s]: connection stuck, detaching - %s", t.name, sess.sid) + dropSessions = append(dropSessions, sess) + continue + } + } + + // Make a copy of msg since messages sent to sessions may differ. msgCopy := msg.copy() // Topic name may be different depending on the user to which the `sess` belongs. t.prepareBroadcastableMessage(msgCopy, pssd.uid, pssd.isChanSub) @@ -2046,6 +2097,48 @@ func (t *Topic) broadcastToSessions(msg *ServerComMessage) { } } +func (t *Topic) metricKind() string { + switch t.cat { + case types.TopicCatGrp: + return "group" + case types.TopicCatP2P: + return "p2p" + default: + return "other" + } +} + +func (t *Topic) metricRole() string { + if t.isProxy { + return "proxy" + } + return "master" +} + +func serverMessageType(msg *ServerComMessage) string { + switch { + case msg == nil: + return "unknown" + case msg.Data != nil: + return "data" + case msg.Ctrl != nil: + return "ctrl" + case msg.Pres != nil: + return "pres" + case msg.Info != nil: + return "info" + case msg.Meta != nil: + return "meta" + default: + return "unknown" + } +} + +func (t *Topic) canShareSerializedBroadcast(msg *ServerComMessage, sess *Session, pssd perSessionData) bool { + return t.cat == types.TopicCatGrp && !t.isChan && sess.proto == WEBSOCK && !pssd.isChanSub && + (msg.Data != nil || msg.Pres != nil || msg.Info != nil) +} + // subscriptionReply generates a response to a subscription request func (t *Topic) subscriptionReply(asChan bool, msg *ClientComMessage) error { // The topic is already initialized by the Hub diff --git a/server/topic_proxy.go b/server/topic_proxy.go index 9ec7b43..d80696e 100644 --- a/server/topic_proxy.go +++ b/server/topic_proxy.go @@ -213,6 +213,7 @@ func (t *Topic) handleProxyLeaveRequest(msg *ClientComMessage, killTimer *time.T // proxyMasterResponse at proxy topic processes a master topic response to an earlier request. func (t *Topic) proxyMasterResponse(msg *ClusterResp, killTimer *time.Timer) { + immetrics.ObserveClusterProxyQueueWait(serverMessageType(msg.SrvMsg), msg.proxyQueuedAt) // Kills topic after a period of inactivity. keepAlive := idleProxyTopicTimeout diff --git a/server/topic_test.go b/server/topic_test.go index 0412cfd..60567c6 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -4334,3 +4334,75 @@ func TestMain(m *testing.M) { globals.maxSubscriberCount = 1_000_000_000 os.Exit(m.Run()) } + +func TestGroupBroadcastReusesSerializedWebSocketPayload(t *testing.T) { + uid1, uid2, uid3 := types.Uid(1), types.Uid(2), types.Uid(3) + sess1 := &Session{sid: "ws-1", proto: WEBSOCK, send: make(chan any, 2)} + sess2 := &Session{sid: "ws-2", proto: WEBSOCK, send: make(chan any, 2)} + sessNoRead := &Session{sid: "ws-no-read", proto: WEBSOCK, send: make(chan any, 2)} + topic := &Topic{ + tenantID: 1, + name: "grpShared", + xoriginal: "grpShared", + cat: types.TopicCatGrp, + sessions: map[*Session]perSessionData{ + sess1: {uid: uid1}, + sess2: {uid: uid2}, + sessNoRead: {uid: uid3}, + }, + perUser: map[types.Uid]perUserData{ + uid1: {modeWant: types.ModeCFull, modeGiven: types.ModeCFull}, + uid2: {modeWant: types.ModeCFull, modeGiven: types.ModeCFull}, + uid3: {modeWant: types.ModeNone, modeGiven: types.ModeNone}, + }, + } + msg := &ServerComMessage{Data: &MsgServerData{ + Topic: "grpShared", From: uid1.UserId(), SeqId: 42, Timestamp: time.Now(), Content: "hello", + }} + + topic.broadcastToSessions(msg) + + payload1, ok := (<-sess1.send).(sharedServerMessage) + if !ok { + t.Fatal("first WebSocket did not receive a serialized payload") + } + payload2, ok := (<-sess2.send).(sharedServerMessage) + if !ok { + t.Fatal("second WebSocket did not receive a serialized payload") + } + if len(payload1) == 0 || len(payload2) == 0 || &payload1[0] != &payload2[0] { + t.Fatal("group WebSocket recipients did not reuse the same serialized payload") + } + var decoded ServerComMessage + if err := json.Unmarshal(payload1, &decoded); err != nil { + t.Fatalf("invalid shared payload: %v", err) + } + if decoded.Data == nil || decoded.Data.Topic != "grpShared" || decoded.Data.SeqId != 42 || decoded.Data.Content != "hello" { + t.Fatalf("unexpected shared payload: %#v", decoded.Data) + } + if len(sessNoRead.send) != 0 { + t.Fatal("non-reader received a group broadcast") + } +} + +func TestSharedBroadcastEligibility(t *testing.T) { + sess := &Session{proto: WEBSOCK} + msg := &ServerComMessage{Data: &MsgServerData{Topic: "grpShared"}} + for _, tc := range []struct { + name string + topic Topic + pssd perSessionData + want bool + }{ + {name: "ordinary group", topic: Topic{cat: types.TopicCatGrp}, want: true}, + {name: "channel", topic: Topic{cat: types.TopicCatGrp, isChan: true}}, + {name: "channel subscription", topic: Topic{cat: types.TopicCatGrp}, pssd: perSessionData{isChanSub: true}}, + {name: "p2p", topic: Topic{cat: types.TopicCatP2P}}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := tc.topic.canShareSerializedBroadcast(msg, sess, tc.pssd); got != tc.want { + t.Fatalf("eligibility=%v want=%v", got, tc.want) + } + }) + } +}