优化3
This commit is contained in:
@@ -17,10 +17,6 @@ use Workerman\Worker;
|
||||
|
||||
final class AgentClient
|
||||
{
|
||||
private const MAX_SEND_BUFFER = 32 * 1024 * 1024;
|
||||
private const BACKPRESSURE_HIGH_WATERMARK = 16 * 1024 * 1024;
|
||||
private const DATA_CHUNK_SIZE = 256 * 1024;
|
||||
|
||||
private ?AsyncTcpConnection $pop = null;
|
||||
private ?FrameParser $parser = null;
|
||||
private bool $authenticated = false;
|
||||
@@ -67,6 +63,9 @@ 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,
|
||||
) {
|
||||
}
|
||||
|
||||
@@ -99,7 +98,7 @@ final class AgentClient
|
||||
$this->parser = new FrameParser();
|
||||
$this->authenticated = false;
|
||||
$connection = new AsyncTcpConnection($this->popAddress);
|
||||
$connection->maxSendBufferSize = self::MAX_SEND_BUFFER;
|
||||
$connection->maxSendBufferSize = $this->maxSendBuffer;
|
||||
$this->pop = $connection;
|
||||
|
||||
$connection->onConnect = function (AsyncTcpConnection $connection): void {
|
||||
@@ -161,7 +160,7 @@ final class AgentClient
|
||||
|
||||
private function onClientConnect(TcpConnection $connection): void
|
||||
{
|
||||
$connection->maxSendBufferSize = self::MAX_SEND_BUFFER;
|
||||
$connection->maxSendBufferSize = $this->maxSendBuffer;
|
||||
$connection->onBufferFull = fn (TcpConnection $connection) => $this->pausePopForClient($connection);
|
||||
$connection->onBufferDrain = fn (TcpConnection $connection) => $this->onClientBufferDrain($connection);
|
||||
$this->initialBuffers[$connection->id] = '';
|
||||
@@ -185,7 +184,7 @@ final class AgentClient
|
||||
$this->closeClient($sessionId);
|
||||
return;
|
||||
}
|
||||
if ($this->pop !== null && $this->pop->getSendBufferQueueSize() >= self::BACKPRESSURE_HIGH_WATERMARK) {
|
||||
if ($this->pop !== null && $this->pop->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
|
||||
$connection->pauseRecv();
|
||||
$this->pausedClientsForPop[$connection->id] = $connection;
|
||||
}
|
||||
@@ -611,7 +610,7 @@ final class AgentClient
|
||||
$this->closeClient($frame->sessionId);
|
||||
return;
|
||||
}
|
||||
if ($client->getSendBufferQueueSize() >= self::BACKPRESSURE_HIGH_WATERMARK) {
|
||||
if ($client->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
|
||||
$this->pausePopForClient($client);
|
||||
}
|
||||
}
|
||||
@@ -833,9 +832,9 @@ final class AgentClient
|
||||
private function sendData(string $sessionId, string $data): bool
|
||||
{
|
||||
$length = strlen($data);
|
||||
for ($offset = 0; $offset < $length; $offset += self::DATA_CHUNK_SIZE) {
|
||||
for ($offset = 0; $offset < $length; $offset += $this->dataChunkSize) {
|
||||
$sent = $this->send(new Frame(FrameType::DATA, $sessionId, [
|
||||
'data_raw' => substr($data, $offset, self::DATA_CHUNK_SIZE),
|
||||
'data_raw' => substr($data, $offset, $this->dataChunkSize),
|
||||
]));
|
||||
if ($sent === false) {
|
||||
return false;
|
||||
|
||||
@@ -24,10 +24,6 @@ use Workerman\Worker;
|
||||
|
||||
final class AgentListener
|
||||
{
|
||||
private const MAX_SEND_BUFFER = 32 * 1024 * 1024;
|
||||
private const BACKPRESSURE_HIGH_WATERMARK = 16 * 1024 * 1024;
|
||||
private const DATA_CHUNK_SIZE = 256 * 1024;
|
||||
|
||||
/** @var array<int, FrameParser> */
|
||||
private array $parsers = [];
|
||||
/** @var array<int, string> */
|
||||
@@ -43,6 +39,9 @@ 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,
|
||||
) {
|
||||
$worker->onConnect = fn (TcpConnection $connection) => $this->onConnect($connection);
|
||||
$worker->onMessage = fn (TcpConnection $connection, string $data) => $this->onMessage($connection, $data);
|
||||
@@ -52,7 +51,7 @@ final class AgentListener
|
||||
|
||||
private function onConnect(TcpConnection $connection): void
|
||||
{
|
||||
$connection->maxSendBufferSize = self::MAX_SEND_BUFFER;
|
||||
$connection->maxSendBufferSize = $this->maxSendBuffer;
|
||||
$connection->onBufferDrain = fn (TcpConnection $connection) => $this->resumeTargetsForAgent($connection);
|
||||
$this->parsers[$connection->id] = new FrameParser();
|
||||
}
|
||||
@@ -231,7 +230,7 @@ final class AgentListener
|
||||
$this->sessions->add($session);
|
||||
|
||||
$target = new AsyncTcpConnection($targetAddress);
|
||||
$target->maxSendBufferSize = self::MAX_SEND_BUFFER;
|
||||
$target->maxSendBufferSize = $this->maxSendBuffer;
|
||||
$session->target = $target;
|
||||
|
||||
$target->onConnect = function () use ($session, $agentConnection): void {
|
||||
@@ -247,7 +246,7 @@ final class AgentListener
|
||||
$this->closeSession($session, 'failed', 'agent_buffer_overflow');
|
||||
return;
|
||||
}
|
||||
if ($agentConnection->getSendBufferQueueSize() >= self::BACKPRESSURE_HIGH_WATERMARK) {
|
||||
if ($agentConnection->getSendBufferQueueSize() >= $this->backpressureHighWatermark) {
|
||||
$target->pauseRecv();
|
||||
$this->pausedTargetsByAgent[$agentConnection->id][$session->sessionId] = $target;
|
||||
}
|
||||
@@ -275,7 +274,7 @@ final class AgentListener
|
||||
$this->closeSession($session, 'failed', 'target_buffer_overflow');
|
||||
return;
|
||||
}
|
||||
if ($session->target->getSendBufferQueueSize() >= self::BACKPRESSURE_HIGH_WATERMARK && $session->agent !== null) {
|
||||
if ($session->target->getSendBufferQueueSize() >= $this->backpressureHighWatermark && $session->agent !== null) {
|
||||
$session->agent->pauseRecv();
|
||||
$session->target->onBufferDrain = function (AsyncTcpConnection $target) use ($session): void {
|
||||
$session->agent?->resumeRecv();
|
||||
@@ -395,9 +394,9 @@ final class AgentListener
|
||||
private function sendData(TcpConnection $connection, string $sessionId, string $data): bool
|
||||
{
|
||||
$length = strlen($data);
|
||||
for ($offset = 0; $offset < $length; $offset += self::DATA_CHUNK_SIZE) {
|
||||
for ($offset = 0; $offset < $length; $offset += $this->dataChunkSize) {
|
||||
$sent = $this->send($connection, new Frame(FrameType::DATA, $sessionId, [
|
||||
'data_raw' => substr($data, $offset, self::DATA_CHUNK_SIZE),
|
||||
'data_raw' => substr($data, $offset, $this->dataChunkSize),
|
||||
]));
|
||||
if ($sent === false) {
|
||||
return false;
|
||||
|
||||
@@ -25,6 +25,9 @@ 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,
|
||||
) {
|
||||
$this->nodes = new NodeRegistry();
|
||||
$this->sessions = new SessionManager();
|
||||
@@ -44,6 +47,9 @@ final class PopServer
|
||||
$this->nodes,
|
||||
$this->sessions,
|
||||
$this->audit,
|
||||
$this->maxSendBuffer,
|
||||
$this->backpressureHighWatermark,
|
||||
$this->dataChunkSize,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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[]
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user