Transferring data between workers

[osrec]

I would like to know how to transfer data between workers (like we can with task workers).

For example, let's say I have a HTTP server with 8 workers. On "workerstart", each worker calls a static method on classA which initialises a random static value in classA.

When I make a http request I want to print out the static value from classA, but for all the workers, regardless of which worker the request is handled by.

Is this possible? If yes, then what's the best way to achieve this?

Thanks very much :) !

[twose]

You can initialise a random static value before $server->start, not in workerStart.
Or you can use shared memory to transfer it, such as swoole_table

$http = new swoole_http_server('0.0.0.0', 8888, SWOOLE_BASE);

$http->on('start', function (swoole_http_server $server) {
    echo "start on http://127.0.0.1:8888\n";
});

$http->on('workerStart', function (swoole_http_server $server) {
    echo "start\n";
});

$http->on('request', function (swoole_http_request $request, swoole_http_response $response) {
    $response->end('hello swooler');
});

$http->on('workerStop', function () {
    echo "stop\n";
});

// init random value before start

$http->start();

[osrec]

Let me try and explain again with an example:

class MyClass
{
	static $randomVal;
	static function generateRandomVal() { self::$randomVal = md5(rand()); }
}

$http = new swoole_http_server('0.0.0.0', 8888, SWOOLE_BASE);

$http->set(['worker_num' => 8]);

$http->on('workerStart', function (swoole_http_server $server) 
{
    MyClass::generateRandomVal(); // This sets  MyClass::$randomVal
    echo "Initialised random val to " . MyClass::$randomVal . "\n";
});

# Write out random vals from all workers

$http->on('request', function (swoole_http_request $request, swoole_http_response $response) use ($http)
{
    $response->write('Random Values');
    for($w = 0; $w < $http->ports[0]->setting['worker_num']; ++$w)
    {
         $response->write("\n Random Val from Worker # $w: ");
         $response->write(/* get MyClass::$randomVal from each worker */);
    }
});

$http->start();

Does the above explain things better? Or should I try again with another example!?

[twose]

You should call MyClass::generateRandomVal() outside of the event callback

class MyClass
{
	static $randomVal;
	static function generateRandomVal() { self::$randomVal = md5(rand()); }
}

MyClass::generateRandomVal(); // This sets  MyClass::$randomVal

$http = new swoole_http_server('0.0.0.0', 8888, SWOOLE_BASE);

$http->set(['worker_num' => 8]);

$http->on('workerStart', function (swoole_http_server $server) 
{
    echo "Initialised random val to " . MyClass::$randomVal . "\n";
});

# Write out random vals from all workers

$http->on('request', function (swoole_http_request $request, swoole_http_response $response) use ($http)
{
    $response->write('Random Values');
    for($w = 0; $w < $http->ports[0]->setting['worker_num']; ++$w)
    {
         $response->write("\n Random Val from Worker # $w: ");
         $response->write(/* get MyClass::$randomVal from each worker */);
    }
});

$http->start();

[osrec]

Setting the values is not a problem. It works even when I do it on worker start. The real issue is getting the values from all threads on request. How do I dynamically access the values from each worker on request, regardless of which worker is handling the request?

[twose]

@osrec
Is that what you want?
https://www.swoole.co.uk/docs/modules/swoole-table

[osrec]

Not quite - what I'm looking for is the ability to transfer data between workers, much like the way we can transfer data between workers and taskWorkers (i.e. we issue a task to the taskWorker and get a result).

[twose]

In what scene it is necessary? the table uses share memory and it's a way to transfer data between workers.

[ghost]

My first intention too - use table? Why not? Please explain @osrec

Are you looking for messaging like in node with an global object cache?

[ghost]

@osrec whats with https://rawgit.com/tchiotludo/swoole-ide-helper/english/docs/classes/swoole_server.html#method_sendMessage ?

You can send serialized data and you can send to all the other workers... hmm.. would not work?

[osrec]

That is only for task workers, not general workers. That is what my original ask was - why can't we communicate with task workers in the same way as general workers?!

[ghost]

But lastly, why do you want to talk between different clients in different states without any state of guaranteed progress? Maybe, there is no client actually on the other worker? :)

[ghost]

Now i am interested in solving this, too. Those sharing-mechanisms couldn't be done actually, table is not flexible enough, @osrec is right, thats a big limitation without. Swoole should work as powerfull as node.js in this case, there you can communicate between processes and threads, up and down, via IPC. So, having this for all scriptable workers would be very powerfull. Lastly, only this would give flexibility of global vars / cache etc. But, the communication mechanism should be done with priority - first the messages, than all scripts.

