PHP asynchrone

L'asynchronie est la capacité d'un système logiciel à ne pas bloquer le thread principal d'exécution.
Une opération asynchrone est une opération qui ne bloque pas le flux d'exécution d'un programme tant qu'elle n'est pas terminée.

Opérations de blocage

public function update(User $user)
{
    try {
        $sql = 'UPDATE users SET ...';
        return $this->connection->execute($sql, $user->data());
    } catch (\PDOException $error) {
        log($error->getMessage());
    }

    return 0;
}

• Système de fichiers : fwrite, file_get_contents.
• Bases de données : PDOConnection, RedisClient. Presque toutes les extensions pour connecter une base de données fonctionnent par défaut en mode blocage.
• Processus : exec, system, proc_open. Ce sont des opérations de blocage, car tout travail avec des processus est construit via des appels système.
• Travailler avec stdin / stdout : readline, echo, print.

Client SQL asynchrone

public function execAsync(string $query, array $params = [])
{
    $socket = stream_socket_client('127.0.0.1:3306', ...);

    stream_set_blocking($socket, false);

    $data = $this->packBinarySQL($query, $params);
    
    socket_write($socket, $data, strlen($data));
}

Promettre

Promise est un objet wrapper sur le résultat d'une opération asynchrone. De plus, le résultat de l'opération nous est encore inconnu.

Fonctionnement de Promise

interface Promise
{
    const
        STATUS_PENDING = 0,
        STATUS_RESOLVED = 1,
        STATUS_REJECTED = 2
    ;

    public function onResolve(callable $callback);
    public function onReject(callable $callback);
    public function resolve($data);
    public function reject(\Throwable $error);
}

Client SQL asynchrone

public function execAsync(string $query, array $params = [])
{
    $deferred = new Deferred;

    $socket = stream_socket_client('127.0.0.1:3306', ...);
    stream_set_blocking($socket, false);

    $data = $this->packBinarySQL($query, $params);
    socket_write($socket, $data, strlen($data));

    return $deferred->promise();
}

• définissez le rappel et obtenez ceux affected rowsqui nous reviennent PDOConnection;
• gérer l'erreur, ajouter au journal;
• Relancez la requête si le serveur SQL répond avec une erreur.

$promise = $this->execAsync($sql, $user->data());

$promise->onResolve(function (int $rows) {
    echo "Affected rows: {$rows}";
});

$promise->onReject(function (\Throwable $error) {
    log($error->getMessage());
});

Boucle d'événement

• Le client indique à Event Loop qu'il est intéressé par une sorte de socket.
• La boucle d'événements interroge le système d'exploitation via un appel système stream_select: le socket est-il prêt, toutes les données sont-elles écrites, les données proviennent-elles de l'autre côté.
• Si le système d'exploitation signale que le socket n'est pas prêt, bloqué, alors la boucle d'événement répète la boucle.
• Lorsque le système d'exploitation notifie que le socket est prêt, la boucle d'événements renvoie le contrôle au client et active ( resolveou reject) Promise.

public static function run()
{
    while (true) {
        stream_select($readSockets, $writeSockets, null, 0);
        
        foreach ($readSockets as $i => $socket) {
            call_user_func(self::readCallbacks[$i], $socket);
        }

        // Do same for write sockets
    }
}

public function execAsync(string $query, array $params = [])
{
    $deferred = new Deferred;
    ...
    Loop::onReadable($socket, function ($socket) use ($deferred) {
        $deferred->resolve(socket_read($socket));
    });

    return $deferred->promise();
}

• JavaScript setTimeout setInterval — . N . Event Loop .
• Event Loop . process control, .

Event Loop

ReactPHP

$connection = (new ConnectionFactory)->createLazyConnection();

$promise = $connection->query('UPDATE users SET ...');
$promise->then(
    function (QueryResult $command) {
        echo count($command->resultRows) . ' row(s) in set.';
    },
    function (Exception $error) {
        echo 'Error: ' . $error->getMessage();
    });

    function (QueryResult $command) {
        echo count($command->resultRows) . ' row(s) in set.';
    },

    function (Exception $error) {
        echo 'Error: ' . $error->getMessage();
    });

