PHP assíncrono

Assincronia é a capacidade de um sistema de software de não bloquear o thread principal de execução.
Uma operação assíncrona é uma operação que não bloqueia o fluxo de execução de um programa até sua conclusão.

Operações de bloqueio

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;
}

• Sistema de arquivos : fwrite, file_get_contents.
• Bases de dados : PDOConnection, RedisClient. Quase todas as extensões para conectar um banco de dados funcionam no modo de bloqueio por padrão.
• Processos : exec, system, proc_open. Essas são operações de bloqueio, pois todo o trabalho com processos é criado por meio de chamadas do sistema.
• Trabalhando com stdin / stdout : readline, echo, print.

Cliente SQL assíncrono

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));
}

Promessa

Promise é um objeto wrapper sobre o resultado de uma operação assíncrona. Além disso, o resultado da operação ainda é desconhecido para nós.

Como a promessa funciona

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);
}

Cliente SQL assíncrono

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();
}

• defina o retorno de chamada e obtenha os affected rowsque retornam para nós PDOConnection;
• lidar com o erro, adicione ao log;
• Repita a consulta se o servidor SQL responder com um erro.

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

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

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

Loop de eventos

• O cliente diz ao Event Loop que está interessado em algum tipo de soquete.
• O Loop de Eventos pesquisa o SO por meio de uma chamada do sistema stream_select: o soquete está pronto, todos os dados gravados, são os dados provenientes do outro lado.
• Se o SO reportar que o soquete não está pronto, bloqueado, o Event Loop repetirá o loop.
• Quando o sistema operacional notifica que o soquete está pronto, o Event Loop retorna o controle ao cliente e ativa ( 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();
}

Geradores

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...'));

Geradores e Promessa

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);
    };

Multitarefa cooperativa

Processamento de solicitações em PHP

• Tratamento simples de erros . Se algo der errado e uma das solicitações falhar, não precisamos fazer nada - a próxima virá, e isso não afetará seu trabalho.
• Nós não pensamos em memória . Não precisamos limpar ou monitorar a memória. Na próxima solicitação, toda a memória será limpa.

• Limite o número de processos . Se tivermos 50 threads de FPM no servidor, assim que a 51ª solicitação chegar, ele aguardará até que um dos threads fique livre.
• Custos para troca de contexto . O SO alterna solicitações entre fluxos FPM. Essa operação no nível do processador é denominada Context Switch. É caro e executa um grande número de medidas. É necessário salvar todos os registros, a pilha de chamadas, tudo o que está no processador, mudar para outro processo, carregar seus registros e sua pilha de chamadas, executar algo lá novamente, alternar novamente, salvar novamente ... Por um longo tempo.

Servidor HTTP assíncrono

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();
});

Problemas

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);
    }
}

Por que tudo isso?

Para velocidade e desempenho.

Em 13 de maio de 2020, nos reuniremos pela segunda vez no PHP Rússia para discutir a linguagem, bibliotecas e estruturas, maneiras de aumentar a produtividade e as armadilhas das soluções de hype. Aceitamos os quatro primeiros relatórios , mas a Chamada de Trabalhos ainda está chegando. Inscreva- se se desejar compartilhar sua experiência com a comunidade.