Skip to content

Complete PgSqlHandle operations only once the connection is idle #75

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
danog wants to merge 1 commit into
base: 2.x
Choose a base branch
from
Open

Complete PgSqlHandle operations only once the connection is idle #75

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 42 additions & 40 deletions src/Internal/PgSqlHandle.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,11 @@ public function __construct(
$deferred = &$this->pendingOperation;
$listeners = &$this->listeners;
$onClose = new DeferredFuture();
$results = [];

$poll = EventLoop::onReadable($socket, static function (string $watcher) use (
&$deferred,
&$results,
&$lastUsedAt,
&$listeners,
&$connection,
Expand Down Expand Up @@ -118,15 +120,32 @@ public function __construct(
return; // No active query, only notification listeners.
}

if (\pg_connection_busy($connection)) {
return;
}
// Collect every result of the operation before completing it: the operation only ends once
// the connection is idle again. Otherwise, another operation could be sent in between, and
// fetching the remaining results of this one would block on and consume its results instead.
$operation = $deferred;
while (!\pg_connection_busy($connection)) {
$result = \pg_get_result($connection);
if ($result !== false) {
$results[] = $result;
$status = \pg_result_status($result);
if ($status !== \PGSQL_COPY_IN && $status !== \PGSQL_COPY_OUT) {
continue;
}
// COPY is not supported, the result is an error and closes the connection.
}

$deferred->complete(\pg_get_result($connection));
$deferred = null;
$operationResults = $results;
$results = [];

$deferred = null;
$operation->complete($operationResults);

if (empty($listeners)) {
EventLoop::unreference($watcher);
}

if (empty($listeners)) {
EventLoop::unreference($watcher);
return;
}
} catch (SqlConnectionException $exception) {
$connection = null; // Marks connection as dead.
Expand Down Expand Up @@ -209,7 +228,7 @@ private function fetchTypes(): Future

EventLoop::queue(function () use ($queryDeferred, $typesDeferred): void {
try {
$result = $queryDeferred->getFuture()->await();
[$result] = $queryDeferred->getFuture()->await();
if (\pg_result_status($result) !== \PGSQL_TUPLES_OK) {
throw new SqlException(\pg_result_error($result) ?: 'Unknown result error');
}
Expand Down Expand Up @@ -265,9 +284,11 @@ public function isClosed(): bool
* @param \Closure $function Function to execute.
* @param mixed ...$args Arguments to pass to function.
*
* @return non-empty-list<\PgSql\Result> All results of the operation.
*
* @throws SqlException
*/
private function send(\Closure $function, mixed ...$args): mixed
private function send(\Closure $function, mixed ...$args): array
{
$this->types ??= $this->fetchTypes()->await();

Expand Down Expand Up @@ -305,18 +326,20 @@ private function send(\Closure $function, mixed ...$args): mixed
}

/**
* @param \PgSql\Result $result PostgreSQL result resource.
* @param non-empty-list<\PgSql\Result> $results All results of the operation.
* @param string $sql Query SQL.
*
* @throws SqlException
* @throws SqlQueryError
*/
private function createResult(\PgSql\Result $result, string $sql): PostgresResult
private function createResult(array $results, string $sql): PostgresResult
{
if ($this->handle === null) {
throw new \Error("The connection to the database has been closed");
}

$result = \array_shift($results);

\assert($this->types !== null, 'Expected type array to be populated before creating a result');

switch ($status = \pg_result_status($result)) {
Expand All @@ -326,28 +349,23 @@ private function createResult(\PgSql\Result $result, string $sql): PostgresResul
case \PGSQL_COMMAND_OK:
return new PostgresCommandResult(
\pg_affected_rows($result),
Future::complete($this->fetchNextResult($sql)),
Future::complete($results ? $this->createResult($results, $sql) : null),
);

case \PGSQL_TUPLES_OK:
return new PgSqlResultSet($result, $this->types, Future::complete($this->fetchNextResult($sql)));
return new PgSqlResultSet(
$result,
$this->types,
Future::complete($results ? $this->createResult($results, $sql) : null),
);

case \PGSQL_NONFATAL_ERROR:
case \PGSQL_FATAL_ERROR:
$diagnostics = [];
foreach (self::DIAGNOSTIC_CODES as $fieldCode => $description) {
$diagnostics[$description] = \pg_result_error_field($result, $fieldCode);
}
$message = \pg_result_error($result) ?: 'Unknown result error';
\set_error_handler(self::getErrorHandler());
try {
while (\pg_connection_busy($this->handle) && \pg_get_result($this->handle)) {
// Clear all outstanding result rows from the connection
}
} finally {
\restore_error_handler();
throw new PostgresQueryError($message, $diagnostics, $sql);
}
throw new PostgresQueryError(\pg_result_error($result) ?: 'Unknown result error', $diagnostics, $sql);

case \PGSQL_BAD_RESPONSE:
$this->close();
Expand All @@ -365,22 +383,6 @@ private function createResult(\PgSql\Result $result, string $sql): PostgresResul
}
}

/**
* @throws SqlException
*/
private function fetchNextResult(string $sql): ?PostgresResult
{
if ($this->handle === null) {
throw new \Error("The connection to the database has been closed");
}

if ($result = \pg_get_result($this->handle)) {
return $this->createResult($result, $sql);
}

return null;
}

#[\Override]
public function statementExecute(string $name, array $params): PostgresResult
{
Expand Down Expand Up @@ -482,7 +484,7 @@ public function prepare(string $sql): PostgresStatement
}

$future = async(function () use ($name, $modifiedSql, $sql): string {
$result = $this->send(\pg_send_prepare(...), $name, $modifiedSql);
[$result] = $this->send(\pg_send_prepare(...), $name, $modifiedSql);

switch ($status = \pg_result_status($result)) {
case \PGSQL_COMMAND_OK:
Expand Down
47 changes: 47 additions & 0 deletions test/PgSqlStatementDeallocateTest.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,47 @@
<?php declare(strict_types=1);

namespace Amp\Postgres\Test;

use Amp\Postgres\PgSqlConnection;
use Amp\Postgres\PostgresConfig;
use PHPUnit\Framework\TestCase;
use Revolt\EventLoop;
use function Amp\delay;

/**
* Not an AsyncTestCase: the test runs on {main}, whose suspension is resumed by interrupting the event loop,
* so the query waiting for the DEALLOCATE to complete is sent before the fiber that sent the DEALLOCATE resumes.
*
* @requires extension pgsql
*/
class PgSqlStatementDeallocateTest extends TestCase
{
public function testQueryWhileDeallocatingStatement(): void
{
if (EventLoop::getDriver()->getHandle() instanceof \EvLoop) {
$this->markTestSkipped("ext-pgsql is not compatible with pecl-ev");
}

$timeout = EventLoop::delay(5, static fn () => throw new \Error('The result of a query was lost'));

try {
$connection = PgSqlConnection::connect(
PostgresConfig::fromString('host=localhost user=postgres password=postgres'),
);

for ($i = 0; $i < 20; ++$i) {
$statement = $connection->prepare("SELECT $i AS value");
self::assertSame($i, $statement->execute()->fetchRow()['value']);
unset($statement); // Deallocates the statement in another fiber.

delay(0); // Lets the DEALLOCATE be sent.

self::assertSame($i, $connection->query("SELECT $i AS value")->fetchRow()['value']);
}

$connection->close();
} finally {
EventLoop::cancel($timeout);
}
}
}
Loading