server request and channel consumer

[mrceperka]

Hello,

I'd like to start http server and also spawn a consumer task on background. Server and consumer task would share channel for communication.

I've manged to do that with following code:

<?php declare(strict_types = 1);

use Swoole\Coroutine\Channel;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Http\Server;

\Co::set(['hook_flags' => SWOOLE_HOOK_TCP]);

// channel stuff
$channel = new Channel(1000);
$consumer = function () use ($channel) {
	echo "consume start\n";
	while (true) {
		
		$data[] = $channel->pop();
		var_dump($data);
	}
};

// SERVER
$host = '0.0.0.0';
$port = 9502;
$workers = 24;
$server = new Server($host, $port);
$server->set(
	[
		'worker_num' => $workers,
		'log_level'  => SWOOLE_LOG_WARNING,
	]
);

$server->on(
	"Start",
	function (Server $server) use ($host, $port) {
		echo "Swoole http server is started at http://$host:$port\n";
		echo "Server settings: \n";
		print_r($server->setting);
	}
);

$server->on(
	"Request",
	function (Request $request, Response $response) use ($channel) {
		$channel->push(['test']);
		
		$response->header("Content-Type", "text/plain");
		$response->status(200);
		$response->write('Ok');
		$response->end();
	}
);

\Co\run($consumer);

$server->start();

But there's weird thing when I run this code. It works, but also emits a fatal error.

===================================================================
 [FATAL ERROR]: all coroutines (count: 1) are asleep - deadlock!
===================================================================

 [Coroutine-1]
--------------------------------------------------------------------
#0  Swoole\Coroutine\Channel->pop() called at [/var/docker/backend/swoole/server2.php:18]

What's wrong? How do I fix this?

[matyhtf]

This is incorrect usage. The $consumer should be used in the Workerstart callback.

<?php declare(strict_types = 1);

use Swoole\Coroutine\Channel;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Http\Server;

\Co::set(['hook_flags' => SWOOLE_HOOK_TCP]);

// channel stuff
$channel = new Channel(1000);
$consumer = function () use ($channel) {
	echo "consume start\n";
	while (true) {
		
		$data[] = $channel->pop();
		var_dump($data);
	}
};

// SERVER
$host = '0.0.0.0';
$port = 9502;
$workers = 24;
$server = new Server($host, $port);
$server->set(
	[
		'worker_num' => $workers,
		'log_level'  => SWOOLE_LOG_WARNING,
	]
);

$server->on(
	"Start",
	function (Server $server) use ($host, $port) {
		echo "Swoole http server is started at http://$host:$port\n";
		echo "Server settings: \n";
		print_r($server->setting);
	}
);


$server->on(
	"Request",
	function (Request $request, Response $response) use ($channel) {
		$channel->push(['test']);
		
		$response->header("Content-Type", "text/plain");
		$response->status(200);
		$response->write('Ok');
		$response->end();
	}
);

$server->on(
	"workerStart",
	function (Request $request, Response $response) use ($channel, $consumer) {
            $consumer();
	}
);


$server->start();

[mrceperka]

I will try that ASAP, thanks. workerStart will spawn new coroutine?

[mrceperka]

@matyhtf It works, but is starts many workers. What if I want to start only one? Should I use global variable to limit that?

💡 Oh, it looks that I could use workerId received in workerStart callback.

[matyhtf]

You are right, that will execute only in the worker process when worker_id is 0

[fakharksa]

It is a v ery good way to communicate in a specific manner (with specific behavior / method). One more suggestion which is important to add here is this thatthe correct name of Consumer (for the sake of Swoole Design Pattern) will be a verb "Consume", not Consumer().

Consumer can actually be any function (any event's callback, in this case) which will make a call to the "consume()" anonymous function.

I think in certain use-case, it is also possible to reverse the flow which means "Technically, the onRequest event's callback can also call Consume() while callback of event onWorkerStart can push the data to the channel. Soo in this case, consume() is technically callable from within onRequest event's callback.