Any progress in mind?

@osrec have you solved it already?

[ghost]

@twose the problem with table is you cannot sort or do all the other things you can do with arrays for example. So if you can share data for arrays, you can do everything with an updated array, but not with table

[ghost]

For example, i want to have a global cache for filemtime and Etag to reduce IO. I do not want to swell redis - thats to much IO, too. Besides redis, i use ramdisk, so i want to increase a count variable, to safe files from redis-cluster to ramdisk or SSD and to choose which files should be served from ramdisk and which could just be served from SSD. But, all in all, table would not help me to calculate and change this permanently.

With priorizized IPC i can handle that in arrays and calculate and sort and do all those things i can do with arrays, without a new foreach table row calculation every time. Although, automatism with timer-tick wouldn't help, it just swell all with no accuracy.

But you cannot say, table needs just a kind of sort - there are much more algorithms needed over time, so, IPC would solve all those things.

[osrec]

Agree with @flddr

[ghost]

Some thoughts or feelings about this? 😊

[ghost]

@osrec

Sorry, maybe i am confused...

After this talk some weeks ago i believe in you that i understand $server->sendMessage($message, $worker_id) wrong - but today i tried to communicate between all workers in SWOOLE_PROCESS and SWOOLE_BASE with in both 10 of workers - this works perfectly and as aspected.

I am able to talk from worker to worker, use and share arrays for example, and to share websocket-communication between all the 10 workers. (so every worker simply talks to its own known fds and i do not need $server->connections for example)

So, why do you mean this is not possible? Exactly what?

btw: i am not talking about taskworkers, i am talking about

$server->set([
  'worker_num' => 10
]);

Sorry, if i do not understand again 😉

As i am not at home anymore this weekend i post a testcode so you can better follow what i am talking about:

// $server = new swoole_websocket_server('xyz', 80, SWOOLE_PROCESS);
// $https = $server->addListener('xyz', 443, SWOOLE_SOCK_TCP | SWOOLE_SSL);

// $server = new swoole_websocket_server('xyz', 80, SWOOLE_BASE);
// $https = $server->addListener('xyz', 443, SWOOLE_SOCK_TCP | SWOOLE_SSL);

// $server->set([
//   'worker_num' => 10 // WORKER_NUM
// ]);

$server->fds = [];

$server->on('open', function (swoole_websocket_server $server, $request) {
  // add fd to scope
  $server->fds[$request->fd] = true; // dummy-bool
});

$server->on('close', function ($server, $fd) {
  // delete fd from scope
  unset($server->fds[$fd]);
});

$server->on('message', function (swoole_websocket_server $server, $frame) {
  $message = "simple test message number: " . rand(1,10000);
  // pipe the message to all 9 other workers
  for ($i=0; $i < 10; $i++) { // WORKER_NUM
    if ($i !== $server->worker_id)
      $server->sendMessage($message, $i); // in this case only workers (no taskworkers)
  }
  // message to all fds in this scope
  testMessageSender($server, $message);
});

$server->on('pipeMessage', function($server, $src_worker_id, $data) {
  // send to your known fds in worker scope
  testMessageSender($server, $data);
});

function testMessageSender(&$server, $message){
  // use only your own scope fds
  foreach ($server->fds as $fd => $dummyBool) {
    // push to your connected clients
    $server->push($fd, $message); // again: no need to iterate $server->connections
  }
}

This is just to share my testing. The next step is correct timing related to requests and so on, but this should be simply related to reactive coordination of all handlings (shared data) and so on. And we can use any type of data / arrays and so on like $server->fds.

So

I would like to know how to transfer data between workers (like we can with task workers).

should be answered? And you can use those arrays instead of table; either all data copied to every scope or handled via IPC in reactive way with all capabilities of arrays as we talked before.

For me my question above #1862 (comment) is answered 😊

[osrec]

Sorry for not responding in a while - been a busy few months. @flddr are you suggesting that you can transfer data between task workers and normal workers? I was not able to achieve this in earlier swoole versions. Perhaps this functionality has been enhanced recently?

[jsanahuja]

I can confirm using $server->sendMessage($message, $workerId) to send and $server->on('pipeMessage', function(\Swoole\Server $server, int $fromWorkerId, mixed $message){ ... } to receive works just fine.

This should be closed.

[twose]

@jsanahuja Thanks for the reminder!