Делаем вебсокеты на PHP с нуля. Часть 2. IPC

После написания моей предыдущей статьи Делаем вебсокеты на PHP с нуля я понял, что у сообщества есть некоторый интерес к поднятой мною теме.

В прошлой статье я обещал, что опишу:

• запуск нескольких процессов для обработки соединений
• межпроцессное взаимодействие
• разделение процессов мастер-воркер
• проксирование вебсокетов с помощью nginx
• запуск из консоли
• интеграция с вашим фреймворком на примере yii
• демонстрация

И, как обычно, — получившийся код и ссылка на демонстрационный чат в конце статьи.

Запуск нескольких процессов для обработки соединений

Для работы простого сервера вебсокетов достаточно одного процесса, но чтобы увеличить количество одновременных соединений (и обойти ограничение 1024 одновременных соединения), а также для использования ресурсов всего процессора (а не только одного ядра), необходимо, чтобы сервер вебсокетов использовал несколько процессов (оптимально — количество процессов = количество ядер процессора).

Для запуска нескольких процессов мы будем использовать функцию pcntl_fork(). Она создаёт новый процесс (дочерний), который является практически полной копией процесса-родителя, выполняющего этот вызов.
После вызова pcntl_fork() алгоритм разветвляется: в случае успешного выполнения функции pcntl_fork() она возвращает PID дочернего процесса родительскому, а NULL дочернему. Если создание форка закончилось неудачей, функция pcntl_fork() возвращает значение −1).

$pid = pcntl_fork(); //делаем форк

//далее весь код будет выполняться в обоих процессах

if ($pid == -1) {
    // Не удалось создать дочерний процесс
} elseif ($pid) {
    // Этот код выполнится родительским процессом
} else {
    // А этот код выполнится дочерним процессом, его PID можно узнать с помощью функции getmypid()
}

Про отличие родительского процесса от дочернего можно почитать на википедии.

Мы можем в цикле создавать столько дочерних процессов, сколько нам необходимо:

$childs = array();

for ($i=0; $i<4; $i++) {
    $pid = pcntl_fork(); //создаём форк

    if ($pid == -1) {
        die("error: pcntl_fork");
    } elseif ($pid) { //родительский процесс
        $childs[] = $pid; //заполняем массив дочерними PID, они нам ещё пригодятся :)
    } else { //дочерний процесс
        break; //выходим из цикла, чтобы дочерние процессы создавались только из родителя
    }
}

Межпроцессное взаимодействие

Для взаимодействия между родительским и дочерним процессом мы будем использовать сокеты, а именно связанные сокеты:
Функция stream_socket_pair() создаёт пару связанных неразличимых потоковых сокетов. Таким образом мы можем писать в один сокет, а считывать данные из второго.

$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); //получаем массив из связанных сокетов

fwrite($pair[0], 'тест'); //пишем в первый сокет
fread($pair[1], mb_strlen('тест')); //читаем из второго

Теперь совмещаем этот код с форками и получаем:

$pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); //создаём связанные сокеты

$pid = pcntl_fork(); //делаем форк

//далее весь код будет выполняться в обоих процессах

if ($pid == -1) {
    die("error: pcntl_fork");
} elseif ($pid) { //родительский процесс
    fclose($pair[0]); //закрываем один из сокетов в родителе
    $child = $pair[1]; //второй будем использовать для связи с потомком
} else { //дочерний процесс
    fclose($pair[1]); //закрываем второй из сокетов в потомке
    $parent = $pair[0]; //первый будем использовать для связи с родителем
}

Итоговый код для создания множества дочерних процессов:

$parent = null;
$childs = array();

for ($i=0; $i<5; $i++) {
    $pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); //создаём связанные сокеты

    $pid = pcntl_fork(); //создаём форк

    if ($pid == -1) {
        die("error: pcntl_fork");
    } elseif ($pid) { //родительский процесс
        fclose($pair[0]); //закрываем один из сокетов в родителе
        $childs[] = $pair[1]; //второй будем использовать для связи с потомком
    } else { //дочерний процесс
        fclose($pair[1]); //закрываем второй из сокетов в потомке
        $parent = $pair[0]; //первый будем использовать для связи с родителем
        break; //выходим из цикла, чтобы дочерние процессы создавались только из родителя
    }
}

В результате работы этого кода в родителе массив $childs будет содержать в себе все сокеты для связи с потомками, а потомки для связи с родителем будут использовать $parent.

