Compare commits

...

8 Commits

Author SHA1 Message Date
enoch 38797fe078 KCP Improving 2026-05-30 18:09:09 +08:00
enoch 56c74337e4 KCP improve 2026-05-30 17:50:45 +08:00
enoch 661ed401da Add KCP support 2026-05-29 01:06:18 +08:00
enoch 6d9299eeb0 优化3 2026-05-29 00:08:15 +08:00
enoch e80fe14624 优化2 2026-05-28 23:59:12 +08:00
enoch 068d3f4be9 压力优化 2026-05-28 23:55:16 +08:00
enoch 9c07b9fadc add more 2026-05-28 23:42:00 +08:00
enoch ad1327bebd add more 2026-05-28 23:41:40 +08:00
44 changed files with 4589 additions and 277 deletions
+44 -4
View File
@@ -9,6 +9,40 @@ LAYLINK_FRAME_ENCRYPTION=none
# Agent 与 POP Server 之间 Frame 加密方式;可选 none、chacha20,两端必须一致。
LAYLINK_FRAME_ENCRYPTION_KEY=
# Frame 加密密钥;LAYLINK_FRAME_ENCRYPTION=chacha20 时必填。可填普通口令,或 hex:32字节十六进制,或 base64:32字节base64。
LAYLINK_DATA_CHUNK_BYTES=1048576
# TCP DATA 帧分片大小;单位字节。较大值可减少帧数量并提升吞吐,建议 262144、524288、1048576、2097152。
LAYLINK_MAX_SEND_BUFFER_BYTES=67108864
# 单连接发送缓冲区上限;单位字节。大文件下载建议 33554432 或 67108864,内存紧张时调小。
LAYLINK_BACKPRESSURE_HIGH_WATERMARK_BYTES=33554432
# 背压触发水位;单位字节。应小于 LAYLINK_MAX_SEND_BUFFER_BYTES,达到后会暂停上游读取直到缓冲排空。
[kcp]
LAYLINK_KCP_BACKEND=ffi
# KCP 实现后端;可选 ffi、php。ffi 使用 native ikcp.c 动态库,生产建议使用;php 是调试回退实现。
LAYLINK_KCP_FFI_LIB=native/kcp/liblaylink_kcp.so
# native KCP 动态库路径;LAYLINK_KCP_BACKEND=ffi 时使用。相对路径按项目根目录解析,先运行 scripts/build-kcp-ffi.sh 构建。
LAYLINK_KCP_NODELAY=1
# KCP nodelay 开关;1 更低延迟,0 更保守。拥堵明显时可保持 1 并调大 interval 或关闭 nc。
LAYLINK_KCP_INTERVAL_MS=10
# KCP 内部 update 间隔;单位毫秒。常用 10、20、30,越小越低延迟但 CPU/发包更高。
LAYLINK_KCP_FAST_RESEND=2
# KCP 快速重传阈值;0 关闭,2 是常见低延迟设置,丢包网络可尝试 2-4。
LAYLINK_KCP_NO_CONGESTION_CONTROL=1
# 是否关闭 KCP 拥塞控制;1 为高吞吐模式,接近早期 100Mbps 默认值;丢包/拥堵明显时改 0。
LAYLINK_KCP_SEND_WINDOW=1024
# KCP 发送窗口;高吞吐建议 1024,保守可用 256 或 512。
LAYLINK_KCP_RECV_WINDOW=1024
# KCP 接收窗口;应不小于发送窗口,高吞吐建议 1024。
LAYLINK_KCP_MTU_BYTES=1350
# KCP MTU;高吞吐建议 1350,若公网链路分片/丢包明显则改 1200。
LAYLINK_KCP_TICK_MS=10
# PHP transport tick 间隔;单位毫秒。通常与 LAYLINK_KCP_INTERVAL_MS 一致。
LAYLINK_KCP_UDP_SEND_QUEUE_BYTES=67108864
# UDP EAGAIN 发送队列上限;单位字节。高吞吐建议 67108864,内存紧张或拥堵明显时调低。
LAYLINK_KCP_UDP_FLUSH_PACKETS=1024
# 每次 tick 最多刷出的 UDP packet 数;高吞吐建议 1024,拥堵时可调低到 256 或 128。
LAYLINK_KCP_OUTPUT_DRAIN_PACKETS=1024
# 每次从 native KCP 输出队列搬到 UDP 发送队列的最大 packet 数;高吞吐建议 1024,单连接影响事件循环时调低。
[client-agent]
NODE_ID=client-01
@@ -20,9 +54,11 @@ NODE_TOKEN=CHANGE_ME
NODE_ZONE=default
# 当前 Agent 所在逻辑区域;可按部署场景填写,例如 local、corp、restricted-a。
POP_SERVER_ADDRESS=tcp://127.0.0.1:9001
# Agent 出站连接 POP Server 的地址;格式为 tcp://host:port,例如 tcp://10.1.0.2:9001。
# Agent 出站连接 POP Server 的地址;格式为 tcp://host:port,例如 tcp://10.1.0.2:9001AGENT_TRANSPORT_PROTOCOL=kcp 时会使用同一 host:port 的 UDP
AGENT_TRANSPORT_PROTOCOL=tcp
# 当前 Agent 到 POP Server 使用的传输协议;可选值 tcp、udp、kcp;必须被 POP_ALLOWED_AGENT_TRANSPORTS 允许,当前可运行值为 tcp。
# 当前 Agent 到 POP Server 使用的传输协议;可选值 tcp、udp、kcp;必须被 POP_ALLOWED_AGENT_TRANSPORTS 允许,当前可运行值为 tcp、kcp
CLIENT_AGENT_POP_CONNECTIONS=1
# Client Agent 到 POP Server 的并行长连接数量;默认 1。提高到 2、4 可分摊多会话,当前 tcp/kcp 生效。
CLIENT_AGENT_AUTH_TOKEN=dev-token
# Client Agent 为 SOCKS5/HTTP 代理入口生成 OPEN 帧时使用的客户端认证 token;当前 MVP 默认 dev-token。
CLIENT_AGENT_USER_ID=admin
@@ -61,5 +97,9 @@ CLIENT_AGENT_RAW_JSON_LISTEN_PORT=9000
[pop-server]
POP_AGENT_LISTEN=0.0.0.0:9001
# POP Server 监听 Agent 长连接的地址;格式为 host:port,例如 0.0.0.0:9001 或 127.0.0.1:9001。
POP_ALLOWED_AGENT_TRANSPORTS=tcp
# POP Server 允许 Client Agent 使用的传输协议;逗号分隔,可选值 tcp、udp、kcp;当前已实现 tcpudp/kcp 为预留实现。
POP_ALLOWED_AGENT_TRANSPORTS=tcp,kcp
# POP Server 允许 Client Agent 使用的传输协议;可写逗号数组 tcp,kcp,也可写 JSON 数组 ["tcp","kcp"];当前已实现 tcp、kcpudp 为预留实现。
POP_AGENT_TCP_WORKERS=1
# POP Server TCP Agent listener 的 worker 数量;TCP 模式可按 CPU 和并发提高到 2、4、8。只使用 KCP 时该值不影响 KCP。
POP_AGENT_KCP_WORKERS=1
# POP Server KCP/UDP Agent listener 的 worker 数量;当前必须为 1,因为 KCP 会话状态在单 worker 内维护,多 worker 会导致 UDP 包跨进程丢状态。
+2 -1
View File
@@ -1,2 +1,3 @@
.env
runtime/*
runtime/*
native/kcp/*.so
+4
View File
@@ -39,6 +39,10 @@ $bootAgent = function (string $protocol, string $listen, string $name) use ($nod
? Env::get('CLIENT_AGENT_SOCKS5_UDP_LISTEN_IP', '127.0.0.1') . ':' . Env::get('CLIENT_AGENT_SOCKS5_UDP_LISTEN_PORT', '1081')
: null,
Env::get('CLIENT_AGENT_SOCKS5_UDP_ADVERTISE_IP', Env::get('CLIENT_AGENT_SOCKS5_LISTEN_IP', '127.0.0.1')),
Env::int('LAYLINK_MAX_SEND_BUFFER_BYTES', 64 * 1024 * 1024, 1024 * 1024),
Env::int('LAYLINK_BACKPRESSURE_HIGH_WATERMARK_BYTES', 32 * 1024 * 1024, 512 * 1024),
Env::int('LAYLINK_DATA_CHUNK_BYTES', 1024 * 1024, 16 * 1024, 8 * 1024 * 1024),
Env::int('CLIENT_AGENT_POP_CONNECTIONS', 1, 1, 16),
);
$agent->boot($name);
};
+5
View File
@@ -24,6 +24,11 @@ $server = new PopServer(
require dirname(__DIR__) . '/config/policies.php',
Env::csv('POP_ALLOWED_AGENT_TRANSPORTS', ['tcp']),
Env::get('AUDIT_LOG', dirname(__DIR__) . '/runtime/audit.log'),
Env::int('LAYLINK_MAX_SEND_BUFFER_BYTES', 64 * 1024 * 1024, 1024 * 1024),
Env::int('LAYLINK_BACKPRESSURE_HIGH_WATERMARK_BYTES', 32 * 1024 * 1024, 512 * 1024),
Env::int('LAYLINK_DATA_CHUNK_BYTES', 1024 * 1024, 16 * 1024, 8 * 1024 * 1024),
Env::int('POP_AGENT_TCP_WORKERS', 1, 1, 64),
Env::int('POP_AGENT_KCP_WORKERS', 1, 1, 1),
);
$server->boot();
+1 -1
View File
@@ -8,7 +8,7 @@ return [
'192.168.0.0/16',
'10.10.0.0/16',
],
'allowed_ports' => [22, 80, 443, 3306, 5432],
'allowed_ports' => [22, 80, 443, '3000-60000', 3306, 5432],
'enabled' => true,
],
];
+1 -1
View File
@@ -5,7 +5,7 @@ return [
'policy_id' => 'public-web-egress',
'users' => ['normal-user', 'admin', 'devops'],
'target_hosts' => ['*'],
'target_ports' => [80, 443],
'target_ports' => [80, 443, '3000-60000'],
'protocol' => 'tcp',
'route_type' => 'direct',
'enabled' => true,
+147 -32
View File
@@ -2,7 +2,7 @@
## Implementation Status
Last updated: 2026-05-28 Asia/Shanghai.
Last updated: 2026-05-30 Asia/Shanghai.
Current phase: MVP bootstrap in progress.
@@ -24,7 +24,7 @@ Direction update under evaluation:
* POP Server uses `POP_ALLOWED_AGENT_TRANSPORTS` to allow one or more Agent-to-POP transports.
* Agent uses `AGENT_TRANSPORT_PROTOCOL` to choose one concrete transport.
* Allowed names are `tcp`, `udp`, and `kcp`.
* Current runnable implementation is `tcp`; `udp` and `kcp` are reserved and must be implemented behind a transport abstraction.
* Current runnable implementations are `tcp` and experimental `kcp`; `udp` is reserved.
* Feasibility:
* Workerman supports long-running async TCP servers and custom protocols; it is suitable for the framed fallback/control channel.
* KCP itself is a UDP-based reliable ARQ protocol, so adding KCP means adding a UDP transport layer and session demultiplexing below the existing LayLink frame protocol.
@@ -33,7 +33,7 @@ Direction update under evaluation:
1. Complete Client Agent naming migration in code, docs, config, and entrypoints.
2. Implement TCP-framed Client Agent -> POP -> public target path.
3. Define `TransportInterface` so frame protocol can run over TCP now and KCP later.
4. Add KCP-over-UDP transport via extension/FFI/proxy after the TCP framed path is stable.
4. Add KCP-over-UDP transport behind the transport abstraction after the TCP framed path is stable.
* Main risk:
* KCP is not a socket by itself. It needs UDP I/O, timers, packet flush/update scheduling, MTU handling, retransmission tuning, and connection/session management.
* PHP-only KCP may work as a prototype but is likely CPU-heavy under concurrency.
@@ -129,6 +129,7 @@ Completed in this checkpoint:
* `https://ip.sb/` for egress IP
* Reorganized `.env.example` into readable sections:
* `[config]`
* `[kcp]`
* `[client-agent]`
* `[pop-server]`
* Section headers are comments-for-humans in practice; the current Env loader ignores lines without `=`.
@@ -151,6 +152,109 @@ Completed in this checkpoint:
* `chacha20` currently uses libsodium XChaCha20 stream encryption with a random nonce per frame.
* Verified `none` and `chacha20` FrameCodec encode/decode round trips.
* Verified POP Server starts with `LAYLINK_FRAME_ENCRYPTION=chacha20`.
* Added port range matching for policy and agent allowlist ports:
* `target_ports` supports exact ports such as `80` and string ranges such as `'8080-10080'`.
* `allowed_ports` supports the same syntax.
* Allowed sample public TCP egress policy on port range `'8080-10080'` for HTTP-alt/speedtest style endpoints.
* Optimized TCP stream `DATA` frames:
* Control frames remain JSON.
* TCP `DATA` payloads now use binary frame encoding when both ends run the updated code.
* This removes base64 expansion and JSON string encoding from the hot TCP data path.
* Verified binary TCP `DATA` frame encode/decode under both `none` and `chacha20`.
* Added POP-side target DNS pre-resolution:
* Domain resolution failures return `OPEN_FAIL` with `dns_resolution_failed`.
* POP no longer lets common target DNS failures bubble up as raw `stream_socket_client()` warnings.
* Added TCP backpressure for large transfers:
* POP pauses target reads when the Agent connection send buffer crosses the high watermark.
* POP resumes target reads when the Agent connection drains.
* Client Agent pauses local client reads when the POP connection send buffer crosses the high watermark.
* Client Agent pauses POP reads while a local client output buffer is full.
* Send buffer limits default to 64 MiB with a 32 MiB backpressure high watermark.
* Tuning envs:
* `LAYLINK_DATA_CHUNK_BYTES`
* `LAYLINK_MAX_SEND_BUFFER_BYTES`
* `LAYLINK_BACKPRESSURE_HIGH_WATERMARK_BYTES`
* Fixed large-download truncation risk:
* Client Agent now treats POP `CLOSE` as a graceful remote EOF and waits for the local client send buffer to drain before closing the local socket.
* TCP `DATA` is split into configurable chunks, defaulting to 1 MiB, to reduce frame overhead while avoiding oversized frames.
* POP refreshes Agent activity on any valid frame, not only `PING`, reducing heartbeat false positives during heavy traffic.
* Started Agent-to-POP transport abstraction:
* Added `FrameClientTransport`.
* Added `TcpFrameClientTransport` as the current TCP implementation.
* `AgentClient` now sends and receives LayLink frames through the transport interface instead of directly owning `AsyncTcpConnection` and `FrameParser`.
* This preserves current TCP behavior while preparing a `KcpFrameClientTransport` implementation.
* Added POP-side frame transport abstraction:
* Added `FrameServerConnection`.
* Added `TcpFrameServerConnection`.
* Added `TcpFrameServerListener`.
* `AgentListener`, `NodeConnection`, `NodeRegistry`, and `TunnelSession` now hold Agent connections through `FrameServerConnection`.
* TCP listener decode/encode details are isolated from POP session, policy, heartbeat, and relay logic.
* Added transport factory/config selection:
* `FrameClientTransportFactory` maps `AGENT_TRANSPORT_PROTOCOL=tcp` to `TcpFrameClientTransport`.
* `FrameServerListenerFactory` maps the implemented POP transport `tcp` to `TcpFrameServerListener`.
* `FrameClientTransportFactory` maps `AGENT_TRANSPORT_PROTOCOL=kcp` to `KcpFrameClientTransport`.
* `FrameServerListenerFactory` maps POP transport `kcp` to `KcpFrameServerListener`.
* `udp` still fails at factory boundaries with explicit not-implemented errors instead of leaking into business logic.
* Added experimental multi-connection Client Agent -> POP support:
* `CLIENT_AGENT_POP_CONNECTIONS` controls how many parallel Agent-to-POP long connections a Client Agent opens.
* New local TCP sessions are distributed round-robin across authenticated POP transports.
* Each session stays bound to its selected POP transport for the whole session lifetime.
* POP `NodeRegistry` now supports multiple live connections under the same `NODE_ID`.
* Heartbeat activity and offline cleanup are tracked per Agent connection.
* KCP/UDP implementation decision:
* Start with `kcp` before raw `udp` for Agent-to-POP frame transport.
* Existing TCP tunnel sessions require ordered, reliable byte-stream semantics; raw UDP would need retransmission, ordering, MTU fragmentation, congestion/window handling, and session cleanup.
* Implementing raw UDP as a general Frame transport would effectively recreate a weaker KCP.
* Keep the existing SOCKS5 `UDP ASSOCIATE`/`UDP_DATA` feature separate: it is application datagram relay over the current reliable Agent-to-POP channel, not the Agent-to-POP transport itself.
* Recommended KCP path is a transport implementation behind `FrameClientTransport` / `FrameServerConnection`, backed by a native extension, FFI binding, or sidecar process rather than pure PHP for production throughput.
* Added KCP Agent-to-POP transport:
* `KcpPacketCodec` defines UDP packet types for `SYN`, `SYN_ACK`, `DATA`, `ACK`, and `CLOSE`.
* `KcpFrameClientTransport` runs Client Agent frames over UDP while preserving the existing `FrameClientTransport` interface.
* `KcpFrameServerListener` and `KcpFrameServerConnection` expose KCP/UDP sessions to POP as `FrameServerConnection`.
* POP can now listen on both TCP and KCP when `POP_ALLOWED_AGENT_TRANSPORTS=tcp,kcp`.
* `NativeKcpSession` uses PHP FFI to call native upstream `ikcp.c` through `native/kcp/liblaylink_kcp.so`.
* `scripts/build-kcp-ffi.sh` builds the native shared library from vendored `native/kcp/ikcp.c`.
* `LAYLINK_KCP_BACKEND=ffi` selects the native KCP backend; `LAYLINK_KCP_BACKEND=php` remains as a debugging fallback through `KcpReliableSession`.
* `LAYLINK_KCP_FFI_LIB` can point to a custom native KCP library path.
* Added array-style env parsing:
* `Env::csv()` accepts traditional comma-separated values such as `tcp,kcp`.
* `Env::csv()` also accepts JSON arrays such as `["tcp","kcp"]`.
* Added KCP congestion and UDP EAGAIN controls:
* `KcpUdpPacketSender` bypasses Workerman `UdpConnection::send()` for KCP packets and uses suppressed `stream_socket_sendto()` directly.
* UDP `EAGAIN` / "Resource temporarily unavailable" no longer emits PHP warnings from the KCP transport path.
* KCP packets that cannot be sent immediately are queued and retried on subsequent transport ticks.
* Added KCP tuning envs:
* `LAYLINK_KCP_NODELAY`
* `LAYLINK_KCP_INTERVAL_MS`
* `LAYLINK_KCP_FAST_RESEND`
* `LAYLINK_KCP_NO_CONGESTION_CONTROL`
* `LAYLINK_KCP_SEND_WINDOW`
* `LAYLINK_KCP_RECV_WINDOW`
* `LAYLINK_KCP_MTU_BYTES`
* `LAYLINK_KCP_TICK_MS`
* `LAYLINK_KCP_UDP_SEND_QUEUE_BYTES`
* `LAYLINK_KCP_UDP_FLUSH_PACKETS`
* `LAYLINK_KCP_OUTPUT_DRAIN_PACKETS`
* Added POP worker count configuration:
* `POP_AGENT_TCP_WORKERS` controls TCP Agent listener worker count.
* `POP_AGENT_KCP_WORKERS` is exposed but currently clamped to `1` in `bin/pop-server.php`.
* KCP/UDP must remain single-worker in the current architecture because KCP session state is process-local and UDP packets for one conv can otherwise be handled by different workers.
* Native KCP output draining is capped per tick by `LAYLINK_KCP_OUTPUT_DRAIN_PACKETS` to reduce single-flow event-loop monopolization during large downloads.
* Fixed KCP POP-side session lookup for real-world UDP/NAT behavior:
* POP no longer depends only on `remote ip:port + conv` for KCP session lookup.
* POP keeps a `conv -> session` index and migrates the current UDP remote address when a known `conv` arrives from a changed source port.
* KCP callbacks now resolve the active connection by `conv`, so migrated sessions continue delivering frames instead of silently dropping `DATA`.
* Fixed native KCP send-buffer accounting after heavy speedtest-style traffic:
* The previous PHP-side native KCP `queuedBytes` counter could grow during large transfers and never fall back to zero unless user payload was received in the opposite direction.
* This could make a long-lived KCP Agent connection permanently appear full after upload/download tests, causing later `OPEN_OK` / `DATA` sends to fail and audit rows to show zero transferred bytes.
* Added native wrapper `laylink_kcp_waitsnd()` around upstream `ikcp_waitsnd()`.
* `NativeKcpSession::getSendBufferQueueSize()` now derives watermarks from real KCP pending segment count plus pending UDP output bytes.
* POP now treats failure to send `OPEN_OK` as `agent_buffer_overflow` instead of leaving a target connection open with no client-visible success.
* Restored KCP high-throughput defaults:
* Early native KCP testing used hardcoded `nodelay=1, interval=10, resend=2, nc=1, sndwnd=1024, rcvwnd=1024, mtu=1350`.
* The first exposed `.env` defaults were more conservative (`nc=0`, smaller windows, `mtu=1200`) and could reduce throughput dramatically on speedtest-style high-BDP paths.
* `NativeKcpSession`, KCP UDP queue/flush fallback defaults, current `.env`, and `.env.example` now use the high-throughput profile by default.
* For lossy or congested paths, tune down to `LAYLINK_KCP_NO_CONGESTION_CONTROL=0`, `LAYLINK_KCP_MTU_BYTES=1200`, smaller windows, or lower flush/drain packet counts.
Known MVP limitations:
@@ -164,6 +268,7 @@ Known MVP limitations:
* No TLS yet.
* No production-grade client identity yet; `dev-token` is hardcoded for MVP development.
* No automated integration test harness yet.
* TCP stream forwarding can now use multiple Agent-to-POP connections per Client Agent, but a single TCP session is still pinned to one POP transport. Binary `DATA` frames, chunking, graceful EOF, and backpressure reduce per-byte overhead and buffer blowups; KCP is experimental and still needs throughput/loss tuning, while multipath and per-session flow-control tuning are future performance work.
* No explicit idle timeout or connect timeout enforcement yet.
* UDP relay is datagram-oriented and currently creates short-lived POP-side UDP sockets per outbound datagram; pooling and stronger timeout accounting are still future work.
* HTTP proxy supports `CONNECT` and ordinary absolute URL HTTP requests; advanced proxy auth and full HTTP/2 proxying are not implemented.
@@ -173,12 +278,22 @@ Next recommended tasks:
1. Add a local integration harness that starts POP, Client Agent, and a mock TCP echo target, then verifies authorized tunnel, policy denial, and agent local denial.
2. Add configurable client auth token or JWT-ready auth interface.
3. Add target connect timeout and session idle timeout.
4. Add buffer full/drain handling with audit result `buffer_overflow`.
4. Add more detailed buffer overflow audit reasons and metrics.
5. Add README quickstart with exact local commands.
6. Optimize UDP relay with POP-side UDP socket pooling.
7. Add UDP association idle timeouts and cleanup.
8. Aggregate UDP audit records per association instead of per datagram.
9. Add UDP and per-user rate limiting.
6. Add a reproducible throughput benchmark script for direct-vs-LayLink comparisons.
7. Keep TCP tuning as an ongoing task:
* benchmark `LAYLINK_DATA_CHUNK_BYTES` at `524288`, `1048576`, `2097152`, and `4194304`
* benchmark buffer pairs such as `64MiB/32MiB` and `128MiB/64MiB`
* record direct-vs-LayLink throughput, CPU, memory, and disconnect behavior
8. Benchmark and tune `CLIENT_AGENT_POP_CONNECTIONS` for 1, 2, 4, and 8 long connections under mixed single-download and multi-session workloads.
9. Benchmark native FFI `kcp` against `tcp` under latency, loss, and high-throughput workloads; tune KCP nodelay, window, MTU, resend, interval, UDP queue, and flush settings.
10. Design KCP horizontal scaling before allowing `POP_AGENT_KCP_WORKERS>1`; options include multiple POP ports/instances, reuseport five-tuple affinity, external session state, or a UDP dispatcher keyed by conv.
11. Add raw UDP Agent-to-POP transport only for explicitly datagram-oriented frame classes, or after a reliability/window design exists.
12. Add per-session flow-control windows to reduce head-of-line blocking on one Agent connection.
13. Optimize UDP relay with POP-side UDP socket pooling.
14. Add UDP association idle timeouts and cleanup.
15. Aggregate UDP audit records per association instead of per datagram.
16. Add UDP and per-user rate limiting.
## 0. Project Name
@@ -421,7 +536,7 @@ Required frame types:
| `OPEN` | Client Agent -> POP | Client Agent requests POP to authorize and open a target stream. |
| `OPEN_OK` | POP -> Client Agent | POP has connected the target and the stream can begin. |
| `OPEN_FAIL` | POP -> Client Agent | POP rejected or failed the requested target stream. |
| `DATA` | Bidirectional | Stream bytes for one `session_id`; MVP payload uses base64. |
| `DATA` | Bidirectional | Stream bytes for one `session_id`; TCP stream payloads use binary frame encoding when both ends are updated. |
| `UDP_DATA` | Bidirectional | UDP datagram bytes for one UDP association; MVP payload uses base64 and includes target metadata. |
| `CLOSE` | Bidirectional | Close one stream session. |
| `ERROR` | Bidirectional | Explicit protocol or session error. |
@@ -447,13 +562,11 @@ payload_length
payload
```
Suggested JSON payload for MVP is acceptable.
Binary optimization may be added later.
Control frames use JSON payloads. TCP stream `DATA` frames may use the binary DATA encoding below.
### 6.3 Frame Encoding
For MVP, use length-prefixed JSON frames.
For control frames, use length-prefixed JSON frames.
Format:
@@ -462,20 +575,33 @@ uint32_be length
json_payload
```
Example decoded frame:
Example decoded control frame:
```json
{
"version": 1,
"type": "DATA",
"type": "OPEN",
"session_id": "018f6f4a-xxxx-xxxx",
"payload": "base64-encoded-binary"
"payload": {
"target_host": "example.com",
"target_port": 443,
"protocol": "tcp"
}
}
```
For `DATA` frames, binary stream data may be base64 encoded in MVP.
TCP stream `DATA` frames use a binary body before optional encryption:
A later version may replace this with binary headers plus raw binary body.
```text
uint32_be encrypted_or_plain_body_length
"LLB1"
uint8 binary_type=1
uint16_be session_id_length
session_id bytes
raw TCP payload bytes
```
Legacy JSON/base64 `DATA` decoding remains accepted for compatibility, but updated senders should emit binary `DATA`.
---
@@ -658,7 +784,7 @@ return [
'policy_id' => 'public-web-egress',
'users' => ['normal-user', 'admin'],
'target_hosts' => ['*'],
'target_ports' => [80, 443],
'target_ports' => [80, 443, '8080-10080'],
'route_type' => 'direct',
'enabled' => true,
],
@@ -696,7 +822,7 @@ return [
'192.168.0.0/16',
'10.10.0.0/16',
],
'allowed_ports' => [22, 80, 443, 3306, 5432],
'allowed_ports' => [22, 80, 443, '8080-10080', 3306, 5432],
'enabled' => true,
],
];
@@ -773,18 +899,7 @@ On failure:
After `OPEN_OK`, data is exchanged with `DATA` frames.
Example:
```json
{
"version": 1,
"type": "DATA",
"session_id": "018f6f4a-xxxx",
"payload": {
"data": "base64-encoded-binary"
}
}
```
Updated implementations encode TCP `DATA` as the binary frame described in section 6.3. Legacy JSON/base64 `DATA` frames may still be decoded during rolling upgrades.
Both POP Server and Agent must map `session_id` to the corresponding local TCP connection.
+1423
View File
File diff suppressed because it is too large Load Diff
+452
View File
@@ -0,0 +1,452 @@
//=====================================================================
//
// KCP - A Better ARQ Protocol Implementation
// skywind3000 (at) gmail.com, 2010-2011
//
// Features:
// + Average RTT reduce 30% - 40% vs traditional ARQ like tcp.
// + Maximum RTT reduce three times vs tcp.
// + Lightweight, distributed as a single source file.
//
//=====================================================================
#ifndef _IKCP_H_
#define _IKCP_H_
#include <stddef.h>
#include <stdlib.h>
#include <assert.h>
//=====================================================================
// 32BIT INTEGER DEFINITION
//=====================================================================
#ifndef __INTEGER_32_BITS__
#define __INTEGER_32_BITS__
#if defined(_WIN64) || defined(WIN64) || defined(__amd64__) || \
defined(__x86_64) || defined(__x86_64__) || defined(_M_IA64) || \
defined(_M_AMD64)
typedef unsigned int ISTDUINT32;
typedef int ISTDINT32;
#elif defined(_WIN32) || defined(WIN32) || defined(__i386__) || \
defined(__i386) || defined(_M_X86)
typedef unsigned long ISTDUINT32;
typedef long ISTDINT32;
#elif defined(__MACOS__)
typedef UInt32 ISTDUINT32;
typedef SInt32 ISTDINT32;
#elif defined(__APPLE__) && defined(__MACH__)
#include <sys/types.h>
typedef u_int32_t ISTDUINT32;
typedef int32_t ISTDINT32;
#elif defined(__BEOS__)
#include <sys/inttypes.h>
typedef u_int32_t ISTDUINT32;
typedef int32_t ISTDINT32;
#elif (defined(_MSC_VER) || defined(__BORLANDC__)) && (!defined(__MSDOS__))
typedef unsigned __int32 ISTDUINT32;
typedef __int32 ISTDINT32;
#elif defined(__GNUC__)
#include <stdint.h>
typedef uint32_t ISTDUINT32;
typedef int32_t ISTDINT32;
#else
typedef unsigned long ISTDUINT32;
typedef long ISTDINT32;
#endif
#endif
//=====================================================================
// Integer Definition
//=====================================================================
#ifndef __IINT8_DEFINED
#define __IINT8_DEFINED
typedef char IINT8;
#endif
#ifndef __IUINT8_DEFINED
#define __IUINT8_DEFINED
typedef unsigned char IUINT8;
#endif
#ifndef __IUINT16_DEFINED
#define __IUINT16_DEFINED
typedef unsigned short IUINT16;
#endif
#ifndef __IINT16_DEFINED
#define __IINT16_DEFINED
typedef short IINT16;
#endif
#ifndef __IINT32_DEFINED
#define __IINT32_DEFINED
typedef ISTDINT32 IINT32;
#endif
#ifndef __IUINT32_DEFINED
#define __IUINT32_DEFINED
typedef ISTDUINT32 IUINT32;
#endif
#ifndef __IINT64_DEFINED
#define __IINT64_DEFINED
#if defined(_MSC_VER) || defined(__BORLANDC__)
typedef __int64 IINT64;
#else
typedef long long IINT64;
#endif
#endif
#ifndef __IUINT64_DEFINED
#define __IUINT64_DEFINED
#if defined(_MSC_VER) || defined(__BORLANDC__)
typedef unsigned __int64 IUINT64;
#else
typedef unsigned long long IUINT64;
#endif
#endif
#ifndef INLINE
#if defined(__GNUC__)
#if (__GNUC__ > 3) || ((__GNUC__ == 3) && (__GNUC_MINOR__ >= 1))
#define INLINE __inline__ __attribute__((always_inline))
#else
#define INLINE __inline__
#endif
#elif (defined(_MSC_VER) || defined(__BORLANDC__) || defined(__WATCOMC__))
#define INLINE __inline
#else
#define INLINE
#endif
#endif
#if (!defined(__cplusplus)) && (!defined(inline))
#define inline INLINE
#endif
//=====================================================================
// QUEUE DEFINITION
//=====================================================================
#ifndef __IQUEUE_DEF__
#define __IQUEUE_DEF__
struct IQUEUEHEAD {
struct IQUEUEHEAD *next, *prev;
};
typedef struct IQUEUEHEAD iqueue_head;
//---------------------------------------------------------------------
// queue init
//---------------------------------------------------------------------
#define IQUEUE_HEAD_INIT(name) { &(name), &(name) }
#define IQUEUE_HEAD(name) \
struct IQUEUEHEAD name = IQUEUE_HEAD_INIT(name)
#define IQUEUE_INIT(ptr) ( \
(ptr)->next = (ptr), (ptr)->prev = (ptr))
#define IOFFSETOF(TYPE, MEMBER) ((size_t) &((TYPE *)0)->MEMBER)
#define ICONTAINEROF(ptr, type, member) ( \
(type*)( ((char*)((type*)ptr)) - IOFFSETOF(type, member)) )
#define IQUEUE_ENTRY(ptr, type, member) ICONTAINEROF(ptr, type, member)
//---------------------------------------------------------------------
// queue operation
//---------------------------------------------------------------------
#define IQUEUE_ADD(node, head) ( \
(node)->prev = (head), (node)->next = (head)->next, \
(head)->next->prev = (node), (head)->next = (node))
#define IQUEUE_ADD_TAIL(node, head) ( \
(node)->prev = (head)->prev, (node)->next = (head), \
(head)->prev->next = (node), (head)->prev = (node))
#define IQUEUE_DEL_BETWEEN(p, n) ((n)->prev = (p), (p)->next = (n))
#define IQUEUE_DEL(entry) (\
(entry)->next->prev = (entry)->prev, \
(entry)->prev->next = (entry)->next, \
(entry)->next = 0, (entry)->prev = 0)
#define IQUEUE_DEL_INIT(entry) do { \
IQUEUE_DEL(entry); IQUEUE_INIT(entry); } while (0)
#define IQUEUE_IS_EMPTY(entry) ((entry) == (entry)->next)
#define iqueue_init IQUEUE_INIT
#define iqueue_entry IQUEUE_ENTRY
#define iqueue_add IQUEUE_ADD
#define iqueue_add_tail IQUEUE_ADD_TAIL
#define iqueue_del IQUEUE_DEL
#define iqueue_del_init IQUEUE_DEL_INIT
#define iqueue_is_empty IQUEUE_IS_EMPTY
#define IQUEUE_FOREACH(iterator, head, TYPE, MEMBER) \
for ((iterator) = iqueue_entry((head)->next, TYPE, MEMBER); \
&((iterator)->MEMBER) != (head); \
(iterator) = iqueue_entry((iterator)->MEMBER.next, TYPE, MEMBER))
#define iqueue_foreach(iterator, head, TYPE, MEMBER) \
IQUEUE_FOREACH(iterator, head, TYPE, MEMBER)
#define iqueue_foreach_entry(pos, head) \
for( (pos) = (head)->next; (pos) != (head) ; (pos) = (pos)->next )
#define __iqueue_splice(list, head) do { \
iqueue_head *first = (list)->next, *last = (list)->prev; \
iqueue_head *at = (head)->next; \
(first)->prev = (head), (head)->next = (first); \
(last)->next = (at), (at)->prev = (last); } while (0)
#define iqueue_splice(list, head) do { \
if (!iqueue_is_empty(list)) __iqueue_splice(list, head); } while (0)
#define iqueue_splice_init(list, head) do { \
iqueue_splice(list, head); iqueue_init(list); } while (0)
#ifdef _MSC_VER
#pragma warning(disable:4311)
#pragma warning(disable:4312)
#pragma warning(disable:4996)
#endif
#endif
//---------------------------------------------------------------------
// BYTE ORDER & ALIGNMENT
//---------------------------------------------------------------------
#ifndef IWORDS_BIG_ENDIAN
#ifdef _BIG_ENDIAN_
#if _BIG_ENDIAN_
#define IWORDS_BIG_ENDIAN 1
#endif
#endif
#ifndef IWORDS_BIG_ENDIAN
#if defined(__hppa__) || \
defined(__m68k__) || defined(mc68000) || defined(_M_M68K) || \
(defined(__MIPS__) && defined(__MIPSEB__)) || \
defined(__ppc__) || defined(__POWERPC__) || defined(_M_PPC) || \
defined(__sparc__) || defined(__powerpc__) || \
defined(__mc68000__) || defined(__s390x__) || defined(__s390__)
#define IWORDS_BIG_ENDIAN 1
#endif
#endif
#ifndef IWORDS_BIG_ENDIAN
#define IWORDS_BIG_ENDIAN 0
#endif
#endif
#ifndef IWORDS_MUST_ALIGN
#if defined(__i386__) || defined(__i386) || defined(_i386_)
#define IWORDS_MUST_ALIGN 0
#elif defined(_M_IX86) || defined(_X86_) || defined(__x86_64__)
#define IWORDS_MUST_ALIGN 0
#elif defined(__amd64) || defined(__amd64__)
#define IWORDS_MUST_ALIGN 0
#else
#define IWORDS_MUST_ALIGN 1
#endif
#endif
//=====================================================================
// Predefine struct
//=====================================================================
struct IKCPCB;
typedef struct IKCPCB ikcpcb;
//=====================================================================
// SEGMENT
//=====================================================================
struct IKCPSEG
{
struct IQUEUEHEAD node;
IUINT32 conv;
IUINT32 cmd;
IUINT32 frg;
IUINT32 wnd;
IUINT32 ts;
IUINT32 sn;
IUINT32 una;
IUINT32 len;
IUINT32 resendts;
IUINT32 rto;
IUINT32 fastack;
IUINT32 xmit;
char data[1];
};
//---------------------------------------------------------------------
// IKCPOPS - pluggable congestion control operations
//---------------------------------------------------------------------
struct IKCPOPS
{
const char *name;
int (*init)(ikcpcb *kcp);
void (*release)(ikcpcb *kcp);
void (*on_ack)(ikcpcb *kcp, IUINT32 acked_segs, IUINT32 acked_bytes,
IUINT32 prior_in_flight);
void (*on_fast_retransmit)(ikcpcb *kcp, IUINT32 fast_retrans,
IUINT32 inflight, IUINT32 prior_cwnd);
void (*on_timeout)(ikcpcb *kcp, IUINT32 prior_cwnd);
void (*on_tick)(ikcpcb *kcp);
void (*on_app_limited)(ikcpcb *kcp, IUINT32 inflight);
void (*on_rtt)(ikcpcb *kcp, IINT32 rtt);
void (*on_pkt_sent)(ikcpcb *kcp, IUINT32 sn, IUINT32 ts,
IUINT32 len, IUINT32 inflight, IUINT32 xmit);
void (*on_pkt_acked)(ikcpcb *kcp, IUINT32 sn, IUINT32 ts,
IUINT32 len, IINT32 rtt, IUINT32 xmit);
IUINT32 (*get_info)(ikcpcb *kcp, void *buf, IUINT32 bufsize);
IUINT32 (*pacing_rate)(ikcpcb *kcp);
};
//---------------------------------------------------------------------
// IKCPCB
//---------------------------------------------------------------------
struct IKCPCB
{
IUINT32 conv, mtu, mss, state;
IUINT32 snd_una, snd_nxt, rcv_nxt;
IUINT32 ts_recent, ts_lastack, ssthresh;
IINT32 rx_rttval, rx_srtt, rx_rto, rx_minrto;
IUINT32 snd_wnd, rcv_wnd, rmt_wnd, cwnd, probe;
IUINT32 current, interval, ts_flush, xmit;
IUINT32 nrcv_buf, nsnd_buf;
IUINT32 nrcv_que, nsnd_que;
IUINT32 nodelay, updated;
IUINT32 ts_probe, probe_wait;
IUINT32 dead_link, incr;
struct IQUEUEHEAD snd_queue;
struct IQUEUEHEAD rcv_queue;
struct IQUEUEHEAD snd_buf;
struct IQUEUEHEAD rcv_buf;
IUINT32 *acklist;
IUINT32 ackcount;
IUINT32 ackblock;
IUINT32 ackedlen;
void *user;
char *buffer;
int fastresend;
int fastlimit;
int nocwnd, stream;
const struct IKCPOPS *ccops;
void *congest;
int logmask;
int (*output)(const char *buf, int len, struct IKCPCB *kcp, void *user);
void (*writelog)(const char *log, struct IKCPCB *kcp, void *user);
};
#define IKCP_LOG_OUTPUT 1
#define IKCP_LOG_INPUT 2
#define IKCP_LOG_SEND 4
#define IKCP_LOG_RECV 8
#define IKCP_LOG_IN_DATA 16
#define IKCP_LOG_IN_ACK 32
#define IKCP_LOG_IN_PROBE 64
#define IKCP_LOG_IN_WINS 128
#define IKCP_LOG_OUT_DATA 256
#define IKCP_LOG_OUT_ACK 512
#define IKCP_LOG_OUT_PROBE 1024
#define IKCP_LOG_OUT_WINS 2048
#ifdef __cplusplus
extern "C" {
#endif
//---------------------------------------------------------------------
// interface
//---------------------------------------------------------------------
// create a new kcp control object, 'conv' must be equal in both endpoints
// of the same connection. 'user' will be passed to the output callback.
// output callback can be set up like this: 'kcp->output = my_udp_output'
ikcpcb* ikcp_create(IUINT32 conv, void *user);
// release kcp control object
void ikcp_release(ikcpcb *kcp);
// set output callback, which will be invoked by kcp
void ikcp_setoutput(ikcpcb *kcp, int (*output)(const char *buf, int len,
ikcpcb *kcp, void *user));
// user/upper level recv: returns size, returns below zero for EAGAIN
int ikcp_recv(ikcpcb *kcp, char *buffer, int len);
// user/upper level send, returns below zero for error
int ikcp_send(ikcpcb *kcp, const char *buffer, int len);
// update state (call it repeatedly, every 10ms-100ms), or you can ask
// ikcp_check when to call it again (without ikcp_input/_send calling).
// 'current' - current timestamp in millisec.
void ikcp_update(ikcpcb *kcp, IUINT32 current);
// Determines when you should invoke ikcp_update next:
// returns the timestamp (in milliseconds) at which you should call
// ikcp_update, assuming no ikcp_input/_send calls occur in between.
// You can call ikcp_update at that time instead of calling it repeatedly.
// Important for reducing unnecessary ikcp_update invocations. Use it to
// schedule ikcp_update (e.g., implementing an epoll-like mechanism,
// or optimizing ikcp_update when handling massive kcp connections).
IUINT32 ikcp_check(const ikcpcb *kcp, IUINT32 current);
// when you receive a low-level packet (e.g., UDP packet), call this
int ikcp_input(ikcpcb *kcp, const char *data, long size);
// flush pending data
void ikcp_flush(ikcpcb *kcp);
// check the size of next message in the recv queue
int ikcp_peeksize(const ikcpcb *kcp);
// change MTU size, default is 1400
int ikcp_setmtu(ikcpcb *kcp, int mtu);
// set maximum window size: sndwnd=32, rcvwnd=32 by default
int ikcp_wndsize(ikcpcb *kcp, int sndwnd, int rcvwnd);
// get how many packets are waiting to be sent
int ikcp_waitsnd(const ikcpcb *kcp);
// fastest: ikcp_nodelay(kcp, 1, 20, 2, 1)
// nodelay: 0:disable (default), 1:enable
// interval: internal update timer interval in ms, default is 100ms
// resend: 0:disable fast resend (default), 1:enable fast resend
// nc: 0:normal congestion control (default), 1:disable congestion control
int ikcp_nodelay(ikcpcb *kcp, int nodelay, int interval, int resend, int nc);
// install congestion control algorithm, NULL restores builtin
int ikcp_setcc(ikcpcb *kcp, const struct IKCPOPS *ops);
// write log with kcp->writelog
void ikcp_log(ikcpcb *kcp, int mask, const char *fmt, ...);
// setup allocator
void ikcp_allocator(void* (*new_malloc)(size_t), void (*new_free)(void*));
// read conv
IUINT32 ikcp_getconv(const void *ptr);
#ifdef __cplusplus
}
#endif
#endif
+200
View File
@@ -0,0 +1,200 @@
#include "laylink_kcp.h"
#include "ikcp.h"
#include <stdlib.h>
#include <string.h>
typedef struct laylink_packet {
int len;
char* data;
struct laylink_packet* next;
} laylink_packet;
struct laylink_kcp {
ikcpcb* kcp;
laylink_packet* output_head;
laylink_packet* output_tail;
};
static void laylink_packet_free(laylink_packet* packet) {
if (packet == NULL) {
return;
}
free(packet->data);
free(packet);
}
static int laylink_kcp_output(const char* buffer, int len, ikcpcb* kcp, void* user) {
(void)kcp;
laylink_kcp* session = (laylink_kcp*)user;
if (session == NULL || len <= 0) {
return -1;
}
laylink_packet* packet = (laylink_packet*)malloc(sizeof(laylink_packet));
if (packet == NULL) {
return -2;
}
packet->data = (char*)malloc((size_t)len);
if (packet->data == NULL) {
free(packet);
return -3;
}
memcpy(packet->data, buffer, (size_t)len);
packet->len = len;
packet->next = NULL;
if (session->output_tail != NULL) {
session->output_tail->next = packet;
} else {
session->output_head = packet;
}
session->output_tail = packet;
return 0;
}
laylink_kcp* laylink_kcp_create(unsigned int conv) {
laylink_kcp* session = (laylink_kcp*)calloc(1, sizeof(laylink_kcp));
if (session == NULL) {
return NULL;
}
session->kcp = ikcp_create(conv, session);
if (session->kcp == NULL) {
free(session);
return NULL;
}
session->kcp->output = laylink_kcp_output;
ikcp_nodelay(session->kcp, 1, 10, 2, 1);
ikcp_wndsize(session->kcp, 1024, 1024);
ikcp_setmtu(session->kcp, 1350);
return session;
}
void laylink_kcp_release(laylink_kcp* session) {
if (session == NULL) {
return;
}
laylink_packet* packet = session->output_head;
while (packet != NULL) {
laylink_packet* next = packet->next;
laylink_packet_free(packet);
packet = next;
}
if (session->kcp != NULL) {
ikcp_release(session->kcp);
}
free(session);
}
int laylink_kcp_nodelay(laylink_kcp* session, int nodelay, int interval, int resend, int nc) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_nodelay(session->kcp, nodelay, interval, resend, nc);
}
int laylink_kcp_wndsize(laylink_kcp* session, int sndwnd, int rcvwnd) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_wndsize(session->kcp, sndwnd, rcvwnd);
}
int laylink_kcp_setmtu(laylink_kcp* session, int mtu) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_setmtu(session->kcp, mtu);
}
int laylink_kcp_send(laylink_kcp* session, const char* buffer, int len) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_send(session->kcp, buffer, len);
}
int laylink_kcp_input(laylink_kcp* session, const char* buffer, long size) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_input(session->kcp, buffer, size);
}
void laylink_kcp_update(laylink_kcp* session, unsigned int current) {
if (session == NULL || session->kcp == NULL) {
return;
}
ikcp_update(session->kcp, current);
}
unsigned int laylink_kcp_check(laylink_kcp* session, unsigned int current) {
if (session == NULL || session->kcp == NULL) {
return current + 10;
}
return ikcp_check(session->kcp, current);
}
int laylink_kcp_peeksize(laylink_kcp* session) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_peeksize(session->kcp);
}
int laylink_kcp_recv(laylink_kcp* session, char* buffer, int len) {
if (session == NULL || session->kcp == NULL) {
return -1;
}
return ikcp_recv(session->kcp, buffer, len);
}
void laylink_kcp_flush(laylink_kcp* session) {
if (session == NULL || session->kcp == NULL) {
return;
}
ikcp_flush(session->kcp);
}
int laylink_kcp_waitsnd(laylink_kcp* session) {
if (session == NULL || session->kcp == NULL) {
return 0;
}
return ikcp_waitsnd(session->kcp);
}
int laylink_kcp_pending_output_size(laylink_kcp* session) {
if (session == NULL || session->output_head == NULL) {
return 0;
}
return session->output_head->len;
}
int laylink_kcp_pop_output(laylink_kcp* session, char* buffer, int len) {
if (session == NULL || session->output_head == NULL) {
return 0;
}
laylink_packet* packet = session->output_head;
if (len < packet->len) {
return -packet->len;
}
memcpy(buffer, packet->data, (size_t)packet->len);
session->output_head = packet->next;
if (session->output_head == NULL) {
session->output_tail = NULL;
}
int packet_len = packet->len;
laylink_packet_free(packet);
return packet_len;
}
+30
View File
@@ -0,0 +1,30 @@
#ifndef LAYLINK_KCP_H
#define LAYLINK_KCP_H
#ifdef __cplusplus
extern "C" {
#endif
typedef struct laylink_kcp laylink_kcp;
laylink_kcp* laylink_kcp_create(unsigned int conv);
void laylink_kcp_release(laylink_kcp* session);
int laylink_kcp_nodelay(laylink_kcp* session, int nodelay, int interval, int resend, int nc);
int laylink_kcp_wndsize(laylink_kcp* session, int sndwnd, int rcvwnd);
int laylink_kcp_setmtu(laylink_kcp* session, int mtu);
int laylink_kcp_send(laylink_kcp* session, const char* buffer, int len);
int laylink_kcp_input(laylink_kcp* session, const char* buffer, long size);
void laylink_kcp_update(laylink_kcp* session, unsigned int current);
unsigned int laylink_kcp_check(laylink_kcp* session, unsigned int current);
int laylink_kcp_peeksize(laylink_kcp* session);
int laylink_kcp_recv(laylink_kcp* session, char* buffer, int len);
void laylink_kcp_flush(laylink_kcp* session);
int laylink_kcp_waitsnd(laylink_kcp* session);
int laylink_kcp_pending_output_size(laylink_kcp* session);
int laylink_kcp_pop_output(laylink_kcp* session, char* buffer, int len);
#ifdef __cplusplus
}
#endif
#endif
+4
View File
@@ -0,0 +1,4 @@
kcp有可供调整的参数吗?
存在堵塞现象。
PHP Warning: stream_socket_sendto(): Resource temporarily unavailable
in /www/laylink/vendor/workerman/workerman/src/Connection/UdpConnection.php on line 85
+81 -11
View File
@@ -25,7 +25,7 @@ LayLink 是一个基于 PHP Workerman 的策略控制型四层反向访问网关
cp .env.example .env
```
`.env.example` 中的 `[config]``[client-agent]``[pop-server]` 是阅读分组标题,当前加载器会忽略这些标题,只读取 `KEY=value` 配置行。
`.env.example` 中的 `[config]``[kcp]``[client-agent]``[pop-server]` 是阅读分组标题,当前加载器会忽略这些标题,只读取 `KEY=value` 配置行。
Agent 与 POP Server 之间的 LayLink Frame 支持加密:
@@ -74,7 +74,9 @@ POP Server 需要配置这些 `.env`
```env
APP_ENV=dev
POP_AGENT_LISTEN=0.0.0.0:9001
POP_ALLOWED_AGENT_TRANSPORTS=tcp
POP_ALLOWED_AGENT_TRANSPORTS=tcp,kcp
POP_AGENT_TCP_WORKERS=1
POP_AGENT_KCP_WORKERS=1
AUDIT_LOG=runtime/audit.log
LOG_LEVEL=debug
```
@@ -87,7 +89,9 @@ LOG_LEVEL=debug
| `LAYLINK_FRAME_ENCRYPTION` | Agent 与 POP Server 之间 Frame 加密方式,两端必须一致。 | `none``chacha20` |
| `LAYLINK_FRAME_ENCRYPTION_KEY` | Frame 加密密钥,启用 `chacha20` 时必填。 | 普通口令、`hex:...``base64:...` |
| `POP_AGENT_LISTEN` | POP Server 给 Client Agent 连接的监听地址。Agent 的 `POP_SERVER_ADDRESS` 应指向这里。 | `0.0.0.0:9001``127.0.0.1:9001` |
| `POP_ALLOWED_AGENT_TRANSPORTS` | POP Server 允许 Agent 使用的底层传输协议,逗号分隔。Agent 认证时会上报自己的选择,不在列表内会被拒绝。 | `tcp``tcp,kcp``tcp,udp,kcp` |
| `POP_ALLOWED_AGENT_TRANSPORTS` | POP Server 允许 Agent 使用的底层传输协议。支持逗号数组,也支持 JSON 数组。Agent 认证时会上报自己的选择,不在列表内会被拒绝。 | `tcp``tcp,kcp``["tcp","kcp"]` |
| `POP_AGENT_TCP_WORKERS` | POP TCP Agent listener 的 worker 数。TCP 模式可按 CPU 和并发提高。 | `1``2``4``8` |
| `POP_AGENT_KCP_WORKERS` | POP KCP/UDP Agent listener 的 worker 数。当前必须保持 `1`。 | `1` |
| `AUDIT_LOG` | 审计日志路径。MVP 使用 JSON Lines 追加写入。 | `runtime/audit.log` |
| `LOG_LEVEL` | 日志级别预留配置。当前 MVP 主要为后续日志工厂使用。 | `debug``info``warning``error` |
@@ -121,6 +125,7 @@ NODE_ID=client-01
NODE_TYPE=client
NODE_TOKEN=CHANGE_ME
AGENT_TRANSPORT_PROTOCOL=tcp
CLIENT_AGENT_POP_CONNECTIONS=1
CLIENT_AGENT_AUTH_TOKEN=dev-token
CLIENT_AGENT_USER_ID=admin
CLIENT_AGENT_SOCKS5_ENABLED=true
@@ -151,6 +156,7 @@ LOG_LEVEL=debug
| `NODE_TYPE` | 当前节点类型。Client Agent 必须配置为 `client`。 | `client` |
| `NODE_TOKEN` | 当前节点认证密钥。必须和 `config/nodes.php` 中同一 `NODE_ID``token` 一致。 | 强随机字符串,开发时可临时用 `CHANGE_ME` |
| `AGENT_TRANSPORT_PROTOCOL` | 当前 Agent 到 POP Server 使用的底层传输协议。必须被 POP Server 的 `POP_ALLOWED_AGENT_TRANSPORTS` 允许。 | `tcp``udp``kcp` |
| `CLIENT_AGENT_POP_CONNECTIONS` | Client Agent 到 POP Server 的并行长连接数量。新 TCP 会话会在已认证连接间轮询分配,适合多并发请求或多线程测速。 | `1``2``4` |
| `CLIENT_AGENT_AUTH_TOKEN` | SOCKS5/HTTP 代理入口生成 `OPEN` 帧时使用的客户端认证 token。 | `dev-token`,生产应替换 |
| `CLIENT_AGENT_USER_ID` | SOCKS5/HTTP 代理入口生成 `OPEN` 帧时使用的默认用户 ID。 | `admin``normal-user` |
| `CLIENT_AGENT_SOCKS5_ENABLED` | 是否启用 SOCKS5 本地入口。 | `true``false` |
@@ -181,7 +187,7 @@ Client Agent 的节点身份不是只写在 `.env` 中,POP Server 侧还必须
'192.168.0.0/16',
'10.10.0.0/16',
],
'allowed_ports' => [22, 80, 443, 3306, 5432],
'allowed_ports' => [22, 80, 443, '8080-10080', 3306, 5432],
'enabled' => true,
],
```
@@ -262,7 +268,7 @@ CLIENT_AGENT_SOCKS5_PASSWORD=change-this-password
| `auth_token` | 客户端认证 token。当前 MVP 固定为 `dev-token`。 | `dev-token` |
| `user_id` | 用户身份。POP Server 会用它匹配 `config/policies.php`。 | `admin``devops``normal-user` |
| `target_host` | 目标主机。 | `192.168.10.20``example.com` |
| `target_port` | 目标端口。 | `22``80``443``5432` |
| `target_port` | 目标端口。 | `22``80``443``8080``5432` |
| `protocol` | 目标协议。当前只支持 TCP。 | `tcp` |
| `route_hint` | 预留字段。新的最小路径由 POP Server 直连公网目标,通常不需要填写。 | `null` |
@@ -277,7 +283,7 @@ CLIENT_AGENT_SOCKS5_PASSWORD=change-this-password
'policy_id' => 'public-web-egress',
'users' => ['normal-user', 'admin', 'devops'],
'target_hosts' => ['*'],
'target_ports' => [80, 443],
'target_ports' => [80, 443, '8080-10080'],
'protocol' => 'tcp',
'route_type' => 'direct',
'enabled' => true,
@@ -286,10 +292,15 @@ CLIENT_AGENT_SOCKS5_PASSWORD=change-this-password
这条策略表示:
* `normal-user``admin``devops` 可以访问任意主机的 `80``443` 端口。
* `normal-user``admin``devops` 可以访问任意主机的 `80``443`,以及 `8080``10080` 端口。
* Client Agent 只负责把请求封装成 Frame 发到 POP Server。
* POP Server 校验策略后直接连接公网目标。
`target_ports``allowed_ports` 都支持两种写法:
* 单端口:`80`
* 端口范围:`'8080-10080'`
路由类型:
| `route_type` | 作用 |
@@ -304,7 +315,7 @@ CLIENT_AGENT_SOCKS5_PASSWORD=change-this-password
```env
APP_ENV=dev
POP_AGENT_LISTEN=127.0.0.1:9001
POP_ALLOWED_AGENT_TRANSPORTS=tcp
POP_ALLOWED_AGENT_TRANSPORTS=tcp,kcp
NODE_ID=client-01
NODE_TYPE=client
NODE_TOKEN=CHANGE_ME
@@ -341,23 +352,64 @@ Agent 到 POP Server 的业务数据始终使用 LayLink 自定义 Frame 协议
| --- | --- | --- |
| `tcp` | Frame over TCP,最容易部署和调试。 | 已实现 |
| `udp` | Frame over UDP,需要额外处理可靠性、顺序和丢包。 | 已预留,未实现 |
| `kcp` | Frame over KCP/UDP用 KCP 做可靠、低延迟传输。 | 已预留,未实现 |
| `kcp` | Frame over KCP/UDP默认通过 FFI 调用 native `ikcp.c`。 | 已实现,需构建动态库 |
POP Server 用 `POP_ALLOWED_AGENT_TRANSPORTS` 控制允许哪些传输协议。例如:
```env
POP_ALLOWED_AGENT_TRANSPORTS=tcp,kcp
# 也可以写成:
POP_ALLOWED_AGENT_TRANSPORTS=["tcp","kcp"]
```
Client Agent 用 `AGENT_TRANSPORT_PROTOCOL` 选择自己实际使用哪种协议。例如:
```env
AGENT_TRANSPORT_PROTOCOL=tcp
AGENT_TRANSPORT_PROTOCOL=kcp
```
如果 Agent 选择的协议不在 POP 允许列表中,POP 会在认证阶段返回 `AUTH_FAIL`,原因是 `transport_not_allowed`
当前代码只实现了 `tcp`。如果 Agent 配置为 `udp` `kcp`,进程会启动失败并明确提示该传输尚未实现。
`kcp` 默认使用 FFI 调用 native `ikcp.c`。首次使用前需要构建动态库:
```bash
scripts/build-kcp-ffi.sh
```
然后两端配置:
```env
LAYLINK_KCP_BACKEND=ffi
LAYLINK_KCP_FFI_LIB=native/kcp/liblaylink_kcp.so
```
使用 `kcp` 时,POP Server 会在 `POP_AGENT_LISTEN` 的同一 host:port 上监听 UDPClient Agent 的 `POP_SERVER_ADDRESS` 仍填写同一地址即可。服务器防火墙需要放行同端口 UDP。
如果运行环境暂时不能启用 FFI,可以配置 `LAYLINK_KCP_BACKEND=php` 使用调试回退实现;该实现不适合作为生产高吞吐路径。
KCP 可调参数:
| 变量 | 作用 | 建议 |
| --- | --- | --- |
| `LAYLINK_KCP_NODELAY` | KCP nodelay 开关。 | 默认 `1` |
| `LAYLINK_KCP_INTERVAL_MS` | KCP update 间隔,越小越低延迟但发包/CPU 更高。 | `10``20``30` |
| `LAYLINK_KCP_FAST_RESEND` | 快速重传阈值。 | `2`,拥堵时试 `3``4` |
| `LAYLINK_KCP_NO_CONGESTION_CONTROL` | 是否关闭 KCP 拥塞控制。 | 公网建议 `0`,内网压测可试 `1` |
| `LAYLINK_KCP_SEND_WINDOW` | KCP 发送窗口。 | `128``256``512` |
| `LAYLINK_KCP_RECV_WINDOW` | KCP 接收窗口。 | `256``512``1024` |
| `LAYLINK_KCP_MTU_BYTES` | KCP MTU。 | 公网建议 `1200`,内网可试 `1350` |
| `LAYLINK_KCP_TICK_MS` | PHP transport tick 间隔。 | 通常等于 `LAYLINK_KCP_INTERVAL_MS` |
| `LAYLINK_KCP_UDP_SEND_QUEUE_BYTES` | UDP 发送遇到 EAGAIN 时的本地排队上限。 | `16777216``33554432` |
| `LAYLINK_KCP_UDP_FLUSH_PACKETS` | 每次 tick 最多刷出的 UDP packet 数。 | 拥堵时 `64`/`128`,吞吐压测 `256`/`512` |
| `LAYLINK_KCP_OUTPUT_DRAIN_PACKETS` | 每次从 native KCP 输出队列搬到 UDP 发送队列的最大 packet 数。 | 延迟敏感用 `64`/`128`,吞吐压测用 `256`/`512` |
如果出现 `stream_socket_sendto(): Resource temporarily unavailable`,说明 UDP socket 发送缓冲暂时满了。LayLink 会把 KCP packet 放入本地发送队列并在后续 tick 重试;同时建议把 `LAYLINK_KCP_NO_CONGESTION_CONTROL` 保持为 `0`,必要时降低 `LAYLINK_KCP_SEND_WINDOW``LAYLINK_KCP_UDP_FLUSH_PACKETS` 或调大 `LAYLINK_KCP_INTERVAL_MS`
KCP/UDP 目前不要把 `POP_AGENT_KCP_WORKERS` 调大。KCP 会话状态存在单个 worker 进程里,UDP 多 worker 会让同一个会话的数据包分散到不同进程,导致找不到会话、重传增加甚至断流。要横向扩展 KCP,当前推荐启动多个 POP 端口或多个 POP 实例,由 Client Agent 配置多 POP/多进程策略;后续可以做 `SO_REUSEPORT` 五元组哈希、外部 session 表或每 conv 固定 worker 分发。
单 worker 并不等于一个大下载会同步阻塞其他请求:目标连接、UDP socket 和本地客户端都是非阻塞 I/O。但大文件会带来大量 KCP 分片和加密/解密/FFI 调用,可能短时间占用事件循环 CPU。延迟敏感场景可降低 `LAYLINK_DATA_CHUNK_BYTES``LAYLINK_KCP_OUTPUT_DRAIN_PACKETS``LAYLINK_KCP_UDP_FLUSH_PACKETS`,让一个大流量会话每次 tick 少占一点时间,换取更好的多会话公平性。
如果 Agent 配置为 `udp`,进程会启动失败并明确提示该传输尚未实现。
启动 POP Server
@@ -373,6 +425,24 @@ php bin/client-agent.php start
然后把应用的代理设置为 SOCKS5 `127.0.0.1:1080`。Client Agent 会解析 SOCKS5 `CONNECT`,封装成 `OPEN` 帧发给 POP ServerPOP Server 校验通过后直连公网目标,随后通过 `DATA` 帧转发原始 TCP 数据。
TCP 大流量 `DATA` 帧使用二进制帧编码;`AUTH``OPEN``CLOSE``ERROR` 等控制帧仍使用 JSON 编码。启用 `chacha20` 时,二进制和 JSON Frame body 都会被加密。
吞吐相关参数由两端共用,POP Server 和 Client Agent 建议保持一致:
```env
LAYLINK_DATA_CHUNK_BYTES=1048576
LAYLINK_MAX_SEND_BUFFER_BYTES=67108864
LAYLINK_BACKPRESSURE_HIGH_WATERMARK_BYTES=33554432
```
`LAYLINK_DATA_CHUNK_BYTES` 越大,每 MB 需要处理的 Frame 越少,单连接下载通常越快;如果多会话公平性变差,可以降到 `262144``524288``LAYLINK_BACKPRESSURE_HIGH_WATERMARK_BYTES` 必须小于 `LAYLINK_MAX_SEND_BUFFER_BYTES`
大文件下载时,LayLink 会使用 Workerman 的 `pauseRecv()` / `resumeRecv()` 做背压:当下游发送缓冲区过高时暂停上游读取,缓冲排空后继续读取。这可以避免单个慢连接无限堆积内存或因为发送缓冲区满而断联。
当 POP 收到目标站关闭连接时,Client Agent 会先等待本地客户端发送缓冲区排空,再关闭本地 socket,避免大文件尾部数据还在缓冲区里时被提前截断。TCP `DATA` 默认按 1 MiB 分片发送,以减少帧开销;可通过 `LAYLINK_DATA_CHUNK_BYTES` 调整。
Client Agent 默认使用 1 条 Agent-to-POP TCP 长连接。可通过 `CLIENT_AGENT_POP_CONNECTIONS` 增加并行长连接数,新 TCP 会话会在已认证 POP 连接之间轮询分配,并在会话生命周期内固定使用同一条连接。它主要改善多并发请求、多线程下载或测速场景;单个 TCP 下载是否变快取决于客户端和目标站是否本身使用多连接。背压可以保护进程不堵死,但单条会话仍受单 TCP 流限制;KCP、per-session window 和更细粒度 TCP 调优仍是后续性能优化方向。
验证 SOCKS5 HTTPS 联通性和出口 IP
```bash
-63
View File
@@ -1,63 +0,0 @@
2026-05-28 10:19:07 pid:478270 Workerman[pop-server.php] start in DEBUG mode
2026-05-28 10:19:09 pid:478270 Workerman[pop-server.php] received signal SIGTERM
2026-05-28 10:19:09 pid:478270 Workerman[pop-server.php] stopping
2026-05-28 10:19:09 pid:478270 Workerman[pop-server.php] has been stopped
2026-05-28 11:13:28 pid:481004 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 11:13:30 pid:481004 Workerman[client-agent.php] received signal SIGTERM
2026-05-28 11:13:30 pid:481004 Workerman[client-agent.php] stopping
2026-05-28 11:13:30 pid:481004 Workerman[client-agent.php] has been stopped
2026-05-28 11:21:31 pid:482154 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 11:21:32 pid:482187 Workerman[pop-server.php] start in DEBUG mode
2026-05-28 11:21:33 pid:482154 Workerman[client-agent.php] received signal SIGTERM
2026-05-28 11:21:33 pid:482154 Workerman[client-agent.php] stopping
2026-05-28 11:21:33 pid:482154 Workerman[client-agent.php] has been stopped
2026-05-28 11:21:34 pid:482187 Workerman[pop-server.php] received signal SIGTERM
2026-05-28 11:21:34 pid:482187 Workerman[pop-server.php] stopping
2026-05-28 11:21:34 pid:482187 Workerman[pop-server.php] has been stopped
2026-05-28 11:32:55 pid:484077 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 11:32:57 pid:484077 Workerman[client-agent.php] received signal SIGTERM
2026-05-28 11:32:57 pid:484077 Workerman[client-agent.php] stopping
2026-05-28 11:32:57 pid:484077 Workerman[client-agent.php] has been stopped
2026-05-28 11:34:04 pid:484561 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 11:34:06 pid:484561 Workerman[client-agent.php] received signal SIGTERM
2026-05-28 11:34:06 pid:484561 Workerman[client-agent.php] stopping
2026-05-28 11:34:06 pid:484561 Workerman[client-agent.php] has been stopped
2026-05-28 11:36:48 pid:485045 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 11:36:50 pid:485045 Workerman[client-agent.php] received signal SIGTERM
2026-05-28 11:36:50 pid:485045 Workerman[client-agent.php] stopping
2026-05-28 11:36:50 pid:485045 Workerman[client-agent.php] has been stopped
2026-05-28 11:41:26 pid:486188 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 11:41:28 pid:486188 Workerman[client-agent.php] received signal SIGTERM
2026-05-28 11:41:28 pid:486188 Workerman[client-agent.php] stopping
2026-05-28 11:41:28 pid:486188 Workerman[client-agent.php] has been stopped
2026-05-28 12:08:09 pid:489103 Workerman[pop-server.php] start in DEBUG mode
2026-05-28 12:08:11 pid:489103 Workerman[pop-server.php] received signal SIGTERM
2026-05-28 12:08:11 pid:489103 Workerman[pop-server.php] stopping
2026-05-28 12:08:11 pid:489103 Workerman[pop-server.php] has been stopped
2026-05-28 12:28:14 pid:489644 Workerman[client-agent.php] start in DEBUG mode
2026-05-28 12:34:47 pid:490774 Workerman[client-agent.php] restart
2026-05-28 12:34:47 pid:490774 Workerman[client-agent.php] is stopping ...
2026-05-28 12:34:47 pid:489644 Workerman[client-agent.php] received signal SIGINT
2026-05-28 12:34:47 pid:489644 Workerman[client-agent.php] stopping
2026-05-28 12:34:48 pid:489644 Workerman[client-agent.php] has been stopped
2026-05-28 12:34:48 pid:490774 Workerman[client-agent.php] stop success
2026-05-28 12:35:17 pid:490876 Workerman[client-agent.php] stop
2026-05-28 12:35:17 pid:490876 Workerman[client-agent.php] is stopping ...
2026-05-28 12:35:17 pid:490774 Workerman[client-agent.php] received signal SIGINT
2026-05-28 12:35:17 pid:490774 Workerman[client-agent.php] stopping
2026-05-28 12:35:19 pid:490774 Workerman[client-agent.php] has been stopped
2026-05-28 12:35:19 pid:490876 Workerman[client-agent.php] stop success
2026-05-28 12:35:19 pid:490845 Workerman[client-agent.php] start in DAEMON mode
2026-05-28 12:40:44 pid:491096 Workerman[client-agent.php] restart
2026-05-28 12:40:44 pid:491096 Workerman[client-agent.php] is stopping ...
2026-05-28 12:40:44 pid:490878 Workerman[client-agent.php] received signal SIGINT
2026-05-28 12:40:44 pid:490878 Workerman[client-agent.php] stopping
2026-05-28 12:40:44 pid:490878 Workerman[client-agent.php] has been stopped
2026-05-28 12:40:44 pid:491096 Workerman[client-agent.php] stop success
2026-05-28 13:04:58 pid:492360 Workerman[pop-server.php] start in DEBUG mode
2026-05-28 13:05:00 pid:492360 Workerman[pop-server.php] received signal SIGTERM
2026-05-28 13:05:00 pid:492360 Workerman[pop-server.php] stopping
2026-05-28 13:05:00 pid:492360 Workerman[pop-server.php] has been stopped
2026-05-28 13:07:21 pid:491096 Workerman[client-agent.php] received signal SIGINT
2026-05-28 13:07:21 pid:491096 Workerman[client-agent.php] stopping
2026-05-28 13:07:21 pid:491096 Workerman[client-agent.php] has been stopped
+14
View File
@@ -0,0 +1,14 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
SRC_DIR="${ROOT_DIR}/native/kcp"
OUT="${SRC_DIR}/liblaylink_kcp.so"
cc="${CC:-gcc}"
"${cc}" -O3 -fPIC -shared \
"${SRC_DIR}/ikcp.c" \
"${SRC_DIR}/laylink_kcp.c" \
-o "${OUT}"
echo "${OUT}"
+1 -1
View File
@@ -24,7 +24,7 @@ echo "Using SOCKS5 proxy: ${proxy}"
echo
echo "[1/2] HTTPS connectivity: https://bing.com/"
bing_code="$(
curl \
curl -vvvv \
--silent \
--show-error \
--location \
+281 -72
View File
@@ -5,11 +5,10 @@ declare(strict_types=1);
namespace LayLink\Agent;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use LayLink\Protocol\FrameType;
use LayLink\Transport\FrameClientTransport;
use LayLink\Transport\FrameClientTransportFactory;
use LayLink\Util\Uuid;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Connection\UdpConnection;
use Workerman\Timer;
@@ -17,8 +16,13 @@ use Workerman\Worker;
final class AgentClient
{
private ?AsyncTcpConnection $pop = null;
private ?FrameParser $parser = null;
/** @var array<int, FrameClientTransport> */
private array $pops = [];
/** @var array<int, true> */
private array $authenticatedPops = [];
/** @var array<string, FrameClientTransport> */
private array $sessionTransports = [];
private int $nextTransportCursor = 0;
private bool $authenticated = false;
/** @var array<int, string> */
private array $initialBuffers = [];
@@ -38,6 +42,14 @@ final class AgentClient
private array $udpClients = [];
/** @var array<string, array{client_address: string, target_host: string, target_port: int}> */
private array $udpSessions = [];
/** @var array<int, TcpConnection> */
private array $pausedClientsForPop = [];
/** @var array<int, true> */
private array $clientsPausingPop = [];
/** @var array<string, true> */
private array $pendingClientCloses = [];
/** @var array<int, true> */
private array $suppressClientCloseFrames = [];
public function __construct(
private readonly string $clientListen,
@@ -55,6 +67,10 @@ final class AgentClient
private readonly string $socks5Password = '',
private readonly ?string $socks5UdpListen = null,
private readonly string $socks5UdpAdvertiseIp = '127.0.0.1',
private readonly int $maxSendBuffer = 64 * 1024 * 1024,
private readonly int $backpressureHighWatermark = 32 * 1024 * 1024,
private readonly int $dataChunkSize = 1024 * 1024,
private readonly int $popConnectionCount = 1,
) {
}
@@ -64,9 +80,6 @@ final class AgentClient
$worker->name = $workerName;
$worker->count = 1;
$worker->onWorkerStart = function (): void {
if ($this->transportProtocol !== 'tcp') {
throw new \RuntimeException("Agent transport '{$this->transportProtocol}' is configured but not implemented yet.");
}
$this->connect();
Timer::add(10, fn () => $this->heartbeat());
};
@@ -84,67 +97,68 @@ final class AgentClient
private function connect(): void
{
$this->parser = new FrameParser();
$this->authenticated = false;
$connection = new AsyncTcpConnection($this->popAddress);
$connection->maxSendBufferSize = 8 * 1024 * 1024;
$this->pop = $connection;
$connection->onConnect = function (AsyncTcpConnection $connection): void {
$this->send(new Frame(FrameType::AUTH, null, [
'node_id' => $this->nodeId,
'node_type' => $this->nodeType,
'node_zone' => $this->nodeZone,
'node_token' => $this->nodeToken,
'transport_protocol' => $this->transportProtocol,
'supported_protocols' => ['tcp'],
'supported_transports' => ['tcp', 'udp', 'kcp'],
]));
};
$connection->onMessage = function (AsyncTcpConnection $connection, string $data): void {
try {
foreach ($this->parser?->push($data) ?? [] as $frame) {
$this->handleFrame($frame);
}
} catch (\Throwable $e) {
$connection->close();
}
};
$connection->onClose = function (): void {
$this->authenticated = false;
foreach ($this->clients as $client) {
$client->close();
}
$this->initialBuffers = [];
$this->connectionSessionIds = [];
$this->clients = [];
$this->sessionStates = [];
$this->pendingData = [];
$this->connectionStages = [];
$this->sessionIngressProtocols = [];
Timer::add(3, fn () => $this->connect(), [], false);
};
$connection->connect();
$this->pops = [];
$this->authenticatedPops = [];
for ($i = 0; $i < $this->popConnectionCount; $i++) {
$this->connectOne();
}
}
private function handleFrame(Frame $frame): void
private function connectOne(): void
{
$transport = (new FrameClientTransportFactory())->create(
$this->transportProtocol,
$this->popAddress,
$this->maxSendBuffer,
function (FrameClientTransport $transport): void {
$transport->send(new Frame(FrameType::AUTH, null, [
'node_id' => $this->nodeId,
'node_type' => $this->nodeType,
'node_zone' => $this->nodeZone,
'node_token' => $this->nodeToken,
'transport_protocol' => $this->transportProtocol,
'supported_protocols' => ['tcp'],
'supported_transports' => ['tcp', 'udp', 'kcp'],
]));
},
function (FrameClientTransport $transport, Frame $frame): void {
$this->handleFrame($transport, $frame);
},
function (FrameClientTransport $transport): void {
$this->handlePopClose($transport);
Timer::add(3, fn () => $this->connectOne(), [], false);
},
function (FrameClientTransport $transport): void {
$this->resumeClientsForPop($transport);
},
function (\Throwable $e): void {
},
);
$this->pops[spl_object_id($transport)] = $transport;
$transport->connect();
}
private function handleFrame(FrameClientTransport $transport, Frame $frame): void
{
match ($frame->type) {
FrameType::AUTH_OK => $this->authenticated = true,
FrameType::AUTH_FAIL => $this->pop?->close(),
FrameType::AUTH_OK => $this->markPopAuthenticated($transport),
FrameType::AUTH_FAIL => $transport->close(),
FrameType::PONG => null,
FrameType::OPEN_OK => $this->openClientSession($frame),
FrameType::OPEN_FAIL => $this->failClientSession($frame),
FrameType::DATA => $this->forwardToClient($frame),
FrameType::UDP_DATA => $this->forwardUdpToClient($frame),
FrameType::CLOSE => $this->closeClient($frame->sessionId),
FrameType::CLOSE => $this->closeClient($frame->sessionId, true),
default => null,
};
}
private function onClientConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = 8 * 1024 * 1024;
$connection->maxSendBufferSize = $this->maxSendBuffer;
$connection->onBufferFull = fn (TcpConnection $connection) => $this->pausePopForClient($connection);
$connection->onBufferDrain = fn (TcpConnection $connection) => $this->onClientBufferDrain($connection);
$this->initialBuffers[$connection->id] = '';
$this->connectionStages[$connection->id] = 'init';
}
@@ -162,7 +176,15 @@ final class AgentClient
return;
}
$this->send(new Frame(FrameType::DATA, $sessionId, ['data' => base64_encode($data)]));
if (!$this->sendData($sessionId, $data)) {
$this->closeClient($sessionId);
return;
}
$transport = $this->sessionTransports[$sessionId] ?? null;
if ($transport !== null && $transport->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
$connection->pauseRecv();
$this->pausedClientsForPop[$connection->id] = $connection;
}
}
private function handleInitialRequest(TcpConnection $connection, string $data): void
@@ -422,7 +444,8 @@ final class AgentClient
private function startPopSession(TcpConnection $connection, array $request, string $payloadBytes, string $ingressProtocol): void
{
if (!$this->authenticated || $this->pop === null) {
$transport = $this->selectTransport();
if (!$this->authenticated || $transport === null) {
$this->failOpeningLocalClient($connection, $ingressProtocol, 'pop_not_connected');
return;
}
@@ -430,13 +453,14 @@ final class AgentClient
$sessionId = Uuid::v4();
$this->connectionSessionIds[$connection->id] = $sessionId;
$this->clients[$sessionId] = $connection;
$this->sessionTransports[$sessionId] = $transport;
$this->sessionStates[$sessionId] = 'opening';
$this->sessionIngressProtocols[$sessionId] = $ingressProtocol;
if ($payloadBytes !== '') {
$this->pendingData[$sessionId] = $payloadBytes;
}
$this->send(new Frame(FrameType::OPEN, $sessionId, [
$transport->send(new Frame(FrameType::OPEN, $sessionId, [
'auth_token' => (string)($request['auth_token'] ?? ''),
'user_id' => (string)($request['user_id'] ?? ''),
'target_host' => (string)($request['target_host'] ?? ''),
@@ -461,7 +485,7 @@ final class AgentClient
private function onSocks5UdpMessage(UdpConnection $connection, string $data): void
{
if (!$this->authenticated || $this->pop === null || strlen($data) < 10) {
if (!$this->authenticated || $this->selectTransport() === null || strlen($data) < 10) {
return;
}
@@ -544,7 +568,7 @@ final class AgentClient
$pending = $this->pendingData[$frame->sessionId] ?? '';
unset($this->pendingData[$frame->sessionId]);
if ($pending !== '') {
$this->send(new Frame(FrameType::DATA, $frame->sessionId, ['data' => base64_encode($pending)]));
$this->sendData($frame->sessionId, $pending);
}
}
@@ -573,38 +597,80 @@ final class AgentClient
return;
}
$data = base64_decode((string)($frame->payload['data'] ?? ''), true);
$data = $this->frameData($frame);
if ($data === false) {
$this->send(new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'invalid_frame']));
return;
}
$this->clients[$frame->sessionId]->send($data);
$client = $this->clients[$frame->sessionId];
$sent = $client->send($data);
if ($sent === false) {
$this->closeClient($frame->sessionId);
return;
}
if ($client->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
$this->pausePopForClient($client);
}
}
private function frameData(Frame $frame): string|false
{
if (isset($frame->payload['data_raw']) && is_string($frame->payload['data_raw'])) {
return $frame->payload['data_raw'];
}
return base64_decode((string)($frame->payload['data'] ?? ''), true);
}
private function onClientClose(TcpConnection $connection): void
{
unset($this->initialBuffers[$connection->id]);
unset($this->connectionStages[$connection->id]);
unset($this->pausedClientsForPop[$connection->id], $this->clientsPausingPop[$connection->id]);
$suppressCloseFrame = isset($this->suppressClientCloseFrames[$connection->id]);
unset($this->suppressClientCloseFrames[$connection->id]);
if (!isset($this->connectionSessionIds[$connection->id])) {
return;
}
$sessionId = $this->connectionSessionIds[$connection->id];
unset($this->pendingClientCloses[$sessionId]);
unset($this->connectionSessionIds[$connection->id]);
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId]);
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId], $this->sessionTransports[$sessionId]);
if ($suppressCloseFrame) {
return;
}
$this->send(new Frame(FrameType::CLOSE, $sessionId, ['reason' => 'client_closed']));
}
private function closeClient(?string $sessionId): void
private function closeClient(?string $sessionId, bool $graceful = false): void
{
if ($sessionId === null || !isset($this->clients[$sessionId])) {
return;
}
$connection = $this->clients[$sessionId];
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId]);
if ($graceful && $connection->getSendBufferQueueSize() > 0) {
$this->pendingClientCloses[$sessionId] = true;
$this->suppressClientCloseFrames[$connection->id] = true;
return;
}
$this->finalizeClientClose($sessionId);
}
private function finalizeClientClose(string $sessionId): void
{
if (!isset($this->clients[$sessionId])) {
return;
}
$connection = $this->clients[$sessionId];
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId], $this->sessionTransports[$sessionId]);
unset($this->connectionSessionIds[$connection->id]);
unset($this->pausedClientsForPop[$connection->id], $this->clientsPausingPop[$connection->id], $this->pendingClientCloses[$sessionId]);
$this->suppressClientCloseFrames[$connection->id] = true;
$connection->close();
}
@@ -745,20 +811,163 @@ final class AgentClient
private function heartbeat(): void
{
if (!$this->authenticated || $this->pop === null) {
if (!$this->authenticated) {
return;
}
$this->send(new Frame(FrameType::PING, null, [
'node_id' => $this->nodeId,
'active_sessions' => count($this->clients),
'load' => sys_getloadavg()[0] ?? 0.0,
'timestamp' => time(),
]));
foreach ($this->authenticatedTransports() as $transport) {
$transport->send(new Frame(FrameType::PING, null, [
'node_id' => $this->nodeId,
'active_sessions' => $this->activeSessionsForTransport($transport),
'load' => sys_getloadavg()[0] ?? 0.0,
'timestamp' => time(),
]));
}
}
private function send(Frame $frame): void
private function send(Frame $frame): bool|null
{
$this->pop?->send(FrameCodec::encode($frame));
return $this->transportForSession($frame->sessionId)?->send($frame);
}
private function sendData(string $sessionId, string $data): bool
{
$length = strlen($data);
for ($offset = 0; $offset < $length; $offset += $this->dataChunkSize) {
$sent = $this->send(new Frame(FrameType::DATA, $sessionId, [
'data_raw' => substr($data, $offset, $this->dataChunkSize),
]));
if ($sent === false) {
return false;
}
}
return true;
}
private function resumeClientsForPop(FrameClientTransport $transport): void
{
foreach ($this->pausedClientsForPop as $connectionId => $client) {
if ($this->transportForConnection($client) !== $transport) {
continue;
}
$client->resumeRecv();
unset($this->pausedClientsForPop[$connectionId]);
}
}
private function pausePopForClient(TcpConnection $connection): void
{
$transport = $this->transportForConnection($connection);
if ($transport === null) {
return;
}
$this->clientsPausingPop[$connection->id] = true;
$transport->pauseRecv();
}
private function resumePopForClient(TcpConnection $connection): void
{
unset($this->clientsPausingPop[$connection->id]);
$transport = $this->transportForConnection($connection);
if ($transport !== null && !$this->hasClientsPausingTransport($transport)) {
$transport->resumeRecv();
}
}
private function onClientBufferDrain(TcpConnection $connection): void
{
$this->resumePopForClient($connection);
$sessionId = $this->connectionSessionIds[$connection->id] ?? null;
if (is_string($sessionId) && isset($this->pendingClientCloses[$sessionId])) {
$this->finalizeClientClose($sessionId);
}
}
private function markPopAuthenticated(FrameClientTransport $transport): void
{
$this->authenticatedPops[spl_object_id($transport)] = true;
$this->authenticated = true;
}
private function handlePopClose(FrameClientTransport $transport): void
{
$transportId = spl_object_id($transport);
unset($this->pops[$transportId], $this->authenticatedPops[$transportId]);
$this->authenticated = $this->authenticatedPops !== [];
foreach ($this->sessionTransports as $sessionId => $sessionTransport) {
if ($sessionTransport !== $transport || !isset($this->clients[$sessionId])) {
continue;
}
$client = $this->clients[$sessionId];
$this->suppressClientCloseFrames[$client->id] = true;
$client->close();
}
}
private function selectTransport(): ?FrameClientTransport
{
$transports = $this->authenticatedTransports();
if ($transports === []) {
return null;
}
$transport = $transports[$this->nextTransportCursor % count($transports)];
$this->nextTransportCursor++;
return $transport;
}
/**
* @return FrameClientTransport[]
*/
private function authenticatedTransports(): array
{
$transports = [];
foreach ($this->authenticatedPops as $transportId => $_) {
if (isset($this->pops[$transportId])) {
$transports[] = $this->pops[$transportId];
}
}
return $transports;
}
private function transportForSession(?string $sessionId): ?FrameClientTransport
{
if ($sessionId !== null && isset($this->sessionTransports[$sessionId])) {
return $this->sessionTransports[$sessionId];
}
return $this->selectTransport();
}
private function transportForConnection(TcpConnection $connection): ?FrameClientTransport
{
$sessionId = $this->connectionSessionIds[$connection->id] ?? null;
return is_string($sessionId) ? ($this->sessionTransports[$sessionId] ?? null) : null;
}
private function hasClientsPausingTransport(FrameClientTransport $transport): bool
{
foreach ($this->clientsPausingPop as $connectionId => $_) {
$sessionId = $this->connectionSessionIds[$connectionId] ?? null;
if (is_string($sessionId) && ($this->sessionTransports[$sessionId] ?? null) === $transport) {
return true;
}
}
return false;
}
private function activeSessionsForTransport(FrameClientTransport $transport): int
{
$count = 0;
foreach ($this->sessionTransports as $sessionId => $sessionTransport) {
if ($sessionTransport === $transport && isset($this->clients[$sessionId])) {
$count++;
}
}
return $count;
}
}
+2 -1
View File
@@ -5,6 +5,7 @@ declare(strict_types=1);
namespace LayLink\Agent;
use LayLink\Auth\PolicyChecker;
use LayLink\Auth\PortMatcher;
final class TargetConnector
{
@@ -14,7 +15,7 @@ final class TargetConnector
public function isAllowed(string $host, int $port): bool
{
if (!in_array($port, $this->nodeConfig['allowed_ports'] ?? [], true)) {
if (!PortMatcher::matches($port, $this->nodeConfig['allowed_ports'] ?? [])) {
return false;
}
+1 -1
View File
@@ -22,7 +22,7 @@ final class PolicyChecker
if (($policy['protocol'] ?? 'tcp') !== $protocol) {
continue;
}
if (!in_array($port, $policy['target_ports'] ?? [], true)) {
if (!PortMatcher::matches($port, $policy['target_ports'] ?? [])) {
continue;
}
if (!$this->hostMatches($host, $policy['target_hosts'] ?? [])) {
+47
View File
@@ -0,0 +1,47 @@
<?php
declare(strict_types=1);
namespace LayLink\Auth;
final class PortMatcher
{
public static function matches(int $port, array $rules): bool
{
foreach ($rules as $rule) {
if (is_int($rule) && $rule === $port) {
return true;
}
if (is_string($rule) && self::stringRuleMatches($port, $rule)) {
return true;
}
}
return false;
}
private static function stringRuleMatches(int $port, string $rule): bool
{
$rule = trim($rule);
if ($rule === '') {
return false;
}
if (ctype_digit($rule)) {
return (int)$rule === $port;
}
if (!preg_match('/^(\d{1,5})\s*-\s*(\d{1,5})$/', $rule, $matches)) {
return false;
}
$start = (int)$matches[1];
$end = (int)$matches[2];
if ($start < 1 || $end > 65535 || $start > $end) {
return false;
}
return $port >= $start && $port <= $end;
}
}
+2 -2
View File
@@ -4,7 +4,7 @@ declare(strict_types=1);
namespace LayLink\Node;
use Workerman\Connection\TcpConnection;
use LayLink\Transport\FrameServerConnection;
final class NodeConnection
{
@@ -15,7 +15,7 @@ final class NodeConnection
public readonly string $nodeId,
public readonly string $nodeType,
public readonly string $nodeZone,
public readonly TcpConnection $connection,
public readonly FrameServerConnection $connection,
) {
$this->lastHeartbeat = time();
}
+38 -10
View File
@@ -4,23 +4,30 @@ declare(strict_types=1);
namespace LayLink\Node;
use Workerman\Connection\TcpConnection;
use LayLink\Transport\FrameServerConnection;
final class NodeRegistry
{
/** @var array<string, NodeConnection> */
/** @var array<string, array<int, NodeConnection>> */
private array $nodes = [];
public function register(NodeConnection $node): void
{
$this->nodes[$node->nodeId] = $node;
$this->nodes[$node->nodeId][$node->connection->id()] = $node;
}
public function unregisterByConnection(TcpConnection $connection): ?NodeConnection
public function unregisterByConnection(FrameServerConnection $connection): ?NodeConnection
{
foreach ($this->nodes as $nodeId => $node) {
if ($node->connection === $connection) {
unset($this->nodes[$nodeId]);
foreach ($this->nodes as $nodeId => $connections) {
foreach ($connections as $connectionId => $node) {
if ($node->connection !== $connection) {
continue;
}
unset($this->nodes[$nodeId][$connectionId]);
if ($this->nodes[$nodeId] === []) {
unset($this->nodes[$nodeId]);
}
return $node;
}
}
@@ -30,12 +37,26 @@ final class NodeRegistry
public function get(string $nodeId): ?NodeConnection
{
return $this->nodes[$nodeId] ?? null;
$connections = $this->nodes[$nodeId] ?? [];
return $connections === [] ? null : array_values($connections)[0];
}
public function getByConnection(FrameServerConnection $connection): ?NodeConnection
{
foreach ($this->nodes as $connections) {
foreach ($connections as $node) {
if ($node->connection === $connection) {
return $node;
}
}
}
return null;
}
public function isOnline(string $nodeId): bool
{
return isset($this->nodes[$nodeId]);
return ($this->nodes[$nodeId] ?? []) !== [];
}
/**
@@ -43,6 +64,13 @@ final class NodeRegistry
*/
public function all(): array
{
return array_values($this->nodes);
$nodes = [];
foreach ($this->nodes as $connections) {
foreach ($connections as $node) {
$nodes[] = $node;
}
}
return $nodes;
}
}
+52 -8
View File
@@ -9,6 +9,8 @@ use InvalidArgumentException;
final class FrameCodec
{
public const MAX_FRAME_LENGTH = 16 * 1024 * 1024;
private const BINARY_MAGIC = "LLB1";
private const BINARY_TYPE_DATA = 1;
private const ENCRYPTION_NONE = 'none';
private const ENCRYPTION_CHACHA20 = 'chacha20';
@@ -40,20 +42,19 @@ final class FrameCodec
public static function encode(Frame $frame): string
{
$json = json_encode($frame->toArray(), JSON_UNESCAPED_SLASHES);
if ($json === false) {
throw new InvalidArgumentException('invalid_frame_payload');
}
$body = self::encrypt($json);
$body = self::encrypt(self::serializeFrame($frame));
return pack('N', strlen($body)) . $body;
}
public static function decode(string $body): Frame
{
$json = self::decrypt($body);
$data = json_decode($json, true);
$plaintext = self::decrypt($body);
if (str_starts_with($plaintext, self::BINARY_MAGIC)) {
return self::decodeBinaryFrame($plaintext);
}
$data = json_decode($plaintext, true);
if (!is_array($data) || !isset($data['type']) || !is_string($data['type'])) {
throw new InvalidArgumentException('invalid_frame');
}
@@ -76,6 +77,49 @@ final class FrameCodec
);
}
private static function serializeFrame(Frame $frame): string
{
if ($frame->type === FrameType::DATA && isset($frame->payload['data_raw']) && is_string($frame->payload['data_raw'])) {
if ($frame->sessionId === null || strlen($frame->sessionId) > 65535) {
throw new InvalidArgumentException('invalid_session_id');
}
return self::BINARY_MAGIC
. chr(self::BINARY_TYPE_DATA)
. pack('n', strlen($frame->sessionId))
. $frame->sessionId
. $frame->payload['data_raw'];
}
$json = json_encode($frame->toArray(), JSON_UNESCAPED_SLASHES);
if ($json === false) {
throw new InvalidArgumentException('invalid_frame_payload');
}
return $json;
}
private static function decodeBinaryFrame(string $plaintext): Frame
{
if (strlen($plaintext) < 7) {
throw new InvalidArgumentException('invalid_binary_frame');
}
$type = ord($plaintext[4]);
$sessionLength = unpack('n', substr($plaintext, 5, 2))[1];
if (strlen($plaintext) < 7 + $sessionLength) {
throw new InvalidArgumentException('invalid_binary_frame');
}
$sessionId = substr($plaintext, 7, $sessionLength);
$data = substr($plaintext, 7 + $sessionLength);
return match ($type) {
self::BINARY_TYPE_DATA => new Frame(FrameType::DATA, $sessionId, ['data_raw' => $data]),
default => throw new InvalidArgumentException('unsupported_binary_frame'),
};
}
private static function encrypt(string $plaintext): string
{
if (self::$encryption === self::ENCRYPTION_NONE) {
+1 -1
View File
@@ -34,7 +34,7 @@ final class FrameType
self::OPEN => 'Client Agent asks POP Server to open an authorized target stream.',
self::OPEN_OK => 'POP Server confirms target stream is open.',
self::OPEN_FAIL => 'POP Server rejects or fails target stream opening.',
self::DATA => 'Bidirectional stream bytes, base64 encoded in MVP.',
self::DATA => 'Bidirectional stream bytes; TCP stream DATA uses binary frame encoding when both ends are updated.',
self::UDP_DATA => 'Bidirectional UDP datagram relay, base64 encoded in MVP.',
self::CLOSE => 'Either side closes a stream session.',
self::ERROR => 'Explicit protocol or session error.',
+157 -55
View File
@@ -10,24 +10,23 @@ use LayLink\Auth\NodeAuthenticator;
use LayLink\Node\NodeConnection;
use LayLink\Node\NodeRegistry;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use LayLink\Protocol\FrameType;
use LayLink\Route\RouteResolver;
use LayLink\Session\SessionManager;
use LayLink\Session\TunnelSession;
use LayLink\Transport\FrameServerConnection;
use LayLink\Transport\FrameServerListenerFactory;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Connection\TcpConnection;
use Workerman\Timer;
use Workerman\Worker;
final class AgentListener
{
/** @var array<int, FrameParser> */
private array $parsers = [];
/** @var array<int, string> */
private array $connectionNodeIds = [];
/** @var array<int, array<string, AsyncTcpConnection>> */
private array $pausedTargetsByAgent = [];
public function __construct(
Worker $worker,
@@ -37,32 +36,41 @@ final class AgentListener
private readonly NodeRegistry $nodes,
private readonly SessionManager $sessions,
private readonly AuditLogger $audit,
private readonly int $maxSendBuffer = 64 * 1024 * 1024,
private readonly int $backpressureHighWatermark = 32 * 1024 * 1024,
private readonly int $dataChunkSize = 1024 * 1024,
private readonly string $agentTransportProtocol = 'tcp',
) {
$worker->onConnect = fn (TcpConnection $connection) => $this->onConnect($connection);
$worker->onMessage = fn (TcpConnection $connection, string $data) => $this->onMessage($connection, $data);
$worker->onClose = fn (TcpConnection $connection) => $this->onClose($connection);
$worker->onWorkerStart = fn () => Timer::add(10, fn () => $this->sweepHeartbeats());
}
private function onConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = 8 * 1024 * 1024;
$this->parsers[$connection->id] = new FrameParser();
}
private function onMessage(TcpConnection $connection, string $data): void
{
try {
foreach ($this->parsers[$connection->id]->push($data) as $frame) {
$this->handleFrame($connection, $frame);
(new FrameServerListenerFactory())->attach(
$this->agentTransportProtocol,
$worker,
$this->maxSendBuffer,
fn (FrameServerConnection $connection) => $this->onConnect($connection),
fn (FrameServerConnection $connection, Frame $frame) => $this->handleFrame($connection, $frame),
fn (FrameServerConnection $connection) => $this->onClose($connection),
fn (FrameServerConnection $connection) => $this->resumeTargetsForAgent($connection),
fn (FrameServerConnection $connection, \Throwable $e) => $this->onInvalidFrame($connection),
);
$transportWorkerStart = $worker->onWorkerStart;
$worker->onWorkerStart = function () use ($transportWorkerStart): void {
if ($transportWorkerStart !== null) {
$transportWorkerStart();
}
} catch (\Throwable $e) {
$this->send($connection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
$connection->close();
}
Timer::add(10, fn () => $this->sweepHeartbeats());
};
}
private function handleFrame(TcpConnection $connection, Frame $frame): void
private function onConnect(FrameServerConnection $connection): void
{
}
private function onInvalidFrame(FrameServerConnection $connection): void
{
$this->send($connection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
$connection->close();
}
private function handleFrame(FrameServerConnection $connection, Frame $frame): void
{
if ($frame->type === FrameType::AUTH) {
$result = $this->authenticator->authenticate($frame->payload);
@@ -80,7 +88,7 @@ final class AgentListener
$connection,
);
$this->nodes->register($node);
$this->connectionNodeIds[$connection->id] = $nodeId;
$this->connectionNodeIds[$connection->id()] = $nodeId;
$this->send($connection, new Frame(FrameType::AUTH_OK, null, [
'node_id' => $nodeId,
'heartbeat_interval' => 10,
@@ -89,19 +97,17 @@ final class AgentListener
return;
}
$nodeId = $this->connectionNodeIds[$connection->id] ?? null;
if (!is_string($nodeId) || $this->nodes->get($nodeId) === null) {
$nodeId = $this->connectionNodeIds[$connection->id()] ?? null;
$node = is_string($nodeId) ? $this->nodes->getByConnection($connection) : null;
if (!is_string($nodeId) || $node === null) {
$this->send($connection, new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'invalid_auth']));
$connection->close();
return;
}
$node->lastHeartbeat = time();
if ($frame->type === FrameType::PING) {
$node = $this->nodes->get($nodeId);
if ($node !== null) {
$node->lastHeartbeat = time();
$node->activeSessions = (int)($frame->payload['active_sessions'] ?? $node->activeSessions);
}
$node->activeSessions = (int)($frame->payload['active_sessions'] ?? $node->activeSessions);
$this->send($connection, new Frame(FrameType::PONG, null, ['timestamp' => time()]));
return;
}
@@ -123,13 +129,13 @@ final class AgentListener
}
match ($frame->type) {
FrameType::DATA => $this->forwardDataToTarget($session, (string)($frame->payload['data'] ?? '')),
FrameType::DATA => $this->forwardDataToTarget($session, $frame),
FrameType::CLOSE => $this->closeSession($session, 'closed', null),
default => null,
};
}
private function forwardUdpDatagram(TcpConnection $agentConnection, string $nodeId, Frame $frame): void
private function forwardUdpDatagram(FrameServerConnection $agentConnection, string $nodeId, Frame $frame): void
{
if ($frame->sessionId === null) {
$this->send($agentConnection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
@@ -173,7 +179,7 @@ final class AgentListener
$target->connect();
}
private function openTargetForAgent(TcpConnection $agentConnection, string $nodeId, Frame $frame): void
private function openTargetForAgent(FrameServerConnection $agentConnection, string $nodeId, Frame $frame): void
{
if ($frame->sessionId === null || $this->sessions->get($frame->sessionId) !== null) {
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $frame->sessionId, ['reason' => 'invalid_frame']));
@@ -204,6 +210,12 @@ final class AgentListener
return;
}
$targetAddress = $this->resolveTargetAddress($host, $port);
if ($targetAddress === null) {
$this->rejectOpen($agentConnection, $frame, 'dns_resolution_failed', (string)$auth['user_id'], $nodeId, $decision->policyId);
return;
}
$session = new TunnelSession(
$frame->sessionId,
(string)$auth['user_id'],
@@ -219,47 +231,80 @@ final class AgentListener
$session->state = TunnelSession::OPENING;
$this->sessions->add($session);
$target = new AsyncTcpConnection("tcp://{$host}:{$port}");
$target->maxSendBufferSize = 8 * 1024 * 1024;
$target = new AsyncTcpConnection($targetAddress);
$target->maxSendBufferSize = $this->maxSendBuffer;
$session->target = $target;
$target->onConnect = function () use ($session, $agentConnection): void {
$session->state = TunnelSession::OPEN;
$this->send($agentConnection, new Frame(FrameType::OPEN_OK, $session->sessionId, [
$sent = $this->send($agentConnection, new Frame(FrameType::OPEN_OK, $session->sessionId, [
'target_host' => $session->targetHost,
'target_port' => $session->targetPort,
]));
if ($sent === false) {
$this->closeSession($session, 'failed', 'agent_buffer_overflow');
}
};
$target->onMessage = function (AsyncTcpConnection $target, string $data) use ($session, $agentConnection): void {
$session->bytesTargetToClient += strlen($data);
$this->send($agentConnection, new Frame(FrameType::DATA, $session->sessionId, [
'data' => base64_encode($data),
]));
if (!$this->sendData($agentConnection, $session->sessionId, $data)) {
$this->closeSession($session, 'failed', 'agent_buffer_overflow');
return;
}
if ($agentConnection->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
$target->pauseRecv();
$this->pausedTargetsByAgent[$agentConnection->id()][$session->sessionId] = $target;
}
};
$target->onClose = fn () => $this->closeSession($session, 'closed', null);
$target->onError = fn () => $this->failOpenSession($agentConnection, $session, 'target_connection_refused');
$target->connect();
}
private function forwardDataToTarget(TunnelSession $session, string $encoded): void
private function forwardDataToTarget(TunnelSession $session, Frame $frame): void
{
$data = base64_decode($encoded, true);
$data = $this->frameData($frame);
if ($data === false) {
$this->closeSession($session, 'failed', 'invalid_frame');
return;
}
$session->bytesClientToTarget += strlen($data);
$session->target?->send($data);
if ($session->target === null) {
return;
}
$sent = $session->target->send($data);
if ($sent === false) {
$this->closeSession($session, 'failed', 'target_buffer_overflow');
return;
}
if ($session->target->getSendBufferQueueSize() >= $this->backpressureHighWatermark && $session->agent !== null) {
$session->agent->pauseRecv();
$session->target->onBufferDrain = function (AsyncTcpConnection $target) use ($session): void {
$session->agent?->resumeRecv();
};
}
}
private function failOpenSession(TcpConnection $agentConnection, TunnelSession $session, string $reason): void
private function failOpenSession(FrameServerConnection $agentConnection, TunnelSession $session, string $reason): void
{
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $session->sessionId, ['reason' => $reason]));
if ($session->state === TunnelSession::OPENING) {
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $session->sessionId, ['reason' => $reason]));
}
$this->closeSession($session, 'failed', $reason);
}
private function rejectOpen(TcpConnection $agentConnection, Frame $frame, string $reason, string $userId, string $nodeId, ?string $policyId = null): void
private function frameData(Frame $frame): string|false
{
if (isset($frame->payload['data_raw']) && is_string($frame->payload['data_raw'])) {
return $frame->payload['data_raw'];
}
return base64_decode((string)($frame->payload['data'] ?? ''), true);
}
private function rejectOpen(FrameServerConnection $agentConnection, Frame $frame, string $reason, string $userId, string $nodeId, ?string $policyId = null): void
{
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $frame->sessionId, ['reason' => $reason]));
$this->audit->write([
@@ -289,6 +334,9 @@ final class AgentListener
$session->state = TunnelSession::CLOSED;
$this->sessions->remove($session->sessionId);
if ($session->agent !== null) {
unset($this->pausedTargetsByAgent[$session->agent->id()][$session->sessionId]);
}
if ($session->agent !== null) {
$this->send($session->agent, new Frame(FrameType::CLOSE, $session->sessionId, ['reason' => $reason ?? $result]));
@@ -316,10 +364,10 @@ final class AgentListener
]);
}
private function onClose(TcpConnection $connection): void
private function onClose(FrameServerConnection $connection): void
{
unset($this->parsers[$connection->id]);
unset($this->connectionNodeIds[$connection->id]);
unset($this->connectionNodeIds[$connection->id()]);
unset($this->pausedTargetsByAgent[$connection->id()]);
$node = $this->nodes->unregisterByConnection($connection);
foreach ($this->sessions->all() as $session) {
if ($session->agent === $connection) {
@@ -342,8 +390,62 @@ final class AgentListener
}
}
private function send(TcpConnection $connection, Frame $frame): void
private function send(FrameServerConnection $connection, Frame $frame): bool|null
{
$connection->send(FrameCodec::encode($frame));
return $connection->send($frame);
}
private function sendData(FrameServerConnection $connection, string $sessionId, string $data): bool
{
$length = strlen($data);
for ($offset = 0; $offset < $length; $offset += $this->dataChunkSize) {
$sent = $this->send($connection, new Frame(FrameType::DATA, $sessionId, [
'data_raw' => substr($data, $offset, $this->dataChunkSize),
]));
if ($sent === false) {
return false;
}
}
return true;
}
private function resumeTargetsForAgent(FrameServerConnection $connection): void
{
foreach ($this->pausedTargetsByAgent[$connection->id()] ?? [] as $sessionId => $target) {
$target->resumeRecv();
unset($this->pausedTargetsByAgent[$connection->id()][$sessionId]);
}
if (($this->pausedTargetsByAgent[$connection->id()] ?? []) === []) {
unset($this->pausedTargetsByAgent[$connection->id()]);
}
}
private function resolveTargetAddress(string $host, int $port): ?string
{
if (filter_var($host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV4) !== false) {
return "tcp://{$host}:{$port}";
}
if (filter_var($host, FILTER_VALIDATE_IP, FILTER_FLAG_IPV6) !== false) {
return "tcp://[{$host}]:{$port}";
}
$records = @dns_get_record($host, DNS_A | DNS_AAAA);
if ($records === false || $records === []) {
return null;
}
foreach ($records as $record) {
if (isset($record['ip']) && is_string($record['ip'])) {
return "tcp://{$record['ip']}:{$port}";
}
}
foreach ($records as $record) {
if (isset($record['ipv6']) && is_string($record['ipv6'])) {
return "tcp://[{$record['ipv6']}]:{$port}";
}
}
return null;
}
}
+29 -12
View File
@@ -25,6 +25,11 @@ final class PopServer
private readonly array $policies,
private readonly array $allowedAgentTransports,
string $auditLog,
private readonly int $maxSendBuffer = 64 * 1024 * 1024,
private readonly int $backpressureHighWatermark = 32 * 1024 * 1024,
private readonly int $dataChunkSize = 1024 * 1024,
private readonly int $tcpWorkerCount = 1,
private readonly int $kcpWorkerCount = 1,
) {
$this->nodes = new NodeRegistry();
$this->sessions = new SessionManager();
@@ -33,17 +38,29 @@ final class PopServer
public function boot(): void
{
$agentWorker = new Worker('tcp://' . $this->agentListen);
$agentWorker->name = 'laylink-pop-agent-listener';
$agentWorker->count = 1;
new AgentListener(
$agentWorker,
new NodeAuthenticator($this->nodeConfig, $this->allowedAgentTransports),
new ClientAuthenticator(),
new RouteResolver(new PolicyChecker($this->policies), $this->nodes),
$this->nodes,
$this->sessions,
$this->audit,
);
$implementedTransports = array_values(array_intersect($this->allowedAgentTransports, ['tcp', 'kcp']));
if ($implementedTransports === []) {
throw new \RuntimeException('no_implemented_pop_transport_enabled');
}
foreach ($implementedTransports as $transport) {
$scheme = $transport === 'kcp' ? 'udp' : 'tcp';
$agentWorker = new Worker($scheme . '://' . $this->agentListen);
$agentWorker->name = 'laylink-pop-agent-listener-' . $transport;
$agentWorker->count = $transport === 'kcp' ? $this->kcpWorkerCount : $this->tcpWorkerCount;
new AgentListener(
$agentWorker,
new NodeAuthenticator($this->nodeConfig, $this->allowedAgentTransports),
new ClientAuthenticator(),
new RouteResolver(new PolicyChecker($this->policies), $this->nodes),
$this->nodes,
$this->sessions,
$this->audit,
$this->maxSendBuffer,
$this->backpressureHighWatermark,
$this->dataChunkSize,
$transport,
);
}
}
}
+2 -1
View File
@@ -4,6 +4,7 @@ declare(strict_types=1);
namespace LayLink\Session;
use LayLink\Transport\FrameServerConnection;
use Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection;
@@ -18,7 +19,7 @@ final class TunnelSession
public string $state = self::NEW;
public ?TcpConnection $client = null;
public ?TcpConnection $agent = null;
public ?FrameServerConnection $agent = null;
public ?AsyncTcpConnection $target = null;
public ?string $nodeId = null;
public string $startTime;
+22
View File
@@ -0,0 +1,22 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
interface FrameClientTransport
{
public function connect(): void;
public function send(Frame $frame): bool|null;
public function close(): void;
public function pauseRecv(): void;
public function resumeRecv(): void;
public function getSendBufferQueueSize(): int;
}
@@ -0,0 +1,44 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use InvalidArgumentException;
final class FrameClientTransportFactory
{
public function create(
string $protocol,
string $address,
int $maxSendBuffer,
\Closure $onConnect,
\Closure $onFrame,
\Closure $onClose,
\Closure $onBufferDrain,
\Closure $onInvalidFrame,
): FrameClientTransport {
return match (strtolower(trim($protocol))) {
'tcp' => new TcpFrameClientTransport(
$address,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'kcp' => new KcpFrameClientTransport(
$address,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'udp' => throw new InvalidArgumentException('agent_transport_not_implemented'),
default => throw new InvalidArgumentException('unsupported_agent_transport'),
};
}
}
+24
View File
@@ -0,0 +1,24 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
interface FrameServerConnection
{
public function id(): int;
public function send(Frame $frame): bool|null;
public function close(): void;
public function pauseRecv(): void;
public function resumeRecv(): void;
public function getSendBufferQueueSize(): int;
public function getRemoteIp(): string;
}
@@ -0,0 +1,45 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use InvalidArgumentException;
use Workerman\Worker;
final class FrameServerListenerFactory
{
public function attach(
string $protocol,
Worker $worker,
int $maxSendBuffer,
\Closure $onConnect,
\Closure $onFrame,
\Closure $onClose,
\Closure $onBufferDrain,
\Closure $onInvalidFrame,
): void {
match (strtolower(trim($protocol))) {
'tcp' => new TcpFrameServerListener(
$worker,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'kcp' => new KcpFrameServerListener(
$worker,
$maxSendBuffer,
$onConnect,
$onFrame,
$onClose,
$onBufferDrain,
$onInvalidFrame,
),
'udp' => throw new InvalidArgumentException('pop_transport_not_implemented'),
default => throw new InvalidArgumentException('unsupported_pop_transport'),
};
}
}
+23
View File
@@ -0,0 +1,23 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
final class KcpConfig
{
public static function int(string $key, int $default, int $min, int $max): int
{
$value = getenv($key);
if ($value === false || trim($value) === '' || !preg_match('/^-?\d+$/', trim($value))) {
return $default;
}
return max($min, min($max, (int)trim($value)));
}
public static function tickSeconds(): float
{
return self::int('LAYLINK_KCP_TICK_MS', 10, 1, 1000) / 1000;
}
}
+185
View File
@@ -0,0 +1,185 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Timer;
final class KcpFrameClientTransport implements FrameClientTransport
{
private ?AsyncUdpConnection $connection = null;
private KcpReliableSession|NativeKcpSession|null $session = null;
private ?KcpUdpPacketSender $sender = null;
private ?int $timerId = null;
private int $conv;
private bool $connected = false;
public function __construct(
private readonly string $address,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$this->conv = random_int(1, 0x7fffffff);
}
public function connect(): void
{
$this->connected = false;
$this->session = null;
$connection = new AsyncUdpConnection($this->normalizeAddress($this->address));
$this->connection = $connection;
$this->sender = new KcpUdpPacketSender(
$connection,
KcpConfig::int('LAYLINK_KCP_UDP_SEND_QUEUE_BYTES', 64 * 1024 * 1024, 1024 * 1024, 512 * 1024 * 1024),
);
$connection->onConnect = function () use ($connection): void {
$this->sender?->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN,
'conv' => $this->conv,
]));
};
$connection->onMessage = function (AsyncUdpConnection $connection, string $data): void {
$this->handlePacket($data);
};
$connection->onClose = function () use ($connection): void {
if ($this->connection === $connection) {
$this->connection = null;
$this->sender = null;
}
$this->stopTimer();
($this->onClose)($this);
};
$connection->connect();
$this->timerId = Timer::add(KcpConfig::tickSeconds(), fn () => $this->tick());
}
public function send(Frame $frame): bool|null
{
return $this->session?->sendFrame($frame, $this->maxSendBuffer) ?? false;
}
public function close(): void
{
$this->session?->close();
$this->connection?->close();
}
public function pauseRecv(): void
{
$this->session?->pause();
}
public function resumeRecv(): void
{
$this->session?->resume();
}
public function getSendBufferQueueSize(): int
{
return ($this->session?->getSendBufferQueueSize() ?? 0) + ($this->sender?->queuedBytes() ?? 0);
}
private function handlePacket(string $data): void
{
$packet = KcpPacketCodec::decode($data);
if ($packet === null) {
$this->session?->receive($data);
return;
}
if ($packet['conv'] !== $this->conv) {
return;
}
if ($packet['type'] === KcpPacketCodec::SYN_ACK && !$this->connected) {
$this->connected = true;
$this->session = $this->createSession();
($this->onConnect)($this);
return;
}
if ($packet['type'] === KcpPacketCodec::CLOSE) {
$this->connection?->close();
return;
}
$this->session?->receive($packet);
if ($this->session?->isClosed()) {
$this->connection?->close();
}
}
private function tick(): void
{
if ($this->connection === null) {
$this->stopTimer();
return;
}
if (!$this->connected) {
$this->sender?->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN,
'conv' => $this->conv,
]));
$this->sender?->flush(KcpConfig::int('LAYLINK_KCP_UDP_FLUSH_PACKETS', 1024, 1, 8192));
return;
}
$before = $this->getSendBufferQueueSize();
$this->sender?->flush(KcpConfig::int('LAYLINK_KCP_UDP_FLUSH_PACKETS', 1024, 1, 8192));
$this->session?->tick();
$this->sender?->flush(KcpConfig::int('LAYLINK_KCP_UDP_FLUSH_PACKETS', 1024, 1, 8192));
if ($before >= $this->maxSendBuffer && $this->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($this);
}
}
private function createSession(): KcpReliableSession|NativeKcpSession
{
$backend = strtolower(trim((string)(getenv('LAYLINK_KCP_BACKEND') ?: 'ffi')));
$libraryPath = (string)(getenv('LAYLINK_KCP_FFI_LIB') ?: dirname(__DIR__, 2) . '/native/kcp/liblaylink_kcp.so');
$args = [
$this->conv,
fn (string $packet): bool|null => $this->sender?->send($packet) ?? false,
fn (Frame $frame) => ($this->onFrame)($this, $frame),
fn (\Throwable $e) => ($this->onInvalidFrame)($e),
];
if ($backend === 'php') {
return new KcpReliableSession(...$args);
}
return new NativeKcpSession(...[...$args, $libraryPath]);
}
private function stopTimer(): void
{
if ($this->timerId === null) {
return;
}
Timer::del($this->timerId);
$this->timerId = null;
}
private function normalizeAddress(string $address): string
{
if (str_starts_with($address, 'udp://')) {
return $address;
}
if (str_starts_with($address, 'tcp://')) {
return 'udp://' . substr($address, 6);
}
return 'udp://' . $address;
}
}
@@ -0,0 +1,87 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use Workerman\Connection\UdpConnection;
final class KcpFrameServerConnection implements FrameServerConnection
{
private bool $closed = false;
public function __construct(
private readonly int $id,
private UdpConnection $connection,
private readonly KcpReliableSession|NativeKcpSession $session,
private readonly KcpUdpPacketSender $sender,
private readonly int $maxSendBuffer,
) {
}
public function id(): int
{
return $this->id;
}
public function send(Frame $frame): bool|null
{
if ($this->closed) {
return false;
}
return $this->session->sendFrame($frame, $this->maxSendBuffer);
}
public function close(): void
{
if ($this->closed) {
return;
}
$this->closed = true;
$this->session->close();
}
public function pauseRecv(): void
{
$this->session->pause();
}
public function resumeRecv(): void
{
$this->session->resume();
}
public function getSendBufferQueueSize(): int
{
return $this->session->getSendBufferQueueSize() + $this->sender->queuedBytes();
}
public function getRemoteIp(): string
{
return $this->connection->getRemoteIp();
}
public function updateConnection(UdpConnection $connection): void
{
$this->connection = $connection;
$this->sender->updateConnection($connection);
}
public function isClosed(): bool
{
return $this->closed || $this->session->isClosed();
}
public function flushUdp(int $maxPackets): void
{
$this->sender->flush($maxPackets);
}
public function sendUdpPacket(string $packet): bool
{
return $this->sender->send($packet);
}
}
+239
View File
@@ -0,0 +1,239 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use Workerman\Connection\UdpConnection;
use Workerman\Timer;
use Workerman\Worker;
final class KcpFrameServerListener
{
/** @var array<string, KcpFrameServerConnection> */
private array $connections = [];
/** @var array<string, KcpReliableSession|NativeKcpSession> */
private array $sessions = [];
/** @var array<int, string> */
private array $keysByConv = [];
private ?int $timerId = null;
private int $nextConnectionId = 1_000_000;
public function __construct(
Worker $worker,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$worker->onMessage = fn (UdpConnection $connection, string $data) => $this->handleMessage($connection, $data);
$worker->onWorkerStart = function (): void {
$this->timerId = Timer::add(KcpConfig::tickSeconds(), fn () => $this->tick());
};
$worker->onWorkerStop = function (): void {
if ($this->timerId !== null) {
Timer::del($this->timerId);
$this->timerId = null;
}
};
}
private function handleMessage(UdpConnection $connection, string $data): void
{
$packet = KcpPacketCodec::decode($data);
if ($packet === null) {
$conv = KcpPacketCodec::rawKcpConv($data);
if ($conv === null) {
return;
}
$key = $this->key($connection, $conv);
$key = $this->knownKey($connection, $conv, $key);
$wrapped = $this->connections[$key] ?? null;
$session = $this->sessions[$key] ?? null;
if ($wrapped === null || $session === null) {
return;
}
$wrapped->updateConnection($connection);
$wasFull = $session->getSendBufferQueueSize() >= $this->maxSendBuffer;
$session->receive($data);
if ($session->isClosed()) {
$this->closeConnection($key);
return;
}
if ($wasFull && $session->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($wrapped);
}
return;
}
$key = $this->key($connection, $packet['conv']);
if ($packet['type'] === KcpPacketCodec::SYN) {
$this->handleSyn($connection, $key, $packet['conv']);
return;
}
$key = $this->knownKey($connection, $packet['conv'], $key);
$wrapped = $this->connections[$key] ?? null;
$session = $this->sessions[$key] ?? null;
if ($wrapped === null || $session === null) {
return;
}
if ($packet['type'] === KcpPacketCodec::CLOSE) {
$this->closeConnection($key);
return;
}
$wrapped->updateConnection($connection);
$wasFull = $session->getSendBufferQueueSize() >= $this->maxSendBuffer;
$session->receive($packet);
if ($session->isClosed()) {
$this->closeConnection($key);
return;
}
if ($wasFull && $session->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($wrapped);
}
}
private function handleSyn(UdpConnection $connection, string $key, int $conv): void
{
if (isset($this->keysByConv[$conv]) && $this->keysByConv[$conv] !== $key) {
$key = $this->migrateConnection($this->keysByConv[$conv], $key, $connection, $conv);
}
if (isset($this->connections[$key])) {
$this->connections[$key]->updateConnection($connection);
$this->connections[$key]->sendUdpPacket(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN_ACK,
'conv' => $conv,
]));
$this->connections[$key]->flushUdp(KcpConfig::int('LAYLINK_KCP_UDP_FLUSH_PACKETS', 1024, 1, 8192));
return;
}
$sender = new KcpUdpPacketSender(
$connection,
KcpConfig::int('LAYLINK_KCP_UDP_SEND_QUEUE_BYTES', 64 * 1024 * 1024, 1024 * 1024, 512 * 1024 * 1024),
);
$session = $this->createSession(
$conv,
fn (string $packet): bool|null => $sender->send($packet),
function (Frame $frame) use ($conv): void {
$wrapped = $this->connectionForConv($conv);
if ($wrapped !== null) {
($this->onFrame)($wrapped, $frame);
}
},
function (\Throwable $e) use ($conv): void {
$wrapped = $this->connectionForConv($conv);
if ($wrapped !== null) {
($this->onInvalidFrame)($wrapped, $e);
}
},
);
$wrapped = new KcpFrameServerConnection($this->nextConnectionId++, $connection, $session, $sender, $this->maxSendBuffer);
$this->sessions[$key] = $session;
$this->connections[$key] = $wrapped;
$this->keysByConv[$conv] = $key;
$sender->send(KcpPacketCodec::encode([
'type' => KcpPacketCodec::SYN_ACK,
'conv' => $conv,
]));
($this->onConnect)($wrapped);
}
private function tick(): void
{
foreach ($this->sessions as $key => $session) {
$wrapped = $this->connections[$key] ?? null;
if ($wrapped === null || $wrapped->isClosed()) {
$this->closeConnection($key);
continue;
}
$before = $session->getSendBufferQueueSize();
$wrapped->flushUdp(KcpConfig::int('LAYLINK_KCP_UDP_FLUSH_PACKETS', 1024, 1, 8192));
$session->tick();
$wrapped->flushUdp(KcpConfig::int('LAYLINK_KCP_UDP_FLUSH_PACKETS', 1024, 1, 8192));
if ($before >= $this->maxSendBuffer && $session->getSendBufferQueueSize() < $this->maxSendBuffer) {
($this->onBufferDrain)($wrapped);
}
}
}
private function closeConnection(string $key): void
{
$wrapped = $this->connections[$key] ?? null;
$session = $this->sessions[$key] ?? null;
unset($this->connections[$key], $this->sessions[$key]);
if ($session !== null) {
unset($this->keysByConv[$session->conv()]);
}
if ($wrapped !== null) {
($this->onClose)($wrapped);
}
}
private function createSession(int $conv, \Closure $sendPacket, \Closure $onFrame, \Closure $onInvalidFrame): KcpReliableSession|NativeKcpSession
{
$backend = strtolower(trim((string)(getenv('LAYLINK_KCP_BACKEND') ?: 'ffi')));
if ($backend === 'php') {
return new KcpReliableSession($conv, $sendPacket, $onFrame, $onInvalidFrame);
}
$libraryPath = (string)(getenv('LAYLINK_KCP_FFI_LIB') ?: dirname(__DIR__, 2) . '/native/kcp/liblaylink_kcp.so');
return new NativeKcpSession($conv, $sendPacket, $onFrame, $onInvalidFrame, $libraryPath);
}
private function key(UdpConnection $connection, int $conv): string
{
return $connection->getRemoteAddress() . '#' . $conv;
}
private function knownKey(UdpConnection $connection, int $conv, string $candidate): string
{
if (isset($this->connections[$candidate])) {
return $candidate;
}
$known = $this->keysByConv[$conv] ?? null;
if ($known === null || !isset($this->connections[$known], $this->sessions[$known])) {
return $candidate;
}
return $this->migrateConnection($known, $candidate, $connection, $conv);
}
private function migrateConnection(string $oldKey, string $newKey, UdpConnection $connection, int $conv): string
{
if ($oldKey === $newKey) {
return $newKey;
}
$wrapped = $this->connections[$oldKey] ?? null;
$session = $this->sessions[$oldKey] ?? null;
if ($wrapped === null || $session === null) {
return $newKey;
}
unset($this->connections[$oldKey], $this->sessions[$oldKey]);
$wrapped->updateConnection($connection);
$this->connections[$newKey] = $wrapped;
$this->sessions[$newKey] = $session;
$this->keysByConv[$conv] = $newKey;
return $newKey;
}
private function connectionForConv(int $conv): ?KcpFrameServerConnection
{
$key = $this->keysByConv[$conv] ?? null;
return $key === null ? null : ($this->connections[$key] ?? null);
}
}
+71
View File
@@ -0,0 +1,71 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
final class KcpPacketCodec
{
public const MAGIC = "LLK1";
public const SYN = 1;
public const SYN_ACK = 2;
public const DATA = 3;
public const ACK = 4;
public const CLOSE = 5;
private const HEADER_LENGTH = 25;
/**
* @param array{type:int, conv:int, seq?:int, ack?:int, message_id?:int, fragment_index?:int, fragment_count?:int, payload?:string} $packet
*/
public static function encode(array $packet): string
{
return pack(
'a4CNNNNnn',
self::MAGIC,
$packet['type'],
$packet['conv'],
$packet['seq'] ?? 0,
$packet['ack'] ?? 0,
$packet['message_id'] ?? 0,
$packet['fragment_index'] ?? 0,
$packet['fragment_count'] ?? 0,
) . ($packet['payload'] ?? '');
}
/**
* @return array{type:int, conv:int, seq:int, ack:int, message_id:int, fragment_index:int, fragment_count:int, payload:string}|null
*/
public static function decode(string $bytes): ?array
{
if (strlen($bytes) < self::HEADER_LENGTH || substr($bytes, 0, 4) !== self::MAGIC) {
return null;
}
$header = unpack('a4magic/Ctype/Nconv/Nseq/Nack/Nmessage_id/nfragment_index/nfragment_count', substr($bytes, 0, self::HEADER_LENGTH));
if ($header === false) {
return null;
}
return [
'type' => (int)$header['type'],
'conv' => (int)$header['conv'],
'seq' => (int)$header['seq'],
'ack' => (int)$header['ack'],
'message_id' => (int)$header['message_id'],
'fragment_index' => (int)$header['fragment_index'],
'fragment_count' => (int)$header['fragment_count'],
'payload' => substr($bytes, self::HEADER_LENGTH),
];
}
public static function rawKcpConv(string $bytes): ?int
{
if (strlen($bytes) < 4 || substr($bytes, 0, 4) === self::MAGIC) {
return null;
}
$decoded = unpack('Vconv', substr($bytes, 0, 4));
return $decoded === false ? null : (int)$decoded['conv'];
}
}
+235
View File
@@ -0,0 +1,235 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
final class KcpReliableSession
{
private const MTU = 1200;
private const RESEND_AFTER_MS = 120;
private FrameParser $parser;
private int $nextSeq = 1;
private int $expectedSeq = 1;
private int $nextMessageId = 1;
/** @var array<int, array{packet:string, sent_at:float, bytes:int}> */
private array $unacked = [];
/** @var array<int, array{message_id:int, fragment_index:int, fragment_count:int, payload:string}> */
private array $pendingSegments = [];
/** @var array<int, array<int, string>> */
private array $messageFragments = [];
/** @var array<int, int> */
private array $messageFragmentCounts = [];
/** @var Frame[] */
private array $pausedFrames = [];
private bool $paused = false;
private bool $closed = false;
public function __construct(
private readonly int $conv,
private readonly \Closure $sendPacket,
private readonly \Closure $onFrame,
private readonly \Closure $onInvalidFrame,
) {
$this->parser = new FrameParser();
}
public function conv(): int
{
return $this->conv;
}
public function sendFrame(Frame $frame, int $maxSendBuffer): bool|null
{
if ($this->closed || $this->getSendBufferQueueSize() >= $maxSendBuffer) {
return false;
}
$bytes = FrameCodec::encode($frame);
$payloadSize = self::MTU - 25;
$fragments = str_split($bytes, $payloadSize);
$messageId = $this->nextMessageId++;
$fragmentCount = count($fragments);
foreach ($fragments as $fragmentIndex => $payload) {
$seq = $this->nextSeq++;
$packet = KcpPacketCodec::encode([
'type' => KcpPacketCodec::DATA,
'conv' => $this->conv,
'seq' => $seq,
'message_id' => $messageId,
'fragment_index' => $fragmentIndex,
'fragment_count' => $fragmentCount,
'payload' => $payload,
]);
$this->unacked[$seq] = [
'packet' => $packet,
'sent_at' => $this->nowMs(),
'bytes' => strlen($packet),
];
($this->sendPacket)($packet);
}
return true;
}
/**
* @param array{type:int, conv:int, seq:int, ack:int, message_id:int, fragment_index:int, fragment_count:int, payload:string}|string $packet
*/
public function receive(array|string $packet): void
{
if ($this->closed || is_string($packet)) {
return;
}
if ($packet['type'] === KcpPacketCodec::ACK) {
unset($this->unacked[$packet['ack']]);
return;
}
if ($packet['type'] === KcpPacketCodec::CLOSE) {
$this->closed = true;
return;
}
if ($packet['type'] !== KcpPacketCodec::DATA) {
return;
}
$this->ack($packet['seq']);
if ($packet['seq'] < $this->expectedSeq) {
return;
}
$this->pendingSegments[$packet['seq']] = [
'message_id' => $packet['message_id'],
'fragment_index' => $packet['fragment_index'],
'fragment_count' => $packet['fragment_count'],
'payload' => $packet['payload'],
];
$this->drainOrderedSegments();
}
public function tick(): void
{
if ($this->closed) {
return;
}
$now = $this->nowMs();
foreach ($this->unacked as $seq => $pending) {
if ($now - $pending['sent_at'] < self::RESEND_AFTER_MS) {
continue;
}
$this->unacked[$seq]['sent_at'] = $now;
($this->sendPacket)($pending['packet']);
}
}
public function pause(): void
{
$this->paused = true;
}
public function resume(): void
{
$this->paused = false;
while (!$this->paused && $this->pausedFrames !== []) {
$frame = array_shift($this->pausedFrames);
($this->onFrame)($frame);
}
}
public function close(): void
{
if ($this->closed) {
return;
}
$this->closed = true;
($this->sendPacket)(KcpPacketCodec::encode([
'type' => KcpPacketCodec::CLOSE,
'conv' => $this->conv,
]));
}
public function isClosed(): bool
{
return $this->closed;
}
public function getSendBufferQueueSize(): int
{
$bytes = 0;
foreach ($this->unacked as $pending) {
$bytes += $pending['bytes'];
}
return $bytes;
}
/**
* @param array{message_id:int, fragment_index:int, fragment_count:int, payload:string} $segment
*/
private function receiveOrderedSegment(array $segment): void
{
$messageId = $segment['message_id'];
$fragmentIndex = $segment['fragment_index'];
$fragmentCount = $segment['fragment_count'];
if ($fragmentCount < 1 || $fragmentIndex < 0 || $fragmentIndex >= $fragmentCount) {
return;
}
$this->messageFragmentCounts[$messageId] = $fragmentCount;
$this->messageFragments[$messageId][$fragmentIndex] = $segment['payload'];
if (count($this->messageFragments[$messageId]) !== $fragmentCount) {
return;
}
ksort($this->messageFragments[$messageId]);
$bytes = implode('', $this->messageFragments[$messageId]);
unset($this->messageFragments[$messageId], $this->messageFragmentCounts[$messageId]);
try {
foreach ($this->parser->push($bytes) as $frame) {
if ($this->paused) {
$this->pausedFrames[] = $frame;
continue;
}
($this->onFrame)($frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($e);
}
}
private function drainOrderedSegments(): void
{
while (isset($this->pendingSegments[$this->expectedSeq])) {
$segment = $this->pendingSegments[$this->expectedSeq];
unset($this->pendingSegments[$this->expectedSeq]);
$this->expectedSeq++;
$this->receiveOrderedSegment($segment);
}
}
private function ack(int $seq): void
{
($this->sendPacket)(KcpPacketCodec::encode([
'type' => KcpPacketCodec::ACK,
'conv' => $this->conv,
'ack' => $seq,
]));
}
private function nowMs(): float
{
return microtime(true) * 1000;
}
}
+89
View File
@@ -0,0 +1,89 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use Workerman\Connection\AsyncUdpConnection;
use Workerman\Connection\UdpConnection;
final class KcpUdpPacketSender
{
/** @var string[] */
private array $queue = [];
private int $queuedBytes = 0;
public function __construct(
private UdpConnection $connection,
private readonly int $maxQueueBytes,
) {
}
public function updateConnection(UdpConnection $connection): void
{
$this->connection = $connection;
}
public function send(string $packet): bool
{
if ($this->queue !== []) {
return $this->enqueue($packet);
}
if ($this->trySend($packet)) {
return true;
}
return $this->enqueue($packet);
}
public function flush(int $maxPackets): void
{
$sent = 0;
while ($this->queue !== [] && $sent < $maxPackets) {
$packet = $this->queue[0];
if (!$this->trySend($packet)) {
return;
}
array_shift($this->queue);
$this->queuedBytes -= strlen($packet);
$sent++;
}
}
public function queuedBytes(): int
{
return $this->queuedBytes;
}
private function enqueue(string $packet): bool
{
$length = strlen($packet);
if ($this->queuedBytes + $length > $this->maxQueueBytes) {
return false;
}
$this->queue[] = $packet;
$this->queuedBytes += $length;
return true;
}
private function trySend(string $packet): bool
{
$socket = $this->connection->getSocket();
if (!is_resource($socket)) {
return false;
}
if ($this->connection instanceof AsyncUdpConnection) {
return @stream_socket_sendto($socket, $packet) === strlen($packet);
}
$remote = $this->connection->isIpV6()
? '[' . $this->connection->getRemoteIp() . ']:' . $this->connection->getRemotePort()
: $this->connection->getRemoteAddress();
return @stream_socket_sendto($socket, $packet, 0, $remote) === strlen($packet);
}
}
+53
View File
@@ -0,0 +1,53 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use FFI;
final class NativeKcpLibrary
{
private const CDEF = <<<'CDEF'
typedef struct laylink_kcp laylink_kcp;
laylink_kcp* laylink_kcp_create(unsigned int conv);
void laylink_kcp_release(laylink_kcp* session);
int laylink_kcp_nodelay(laylink_kcp* session, int nodelay, int interval, int resend, int nc);
int laylink_kcp_wndsize(laylink_kcp* session, int sndwnd, int rcvwnd);
int laylink_kcp_setmtu(laylink_kcp* session, int mtu);
int laylink_kcp_send(laylink_kcp* session, const char* buffer, int len);
int laylink_kcp_input(laylink_kcp* session, const char* buffer, long size);
void laylink_kcp_update(laylink_kcp* session, unsigned int current);
unsigned int laylink_kcp_check(laylink_kcp* session, unsigned int current);
int laylink_kcp_peeksize(laylink_kcp* session);
int laylink_kcp_recv(laylink_kcp* session, char* buffer, int len);
void laylink_kcp_flush(laylink_kcp* session);
int laylink_kcp_waitsnd(laylink_kcp* session);
int laylink_kcp_pending_output_size(laylink_kcp* session);
int laylink_kcp_pop_output(laylink_kcp* session, char* buffer, int len);
CDEF;
private static ?FFI $ffi = null;
public static function load(?string $libraryPath = null): FFI
{
if (self::$ffi !== null) {
return self::$ffi;
}
if (!class_exists(FFI::class)) {
throw new \RuntimeException('ffi_extension_not_loaded');
}
$libraryPath ??= dirname(__DIR__, 2) . '/native/kcp/liblaylink_kcp.so';
if (!str_starts_with($libraryPath, '/')) {
$libraryPath = dirname(__DIR__, 2) . '/' . $libraryPath;
}
if (!is_file($libraryPath)) {
throw new \RuntimeException('kcp_ffi_library_not_found');
}
self::$ffi = FFI::cdef(self::CDEF, $libraryPath);
return self::$ffi;
}
}
+211
View File
@@ -0,0 +1,211 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use FFI;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
final class NativeKcpSession
{
private const RECEIVE_BUFFER_BYTES = 8 * 1024 * 1024;
private \FFI $ffi;
private mixed $session;
private FrameParser $parser;
private bool $paused = false;
private bool $closed = false;
private int $mtuBytes;
/** @var Frame[] */
private array $pausedFrames = [];
public function __construct(
private readonly int $conv,
private readonly \Closure $sendPacket,
private readonly \Closure $onFrame,
private readonly \Closure $onInvalidFrame,
private readonly ?string $libraryPath = null,
) {
$this->ffi = NativeKcpLibrary::load($this->libraryPath);
$this->session = $this->ffi->laylink_kcp_create($conv);
if ($this->session === null) {
throw new \RuntimeException('kcp_create_failed');
}
$this->ffi->laylink_kcp_nodelay(
$this->session,
KcpConfig::int('LAYLINK_KCP_NODELAY', 1, 0, 1),
KcpConfig::int('LAYLINK_KCP_INTERVAL_MS', 10, 1, 1000),
KcpConfig::int('LAYLINK_KCP_FAST_RESEND', 2, 0, 10),
KcpConfig::int('LAYLINK_KCP_NO_CONGESTION_CONTROL', 1, 0, 1),
);
$this->ffi->laylink_kcp_wndsize(
$this->session,
KcpConfig::int('LAYLINK_KCP_SEND_WINDOW', 1024, 16, 8192),
KcpConfig::int('LAYLINK_KCP_RECV_WINDOW', 1024, 16, 8192),
);
$this->mtuBytes = KcpConfig::int('LAYLINK_KCP_MTU_BYTES', 1350, 576, 1400);
$this->ffi->laylink_kcp_setmtu($this->session, $this->mtuBytes);
$this->parser = new FrameParser();
}
public function __destruct()
{
if (!$this->closed) {
$this->closed = true;
$this->ffi->laylink_kcp_release($this->session);
}
}
public function conv(): int
{
return $this->conv;
}
public function sendFrame(Frame $frame, int $maxSendBuffer): bool|null
{
if ($this->closed || $this->getSendBufferQueueSize() >= $maxSendBuffer) {
return false;
}
$bytes = FrameCodec::encode($frame);
$result = $this->ffi->laylink_kcp_send($this->session, $bytes, strlen($bytes));
if ($result < 0) {
return false;
}
$this->ffi->laylink_kcp_flush($this->session);
$this->drainOutput();
return true;
}
public function receive(string|array $packet): void
{
if ($this->closed || !is_string($packet)) {
return;
}
if ($this->ffi->laylink_kcp_input($this->session, $packet, strlen($packet)) < 0) {
return;
}
$this->drainFrames();
$this->drainOutput();
}
public function tick(): void
{
if ($this->closed) {
return;
}
$this->ffi->laylink_kcp_update($this->session, $this->nowMs());
$this->drainFrames();
$this->drainOutput();
}
public function pause(): void
{
$this->paused = true;
}
public function resume(): void
{
$this->paused = false;
while (!$this->paused && $this->pausedFrames !== []) {
$frame = array_shift($this->pausedFrames);
($this->onFrame)($frame);
}
}
public function close(): void
{
if ($this->closed) {
return;
}
$this->closed = true;
($this->sendPacket)(KcpPacketCodec::encode([
'type' => KcpPacketCodec::CLOSE,
'conv' => $this->conv,
]));
$this->ffi->laylink_kcp_release($this->session);
}
public function isClosed(): bool
{
return $this->closed;
}
public function getSendBufferQueueSize(): int
{
$waitSegments = $this->ffi->laylink_kcp_waitsnd($this->session);
$waitBytes = max(0, $waitSegments) * $this->mtuBytes;
return $waitBytes + $this->pendingOutputBytes();
}
private function drainFrames(): void
{
while (($size = $this->ffi->laylink_kcp_peeksize($this->session)) > 0) {
if ($size > self::RECEIVE_BUFFER_BYTES) {
($this->onInvalidFrame)(new \RuntimeException('kcp_message_too_large'));
$this->close();
return;
}
$buffer = $this->ffi->new("char[$size]");
$read = $this->ffi->laylink_kcp_recv($this->session, $buffer, $size);
if ($read <= 0) {
return;
}
$bytes = FFI::string($buffer, $read);
try {
foreach ($this->parser->push($bytes) as $frame) {
if ($this->paused) {
$this->pausedFrames[] = $frame;
continue;
}
($this->onFrame)($frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($e);
}
}
}
private function drainOutput(): void
{
$drained = 0;
$maxPackets = KcpConfig::int('LAYLINK_KCP_OUTPUT_DRAIN_PACKETS', 1024, 1, 8192);
while ($drained < $maxPackets && ($size = $this->ffi->laylink_kcp_pending_output_size($this->session)) > 0) {
$buffer = $this->ffi->new("char[$size]");
$read = $this->ffi->laylink_kcp_pop_output($this->session, $buffer, $size);
if ($read <= 0) {
return;
}
$drained++;
if (($this->sendPacket)(FFI::string($buffer, $read)) === false) {
($this->onInvalidFrame)(new \RuntimeException('kcp_udp_send_queue_full'));
$this->close();
return;
}
}
}
private function pendingOutputBytes(): int
{
$size = $this->ffi->laylink_kcp_pending_output_size($this->session);
return $size > 0 ? $size : 0;
}
private function nowMs(): int
{
return (int)floor(microtime(true) * 1000);
}
}
+85
View File
@@ -0,0 +1,85 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use Workerman\Connection\AsyncTcpConnection;
final class TcpFrameClientTransport implements FrameClientTransport
{
private ?AsyncTcpConnection $connection = null;
private FrameParser $parser;
public function __construct(
private readonly string $address,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$this->parser = new FrameParser();
}
public function connect(): void
{
$this->parser = new FrameParser();
$connection = new AsyncTcpConnection($this->address);
$connection->maxSendBufferSize = $this->maxSendBuffer;
$this->connection = $connection;
$connection->onConnect = function () use ($connection): void {
($this->onConnect)($this);
};
$connection->onMessage = function (AsyncTcpConnection $connection, string $data): void {
try {
foreach ($this->parser->push($data) as $frame) {
($this->onFrame)($this, $frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($e);
$connection->close();
}
};
$connection->onBufferDrain = function () use ($connection): void {
($this->onBufferDrain)($this);
};
$connection->onClose = function () use ($connection): void {
if ($this->connection === $connection) {
$this->connection = null;
}
($this->onClose)($this);
};
$connection->connect();
}
public function send(Frame $frame): bool|null
{
return $this->connection?->send(FrameCodec::encode($frame));
}
public function close(): void
{
$this->connection?->close();
}
public function pauseRecv(): void
{
$this->connection?->pauseRecv();
}
public function resumeRecv(): void
{
$this->connection?->resumeRecv();
}
public function getSendBufferQueueSize(): int
{
return $this->connection?->getSendBufferQueueSize() ?? 0;
}
}
@@ -0,0 +1,63 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use LayLink\Protocol\Frame;
use LayLink\Protocol\FrameCodec;
use LayLink\Protocol\FrameParser;
use Workerman\Connection\TcpConnection;
final class TcpFrameServerConnection implements FrameServerConnection
{
private FrameParser $parser;
public function __construct(private readonly TcpConnection $connection)
{
$this->parser = new FrameParser();
}
public function id(): int
{
return $this->connection->id;
}
/**
* @return Frame[]
*/
public function push(string $data): array
{
return $this->parser->push($data);
}
public function send(Frame $frame): bool|null
{
return $this->connection->send(FrameCodec::encode($frame));
}
public function close(): void
{
$this->connection->close();
}
public function pauseRecv(): void
{
$this->connection->pauseRecv();
}
public function resumeRecv(): void
{
$this->connection->resumeRecv();
}
public function getSendBufferQueueSize(): int
{
return $this->connection->getSendBufferQueueSize();
}
public function getRemoteIp(): string
{
return $this->connection->getRemoteIp();
}
}
+66
View File
@@ -0,0 +1,66 @@
<?php
declare(strict_types=1);
namespace LayLink\Transport;
use Workerman\Connection\TcpConnection;
use Workerman\Worker;
final class TcpFrameServerListener
{
/** @var array<int, TcpFrameServerConnection> */
private array $connections = [];
public function __construct(
Worker $worker,
private readonly int $maxSendBuffer,
private readonly \Closure $onConnect,
private readonly \Closure $onFrame,
private readonly \Closure $onClose,
private readonly \Closure $onBufferDrain,
private readonly \Closure $onInvalidFrame,
) {
$worker->onConnect = fn (TcpConnection $connection) => $this->handleConnect($connection);
$worker->onMessage = fn (TcpConnection $connection, string $data) => $this->handleMessage($connection, $data);
$worker->onClose = fn (TcpConnection $connection) => $this->handleClose($connection);
}
private function handleConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = $this->maxSendBuffer;
$wrapped = new TcpFrameServerConnection($connection);
$this->connections[$connection->id] = $wrapped;
$connection->onBufferDrain = function () use ($wrapped): void {
($this->onBufferDrain)($wrapped);
};
($this->onConnect)($wrapped);
}
private function handleMessage(TcpConnection $connection, string $data): void
{
$wrapped = $this->connections[$connection->id] ?? null;
if ($wrapped === null) {
return;
}
try {
foreach ($wrapped->push($data) as $frame) {
($this->onFrame)($wrapped, $frame);
}
} catch (\Throwable $e) {
($this->onInvalidFrame)($wrapped, $e);
}
}
private function handleClose(TcpConnection $connection): void
{
$wrapped = $this->connections[$connection->id] ?? null;
unset($this->connections[$connection->id]);
if ($wrapped === null) {
return;
}
($this->onClose)($wrapped);
}
}
+26
View File
@@ -49,6 +49,21 @@ final class Env
return in_array(strtolower(trim($value)), ['1', 'true', 'yes', 'on'], true);
}
public static function int(string $key, int $default, int $min = PHP_INT_MIN, int $max = PHP_INT_MAX): int
{
$value = getenv($key);
if ($value === false || trim($value) === '') {
return $default;
}
$value = trim($value);
if (!preg_match('/^-?\d+$/', $value)) {
return $default;
}
return max($min, min($max, (int)$value));
}
/**
* @return string[]
*/
@@ -59,6 +74,17 @@ final class Env
return $default;
}
$value = trim($value);
if (str_starts_with($value, '[')) {
$decoded = json_decode($value, true);
if (is_array($decoded)) {
return array_values(array_filter(array_map(
static fn (mixed $item): string => strtolower(trim((string)$item)),
$decoded,
)));
}
}
return array_values(array_filter(array_map(
static fn (string $item): string => strtolower(trim($item)),
explode(',', $value),