PHP asinkron

Asynchrony adalah kemampuan sistem perangkat lunak untuk tidak memblokir utas eksekusi.
Operasi asinkron adalah operasi yang tidak memblokir aliran eksekusi program sampai selesai.

Operasi pemblokiran

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

• Sistem File : fwrite, file_get_contents.
• Basis data : PDOConnection, RedisClient. Hampir semua ekstensi untuk menyambungkan kerja basis data dalam mode pemblokiran secara default.
• Proses : exec, system, proc_open. Ini memblokir operasi, karena semua pekerjaan dengan proses dibangun melalui panggilan sistem.
• Bekerja dengan stdin / stdout : readline, echo, print.

Asynchronous SQL Client

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

Janji

Promise adalah objek pembungkus dari hasil operasi asinkron. Selain itu, hasil operasi masih belum diketahui oleh kami.

Bagaimana Janji Bekerja

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

Asynchronous SQL Client

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

• atur callback dan dapatkan affected rowsyang kembali kepada kami PDOConnection;
• menangani kesalahan, tambahkan ke log;
• Coba lagi permintaan jika server SQL merespons dengan kesalahan.

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

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

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

Perulangan acara

• Klien memberi tahu Event Loop bahwa ia tertarik pada beberapa jenis soket.
• Perulangan Kejadian polling OS melalui system call stream_select: apakah soket siap, semua data tertulis, adalah data yang berasal dari sisi lain.
• Jika OS melaporkan bahwa soket tidak siap, diblokir, maka Event Loop mengulangi loop.
• Ketika OS memberitahukan bahwa soket sudah siap, Event Loop mengembalikan kontrol ke klien dan mengaktifkan ( resolveatau reject) Janji.

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

Generator

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

Coroutine

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

Generator dan Janji

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

Multitasking kooperatif

Meminta pemrosesan dalam PHP

• Penanganan kesalahan sederhana . Jika ada yang tidak beres dan salah satu permintaan jatuh, kita tidak perlu melakukan apa pun - yang berikutnya akan datang, dan ini tidak akan memengaruhi pekerjaannya.
• Kami tidak memikirkan ingatan . Kami tidak perlu membersihkan atau memantau memori. Atas permintaan berikutnya, semua memori akan dihapus.

• Batasi jumlah proses . Jika kami memiliki 50 utas FPM di server, maka segera setelah permintaan ke-51 tiba, itu akan menunggu sampai salah satu utas menjadi bebas.
• Biaya untuk Beralih Konteks . OS mengalihkan permintaan di antara aliran FPM. Operasi tingkat prosesor ini disebut Context Switch. Itu mahal dan menjalankan sejumlah besar tindakan. Penting untuk menyimpan semua register, tumpukan panggilan, semua yang ada di prosesor, kemudian beralih ke proses lain, muat register dan tumpukan panggilannya, lakukan sesuatu di sana lagi, beralih lagi, simpan lagi ... Untuk waktu yang lama.

Server HTTP Asinkron

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

Masalah

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

Mengapa semua ini?

Untuk kecepatan dan kinerja.

13 Mei 2020, kami akan berkumpul untuk kedua kalinya di PHP Rusia untuk membahas bahasa, perpustakaan, dan kerangka kerja, cara untuk meningkatkan produktivitas dan perangkap solusi hype. Kami telah menerima 4 laporan pertama , tetapi Call for Papers masih akan datang. Terapkan jika Anda ingin berbagi pengalaman dengan komunitas.