$promise
    ->then(function ($data) {
        return new Promise(...);
    })
    ->then(function ($data) {
        ...
    }, function ($error) {
        log($error);
    })
    ...

Amphp

$pool = Mysql\pool("host=127.0.0.1 port=3306 db=test");

try {
    $result = yield $pool->query("UPDATE users SET ...");

    echo $result->affectedRows . ' row(s) in set.';
} catch (\Throwable $error) {
    echo 'Error: ' . $error->getMessage();
}

Générateurs

function generator($counter = 1)
{
    yield $counter++;

    echo "A";

    yield $counter;

    echo "B";

    yield ++$counter;
}

$generator = generator(1);

foreach ($generator as $value) {
    echo $value;
}

while ($generator->valid()) {
    echo $generator->current();

    $generator->next();
}

Coroutines

function printer() {  
    while (true) {     
        echo yield;       
    }                             
}                                

$print = printer();
$print->send('Hello');
$print->send(' PHPRussia');
$print->send(' 2019');
$print->send('!');

function printer() {
    try {
        echo yield;
    } catch (\Throwable $e) {
        echo $e->getMessage();
    }
}

printer()->throw(new \Exception('Ooops...'));

Générateurs et promesse

class Generator
{
    public function send($data);
    public function throw(\Throwable $error);
}

class Promise
{
    public function resolve($data);
    public function reject(\Throwable $error);
}

function recoil(\Generator $generator)
{
    $promise = $generator->current();

    $promise->onResolve(function($data) use ($generator) {
        $generator->send($data);
        recoil($generator);
    };

    $promise->onReject(function ($error) use ($generator) {
        $generator->throw($error);
        recoil($generator);
    });
}

    $promise->onResolve(function($data) use ($generator) {
        $generator->send($data);
        recoil($generator);
    };

Multitâche coopératif

Traitement des demandes en PHP

• Gestion simple des erreurs . Si quelque chose a mal tourné et qu'une des demandes est tombée, nous n'avons rien à faire - la prochaine viendra, et cela n'affectera pas son travail.
• Nous ne pensons pas à la mémoire . Nous n'avons pas besoin de nettoyer ou de surveiller la mémoire. À la prochaine demande, toute la mémoire sera effacée.

• Limitez le nombre de processus . Si nous avons 50 threads FPM sur le serveur, dès que la 51e requête arrive, il attendra que l'un des threads devienne libre.
• Coûts pour le changement de contexte . Le système d'exploitation bascule les demandes entre les flux FPM. Cette opération au niveau du processeur est appelée Context Switch. Il coûte cher et exécute un grand nombre de mesures. Il faut sauvegarder tous les registres, la pile d'appels, tout ce qui se trouve dans le processeur, puis passer à un autre processus, charger ses registres et sa pile d'appels, y refaire quelque chose, basculer à nouveau, sauvegarder encore ... Longtemps.

Serveur HTTP asynchrone

Loop::run(function () {
    $app = new Application();
    $app->bootstrap();

    $sockets = [Socket\listen('0.0.0.0:80')];

    $server = new Server($sockets, new CallableRequestHandler(
        function (Request $request) use ($app) {
            $response = yield $app->dispatch($request);

            return new Response(Status::OK, [], $response);
        })
    );

    yield $server->start();
});

Problèmes

class UserRepository
{
    public function find(int $id): \Generator
    {
        $data = yield $this->db->query('SELECT ...', $id);

        return User::fill($data);
    }
}

PHP 8

class UserRepository
{
    public function find(int $id): Promise<User>
    {
        $data = await $this->db->query('SELECT ...', $id);

        return User::fill($data);
    }
}

class UserRepository
{
    public function find(int $id): Promise<User>
    {
        $data = yield $this->db->query('SELECT ...', $id);

        return User::fill($data);
    }
}

Pourquoi tout ça?

Pour la vitesse et les performances.

Le 13 mai 2020, nous nous réunirons pour la deuxième fois à PHP Russie pour discuter du langage, des bibliothèques et des cadres, des façons d'augmenter la productivité et des pièges des solutions de battage médiatique. Nous avons accepté les 4 premiers rapports , mais l'appel à communications arrive toujours. Postulez si vous souhaitez partager votre expérience avec la communauté.