Distributed work pool / pipeline example

This example show how we can implement distributed work pool.

First file "master.php" distribute the work.

For each work it create a hash with work parameters (hash fields"a" and "b") and put the work id into a set called "pool:work".

Second file called "slave.php":

pop random work,
check the work parameters - e.g. hash fields,
do the work (e.g. calculate a + b),
store result into "result" hash field,
put it back into different set "pool:finishedwork"

After that the slave get new task.

Because Redis is single threaded and atomic, we are insured that when some slave gets the work, no one slave will get same work.

Note that you can have many slaves and there is no requirement those slaves to be on same machine, so you can scale on as many machines as you want.

Also note that slaves may work in a background as a server and it will stay and wait for work:

[~]$ php slave.php > /dev/null 2> /dev/null &

If job is big you can easyly hire thousend machines in Amazon EC2 and you can easyly do 1000 hours jobs in minutes

master.php

$r = new Redis();
$r->connect("127.0.0.1", "6379");

for($i = 0; $i < 1000; $i++){
	$x = $r->incr("id:work");

	$r->hmset(
		"work:$x", array(
			"id"    => $i           ,
			"a"     => rand()       ,
			"b"     => rand()
		)
	);

	$r->sadd("pool:work", $x);
}

echo "Work distributed\n";

slave.php

$r = new Redis();
$r->connect("192.168.0.1", "6379");

while(true){
	// Get random work id...
	$x = $r->spop("pool:work");

	if ($x === false){
		echo "No more work to do\n";
		sleep(10);
		continue;
	}

	// Get work description...
	$data = $r->hgetall("work:$x");

	if (!count($data))
		continue;

	// Do the work...
	$total = $data["a"] + $data["b"];

	// Store back...
	$r->hset("work:$x", "total", $total);

	// And store id in finished pool...
	$r->sadd("pool:finishedwork", $x);

	echo "Work #$x is finished...\n";
	// sleep(2);
}

Screen tesing

[~]$ php master.php
Work distributed

[~]$ 
[~]$ 
[~]$ redis-cli
redis 127.0.0.1> scard pool:work
(integer) 1000
redis 127.0.0.1> scard pool:finishedwork
(integer) 0
redis 127.0.0.1> quit

[~]$ 
[~]$ 
[~]$ php slave.php
...
Work #238 is finished...
Work #291 is finished...
Work #453 is finished...
Work #858 is finished...
Work #777 is finished...
No more work to do

[~]$ 
[~]$ 
[~]$ redis-cli
redis 127.0.0.1> scard pool:work
(integer) 0
redis 127.0.0.1> scard pool:finishedwork
(integer) 1000
redis 127.0.0.1> quit

[~]$

Redis connection note

The example uses generic connection to the Redis server. To make the example work with our service you will need to use code similar to this one:

// change these parameter according to the information in your instance list
$host = '85.25.11.9';
$port = 1234;
$password = 'somehashcode';
$db = 0;

$r = new Redis();
$r->connect($host, $port);
$r->auth($password);
$r->select($db);

blog comments powered by Disqus

Code library

Note:
Because we want this page to be useful for memcached users, we tagged with memcached all examples that may be "recreated" for memcached server.

Distributed work pool / pipeline example

master.php

slave.php

Screen tesing

Redis connection note

Code library

MySQL cache in PHP

How to prevent website to be web-scraped

'Rotating' news

Show random element

Accounting / Vote example

Scoreboard example

Scaling example

Distributed work pool / pipeline example

Simple realtime web counter

Emulation of expiration of the set members

Lock example

Message queue example

Index example

Using Redis for MySQL autoincrement

Using Redis as "data buffer"

Calculating how many visitors are on the same webpage using rolling average