Прием потока данных через TCP сокет на PHP?

Question

[SSar]

На nix сервере постоянно запущен php скрипт, принимающий данные через Интернет от удаленных датчиков по GPRS/EDGE каждые несколько секунд.



В Сети нашлось множество примеров работы с сокетами в php однако у всех них общий недостаток — через несколько десятков минут непрерывной работы скрипт «зависает» и сокеты порожденные им висят в статусе FIN_WAIT_1.



Я временно решил данную проблему установив режим сокета в неблокирующий режим и включив SO_LINGER ('l_onoff'=>1, 'l_linger'=>0), т.е. грубо закрываю сокет не дожидаясь ответа после получения данных.



Однако данная мера вызывает реконнект дорогостоящего GPRS соединения ибо операторы округляют трафик по 100кб обычно, т.е. факт соединения уже 100кб считай утекло. Хотелось бы увидеть реально рабочий пример как можно НЕ разрывая соединения стабильно принимать данные от датчиков и при этом не вызывать «зависание» сокетов и самого скрипта.



Ниже привожу часть кода с учетом моих поправок по приему данных на PHP:

$socket = socket_create_listen($port, SOMAXCONN) ;
socket_set_nonblock($socket);
$arrOpt = array('l_onoff'=>1, 'l_linger'=>0);

while(1) {
	usleep(100000);
	if((time()-$time) >= 60) { $time = time(); UpdatePID($pid); }
	$client = @socket_accept($socket);
	if(!$client) continue;
	socket_set_option($client, SOL_SOCKET, SO_LINGER, $arrOpt);
	$data = socket_read($client, 4096);
	$data = trim($data);
	Save2DB($data);
	socket_close($client);
	if(strtolower($input)=="exit") break; }

socket_close($socket);

UpdatePID — обновляет pid файл, необходимый для предотвращения повторного запуска скрипта.

Save2DB — парсит и пишет данные в базу.



С использованием socket_set_option(..SO_KEEPALIVE...) никто не работал?

Answer 1

[galaxy]

Вы что-то не то делаете с сокетами.
FIN_WAIT_1 говорит о том, что ваш сервер соединение закрыл со своей стороны и ждет закрытия со стороны клиента. Имхо, происходит такое потому, что, сделав socket_read(), вы делаете socket_accept() следующему соединению, переменная $client (предыдущий сокет) попадает в сборщик мусора, который пытается корректно завершить соединение. Нельзя так выбрасывать заакцепченный сокет, надо продолжать пытаться читать из него данные, одновременно проверяя, нет ли новых запросов на соединение. Можно организовать это и через sleep'ы, но классический подход состоит в использовании socket_select(). Следуя вашему подходу, можно написать что-то в этом духе:

while(1) {
  usleep(100000);
  if((time()-$time) >= 60) { $time = time(); UpdatePID($pid); }
  $client = @socket_accept($socket);
  if($client) {
    $clients[] = $client;
  }
  foreach($clients as $i => $c) {
    $data = socket_read($client, 4096);
    if ($data === false) { // error or closed connection
      socket_close($c);
      unset($clients[$i]);
    } elseif (strlen($data)) {
      Save2DB(trim($data));
    }
  }

  ...

Comments

[SSar]

хм… после socket_read($client, 4096) дальше по коду у меня идет нормальный socket_close($client), но видимо он не успевает закрыть сокет до следующего соединения (судя по мануалу по сокетам в php такое бывает).
понаблюдал за сокетами через netstat — чаще всего зависают в статусе TIME_WAIT, до FIN_WAIT_1 дело доходит когда незакрытых сокетов становится несколько.

[SSar]

А можно посмотреть на пример с использованием socket_select(), если он Вам хорошо знаком?
PS Спс за участие.

[SSar]

и еще: в вашем примере после строки foreach нужно наверное socket_read($c...) использовать вместо socket_read($client)?
и еще reset($clients) перед циклом чтоб указатель на 0й элемент массива установить после предыдущего перебора.

[galaxy]

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

TIME_WAIT — это нормально, никуда не деться от них. Не забывайте про SO_REUSEADDR и проблем быть не должно.

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

Да, про socket_read(), конечно так. Reset не нужен, foreach его все равно сделает.

И да, я свои примеры, естественно, не тестировал :)

Вообще, написание сетевых серверных приложений, особенно работающих с длинными соединениями и многими клиентами, требует большой аккуратности. Т.е. программа-пример из любого howto по сокетам будет работать как надо, но только до тех пор, пока не заглючит сеть, клиенты не начнут неожиданно рвать соединения и прочая, и прочая.

