Fix bug swoole#5635

ext-src/swoole_pgsql.cc

@@ -23,30 +23,48 @@
 
 #ifdef SW_USE_PGSQL
 
-using swoole::Reactor;
 using swoole::Coroutine;
+using swoole::Reactor;
 using swoole::coroutine::Socket;
 using swoole::coroutine::translate_events_to_poll;
 
 static SW_THREAD_LOCAL bool swoole_pgsql_blocking = true;
 
-static int swoole_pgsql_socket_poll(PGconn *conn, swEventType event, double timeout = -1) {
+static int swoole_pgsql_socket_poll(PGconn *conn, swEventType event, double timeout = -1, bool check_nonblock = false) {
     if (swoole_pgsql_blocking) {
         struct pollfd fds[1];
         fds[0].fd = PQsocket(conn);
         fds[0].events |= translate_events_to_poll(event);
 
         int result = 0;
         do {
-             result = poll(fds, 1, timeout);
+            result = poll(fds, 1, timeout);
         } while (result < 0 && errno == EINTR);
 
         return result > 0 ? 1 : errno == ETIMEDOUT ? 0 : -1;
     }
 
     Socket sock(PQsocket(conn), SW_SOCK_RAW);
     sock.get_socket()->nonblock = 1;
-    bool retval = sock.poll(event, timeout);
+
+    bool retval = false;
+    while (true) {
+        retval = sock.poll(event, timeout);
+
+        if (!(check_nonblock && event == SW_EVENT_READ)) {
+            break;
+        }
+
+        if (PQconsumeInput(conn) == 0) {
+            retval = false;
+            break;
+        }
+
+        if (PQisBusy(conn) == 0) {
+            break;
+        }
+    }
+
     sock.move_fd();
     return retval ? 1 : sock.errCode == ETIMEDOUT ? 0 : -1;
 }
@@ -68,7 +86,8 @@ static int swoole_pgsql_flush(PGconn *conn) {
 
 static PGresult *swoole_pgsql_get_result(PGconn *conn) {
     PGresult *result, *last_result = nullptr;
-    int poll_ret = swoole_pgsql_socket_poll(conn, SW_EVENT_READ);
+    // PQgetResult will block the process; it is necessary to forcibly check if the data is ready.
+    int poll_ret = swoole_pgsql_socket_poll(conn, SW_EVENT_READ, -1, true);
     if (sw_unlikely(poll_ret == SW_ERR)) {
         return nullptr;
     }

tests/swoole_pdo_pgsql/bug_5635.phpt

@@ -0,0 +1,64 @@
+--TEST--
+swoole_pdo_pgsql: Github bug #5635
+--SKIPIF--
+<?php require __DIR__ . '/../include/skipif.inc'; ?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+require __DIR__ . '/pdo_pgsql.inc';
+
+use Swoole\Coroutine;
+use Swoole\Coroutine\WaitGroup;
+use Swoole\Coroutine\Channel;
+use function Swoole\Coroutine\run;
+
+ini_set("memory_limit", "-1");
+
+$pdo = pdo_pgsql_test_inc::create();
+$pdo->exec('create table bug_5635 (id int, data varchar(1024));');
+$pdo->exec(<<<EOL
+DO $$
+BEGIN
+FOR i IN 1..4000000 LOOP
+INSERT INTO bug_5635(id, data) VALUES (i, 'data' || i);
+END LOOP;
+END $$;
+EOL);
+
+Coroutine::set(['hook_flags' => SWOOLE_HOOK_PDO_PGSQL]);
+run(function() {
+    $waitGroup = new WaitGroup();
+    $channel = new Channel(1);
+
+    Coroutine::create(function() use ($waitGroup, $channel) {
+        $start = time();
+        $waitGroup->add();
+        $pdo = pdo_pgsql_test_inc::create();
+        $stmt = $pdo->query("select * from bug_5635;");
+        $data = $stmt->fetchAll();
+        Assert::true(count($data) == 4000000);
+        $channel->push($data ?? [], 10);
+        $waitGroup->done();
+        echo 'DONE' . PHP_EOL;
+    });
+
+    Coroutine::create(function() use ($waitGroup, $channel) {
+        $waitGroup->add();
+        $result = $channel->pop(1.5);
+        if (!$result) {
+            echo 'channel pop timeout' . PHP_EOL;
+        }
+        $waitGroup->done();
+    });
+
+    var_dump(1);
+    Coroutine::sleep(1);
+    var_dump(2);
+    $waitGroup->wait();
+});
+?>
+--EXPECTF--
+int(1)
+int(2)
+channel pop timeout
+DONE