Timer not firing with Ev::RUN_NOWAIT

[Former user]

I can get the blocking timer to fire with

        // Create and start timer firing after 2 seconds
        $w1 = new EvTimer(2, 0, function ($starttime) {

           error_log("Finished Waiting\n",3,"/logs/log");

         });

        // Loop until Ev::stop() is called or all of watchers stop
        Ev::run();

but if I use

Ev::run(Ev::RUN_NOWAIT);

The timer does't fire.

[Ruslan Osmanov]

This is expected behavior. As described in the documentation, RUN_NOWAIT handles already outstanding events in a single iteration of the loop. It will not wait for any events to become outstanding (unlike RUN_ONCE, for instance).

If Ev::run(Ev::RUN_NOWAIT) is called before $w1 becomes outstanding (2 seconds elapsed), then the event will not be handled, and the callback will not be triggered.

[patrick Shirkey]

Thanks for your quick reply. Sorry if this is dense.

Is the correct method to wrap it with pthread or is there another way to get an asynchronous execution?

[Ruslan Osmanov]

@boosth, if you mean integration with pthreads extension, it depends on the specific task. It is possible to run one Ev loop per thread without locking, but operations on the same Ev loop in different threads require a mutex locking. If you only need to run a timer and another block of code asynchronously, then create a new watcher for the block before calling Ev::run.

[patrick Shirkey]

Thank you for this advice. It is very helpful.

Another thing I am having trouble groking is how to pass variables to the loop?

ex.

$var1 = xxx;
$var2 = yyy;
$var3 = zzz;

        // Create and start timer firing after 2 seconds
        $w1 = new EvTimer(2, 0, function ($starttime) {

           error_log($var1 . $var2 . $var3"\n",3,"/logs/log");

         });

[patrick Shirkey]

I'm not able to get past the blocking mechanism. Maybe my logic is incorrect for the way I am using EvTimer?

This code blocks the output stream until the timer has completed. My intention is to stream the first burst while the timer is active and only complete the stream once the timer is complete.

   $in  = fopen($track_name, "r");
   $out = fopen("php://output", "w");

    while ($size1 < $limit){
      $size1 += fwrite($out,fread($in,8192));
    }

        // Create and start timer firing after 10 seconds
        $w1 = new EvTimer(10, 0, function ($w) {

           error_log("Finished Waiting\n",3,"/logs/log");

               while (!feof($in)){
                     $size1 += fwrite($out,fread($in,8192));
               }

               fclose($in);
               fclose($out);
         });

        // Loop until Ev::stop() is called or all of watchers stop
        Ev::run(Ev::RUN_ONCE);

[Ruslan Osmanov]

There are many ways to pass data into the watcher callbacks:

passing variables from the current scope into the scope of anonymous function:

<?php
$var1 = 'xxx';
$var2 = 'yyy';
$var3 = 'zzz';

$w = new EvTimer(2, 0, function ($watcher, $revents) use ($var1, $var2, $var3) {
    printf("%s %s %s\n", $var1, $var2, $var3);
});

Ev::run();

using the data property:

<?php
class MyTimerData {
    private $x, $y, $z;

    public function __construct($x, $y, $z) {
        $this->x = $x;
        $this->y = $y;
        $this->z = $z;
    }

    public function __get($key) {
        return $this->$key ?? null;
    }
}

$w = new EvTimer(2, 0, function ($watcher, $revents) {
    $d = $watcher->data;
    printf("%s %s %s\n", $d->x, $d->y, $d->z);
});
$w->data = new MyTimerData('xxx', 'yyy', 'zzz');

Ev::run();

using a custom class encapsulating an EvWatcher:

<?php
class MyTimer {
    private $timer;
    private $x, $y, $z;

    public function __construct($x, $y, $z) {
        $this->x = $x;
        $this->y = $y;
        $this->z = $z;

        $this->timer = new EvTimer(2, 0, [$this, 'callback']);
    }

    public function __destruct() {
        // Prevent memory leak by explicitly destroying the watcher object
        $this->timer = null;
    }

    public function callback($watcher, $revents) {
        printf("%s %s %s\n", $this->x, $this->y, $this->z);
    }
}

$timer = new MyTimer('xxx', 'yyy', 'zzz');

Ev::run();

[Ruslan Osmanov]

This code blocks the output stream until the timer has completed. My intention is to stream the first burst while the timer is active and only complete the stream once the timer is complete.

You shouldn't run long-term blocking I/O within the watcher callback, unless the loop is running in separate thread, and doesn't interfere with other async operations.

If you want to perform I/O asynchronously, you should: 1) set the file descriptor (or PHP stream) to non-blocking mode (otherwise, the process may be blocked because of a spurious Ev::READ event, for instance); 2) register an EvIo watcher for the file descriptor/stream using Ev::READ, or Ev::WRITE flags and a callback; 3) in the callback (which is called when the file descriptor becomes readable/writable) read/write a reasonably small chunk of data and leave the file descriptor alone as soon as possible so the other watchers could use it.

In the following code an EvIo watcher asynchronously reads from input stream, until either internal limit is reached, or EvTimer stops it:

<?php
class TestReader
{
    const MAX_BYTES = 1; //8192;

    private $loop;
    private $in, $out;
    private $reader, $timer;
    private $size = 0;
    private $limit = 100;
    private $timeout = 2;

    public function __construct(string $istream, string $ostream) {
        $this->loop = new EvLoop();

        $this->in = fopen($istream, 'r');
        stream_set_blocking($this->in, false);

        $this->out = fopen($ostream, 'w');

        $this->reader = $this->loop->io($this->in, Ev::READ, [$this, 'readerCallback']);
        $this->timer = $this->loop->timer($this->timeout, 0, [$this, 'timerCallback']);
    }

    public function __destruct() {
        $this->loop->stop(Ev::BREAK_ALL);

        $this->reader = null;
        $this->timer = null;
        $this->loop = null;

        fclose($this->in);
        fclose($this->out);
    }


    public function readerCallback(EvWatcher $w, int $revents) {
        if (!feof($this->in) && $this->size < $this->limit) {
            $this->size += fwrite($this->out, fread($this->in, static::MAX_BYTES));
        }
    }

    public function timerCallback(EvWatcher $w, int $revents) {
        echo __METHOD__, PHP_EOL;

        $this->reader->stop();

        while (!feof($this->in)) {
            $this->size += fwrite($this->out, fread($this->in, static::MAX_BYTES));
        }
    }

    public function run() {
        $this->loop->run();
    }
}

$r = new TestReader($track_name, 'php://output');
$r->run();

Note, the readerCallback is reading only a portion of bytes at the time when the file descriptor is readable, while the timerCallback blocks until the end of file is reached.

[patrick Shirkey]

Thank you again for your very detailed replies. It's very helpful.

[patrick Shirkey]

Unfortunately I'm still seeing blocking behaviour on the output stream with this solution.

With this logic I see the connection as pending AND the initial burst is not released until after the timerCallback has fired.

I added this call :

stream_set_blocking($this->out, false);

I'm wondering if there is a system config issue which is stopping asynchronous behaviour in php?

I am running the following versions:

php5-fpm --version

PHP 5.6.30-0+deb8u1 (fpm-fcgi) (built: Feb 8 2017 08:51:18)

pecl info ev

About pecl.php.net/ev-1.0.4

Release Type PECL-style PHP extension (source code) Name ev Channel pecl.php.net Summary Provides interface to libev library Description ev provides interface to libev library - high performance full-featured event loop written in C. Maintainers Ruslan Osmanov [e-mail] (lead) Release Date 2016-12-02 12:22:49 Release Version 1.0.4 (stable) API Version 0.3.0 (stable) License PHP (http://www.php.net/license) Release Notes Fix: memory leak in PHP 7.0.13 Required Dependencies PHP version 5.4.0 PEAR installer version 1.4.0a1 or newer package.xml version 2.0 Last Modified 2017-02-22 09:25 Previous Installed - None - Version

[Ruslan Osmanov]

In PHP-FPM, the output is buffered. You need to use fflush to actually send the buffer contents to the target file descriptor.

However, Ev is mainly designed for CLI scripts usually running as background processes. CLI is convenient for long-running operations, especially when there is a need in keeping states (chats, WebSocket-servers, etc.). You may want to look at the event loop implementation in Icicle project.

[patrick Shirkey]

I was tentatively hoping that I could use ev for this case.

Adding fflush to the loops didn't help.

Can you see any possible path to a solution with php5-fpm and the streaming mechanism?

My goal is to only serve the full stream after a notification has been validated post timeout.

It's an unusual case. I need to send the initial burst without any delay and then only send the remaining data if the notification is validated after the timeout has expired.

Maybe I need a custom C extension for this functionality?

[Ruslan Osmanov]

I don't quite understand why you need this, but you can 1) disable compression on the server side, for example in Nginx:

fastcgi_keep_conn on;
gzip off;
proxy_buffering off;

and 2) the following changes:

<?php
//...

    public function readerCallback(EvWatcher $w, int $revents) {
        if (!feof($this->in) && $this->size < $this->limit) {
            $this->size += fwrite($this->out, fread($this->in, static::MAX_BYTES));
            ob_flush();
            flush();
            // Should work without this fflush($this->out);
        }
    }

header('Content-Encoding: none');
header('Cache-Contol: no-cache');
$r = new TestReader($track_name, 'php://output');
$r->run();

(Tested on Nginx 1.10.2 + PHP 7.1 FPM.)

If your code is not just for learning/test purposes, consider writing a CLI server and some kind of inter-process communication.

[patrick Shirkey]

Thank you again for your detailed answers.

I am serving the following:

header("Content-Type: audio/mpeg");

My goal is to stop malicious parties from bypassing the frontend UI logic and attempting to slurp files from the server.

I would like to start an initial burst of data to get the audio stream going instantly on the frontend and wait for verification in the backend that a timeout has been reached on the frontend before completing the remainder of the data transfer. The initial burst should provide enough buffer to allow the track to be played until the timeout is verified on the backend. All of this has to happen without causing an audible dropout in the audio stream for the listener. It's a bit awkward but the overall logic of the system relies on the delayed notification from the frontend.