Documentation

[krakjoe]

I've started documentation in the manual, it needs more work and examples added. We can clean up the readme now also.

@remicollet do you want to do the readme changes with install info as normal please ?

I'll close this when I'm happy that there's enough examples in the manual ...

[remicollet]

Example proposal.

A background Logger, to not have to wait for log to be written to continue task execution

<?php
class BackgroundLogger {
	private $worker  = NULL;
	private $file    = NULL;
	private $futures = array();

	function __construct(string $file) {
		$this->file = $file;
		$this->worker = new \parallel\Runtime();
	}

	public function __destruct() {
		$this->worker->close();
	}

	public function log(string $text) {
		// clean up running tasks to save memory
		foreach($this->futures as $i => $f) {
			if ($f->done()) {
				unset($this->futures[$i]);
			}
		}
		// New task
		$this->futures[] = $this->worker->run(function($file, $text) {
			$d = date(DATE_RFC2822);
			error_log("$d $text\n", 3, $file);
			usleep(10000);
			return true;
		}, [$this->file, $text]);
	}
}

echo "Start\n";
$logger = new BackgroundLogger($l = __FILE__ . '.log');

$logger->log("Start");
for ($i=1; $i<=20 ; $i++) {
	$logger->log("Continue $i");
}
$logger->log("Done");


echo "Wait for logger\n";
unset($logger);
echo "Done\n\n";
readfile($l);

[remicollet]

Another proposal:

Task runner in parallel, with maximum

<?php
class Runner {
	private $next;
	private $max;
	private $workers = array();
	private $tasks   = array();

	function __construct(int $max, bool $debug) {
		$this->max = $max;
		$this->next = 0;
		$this->debug = $debug;
	}

	public function __destruct() {
		// Wait all tasks
		$done = $err = array();
		while (\parallel\Future::select($this->tasks, $done, $err)) {
			foreach($done as $id => $task) {
				unset($this->workers[$id]);
				if ($this->debug) echo "+ {$id} ended\n";
			}
		}
	}

	public function run(callable $fct, ...$args) {
		if (count($this->tasks) >= $this->max) {
			// Wait 1 task to stop
			$done = $err = array();
			if (\parallel\Future::select($this->tasks, $done, $err)) {
				foreach($done as $id => $task) {
					unset($this->workers[$id]);
				if ($this->debug) echo "+ {$id} ended\n";
				}
			}
		}
		// Create a new task
		$this->workers[$this->next] = new \parallel\Runtime();
		$this->tasks[$this->next]   = $this->workers[$this->next]->run(function($fct, $args) {
			call_user_func_array($fct, $args);
			return true;
		}, [$fct, $args]);
		if ($this->debug) echo "+ {$this->next} started\n";
		$this->next++;
	}
}

echo "Start\n";

$runner = new Runner(3, true);
$runner->run('sleep', 3);
$runner->run('sleep', 2);
$runner->run('sleep', 1);
$runner->run('sleep', 1);
$runner->run('sleep', 1);

echo "Done\n\n";

[krakjoe]

I would write the first one like this:

<?php
class BackgroundLogger {
	private $worker  = NULL;
	private $file    = NULL;

	public function __construct(string $file) {
		$this->file = $file;
		$this->worker = new \parallel\Runtime();
	}

	public function log(string $text) {
		$this->worker->run(function($file, $text) {
			\error_log(\sprintf(
				"%s %s\n", 
				\date(\DATE_RFC2822), 
				$text), 
			 3, 
			 $file);
		}, [$this->file, $text]);
	}
	
	public function __destruct() {
		$this->worker->close();
	}
}

echo "Start\n";
$logger = new BackgroundLogger($l = __FILE__ . '.log');

$logger->log("Start");
for ($i=1; $i<=20 ; $i++) {
	$logger->log("Continue $i");
}
$logger->log("Done");


echo "Wait for logger\n";
unset($logger);
echo "Done\n\n";
readfile($l);

Sleeping in the thread seems superfluous, in addition, so are the futures, since they are not returning any useful value. Keeping them is just a waste of memory. In a real application when the stack gets to 0 size, it will be cleaned automatically, freeing the memory in the background thread.

[krakjoe]

Creating a whole thread to execute one task is awfully wasteful, and probably not something we should encourage by example.

here is an example of pooling that may be worked into a nice example ... you might extend it to create threads on demand up to a maximum ...