Так, в примере с селектом я делаю socket_read($fd, 4096) и перехожу к обработке данных. По идее никто не гарантирует, что я так вычитаю сообщение целиком (на это можно только надеяться, если сообщение достаточно короткие). По-хорошему, надо как-то договариваться о длине сообщения и читать из сокета (через select) пока не наберется эта длина. Для сообщений переменной длины можно ее передавать перед самим сообщением (как первые 4 байта пакета, например).

[SSar]

Я смотрю вы применяете SO_REUSEADDR только для серверного сокета, а для клиентских socket_accept() не надо?

[galaxy]

Нет, не надо, они все (вместе с серверным) на одном локальном адресе сидят.

[SSar]

Я переделал скрипт согласно вашему примеру с SO_REUSEADDR, однако от FIN_WAIT1 избавиться увы не получилось, более того невозможно создать сокет на прослушивание пока висит этот FIN_WAIT1.
Неужто нельзя где-нить таймаут ожидания данных и закрытия соединения уменьшить?

[SSar]

Также netstat показывает все нарастающее число сокетов в состоянии ESTABLISHED, хотя клиент всего 1 был.

[galaxy]

А покажите весь скрипт, если не сложно. И кстати, тут есть тег <source>

[SSar]

В курсе, уже использовал в 1м посте:

$time = time();
while(1) {
	usleep(100000);
	if((time()-$time) >= 60) { $time = time(); UpdatePID($pid); }
	$client = @socket_accept($socket);
	if($client) $clients[] = $client;
	reset($clients);
	while(list($i,$client) = each($clients)) {
		if((time()-$time) >= 60) { $time = time(); UpdatePID($pid); }
		$data = socket_read($client, 4096);
		if($data === false) { // error or closed connection
			socket_close($client);
			unset($clients[$i]); }
		elseif($data!="") {
			$data = trim($data);
			socket_getpeername($client, $peer);
			echo $peer."\n";
			echo $data;
			if(strtolower($data)=="exit") Quit("Aborted by user");
			SaveInDB($data); }
		}
	}

[galaxy]

Задебажил немного, вот рабочий (вроде) вариант:

define('EAGAIN', 11);

$socket = socket_create_listen(1234, SOMAXCONN) ;
socket_set_nonblock($socket);

$clients = array();

while(1) {
    usleep(300000);
    $client = @socket_accept($socket);
    if($client) {
      socket_set_nonblock($client);
      $clients[] = $client;
      socket_getpeername($client, $peer);
      echo "$client ($peer) connected\n";
    }
    reset($clients);
    while(list($i,$client) = each($clients)) {
        $data = socket_read($client, 4096);
        if($data === false && socket_last_error() != EAGAIN) {
            echo "error reading from $client: " . socket_strerror(socket_last_error()).", disconnecting\n";
            socket_close($client);
            unset($clients[$i]); }
        elseif($data!="") {
            $data = trim($data);
            socket_getpeername($client, $peer);
            echo "$client ($peer) says: $data\n";
            if(strtolower($data)=="exit") {
                socket_close($client);
                socket_close($socket);
                die("Aborted by user");
            }
        } elseif ($data === "") {
            echo "$client ($peer) disconnected\n";
            socket_close($client);
            unset($clients[$i]);
        }
    }
}

Следующие моменты исправил:
1. После accept клиентский сокет надо делать неблокирующим: socket_set_nonblock($client);
у вас скрипт, видимо, постоянно висел на read'e
2. socket_read на неблокирующем сокете возвращает ошибку с кодом 11 (EAGAIN), если нечего читать, что учтено в коде.
3. Сигналом того, что соединение закрыто клиентом, является отсутствие ошибки на read и прочитанная пустая строка: elseif ($data === "")
(в мануале по php ерунда, честно говоря, написана)

[SSar]

Супер, только что хотел дописать поправки по п1-2 про неблокирующий клиентский сокет и код 11.
Однако с решением старых проблем вновь полезли новые:

1. Одна и та же железячка (клиент) не собирается вовсе закрывать соединение через GPRS, но при этом шлет каждый раз данные с разных IP, которые повисают в статусе ESTABLISHED.

2. Через telnet эмулировал несколько одновременных подключений — все отлично работает одновременно, НО стоит закрыть соединения и сокеты повисают в статусе CLOSE_WAIT.

В итоге очередь соединений как и массив $clients все растет и растет…

PS мануал по php уже перелопатил еще до постинга на хабр ибо ответа там нет.

[galaxy]

