Strange behavior EvIo

Issue #26 invalid
Alex Sky
created an issue

Здравствуйте.

Я напишу на русском, потому что сложно сформулировать на английском, надеюсь вы не против :)

Мы используем EvIo, для получения асинхронных ответов на сокете PostgreSQL клиента.

Так же мы используем EvIo для получения HTTP запросов.

В результате у нас есть два объекта EvIo - WatcherDB и WatcherHTTP .

Есть проблема с WatcherDB, после реконекта PostgreSQL клиента, WatcherDB перестает вызывать callback обработчик, когда в сокет PostgreSQL клиента приходят новые данные для чтения.

Но самое интересное, что WatcherDB после реконекта PostgreSQL клиента начинает странно себя вести, он вызывает обработчик из WatcherDB когда приходят новые данные в сокете WatcherHTTP, такое ощущения что после реконекта PostgreSQL клиента, EvIo перепутывает обработчики и сокеты между собой.

Если нужно я могу написать упрощенный тест код РНР, но он будет работать если есть PostgreSQL сервер и PostgreSQL клиент

Возможно это важно, после реконекта PostgreSQL клиента, сокет PostgreSQL клиента имеет тот же Resource id т.е. он не меняется.

Comments (14)

  1. Ruslan Osmanov repo owner

    Здравствуйте.

    Давайте посмотрим на PHP-код(желательно рабочий), как именно объект EvIo связан с pqconn::resetAsync(). Я установлю и сервер PostgreSQL, и ext-pq.

  2. Alex Sky reporter

    Уже третий час пытаюсь воспроизвести ошибку в тест скрипте, но пока что не получается.

    В реальном приложении много разных сокетов, все работает на корутинах, воспроизвести такие же условия в тест скрипте сложно, но я сделаю.

  3. Ruslan Osmanov repo owner

    Предполагаю, что после $c->resetAsync() сокет соединения socket = PQsocket(obj->intern->conn) принимает новое значение. Интересно, обновляется ли при этом число, обёрнутое в $c->socket. Обработчик read_property свойства socket равен NULL, т.е. в свойстве socket мы получаем то, что записали явно посредством обработчика "write_property". Похоже что свойство сокета обновляется только в конструкторе соединения.

    С другой стороны, файловый дескриптор сокета, хранящийся в EvIo, не меняется - мы инициализируем его в конструкторе EvIo. Конструктор EvIo получает файловый дескриптор(число) из первого аргумента($fd) и сохраняет его. Далее, объект EvIo повсюду использует именно это число - он не хранит zval. В частности, обработчик "read_property" свойства EvIo::$fd оборачивает сокет в PHP stream(который на самом деле не многим лучше числа). Возможно, после реконнекта, следует пересоздавать объекты EvIo, либо добавить в EvIo метод, обновляющий $fd.

  4. Alex Sky reporter

    Интересно, обновляется ли при этом число, обёрнутое в $c->socket.

    Нет, число не обновляется, в моем случаи после вызовов $c->resetAsync() всегда равен Resource id #21.

    Возможно, после реконнекта, следует пересоздавать объекты EvIo, либо добавить в EvIo метод, обновляющий $fd.

    Да, верно, мы это делаем всегда, там даже достаточно вызывать EvIo->stop ... EvIo->start

    Но, проблема которую я описал, возникают если $c->resetAsync() и последующие EvIo->stop ... EvIo->start вызвать из обработчика другого объекта EvIo.

    Use case нашего приложения, если соединения с PostgreSQL оборвалось и в этот момент нет HTTP запросов которые ждали ответов из PostgreSQL, мы не вызываем сразу $c->resetAsync(), мы его вызовем когда придет новый HTTP запрос которому нужно работать с PostgreSQL, таким образом $c->resetAsync() вызывается из EvIo обработчика HTTP сокета, вот тогда появляется странное поведения EvIo про которое я говорил.

    Мы пока что решили эту проблему просто, вызываем $c->resetAsync() из EvTimer когда пропадает соединения с PostgreSQL. Но я постараюсь воспроизвести условия и сделать простой пример.

  5. Alex Sky reporter

    Возможно это важно, наш РНР демон, запускается через systemd.socket активацию, systemd запускает РНР процес и сам добавляет в РНР процес все fd (HTTP сокеты) т.е. РНР процес изначально имеет больше fd чем обычно. Мы открываем эти fd вот так $fd = socket_import_stream(fopen("php://fd/$i", 'r+b')) и создаем наблюдателей EvIo

  6. Alex Sky reporter

    Я вроде воспроизвел проблему, в 60 строке кода в reset_bug.php вызывается $db->resetAsync() он отрабатывает коректно, создается новое соединения, но EvIo не будет обрабатывать новые события в этом сокете.

    В файле reset_ok.php изменен алгоритм вызова $db->resetAsync() он вызывается в другой итерации STDIN и тогда все работает нормально.

  7. Ruslan Osmanov repo owner

    По-моему всё работает как положено.

    На сколько я понял из документации ext-pq, любая асинхронная операция требует поллинга. В скриптах reset_bug.php и reset_ok.php поллинг выполняется некорректно - $db->poll() вызывается без проверок на $db->busy и без проверок на доступность сокета на запись.

    К слову, колбек $dbIo срабатывает. Скрипт не выводит данных из базы как раз из-за отсутствия корректной обработки поллинга. Следующий скрипт должен работать как ожидалось.

    <?php
    use pq\Connection;
    
    $loop = EvLoop::defaultLoop();
    $db   = new Connection('user=postgres application_name=Test', Connection::ASYNC);
    
    
    function prompt_reconnect() {
        echo "Press Enter to reset connection...\n";
    }
    
    
    $dbIo = $loop->io($db->socket, Ev::READ, function($w) use($db) {
        try {
            echo "# DBIO: Polling\n";
            switch ($state = $db->poll()) {
            case Connection::POLLING_READING:
                // we should wait for the stream to be read-ready
                echo "# DBIO: POLLING_READING\n";
                $w->set($db->socket, Ev::READ);
                break;
    
            case Connection::POLLING_WRITING:
                // we should wait for the stream to be write-ready
                echo "# DBIO::POLLING_WRITING\n";
                $w->set($db->socket, Ev::WRITE);
                break;
    
            case Connection::POLLING_FAILED:
                fprintf(STDERR, "# DBIO: Connection failed: " . $db->errorMessage);
                $w->stop();
                break;
    
            case Connection::POLLING_OK:
                echo "# DBIO: Connection OK\n";
                $w->stop();
                $w->set($db->socket, Ev::WRITE);
    
                if (!$db->busy) {
                    echo "# DBIO: Running async query\n";
                    $db->execAsync("SELECT 'OK'");
                    $w->start();
                } else {
                    echo "# DBIO: Fetching result\n";
                    $r = $db->getResult();
                    printf("# DBIO: READ: %s\n", $r ? $r->fetchRow()[0] : '(none)');
                    prompt_reconnect();
                }
                break;
    
            default:
                printf("Unknown polling state: %s. Stopping socket watcher.\n",
                    var_export($state, true));
                $w->stop();
                break;
            }
        } catch(Throwable $e) {
            echo "Error: ".$e->getMessage()."\r\n";
        }
    });
    
    
    $stdIo = $loop->io(STDIN, Ev::READ, function() use($db, $dbIo) {
        try {
            fgets(STDIN);
    
            echo "# STDINIO: Restarting DB server\n";
            exec('sudo systemctl restart postgresql-9.5');
    
            echo "# STDINIO: Running async reset\n";
            $db->resetAsync();
    
            $dbIo->set($db->socket, Ev::READ);
            $dbIo->start();
    
            // Trigger DBIO read callback
            $db->poll();
        } catch(Throwable $e) {
            echo "Error: ".$e->getMessage()."\r\n";
        }
    });
    
    prompt_reconnect();
    $loop->run();
    
  8. Log in to comment