Разделение процессов на мастера и воркеров

Так как дочерние процессы в нашей реализации не связаны друг с другом напрямую и могут взаимодействовать только через родителя, то целесообразно разделение обязанностей между родителем и потомками:

• родитель будет отвечать за взаимодействие между дочерними процессами (будет мастером)
• дочерние процессы будут выполнять всю работу (будут воркерами)

Также воркер у нас будет заниматься пересылкой сообщений из скриптов со страниц сайта или из крона. Для этого мы создадим дополнительный сокет, и добавим его в массив, прослушиваемых сокетов. Например, можно создать unix-сокет:

$service = stream_socket_server('unix:///tmp/websocket.sock', $errorNumber, $errorString);

Проксирование вебсокетов с помощью nginx

Nginx поддерживает проксирование вебсокетов начиная с версии 1.3.13. Благодаря nginx можно обрабатывать соединения к серверу вебсокетов на том же порту, что и сайт, а также ограничить количество открытых вебсокетов с одного ip и другие полюбившиеся вам плюшки.

Пример nginx-конфига, который это позволяет:

limit_conn_zone $binary_remote_addr zone=perip:10m;

server {
    listen 5.135.163.218:80;
    server_name sharoid.ru;

    location / {
        limit_conn perip 5; #делаем ограничение 5 вебсокетов на 1 ip
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_read_timeout 3600s; #увеличиваем таймаут для вебсокетов
    }
}

Запуск из консоли

Выполняем команду php websocket.php или ./websocket.php (предварительно дав права на выполнение)
Если использовать nohup, например, nohup ./websocket.php &, то скрипт продолжит работать после закрытия консоли.

По-умолчанию есть два ограничения количества соединений на один процесс.

• Первое — на уровне операционной системы: в одном процессе нельзя открыть более чем 1024 соединения. Чтобы обрабатывать больше одновременных соединений, выполните команду: ulimit -n 65535, а если у пользователя недостаточно привилегий, то sudo sh -c "ulimit -n 65535 && exec su $LOGNAME". Текущее значение можно посмотреть используя команду ulimit -n
• Второе — у функции stream_select(): она не принимает больше чем 1024 соединения. Здесь всё сложнее — нужно перекомпилировать php c увеличенным FD_SETSIZE

Как я уже писал, эти ограничения можно обойти, используя дочерние процессы (воркеры).

Интеграция с вашим фреймворком на примере yii

Так как наш мастер прослушивает дополнительный сокет для связи с нашими скриптами (в примере выше был unix:///tmp/websocket.sock), мы можем в любом месте нашего сайта или в кроне соединиться с этим сокетом и отправить сообщение, которое мастер разошлёт всем воркерам, а они, в свою очередь, все клиентам:

$service = stream_socket_client ('unix:///tmp/websocket.sock', $errno, $errstr);
fwrite($service, 'всем привет');

С использованием компонента yii это будет выглядеть вот так:

Yii::app()->websocket->send('всем привет');

Подробнее для yii

Скачиваем экстеншн, кладём его в папку extensions/websocket
В папку components кладём Websocket.php, WebsocketMasterHandler.php и WebsocketWorkerHandler.php из папки sample/yii.
В папку commands кладём из WebsocketCommand.php из папки sample/yii.
В конфиги main.php и console.php вставляем в секцию components:

'websocket' => array(
    'class' => 'Websocket',
    //'websocket' => 'tcp://127.0.0.1:8000',
    //'localsocket' => 'tcp://127.0.0.1:8001',// unix:///tmp/mysock
    //'workers' => 1
),

В конфиг console.php также вставляем в секцию import:

'ext.websocket.*'

Демонстрация

Демонстрационный чат 2.0 (добавлен список пользователей, добавлено ограничение: 1 сообщение в секунду с одного IP)
В нём были использованы описанные выше функции, а также исправлены недостатки, выявленные после публикации предыдущей статьи.
Демонстрационный чат 1.0 (без списка пользователей, без ограничений)

Все исходники я оформил в виде библиотеки и выложил на github

Update: Если сообществу интересна эта тема, то следующая статья будет про то как сделать простую игру, в которой все участники будут находиться на одном игровом поле и взаимодействовать друг с другом в реальном времени (демка уже почти готова).

Третья часть статьи: От чата до игры: Battle City