When it comes to a good example that uses futures/select ... you might extend a pooling example where the tasks do actually return a useful value, and use select for those ... one example I wrote a long time ago which you might use for inspiration: Fetch a bunch of pages from php.net, use DOM/XPath to parse out method prototypes or something of that sort, then you may use timed select() to process the results available first, possibly combine them into some page, or something of that sort ...

We want to try and steer clear of examples that use Future for the sake of using it, it sort of suggests that you must always use a future, and you actually shouldn't unless the Closure returns something useful, you should just fire and forget in that case.

[remicollet]

example of pooling

But you example simply submit task to next worker (each worker will execute same number of tasks), not on a free worker... the reason one I want to use select, but was bitten by bug #12

[krakjoe]

Maybe we could add a ::busy (or some other name) method to the runtime ? But still it might result in creating a worker per task, wouldn't it ?

[remicollet]

When all workers are busy, we should not create a new one, as we want a maximum number of parallel tasks, so we have to wait for 1 to be available, thus the "wait" method, but this one doesn't really work as expected for now (see "select" or "pool" behavior for files)

[krakjoe]

That background logger with Channels ...

<?php
use \parallel\Runtime;
use \parallel\Channel;

class BackgroundLogger {

    public function __construct(string $file) {
        $this->runtime = new Runtime();
        $this->channel = 
            Channel::make($file, Channel::Infinite);
            
        $this->runtime->run(function($file){
            $channel = Channel::open($file);
            $handle  = fopen($file, "rb");
            
            if (!is_resource($handle)) {
                throw new \RuntimeException(
                    "could not open {$file}");
            }
            
            while (($input = $channel->recv())) {
                fwrite($handle, "{$input}\n");
            }
            
            fclose($handle);
        }, [$file]);
    }
    
    public function log($message, ... $args) {
        $this->channel->send(
            vsprintf($message, $args));
    }
    
    public function __destruct() {
        $this->channel->send(false);
    }
    
    private $channel;
    private $runtime;
}

$logger = new BackgroundLogger("php://stdout");

$logger->log("hello world");
$logger->log("I am %s", "here");
?>

I'm pretty sure we are ready to start working on examples ... the events thing is going to take some explaining ...

[krakjoe]

That runner thing, with Events:

use \parallel\Runtime;
use \parallel\Events;

class Runner {

    public function __construct($max) {
        $this->max = $max;
        $this->running = new Events();
        $this->runtime = new Runtime();
    }
    
    public function run(\Closure $task, array $args = []) {
        if (count($this->running) >= $this->max) {
            foreach ($this->running as $event) {
                if (count($this->running) < $this->max)
                    break;
            }
        }
        
        $this->running->addFuture(
            md5(mt_rand()), 
            $this->runtime->run($task, $args));
    }
}

That's just one worker, so it's clear, but you can extend it with more workers ...

[dktapps]

@krakjoe won't the logger example need an exception catch when the channel is closed?

never mind, I did not see your destructor. Seems like it could have problems if an empty string is pushed into the channel though.

[krakjoe]

I've completed the manual for 0.9.0, I didn't add examples yet, but did add explanations for everything I think needs explaining ...

I'm going to close this now as the manual is in pretty good shape, even without examples ...

I don't object to adding some code examples to the manual, maybe we could have a section for that ... if anyone wants to find the time to do that, open a new issue so the examples can be reviewed ...

[maxgalbu]

For those trying to run the background logger example and finding out it doesn't run on parallel 1.1.4... here's a working version:

<?php

use \parallel\Events;
use \parallel\Runtime;
use \parallel\Channel;

class BackgroundLogger {
    private $channel;
    private $runtime;

    public function __construct(string $file) {
        $this->runtime = new Runtime();
        $this->channel = Channel::make($file, Channel::Infinite);
        $this->events = new Events();

        $future = $this->runtime->run(function($file){
            $channel = Channel::open($file);
            $handle = fopen($file, "rb");

            if (!is_resource($handle)) {
                throw new \RuntimeException("could not open {$file}");
            }

            while (($input = $channel->recv())) {
                fwrite($handle, "{$input}\n");
            }

            fclose($handle);
        }, [$file]);

        $this->events->addFuture(md5(mt_rand()), $future);
    }

    public function log($message, ... $args) {
        $this->channel->send(vsprintf($message, $args));
    }

    public function __destruct() {
        $this->channel->send(false);
    }
}

$logger = new BackgroundLogger("php://stdout");
$logger->log("hello world");
$logger->log("I am %s", "here");

Hint: you can't ignore Runtime::run() return value (a Future instance)