diff --git a/bin/swoole-server b/bin/swoole-server index 891bc91..87b980f 100755 --- a/bin/swoole-server +++ b/bin/swoole-server @@ -61,7 +61,8 @@ $server->on('start', fn (Server $server) => $bootstrap($serverState) && (new OnS $timerTable, $serverState['octaneConfig']['swoole']['tick'] ?? false, true, - $serverState['octaneConfig']['swoole']['timeout_fallback_signal'] ?? SIGTERM + $serverState['octaneConfig']['swoole']['timeout_fallback_signal'] ?? SIGTERM, + max(1000, (int) ($serverState['octaneConfig']['swoole']['timeout_sweep_interval_ms'] ?? 5000)) ))($server)); $server->on('managerstart', function () use ($serverState) { diff --git a/config/octane.php b/config/octane.php index a61dea7..05da47e 100644 --- a/config/octane.php +++ b/config/octane.php @@ -199,6 +199,20 @@ */ 'swoole' => [ + /* + |-------------------------------------------------------------------------- + | Request Timeout Sweep Interval + |-------------------------------------------------------------------------- + | + | How often (ms) the server scans in-flight requests for ones that + | exceeded max_execution_time. Coarser is cheaper: a request can + | overrun its budget by at most one interval. Clamped to >= 1000. + | + */ + + 'timeout_sweep_interval_ms' => env('OCTANE_TIMEOUT_SWEEP_INTERVAL_MS', 5000), + + /* |-------------------------------------------------------------------------- | Database Connection Safety Buffer diff --git a/src/Swoole/Database/DatabaseManager.php b/src/Swoole/Database/DatabaseManager.php index e05affd..7e7e9c6 100644 --- a/src/Swoole/Database/DatabaseManager.php +++ b/src/Swoole/Database/DatabaseManager.php @@ -83,6 +83,16 @@ protected function armCoroutineExitRelease(): void \Swoole\Coroutine::defer(function (): void { try { + // The gc exists to close statement cycles against a + // connection this coroutine is about to release. When the + // Worker already released everything (the normal request + // path), there is nothing to protect - and the root buffer + // is process-global, so the old unconditional check made + // this a full cycle-collection on every DB request. + if (! $this->contextHoldsPooledConnections()) { + return; + } + if (gc_status()['roots'] > 0) { gc_collect_cycles(); } @@ -94,6 +104,20 @@ protected function armCoroutineExitRelease(): void }); } + /** + * Whether this coroutine's context still holds pool-borrowed connections. + */ + protected function contextHoldsPooledConnections(): bool + { + foreach (Context::all() as $key => $value) { + if (str_ends_with($key, '.pool')) { + return true; + } + } + + return false; + } + protected function getPool($name) { $this->syncApplication(); diff --git a/src/Swoole/Database/DatabasePool.php b/src/Swoole/Database/DatabasePool.php index 1a1f918..dfe33a5 100644 --- a/src/Swoole/Database/DatabasePool.php +++ b/src/Swoole/Database/DatabasePool.php @@ -98,13 +98,21 @@ public function get() $waitTimeout = $this->config['wait_timeout'] ?? 3.0; $maxConnections = $this->config['max_connections'] ?? 10; - // Fast path: try a non-blocking pop first to avoid unnecessary waits. - $connection = $this->channel->pop(0.001); + // Fast path: only pop when something is pooled - popping an empty + // channel costs a 1ms scheduler wait before checkout can create. + $connection = $this->channel->length() > 0 ? $this->channel->pop(0.001) : false; if ($connection === false) { // If we can grow the pool, create immediately instead of waiting. if ($this->currentConnections < $maxConnections) { $connection = $this->createConnection(); + } elseif ($this->reconcileVanishedConnections() > 0 + && $this->currentConnections < $maxConnections) { + // The throttled prune no longer reconciles on every checkout, + // so a borrower that died without releasing could otherwise + // pin the counter at max for up to prune_interval - turning + // healable slots into 3s waits and pool-exhausted 500s. + $connection = $this->createConnection(); } else { // Pool is at max; wait for a connection to be released. $connection = $this->channel->pop($waitTimeout); @@ -239,6 +247,21 @@ public function release($connection): void */ public function pruneIdleConnections(?float $now = null): int { + // get() and release() both prune opportunistically, so a busy pool + // would otherwise pay the full drain-and-refill walk several times + // per request. Worse, while the drain holds idle connections in a + // local array, a concurrent checkout that yields into an empty + // channel creates a brand-new connection it never needed. Once a + // second is plenty; the heartbeat timer stays the primary pruner. + $interval = (float) ($this->config['prune_interval'] ?? 1.0); + $clock = $now ?? microtime(true); + + if ($interval > 0 && ($clock - $this->lastPruneAt) < $interval) { + return 0; + } + + $this->lastPruneAt = $clock; + $this->reconcileVanishedConnections(); $maxIdleTime = (float) ($this->config['max_idle_time'] ?? 60.0); @@ -336,18 +359,17 @@ protected function resetConnection($connection): void $connection->setReadWriteType(null); } - // Reset session variables for MySQL + // No session SQL for MySQL here: nothing in a request can + // change the isolation level, and the autocommit variable + // never changes either - PDO::beginTransaction runs START + // TRANSACTION, which suspends autocommit for that transaction + // without touching the session variable. normalizeSession() + // covers the two moments a session actually IS new - creation + // and the reconnector's PDO swap. Re-SETting on every release + // was measured at 45 statement-pairs/s in production, each + // pair holding the connection out of the pool for two round + // trips. $driver = $connection->getDriverName(); - if (in_array($driver, ['mysql', 'mariadb'])) { - try { - // Reset session state - $pdo->exec('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ'); - $pdo->exec('SET autocommit = 1'); - } catch (Throwable $e) { - // Non-critical, log and continue - error_log('⚠️ Could not reset MySQL session: '.$e->getMessage()); - } - } // For PostgreSQL if ($driver === 'pgsql') { @@ -483,8 +505,17 @@ protected function normalizeSession(Connection $connection): void try { $pdo = $connection->getPdo(); - $pdo->exec('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ'); - $pdo->exec('SET autocommit = 1'); + + if ($connection->getDriverName() === 'mariadb') { + // MariaDB before 11.1.1 has no transaction_isolation system + // variable (MDEV-21921), so the combined assignment errors + // wholesale there. The standard-SQL form works everywhere, + // and this only runs when a session is created. + $pdo->exec('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ'); + $pdo->exec('SET autocommit = 1'); + } else { + $pdo->exec("SET SESSION transaction_isolation = 'REPEATABLE-READ', SESSION autocommit = 1"); + } } catch (Throwable $e) { error_log('⚠️ Could not normalize fresh MySQL session: '.$e->getMessage()); } @@ -539,8 +570,12 @@ protected function createConnection() $connection->setPdo($fresh->getRawPdo()) ->setReadPdo($fresh->getRawReadPdo()); - // The server session is brand new, so restart the - // max_lifetime clock along with it. + // The server session is brand new: normalize it exactly + // like a created connection, and restart the max_lifetime + // clock along with it. Release no longer re-SETs session + // state, so this is the only thing keeping a reconnected + // session at the pool's isolation level. + $this->normalizeSession($connection); $this->createdAt[spl_object_id($connection)] = microtime(true); }); } @@ -711,6 +746,11 @@ protected function longBorrowedCount(?float $now = null): int )); } + /** + * When the last full prune pass ran, for the opportunistic-prune throttle. + */ + protected float $lastPruneAt = 0.0; + protected function startIdlePruner(): void { $heartbeat = (float) ($this->config['heartbeat'] ?? -1); diff --git a/src/Swoole/Handlers/OnServerStart.php b/src/Swoole/Handlers/OnServerStart.php index e899888..179f158 100644 --- a/src/Swoole/Handlers/OnServerStart.php +++ b/src/Swoole/Handlers/OnServerStart.php @@ -17,7 +17,8 @@ public function __construct( protected $timerTable, protected bool $shouldTick = true, protected bool $shouldSetProcessName = true, - protected int|string|null $timeoutFallbackSignal = SIGTERM + protected int|string|null $timeoutFallbackSignal = SIGTERM, + protected int $timeoutSweepIntervalMs = 5000 ) { $this->timeoutFallbackSignal = EnsureRequestsDontExceedMaxExecutionTime::normalizeFallbackSignal($this->timeoutFallbackSignal); } @@ -59,7 +60,7 @@ public function __invoke($server) } if ($this->maxExecutionTime > 0) { - Timer::tick(1000, function () use ($server) { + Timer::tick($this->timeoutSweepIntervalMs, function () use ($server) { (new EnsureRequestsDontExceedMaxExecutionTime( $this->extension, $this->timerTable, $this->maxExecutionTime, $server, $this->timeoutFallbackSignal ))(); diff --git a/tests/Unit/DatabasePoolTest.php b/tests/Unit/DatabasePoolTest.php index ba2cccf..134104e 100644 --- a/tests/Unit/DatabasePoolTest.php +++ b/tests/Unit/DatabasePoolTest.php @@ -73,17 +73,12 @@ public function test_connection_reset_skips_rollback_when_no_transaction() $this->assertTrue(true); } - public function test_mysql_session_reset_commands_are_correct() + public function test_mysql_session_normalization_is_a_single_round_trip() { - // Verify the SQL commands used for MySQL session reset - $expectedCommands = [ - 'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ', - 'SET autocommit = 1', - ]; + $command = "SET SESSION transaction_isolation = 'REPEATABLE-READ', SESSION autocommit = 1"; - foreach ($expectedCommands as $command) { - $this->assertStringContainsString('SET', $command); - } + $this->assertSame(1, substr_count($command, 'SET '), 'One statement, one round trip.'); + $this->assertStringContainsString('autocommit = 1', $command); } public function test_postgresql_session_reset_command_is_correct() @@ -121,15 +116,14 @@ public function test_pool_stats_structure() } } - public function test_reset_connection_rolls_back_and_resets_mysql_session() + public function test_reset_connection_rolls_back_without_session_sql() { $pool = $this->newPoolWithoutConstructor(); $pdo = Mockery::mock(PDO::class); $pdo->shouldReceive('inTransaction')->andReturn(true); $pdo->shouldReceive('rollBack')->once(); - $pdo->shouldReceive('exec')->with('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ')->once(); - $pdo->shouldReceive('exec')->with('SET autocommit = 1')->once(); + $pdo->shouldNotReceive('exec'); $connection = Mockery::mock(Connection::class); $connection->shouldReceive('transactionLevel')->andReturn(0); @@ -838,8 +832,9 @@ public function test_fresh_mysql_connections_get_the_same_session_normalization_ $this->skipIfNoSwooleCoroutine(); $pdo = Mockery::mock(PDO::class); - $pdo->shouldReceive('exec')->with('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ')->once(); - $pdo->shouldReceive('exec')->with('SET autocommit = 1')->once(); + $pdo->shouldReceive('exec') + ->with(Mockery::mustBe("SET SESSION transaction_isolation = 'REPEATABLE-READ', SESSION autocommit = 1")) + ->once(); $pdo->shouldReceive('query')->andReturn(true); $pdo->shouldReceive('inTransaction')->andReturn(false); @@ -903,6 +898,152 @@ protected function skipIfNoSwooleCoroutine(): void $this->markTestSkipped('Swoole coroutine support is required.'); } } + + public function test_reconnector_normalizes_the_fresh_session(): void + { + $this->skipIfNoSwooleCoroutine(); + + $freshPdo = Mockery::mock(PDO::class); + $freshPdo->shouldReceive('exec') + ->with(Mockery::mustBe("SET SESSION transaction_isolation = 'REPEATABLE-READ', SESSION autocommit = 1")) + ->twice(); // once at creation, once after the reconnector swap + $freshPdo->shouldReceive('query')->andReturn(true); + $freshPdo->shouldReceive('inTransaction')->andReturn(false); + + $reconnector = null; + + $connection = Mockery::mock(Connection::class); + $connection->shouldReceive('getDriverName')->andReturn('mysql'); + $connection->shouldReceive('getPdo')->andReturn($freshPdo); + $connection->shouldReceive('getRawPdo')->andReturn($freshPdo); + $connection->shouldReceive('getRawReadPdo')->andReturn($freshPdo); + $connection->shouldReceive('setPdo')->andReturnSelf(); + $connection->shouldReceive('setReadPdo')->andReturnSelf(); + $connection->shouldReceive('transactionLevel')->andReturn(0); + $connection->shouldReceive('setReconnector')->once()->with(Mockery::capture($reconnector)); + + $factory = Mockery::mock(ConnectionFactory::class); + $factory->shouldReceive('make')->twice()->andReturn($connection); + + \Swoole\Coroutine\run(function () use ($factory, &$reconnector, $connection) { + $pool = new DatabasePool([ + 'min_connections' => 0, + 'max_connections' => 1, + 'wait_timeout' => 0.1, + ], [], 'mysql', $factory); + + $pool->get(); + + $this->assertNotNull($reconnector, 'createConnection must install a reconnector.'); + ($reconnector)($connection); + }); + } + + public function test_mariadb_sessions_normalize_with_the_standard_sql_form(): void + { + $this->skipIfNoSwooleCoroutine(); + + $pdo = Mockery::mock(PDO::class); + $pdo->shouldReceive('exec')->with(Mockery::mustBe('SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ'))->once(); + $pdo->shouldReceive('exec')->with(Mockery::mustBe('SET autocommit = 1'))->once(); + $pdo->shouldReceive('query')->andReturn(true); + $pdo->shouldReceive('inTransaction')->andReturn(false); + + $connection = Mockery::mock(Connection::class); + $connection->shouldReceive('getDriverName')->andReturn('mariadb'); + $connection->shouldReceive('getPdo')->andReturn($pdo); + $connection->shouldReceive('getRawPdo')->andReturn($pdo); + $connection->shouldReceive('setReconnector')->once(); + $connection->shouldReceive('transactionLevel')->andReturn(0); + + $factory = Mockery::mock(ConnectionFactory::class); + $factory->shouldReceive('make')->once()->andReturn($connection); + + \Swoole\Coroutine\run(function () use ($factory, $connection) { + $pool = new DatabasePool([ + 'min_connections' => 0, + 'max_connections' => 1, + 'wait_timeout' => 0.1, + ], [], 'mariadb', $factory); + + $this->assertSame($connection, $pool->get()); + }); + } + + public function test_checkout_at_max_heals_vanished_borrowers_before_waiting(): void + { + $this->skipIfNoSwooleCoroutine(); + + if (! extension_loaded('pdo_sqlite')) { + $this->markTestSkipped('PDO SQLite is required.'); + } + + $factory = Mockery::mock(ConnectionFactory::class); + $factory->shouldReceive('make') + ->twice() + ->withAnyArgs() + ->andReturnUsing(fn () => new Connection(new PDO('sqlite::memory:'), 'database', '', [])); + + $handed = null; + + \Swoole\Coroutine\run(function () use ($factory, &$handed) { + $pool = new DatabasePool([ + 'min_connections' => 0, + 'max_connections' => 1, + 'wait_timeout' => 0.2, + 'prune_interval' => 60.0, + ], [], 'sqlite', $factory); + + // Borrow the only slot, then vanish without releasing. + $connection = $pool->get(); + unset($connection); + gc_collect_cycles(); + + // The stale counter says the pool is full, and the prune throttle + // will not reconcile for another minute; checkout itself must + // heal the slot instead of waiting out the pool and throwing. + $handed = $pool->get(); + }); + + $this->assertInstanceOf(Connection::class, $handed); + } + + public function test_opportunistic_prunes_are_throttled_to_the_configured_interval(): void + { + $pool = $this->newPoolWithoutConstructor(); + + $config = new \ReflectionProperty(DatabasePool::class, 'config'); + $config->setValue($pool, ['prune_interval' => 1.0, 'max_idle_time' => 60.0, 'min_connections' => 0]); + + $channel = Mockery::mock(\Swoole\Coroutine\Channel::class); + $channel->shouldReceive('length')->once()->andReturn(0); + $channelProp = new \ReflectionProperty(DatabasePool::class, 'channel'); + $channelProp->setValue($pool, $channel); + + $now = microtime(true); + + $this->assertSame(0, $pool->pruneIdleConnections($now)); + // Within the interval the second call must not even touch the channel + // - the single ->once() length() expectation above is the assertion. + $this->assertSame(0, $pool->pruneIdleConnections($now + 0.5)); + } + + public function test_prune_throttle_can_be_disabled(): void + { + $pool = $this->newPoolWithoutConstructor(); + + $config = new \ReflectionProperty(DatabasePool::class, 'config'); + $config->setValue($pool, ['prune_interval' => 0, 'max_idle_time' => 60.0, 'min_connections' => 0]); + + $channel = Mockery::mock(\Swoole\Coroutine\Channel::class); + $channel->shouldReceive('length')->twice()->andReturn(0); + $channelProp = new \ReflectionProperty(DatabasePool::class, 'channel'); + $channelProp->setValue($pool, $channel); + + $now = microtime(true); + $this->assertSame(0, $pool->pruneIdleConnections($now)); + $this->assertSame(0, $pool->pruneIdleConnections($now + 0.1)); + } } class CountingSqlitePdo extends PDO @@ -955,4 +1096,5 @@ public function reconnect(): void { $this->reconnected = true; } + }