Re: While loops blocking send?

[Igor Wiedler]

Hi,

This is expected behaviour. The answer is: When using Ratchet, do not use sleep. Or anything that takes a long time. Ever.

PHP is not multithreaded, so you cannot make any blocking calls. If you want to do blocking calls, move them to a separate process that is not tied to the event loop, and communicate over a message queue (see react/zmq and react/stomp).

Ratchet is based on the React event loop. Since you cannot block, you need to use non-blocking I/O. That's exactly what React provides. For any I/O (reading/writing files, network, etc) you need to use libraries that integrate with that event loop. See the React website for a list of all libraries that do so.

I hope I could answer your question. But feel free to ask here or in #reactphp on freenode if anything is still unclear.

Regards,

Igor

On Aug 30, 2012, at 4:57 PM, sch...@gmail.com wrote:

Using a very simple websocket interface, I'm running into a bit of trouble -- if I'm stuck in a loop with a long execution time, no data is flushed out of the sockets (despite calling $client->send() from within the loop).  

Here's the problematic snippet of code from my class:

    public function onMessage(ConnectionInterface $from, $msg) {

        $from->send("Starting Up");

        $x=0;

        while($x<10){

            echo ($x);

            $from->send($x);

            $x = $x+1;

            sleep(1);

        }

    }

If I run this, none of the output (including the first "Starting Up" message) is sent over the socket until after the loop has completed.  Curiously, however, if I look at my console output on the server, echo seems to work just fine.

Has anyone else encountered this?  Is there a way to flush the buffers within the loop?

--
 
 
 

[Andrew Schmadel]

Hi Igor,

Thanks for pointing me in the right direction.  I think I understand Ratchet (and React) a bit better now.

I've rearranged my code to properly use React's event loop (shared with Ratchet), and things are working much better now that I understand how the event loop works.  However, there's one place where I'm still seeing some blocking, and it doesn't make any sense.

Apologies that this has gotten less ratchet-specific, but I'm approaching a dead-end here.

Within my Ratchet application, I have the following code set to execute in response to a message:

        use React\Stream\Stream as Stream;

 

        $ping = new Stream(popen("ping -t 127.0.0.1",'r'),$server->loop);

        $ping->on('data',function($data) use ($from){

            $from->send($data);

        });

        $f4vpp = new Stream(popen("f4vpp -v -i in.f4v -o out.mp4",'r'),$server->loop);
        $f4vpp->on('data',function($data) use ($from){
            $from->send($data);
        });

 

Ping works as expected, and happily sends its output through Ratchet as it's received.  

However, if I include the second half of that code (f4vpp), everything is blocked until it finishes executing (at which point it's delivered all at once).  What's going on here?  Why is one of these long-running processes blocking PHP, while the other works just fine?

I'm using PHP5.4.3 on win32, for whatever that's worth.

Thanks!

Andrew

[weslley ceresa]

Hi,

Try: $this->loop->tick() after $from->send($data);

[Mostafa...@hotmail.com]

Here are some non-blocking code with Ratchet:

<?php
	/*
	* To change this license header, choose License Headers in Project Properties.
	* To change this template file, choose Tools | Templates
	* and open the template in the editor.
	*/
	namespace App;
	use Ratchet\MessageComponentInterface;
	use Ratchet\ConnectionInterface;
	/**
	* Description of Socket
	*
	* @author <mostafa...@hotmail.com>
	*/
	class Socket implements MessageComponentInterface{
	/**
	*
	* @var $clients array of each client
	*/
	protected $clients;
	protected $loop;
	/**
	* constructor
	*/
	public function __construct(\React\EventLoop\LoopInterface $loop) {
	$this->loop = $loop;
	$this->clients = new \SplObjectStorage;
	}
	/**
	*
	* @param ConnectionInterface $conn attaches clients
	*/
	public function onOpen(ConnectionInterface $conn) {
	echo "\nNew Client Connected...\n";
	$this->clients->attach($conn);
	}
	/**
	*
	* @param ConnectionInterface $from
	* @param $msg
	*/

	public function onMessage(ConnectionInterface $from, $msg) {

	echo "\nNew Message: " . $msg . "\n";
	foreach($this->clients as $clientSocket){
	$this->loop->futureTick(function() use ($clientSocket){
	$clientSocket->send("\ndata data start\n");
	});
	$this->loop->futureTick(function() use ($clientSocket){
	$clientSocket->send("\ndata data two\n");
	});
	$stream = new \React\Stream\ReadableResourceStream(fopen(storage_path('logs/laravel.log'), 'r'), $this->loop);
	$loop = $this->loop;
	$stream->on('data', function($data) use ($clientSocket, $loop){
	$loop->futureTick(function() use ($clientSocket, $data){
	$returnValue = file_get_contents(storage_path('logs/laravel.log')) .
	file_get_contents(storage_path('logs/laravel.log')) .
	file_get_contents(storage_path('logs/laravel.log')) .
	file_get_contents(storage_path('logs/laravel.log')) .
	file_get_contents(storage_path('logs/laravel.log')) .
	file_get_contents(storage_path('logs/laravel.log'));
	$clientSocket->send($returnValue);
	$clientSocket->send($data);
	});
	});
	$stream->on('end', function() use ($clientSocket, $loop){
	$loop->futureTick(function() use ($clientSocket){
	$clientSocket->send('finished');
	});
	});
	$this->loop->run();
	}
	}
	/**
	*
	* @param ConnectionInterface $conn
	*/
	public function onClose(ConnectionInterface $conn) {
	echo "\nConnection Closed \n";
	$this->clients->detach($conn);
	}
	/**
	*
	* @param ConnectionInterface $conn
	* @param \Exception $e
	*/
	public function onError(ConnectionInterface $conn, \Exception $e) {
	echo "\nError on Connection: {$e->getMessage()} on line {$e->getLine()} in file {$e->getFile()} \n";
	$conn->close();
	}
	}