From 9eb360926b46a351e31aeacbf22dc13424bd439b Mon Sep 17 00:00:00 2001 From: meiseayoung Date: Sat, 3 Jan 2026 20:45:30 +0800 Subject: [PATCH] feat(usockets): optimize high concurrency support up to 10,000+ connections - Increase uSockets listen() backlog from 512 to 16384 in bsd.c - Add high concurrency test suite (5000-10000 concurrent connections) - Add HIGH_CONCURRENCY.md documentation with optimization guide - Add usockets/build.sh script for rebuilding uSockets library - Update README with latest benchmark results Performance (macOS M3 Max): - 5,000 concurrent: 81,524 RPS, 100% success - 10,000 concurrent: 24,198 RPS, 100% success System requirements: - kern.ipc.somaxconn >= 8192 - ulimit -n 65535 --- README.md | 29 ++ docs/HIGH_CONCURRENCY.md | 115 +++++ tests/test_high_concurrency.go | 245 +++++++++++ usockets/README.md | 25 ++ usockets/build.sh | 55 +++ usockets/src/bsd.c | 767 +++++++++++++++++++++++++++++++++ usockets/usockets.v | 2 + 7 files changed, 1238 insertions(+) create mode 100644 docs/HIGH_CONCURRENCY.md create mode 100644 tests/test_high_concurrency.go create mode 100755 usockets/build.sh create mode 100644 usockets/src/bsd.c diff --git a/README.md b/README.md index 8604f88..3b70d59 100644 --- a/README.md +++ b/README.md @@ -110,6 +110,35 @@ fn main() { See [benchmark/README.md](benchmark/README.md) for performance benchmarks and [usockets/README.md](usockets/README.md) for uSockets integration details. +## High Concurrency Support + +v-hono with uSockets backend supports **10,000+ concurrent connections** out of the box. + +### Performance Results + +| Concurrent | RPS | Avg Latency | P99 Latency | Success Rate | +|------------|-----|-------------|-------------|--------------| +| 5,000 | 81,524 | 60.77ms | 295.17ms | 100% | +| 6,000 | 56,290 | 98.46ms | 557.47ms | 100% | +| 7,000 | 42,895 | 130.04ms | 824.80ms | 100% | +| 8,000 | 30,537 | 165.27ms | 1058.05ms | 100% | +| 9,000 | 28,693 | 215.86ms | 1422.67ms | 100% | +| 10,000 | 24,198 | 240.90ms | 1602.10ms | 100% | + +### System Configuration (macOS) + +For optimal high-concurrency performance: + +```bash +# Increase socket backlog limit +sudo sysctl -w kern.ipc.somaxconn=8192 + +# Increase file descriptor limit +ulimit -n 65535 +``` + +See [docs/HIGH_CONCURRENCY.md](docs/HIGH_CONCURRENCY.md) for detailed optimization guide. + ## Quick Start ```v diff --git a/docs/HIGH_CONCURRENCY.md b/docs/HIGH_CONCURRENCY.md new file mode 100644 index 0000000..53be64e --- /dev/null +++ b/docs/HIGH_CONCURRENCY.md @@ -0,0 +1,115 @@ +# v-hono 高并发优化指南 + +## 概述 + +v-hono 使用 uSockets 作为高性能网络后端，经过优化后可以稳定支持 **10,000+ 并发连接**。 + +## 性能测试结果 + +| 并发数 | RPS | 平均延迟 | P99延迟 | 成功率 | +|--------|-----|----------|---------|--------| +| 5,000 | 81,524 | 60.77ms | 295.17ms | 100% | +| 6,000 | 56,290 | 98.46ms | 557.47ms | 100% | +| 7,000 | 42,895 | 130.04ms | 824.80ms | 100% | +| 8,000 | 30,537 | 165.27ms | 1058.05ms | 100% | +| 9,000 | 28,693 | 215.86ms | 1422.67ms | 100% | +| 10,000 | 24,198 | 240.90ms | 1602.10ms | 100% | + +测试环境：macOS (Apple Silicon M3 Max) + +**重要提示**：每次测试后需等待 60 秒让 TIME_WAIT 连接释放，否则会影响下一次测试结果。 + +## 关键优化 + +### 1. uSockets backlog 参数 + +原始 uSockets 库的 `listen()` backlog 参数硬编码为 512，限制了高并发能力。 + +**修复位置**：`v-hono/usockets/src/bsd.c` (第 532 行和第 578 行) + +```c +// 修改前 +listen(listenFd, 512) + +// 修改后 +listen(listenFd, 16384) +``` + +修改后的源码保存在 `v-hono/usockets/src/bsd.c`，编译后的库文件在 `v-hono/lib/libusockets_full.a`。 + +**重新编译**：运行 `v-hono/usockets/build.sh` 脚本。 + +### 2. 系统参数配置 + +macOS 需要调整以下系统参数： + +```bash +# 查看当前值 +sysctl kern.ipc.somaxconn + +# 临时修改（重启后失效） +sudo sysctl -w kern.ipc.somaxconn=8192 + +# 永久修改（添加到 /etc/sysctl.conf） +kern.ipc.somaxconn=8192 +``` + +**注意**：实际 backlog 值会被 `kern.ipc.somaxconn` 截断，所以系统参数也需要相应调整。 + +## 重新编译 uSockets（可选） + +如果需要自定义 backlog 值，可以重新编译 uSockets： + +```bash +# 1. 克隆 uSockets +git clone https://github.com/uNetworking/uSockets.git + +# 2. 修改 src/bsd.c 中的 listen 调用 +# 将 listen(listenFd, 512) 改为 listen(listenFd, 16384) + +# 3. 编译（macOS with libuv） +WITH_LIBUV=1 CFLAGS="-I/opt/homebrew/include" make + +# 4. 复制到 v-hono +cp uSockets.a /path/to/v-hono/lib/libusockets_full.a + +# 5. 重新编译 v-hono 应用 +v -enable-globals -prod -o your_app your_app.v +``` + +## 使用建议 + +1. **生产环境**：建议使用 uSockets 后端（`app.listen_usockets(port)`） +2. **系统调优**：确保 `kern.ipc.somaxconn` >= 8192 +3. **文件描述符**：服务器和客户端都需要设置 `ulimit -n 65535` +4. **启动命令**：`ulimit -n 65535 && ./your_server` + +## 高并发测试 + +运行高并发测试： + +```bash +# 启动服务器 +ulimit -n 65535 && ./bench_server_usockets + +# 运行测试 (5000-10000 并发) +ulimit -n 65535 && go run v-hono/tests/test_high_concurrency.go + +# 单独测试某个级别 +ulimit -n 65535 && go run v-hono/tests/test_high_concurrency.go 8000 +``` + +## 问题排查 + +如果高并发测试失败，检查： + +1. 服务器是否正常响应：`curl http://localhost:8080/health` +2. 系统参数：`sysctl kern.ipc.somaxconn` +3. 文件描述符限制：`ulimit -n` +4. 测试客户端是否有端口耗尽问题 + +## 版本信息 + +- uSockets backlog: 16384 +- 测试日期: 2026-01-03 +- 测试通过最大并发: 10,000（不使用渐进式启动） diff --git a/tests/test_high_concurrency.go b/tests/test_high_concurrency.go new file mode 100644 index 0000000..db3e59f --- /dev/null +++ b/tests/test_high_concurrency.go @@ -0,0 +1,245 @@ +// 高并发测试 (5000-10000 并发) +// +// 使用方法: +// 1. 启动服务器 (必须设置 ulimit): ulimit -n 65535 && ./bench_server_usockets +// 2. 运行测试: ulimit -n 65535 && go run v-hono/tests/test_high_concurrency.go +// 3. 单独测试某个级别: ulimit -n 65535 && go run v-hono/tests/test_high_concurrency.go 8000 +// +// 重要提示: +// - 服务器和客户端都需要设置 ulimit -n 65535 +// - 连续测试多个并发级别时，建议每次测试前重启服务器 +// - TIME_WAIT 连接会影响后续测试，建议等待 10 秒后再测试下一个级别 +// +// 系统配置要求 (macOS): +// sudo sysctl -w kern.ipc.somaxconn=8192 +// ulimit -n 65535 + +package main + +import ( + "fmt" + "io" + "net/http" + "os" + "sort" + "strconv" + "sync" + "sync/atomic" + "time" +) + +type BenchResult struct { + Concurrency int + TotalReqs int64 + SuccessReqs int64 + RPS float64 + AvgLatency float64 + P99Latency float64 + SuccessRate float64 +} + +func main() { + baseURL := "http://127.0.0.1:8080/" + duration := 10 * time.Second + + // 默认测试级别: 5000, 6000, 7000, 8000, 9000, 10000 + concurrencyLevels := []int{5000, 6000, 7000, 8000, 9000, 10000} + + // 支持命令行参数指定单个并发级别 + if len(os.Args) > 1 { + level, err := strconv.Atoi(os.Args[1]) + if err == nil { + concurrencyLevels = []int{level} + } + } + + fmt.Println("╔═══════════════════════════════════════════════════════════════════════════╗") + fmt.Println("║ v-hono 高并发测试 (5000-10000 并发) ║") + fmt.Println("╚═══════════════════════════════════════════════════════════════════════════╝") + fmt.Println() + + // 检查服务器是否运行 + if !checkServer(baseURL) { + fmt.Println("❌ 服务器未运行") + fmt.Println(" 请先启动: ./bench_server_usockets") + fmt.Println() + fmt.Println(" 系统配置要求:") + fmt.Println(" - macOS: sudo sysctl -w kern.ipc.somaxconn=8192") + fmt.Println(" - ulimit -n 65535") + return + } + fmt.Println("✅ 服务器已就绪") + fmt.Println() + + results := make([]BenchResult, 0, len(concurrencyLevels)) + + for _, conns := range concurrencyLevels { + result := runBenchmark(baseURL, conns, duration) + results = append(results, result) + + // 测试间隔，让系统恢复 (TIME_WAIT 连接需要时间释放) + // macOS TIME_WAIT 默认 15-30 秒，建议等待 60 秒 + if len(concurrencyLevels) > 1 { + fmt.Println(" ⏳ 等待 60 秒让系统恢复 (TIME_WAIT 连接释放)...") + time.Sleep(60 * time.Second) + } + } + + // 输出汇总报告 + printSummary(results) +} + +func checkServer(baseURL string) bool { + client := &http.Client{Timeout: 2 * time.Second} + for i := 0; i < 5; i++ { + resp, err := client.Get(baseURL) + if err == nil && resp.StatusCode < 400 { + resp.Body.Close() + return true + } + time.Sleep(200 * time.Millisecond) + } + return false +} + +func runBenchmark(baseURL string, conns int, duration time.Duration) BenchResult { + fmt.Printf("🔥 测试 %d 并发连接 (持续 %v)\n", conns, duration) + + var total, success int64 + var totalLatencyNs int64 + latencies := make([]int64, 0, 1000000) + var mu sync.Mutex + + client := &http.Client{ + Timeout: 60 * time.Second, + Transport: &http.Transport{ + MaxIdleConns: conns * 2, + MaxIdleConnsPerHost: conns * 2, + MaxConnsPerHost: 0, + IdleConnTimeout: 120 * time.Second, + DisableKeepAlives: false, + ForceAttemptHTTP2: false, + }, + } + + var wg sync.WaitGroup + stop := make(chan struct{}) + start := time.Now() + + // 同时启动所有连接 + for i := 0; i < conns; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + t0 := time.Now() + resp, err := client.Get(baseURL) + lat := time.Since(t0).Nanoseconds() + + atomic.AddInt64(&total, 1) + atomic.AddInt64(&totalLatencyNs, lat) + + mu.Lock() + latencies = append(latencies, lat) + mu.Unlock() + + if err != nil { + continue + } + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode < 400 { + atomic.AddInt64(&success, 1) + } + } + } + }() + } + + time.Sleep(duration) + close(stop) + wg.Wait() + + elapsed := time.Since(start) + + // 计算 P99 + sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] }) + p99 := int64(0) + if len(latencies) > 0 { + idx := len(latencies) * 99 / 100 + p99 = latencies[idx] + } + + avgLat := float64(totalLatencyNs) / float64(total) / 1e6 + rps := float64(total) / elapsed.Seconds() + successRate := float64(success) / float64(total) * 100 + + // 判断测试结果 + status := "✅" + if successRate < 99.0 { + status = "⚠️" + } + if successRate < 90.0 { + status = "❌" + } + + fmt.Printf(" %s RPS: %8.0f Avg: %6.2fms P99: %6.2fms Success: %.1f%%\n", + status, rps, avgLat, float64(p99)/1e6, successRate) + fmt.Println() + + return BenchResult{ + Concurrency: conns, + TotalReqs: total, + SuccessReqs: success, + RPS: rps, + AvgLatency: avgLat, + P99Latency: float64(p99) / 1e6, + SuccessRate: successRate, + } +} + +func printSummary(results []BenchResult) { + if len(results) <= 1 { + return + } + + fmt.Println("═══════════════════════════════════════════════════════════════════════════") + fmt.Println(" 📊 测试结果汇总") + fmt.Println("═══════════════════════════════════════════════════════════════════════════") + fmt.Println() + fmt.Println("┌──────────────┬──────────────┬──────────────┬──────────────┬──────────────┐") + fmt.Println("│ 并发数 │ RPS │ 平均延迟 │ P99延迟 │ 成功率 │") + fmt.Println("├──────────────┼──────────────┼──────────────┼──────────────┼──────────────┤") + + allPassed := true + for _, r := range results { + status := "✅" + if r.SuccessRate < 99.0 { + status = "⚠️" + allPassed = false + } + if r.SuccessRate < 90.0 { + status = "❌" + allPassed = false + } + + fmt.Printf("│ %s %6d │ %10.0f │ %8.2fms │ %8.2fms │ %8.1f%% │\n", + status, r.Concurrency, r.RPS, r.AvgLatency, r.P99Latency, r.SuccessRate) + } + + fmt.Println("└──────────────┴──────────────┴──────────────┴──────────────┴──────────────┘") + fmt.Println() + + if allPassed { + fmt.Println("🎉 所有高并发测试通过！v-hono uSockets 后端表现优秀！") + } else { + fmt.Println("⚠️ 部分测试未达到 99% 成功率，请检查系统配置") + fmt.Println(" - macOS: sudo sysctl -w kern.ipc.somaxconn=8192") + fmt.Println(" - ulimit -n 65535") + } + fmt.Println() +} diff --git a/usockets/README.md b/usockets/README.md index 7ce954c..e194047 100644 --- a/usockets/README.md +++ b/usockets/README.md @@ -2,6 +2,31 @@ High-performance server backend using [uSockets](https://github.com/uNetworking/uSockets) library. +## High Concurrency Performance + +The included `libusockets_full.a` has been optimized with **backlog=16384** to support high concurrency: + +| Concurrent | RPS | Success Rate | +|------------|-----|--------------| +| 4,000 | 112,614 | 100% | +| 6,000 | 56,728 | 100% | +| 8,000 | 33,255 | 100% | +| 10,000 | 25,964 | 100% | + +### System Requirements + +For 10,000+ concurrent connections: + +```bash +# macOS +sudo sysctl -w kern.ipc.somaxconn=8192 +ulimit -n 65535 + +# Linux +sudo sysctl -w net.core.somaxconn=8192 +ulimit -n 65535 +``` + ## Performance Based on benchmark tests (200 connections, 100K requests): diff --git a/usockets/build.sh b/usockets/build.sh new file mode 100755 index 0000000..506adbf --- /dev/null +++ b/usockets/build.sh @@ -0,0 +1,55 @@ +#!/bin/bash +# uSockets 编译脚本 +# 用于重新编译 libusockets_full.a + +set -e + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +USOCKETS_DIR="/tmp/uSockets" + +echo "=== uSockets 编译脚本 ===" + +# 检查是否有 uSockets 源码 +if [ ! -d "$USOCKETS_DIR" ]; then + echo "克隆 uSockets..." + git clone --depth 1 https://github.com/uNetworking/uSockets.git "$USOCKETS_DIR" +fi + +# 应用 backlog 修改 +echo "应用 backlog=16384 修改..." +if [ -f "$SCRIPT_DIR/src/bsd.c" ]; then + cp "$SCRIPT_DIR/src/bsd.c" "$USOCKETS_DIR/src/bsd.c" +else + # 如果没有本地修改的源码，直接修改 + sed -i.bak 's/listen(listenFd, 512)/listen(listenFd, 16384)/g' "$USOCKETS_DIR/src/bsd.c" +fi + +# 编译 +echo "编译 uSockets..." +cd "$USOCKETS_DIR" +make clean + +# 根据系统选择编译参数 +if [[ "$OSTYPE" == "darwin"* ]]; then + # macOS + if [[ $(uname -m) == "arm64" ]]; then + # Apple Silicon + WITH_LIBUV=1 CFLAGS="-I/opt/homebrew/include" make + else + # Intel Mac + WITH_LIBUV=1 CFLAGS="-I/usr/local/include" make + fi +else + # Linux + WITH_LIBUV=1 make +fi + +# 复制到 v-hono +echo "复制库文件..." +cp "$USOCKETS_DIR/uSockets.a" "$SCRIPT_DIR/../lib/libusockets_full.a" + +echo "=== 编译完成 ===" +echo "库文件: $SCRIPT_DIR/../lib/libusockets_full.a" +echo "" +echo "重新编译 v-hono 应用:" +echo " v -enable-globals -prod -o your_app your_app.v" diff --git a/usockets/src/bsd.c b/usockets/src/bsd.c new file mode 100644 index 0000000..5998e20 --- /dev/null +++ b/usockets/src/bsd.c @@ -0,0 +1,767 @@ +/* + * Authored by Alex Hultman, 2018-2021. + * Intellectual property of third-party. + + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + + * http://www.apache.org/licenses/LICENSE-2.0 + + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/* Todo: this file should lie in networking/bsd.c */ + +#define __APPLE_USE_RFC_3542 + +#include "libusockets.h" +#include "internal/internal.h" + +#include +#include + +#ifndef _WIN32 +//#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include +#include +#endif + +/* Internal structure of packet buffer */ +struct us_internal_udp_packet_buffer { +#if defined(_WIN32) || defined(__APPLE__) + char *buf[LIBUS_UDP_MAX_NUM]; + size_t len[LIBUS_UDP_MAX_NUM]; + struct sockaddr_storage addr[LIBUS_UDP_MAX_NUM]; +#else + struct mmsghdr msgvec[LIBUS_UDP_MAX_NUM]; + struct iovec iov[LIBUS_UDP_MAX_NUM]; + struct sockaddr_storage addr[LIBUS_UDP_MAX_NUM]; + char control[LIBUS_UDP_MAX_NUM][256]; +#endif +}; + +/* We need to emulate sendmmsg, recvmmsg on platform who don't have it */ +int bsd_sendmmsg(LIBUS_SOCKET_DESCRIPTOR fd, void *msgvec, unsigned int vlen, int flags) { +#if defined(__APPLE__) + +struct mmsghdr { + struct msghdr msg_hdr; /* Message header */ + unsigned int msg_len; /* Number of bytes transmitted */ +}; + + struct mmsghdr *hdrs = (struct mmsghdr *) msgvec; + + for (int i = 0; i < vlen; i++) { + int ret = sendmsg(fd, &hdrs[i].msg_hdr, flags); + if (ret == -1) { + if (i) { + return i; + } else { + return -1; + } + } else { + hdrs[i].msg_len = ret; + } + } + + return vlen; + +#elif defined(_WIN32) + + struct us_internal_udp_packet_buffer *packet_buffer = (struct us_internal_udp_packet_buffer *) msgvec; + + /* Let's just use sendto here */ + /* Winsock does not have sendmsg, while macOS has, however, we simply use sendto since both macOS and Winsock has it. + * Besides, you should use Linux either way to get best performance with the sendmmsg */ + + + // while we do not get error, send next + + for (int i = 0; i < LIBUS_UDP_MAX_NUM; i++) { + // need to support ipv6 addresses also! + int ret = sendto(fd, packet_buffer->buf[i], packet_buffer->len[i], flags, (struct sockaddr *)&packet_buffer->addr[i], sizeof(struct sockaddr_in)); + + if (ret == -1) { + // if we fail then we need to buffer up, no that's not our problem + // we do need to register poll out though and have a callback for it + return i; + } + + //printf("sendto: %d\n", ret); + } + + return LIBUS_UDP_MAX_NUM; // one message +#else + return sendmmsg(fd, (struct mmsghdr *)msgvec, vlen, flags | MSG_NOSIGNAL); +#endif +} + +int bsd_recvmmsg(LIBUS_SOCKET_DESCRIPTOR fd, void *msgvec, unsigned int vlen, int flags, void *timeout) { +#if defined(_WIN32) || defined(__APPLE__) + struct us_internal_udp_packet_buffer *packet_buffer = (struct us_internal_udp_packet_buffer *) msgvec; + + + for (int i = 0; i < LIBUS_UDP_MAX_NUM; i++) { + socklen_t addr_len = sizeof(struct sockaddr_storage); + int ret = recvfrom(fd, packet_buffer->buf[i], LIBUS_UDP_MAX_SIZE, flags, (struct sockaddr *)&packet_buffer->addr[i], &addr_len); + + if (ret == -1) { + return i; + } + + packet_buffer->len[i] = ret; + } + + return LIBUS_UDP_MAX_NUM; +#else + // we need to set controllen for ip packet + for (int i = 0; i < vlen; i++) { + ((struct mmsghdr *)msgvec)[i].msg_hdr.msg_controllen = 256; + } + + return recvmmsg(fd, (struct mmsghdr *)msgvec, vlen, flags, 0); +#endif +} + +// this one is needed for knowing the destination addr of udp packet +// an udp socket can only bind to one port, and that port never changes +// this function returns ONLY the IP address, not any port +int bsd_udp_packet_buffer_local_ip(void *msgvec, int index, char *ip) { +#if defined(_WIN32) || defined(__APPLE__) + return 0; // not supported +#else + struct msghdr *mh = &((struct mmsghdr *) msgvec)[index].msg_hdr; + for (struct cmsghdr *cmsg = CMSG_FIRSTHDR(mh); cmsg != NULL; cmsg = CMSG_NXTHDR(mh, cmsg)) { + // Linux ipv4 + #ifdef IP_PKTINFO + if (cmsg->cmsg_level == IPPROTO_IP && cmsg->cmsg_type == IP_PKTINFO) { + struct in_pktinfo *pi = (struct in_pktinfo *) CMSG_DATA(cmsg); + memcpy(ip, &pi->ipi_addr, 4); + return 4; + } + // FreeBSD ipv4 + #elif IP_RECVDSTADDR + if (cmsg->cmsg_level == IPPROTO_IP && cmsg->cmsg_type == IP_RECVDSTADDR) { + struct in_addr *addr = (struct in_addr *) CMSG_DATA(cmsg); + memcpy(ip, addr, 4); + return 4; + } + #endif + + // Linux, FreeBSD ipv6 + if (cmsg->cmsg_level == IPPROTO_IPV6 && cmsg->cmsg_type == IPV6_PKTINFO) { + struct in6_pktinfo *pi6 = (struct in6_pktinfo *) CMSG_DATA(cmsg); + memcpy(ip, &pi6->ipi6_addr, 16); + return 16; + } + } + + return 0; // no length + +#endif +} + +char *bsd_udp_packet_buffer_peer(void *msgvec, int index) { +#if defined(_WIN32) || defined(__APPLE__) + struct us_internal_udp_packet_buffer *packet_buffer = (struct us_internal_udp_packet_buffer *) msgvec; + return (char *)&packet_buffer->addr[index]; +#else + return ((struct mmsghdr *) msgvec)[index].msg_hdr.msg_name; +#endif +} + +char *bsd_udp_packet_buffer_payload(void *msgvec, int index) { +#if defined(_WIN32) || defined(__APPLE__) + struct us_internal_udp_packet_buffer *packet_buffer = (struct us_internal_udp_packet_buffer *) msgvec; + return packet_buffer->buf[index]; +#else + return ((struct mmsghdr *) msgvec)[index].msg_hdr.msg_iov[0].iov_base; +#endif +} + +int bsd_udp_packet_buffer_payload_length(void *msgvec, int index) { +#if defined(_WIN32) || defined(__APPLE__) + struct us_internal_udp_packet_buffer *packet_buffer = (struct us_internal_udp_packet_buffer *) msgvec; + return packet_buffer->len[index]; +#else + return ((struct mmsghdr *) msgvec)[index].msg_len; +#endif +} + +void bsd_udp_buffer_set_packet_payload(struct us_udp_packet_buffer_t *send_buf, int index, int offset, void *payload, int length, void *peer_addr) { +#if defined(_WIN32) || defined(__APPLE__) + struct us_internal_udp_packet_buffer *packet_buffer = (struct us_internal_udp_packet_buffer *) send_buf; + + memcpy(packet_buffer->buf[index], payload, length); + memcpy(&packet_buffer->addr[index], peer_addr, sizeof(struct sockaddr_storage)); + + packet_buffer->len[index] = length; +#else + //printf("length: %d, offset: %d\n", length, offset); + + struct mmsghdr *ss = (struct mmsghdr *) send_buf; + + // copy the peer address + memcpy(ss[index].msg_hdr.msg_name, peer_addr, /*ss[index].msg_hdr.msg_namelen*/ sizeof(struct sockaddr_in)); + + // set control length to 0 + ss[index].msg_hdr.msg_controllen = 0; + + // copy the payload + + ss[index].msg_hdr.msg_iov->iov_len = length + offset; + + + memcpy(((char *) ss[index].msg_hdr.msg_iov->iov_base) + offset, payload, length); +#endif +} + +/* The maximum UDP payload size is 64kb, but in IPV6 you can have jumbopackets larger than so. + * We do not support those jumbo packets currently, but will safely ignore them. + * Any sane sender would assume we don't support them if we consistently drop them. + * Therefore a udp_packet_buffer_t will be 64 MB in size (64kb * 1024). */ +void *bsd_create_udp_packet_buffer() { +#if defined(_WIN32) || defined(__APPLE__) + struct us_internal_udp_packet_buffer *b = malloc(sizeof(struct us_internal_udp_packet_buffer) + LIBUS_UDP_MAX_SIZE * LIBUS_UDP_MAX_NUM); + + for (int i = 0; i < LIBUS_UDP_MAX_NUM; i++) { + b->buf[i] = ((char *) b) + sizeof(struct us_internal_udp_packet_buffer) + LIBUS_UDP_MAX_SIZE * i; + } + + return (struct us_udp_packet_buffer_t *) b; +#else + /* Allocate 64kb times 1024 */ + struct us_internal_udp_packet_buffer *b = malloc(sizeof(struct us_internal_udp_packet_buffer) + LIBUS_UDP_MAX_SIZE * LIBUS_UDP_MAX_NUM); + + for (int n = 0; n < LIBUS_UDP_MAX_NUM; ++n) { + + b->iov[n].iov_base = &((char *) (b + 1))[n * LIBUS_UDP_MAX_SIZE]; + b->iov[n].iov_len = LIBUS_UDP_MAX_SIZE; + + b->msgvec[n].msg_hdr = (struct msghdr) { + .msg_name = &b->addr[n], + .msg_namelen = sizeof (struct sockaddr_storage), + + .msg_iov = &b->iov[n], + .msg_iovlen = 1, + + .msg_control = b->control[n], + .msg_controllen = 256, + }; + } + + return (struct us_udp_packet_buffer_t *) b; +#endif +} + +LIBUS_SOCKET_DESCRIPTOR apple_no_sigpipe(LIBUS_SOCKET_DESCRIPTOR fd) { +#ifdef __APPLE__ + if (fd != LIBUS_SOCKET_ERROR) { + int no_sigpipe = 1; + setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, (void *) &no_sigpipe, sizeof(int)); + } +#endif + return fd; +} + +LIBUS_SOCKET_DESCRIPTOR bsd_set_nonblocking(LIBUS_SOCKET_DESCRIPTOR fd) { +#ifdef _WIN32 + /* Libuv will set windows sockets as non-blocking */ +#else + fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0) | O_NONBLOCK); +#endif + return fd; +} + +void bsd_socket_nodelay(LIBUS_SOCKET_DESCRIPTOR fd, int enabled) { + setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (void *) &enabled, sizeof(enabled)); +} + +void bsd_socket_flush(LIBUS_SOCKET_DESCRIPTOR fd) { + // Linux TCP_CORK has the same underlying corking mechanism as with MSG_MORE +#ifdef TCP_CORK + int enabled = 0; + setsockopt(fd, IPPROTO_TCP, TCP_CORK, (void *) &enabled, sizeof(int)); +#endif +} + +LIBUS_SOCKET_DESCRIPTOR bsd_create_socket(int domain, int type, int protocol) { + // returns INVALID_SOCKET on error + int flags = 0; +#if defined(SOCK_CLOEXEC) && defined(SOCK_NONBLOCK) + flags = SOCK_CLOEXEC | SOCK_NONBLOCK; +#endif + + LIBUS_SOCKET_DESCRIPTOR created_fd = socket(domain, type | flags, protocol); + + return bsd_set_nonblocking(apple_no_sigpipe(created_fd)); +} + +void bsd_close_socket(LIBUS_SOCKET_DESCRIPTOR fd) { +#ifdef _WIN32 + closesocket(fd); +#else + close(fd); +#endif +} + +void bsd_shutdown_socket(LIBUS_SOCKET_DESCRIPTOR fd) { +#ifdef _WIN32 + shutdown(fd, SD_SEND); +#else + shutdown(fd, SHUT_WR); +#endif +} + +void bsd_shutdown_socket_read(LIBUS_SOCKET_DESCRIPTOR fd) { +#ifdef _WIN32 + shutdown(fd, SD_RECEIVE); +#else + shutdown(fd, SHUT_RD); +#endif +} + +void internal_finalize_bsd_addr(struct bsd_addr_t *addr) { + // parse, so to speak, the address + if (addr->mem.ss_family == AF_INET6) { + addr->ip = (char *) &((struct sockaddr_in6 *) addr)->sin6_addr; + addr->ip_length = sizeof(struct in6_addr); + addr->port = ntohs(((struct sockaddr_in6 *) addr)->sin6_port); + } else if (addr->mem.ss_family == AF_INET) { + addr->ip = (char *) &((struct sockaddr_in *) addr)->sin_addr; + addr->ip_length = sizeof(struct in_addr); + addr->port = ntohs(((struct sockaddr_in *) addr)->sin_port); + } else { + addr->ip_length = 0; + addr->port = -1; + } +} + +int bsd_local_addr(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr) { + addr->len = sizeof(addr->mem); + if (getsockname(fd, (struct sockaddr *) &addr->mem, &addr->len)) { + return -1; + } + internal_finalize_bsd_addr(addr); + return 0; +} + +int bsd_remote_addr(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr) { + addr->len = sizeof(addr->mem); + if (getpeername(fd, (struct sockaddr *) &addr->mem, &addr->len)) { + return -1; + } + internal_finalize_bsd_addr(addr); + return 0; +} + +char *bsd_addr_get_ip(struct bsd_addr_t *addr) { + return addr->ip; +} + +int bsd_addr_get_ip_length(struct bsd_addr_t *addr) { + return addr->ip_length; +} + +int bsd_addr_get_port(struct bsd_addr_t *addr) { + return addr->port; +} + +// called by dispatch_ready_poll +LIBUS_SOCKET_DESCRIPTOR bsd_accept_socket(LIBUS_SOCKET_DESCRIPTOR fd, struct bsd_addr_t *addr) { + LIBUS_SOCKET_DESCRIPTOR accepted_fd; + addr->len = sizeof(addr->mem); + +#if defined(SOCK_CLOEXEC) && defined(SOCK_NONBLOCK) + // Linux, FreeBSD + accepted_fd = accept4(fd, (struct sockaddr *) addr, &addr->len, SOCK_CLOEXEC | SOCK_NONBLOCK); +#else + // Windows, OS X + accepted_fd = accept(fd, (struct sockaddr *) addr, &addr->len); + +#endif + + /* We cannot rely on addr since it is not initialized if failed */ + if (accepted_fd == LIBUS_SOCKET_ERROR) { + return LIBUS_SOCKET_ERROR; + } + + internal_finalize_bsd_addr(addr); + + return bsd_set_nonblocking(apple_no_sigpipe(accepted_fd)); +} + +int bsd_recv(LIBUS_SOCKET_DESCRIPTOR fd, void *buf, int length, int flags) { + return recv(fd, buf, length, flags); +} + +#if !defined(_WIN32) +#include + +int bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_length, const char *payload, int payload_length) { + struct iovec chunks[2]; + + chunks[0].iov_base = (char *)header; + chunks[0].iov_len = header_length; + chunks[1].iov_base = (char *)payload; + chunks[1].iov_len = payload_length; + + return writev(fd, chunks, 2); +} +#else +int bsd_write2(LIBUS_SOCKET_DESCRIPTOR fd, const char *header, int header_length, const char *payload, int payload_length) { + int written = bsd_send(fd, header, header_length, 0); + if (written == header_length) { + int second_write = bsd_send(fd, payload, payload_length, 0); + if (second_write > 0) { + written += second_write; + } + } + return written; +} +#endif + +int bsd_send(LIBUS_SOCKET_DESCRIPTOR fd, const char *buf, int length, int msg_more) { + + // MSG_MORE (Linux), MSG_PARTIAL (Windows), TCP_NOPUSH (BSD) + +#ifndef MSG_NOSIGNAL +#define MSG_NOSIGNAL 0 +#endif + +#ifdef MSG_MORE + + // for Linux we do not want signals + return send(fd, buf, length, ((msg_more != 0) * MSG_MORE) | MSG_NOSIGNAL); + +#else + + // use TCP_NOPUSH + + return send(fd, buf, length, MSG_NOSIGNAL); + +#endif +} + +int bsd_would_block() { +#ifdef _WIN32 + return WSAGetLastError() == WSAEWOULDBLOCK; +#else + return errno == EWOULDBLOCK;// || errno == EAGAIN; +#endif +} + +// return LIBUS_SOCKET_ERROR or the fd that represents listen socket +// listen both on ipv6 and ipv4 +LIBUS_SOCKET_DESCRIPTOR bsd_create_listen_socket(const char *host, int port, int options) { + struct addrinfo hints, *result; + memset(&hints, 0, sizeof(struct addrinfo)); + + hints.ai_flags = AI_PASSIVE; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + char port_string[16]; + snprintf(port_string, 16, "%d", port); + + if (getaddrinfo(host, port_string, &hints, &result)) { + return LIBUS_SOCKET_ERROR; + } + + LIBUS_SOCKET_DESCRIPTOR listenFd = LIBUS_SOCKET_ERROR; + struct addrinfo *listenAddr; + for (struct addrinfo *a = result; a && listenFd == LIBUS_SOCKET_ERROR; a = a->ai_next) { + if (a->ai_family == AF_INET6) { + listenFd = bsd_create_socket(a->ai_family, a->ai_socktype, a->ai_protocol); + listenAddr = a; + } + } + + for (struct addrinfo *a = result; a && listenFd == LIBUS_SOCKET_ERROR; a = a->ai_next) { + if (a->ai_family == AF_INET) { + listenFd = bsd_create_socket(a->ai_family, a->ai_socktype, a->ai_protocol); + listenAddr = a; + } + } + + if (listenFd == LIBUS_SOCKET_ERROR) { + freeaddrinfo(result); + return LIBUS_SOCKET_ERROR; + } + + if (port != 0) { + /* Otherwise, always enable SO_REUSEPORT and SO_REUSEADDR _unless_ options specify otherwise */ +#ifdef _WIN32 + if (options & LIBUS_LISTEN_EXCLUSIVE_PORT) { + int optval2 = 1; + setsockopt(listenFd, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, (void *) &optval2, sizeof(optval2)); + } else { + int optval3 = 1; + setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, (void *) &optval3, sizeof(optval3)); + } +#else + #if /*defined(__linux) &&*/ defined(SO_REUSEPORT) + if (!(options & LIBUS_LISTEN_EXCLUSIVE_PORT)) { + int optval = 1; + setsockopt(listenFd, SOL_SOCKET, SO_REUSEPORT, (void *) &optval, sizeof(optval)); + } + #endif + int enabled = 1; + setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, (void *) &enabled, sizeof(enabled)); +#endif + + } + +#ifdef IPV6_V6ONLY + int disabled = 0; + setsockopt(listenFd, IPPROTO_IPV6, IPV6_V6ONLY, (void *) &disabled, sizeof(disabled)); +#endif + + if (bind(listenFd, listenAddr->ai_addr, (socklen_t) listenAddr->ai_addrlen) || listen(listenFd, 16384)) { + bsd_close_socket(listenFd); + freeaddrinfo(result); + return LIBUS_SOCKET_ERROR; + } + + freeaddrinfo(result); + return listenFd; +} + +#ifndef _WIN32 +#include +#else +#include +#include +#endif +#include +#include +LIBUS_SOCKET_DESCRIPTOR bsd_create_listen_socket_unix(const char *path, int options) { + + LIBUS_SOCKET_DESCRIPTOR listenFd = LIBUS_SOCKET_ERROR; + + listenFd = bsd_create_socket(AF_UNIX, SOCK_STREAM, 0); + + if (listenFd == LIBUS_SOCKET_ERROR) { + return LIBUS_SOCKET_ERROR; + } + +#ifndef _WIN32 + // 700 permission by default + fchmod(listenFd, S_IRWXU); +#else + _chmod(path, S_IREAD | S_IWRITE | S_IEXEC); +#endif + + struct sockaddr_un server_address; + memset(&server_address, 0, sizeof(server_address)); + server_address.sun_family = AF_UNIX; + strcpy(server_address.sun_path, path); + int size = offsetof(struct sockaddr_un, sun_path) + strlen(server_address.sun_path); +#ifdef _WIN32 + _unlink(path); +#else + unlink(path); +#endif + + if (bind(listenFd, (struct sockaddr *)&server_address, size) || listen(listenFd, 16384)) { + bsd_close_socket(listenFd); + return LIBUS_SOCKET_ERROR; + } + + return listenFd; +} + +LIBUS_SOCKET_DESCRIPTOR bsd_create_udp_socket(const char *host, int port) { + struct addrinfo hints, *result; + memset(&hints, 0, sizeof(struct addrinfo)); + + hints.ai_flags = AI_PASSIVE; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_DGRAM; + + char port_string[16]; + snprintf(port_string, 16, "%d", port); + + if (getaddrinfo(host, port_string, &hints, &result)) { + return LIBUS_SOCKET_ERROR; + } + + LIBUS_SOCKET_DESCRIPTOR listenFd = LIBUS_SOCKET_ERROR; + struct addrinfo *listenAddr; + for (struct addrinfo *a = result; a && listenFd == LIBUS_SOCKET_ERROR; a = a->ai_next) { + if (a->ai_family == AF_INET6) { + listenFd = bsd_create_socket(a->ai_family, a->ai_socktype, a->ai_protocol); + listenAddr = a; + } + } + + for (struct addrinfo *a = result; a && listenFd == LIBUS_SOCKET_ERROR; a = a->ai_next) { + if (a->ai_family == AF_INET) { + listenFd = bsd_create_socket(a->ai_family, a->ai_socktype, a->ai_protocol); + listenAddr = a; + } + } + + if (listenFd == LIBUS_SOCKET_ERROR) { + freeaddrinfo(result); + return LIBUS_SOCKET_ERROR; + } + + if (port != 0) { + /* Should this also go for UDP? */ + int enabled = 1; + setsockopt(listenFd, SOL_SOCKET, SO_REUSEADDR, (void *) &enabled, sizeof(enabled)); + } + +#ifdef IPV6_V6ONLY + int disabled = 0; + setsockopt(listenFd, IPPROTO_IPV6, IPV6_V6ONLY, (void *) &disabled, sizeof(disabled)); +#endif + + /* We need destination address for udp packets in both ipv6 and ipv4 */ + +/* On FreeBSD this option seems to be called like so */ +#ifndef IPV6_RECVPKTINFO +#define IPV6_RECVPKTINFO IPV6_PKTINFO +#endif + + int enabled = 1; + if (setsockopt(listenFd, IPPROTO_IPV6, IPV6_RECVPKTINFO, (void *) &enabled, sizeof(enabled)) == -1) { + if (errno == 92) { + // Linux ipv4 + #ifdef IP_PKTINFO + if (setsockopt(listenFd, IPPROTO_IP, IP_PKTINFO, (void *) &enabled, sizeof(enabled)) != 0) { + printf("Error setting IPv4 pktinfo!\n"); + } + // FreeBSD ipv4 + #elif IP_RECVDSTADDR + if (setsockopt(listenFd, IPPROTO_IP, IP_RECVDSTADDR, (void *) &enabled, sizeof(enabled)) != 0) { + printf("Error setting IPv4 pktinfo!\n"); + } + #endif + } else { + printf("Error setting IPv6 pktinfo!\n"); + } + } + + /* These are used for getting the ECN */ + if (setsockopt(listenFd, IPPROTO_IPV6, IPV6_RECVTCLASS, (void *) &enabled, sizeof(enabled)) == -1) { + if (errno == 92) { + if (setsockopt(listenFd, IPPROTO_IP, IP_RECVTOS, (void *) &enabled, sizeof(enabled)) != 0) { + printf("Error setting IPv4 ECN!\n"); + } + } else { + printf("Error setting IPv6 ECN!\n"); + } + } + + /* We bind here as well */ + if (bind(listenFd, listenAddr->ai_addr, (socklen_t) listenAddr->ai_addrlen)) { + bsd_close_socket(listenFd); + freeaddrinfo(result); + return LIBUS_SOCKET_ERROR; + } + + freeaddrinfo(result); + return listenFd; +} + +int bsd_udp_packet_buffer_ecn(void *msgvec, int index) { + +#if defined(_WIN32) || defined(__APPLE__) + printf("ECN not supported!\n"); +#else + // we should iterate all control messages once, after recvmmsg and then only fetch them with these functions + struct msghdr *mh = &((struct mmsghdr *) msgvec)[index].msg_hdr; + for (struct cmsghdr *cmsg = CMSG_FIRSTHDR(mh); cmsg != NULL; cmsg = CMSG_NXTHDR(mh, cmsg)) { + // do we need to get TOS from ipv6 also? + if (cmsg->cmsg_level == IPPROTO_IP) { + if (cmsg->cmsg_type == IP_TOS) { + uint8_t tos = *(uint8_t *)CMSG_DATA(cmsg); + return tos & 3; + } + } + + if (cmsg->cmsg_level == IPPROTO_IPV6) { + if (cmsg->cmsg_type == IPV6_TCLASS) { + // is this correct? + uint8_t tos = *(uint8_t *)CMSG_DATA(cmsg); + return tos & 3; + } + } + } +#endif + + printf("We got no ECN!\n"); + + return 0; // no ecn defaults to 0 +} + +LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket(const char *host, int port, const char *source_host, int options) { + struct addrinfo hints, *result; + memset(&hints, 0, sizeof(struct addrinfo)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + + char port_string[16]; + snprintf(port_string, 16, "%d", port); + + if (getaddrinfo(host, port_string, &hints, &result) != 0) { + return LIBUS_SOCKET_ERROR; + } + + LIBUS_SOCKET_DESCRIPTOR fd = bsd_create_socket(result->ai_family, result->ai_socktype, result->ai_protocol); + if (fd == LIBUS_SOCKET_ERROR) { + freeaddrinfo(result); + return LIBUS_SOCKET_ERROR; + } + + if (source_host) { + struct addrinfo *interface_result; + if (!getaddrinfo(source_host, NULL, NULL, &interface_result)) { + int ret = bind(fd, interface_result->ai_addr, (socklen_t) interface_result->ai_addrlen); + freeaddrinfo(interface_result); + if (ret == LIBUS_SOCKET_ERROR) { + bsd_close_socket(fd); + freeaddrinfo(result); + return LIBUS_SOCKET_ERROR; + } + } + } + + connect(fd, result->ai_addr, (socklen_t) result->ai_addrlen); + freeaddrinfo(result); + + return fd; +} + +LIBUS_SOCKET_DESCRIPTOR bsd_create_connect_socket_unix(const char *server_path, int options) { + + struct sockaddr_un server_address; + memset(&server_address, 0, sizeof(server_address)); + server_address.sun_family = AF_UNIX; + strcpy(server_address.sun_path, server_path); + int size = offsetof(struct sockaddr_un, sun_path) + strlen(server_address.sun_path); + + LIBUS_SOCKET_DESCRIPTOR fd = bsd_create_socket(AF_UNIX, SOCK_STREAM, 0); + + if (fd == LIBUS_SOCKET_ERROR) { + return LIBUS_SOCKET_ERROR; + } + + connect(fd, (struct sockaddr *)&server_address, size); + + return fd; +} diff --git a/usockets/usockets.v b/usockets/usockets.v index 5495cac..700cffa 100644 --- a/usockets/usockets.v +++ b/usockets/usockets.v @@ -105,6 +105,8 @@ pub fn (ctx SocketContext) on_writable(h voidptr) { C.us_socket_context_on_writa pub fn (ctx SocketContext) on_timeout(h voidptr) { C.us_socket_context_on_timeout(0, ctx, h) } pub fn (ctx SocketContext) on_end(h voidptr) { C.us_socket_context_on_end(0, ctx, h) } pub fn (ctx SocketContext) listen(port int) ListenSocket { + // 参数: ssl=0, context, host=nil, port, options=LIBUS_LISTEN_DEFAULT(0), socket_ext_size=0 + // 注意: backlog 是在 uSockets 库内部硬编码的，不通过此函数传递 return C.us_socket_context_listen(0, ctx, unsafe { nil }, port, 0, 0) }