Против CLOSE_WAIT'a должен помочь п3.

Непонятно, что такое «повисают в статусе ESTABLISHED», поскольку это живое соединение, и либо они там общаются, либо оба ждут, когда кто-нибудь что-нибудь пошлет. Если это ведет к зависанию, то тут бага на уровне логики приложения.

[SSar]

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

[galaxy]

Не волнуйтесь, 0 байт, считанных с сокета, готового для чтения — это, в общем-то, единственный и рекомендованный способ проверить, закрыто ли соединение с удаленной стороны.

Answer 2

[zizop]

Думаю вам стоит попробовать работу с неблокирующими сокетами через libevent.

<?php
$socket = stream_socket_server ('tcp://0.0.0.0:2000', $errno, $errstr);
stream_set_blocking($socket, 0);
$base = event_base_new();
$event = event_new();
event_set($event, $socket, EV_READ | EV_PERSIST, 'ev_accept', $base);
event_base_set($event, $base);
event_add($event);
event_base_loop($base);

$GLOBALS['connections'] = array();
$GLOBALS['buffers'] = array();

function ev_accept($socket, $flag, $base) {
static $id = 0;

$connection = stream_socket_accept($socket);
stream_set_blocking($connection, 0);

$id += 1;

$buffer = event_buffer_new($connection, 'ev_read', NULL, 'ev_error', $id);
event_buffer_base_set($buffer, $base);
event_buffer_timeout_set($buffer, 30, 30);
event_buffer_watermark_set($buffer, EV_READ, 0, 0xffffff);
event_buffer_priority_set($buffer, 10);
event_buffer_enable($buffer, EV_READ | EV_PERSIST);

// we need to save both buffer and connection outside
$GLOBALS['connections'][$id] = $connection;
$GLOBALS['buffers'][$id] = $buffer;
}

function ev_error($buffer, $error, $id) {
event_buffer_disable($GLOBALS['buffers'][$id], EV_READ | EV_WRITE);
event_buffer_free($GLOBALS['buffers'][$id]);
fclose($GLOBALS['connections'][$id]);
unset($GLOBALS['buffers'][$id], $GLOBALS['connections'][$id]);
}

function ev_read($buffer, $id) {
while ($read = event_buffer_read($buffer, 256)) {
var_dump($read);
}
}
?>

См. ru2.php.net/manual/en/book.libevent.php и ru2.php.net/manual/en/libevent.examples.php

Comments

[SSar]

спасибо, попробую

Answer 3

[Ramzeska]

Попробуйте переписать на inetd — на нем все получается в разы проще и компактнее.

Регистрируете свой порт в /etc/services:
myrpc 8000/tcp
Добавляете свой демон в /etc/inetd.conf:
myrpc stream tcp nowait ramzes /home/ramzes/tmp/myrpc.php myrpc.php

#!/usr/local/bin/php
<?

$f=fopen("php://stdin","r");

$data=fread($f,1024);

Save2DB($data);

fclose($f);

Comments

[SSar]

Спс за участие, но хотелось бы более подробнее увидеть. Постом выше учтено уже множество особенностей.

[Ramzeska]

Тут все получается достаточно просто — при установлении соединения на порт, сервер inetd перенаправляет все что приходит в сокет в stdin приложения, а все, что приложение пишет в стандартный вывод перенаправляется клиенту в сокет. На каждый коннект по копии приложения — т.е. 10 коннектов — 10 скриптов запустятся. Ну и когда скрипт завершает работу коннект закрывается. Все отличие от скрипта выше — вся сетевая обвязка остается внутри inetd, а код работает только непосредственно с потоками ввода/вывода

[RomanGPS]

А как в это случае ответить обратно трекеру, у меня GT06 модель, которая шлет каждые 10 секунд свой imei в бинарном формате и просит потвердить обратно получение, просто echo в данном php не срабатывает

Answer 4

[RomanGPS]

Получить поток через inetd действительно довольно просто, правда несколько дней длилось прежде чем провайдер настроит это в VPS. у меня еще трекер который шлет все в бинарном формате и просит ответ. Вот с ответом пока что то не получается ответить

$f=fopen("php://stdin","r");
$data=fread($f,1024);
$array = unpack("c*", $data);
//составляем ответ из полученной строки
$talkback=pack("c*",0x78,0x78,0x05,0x01,$array[13],$array[14],0xD9,0xDC,0x0D,0x0A);
echo $talkback;
//далье после ответа должен начать присылать данные о кооординатах но он упорно шлет первую строку логина, видимо не получает ответа через echo
fclose($f);