laysense/laylink
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

压力优化

Browse Source
This commit is contained in:
Ziki Shay
2026-05-28 23:55:16 +08:00
parent 9c07b9fadc
commit 068d3f4be9
4 changed files with 167 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Agent/AgentClient.php
+58 -6
View File
@@ -17,6 +17,9 @@ use Workerman\Worker;
final class AgentClient
{
private const MAX_SEND_BUFFER = 32 * 1024 * 1024;
private const BACKPRESSURE_HIGH_WATERMARK = 16 * 1024 * 1024;
private ?AsyncTcpConnection $pop = null;
private ?FrameParser $parser = null;
private bool $authenticated = false;
@@ -38,6 +41,10 @@ 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 = [];
public function __construct(
private readonly string $clientListen,
@@ -87,7 +94,7 @@ final class AgentClient
$this->parser = new FrameParser();
$this->authenticated = false;
$connection = new AsyncTcpConnection($this->popAddress);
$connection->maxSendBufferSize = 8 * 1024 * 1024;
$connection->maxSendBufferSize = self::MAX_SEND_BUFFER;
$this->pop = $connection;
$connection->onConnect = function (AsyncTcpConnection $connection): void {
@@ -110,6 +117,7 @@ final class AgentClient
$connection->close();
}
};
$connection->onBufferDrain = fn (AsyncTcpConnection $connection) => $this->resumeClientsForPop();
$connection->onClose = function (): void {
$this->authenticated = false;
foreach ($this->clients as $client) {
@@ -122,6 +130,8 @@ final class AgentClient
$this->pendingData = [];
$this->connectionStages = [];
$this->sessionIngressProtocols = [];
$this->pausedClientsForPop = [];
$this->clientsPausingPop = [];
Timer::add(3, fn () => $this->connect(), [], false);
};
$connection->connect();
@@ -144,7 +154,9 @@ final class AgentClient
private function onClientConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = 8 * 1024 * 1024;
$connection->maxSendBufferSize = self::MAX_SEND_BUFFER;
$connection->onBufferFull = fn (TcpConnection $connection) => $this->pausePopForClient($connection);
$connection->onBufferDrain = fn (TcpConnection $connection) => $this->resumePopForClient($connection);
$this->initialBuffers[$connection->id] = '';
$this->connectionStages[$connection->id] = 'init';
}
@@ -162,7 +174,15 @@ final class AgentClient
return;
}
$this->send(new Frame(FrameType::DATA, $sessionId, ['data_raw' => $data]));
$sent = $this->send(new Frame(FrameType::DATA, $sessionId, ['data_raw' => $data]));
if ($sent === false) {
$this->closeClient($sessionId);
return;
}
if ($this->pop !== null && $this->pop->getSendBufferQueueSize() >= self::BACKPRESSURE_HIGH_WATERMARK) {
$connection->pauseRecv();
$this->pausedClientsForPop[$connection->id] = $connection;
}
}
private function handleInitialRequest(TcpConnection $connection, string $data): void
@@ -579,7 +599,15 @@ final class AgentClient
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() >= self::BACKPRESSURE_HIGH_WATERMARK) {
$this->pausePopForClient($client);
}
}
private function frameData(Frame $frame): string|false
@@ -595,6 +623,7 @@ final class AgentClient
{
unset($this->initialBuffers[$connection->id]);
unset($this->connectionStages[$connection->id]);
unset($this->pausedClientsForPop[$connection->id], $this->clientsPausingPop[$connection->id]);
if (!isset($this->connectionSessionIds[$connection->id])) {
return;
}
@@ -614,6 +643,7 @@ final class AgentClient
$connection = $this->clients[$sessionId];
unset($this->clients[$sessionId], $this->sessionStates[$sessionId], $this->pendingData[$sessionId], $this->sessionIngressProtocols[$sessionId]);
unset($this->connectionSessionIds[$connection->id]);
unset($this->pausedClientsForPop[$connection->id], $this->clientsPausingPop[$connection->id]);
$connection->close();
}
@@ -766,8 +796,30 @@ final class AgentClient
]));
}
private function send(Frame $frame): void
private function send(Frame $frame): bool|null
{
$this->pop?->send(FrameCodec::encode($frame));
return $this->pop?->send(FrameCodec::encode($frame));
}
private function resumeClientsForPop(): void
{
foreach ($this->pausedClientsForPop as $connectionId => $client) {
$client->resumeRecv();
unset($this->pausedClientsForPop[$connectionId]);
}
}
private function pausePopForClient(TcpConnection $connection): void
{
$this->clientsPausingPop[$connection->id] = true;
$this->pop?->pauseRecv();
}
private function resumePopForClient(TcpConnection $connection): void
{
unset($this->clientsPausingPop[$connection->id]);
if ($this->clientsPausingPop === []) {
$this->pop?->resumeRecv();
}
}
}
src/Server/AgentListener.php
+87 -8
View File
@@ -24,10 +24,15 @@ use Workerman\Worker;
final class AgentListener
{
private const MAX_SEND_BUFFER = 32 * 1024 * 1024;
private const BACKPRESSURE_HIGH_WATERMARK = 16 * 1024 * 1024;
/** @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,
@@ -46,7 +51,8 @@ final class AgentListener
private function onConnect(TcpConnection $connection): void
{
$connection->maxSendBufferSize = 8 * 1024 * 1024;
$connection->maxSendBufferSize = self::MAX_SEND_BUFFER;
$connection->onBufferDrain = fn (TcpConnection $connection) => $this->resumeTargetsForAgent($connection);
$this->parsers[$connection->id] = new FrameParser();
}
@@ -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,8 +231,8 @@ 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 = self::MAX_SEND_BUFFER;
$session->target = $target;
$target->onConnect = function () use ($session, $agentConnection): void {
@@ -232,9 +244,17 @@ final class AgentListener
};
$target->onMessage = function (AsyncTcpConnection $target, string $data) use ($session, $agentConnection): void {
$session->bytesTargetToClient += strlen($data);
$this->send($agentConnection, new Frame(FrameType::DATA, $session->sessionId, [
$sent = $this->send($agentConnection, new Frame(FrameType::DATA, $session->sessionId, [
'data_raw' => $data,
]));
if ($sent === false) {
$this->closeSession($session, 'failed', 'agent_buffer_overflow');
return;
}
if ($agentConnection->getSendBufferQueueSize() >= self::BACKPRESSURE_HIGH_WATERMARK) {
$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');
@@ -250,12 +270,28 @@ final class AgentListener
}
$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() >= self::BACKPRESSURE_HIGH_WATERMARK && $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
{
$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);
}
@@ -298,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]));
@@ -329,6 +368,7 @@ final class AgentListener
{
unset($this->parsers[$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) {
@@ -351,8 +391,47 @@ final class AgentListener
}
}
private function send(TcpConnection $connection, Frame $frame): void
private function send(TcpConnection $connection, Frame $frame): bool|null
{
$connection->send(FrameCodec::encode($frame));
return $connection->send(FrameCodec::encode($frame));
}
private function resumeTargetsForAgent(TcpConnection $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;
}
}