refactor: 收口为hub/peer/bridge 三程序并统一 支持 tcp/udp/kcp"

This commit is contained in:
nnbcccscdscdsc
2026-03-17 16:28:35 +08:00
parent 6c975d9ae3
commit 20b2050706
17 changed files with 2766 additions and 6398 deletions

1
.gitignore vendored
View File

@@ -25,3 +25,4 @@ compile_commands.json
*.mp4
*.avi
*.dat
*.jsonl

View File

@@ -14,79 +14,46 @@ BUILD_DIR ?= build
CFLAGS ?= -O2 -g -Wall -Wextra -std=c11
CPPFLAGS ?= -D_DEFAULT_SOURCE -D_POSIX_C_SOURCE=200809L
LDFLAGS ?=
# 链接 pthreadapps 中有多线程)
# 链接 pthread。
LDLIBS ?= -lpthread
INCLUDES := -Iinclude
# 所有二进制共享的核心源码(网络核心 + 协议实现 + ikcp
COMMON_SRCS := \
src/core/network.c \
# hub / peer / bridge 共用的核心源码
PEER_STACK_SRCS := \
src/core/logger.c \
src/protocols/tcp_impl.c \
src/protocols/udp_impl.c \
src/protocols/kcp_impl.c \
src/core/peer_transport.c \
src/protocols/ikcp.c
# 应用入口源文件。
APP_TEST_SRC := src/apps/test_main.c
APP_CLIENT_SRC := src/apps/client_main.c
APP_SERVER_SRC := src/apps/server_main.c
APP_RELAY_SRC := src/apps/relay_main.c
# 当前保留的应用入口源文件。
APP_HUB_SRC := src/apps/hub_main.c
APP_PEER_SRC := src/apps/peer_main.c
APP_BRIDGE_SRC := src/apps/bridge_main.c
# 将源文件映射到 BUILD_DIR 下的对象文件路径。
COMMON_OBJS := $(patsubst %.c,$(BUILD_DIR)/%.o,$(COMMON_SRCS))
TEST_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_TEST_SRC))
CLIENT_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_CLIENT_SRC))
SERVER_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_SERVER_SRC))
RELAY_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_RELAY_SRC))
PEER_STACK_OBJS := $(patsubst %.c,$(BUILD_DIR)/%.o,$(PEER_STACK_SRCS))
HUB_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_HUB_SRC))
PEER_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_PEER_SRC))
BRIDGE_OBJ := $(patsubst %.c,$(BUILD_DIR)/%.o,$(APP_BRIDGE_SRC))
# 按目标清理时,仅删除对应可执行文件与专属入口对象,避免影响其它产物
CLIENT_CLEAN_FILES := $(BUILD_DIR)/omni_client $(CLIENT_OBJ)
ARM64_BUILD_DIR ?= build/arm64
ARM64_CLIENT_OBJ := $(patsubst %.c,$(ARM64_BUILD_DIR)/%.o,$(APP_CLIENT_SRC))
ARM64_CLIENT_CLEAN_FILES := $(ARM64_BUILD_DIR)/omni_client $(ARM64_CLIENT_OBJ)
# 默认构建目标4 个可执行程序。
# 默认构建目标:仅保留 hub / peer / bridge
TARGETS := \
$(BUILD_DIR)/omni_test \
$(BUILD_DIR)/omni_client \
$(BUILD_DIR)/omni_server \
$(BUILD_DIR)/omni_relay \
$(BUILD_DIR)/omni_hub \
$(BUILD_DIR)/omni_peer \
$(BUILD_DIR)/omni_bridge
.PHONY: all arm arm64 clean clean-client clean-arm64-client help
.PHONY: all arm arm64 clean help
# 本机构建入口。
all: $(TARGETS)
# 各可执行程序链接规则。
$(BUILD_DIR)/omni_test: $(COMMON_OBJS) $(TEST_OBJ)
$(BUILD_DIR)/omni_hub: $(PEER_STACK_OBJS) $(HUB_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
$(BUILD_DIR)/omni_client: $(COMMON_OBJS) $(CLIENT_OBJ)
$(BUILD_DIR)/omni_peer: $(PEER_STACK_OBJS) $(PEER_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
$(BUILD_DIR)/omni_server: $(COMMON_OBJS) $(SERVER_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
$(BUILD_DIR)/omni_relay: $(COMMON_OBJS) $(RELAY_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
$(BUILD_DIR)/omni_hub: $(BUILD_DIR)/src/core/logger.o $(HUB_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
$(BUILD_DIR)/omni_peer: $(BUILD_DIR)/src/core/logger.o $(PEER_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
$(BUILD_DIR)/omni_bridge: $(BUILD_DIR)/src/core/logger.o $(BRIDGE_OBJ)
$(BUILD_DIR)/omni_bridge: $(PEER_STACK_OBJS) $(BRIDGE_OBJ)
$(CC) $(CFLAGS) $(LDFLAGS) -o $@ $^ $(LDLIBS)
# 通用编译规则:
@@ -104,14 +71,6 @@ arm:
arm64:
$(MAKE) BUILD_DIR=build/arm64 CC=$(ARM64_CC) all
# 仅清理当前 BUILD_DIR 下的 omni_client 与其入口对象。
clean-client:
rm -f $(CLIENT_CLEAN_FILES)
# 仅清理 build/arm64 下的 omni_client 与其入口对象。
clean-arm64-client:
rm -f $(ARM64_CLIENT_CLEAN_FILES)
# 清理构建目录。
clean:
rm -rf build
@@ -121,7 +80,5 @@ help:
@echo "make -> build native binaries in build/"
@echo "make arm -> build ARM binaries in build/arm (arm-linux-gnueabihf-gcc)"
@echo "make arm64 -> build ARM64 binaries in build/arm64 (aarch64-linux-gnu-gcc)"
@echo "generated binaries: omni_test omni_client omni_server omni_relay omni_hub omni_peer omni_bridge"
@echo "make clean-client -> remove omni_client and client_main.o in BUILD_DIR"
@echo "make clean-arm64-client -> remove omni_client and client_main.o in build/arm64"
@echo "generated binaries: omni_hub omni_peer omni_bridge"
@echo "make clean -> remove build artifacts"

435
README.md
View File

@@ -1,274 +1,213 @@
# OmniSocket
统一的 TCP / UDP / KCP 传输框架,包含
- 协议抽象层(`omni_init / omni_send / omni_recv`
- 客户端:文件分片发送 + 异步接收服务端 ASCII 指令
- 服务端:接收并写文件 + 交互输入指令下发客户端
- 转发器A->B 中转,支持运行时动态修改目标端口
- Hub云端多客户端注册/绑定/路由,支持 A ↔ C ↔ B 命令透传
- Peer主动连接 Hub 的对等端,支持 `register / bind / send / say`
- 启动前时钟同步:客户端先测 `RTT + offset`,服务端据此输出补偿后的端到端时延
OmniSocket 当前包含 3 个核心程序
- `omni_peer`
- `omni_hub`
- `omni_bridge`
## 目录结构
```text
OmniSocket/
├── include/
│ ├── common.h # MsgHeader(type,len,timestamp)、消息类型、通用宏
│ ├── network.h # 统一协议接口定义
│ ├── kcp/ikcp.h # KCP 头文件
│ └── logger.h # 日志与统计接口
├── src/
│ ├── protocols/
│ │ ├── tcp_impl.c # TCP 实现16字节头 + 粘包拆包)
│ │ ├── udp_impl.c # UDP 实现sendto/recvfrom
│ │ ├── kcp_impl.c # KCP 实现(基于 UDP + ikcp
│ │ └── ikcp.c # KCP 源码
│ ├── core/
│ │ ├── network.c # 协议工厂分发
│ │ └── logger.c # 性能统计日志
│ └── apps/
│ ├── client_main.c # 客户端入口
│ ├── hub_main.c # Hub 入口(多客户端注册/路由)
│ ├── peer_main.c # Peer 入口(连接 Hub 的对等端)
│ ├── server_main.c # 服务端入口
│ ├── relay_main.c # 转发器入口
│ └── test_main.c # 简易协议连通性测试
├── scripts/
│ └── local_smoke_test.sh # 本机一键 smoke 测试
├── build/ # 编译产物目录
├── Makefile
└── README.md
```
三者统一支持 `tcp | udp | kcp` 三种传输协议。
## 构建
### 本机构建
本地构建:
```bash
make
```
生成:
- `build/omni_client`
- `build/omni_hub`
生成文件
- `build/omni_peer`
- `build/omni_server`
- `build/omni_relay`
- `build/omni_test`
- `build/omni_hub`
- `build/omni_bridge`
### ARM 交叉编译
默认使用 `arm-linux-gnueabihf-gcc`
Jetson 场景常用 ARM64 交叉编译
```bash
make arm
make arm64
```
生成`build/arm/` 目录。
生成文件:
- `build/arm64/omni_peer`
- `build/arm64/omni_hub`
- `build/arm64/omni_bridge`
## 程序参数
### `omni_server`
```bash
build/omni_server -p tcp|udp|kcp -P <listen_port> -o <output_file> [-b <bind_ip>]
```
说明:
- 接收客户端发送的文件分片并写入 `output_file`
- 若在交互终端运行,可在标准输入输入 ASCII 文本并回发给客户端
- 输入 `quit` 可退出服务端交互循环
### `omni_client`
```bash
build/omni_client -p tcp|udp|kcp -H <server_ip> -P <server_port> -f <file> [-b <bind_port>] [-m <chunk_mtu>] [-w <wait_seconds|-1>]
```
说明:
- 读取 `file`,按 `chunk_mtu`(默认 1400分片发送
- 发送结束后额外发送 `FILE_END` 控制包
- 后台线程持续接收并打印服务端 ASCII 指令
- `-w -1` 表示常驻模式,直到手动 `Ctrl+C`
- 建连后会先自动发送 `TIME_SYNC_REQ/RESP/REPORT`,以最小 RTT 样本估算 `server_time - client_time`
- 若同步响应不可达(例如经过当前实现的单向 relay文件传输仍继续但服务端不会产出补偿后的 `end_to_end_delay_ms`
### `omni_relay`
```bash
build/omni_relay -p tcp|udp|kcp -L <listen_port> -H <target_ip> -P <target_port>
```
标准输入支持命令:
- `set <ip> <port>`:动态修改转发目标
- `show`:显示当前目标
- `quit`:退出 relay
### `omni_hub`
```bash
build/omni_hub -P <listen_port> [-b <bind_ip>] [-p tcp]
```
说明:
- 当前阶段只实现 TCP 控制面
- 多个 `omni_peer` 主动连接到 hub 后,先用 `client_id` 注册
- hub 维护 `client_id -> socket` 映射,并按 `dst_id` 转发 `PEER_TUNNEL`
## 程序说明
### `omni_peer`
`omni_peer` 支持两种工作模式:
- `hub` 模式:连接 `hub``bridge`
- `direct` 模式:`peer` 之间直接互连
常见用途:
- 发送文本命令
- 发送文件
- 接收文件
### `omni_hub`
`omni_hub` 是中心注册与转发节点,通常部署在公网服务器或中心网络位置。
主要职责:
- 维护 `client_id -> session`
- 转发 `peer` 之间的 `bind / tunnel / status`
### `omni_bridge`
`omni_bridge` 是桥接节点,用于将远端 `peer` 接入上游 `hub`
主要职责:
- 上游连接 `hub`
- 下游监听本地入口供 `peer` 接入
- 适用于 `A <-> C <-> D <-> B` 这类桥接链路
当前限制:
- 一个 `bridge` 仅支持一个下游 `peer`
- 下游 `peer``-i` 必须与 `bridge -i` 保持一致
- 上下游必须使用同一种协议,不支持协议转换
- 当前实现更接近“单下游、单身份桥接”,而不是通用多租户中继
## 角色说明
- `A`:本地电脑
- `B`Jetson
- `C`:公网 Hub 服务器
- `D`:公网 Bridge 服务器
下面示例中的 `<proto>` 可替换为 `tcp``udp``kcp`
说明:
- 以下 `omni_peer` 示例统一采用长连接交互模式
- 启动后连接会保持,不再使用“传输一次后自动退出”的一次性写法
- 文本和文件传输通过终端中的交互命令完成
- 若启动命令中使用了 `-m``-F`,程序会执行启动动作模式,而不是进入当前 README 使用的交互模式
## 常用参数与命令
| 分类 | 写法 | 含义 |
| --- | --- | --- |
| 启动参数 | `-i <client_id>` | 当前 `peer` 的逻辑身份。Hub 会根据这个 ID 记录 `client_id -> session` 映射,例如 `-i pc` 表示“我是 pc”`-i jetson` 表示“我是 jetson”。 |
| 启动参数 | `-b <peer_id>` | 启动后默认绑定的目标 `peer`。例如 `-b jetson` 表示后续直接输入 `send ...``put ...` 时,默认发给 `jetson`。 |
| 启动参数 | `-d <peer_id>` | 启动动作模式下的显式目标。通常与 `-m``-F` 配合使用,表示把启动时的那条消息或那个文件直接发给指定目标。 |
| 启动参数 | `-o <output_file>` | 本地接收文件时的落盘路径。收到文件后会写入当前机器上的这个路径,例如 `-o /tmp/from_pc.bin`。 |
| 交互命令 | `bind <peer_id>` | 将默认目标切换到指定 `peer`,后续 `send``put` 默认发给它。 |
| 交互命令 | `send <text>` | 向当前默认目标发送一条文本消息。 |
| 交互命令 | `say <peer_id> <text>` | 向指定 `peer` 发送一条文本消息,不修改当前默认目标。 |
| 交互命令 | `put <file>` | 将当前机器上的文件发送给当前默认目标。 |
| 交互命令 | `push <peer_id> <file>` | 将当前机器上的文件发送给指定 `peer`,不修改当前默认目标。 |
| 交互命令 | `show` | 显示当前本地状态,例如 `client_id`、当前绑定目标和输出路径。 |
| 交互命令 | `quit` | 退出当前 `omni_peer` 进程。 |
## 场景 1点对点直传
`A <-> B`
B 端监听:
```bash
build/omni_peer -H <hub_ip> -P <hub_port> -i <client_id> [-b <peer_id>] [-d <peer_id>] [-m <text>] [-w <wait_seconds|-1>]
./build/omni_peer -M direct -p <proto> -L 9001 -i jetson -o /tmp/from_pc.bin
```
A 端连接:
```bash
./build/omni_peer -M direct -p <proto> -H <B_IP> -P 9001 -i pc -b jetson -o /tmp/from_jetson.bin
```
连接建立后,可在 A 端输入:
```text
send start
put /tmp/input.bin
```
如需反向 `B -> A`,可在 B 端输入:
```text
put /path/to/file.bin
```
## 场景 2通过 Hub 中转
`A <-> C <-> B`
C 端启动 Hub
- C 维护着一张 `client_id -> session` 映射表,用于记录谁是 `pc`、谁是 `jetson`,并据此转发 `bind / tunnel / status`
```bash
./build/omni_hub -p <proto> -P 9002
```
B 端连接 Hub
```bash
./build/omni_peer -p <proto> -H <C_IP> -P 9002 -i jetson -o /tmp/from_pc.bin
```
A 端连接 Hub
```bash
./build/omni_peer -p <proto> -H <C_IP> -P 9002 -i pc -b jetson -o /tmp/from_jetson.bin
```
连接建立后,可在 A 端输入:
```text
send start
put /tmp/input.bin
```
如需反向 `B -> A`,可在 B 端输入:
```text
bind pc
put /path/to/file.bin
```
## 场景 3通过 Bridge 桥接
`A <-> C <-> D <-> B`
C 端启动 Hub
- C 仍然维护 `client_id -> session` 映射表A 以 `pc` 注册到 CBridge 以 `jetson` 这个逻辑身份注册到 C
```bash
./build/omni_hub -p <proto> -P 9003
```
D 端启动 Bridge
```bash
./build/omni_bridge -p <proto> -H <C_IP> -P 9003 -i jetson -L 9004
```
B 端连接 Bridge
```bash
./build/omni_peer -p <proto> -H <D_IP> -P 9004 -i jetson -o /tmp/from_pc.bin
```
A 端连接 Hub
```bash
./build/omni_peer -p <proto> -H <C_IP> -P 9003 -i pc -b jetson -o /tmp/from_jetson.bin
```
连接建立后,可在 A 端输入:
```text
send start
put /tmp/input.bin
```
如需反向 `B -> A`,可在 B 端输入:
```text
bind pc
put /path/to/file.bin
```
说明:
- `-i`:当前 peer 的逻辑 ID后续所有路由都依赖它不依赖私网 IP
- `-b`:启动后先请求绑定默认目标
- `-m`:启动后立即发一条命令;若同时给了 `-d`,则直接发给该目标
- `-w`one-shot 模式下等待若干秒再退出,`-1` 表示常驻
- 该场景已经实现 `A -> C -> D -> B``B -> D -> C -> A` 的桥接转发
- 但当前 `bridge` 仍是单下游、单身份模型,不是完整的多节点桥接网络
交互命令:
- `bind <peer_id>`:绑定默认目标
- `send <text>`:发给当前绑定目标
- `say <peer_id> <text>`:显式指定目标
- `show`:显示本地 `client_id / bound_peer`
- `quit`:退出
## 快速启动(本机)
先准备一个测试文件:
```bash
dd if=/dev/urandom of=/tmp/input.bin bs=1400 count=64
```
### TCP 直连2 个终端)
终端 1
```bash
build/omni_server -p tcp -P 9000 -o /tmp/out_tcp.bin
```
终端 2
```bash
build/omni_client -p tcp -H 127.0.0.1 -P 9000 -f /tmp/input.bin
```
校验:
```bash
cmp -s /tmp/input.bin /tmp/out_tcp.bin && echo OK || echo FAIL
```
日志观察:
- client / server 的 `summary` 日志会新增 `clock_sync_ok``clock_offset_ms``clock_sync_rtt_ms``clock_sync_samples`
- server 侧的 `end_to_end_avg_ms``clock_sync_ok=1` 时表示已经按 offset 补偿后的端到端时延
### UDP 直连2 个终端)
终端 1
```bash
build/omni_server -p udp -P 9001 -o /tmp/out_udp.bin
```
终端 2
```bash
build/omni_client -p udp -H 127.0.0.1 -P 9001 -f /tmp/input.bin
```
校验:
```bash
cmp -s /tmp/input.bin /tmp/out_udp.bin && echo OK || echo FAIL
```
### KCP 直连2 个终端)
终端 1
```bash
build/omni_server -p kcp -P 9002 -o /tmp/out_kcp.bin
```
终端 2
```bash
build/omni_client -p kcp -H 127.0.0.1 -P 9002 -f /tmp/input.bin
```
校验:
```bash
cmp -s /tmp/input.bin /tmp/out_kcp.bin && echo OK || echo FAIL
```
## Relay 场景示例3 个终端)
终端 1最终接收端 B
```bash
build/omni_server -p udp -P 9102 -o /tmp/out_relay.bin
```
终端 2relay
```bash
build/omni_relay -p udp -L 9101 -H 127.0.0.1 -P 9102
```
终端 3发送端 A
```bash
build/omni_client -p udp -H 127.0.0.1 -P 9101 -f /tmp/input.bin
```
relay 终端可输入:
```text
show
set 127.0.0.1 9103
```
## Hub / Peer 场景3 个终端)
终端 1cloud hub C
```bash
build/omni_hub -P 9200
```
终端 2peer B
```bash
build/omni_peer -H 127.0.0.1 -P 9200 -i beta
```
终端 3peer A
```bash
build/omni_peer -H 127.0.0.1 -P 9200 -i alpha
```
在 A 终端输入:
```text
bind beta
send hello-from-alpha
```
此时 B 终端会打印:
```text
[peer alpha -> beta] hello-from-alpha
```
本地一键 smoke
```bash
./scripts/local_peer_smoke_test.sh
```

View File

@@ -80,6 +80,13 @@ typedef struct OmniStats {
/* 初始化统计模块,在程序启动时调用一次 */
void logger_init(void);
/* 设置当前进程的日志上下文,便于 perf/jsonl 日志携带协议与节点信息。 */
void logger_set_context(const char *app,
const char *proto,
const char *mode,
const char *role,
const char *self_id);
/* 记录一次发送/接收 */
void logger_on_send(size_t bytes);
void logger_on_recv(size_t bytes);

59
include/peer_transport.h Normal file
View File

@@ -0,0 +1,59 @@
#ifndef OMNISOCKET_PEER_TRANSPORT_H
#define OMNISOCKET_PEER_TRANSPORT_H
#include "common.h"
#include "network.h"
#include <stddef.h>
#include <stdint.h>
typedef struct PeerTransport PeerTransport;
typedef struct PeerTransportSession PeerTransportSession;
typedef struct PeerTransportEvent {
int kind;
PeerTransportSession *session;
MsgHeader header;
} PeerTransportEvent;
enum {
PEER_TRANSPORT_EVENT_NONE = 0,
PEER_TRANSPORT_EVENT_MESSAGE = 1,
PEER_TRANSPORT_EVENT_CLOSED = 2
};
PeerTransport *peer_transport_open(OmniRole role,
OmniProtocol proto,
const char *bind_ip,
uint16_t bind_port,
const char *peer_ip,
uint16_t peer_port);
void peer_transport_close(PeerTransport *transport);
PeerTransportSession *peer_transport_default_session(PeerTransport *transport);
int peer_transport_send(PeerTransport *transport,
PeerTransportSession *session,
uint32_t type,
const void *payload,
uint32_t payload_len);
int peer_transport_next_event(PeerTransport *transport,
PeerTransportEvent *event,
uint8_t *payload_buf,
size_t payload_cap,
int timeout_ms);
void peer_transport_close_session(PeerTransport *transport,
PeerTransportSession *session);
const char *peer_transport_session_remote(const PeerTransportSession *session,
char *buf,
size_t buf_sz);
uint16_t peer_transport_session_remote_port(const PeerTransportSession *session);
const char *peer_transport_proto_name(OmniProtocol proto);
#endif

File diff suppressed because it is too large Load Diff

View File

@@ -1,122 +0,0 @@
#!/usr/bin/env bash
# 本机一键 smoke 测试:
# - test1: TCP 直连 client -> server 文件一致性
# - test2: UDP client -> relay -> server包含动态目标切换
set -euo pipefail
# 根目录与构建产物目录。
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
BUILD_DIR="$ROOT_DIR/build"
# 每次测试创建独立临时目录,避免互相污染。
TMP_DIR="$(mktemp -d /tmp/omnisocket-smoke.XXXXXX)"
# 随机选择一组端口,降低被系统中已有进程占用的概率。
BASE_PORT=$((20000 + (RANDOM % 20000)))
DIRECT_PORT="$BASE_PORT"
RELAY_PORT=$((BASE_PORT + 1))
SINK1_PORT=$((BASE_PORT + 2))
SINK2_PORT=$((BASE_PORT + 3))
# 记录后台进程 PID统一在 cleanup 中回收。
PIDS=()
cleanup() {
# 无论脚本成功/失败,都尽量回收子进程,避免残留占端口。
for pid in "${PIDS[@]:-}"; do
kill "$pid" 2>/dev/null || true
wait "$pid" 2>/dev/null || true
done
# 删除临时目录与中间文件。
rm -rf "$TMP_DIR"
}
trap cleanup EXIT
log() {
printf '[smoke] %s\n' "$1"
}
wait_with_timeout() {
# 轮询等待某个 PID 退出,超时返回非 0。
# 参数:
# $1 pid
# $2 timeout_s
local pid="$1"
local timeout_s="$2"
local i
for ((i = 0; i < timeout_s * 10; ++i)); do
if ! kill -0 "$pid" 2>/dev/null; then
wait "$pid" 2>/dev/null || true
return 0
fi
sleep 0.1
done
return 1
}
log "ports direct=$DIRECT_PORT relay=$RELAY_PORT sink1=$SINK1_PORT sink2=$SINK2_PORT"
log "building native binaries"
# 统一从干净状态构建。
make -C "$ROOT_DIR" clean all >/dev/null
# 测试输入与输出文件路径。
INPUT_FILE="$TMP_DIR/input.bin"
DIRECT_OUT="$TMP_DIR/direct_out.bin"
RELAY1_OUT="$TMP_DIR/relay_sink1.bin"
RELAY2_OUT="$TMP_DIR/relay_sink2.bin"
# 准备随机输入文件32 * 1400 = 44800 bytes
dd if=/dev/urandom of="$INPUT_FILE" bs=1400 count=32 status=none
log "test1: direct tcp client -> server"
# 启动 TCP 服务端接收文件。
"$BUILD_DIR/omni_server" -p tcp -P "$DIRECT_PORT" -o "$DIRECT_OUT" >"$TMP_DIR/direct_server.log" 2>&1 &
DIRECT_SERVER_PID=$!
PIDS+=("$DIRECT_SERVER_PID")
sleep 1
# 启动客户端发送文件。
"$BUILD_DIR/omni_client" -p tcp -H 127.0.0.1 -P "$DIRECT_PORT" -f "$INPUT_FILE" -w 1 >"$TMP_DIR/direct_client.log" 2>&1
wait_with_timeout "$DIRECT_SERVER_PID" 10
# 校验接收文件与输入文件一致。
cmp -s "$INPUT_FILE" "$DIRECT_OUT"
log "test1 passed"
log "test2: udp relay forwarding with dynamic port switch"
# sink1relay 初始目标(预期可能不再接收最终数据)。
"$BUILD_DIR/omni_server" -p udp -P "$SINK1_PORT" -o "$RELAY1_OUT" >"$TMP_DIR/relay_sink1.log" 2>&1 &
SINK1_PID=$!
PIDS+=("$SINK1_PID")
# sink2relay 切换后的目标(最终校验对象)。
"$BUILD_DIR/omni_server" -p udp -P "$SINK2_PORT" -o "$RELAY2_OUT" >"$TMP_DIR/relay_sink2.log" 2>&1 &
SINK2_PID=$!
PIDS+=("$SINK2_PID")
# 预置 relay 控制命令:启动后立即切到 sink2。
CTRL_FILE="$TMP_DIR/relay_ctrl.txt"
printf 'set 127.0.0.1 %s\n' "$SINK2_PORT" >"$CTRL_FILE"
# 启动 relayUDP 监听 RELAY_PORT
"$BUILD_DIR/omni_relay" -p udp -L "$RELAY_PORT" -H 127.0.0.1 -P "$SINK1_PORT" <"$CTRL_FILE" >"$TMP_DIR/relay.log" 2>&1 &
RELAY_PID=$!
PIDS+=("$RELAY_PID")
sleep 1
# 客户端发送到 relay由 relay 中转到目标 sink。
"$BUILD_DIR/omni_client" -p udp -H 127.0.0.1 -P "$RELAY_PORT" -f "$INPUT_FILE" -w 1 >"$TMP_DIR/relay_client.log" 2>&1
wait_with_timeout "$SINK2_PID" 10
# 校验 relay 最终接收端文件一致。
cmp -s "$INPUT_FILE" "$RELAY2_OUT"
if [[ -s "$RELAY1_OUT" ]]; then
# 如果 sink1 收到数据,通常是切换命令生效前的短暂窗口内到达。
log "warning: sink1 received data before switch (relay reconfiguration happened mid-flight)"
fi
# relay/sink1 不一定会自然退出,这里主动结束避免脚本挂住。
kill "$RELAY_PID" 2>/dev/null || true
wait "$RELAY_PID" 2>/dev/null || true
kill "$SINK1_PID" 2>/dev/null || true
log "test2 passed"
log "all smoke tests passed"

View File

@@ -1,44 +1,28 @@
/*
* bridge_main.c
* 固定多跳代理
* 固定多跳桥接
* - 上游作为一个 peer 主动连接远端 hub
* - 下游作为一个轻量 hub 接入本地 peer
* - 将 bind / tunnel / status 在上下游之间原样转发
*
* 适用场景:
* - A 连接 C(hub)
* - B 连接 D(bridge)
* - D 再连接 C
* - C 只看见 bridge 暴露出来的逻辑 client_id
* - 将 bind / tunnel / status 在上下游之间转发
*/
#include "common.h"
#include "logger.h"
#include "peer_transport.h"
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#define BRIDGE_MAX_PAYLOAD (PEER_TUNNEL_META_SIZE + 65536u)
typedef struct BridgeRuntime {
int listen_fd;
int upstream_fd;
int downstream_fd;
pthread_mutex_t upstream_mu;
pthread_mutex_t downstream_mu;
pthread_mutex_t state_mu;
atomic_int running;
PeerTransport *upstream;
PeerTransport *downstream;
PeerTransportSession *downstream_session;
char client_id[OMNI_PEER_ID_SIZE];
} BridgeRuntime;
@@ -66,10 +50,31 @@ static void usage(const char *prog)
{
fprintf(stderr,
"Usage:\n"
" %s -H <upstream_hub_ip> -P <upstream_hub_port> -i <client_id> -L <listen_port> [-b <bind_ip>]\n",
" %s -H <upstream_hub_ip> -P <upstream_hub_port> -i <client_id> -L <listen_port>\n"
" [-b <bind_ip>] [-p tcp|udp|kcp]\n",
prog);
}
static int parse_proto(const char *s, OmniProtocol *out_proto)
{
if (!s || !out_proto) {
return 0;
}
if (strcmp(s, "tcp") == 0) {
*out_proto = OMNI_PROTO_TCP;
return 1;
}
if (strcmp(s, "udp") == 0) {
*out_proto = OMNI_PROTO_UDP;
return 1;
}
if (strcmp(s, "kcp") == 0) {
*out_proto = OMNI_PROTO_KCP;
return 1;
}
return 0;
}
static int peer_id_is_valid(const char *id)
{
size_t len = 0;
@@ -93,127 +98,6 @@ static int peer_id_is_valid(const char *id)
return 1;
}
static ssize_t read_n(int fd, void *buf, size_t n)
{
uint8_t *p = (uint8_t *)buf;
size_t done = 0;
while (done < n) {
ssize_t rc = recv(fd, p + done, n - done, 0);
if (rc == 0) {
return 0;
}
if (rc < 0) {
if (errno == EINTR) {
continue;
}
return -1;
}
done += (size_t)rc;
}
return (ssize_t)done;
}
static ssize_t write_n(int fd, const void *buf, size_t n)
{
const uint8_t *p = (const uint8_t *)buf;
size_t done = 0;
while (done < n) {
ssize_t rc = send(fd, p + done, n - done, 0);
if (rc < 0) {
if (errno == EINTR) {
continue;
}
return -1;
}
done += (size_t)rc;
}
return (ssize_t)done;
}
static int recv_app_message(int fd, MsgHeader *out_hdr, uint8_t *payload_buf, size_t payload_cap)
{
MsgHeader net_hdr;
ssize_t n;
n = read_n(fd, &net_hdr, MSG_HEADER_SIZE);
if (n == 0) {
return 0;
}
if (n != (ssize_t)MSG_HEADER_SIZE) {
return OMNI_ERR_IO;
}
omni_msg_header_decode(&net_hdr, out_hdr);
if (out_hdr->len > payload_cap) {
return OMNI_ERR_IO;
}
if (out_hdr->len > 0) {
n = read_n(fd, payload_buf, out_hdr->len);
if (n != (ssize_t)out_hdr->len) {
return OMNI_ERR_IO;
}
}
logger_on_recv(MSG_HEADER_SIZE + out_hdr->len);
logger_maybe_print_performance_log("bridge_recv");
return 1;
}
static int send_app_message_fd_locked(int fd,
uint32_t type,
const void *payload,
uint32_t payload_len)
{
MsgHeader hdr;
uint8_t header_buf[MSG_HEADER_SIZE];
omni_msg_header_encode(&hdr, type, payload_len, omni_now_ms());
memcpy(header_buf, &hdr, sizeof(header_buf));
if (write_n(fd, header_buf, sizeof(header_buf)) != (ssize_t)sizeof(header_buf)) {
return OMNI_ERR_IO;
}
if (payload_len > 0 && payload) {
if (write_n(fd, payload, payload_len) != (ssize_t)payload_len) {
return OMNI_ERR_IO;
}
}
logger_on_send(MSG_HEADER_SIZE + payload_len);
logger_maybe_print_performance_log("bridge_send");
return OMNI_OK;
}
static int bridge_send_upstream(BridgeRuntime *rt,
uint32_t type,
const void *payload,
uint32_t payload_len)
{
int rc;
pthread_mutex_lock(&rt->upstream_mu);
rc = send_app_message_fd_locked(rt->upstream_fd, type, payload, payload_len);
pthread_mutex_unlock(&rt->upstream_mu);
return rc;
}
static int bridge_send_downstream(BridgeRuntime *rt,
uint32_t type,
const void *payload,
uint32_t payload_len)
{
int rc = OMNI_ERR_IO;
pthread_mutex_lock(&rt->downstream_mu);
if (rt->downstream_fd >= 0) {
rc = send_app_message_fd_locked(rt->downstream_fd, type, payload, payload_len);
}
pthread_mutex_unlock(&rt->downstream_mu);
return rc;
}
static int send_status_to_downstream(BridgeRuntime *rt,
uint32_t code,
const char *self_id,
@@ -222,209 +106,182 @@ static int send_status_to_downstream(BridgeRuntime *rt,
{
PeerStatusMeta status_meta;
if (!rt || !rt->downstream || !rt->downstream_session) {
return OMNI_ERR_PARAM;
}
omni_peer_status_meta_encode(&status_meta, code, self_id, peer_id, detail);
return bridge_send_downstream(rt,
return peer_transport_send(rt->downstream,
rt->downstream_session,
MSG_TYPE_PEER_STATUS,
&status_meta,
PEER_STATUS_META_SIZE);
}
static int create_connected_socket(const char *host, uint16_t port)
{
int fd;
struct sockaddr_in addr;
fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) {
return -1;
}
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
if (inet_pton(AF_INET, host, &addr.sin_addr) != 1) {
close(fd);
errno = EINVAL;
return -1;
}
if (connect(fd, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
close(fd);
return -1;
}
return fd;
}
static int create_listen_socket(const char *bind_ip, uint16_t port)
{
int fd;
int reuse = 1;
struct sockaddr_in addr;
fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd < 0) {
return -1;
}
(void)setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_port = htons(port);
if (bind_ip && bind_ip[0] != '\0') {
if (inet_pton(AF_INET, bind_ip, &addr.sin_addr) != 1) {
close(fd);
return -1;
}
} else {
addr.sin_addr.s_addr = htonl(INADDR_ANY);
}
if (bind(fd, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
close(fd);
return -1;
}
if (listen(fd, 8) != 0) {
close(fd);
return -1;
}
return fd;
}
static int bridge_send_upstream_register(BridgeRuntime *rt)
static int send_upstream_register(BridgeRuntime *rt)
{
PeerRegisterMeta meta;
if (!rt || !rt->upstream) {
return OMNI_ERR_PARAM;
}
omni_peer_register_meta_encode(&meta, rt->client_id);
return bridge_send_upstream(rt, MSG_TYPE_PEER_REGISTER, &meta, PEER_REGISTER_META_SIZE);
return peer_transport_send(rt->upstream,
NULL,
MSG_TYPE_PEER_REGISTER,
&meta,
PEER_REGISTER_META_SIZE);
}
static void *upstream_thread_main(void *arg)
static int handle_upstream_event(BridgeRuntime *rt,
const PeerTransportEvent *event,
const uint8_t *payload)
{
BridgeRuntime *rt = (BridgeRuntime *)arg;
uint8_t payload[BRIDGE_MAX_PAYLOAD];
if (!rt || !event) {
return OMNI_ERR_PARAM;
}
while (atomic_load(&rt->running)) {
MsgHeader hdr;
int rc = recv_app_message(rt->upstream_fd, &hdr, payload, sizeof(payload));
if (rc == 0) {
if (event->kind == PEER_TRANSPORT_EVENT_CLOSED) {
logger_log("INFO", "bridge", "upstream_closed");
break;
}
if (rc < 0) {
logger_log("ERROR", "bridge", "upstream_recv_failed rc=%d", rc);
break;
return OMNI_ERR_IO;
}
switch (hdr.type) {
switch (event->header.type) {
case MSG_TYPE_PEER_STATUS:
case MSG_TYPE_PEER_TUNNEL:
if (bridge_send_downstream(rt, hdr.type, payload, hdr.len) != OMNI_OK) {
if (!rt->downstream_session) {
logger_log("WARN", "bridge",
"downstream_forward_skipped type=%u len=%u",
(unsigned)hdr.type,
(unsigned)hdr.len);
(unsigned)event->header.type,
(unsigned)event->header.len);
return OMNI_OK;
}
break;
return peer_transport_send(rt->downstream,
rt->downstream_session,
event->header.type,
payload,
event->header.len);
default:
logger_log("WARN", "bridge", "unexpected_upstream_type=%u len=%u",
(unsigned)hdr.type, (unsigned)hdr.len);
break;
logger_log("WARN", "bridge",
"unexpected_upstream_type=%u len=%u",
(unsigned)event->header.type,
(unsigned)event->header.len);
return OMNI_OK;
}
}
atomic_store(&rt->running, 0);
return NULL;
}
static void clear_downstream_fd(BridgeRuntime *rt, int fd)
static int handle_downstream_register(BridgeRuntime *rt,
PeerTransportSession *session,
const uint8_t *payload,
uint32_t payload_len)
{
pthread_mutex_lock(&rt->downstream_mu);
if (rt->downstream_fd == fd) {
rt->downstream_fd = -1;
}
pthread_mutex_unlock(&rt->downstream_mu);
}
static void handle_downstream_connection(BridgeRuntime *rt, int fd)
{
uint8_t payload[BRIDGE_MAX_PAYLOAD];
int registered = 0;
while (atomic_load(&rt->running)) {
MsgHeader hdr;
int rc = recv_app_message(fd, &hdr, payload, sizeof(payload));
if (rc == 0) {
logger_log("INFO", "bridge", "downstream_closed");
break;
}
if (rc < 0) {
logger_log("ERROR", "bridge", "downstream_recv_failed rc=%d", rc);
break;
}
switch (hdr.type) {
case MSG_TYPE_PEER_REGISTER: {
PeerRegisterMeta register_meta;
if (hdr.len < PEER_REGISTER_META_SIZE) {
(void)send_status_to_downstream(rt, PEER_STATUS_ERROR, NULL, NULL, "short_register_payload");
break;
if (!rt || !session) {
return OMNI_ERR_PARAM;
}
if (payload_len < PEER_REGISTER_META_SIZE) {
rt->downstream_session = session;
return send_status_to_downstream(rt,
PEER_STATUS_ERROR,
NULL,
NULL,
"short_register_payload");
}
omni_peer_register_meta_decode((const PeerRegisterMeta *)payload, &register_meta);
if (strcmp(register_meta.client_id, rt->client_id) != 0) {
(void)send_status_to_downstream(rt,
PEER_STATUS_ERROR,
register_meta.client_id,
NULL,
"client_id_mismatch");
rt->downstream_session = session;
logger_log("WARN", "bridge",
"downstream_register_mismatch got=%s expect=%s",
register_meta.client_id,
rt->client_id);
break;
return send_status_to_downstream(rt,
PEER_STATUS_ERROR,
register_meta.client_id,
NULL,
"client_id_mismatch");
}
registered = 1;
(void)send_status_to_downstream(rt,
rt->downstream_session = session;
return send_status_to_downstream(rt,
PEER_STATUS_REGISTERED,
rt->client_id,
NULL,
"bridge_register_ok");
break;
}
static int handle_downstream_event(BridgeRuntime *rt,
const PeerTransportEvent *event,
const uint8_t *payload)
{
if (!rt || !event) {
return OMNI_ERR_PARAM;
}
if (event->kind == PEER_TRANSPORT_EVENT_CLOSED) {
if (rt->downstream_session == event->session) {
logger_log("INFO", "bridge", "downstream_closed");
rt->downstream_session = NULL;
}
peer_transport_close_session(rt->downstream, event->session);
return OMNI_OK;
}
if (rt->downstream_session && rt->downstream_session != event->session) {
PeerStatusMeta status_meta;
omni_peer_status_meta_encode(&status_meta,
PEER_STATUS_ERROR,
rt->client_id,
NULL,
"bridge_busy");
(void)peer_transport_send(rt->downstream,
event->session,
MSG_TYPE_PEER_STATUS,
&status_meta,
PEER_STATUS_META_SIZE);
peer_transport_close_session(rt->downstream, event->session);
logger_log("WARN", "bridge", "reject_extra_downstream");
return OMNI_OK;
}
switch (event->header.type) {
case MSG_TYPE_PEER_REGISTER:
return handle_downstream_register(rt,
event->session,
payload,
event->header.len);
case MSG_TYPE_PEER_BIND:
case MSG_TYPE_PEER_TUNNEL:
if (!registered) {
if (rt->downstream_session != event->session) {
rt->downstream_session = event->session;
(void)send_status_to_downstream(rt,
PEER_STATUS_ERROR,
rt->client_id,
NULL,
"register_first");
break;
return OMNI_OK;
}
if (bridge_send_upstream(rt, hdr.type, payload, hdr.len) != OMNI_OK) {
logger_log("ERROR", "bridge",
"upstream_forward_failed type=%u len=%u",
(unsigned)hdr.type,
(unsigned)hdr.len);
if (peer_transport_send(rt->upstream,
NULL,
event->header.type,
payload,
event->header.len) != OMNI_OK) {
(void)send_status_to_downstream(rt,
PEER_STATUS_ERROR,
rt->client_id,
NULL,
"upstream_forward_failed");
}
break;
return OMNI_OK;
default:
logger_log("WARN", "bridge", "unexpected_downstream_type=%u len=%u",
(unsigned)hdr.type, (unsigned)hdr.len);
break;
logger_log("WARN", "bridge",
"unexpected_downstream_type=%u len=%u",
(unsigned)event->header.type,
(unsigned)event->header.len);
return OMNI_OK;
}
}
clear_downstream_fd(rt, fd);
shutdown(fd, SHUT_RDWR);
close(fd);
}
int main(int argc, char **argv)
@@ -432,13 +289,16 @@ int main(int argc, char **argv)
const char *upstream_ip = NULL;
const char *bind_ip = NULL;
const char *client_id = NULL;
const char *proto_str = "tcp";
OmniProtocol proto = OMNI_PROTO_TCP;
int upstream_port = 0;
int listen_port = 0;
int opt;
BridgeRuntime rt;
pthread_t upstream_tid;
uint8_t upstream_payload[BRIDGE_MAX_PAYLOAD];
uint8_t downstream_payload[BRIDGE_MAX_PAYLOAD];
while ((opt = getopt(argc, argv, "H:P:i:L:b:")) != -1) {
while ((opt = getopt(argc, argv, "H:P:i:L:b:p:")) != -1) {
switch (opt) {
case 'H':
upstream_ip = optarg;
@@ -455,113 +315,97 @@ int main(int argc, char **argv)
case 'b':
bind_ip = optarg;
break;
case 'p':
proto_str = optarg;
break;
default:
usage(argv[0]);
return 1;
}
}
if (!upstream_ip || upstream_port <= 0 || listen_port <= 0 || !peer_id_is_valid(client_id)) {
if (!upstream_ip || upstream_port <= 0 || listen_port <= 0 ||
!peer_id_is_valid(client_id) || !parse_proto(proto_str, &proto)) {
usage(argv[0]);
return 1;
}
logger_init();
install_signal_handlers();
logger_set_context("bridge", proto_str, "bridge", "relay", client_id);
memset(&rt, 0, sizeof(rt));
rt.listen_fd = -1;
rt.upstream_fd = -1;
rt.downstream_fd = -1;
omni_copy_fixed_ascii(rt.client_id, sizeof(rt.client_id), client_id);
atomic_init(&rt.running, 1);
pthread_mutex_init(&rt.upstream_mu, NULL);
pthread_mutex_init(&rt.downstream_mu, NULL);
pthread_mutex_init(&rt.state_mu, NULL);
rt.upstream_fd = create_connected_socket(upstream_ip, (uint16_t)upstream_port);
if (rt.upstream_fd < 0) {
perror("bridge upstream connect");
goto fail;
rt.upstream = peer_transport_open(OMNI_ROLE_CLIENT,
proto,
NULL,
0,
upstream_ip,
(uint16_t)upstream_port);
if (!rt.upstream) {
perror("bridge upstream");
return 1;
}
if (bridge_send_upstream_register(&rt) != OMNI_OK) {
if (send_upstream_register(&rt) != OMNI_OK) {
fprintf(stderr, "bridge upstream register failed\n");
goto fail;
peer_transport_close(rt.upstream);
return 1;
}
rt.listen_fd = create_listen_socket(bind_ip, (uint16_t)listen_port);
if (rt.listen_fd < 0) {
perror("bridge listen");
goto fail;
}
if (pthread_create(&upstream_tid, NULL, upstream_thread_main, &rt) != 0) {
perror("bridge pthread_create");
goto fail;
rt.downstream = peer_transport_open(OMNI_ROLE_SERVER,
proto,
bind_ip,
(uint16_t)listen_port,
NULL,
0);
if (!rt.downstream) {
perror("bridge downstream");
peer_transport_close(rt.upstream);
return 1;
}
logger_log("INFO", "bridge",
"listening bind_ip=%s listen_port=%u upstream=%s:%u client_id=%s",
"listening bind_ip=%s listen_port=%u upstream=%s:%u proto=%s client_id=%s",
bind_ip ? bind_ip : "0.0.0.0",
(unsigned)listen_port,
upstream_ip,
(unsigned)upstream_port,
peer_transport_proto_name(proto),
rt.client_id);
while (atomic_load(&rt.running) && !g_stop) {
struct sockaddr_in peer_addr;
socklen_t peer_len = sizeof(peer_addr);
int cfd = accept(rt.listen_fd, (struct sockaddr *)&peer_addr, &peer_len);
while (!g_stop) {
PeerTransportEvent event;
int rc;
if (cfd < 0) {
if (errno == EINTR && !g_stop) {
continue;
}
if (g_stop) {
rc = peer_transport_next_event(rt.upstream,
&event,
upstream_payload,
sizeof(upstream_payload),
50);
if (rc < 0) {
logger_log("ERROR", "bridge", "upstream_recv_failed rc=%d", rc);
break;
}
perror("bridge accept");
if (rc > 0 && handle_upstream_event(&rt, &event, upstream_payload) != OMNI_OK) {
break;
}
pthread_mutex_lock(&rt.downstream_mu);
if (rt.downstream_fd >= 0) {
pthread_mutex_unlock(&rt.downstream_mu);
close(cfd);
logger_log("WARN", "bridge", "reject_extra_downstream");
continue;
rc = peer_transport_next_event(rt.downstream,
&event,
downstream_payload,
sizeof(downstream_payload),
50);
if (rc < 0) {
logger_log("ERROR", "bridge", "downstream_recv_failed rc=%d", rc);
break;
}
if (rc > 0) {
(void)handle_downstream_event(&rt, &event, downstream_payload);
}
rt.downstream_fd = cfd;
pthread_mutex_unlock(&rt.downstream_mu);
logger_log("INFO", "bridge", "downstream_connected");
handle_downstream_connection(&rt, cfd);
}
atomic_store(&rt.running, 0);
if (rt.listen_fd >= 0) {
close(rt.listen_fd);
}
pthread_join(upstream_tid, NULL);
if (rt.upstream_fd >= 0) {
shutdown(rt.upstream_fd, SHUT_RDWR);
close(rt.upstream_fd);
}
peer_transport_close(rt.downstream);
peer_transport_close(rt.upstream);
logger_print_performance_log("final");
pthread_mutex_destroy(&rt.state_mu);
pthread_mutex_destroy(&rt.downstream_mu);
pthread_mutex_destroy(&rt.upstream_mu);
return 0;
fail:
if (rt.listen_fd >= 0) {
close(rt.listen_fd);
}
if (rt.upstream_fd >= 0) {
close(rt.upstream_fd);
}
pthread_mutex_destroy(&rt.state_mu);
pthread_mutex_destroy(&rt.downstream_mu);
pthread_mutex_destroy(&rt.upstream_mu);
return 1;
}

View File

@@ -1,844 +0,0 @@
/*
* client_main.c
* 客户端:读取大文件分片发送,同时后台接收服务端控制/确认消息
*
* 整体模型:
* 1) 主线程负责读文件、切 chunk、封装业务帧并发送。
* 2) 接收线程负责监听服务端命令和最终 ACK。
* 3) 两个线程通过原子变量共享“运行状态”和“ACK 结果”。
*/
#include "common.h"
#include "network.h"
#include "logger.h"
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
/* 接收线程的单帧缓冲区上限:业务头 + chunk 元数据 + 最大 payload。 */
#define CLIENT_FRAME_BUF_SIZE (MSG_HEADER_SIZE + TRANSFER_CHUNK_META_SIZE + 65536u)
#define OMNI_TIME_SYNC_PROBE_COUNT 5u
#define OMNI_TIME_SYNC_REPLY_TIMEOUT_MS 500u
typedef struct ClientRuntime {
OmniContext *ctx; /* 底层协议上下文。 */
atomic_int running; /* 共享退出标志,主线程和接收线程都会检查。 */
atomic_int ack_received; /* 是否已经收到服务端的传输完成确认。 */
atomic_ullong ack_rtt_ms; /* FILE_END -> TRANSFER_ACK 这一来一回的 RTT。 */
atomic_ullong ack_bytes_written; /* 服务端实际写盘字节数。 */
atomic_uint ack_transfer_id; /* 收到 ACK 对应的传输 ID。 */
atomic_uint sync_expected_probe_id; /* 当前正在等的 probe。 */
atomic_ullong sync_expected_client_send_ts_ms; /* 当前 probe 的客户端 t0。 */
atomic_int sync_reply_ready; /* 后台线程是否已经收到匹配响应。 */
atomic_ullong sync_reply_rtt_ms; /* 当前 probe 的 RTT。 */
atomic_llong sync_reply_offset_ms; /* 当前 probe 的 offset。 */
} ClientRuntime;
typedef struct ClockSyncResult {
int valid; /* 是否已拿到可用 offset。 */
int64_t server_minus_client_offset_ms; /* server_time - client_time */
uint64_t best_rtt_ms; /* 选中样本的 RTT。 */
uint32_t sample_count; /* 实际成功样本数。 */
} ClockSyncResult;
/* 信号处理只做最轻量的事情:设置停止标志,由主流程自己收尾。 */
static volatile sig_atomic_t g_stop = 0;
/* SIGINT/SIGTERM 的处理函数:通知发送循环尽快退出。 */
static void on_signal(int signo)
{
(void)signo;
g_stop = 1;
}
/* 注册 Ctrl+C / kill 的处理逻辑,避免被粗暴中断后缺少收尾日志。 */
static void install_signal_handlers(void)
{
struct sigaction sa;
memset(&sa, 0, sizeof(sa));
sa.sa_handler = on_signal;
sigemptyset(&sa.sa_mask);
sa.sa_flags = 0;
(void)sigaction(SIGINT, &sa, NULL);
(void)sigaction(SIGTERM, &sa, NULL);
}
/* 打印客户端命令行帮助。 */
static void usage(const char *prog)
{
fprintf(stderr,
"Usage:\n"
" %s -p tcp|udp|kcp -H <server_ip> -P <server_port> -f <file>\n"
" [-b <bind_port>] [-m <chunk_mtu>] [-w <wait_seconds|-1>]\n",
prog);
}
/* 将字符串协议名转换为内部枚举;非法输入默认回退 TCP。 */
static OmniProtocol parse_proto(const char *s)
{
if (!s) return OMNI_PROTO_TCP;
if (strcmp(s, "tcp") == 0) return OMNI_PROTO_TCP;
if (strcmp(s, "udp") == 0) return OMNI_PROTO_UDP;
if (strcmp(s, "kcp") == 0) return OMNI_PROTO_KCP;
return OMNI_PROTO_TCP;
}
/*
* 发送一个完整的业务帧。
* 这里统一负责:拼 MsgHeader -> 拷贝 payload -> 调底层 omni_send。
*/
static int send_app_message_with_timestamp(OmniContext *ctx,
uint32_t type,
const void *payload,
uint32_t payload_len,
uint64_t timestamp_ms)
{
size_t total_len = MSG_HEADER_SIZE + (size_t)payload_len;
uint8_t *frame = (uint8_t *)malloc(total_len);
if (!frame) {
logger_log("ERROR", "client", "malloc_frame_failed len=%zu", total_len);
return OMNI_ERR_GENERIC;
}
MsgHeader hdr;
/* 业务层总是按“头 + 载荷”的统一格式发给对端。 */
omni_msg_header_encode(&hdr, type, payload_len, timestamp_ms);
memcpy(frame, &hdr, MSG_HEADER_SIZE);
if (payload_len > 0 && payload) {
memcpy(frame + MSG_HEADER_SIZE, payload, payload_len);
}
ssize_t n = omni_send(ctx, frame, total_len);
free(frame);
if (n != (ssize_t)total_len) {
logger_log("ERROR", "client",
"omni_send_failed expect=%zu got=%zd type=%u",
total_len, n, (unsigned)type);
return OMNI_ERR_IO;
}
return OMNI_OK;
}
/* 从接收到的一整帧中拆出应用层头和 payload 指针,并做长度一致性校验。 */
static int decode_app_message(const uint8_t *frame,
size_t frame_len,
MsgHeader *out_hdr,
const uint8_t **out_payload)
{
if (!frame || frame_len < MSG_HEADER_SIZE || !out_hdr || !out_payload) {
return OMNI_ERR_PARAM;
}
MsgHeader net_hdr;
memcpy(&net_hdr, frame, MSG_HEADER_SIZE);
omni_msg_header_decode(&net_hdr, out_hdr);
if ((size_t)out_hdr->len + MSG_HEADER_SIZE != frame_len) {
return OMNI_ERR_IO;
}
*out_payload = frame + MSG_HEADER_SIZE;
return OMNI_OK;
}
/* 用四时间戳公式估算 server_time - client_time。 */
static int64_t compute_server_minus_client_offset_ms(const TimeSyncReplyMeta *reply,
uint64_t client_recv_ts_ms)
{
int64_t req_leg = (int64_t)reply->server_recv_ts_ms -
(int64_t)reply->client_send_ts_ms;
int64_t resp_leg = (int64_t)reply->server_send_ts_ms -
(int64_t)client_recv_ts_ms;
return (req_leg + resp_leg) / 2;
}
/*
* 在正式发文件前做几轮时钟探测:
* - 主线程发 t0
* - 服务端回 t1/t2
* - 后台接收线程在 t3 处写回 RTT/offset
* - 主线程带超时地等待,保证经由单向 relay 时不会卡死
*/
static int perform_time_sync(OmniContext *ctx,
ClientRuntime *rt,
ClockSyncResult *out_result)
{
uint64_t best_rtt_ms = UINT64_MAX;
int64_t best_offset_ms = 0;
uint32_t sample_count = 0;
if (!ctx || !rt || !out_result) {
return OMNI_ERR_PARAM;
}
memset(out_result, 0, sizeof(*out_result));
for (uint32_t probe_id = 1; probe_id <= OMNI_TIME_SYNC_PROBE_COUNT; ++probe_id) {
TimeSyncProbeMeta probe_meta;
uint64_t client_send_ts_ms = omni_now_ms();
int got_reply = 0;
atomic_store(&rt->sync_reply_ready, 0);
atomic_store(&rt->sync_expected_probe_id, probe_id);
atomic_store(&rt->sync_expected_client_send_ts_ms, client_send_ts_ms);
omni_time_sync_probe_meta_encode(&probe_meta, probe_id, client_send_ts_ms);
if (send_app_message_with_timestamp(ctx,
MSG_TYPE_TIME_SYNC_REQ,
&probe_meta,
TIME_SYNC_PROBE_META_SIZE,
client_send_ts_ms) != OMNI_OK) {
logger_log("ERROR", "client",
"time_sync_send_failed probe_id=%u",
(unsigned)probe_id);
return OMNI_ERR_IO;
}
for (uint32_t waited_ms = 0;
waited_ms < OMNI_TIME_SYNC_REPLY_TIMEOUT_MS;
waited_ms += 5u) {
if (atomic_load(&rt->sync_reply_ready)) {
uint64_t rtt_ms = atomic_load(&rt->sync_reply_rtt_ms);
int64_t offset_ms = atomic_load(&rt->sync_reply_offset_ms);
sample_count++;
got_reply = 1;
logger_on_rtt(rtt_ms);
logger_log("INFO", "client",
"time_sync_sample probe_id=%u rtt_ms=%llu offset_ms=%lld",
(unsigned)probe_id,
(unsigned long long)rtt_ms,
(long long)offset_ms);
if (rtt_ms < best_rtt_ms) {
best_rtt_ms = rtt_ms;
best_offset_ms = offset_ms;
}
break;
}
usleep(5 * 1000);
}
atomic_store(&rt->sync_expected_probe_id, 0);
atomic_store(&rt->sync_expected_client_send_ts_ms, 0);
atomic_store(&rt->sync_reply_ready, 0);
if (!got_reply) {
logger_log("WARN", "client",
"time_sync_probe_timeout probe_id=%u timeout_ms=%u",
(unsigned)probe_id,
(unsigned)OMNI_TIME_SYNC_REPLY_TIMEOUT_MS);
}
}
if (sample_count == 0 || best_rtt_ms == UINT64_MAX) {
logger_log("WARN", "client", "time_sync_no_valid_sample");
return OMNI_ERR_IO;
}
out_result->valid = 1;
out_result->server_minus_client_offset_ms = best_offset_ms;
out_result->best_rtt_ms = best_rtt_ms;
out_result->sample_count = sample_count;
{
TimeSyncReportMeta report_meta;
uint64_t report_ts_ms = omni_now_ms();
omni_time_sync_report_meta_encode(&report_meta,
best_offset_ms,
best_rtt_ms,
sample_count);
if (send_app_message_with_timestamp(ctx,
MSG_TYPE_TIME_SYNC_REPORT,
&report_meta,
TIME_SYNC_REPORT_META_SIZE,
report_ts_ms) != OMNI_OK) {
logger_log("ERROR", "client",
"time_sync_report_send_failed offset_ms=%lld best_rtt_ms=%llu",
(long long)best_offset_ms,
(unsigned long long)best_rtt_ms);
return OMNI_ERR_IO;
}
}
logger_log("INFO", "client",
"time_sync_selected offset_ms=%lld best_rtt_ms=%llu samples=%u",
(long long)best_offset_ms,
(unsigned long long)best_rtt_ms,
(unsigned)sample_count);
return OMNI_OK;
}
/* 按需扩容“每秒发送多少片”的统计数组。 */
static int ensure_window_capacity(uint64_t **counts, size_t *cap, uint32_t window_id)
{
size_t need = (size_t)window_id + 1u;
size_t new_cap;
uint64_t *new_counts;
if (need <= *cap) {
return OMNI_OK;
}
new_cap = (*cap == 0) ? 8u : *cap;
while (new_cap < need) {
new_cap *= 2u;
}
new_counts = (uint64_t *)realloc(*counts, new_cap * sizeof(uint64_t));
if (!new_counts) {
return OMNI_ERR_GENERIC;
}
memset(new_counts + *cap, 0, (new_cap - *cap) * sizeof(uint64_t));
*counts = new_counts;
*cap = new_cap;
return OMNI_OK;
}
/* 将窗口分布统计转成 "0:12,1:15,3:9" 这类便于日志输出的字符串。 */
static char *format_window_distribution(const uint64_t *counts, uint32_t total_windows)
{
size_t cap = 256;
size_t len = 0;
char *buf = (char *)malloc(cap);
if (!buf) {
return NULL;
}
buf[0] = '\0';
for (uint32_t i = 0; i < total_windows; ++i) {
char tmp[64];
int n;
if (counts[i] == 0) {
continue;
}
n = snprintf(tmp, sizeof(tmp), "%s%u:%llu",
(len == 0) ? "" : ",",
(unsigned)i,
(unsigned long long)counts[i]);
if (n <= 0) {
continue;
}
while (len + (size_t)n + 1 > cap) {
char *new_buf;
cap *= 2u;
new_buf = (char *)realloc(buf, cap);
if (!new_buf) {
free(buf);
return NULL;
}
buf = new_buf;
}
memcpy(buf + len, tmp, (size_t)n);
len += (size_t)n;
buf[len] = '\0';
}
if (len == 0) {
snprintf(buf, cap, "none");
}
return buf;
}
/* 通过保存/恢复文件指针位置计算总文件大小,不影响后续顺序读取。 */
static uint64_t compute_file_size(FILE *fp)
{
off_t cur = ftello(fp);
off_t end;
if (cur < 0) {
return 0;
}
if (fseeko(fp, 0, SEEK_END) != 0) {
return 0;
}
end = ftello(fp);
(void)fseeko(fp, cur, SEEK_SET);
if (end < 0) {
return 0;
}
return (uint64_t)end;
}
/* 计算百分比时统一处理分母为 0 的情况。 */
static double rate_percent(uint64_t numerator, uint64_t denominator)
{
if (denominator == 0) {
return 0.0;
}
return ((double)numerator * 100.0) / (double)denominator;
}
static uint64_t saturating_sub_u64(uint64_t total, uint64_t delta)
{
return (total > delta) ? (total - delta) : 0;
}
/*
* 后台接收线程:
* - 接收服务端主动下发的命令
* - 接收传输结束 ACK并把结果写入原子变量
*/
static void *recv_thread_main(void *arg)
{
ClientRuntime *rt = (ClientRuntime *)arg;
uint8_t frame[CLIENT_FRAME_BUF_SIZE];
/* 允许主线程在退出时 cancel 本线程。 */
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
while (atomic_load(&rt->running)) {
ssize_t n = omni_recv(rt->ctx, frame, sizeof(frame));
if (n < 0) {
logger_log("ERROR", "client", "recv_failed n=%zd", n);
break;
}
if (n == 0) {
usleep(2 * 1000);
continue;
}
MsgHeader hdr;
const uint8_t *payload = NULL;
int rc = decode_app_message(frame, (size_t)n, &hdr, &payload);
if (rc != OMNI_OK) {
logger_log("ERROR", "client", "invalid_app_frame bytes=%zd rc=%d", n, rc);
continue;
}
if (hdr.type == MSG_TYPE_COMMAND) {
/* 命令消息只是做演示打印,不参与文件传输状态机。 */
char cmd[2048];
size_t cpy = hdr.len < (uint32_t)(sizeof(cmd) - 1) ? hdr.len : (sizeof(cmd) - 1);
memcpy(cmd, payload, cpy);
cmd[cpy] = '\0';
printf("[server-cmd] %s\n", cmd);
fflush(stdout);
} else if (hdr.type == MSG_TYPE_TRANSFER_ACK) {
TransferAckMeta ack_meta;
uint64_t rtt_ms;
if (hdr.len < TRANSFER_ACK_META_SIZE) {
logger_log("WARN", "client", "short_transfer_ack len=%u", (unsigned)hdr.len);
continue;
}
omni_transfer_ack_meta_decode((const TransferAckMeta *)payload, &ack_meta);
/* 服务端把 FILE_END 的发送时间原样回显回来,因此客户端可直接测一轮 ACK RTT。 */
rtt_ms = omni_now_ms() - ack_meta.echoed_end_ts_ms;
atomic_store(&rt->ack_received, 1);
atomic_store(&rt->ack_rtt_ms, rtt_ms);
atomic_store(&rt->ack_bytes_written, ack_meta.bytes_written);
atomic_store(&rt->ack_transfer_id, ack_meta.transfer_id);
logger_on_rtt(rtt_ms);
logger_log("INFO", "client",
"transfer_ack transfer_id=%u bytes_written=%llu rtt_ms=%llu",
(unsigned)ack_meta.transfer_id,
(unsigned long long)ack_meta.bytes_written,
(unsigned long long)rtt_ms);
} else if (hdr.type == MSG_TYPE_TIME_SYNC_RESP) {
TimeSyncReplyMeta reply_meta;
uint32_t expected_probe_id;
uint64_t expected_client_send_ts_ms;
uint64_t client_recv_ts_ms;
uint64_t rtt_ms;
int64_t offset_ms;
if (hdr.len < TIME_SYNC_REPLY_META_SIZE) {
logger_log("WARN", "client", "short_time_sync_resp len=%u", (unsigned)hdr.len);
continue;
}
omni_time_sync_reply_meta_decode((const TimeSyncReplyMeta *)payload, &reply_meta);
expected_probe_id = atomic_load(&rt->sync_expected_probe_id);
expected_client_send_ts_ms = atomic_load(&rt->sync_expected_client_send_ts_ms);
if (expected_probe_id == 0 ||
reply_meta.probe_id != expected_probe_id ||
reply_meta.client_send_ts_ms != expected_client_send_ts_ms) {
logger_log("WARN", "client",
"unexpected_time_sync_resp probe_id=%u expected_probe=%u",
(unsigned)reply_meta.probe_id,
(unsigned)expected_probe_id);
continue;
}
client_recv_ts_ms = omni_now_ms();
rtt_ms = client_recv_ts_ms - reply_meta.client_send_ts_ms;
offset_ms = compute_server_minus_client_offset_ms(&reply_meta,
client_recv_ts_ms);
atomic_store(&rt->sync_reply_rtt_ms, rtt_ms);
atomic_store(&rt->sync_reply_offset_ms, offset_ms);
atomic_store(&rt->sync_reply_ready, 1);
} else {
logger_log("INFO", "client",
"recv_non_command type=%u len=%u",
(unsigned)hdr.type, (unsigned)hdr.len);
}
}
atomic_store(&rt->running, 0);
return NULL;
}
/*
* 客户端主流程:
* 1) 解析参数并建立连接
* 2) 启动接收线程
* 3) 循环读取文件并分片发送
* 4) 发送 FILE_END
* 5) 等待 ACK 或超时,然后输出汇总
*/
int main(int argc, char **argv)
{
const char *proto_str = "tcp";
const char *server_ip = NULL;
const char *file_path = NULL;
OmniProtocol proto;
FILE *fp = NULL;
OmniContext *ctx = NULL;
pthread_t recv_tid;
ClientRuntime rt;
uint8_t *chunk = NULL;
uint64_t *window_counts = NULL;
size_t window_cap = 0;
uint32_t total_windows = 0;
uint64_t total_bytes = 0;
uint32_t total_chunks = 0;
uint32_t transfer_id;
uint64_t total_sent = 0;
uint64_t offset = 0;
uint64_t transfer_start_send_ms = 0;
uint64_t file_end_send_ts_ms = 0;
ClockSyncResult clock_sync;
int server_port = 0;
int bind_port = 0;
unsigned chunk_size = OMNI_DEFAULT_MTU;
int wait_seconds = 2;
int opt;
int exit_code = 0;
install_signal_handlers();
while ((opt = getopt(argc, argv, "p:H:P:f:b:m:w:")) != -1) {
switch (opt) {
case 'p':
proto_str = optarg;
break;
case 'H':
server_ip = optarg;
break;
case 'P':
server_port = atoi(optarg);
break;
case 'f':
file_path = optarg;
break;
case 'b':
bind_port = atoi(optarg);
break;
case 'm':
chunk_size = (unsigned)strtoul(optarg, NULL, 10);
break;
case 'w':
wait_seconds = atoi(optarg);
break;
default:
usage(argv[0]);
return 1;
}
}
if (!server_ip || server_port <= 0 || !file_path) {
usage(argv[0]);
return 1;
}
if (chunk_size == 0 || chunk_size > 65536u) {
fprintf(stderr, "invalid chunk size: %u\n", chunk_size);
return 1;
}
fp = fopen(file_path, "rb");
if (!fp) {
perror("fopen");
return 1;
}
total_bytes = compute_file_size(fp);
total_chunks = (chunk_size == 0) ? 0u :
(uint32_t)((total_bytes + (uint64_t)chunk_size - 1u) / (uint64_t)chunk_size);
transfer_id = (uint32_t)((omni_now_ms() ^ (uint64_t)getpid() ^ total_bytes) & 0xffffffffu);
proto = parse_proto(proto_str);
ctx = omni_init(OMNI_ROLE_CLIENT, proto,
NULL, (uint16_t)bind_port,
server_ip, (uint16_t)server_port);
if (!ctx) {
fclose(fp);
fprintf(stderr, "omni_init failed\n");
return 1;
}
logger_set_transfer_total(total_bytes);
logger_set_progress(0);
memset(&clock_sync, 0, sizeof(clock_sync));
/* 接收线程与主线程共享一个 runtime 结构。 */
rt.ctx = ctx;
atomic_init(&rt.running, 1);
atomic_init(&rt.ack_received, 0);
atomic_init(&rt.ack_rtt_ms, 0);
atomic_init(&rt.ack_bytes_written, 0);
atomic_init(&rt.ack_transfer_id, 0);
atomic_init(&rt.sync_expected_probe_id, 0);
atomic_init(&rt.sync_expected_client_send_ts_ms, 0);
atomic_init(&rt.sync_reply_ready, 0);
atomic_init(&rt.sync_reply_rtt_ms, 0);
atomic_init(&rt.sync_reply_offset_ms, 0);
if (pthread_create(&recv_tid, NULL, recv_thread_main, &rt) != 0) {
perror("pthread_create");
fclose(fp);
omni_close(ctx);
return 1;
}
if (perform_time_sync(ctx, &rt, &clock_sync) != OMNI_OK) {
logger_log("WARN", "client",
"time_sync_unavailable transfer_will_continue_without_compensated_server_metrics");
}
chunk = (uint8_t *)malloc(chunk_size);
if (!chunk) {
logger_log("ERROR", "client", "malloc_chunk_failed size=%u", chunk_size);
atomic_store(&rt.running, 0);
pthread_cancel(recv_tid);
pthread_join(recv_tid, NULL);
fclose(fp);
omni_close(ctx);
return 1;
}
for (uint32_t seq = 1; atomic_load(&rt.running); ++seq) {
uint64_t origin_ts_ms;
size_t nread;
if (g_stop) {
logger_log("INFO", "client", "signal_received_stop_sending");
atomic_store(&rt.running, 0);
exit_code = 1;
break;
}
origin_ts_ms = omni_now_ms();
nread = fread(chunk, 1, chunk_size, fp);
if (nread == 0) {
if (feof(fp)) {
break;
}
if (ferror(fp)) {
logger_log("ERROR", "client", "fread_failed");
atomic_store(&rt.running, 0);
exit_code = 1;
break;
}
}
if (nread > 0) {
uint64_t process_t0 = omni_now_ms();
uint64_t send_ts_ms = omni_now_ms();
uint32_t window_id;
size_t payload_len = TRANSFER_CHUNK_META_SIZE + nread;
uint8_t *payload = (uint8_t *)malloc(payload_len);
TransferChunkMeta meta;
int rc;
if (!payload) {
logger_log("ERROR", "client", "malloc_payload_failed len=%zu", payload_len);
atomic_store(&rt.running, 0);
exit_code = 1;
break;
}
if (transfer_start_send_ms == 0) {
transfer_start_send_ms = send_ts_ms;
}
/* 以“首次发送时间”为零点,将分片映射到按秒划分的发送窗口。 */
window_id = (uint32_t)((send_ts_ms - transfer_start_send_ms) / 1000u);
if (ensure_window_capacity(&window_counts, &window_cap, window_id) != OMNI_OK) {
free(payload);
logger_log("ERROR", "client", "window_counter_alloc_failed window=%u",
(unsigned)window_id);
atomic_store(&rt.running, 0);
exit_code = 1;
break;
}
omni_transfer_chunk_meta_encode(&meta,
transfer_id,
seq,
total_chunks,
window_id,
total_bytes,
offset,
(uint32_t)nread,
origin_ts_ms);
/* payload = chunk 元数据 + 实际文件数据。 */
memcpy(payload, &meta, TRANSFER_CHUNK_META_SIZE);
memcpy(payload + TRANSFER_CHUNK_META_SIZE, chunk, nread);
logger_on_processing_latency((double)(omni_now_ms() - process_t0));
rc = send_app_message_with_timestamp(ctx, MSG_TYPE_FILE_CHUNK,
payload, (uint32_t)payload_len,
send_ts_ms);
free(payload);
if (rc != OMNI_OK) {
atomic_store(&rt.running, 0);
exit_code = 1;
break;
}
window_counts[window_id]++;
if (window_id + 1u > total_windows) {
total_windows = window_id + 1u;
}
total_sent += nread;
offset += nread;
logger_set_progress(total_sent);
}
}
if (atomic_load(&rt.running)) {
TransferEndMeta end_meta;
file_end_send_ts_ms = omni_now_ms();
/* FILE_END 表示“数据流已经发完”,不是文件内容本身。 */
omni_transfer_end_meta_encode(&end_meta,
transfer_id,
total_chunks,
total_bytes,
total_windows);
if (send_app_message_with_timestamp(ctx, MSG_TYPE_FILE_END,
&end_meta, TRANSFER_END_META_SIZE,
file_end_send_ts_ms) != OMNI_OK) {
atomic_store(&rt.running, 0);
exit_code = 1;
}
}
logger_log("INFO", "client",
"file_transfer_done transfer_id=%u bytes=%llu total_chunks=%u",
(unsigned)transfer_id,
(unsigned long long)total_sent,
(unsigned)total_chunks);
if (wait_seconds < 0) {
/* keepalive 模式:发送完成后不主动退出,便于继续观察控制消息。 */
logger_log("INFO", "client", "keepalive_mode=on press_ctrl_c_to_exit");
while (atomic_load(&rt.running) && !g_stop) {
sleep(1);
}
} else {
/* 普通模式:等待一小段时间给服务端回 ACK。 */
for (int i = 0; i < wait_seconds && atomic_load(&rt.running) && !g_stop; ++i) {
if (atomic_load(&rt.ack_received)) {
break;
}
sleep(1);
}
}
{
/* 输出客户端视角的最终汇总,包括发送窗口分布和 ACK 结果。 */
OmniStats snapshot = logger_get_snapshot();
char *window_dist = format_window_distribution(window_counts, total_windows);
uint64_t tcp_original_bytes = saturating_sub_u64(snapshot.tcp_data_bytes_sent,
snapshot.tcp_retrans_bytes);
uint64_t kcp_original_bytes = saturating_sub_u64(snapshot.kcp_data_bytes_sent,
snapshot.kcp_retrans_bytes);
double tcp_retrans_rate = rate_percent(snapshot.tcp_retrans_bytes, tcp_original_bytes);
double kcp_retrans_rate = rate_percent(snapshot.kcp_retrans_bytes, kcp_original_bytes);
uint64_t ack_rtt_ms = atomic_load(&rt.ack_rtt_ms);
uint64_t ack_bytes_written = atomic_load(&rt.ack_bytes_written);
unsigned ack_received = (unsigned)atomic_load(&rt.ack_received);
logger_log("INFO", "summary",
"event=transfer_summary role=client proto=%s transfer_id=%u "
"total_bytes=%llu total_chunks=%u sent_bytes=%llu progress_bytes=%llu "
"tx_avg_mbps=%.3f tx_current_mbps=%.3f rx_avg_mbps=%.3f "
"processing_avg_ms=%.3f processing_max_ms=%.3f "
"queue_avg_ms=%.3f transmission_avg_ms=%.3f propagation_avg_ms=%.3f "
"last_rtt_ms=%llu min_rtt_ms=%llu "
"send_buffer_avg_pct=%.2f recv_buffer_avg_pct=%.2f "
"cwnd_avg=%.2f "
"tcp_retrans=%llu tcp_data_segs_out=%llu tcp_original_bytes=%llu "
"tcp_retrans_bytes=%llu tcp_retrans_rate_pct=%.2f "
"kcp_retrans=%llu kcp_data_segs_out=%llu kcp_original_bytes=%llu "
"kcp_retrans_bytes=%llu kcp_retrans_rate_pct=%.2f "
"send_windows=%u send_window_distribution=%s "
"ack_received=%u ack_rtt_ms=%llu ack_bytes_written=%llu "
"clock_sync_ok=%u clock_offset_ms=%lld clock_sync_rtt_ms=%llu clock_sync_samples=%u",
proto_str,
(unsigned)transfer_id,
(unsigned long long)total_bytes,
(unsigned)total_chunks,
(unsigned long long)total_sent,
(unsigned long long)snapshot.progress_bytes,
snapshot.tx_avg_mbps,
snapshot.tx_current_mbps,
snapshot.rx_avg_mbps,
(snapshot.processing_delay_ms.count == 0) ? 0.0 :
snapshot.processing_delay_ms.sum / (double)snapshot.processing_delay_ms.count,
snapshot.processing_delay_ms.max,
(snapshot.queue_delay_ms.count == 0) ? 0.0 :
snapshot.queue_delay_ms.sum / (double)snapshot.queue_delay_ms.count,
(snapshot.transmission_delay_ms.count == 0) ? 0.0 :
snapshot.transmission_delay_ms.sum / (double)snapshot.transmission_delay_ms.count,
(snapshot.propagation_delay_ms.count == 0) ? 0.0 :
snapshot.propagation_delay_ms.sum / (double)snapshot.propagation_delay_ms.count,
(unsigned long long)snapshot.last_rtt_ms,
(unsigned long long)((snapshot.min_rtt_ms == UINT64_MAX) ? 0 : snapshot.min_rtt_ms),
(snapshot.send_buffer_pct.count == 0) ? 0.0 :
snapshot.send_buffer_pct.sum / (double)snapshot.send_buffer_pct.count,
(snapshot.recv_buffer_pct.count == 0) ? 0.0 :
snapshot.recv_buffer_pct.sum / (double)snapshot.recv_buffer_pct.count,
(snapshot.cwnd.count == 0) ? 0.0 :
snapshot.cwnd.sum / (double)snapshot.cwnd.count,
(unsigned long long)snapshot.tcp_retrans,
(unsigned long long)snapshot.tcp_data_segs_out,
(unsigned long long)tcp_original_bytes,
(unsigned long long)snapshot.tcp_retrans_bytes,
tcp_retrans_rate,
(unsigned long long)snapshot.kcp_retrans,
(unsigned long long)snapshot.kcp_data_segs_out,
(unsigned long long)kcp_original_bytes,
(unsigned long long)snapshot.kcp_retrans_bytes,
kcp_retrans_rate,
(unsigned)total_windows,
window_dist ? window_dist : "alloc_failed",
ack_received,
(unsigned long long)ack_rtt_ms,
(unsigned long long)ack_bytes_written,
(unsigned)clock_sync.valid,
(long long)clock_sync.server_minus_client_offset_ms,
(unsigned long long)clock_sync.best_rtt_ms,
(unsigned)clock_sync.sample_count);
free(window_dist);
}
/* 主线程统一做收尾:停线程、释放缓冲、关闭文件和连接。 */
atomic_store(&rt.running, 0);
pthread_cancel(recv_tid);
pthread_join(recv_tid, NULL);
free(window_counts);
free(chunk);
fclose(fp);
omni_close(ctx);
return exit_code;
}

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@@ -1,278 +0,0 @@
/*
* relay_main.c
* 中转站:从 A 接收数据后立即转发到 B支持运行时动态修改转发目标
*
* 并发模型:
* - 主线程:阻塞接收上游流量并转发到当前目标
* - 控制线程:读取 stdin 命令,动态切换目标地址
*
* 线程安全策略:
* - tx_ctx / target_ip / target_port 受 tx_mu 互斥锁保护
* - 主线程转发发送与控制线程切换目标不会并发踩内存
*
* 控制命令stdin
* set <ip> <port> 修改目标地址
* show 打印当前目标
* quit 退出
*/
#include "common.h"
#include "network.h"
#include "logger.h"
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#define RELAY_BUF_SIZE (MSG_HEADER_SIZE + TRANSFER_CHUNK_META_SIZE + 65536u)
typedef struct RelayState {
/* 当前 relay 工作协议。 */
OmniProtocol proto;
/* 上游接收上下文(通常是服务端角色)。 */
OmniContext *rx_ctx;
/* 下游发送上下文(通常是客户端角色,可动态替换)。 */
OmniContext *tx_ctx;
/* 保护 tx_ctx 与目标地址信息。 */
pthread_mutex_t tx_mu;
/* 运行标志。 */
atomic_int running;
/* 当前目标地址快照(用于 show 命令与日志)。 */
char target_ip[64];
uint16_t target_port;
} RelayState;
static void usage(const char *prog)
{
fprintf(stderr,
"Usage:\n"
" %s -p tcp|udp|kcp -L <listen_port> -H <target_ip> -P <target_port>\n",
prog);
}
/* 将命令行协议字符串映射为内部协议枚举。 */
static OmniProtocol parse_proto(const char *s)
{
/* 非法输入回退 TCP。 */
if (!s) return OMNI_PROTO_TCP;
if (strcmp(s, "tcp") == 0) return OMNI_PROTO_TCP;
if (strcmp(s, "udp") == 0) return OMNI_PROTO_UDP;
if (strcmp(s, "kcp") == 0) return OMNI_PROTO_KCP;
return OMNI_PROTO_TCP;
}
/* 在运行期安全地切换下游转发目标。 */
static int relay_set_target(RelayState *st, const char *ip, uint16_t port)
{
/*
* 动态切换目标步骤:
* 1) 先建立新 tx_ctx失败时保持旧目标不变
* 2) 加锁替换指针与目标参数
* 3) 解锁后关闭旧 tx_ctx避免持锁做慢操作
*/
OmniContext *new_tx = omni_init(OMNI_ROLE_CLIENT, st->proto,
NULL, 0,
ip, port);
if (!new_tx) {
logger_log("ERROR", "relay", "connect_target_failed ip=%s port=%u",
ip, (unsigned)port);
return OMNI_ERR_IO;
}
pthread_mutex_lock(&st->tx_mu);
OmniContext *old_tx = st->tx_ctx;
st->tx_ctx = new_tx;
snprintf(st->target_ip, sizeof(st->target_ip), "%s", ip);
st->target_port = port;
pthread_mutex_unlock(&st->tx_mu);
if (old_tx) {
omni_close(old_tx);
}
logger_log("INFO", "relay", "target_updated ip=%s port=%u", ip, (unsigned)port);
return OMNI_OK;
}
/* 控制线程:解析 stdin 指令并驱动 show / set / quit。 */
static void *control_thread_main(void *arg)
{
/* 控制线程负责解析 stdin 命令。 */
RelayState *st = (RelayState *)arg;
char line[256];
while (atomic_load(&st->running)) {
if (!fgets(line, sizeof(line), stdin)) {
/*
* 管道/重定向 EOF 时不要立刻退出 relay
* - 清理 EOF 状态
* - 短暂休眠后继续循环
* 这样 relay 仍可继续处理主数据面转发。
*/
clearerr(stdin);
usleep(100 * 1000);
continue;
}
size_t len = strlen(line);
while (len > 0 && (line[len - 1] == '\n' || line[len - 1] == '\r')) {
line[--len] = '\0';
}
if (len == 0) {
continue;
}
if (strcmp(line, "quit") == 0) {
/* 通知主线程退出。 */
atomic_store(&st->running, 0);
break;
}
if (strcmp(line, "show") == 0) {
/* 在锁保护下读取目标快照,避免与 set 并发冲突。 */
pthread_mutex_lock(&st->tx_mu);
fprintf(stderr, "relay target: %s:%u\n",
st->target_ip[0] ? st->target_ip : "N/A",
(unsigned)st->target_port);
pthread_mutex_unlock(&st->tx_mu);
continue;
}
char ip[64];
unsigned port = 0;
if (sscanf(line, "set %63s %u", ip, &port) == 2 && port > 0 && port <= 65535u) {
/* 动态切目标。 */
relay_set_target(st, ip, (uint16_t)port);
continue;
}
fprintf(stderr, "unknown command: %s\n", line);
fprintf(stderr, "commands: set <ip> <port> | show | quit\n");
}
return NULL;
}
/*
* relay 主流程:
* - 上游用 server 角色收数据
* - 下游用 client 角色发数据
* - 主线程搬运数据,控制线程负责改目标
*/
int main(int argc, char **argv)
{
/* 命令行参数默认值。 */
const char *proto_str = "tcp";
const char *target_ip = NULL;
int listen_port = 0;
int target_port = 0;
int opt;
while ((opt = getopt(argc, argv, "p:L:H:P:")) != -1) {
switch (opt) {
case 'p':
proto_str = optarg;
break;
case 'L':
listen_port = atoi(optarg);
break;
case 'H':
target_ip = optarg;
break;
case 'P':
target_port = atoi(optarg);
break;
default:
usage(argv[0]);
return 1;
}
}
if (!target_ip || listen_port <= 0 || target_port <= 0) {
usage(argv[0]);
return 1;
}
RelayState st;
memset(&st, 0, sizeof(st));
st.proto = parse_proto(proto_str);
atomic_init(&st.running, 1);
pthread_mutex_init(&st.tx_mu, NULL);
/*
* rx_ctx 作为上游入口server 角色):
* - TCP: 等待上游 connect
* - UDP/KCP: 绑定监听端口接收上游包
*/
st.rx_ctx = omni_init(OMNI_ROLE_SERVER, st.proto, NULL, (uint16_t)listen_port, NULL, 0);
if (!st.rx_ctx) {
fprintf(stderr, "relay: omni_init rx failed\n");
pthread_mutex_destroy(&st.tx_mu);
return 1;
}
/*
* 初始目标连接失败不直接退出:
* - relay 可先启动数据入口
* - 后续通过 set 命令修复目标地址
*/
(void)relay_set_target(&st, target_ip, (uint16_t)target_port);
/* 启动控制面线程。 */
pthread_t ctrl_tid;
if (pthread_create(&ctrl_tid, NULL, control_thread_main, &st) != 0) {
perror("pthread_create");
omni_close(st.rx_ctx);
if (st.tx_ctx) omni_close(st.tx_ctx);
pthread_mutex_destroy(&st.tx_mu);
return 1;
}
uint8_t buf[RELAY_BUF_SIZE];
while (atomic_load(&st.running)) {
/* 数据面:收上游 -> 转发下游。 */
ssize_t n = omni_recv(st.rx_ctx, buf, sizeof(buf));
if (n < 0) {
logger_log("ERROR", "relay", "recv_failed n=%zd", n);
break;
}
if (n == 0) {
/* 暂时无数据时短暂退避。 */
usleep(2 * 1000);
continue;
}
ssize_t m = OMNI_ERR_PARAM;
/* 发送上下文受锁保护,防止与 set 命令并发替换。 */
pthread_mutex_lock(&st.tx_mu);
if (st.tx_ctx) {
m = omni_send(st.tx_ctx, buf, (size_t)n);
}
pthread_mutex_unlock(&st.tx_mu);
if (m != n) {
logger_log("ERROR", "relay", "forward_failed in=%zd out=%zd", n, m);
} else {
logger_log("INFO", "relay", "forward_ok bytes=%zd", n);
}
}
/* 收尾:先停主循环,再依次释放 rx / 控制线程 / tx。 */
atomic_store(&st.running, 0);
omni_close(st.rx_ctx);
pthread_join(ctrl_tid, NULL);
pthread_mutex_lock(&st.tx_mu);
OmniContext *tx = st.tx_ctx;
st.tx_ctx = NULL;
pthread_mutex_unlock(&st.tx_mu);
if (tx) {
omni_close(tx);
}
pthread_mutex_destroy(&st.tx_mu);
return 0;
}

View File

@@ -1,858 +0,0 @@
/*
* server_main.c
* 服务端:接收文件写盘,输出结构化传输统计
*/
#include "common.h"
#include "network.h"
#include "logger.h"
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#define SERVER_FRAME_BUF_SIZE (MSG_HEADER_SIZE + TRANSFER_CHUNK_META_SIZE + 65536u)
/*
* 一次文件传输在服务端视角下的累计状态。
* 它主要用来回答三个问题:
* 1) 收到了多少唯一分片
* 2) 有没有重复/乱序/缺失
* 3) 最终写盘了多少字节
*/
typedef struct TransferState {
uint32_t transfer_id; /* 当前传输 ID。 */
uint32_t total_chunks; /* 期望收到的总分片数。 */
uint32_t total_windows; /* 发送端告诉我们的总窗口数。 */
uint64_t total_bytes; /* 期望收到的总字节数。 */
uint32_t highest_seq_seen; /* 目前见过的最大序号,用于判断乱序。 */
uint32_t unique_chunks; /* 成功去重后收到的分片数。 */
uint32_t duplicate_chunks; /* 重复到达的分片数。 */
uint32_t out_of_order_chunks; /* 序号回退的分片数,表示存在乱序。 */
uint64_t unique_bytes_written; /* 实际写盘的唯一数据量。 */
uint8_t *seen; /* 位图/标记数组:某个 seq 是否已收到。 */
size_t seen_cap; /* seen 数组容量。 */
uint64_t *recv_window_counts; /* 每个时间窗口收到的分片数。 */
size_t recv_window_cap; /* recv_window_counts 容量。 */
} TransferState;
typedef struct ClockSyncState {
int valid; /* 是否已经拿到客户端确认的 offset。 */
int64_t server_minus_client_offset_ms; /* server_time - client_time */
uint64_t best_rtt_ms; /* 客户端选中的最优 RTT 样本。 */
uint32_t sample_count; /* 参与选择的有效样本数。 */
} ClockSyncState;
/* 服务端运行期上下文:连接、写盘目标、线程状态和传输统计都放在这里。 */
typedef struct ServerRuntime {
OmniContext *ctx;
OmniProtocol proto;
FILE *out_fp;
atomic_int running;
atomic_int transfer_done;
uint64_t bytes_written;
TransferState transfer;
ClockSyncState clock_sync;
} ServerRuntime;
/* 打印服务端命令行帮助。 */
static void usage(const char *prog)
{
fprintf(stderr,
"Usage:\n"
" %s -p tcp|udp|kcp -P <listen_port> -o <output_file> [-b <bind_ip>]\n",
prog);
}
/* 将字符串协议名解析为内部枚举,非法值默认回退 TCP。 */
static OmniProtocol parse_proto(const char *s)
{
if (!s) return OMNI_PROTO_TCP;
if (strcmp(s, "tcp") == 0) return OMNI_PROTO_TCP;
if (strcmp(s, "udp") == 0) return OMNI_PROTO_UDP;
if (strcmp(s, "kcp") == 0) return OMNI_PROTO_KCP;
return OMNI_PROTO_TCP;
}
/* 与客户端对称:负责发送完整业务帧。 */
static int send_app_message_with_timestamp(OmniContext *ctx,
uint32_t type,
const void *payload,
uint32_t payload_len,
uint64_t timestamp_ms)
{
size_t total_len = MSG_HEADER_SIZE + (size_t)payload_len;
uint8_t *frame = (uint8_t *)malloc(total_len);
if (!frame) {
logger_log("ERROR", "server", "malloc_frame_failed len=%zu", total_len);
return OMNI_ERR_GENERIC;
}
MsgHeader hdr;
omni_msg_header_encode(&hdr, type, payload_len, timestamp_ms);
memcpy(frame, &hdr, MSG_HEADER_SIZE);
if (payload_len > 0 && payload) {
memcpy(frame + MSG_HEADER_SIZE, payload, payload_len);
}
ssize_t n = omni_send(ctx, frame, total_len);
free(frame);
if (n != (ssize_t)total_len) {
logger_log("ERROR", "server",
"omni_send_failed expect=%zu got=%zd type=%u",
total_len, n, (unsigned)type);
return OMNI_ERR_IO;
}
return OMNI_OK;
}
/* 与客户端对称:拆包并校验应用层帧长度。 */
static int decode_app_message(const uint8_t *frame,
size_t frame_len,
MsgHeader *out_hdr,
const uint8_t **out_payload)
{
if (!frame || frame_len < MSG_HEADER_SIZE || !out_hdr || !out_payload) {
return OMNI_ERR_PARAM;
}
MsgHeader net_hdr;
memcpy(&net_hdr, frame, MSG_HEADER_SIZE);
omni_msg_header_decode(&net_hdr, out_hdr);
if ((size_t)out_hdr->len + MSG_HEADER_SIZE != frame_len) {
return OMNI_ERR_IO;
}
*out_payload = frame + MSG_HEADER_SIZE;
return OMNI_OK;
}
/* 用客户端确认的 offset把 origin_ts_ms 转成服务端时钟系并计算端到端时延。 */
static double compute_compensated_end_to_end_ms(const ServerRuntime *rt,
uint64_t server_recv_ts_ms,
const TransferChunkMeta *meta)
{
int64_t e2e_ms;
if (!rt || !meta || !rt->clock_sync.valid) {
return -1.0;
}
e2e_ms = (int64_t)server_recv_ts_ms -
((int64_t)meta->origin_ts_ms + rt->clock_sync.server_minus_client_offset_ms);
if (e2e_ms < 0) {
return 0.0;
}
return (double)e2e_ms;
}
/* 扩容 seen 标记数组,下标直接使用 seq 值。 */
static int ensure_seen_capacity(TransferState *st, uint32_t total_chunks)
{
size_t need = (size_t)total_chunks + 1u;
uint8_t *new_seen;
if (need <= st->seen_cap) {
return OMNI_OK;
}
new_seen = (uint8_t *)realloc(st->seen, need);
if (!new_seen) {
return OMNI_ERR_GENERIC;
}
memset(new_seen + st->seen_cap, 0, need - st->seen_cap);
st->seen = new_seen;
st->seen_cap = need;
return OMNI_OK;
}
/* 扩容按窗口聚合的接收计数数组。 */
static int ensure_window_capacity(uint64_t **counts, size_t *cap, uint32_t window_id)
{
size_t need = (size_t)window_id + 1u;
size_t new_cap;
uint64_t *new_counts;
if (need <= *cap) {
return OMNI_OK;
}
new_cap = (*cap == 0) ? 8u : *cap;
while (new_cap < need) {
new_cap *= 2u;
}
new_counts = (uint64_t *)realloc(*counts, new_cap * sizeof(uint64_t));
if (!new_counts) {
return OMNI_ERR_GENERIC;
}
memset(new_counts + *cap, 0, (new_cap - *cap) * sizeof(uint64_t));
*counts = new_counts;
*cap = new_cap;
return OMNI_OK;
}
/* 将接收窗口分布转成日志友好的字符串。 */
static char *format_window_distribution(const uint64_t *counts, uint32_t total_windows)
{
size_t cap = 256;
size_t len = 0;
char *buf = (char *)malloc(cap);
if (!buf) {
return NULL;
}
buf[0] = '\0';
for (uint32_t i = 0; i < total_windows; ++i) {
char tmp[64];
int n;
if (counts[i] == 0) {
continue;
}
n = snprintf(tmp, sizeof(tmp), "%s%u:%llu",
(len == 0) ? "" : ",",
(unsigned)i,
(unsigned long long)counts[i]);
if (n <= 0) {
continue;
}
while (len + (size_t)n + 1 > cap) {
char *new_buf;
cap *= 2u;
new_buf = (char *)realloc(buf, cap);
if (!new_buf) {
free(buf);
return NULL;
}
buf = new_buf;
}
memcpy(buf + len, tmp, (size_t)n);
len += (size_t)n;
buf[len] = '\0';
}
if (len == 0) {
snprintf(buf, cap, "none");
}
return buf;
}
static uint64_t saturating_sub_u64(uint64_t total, uint64_t delta)
{
return (total > delta) ? (total - delta) : 0;
}
static double rate_percent(uint64_t numerator, uint64_t denominator)
{
if (denominator == 0) {
return 0.0;
}
return ((double)numerator * 100.0) / (double)denominator;
}
/*
* 根据 seen 位图生成缺失区间字符串,例如 "3-5,9,12-14"。
* 同时顺手统计缺失 chunk 数、突发丢包次数和最长突发长度。
*/
static char *format_missing_ranges(const TransferState *st,
uint32_t *out_missing_chunks,
uint32_t *out_burst_count,
uint32_t *out_max_burst_len)
{
size_t cap = 256;
size_t len = 0;
char *buf = (char *)malloc(cap);
uint32_t missing_chunks = 0;
uint32_t burst_count = 0;
uint32_t max_burst_len = 0;
if (!buf) {
return NULL;
}
buf[0] = '\0';
if (st->total_chunks == 0 || !st->seen) {
snprintf(buf, cap, "none");
if (out_missing_chunks) *out_missing_chunks = 0;
if (out_burst_count) *out_burst_count = 0;
if (out_max_burst_len) *out_max_burst_len = 0;
return buf;
}
for (uint32_t seq = 1; seq <= st->total_chunks; ++seq) {
if (st->seen[seq] != 0) {
continue;
}
uint32_t start = seq;
uint32_t end = seq;
char tmp[64];
int n;
/* 把连续缺失的 seq 合并成一个区间,便于读日志时看出 burst loss。 */
while (end + 1u <= st->total_chunks && st->seen[end + 1u] == 0) {
end++;
}
missing_chunks += (end - start + 1u);
burst_count++;
if ((end - start + 1u) > max_burst_len) {
max_burst_len = end - start + 1u;
}
if (start == end) {
n = snprintf(tmp, sizeof(tmp), "%s%u",
(len == 0) ? "" : ",",
(unsigned)start);
} else {
n = snprintf(tmp, sizeof(tmp), "%s%u-%u",
(len == 0) ? "" : ",",
(unsigned)start,
(unsigned)end);
}
if (n > 0) {
while (len + (size_t)n + 1 > cap) {
char *new_buf;
cap *= 2u;
new_buf = (char *)realloc(buf, cap);
if (!new_buf) {
free(buf);
return NULL;
}
buf = new_buf;
}
memcpy(buf + len, tmp, (size_t)n);
len += (size_t)n;
buf[len] = '\0';
}
seq = end;
}
if (len == 0) {
snprintf(buf, cap, "none");
}
if (out_missing_chunks) *out_missing_chunks = missing_chunks;
if (out_burst_count) *out_burst_count = burst_count;
if (out_max_burst_len) *out_max_burst_len = max_burst_len;
return buf;
}
/* 用首个有效分片补齐传输的基础元数据,后续分片只做一致性延续。 */
static void transfer_state_init(TransferState *st, const TransferChunkMeta *meta)
{
if (st->transfer_id == 0) {
st->transfer_id = meta->transfer_id;
}
if (st->total_chunks == 0) {
st->total_chunks = meta->total_chunks;
}
if (st->total_bytes == 0) {
st->total_bytes = meta->total_bytes;
}
if (meta->window_id + 1u > st->total_windows) {
st->total_windows = meta->window_id + 1u;
}
}
/*
* 处理一个 FILE_CHUNK
* - 校验元数据
* - 去重 / 统计乱序
* - 估算时延拆分
* - 按 offset 随机写盘
*/
static int server_record_chunk(ServerRuntime *rt,
const MsgHeader *hdr,
const uint8_t *payload,
uint32_t payload_len)
{
TransferChunkMeta meta;
const uint8_t *chunk_data;
uint64_t process_t0;
(void)hdr;
if (payload_len < TRANSFER_CHUNK_META_SIZE) {
logger_log("ERROR", "server", "short_chunk_payload len=%u", (unsigned)payload_len);
return OMNI_ERR_IO;
}
process_t0 = omni_now_ms();
omni_transfer_chunk_meta_decode((const TransferChunkMeta *)payload, &meta);
if (meta.chunk_bytes > payload_len - TRANSFER_CHUNK_META_SIZE) {
logger_log("ERROR", "server",
"invalid_chunk_meta transfer_id=%u seq=%u chunk_bytes=%u payload_len=%u",
(unsigned)meta.transfer_id,
(unsigned)meta.seq,
(unsigned)meta.chunk_bytes,
(unsigned)payload_len);
return OMNI_ERR_IO;
}
transfer_state_init(&rt->transfer, &meta);
if (ensure_seen_capacity(&rt->transfer, meta.total_chunks) != OMNI_OK) {
logger_log("ERROR", "server", "seen_bitmap_alloc_failed total_chunks=%u",
(unsigned)meta.total_chunks);
return OMNI_ERR_GENERIC;
}
if (ensure_window_capacity(&rt->transfer.recv_window_counts,
&rt->transfer.recv_window_cap,
meta.window_id) != OMNI_OK) {
logger_log("ERROR", "server", "recv_window_alloc_failed window=%u",
(unsigned)meta.window_id);
return OMNI_ERR_GENERIC;
}
if (meta.seq == 0 || meta.seq > meta.total_chunks) {
logger_log("WARN", "server", "invalid_seq transfer_id=%u seq=%u total_chunks=%u",
(unsigned)meta.transfer_id,
(unsigned)meta.seq,
(unsigned)meta.total_chunks);
return OMNI_ERR_IO;
}
/* seen 位图用于去重:重复片不再写盘,但会计入 duplicate_chunks。 */
if (rt->transfer.seen[meta.seq] != 0) {
rt->transfer.duplicate_chunks++;
return OMNI_OK;
}
rt->transfer.seen[meta.seq] = 1;
rt->transfer.unique_chunks++;
rt->transfer.recv_window_counts[meta.window_id]++;
/* 若当前序号小于历史最大序号,说明该片到达顺序发生了回退。 */
if (meta.seq < rt->transfer.highest_seq_seen) {
rt->transfer.out_of_order_chunks++;
}
if (meta.seq > rt->transfer.highest_seq_seen) {
rt->transfer.highest_seq_seen = meta.seq;
}
{
double e2e_ms = compute_compensated_end_to_end_ms(rt, process_t0, &meta);
if (e2e_ms >= 0.0) {
logger_on_end_to_end_latency(e2e_ms);
}
}
logger_set_transfer_total(meta.total_bytes);
chunk_data = payload + TRANSFER_CHUNK_META_SIZE;
/*
* 必须按 offset 写盘,而不是简单 append。
* 这样即便 UDP/KCP 发生乱序,输出文件仍能恢复到正确位置。
*/
if (fseeko(rt->out_fp, (off_t)meta.offset_bytes, SEEK_SET) != 0) {
logger_log("ERROR", "server",
"fseeko_failed transfer_id=%u seq=%u offset=%llu",
(unsigned)meta.transfer_id,
(unsigned)meta.seq,
(unsigned long long)meta.offset_bytes);
return OMNI_ERR_IO;
}
if (fwrite(chunk_data, 1, meta.chunk_bytes, rt->out_fp) != meta.chunk_bytes) {
logger_log("ERROR", "server",
"fwrite_failed transfer_id=%u seq=%u chunk_bytes=%u",
(unsigned)meta.transfer_id,
(unsigned)meta.seq,
(unsigned)meta.chunk_bytes);
return OMNI_ERR_IO;
}
logger_on_processing_latency((double)(omni_now_ms() - process_t0));
rt->transfer.unique_bytes_written += meta.chunk_bytes;
rt->bytes_written = rt->transfer.unique_bytes_written;
logger_set_progress(rt->bytes_written);
return OMNI_OK;
}
/* 输出服务端视角的最终汇总,包括丢包区间、乱序和时延统计。 */
static void server_log_transfer_summary(ServerRuntime *rt)
{
OmniStats snapshot = logger_get_snapshot();
char *missing_ranges = NULL;
char *recv_window_dist = NULL;
uint32_t missing_chunks = 0;
uint32_t burst_count = 0;
uint32_t max_burst_len = 0;
double loss_rate = 0.0;
double tcp_retrans_rate = 0.0;
double kcp_retrans_rate = 0.0;
uint64_t tcp_original_bytes = saturating_sub_u64(snapshot.tcp_data_bytes_sent,
snapshot.tcp_retrans_bytes);
uint64_t kcp_original_bytes = saturating_sub_u64(snapshot.kcp_data_bytes_sent,
snapshot.kcp_retrans_bytes);
missing_ranges = format_missing_ranges(&rt->transfer,
&missing_chunks,
&burst_count,
&max_burst_len);
recv_window_dist = format_window_distribution(rt->transfer.recv_window_counts,
rt->transfer.total_windows);
if (rt->transfer.total_chunks > 0) {
loss_rate = ((double)missing_chunks * 100.0) / (double)rt->transfer.total_chunks;
}
tcp_retrans_rate = rate_percent(snapshot.tcp_retrans_bytes, tcp_original_bytes);
kcp_retrans_rate = rate_percent(snapshot.kcp_retrans_bytes, kcp_original_bytes);
logger_log("INFO", "summary",
"event=transfer_summary role=server proto=%d transfer_id=%u "
"expected_bytes=%llu written_bytes=%llu expected_chunks=%u unique_chunks=%u "
"duplicate_chunks=%u out_of_order_chunks=%u "
"loss_rate_pct=%.2f missing_chunks=%u missing_ranges=%s "
"burst_loss_count=%u max_burst_loss_len=%u recv_windows=%u recv_window_distribution=%s "
"rx_avg_mbps=%.3f rx_current_mbps=%.3f "
"processing_avg_ms=%.3f queue_avg_ms=%.3f transmission_avg_ms=%.3f "
"propagation_avg_ms=%.3f end_to_end_avg_ms=%.3f "
"send_buffer_avg_pct=%.2f recv_buffer_avg_pct=%.2f "
"cwnd_avg=%.2f "
"local_tcp_retrans=%llu local_tcp_data_segs_out=%llu local_tcp_original_bytes=%llu "
"local_tcp_retrans_bytes=%llu local_tcp_retrans_rate_pct=%.2f "
"local_kcp_retrans=%llu local_kcp_data_segs_out=%llu local_kcp_original_bytes=%llu "
"local_kcp_retrans_bytes=%llu local_kcp_retrans_rate_pct=%.2f "
"clock_sync_ok=%u clock_offset_ms=%lld clock_sync_rtt_ms=%llu clock_sync_samples=%u",
(int)rt->proto,
(unsigned)rt->transfer.transfer_id,
(unsigned long long)rt->transfer.total_bytes,
(unsigned long long)rt->bytes_written,
(unsigned)rt->transfer.total_chunks,
(unsigned)rt->transfer.unique_chunks,
(unsigned)rt->transfer.duplicate_chunks,
(unsigned)rt->transfer.out_of_order_chunks,
loss_rate,
(unsigned)missing_chunks,
missing_ranges ? missing_ranges : "alloc_failed",
(unsigned)burst_count,
(unsigned)max_burst_len,
(unsigned)rt->transfer.total_windows,
recv_window_dist ? recv_window_dist : "alloc_failed",
snapshot.rx_avg_mbps,
snapshot.rx_current_mbps,
(snapshot.processing_delay_ms.count == 0) ? 0.0 :
snapshot.processing_delay_ms.sum / (double)snapshot.processing_delay_ms.count,
(snapshot.queue_delay_ms.count == 0) ? 0.0 :
snapshot.queue_delay_ms.sum / (double)snapshot.queue_delay_ms.count,
(snapshot.transmission_delay_ms.count == 0) ? 0.0 :
snapshot.transmission_delay_ms.sum / (double)snapshot.transmission_delay_ms.count,
(snapshot.propagation_delay_ms.count == 0) ? 0.0 :
snapshot.propagation_delay_ms.sum / (double)snapshot.propagation_delay_ms.count,
(snapshot.end_to_end_delay_ms.count == 0) ? 0.0 :
snapshot.end_to_end_delay_ms.sum / (double)snapshot.end_to_end_delay_ms.count,
(snapshot.send_buffer_pct.count == 0) ? 0.0 :
snapshot.send_buffer_pct.sum / (double)snapshot.send_buffer_pct.count,
(snapshot.recv_buffer_pct.count == 0) ? 0.0 :
snapshot.recv_buffer_pct.sum / (double)snapshot.recv_buffer_pct.count,
(snapshot.cwnd.count == 0) ? 0.0 :
snapshot.cwnd.sum / (double)snapshot.cwnd.count,
(unsigned long long)snapshot.tcp_retrans,
(unsigned long long)snapshot.tcp_data_segs_out,
(unsigned long long)tcp_original_bytes,
(unsigned long long)snapshot.tcp_retrans_bytes,
tcp_retrans_rate,
(unsigned long long)snapshot.kcp_retrans,
(unsigned long long)snapshot.kcp_data_segs_out,
(unsigned long long)kcp_original_bytes,
(unsigned long long)snapshot.kcp_retrans_bytes,
kcp_retrans_rate,
(unsigned)rt->clock_sync.valid,
(long long)rt->clock_sync.server_minus_client_offset_ms,
(unsigned long long)rt->clock_sync.best_rtt_ms,
(unsigned)rt->clock_sync.sample_count);
free(missing_ranges);
free(recv_window_dist);
}
/* 释放 TransferState 持有的动态内存。 */
static void transfer_state_free(TransferState *st)
{
free(st->seen);
free(st->recv_window_counts);
memset(st, 0, sizeof(*st));
}
/*
* 服务端接收线程:
* - 接收 FILE_CHUNK 并写盘
* - 接收 FILE_END 并输出汇总、回 ACK
* - 接收 COMMAND 仅记录日志
*/
static void *recv_thread_main(void *arg)
{
ServerRuntime *rt = (ServerRuntime *)arg;
uint8_t frame[SERVER_FRAME_BUF_SIZE];
pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);
while (atomic_load(&rt->running)) {
ssize_t n = omni_recv(rt->ctx, frame, sizeof(frame));
if (n < 0) {
logger_log("ERROR", "server", "recv_failed n=%zd", n);
break;
}
if (n == 0) {
if (rt->proto == OMNI_PROTO_TCP) {
logger_log("INFO", "server", "tcp_peer_closed");
break;
}
usleep(2 * 1000);
continue;
}
MsgHeader hdr;
const uint8_t *payload = NULL;
int rc = decode_app_message(frame, (size_t)n, &hdr, &payload);
if (rc != OMNI_OK) {
logger_log("ERROR", "server", "invalid_app_frame bytes=%zd rc=%d", n, rc);
continue;
}
if (hdr.type == MSG_TYPE_FILE_CHUNK) {
rc = server_record_chunk(rt, &hdr, payload, hdr.len);
if (rc != OMNI_OK) {
break;
}
} else if (hdr.type == MSG_TYPE_TIME_SYNC_REQ) {
TimeSyncProbeMeta probe_meta;
TimeSyncReplyMeta reply_meta;
uint64_t server_recv_ts_ms;
uint64_t server_send_ts_ms;
if (hdr.len < TIME_SYNC_PROBE_META_SIZE) {
logger_log("WARN", "server",
"short_time_sync_probe len=%u",
(unsigned)hdr.len);
continue;
}
server_recv_ts_ms = omni_now_ms();
omni_time_sync_probe_meta_decode((const TimeSyncProbeMeta *)payload, &probe_meta);
server_send_ts_ms = omni_now_ms();
omni_time_sync_reply_meta_encode(&reply_meta,
probe_meta.probe_id,
probe_meta.client_send_ts_ms,
server_recv_ts_ms,
server_send_ts_ms);
if (send_app_message_with_timestamp(rt->ctx,
MSG_TYPE_TIME_SYNC_RESP,
&reply_meta,
TIME_SYNC_REPLY_META_SIZE,
server_send_ts_ms) != OMNI_OK) {
logger_log("ERROR", "server",
"time_sync_reply_failed probe_id=%u",
(unsigned)probe_meta.probe_id);
break;
}
logger_log("INFO", "server",
"time_sync_reply probe_id=%u server_recv_ts_ms=%llu server_send_ts_ms=%llu",
(unsigned)probe_meta.probe_id,
(unsigned long long)server_recv_ts_ms,
(unsigned long long)server_send_ts_ms);
continue;
} else if (hdr.type == MSG_TYPE_TIME_SYNC_REPORT) {
TimeSyncReportMeta report_meta;
if (hdr.len < TIME_SYNC_REPORT_META_SIZE) {
logger_log("WARN", "server",
"short_time_sync_report len=%u",
(unsigned)hdr.len);
continue;
}
omni_time_sync_report_meta_decode((const TimeSyncReportMeta *)payload, &report_meta);
rt->clock_sync.valid = 1;
rt->clock_sync.server_minus_client_offset_ms =
report_meta.server_minus_client_offset_ms;
rt->clock_sync.best_rtt_ms = report_meta.best_rtt_ms;
rt->clock_sync.sample_count = report_meta.sample_count;
logger_on_rtt(report_meta.best_rtt_ms);
logger_log("INFO", "server",
"time_sync_ready offset_ms=%lld best_rtt_ms=%llu samples=%u",
(long long)rt->clock_sync.server_minus_client_offset_ms,
(unsigned long long)rt->clock_sync.best_rtt_ms,
(unsigned)rt->clock_sync.sample_count);
continue;
} else if (hdr.type == MSG_TYPE_FILE_END) {
TransferEndMeta end_meta;
TransferAckMeta ack_meta;
if (hdr.len >= TRANSFER_END_META_SIZE) {
/* FILE_END 会补齐整次传输的最终期望规模信息。 */
omni_transfer_end_meta_decode((const TransferEndMeta *)payload, &end_meta);
if (rt->transfer.transfer_id == 0) {
rt->transfer.transfer_id = end_meta.transfer_id;
}
if (rt->transfer.total_chunks == 0) {
rt->transfer.total_chunks = end_meta.total_chunks;
}
if (rt->transfer.total_bytes == 0) {
rt->transfer.total_bytes = end_meta.total_bytes;
}
if (end_meta.total_windows > rt->transfer.total_windows) {
rt->transfer.total_windows = end_meta.total_windows;
}
logger_set_transfer_total(rt->transfer.total_bytes);
}
/* 先确保文件内容刷到磁盘,再输出汇总并回 ACK。 */
fflush(rt->out_fp);
server_log_transfer_summary(rt);
atomic_store(&rt->transfer_done, 1);
logger_log("INFO", "server", "file_transfer_end bytes=%llu",
(unsigned long long)rt->bytes_written);
omni_transfer_ack_meta_encode(&ack_meta,
rt->transfer.transfer_id,
rt->transfer.total_chunks,
rt->transfer.total_bytes,
rt->bytes_written,
hdr.timestamp);
(void)send_app_message_with_timestamp(rt->ctx,
MSG_TYPE_TRANSFER_ACK,
&ack_meta,
TRANSFER_ACK_META_SIZE,
omni_now_ms());
continue;
} else if (hdr.type == MSG_TYPE_COMMAND) {
logger_log("INFO", "server",
"recv_command_from_peer len=%u (ignored)",
(unsigned)hdr.len);
} else {
logger_log("INFO", "server",
"recv_unknown_type type=%u len=%u",
(unsigned)hdr.type, (unsigned)hdr.len);
}
}
atomic_store(&rt->running, 0);
return NULL;
}
/*
* 服务端主流程:
* 1) 建立监听/接收上下文
* 2) 启动后台接收线程
* 3) 如果前台是交互终端,则允许输入命令发给客户端
* 4) 等待传输完成后统一收尾
*/
int main(int argc, char **argv)
{
const char *proto_str = "tcp";
const char *bind_ip = NULL;
const char *output_path = NULL;
OmniProtocol proto;
FILE *out_fp = NULL;
OmniContext *ctx = NULL;
ServerRuntime rt;
pthread_t recv_tid;
int listen_port = 0;
int opt;
while ((opt = getopt(argc, argv, "p:b:P:o:")) != -1) {
switch (opt) {
case 'p':
proto_str = optarg;
break;
case 'b':
bind_ip = optarg;
break;
case 'P':
listen_port = atoi(optarg);
break;
case 'o':
output_path = optarg;
break;
default:
usage(argv[0]);
return 1;
}
}
if (listen_port <= 0 || !output_path) {
usage(argv[0]);
return 1;
}
out_fp = fopen(output_path, "wb+");
if (!out_fp) {
perror("fopen");
return 1;
}
proto = parse_proto(proto_str);
ctx = omni_init(OMNI_ROLE_SERVER, proto,
bind_ip, (uint16_t)listen_port,
NULL, 0);
if (!ctx) {
fclose(out_fp);
fprintf(stderr, "omni_init failed\n");
return 1;
}
memset(&rt, 0, sizeof(rt));
rt.ctx = ctx;
rt.proto = proto;
rt.out_fp = out_fp;
atomic_init(&rt.running, 1);
atomic_init(&rt.transfer_done, 0);
if (pthread_create(&recv_tid, NULL, recv_thread_main, &rt) != 0) {
perror("pthread_create");
omni_close(ctx);
fclose(out_fp);
return 1;
}
if (isatty(STDIN_FILENO)) {
/* 交互模式:允许服务端向客户端发送文本命令。 */
char line[2048];
while (atomic_load(&rt.running)) {
if (!fgets(line, sizeof(line), stdin)) {
break;
}
size_t len = strlen(line);
while (len > 0 && (line[len - 1] == '\n' || line[len - 1] == '\r')) {
line[--len] = '\0';
}
if (len == 0) {
continue;
}
if (strcmp(line, "quit") == 0) {
break;
}
if (send_app_message_with_timestamp(ctx,
MSG_TYPE_COMMAND,
line,
(uint32_t)len,
omni_now_ms()) != OMNI_OK) {
logger_log("ERROR", "server", "send_command_failed");
break;
}
}
} else {
/* 非交互模式:例如脚本执行时,只等待传输结束即可。 */
while (atomic_load(&rt.running) && !atomic_load(&rt.transfer_done)) {
usleep(100 * 1000);
}
}
/* 主线程统一收尾并释放接收线程持有的资源。 */
atomic_store(&rt.running, 0);
pthread_cancel(recv_tid);
pthread_join(recv_tid, NULL);
omni_close(ctx);
fclose(out_fp);
transfer_state_free(&rt.transfer);
return 0;
}

View File

@@ -1,190 +0,0 @@
/*
* test_main.c
* 简单测试程序:客户端/服务端双向传输 + 日志观测
*
* 使用方式示例(同一机器上,两个终端):
* 终端1./omni_test -r server -p tcp -P 9000
* 终端2./omni_test -r client -p tcp -P 9000 -H 127.0.0.1
*
* 协议可选tcp / udp / kcp
*/
#include "common.h"
#include "network.h"
#include "logger.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
/* 打印测试程序命令行帮助。 */
static void usage(const char *prog)
{
fprintf(stderr,
"Usage:\n"
" %s -r server -p tcp|udp|kcp -P <port>\n"
" %s -r client -p tcp|udp|kcp -P <port> -H <host>\n",
prog, prog);
}
/* 解析协议名,非法输入时默认回退 TCP。 */
static OmniProtocol parse_proto(const char *s)
{
if (strcmp(s, "tcp") == 0) return OMNI_PROTO_TCP;
if (strcmp(s, "udp") == 0) return OMNI_PROTO_UDP;
if (strcmp(s, "kcp") == 0) return OMNI_PROTO_KCP;
return OMNI_PROTO_TCP;
}
/*
* 测试服务端:
* 持续收消息并原样 echo 回去,主要用于验证双向收发路径是否通畅。
*/
static void run_server(OmniProtocol proto, uint16_t port)
{
OmniContext *ctx = omni_init(OMNI_ROLE_SERVER, proto,
NULL, port,
NULL, 0);
if (!ctx) {
fprintf(stderr, "server: omni_init failed\n");
return;
}
logger_log("INFO", "test", "server_started proto=%d port=%u",
(int)proto, (unsigned)port);
char buf[4096];
for (;;) {
ssize_t n = omni_recv(ctx, buf, sizeof(buf));
if (n < 0) {
logger_log("INFO", "test", "server_recv_end n=%zd", n);
break;
}
if (n == 0) {
/* KCP 在“暂时没拼出完整消息”时可能返回 0因此这里不能立刻判定连接结束。 */
if (proto == OMNI_PROTO_KCP) {
usleep(10 * 1000);
continue;
}
logger_log("INFO", "test", "server_recv_end n=%zd", n);
break;
}
logger_log("INFO", "test", "server_recv bytes=%zd", n);
/* 简单 echo 回客户端,验证双向通信 */
ssize_t m = omni_send(ctx, buf, (size_t)n);
logger_log("INFO", "test", "server_echo bytes=%zd", m);
}
omni_close(ctx);
}
/*
* 测试客户端:
* 连续发 100 条消息,每发一条就等一条 echo用来观测协议往返行为。
*/
static void run_client(OmniProtocol proto, const char *host, uint16_t port)
{
if (!host) {
fprintf(stderr, "client: host is required\n");
return;
}
OmniContext *ctx = omni_init(OMNI_ROLE_CLIENT, proto,
NULL, 0,
host, port);
if (!ctx) {
fprintf(stderr, "client: omni_init failed\n");
return;
}
logger_log("INFO", "test", "client_started proto=%d host=%s port=%u",
(int)proto, host, (unsigned)port);
char send_buf[2048];
char recv_buf[4096];
for (int i = 0; i < 100; ++i) {
int len = snprintf(send_buf, sizeof(send_buf),
"msg=%d time_ms=%llu payload_size=%zu",
i,
(unsigned long long)omni_now_ms(),
sizeof(send_buf));
ssize_t n = omni_send(ctx, send_buf, (size_t)len);
logger_log("INFO", "test", "client_send i=%d bytes=%zd", i, n);
if (n <= 0) break;
ssize_t m = omni_recv(ctx, recv_buf, sizeof(recv_buf));
if (m < 0) {
logger_log("INFO", "test", "client_recv_end i=%d bytes=%zd", i, m);
break;
}
if (m == 0) {
/* 对 KCP 来说0 更像“此刻没取到完整消息”,而不是 socket 关闭。 */
if (proto == OMNI_PROTO_KCP) {
usleep(10 * 1000);
--i;
continue;
}
logger_log("INFO", "test", "client_recv_end i=%d bytes=%zd", i, m);
break;
}
logger_log("INFO", "test",
"client_recv_echo i=%d bytes=%zd first_bytes=\"%.32s\"",
i, m, recv_buf);
usleep(10 * 1000); /* 10ms 间隔,模拟稳定流量 */
}
omni_close(ctx);
}
/* 程序入口:根据角色参数派发到测试客户端或测试服务端。 */
int main(int argc, char **argv)
{
const char *role_str = NULL;
const char *proto_str = "tcp";
const char *host = NULL;
int port = 0;
int opt;
while ((opt = getopt(argc, argv, "r:p:P:H:")) != -1) {
switch (opt) {
case 'r':
role_str = optarg;
break;
case 'p':
proto_str = optarg;
break;
case 'P':
port = atoi(optarg);
break;
case 'H':
host = optarg;
break;
default:
usage(argv[0]);
return 1;
}
}
if (!role_str || port <= 0) {
usage(argv[0]);
return 1;
}
OmniProtocol proto = parse_proto(proto_str);
if (strcmp(role_str, "server") == 0) {
run_server(proto, (uint16_t)port);
} else if (strcmp(role_str, "client") == 0) {
run_client(proto, host, (uint16_t)port);
} else {
usage(argv[0]);
return 1;
}
return 0;
}

View File

@@ -23,6 +23,11 @@ static FILE *g_json_fp = NULL;
static int g_min_level = 1; /* default INFO */
static int g_initialized = 0;
static pthread_mutex_t g_mu = PTHREAD_MUTEX_INITIALIZER;
static char g_ctx_app[32];
static char g_ctx_proto[16];
static char g_ctx_mode[16];
static char g_ctx_role[16];
static char g_ctx_self_id[OMNI_PEER_ID_SIZE];
/* 对 common.h 的时间接口做一层薄包装,统一 logger 模块内部的调用入口。 */
static uint64_t now_ms(void)
@@ -30,6 +35,23 @@ static uint64_t now_ms(void)
return omni_now_ms();
}
static void copy_context_field(char *dst,
size_t dst_sz,
const char *src,
const char *fallback)
{
if (!dst || dst_sz == 0) {
return;
}
if (src && src[0] != '\0') {
omni_copy_fixed_ascii(dst, dst_sz, src);
return;
}
if (fallback) {
omni_copy_fixed_ascii(dst, dst_sz, fallback);
}
}
/* 将字符串日志级别映射为可比较的整数,数值越大表示级别越高。 */
static int level_to_int(const char *level)
{
@@ -199,6 +221,11 @@ void logger_init(void)
pthread_mutex_lock(&g_mu);
if (!g_initialized) {
memset(&g_stats, 0, sizeof(g_stats));
memset(g_ctx_app, 0, sizeof(g_ctx_app));
memset(g_ctx_proto, 0, sizeof(g_ctx_proto));
memset(g_ctx_mode, 0, sizeof(g_ctx_mode));
memset(g_ctx_role, 0, sizeof(g_ctx_role));
memset(g_ctx_self_id, 0, sizeof(g_ctx_self_id));
g_stats.start_ms = now_ms();
g_stats.last_report_ms = g_stats.start_ms;
g_stats.window_start_ms = g_stats.start_ms;
@@ -213,6 +240,10 @@ void logger_init(void)
metric_reset(&g_stats.send_buffer_pct);
metric_reset(&g_stats.recv_buffer_pct);
metric_reset(&g_stats.cwnd);
copy_context_field(g_ctx_app, sizeof(g_ctx_app), NULL, "unknown");
copy_context_field(g_ctx_proto, sizeof(g_ctx_proto), NULL, "unknown");
copy_context_field(g_ctx_mode, sizeof(g_ctx_mode), NULL, "unknown");
copy_context_field(g_ctx_role, sizeof(g_ctx_role), NULL, "unknown");
{
const char *lvl_env = getenv("OMNI_LOG_LEVEL");
@@ -227,6 +258,36 @@ void logger_init(void)
pthread_mutex_unlock(&g_mu);
}
void logger_set_context(const char *app,
const char *proto,
const char *mode,
const char *role,
const char *self_id)
{
pthread_mutex_lock(&g_mu);
if (!g_initialized) {
pthread_mutex_unlock(&g_mu);
logger_init();
pthread_mutex_lock(&g_mu);
}
if (app) {
copy_context_field(g_ctx_app, sizeof(g_ctx_app), app, "unknown");
}
if (proto) {
copy_context_field(g_ctx_proto, sizeof(g_ctx_proto), proto, "unknown");
}
if (mode) {
copy_context_field(g_ctx_mode, sizeof(g_ctx_mode), mode, "unknown");
}
if (role) {
copy_context_field(g_ctx_role, sizeof(g_ctx_role), role, "unknown");
}
if (self_id) {
copy_context_field(g_ctx_self_id, sizeof(g_ctx_self_id), self_id, "");
}
pthread_mutex_unlock(&g_mu);
}
/* 记录一次发送事件,更新累计发送字节和窗口内发送字节。 */
void logger_on_send(size_t bytes)
{
@@ -540,6 +601,11 @@ void logger_print_performance_log(const char *tag)
uint64_t now;
OmniStats snapshot;
double progress_pct;
char ctx_app[sizeof(g_ctx_app)];
char ctx_proto[sizeof(g_ctx_proto)];
char ctx_mode[sizeof(g_ctx_mode)];
char ctx_role[sizeof(g_ctx_role)];
char ctx_self_id[sizeof(g_ctx_self_id)];
pthread_mutex_lock(&g_mu);
now = now_ms();
@@ -555,6 +621,11 @@ void logger_print_performance_log(const char *tag)
snapshot = g_stats;
snapshot.last_report_ms = now;
g_stats.last_report_ms = now;
omni_copy_fixed_ascii(ctx_app, sizeof(ctx_app), g_ctx_app);
omni_copy_fixed_ascii(ctx_proto, sizeof(ctx_proto), g_ctx_proto);
omni_copy_fixed_ascii(ctx_mode, sizeof(ctx_mode), g_ctx_mode);
omni_copy_fixed_ascii(ctx_role, sizeof(ctx_role), g_ctx_role);
omni_copy_fixed_ascii(ctx_self_id, sizeof(ctx_self_id), g_ctx_self_id);
pthread_mutex_unlock(&g_mu);
progress_pct = 0.0;
@@ -568,7 +639,7 @@ void logger_print_performance_log(const char *tag)
FILE *fp = stderr;
print_timestamp(fp, now);
fprintf(fp,
"level=INFO component=perf tag=%s "
"level=INFO component=perf app=%s proto=%s mode=%s role=%s self_id=%s tag=%s "
"elapsed_ms=%llu bytes_sent=%llu bytes_recv=%llu "
"send_count=%llu recv_count=%llu "
"tx_current_mbps=%.3f rx_current_mbps=%.3f "
@@ -582,6 +653,11 @@ void logger_print_performance_log(const char *tag)
"last_rtt_ms=%llu min_rtt_ms=%llu max_rtt_ms=%llu "
"tcp_retrans=%llu tcp_data_segs_out=%llu tcp_data_bytes_sent=%llu tcp_retrans_bytes=%llu "
"kcp_retrans=%llu kcp_data_segs_out=%llu kcp_data_bytes_sent=%llu kcp_retrans_bytes=%llu\n",
ctx_app[0] ? ctx_app : "unknown",
ctx_proto[0] ? ctx_proto : "unknown",
ctx_mode[0] ? ctx_mode : "unknown",
ctx_role[0] ? ctx_role : "unknown",
ctx_self_id[0] ? ctx_self_id : "-",
tag ? tag : "periodic",
(unsigned long long)(now - snapshot.start_ms),
(unsigned long long)snapshot.bytes_sent,
@@ -632,6 +708,11 @@ void logger_print_performance_log(const char *tag)
"{\"ts_ms\":%llu,"
"\"level\":\"INFO\","
"\"component\":\"perf\","
"\"app\":\"%s\","
"\"proto\":\"%s\","
"\"mode\":\"%s\","
"\"role\":\"%s\","
"\"self_id\":\"%s\","
"\"tag\":\"%s\","
"\"elapsed_ms\":%llu,"
"\"bytes_sent\":%llu,"
@@ -691,6 +772,11 @@ void logger_print_performance_log(const char *tag)
"\"kcp_data_bytes_sent\":%llu,"
"\"kcp_retrans_bytes\":%llu}\n",
(unsigned long long)now,
ctx_app[0] ? ctx_app : "unknown",
ctx_proto[0] ? ctx_proto : "unknown",
ctx_mode[0] ? ctx_mode : "unknown",
ctx_role[0] ? ctx_role : "unknown",
ctx_self_id[0] ? ctx_self_id : "",
tag ? tag : "periodic",
(unsigned long long)(now - snapshot.start_ms),
(unsigned long long)snapshot.bytes_sent,

View File

@@ -36,6 +36,20 @@ static const struct ProtoVTable *select_vtable(OmniProtocol proto)
}
}
static const char *proto_name(OmniProtocol proto)
{
switch (proto) {
case OMNI_PROTO_TCP:
return "tcp";
case OMNI_PROTO_UDP:
return "udp";
case OMNI_PROTO_KCP:
return "kcp";
default:
return "unknown";
}
}
OmniContext *omni_init(OmniRole role,
OmniProtocol proto,
const char *bind_ip,
@@ -44,6 +58,11 @@ OmniContext *omni_init(OmniRole role,
uint16_t peer_port)
{
logger_init();
logger_set_context("network",
proto_name(proto),
NULL,
role == OMNI_ROLE_SERVER ? "server" : "client",
NULL);
const struct ProtoVTable *vt = select_vtable(proto);
if (!vt || !vt->init) {

1141
src/core/peer_transport.c Normal file

File diff suppressed because it is too large Load Diff