Use counter to prevent race conditions in UvDriver

Fixes #16.

lib/UvDriver.php

@@ -18,6 +18,9 @@ class UvDriver implements Driver {
     /** @var string Loop onReadable watcher. */
     private $busy;
 
+    /** @var int Number of pending operations for disabling / enabling the busy watcher. */
+    private $pendingOperations = 0;
+
     /**
      * @param \Amp\Loop\UvDriver $driver
      */
@@ -25,7 +28,8 @@ class UvDriver implements Driver {
         $this->driver = $driver;
         $this->loop = $driver->getHandle();
 
-        // dummy handle to be able to tell the loop that there is work being done and it shouldn't abort if there are no other watchers at a given moment
+        // dummy handle to be able to tell the loop that there is work being done and it shouldn't abort if there are no
+        // other watchers at a given moment
         $this->busy = $driver->repeat(PHP_INT_MAX, function () { });
         $driver->unreference($this->busy);
     }
@@ -37,14 +41,12 @@ class UvDriver implements Driver {
         $flags = $this->parseMode($mode);
 
         $chmod = ($flags & \UV::O_CREAT) ? 0644 : 0;
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         $openArr = [$mode, $path, $deferred];
         \uv_fs_open($this->loop, $path, $flags, $chmod, function ($fh) use ($openArr) {
             if ($fh) {
                 $this->onOpenHandle($fh, $openArr);
             } else {
-                $this->driver->unreference($this->busy);
                 list(, $path, $deferred) = $openArr;
                 $deferred->fail(new FilesystemException(
                     "Failed opening file handle to $path"
@@ -52,7 +54,10 @@ class UvDriver implements Driver {
             }
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     private function parseMode(string $mode): int {
@@ -79,7 +84,6 @@ class UvDriver implements Driver {
         list($mode) = $openArr;
         if ($mode[0] === "w") {
             \uv_fs_ftruncate($this->loop, $fh, $length = 0, function ($fh) use ($openArr) {
-                $this->driver->unreference($this->busy);
                 if ($fh) {
                     $this->finalizeHandle($fh, $size = 0, $openArr);
                 } else {
@@ -91,7 +95,6 @@ class UvDriver implements Driver {
             });
         } else {
             \uv_fs_fstat($this->loop, $fh, function ($fh, $stat) use ($openArr) {
-                $this->driver->unreference($this->busy);
                 if ($fh) {
                     StatCache::set($openArr[1], $stat);
                     $this->finalizeHandle($fh, $stat["size"], $openArr);
@@ -119,7 +122,6 @@ class UvDriver implements Driver {
             return new Success($stat);
         }
 
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_stat($this->loop, $path, function ($fh, $stat) use ($deferred, $path) {
             if (empty($fh)) {
@@ -127,11 +129,13 @@ class UvDriver implements Driver {
             } else {
                 StatCache::set($path, $stat);
             }
-            $this->driver->unreference($this->busy);
             $deferred->resolve($stat);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
@@ -258,96 +262,101 @@ class UvDriver implements Driver {
      * {@inheritdoc}
      */
     public function lstat(string $path): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_lstat($this->loop, $path, function ($fh, $stat) use ($deferred) {
             if (empty($fh)) {
                 $stat = null;
             }
-            $this->driver->unreference($this->busy);
             $deferred->resolve($stat);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function symlink(string $target, string $link): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_symlink($this->loop, $target, $link, \UV::S_IRWXU | \UV::S_IRUSR, function ($fh) use ($deferred) {
-            $this->driver->unreference($this->busy);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function link(string $target, string $link): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_link($this->loop, $target, $link, \UV::S_IRWXU | \UV::S_IRUSR, function ($fh) use ($deferred) {
-            $this->driver->unreference($this->busy);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function readlink(string $path): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_readlink($this->loop, $path, function ($fh) use ($deferred) {
-            $this->driver->unreference($this->busy);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function rename(string $from, string $to): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_rename($this->loop, $from, $to, function ($fh) use ($deferred, $from) {
-            $this->driver->unreference($this->busy);
             StatCache::clear($from);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function unlink(string $path): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_unlink($this->loop, $path, function ($fh) use ($deferred, $path) {
-            $this->driver->unreference($this->busy);
             StatCache::clear($path);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function mkdir(string $path, int $mode = 0644, bool $recursive = false): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
 
         if ($recursive) {
@@ -362,7 +371,6 @@ class UvDriver implements Driver {
 
                 if (empty($arrayPath)) {
                     \uv_fs_mkdir($this->loop, $tmpPath, $mode, function ($fh) use ($deferred) {
-                        $this->driver->unreference($this->busy);
                         $deferred->resolve((bool) $fh);
                     });
                 } else {
@@ -381,37 +389,38 @@ class UvDriver implements Driver {
             $callback();
         } else {
             \uv_fs_mkdir($this->loop, $path, $mode, function ($fh) use ($deferred) {
-                $this->driver->unreference($this->busy);
                 $deferred->resolve((bool) $fh);
             });
         }
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function rmdir(string $path): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_rmdir($this->loop, $path, function ($fh) use ($deferred, $path) {
-            $this->driver->unreference($this->busy);
             StatCache::clear($path);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function scandir(string $path): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         uv_fs_readdir($this->loop, $path, 0, function ($fh, $data) use ($deferred, $path) {
-            $this->driver->unreference($this->busy);
             if (empty($fh)) {
                 $deferred->fail(new FilesystemException(
                     "Failed reading contents from {$path}"
@@ -421,21 +430,25 @@ class UvDriver implements Driver {
             }
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function chmod(string $path, int $mode): Promise {
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_chmod($this->loop, $path, $mode, function ($fh) use ($deferred) {
-            $this->driver->unreference($this->busy);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
@@ -443,59 +456,60 @@ class UvDriver implements Driver {
      */
     public function chown(string $path, int $uid, int $gid): Promise {
         // @TODO Return a failure in windows environments
-        $this->driver->reference($this->busy);
         $deferred = new Deferred;
         \uv_fs_chown($this->loop, $path, $uid, $gid, function ($fh) use ($deferred) {
-            $this->driver->unreference($this->busy);
             $deferred->resolve((bool) $fh);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function touch(string $path): Promise {
-        $this->driver->reference($this->busy);
         $atime = $mtime = time();
         $deferred = new Deferred;
         \uv_fs_utime($this->loop, $path, $mtime, $atime, function () use ($deferred) {
             // The uv_fs_utime() callback does not receive any args at this time
-            $this->driver->unreference($this->busy);
             $deferred->resolve(true);
         });
 
-        return $deferred->promise();
+        $promise = $deferred->promise();
+        $this->watchOperation($promise);
+
+        return $promise;
     }
 
     /**
      * {@inheritdoc}
      */
     public function get(string $path): Promise {
-        return new Coroutine($this->doGet($path));
+        $promise = new Coroutine($this->doGet($path));
+        $this->watchOperation($promise);
+        return $promise;
     }
 
     private function doGet($path): \Generator {
-        $this->driver->reference($this->busy);
         $promise = $this->doFsOpen($path, $flags = \UV::O_RDONLY, $mode = 0);
         if (!$fh = yield $promise) {
-            $this->driver->unreference($this->busy);
             throw new FilesystemException(
                 "Failed opening file handle: {$path}"
             );
         }
 
         $deferred = new Deferred;
+
         $stat = (yield $this->doFsStat($fh));
         if (empty($stat)) {
-            $this->driver->unreference($this->busy);
             $deferred->fail(new FilesystemException(
                 "stat operation failed on open file handle"
             ));
         } elseif (!$stat["isfile"]) {
             \uv_fs_close($this->loop, $fh, function () use ($deferred) {
-                $this->driver->unreference($this->busy);
                 $deferred->fail(new FilesystemException(
                     "cannot buffer contents: path is not a file"
                 ));
@@ -504,20 +518,18 @@ class UvDriver implements Driver {
             $buffer = (yield $this->doFsRead($fh, $offset = 0, $stat["size"]));
             if ($buffer === false) {
                 \uv_fs_close($this->loop, $fh, function () use ($deferred) {
-                    $this->driver->unreference($this->busy);
                     $deferred->fail(new FilesystemException(
                         "read operation failed on open file handle"
                     ));
                 });
             } else {
                 \uv_fs_close($this->loop, $fh, function () use ($deferred, $buffer) {
-                    $this->driver->unreference($this->busy);
                     $deferred->resolve($buffer);
                 });
             }
         }
 
-        return yield $deferred->promise();
+        return $deferred->promise();
     }
 
     private function doFsOpen($path, $flags, $mode) {
@@ -557,16 +569,16 @@ class UvDriver implements Driver {
      * {@inheritdoc}
      */
     public function put(string $path, string $contents): Promise {
-        return new Coroutine($this->doPut($path, $contents));
+        $promise = new Coroutine($this->doPut($path, $contents));
+        $this->watchOperation($promise);
+        return $promise;
     }
 
     private function doPut($path, $contents): \Generator {
         $flags = \UV::O_WRONLY | \UV::O_CREAT;
         $mode = \UV::S_IRWXU | \UV::S_IRUSR;
-        $this->driver->reference($this->busy);
         $promise = $this->doFsOpen($path, $flags, $mode);
         if (!$fh = yield $promise) {
-            $this->driver->unreference($this->busy);
             throw new FilesystemException(
                 "Failed opening write file handle"
             );
@@ -576,7 +588,6 @@ class UvDriver implements Driver {
         $len = strlen($contents);
         \uv_fs_write($this->loop, $fh, $contents, $offset = 0, function ($fh, $result) use ($deferred, $len) {
             \uv_fs_close($this->loop, $fh, function () use ($deferred, $result, $len) {
-                $this->driver->unreference($this->busy);
                 if ($result < 0) {
                     $deferred->fail(new FilesystemException(
                         \uv_strerror($result)
@@ -587,6 +598,22 @@ class UvDriver implements Driver {
             });
         });
 
-        return yield $deferred->promise();
+        return $deferred->promise();
+    }
+
+    private function watchOperation(Promise $promise) {
+        $this->pendingOperations++;
+
+        if ($this->pendingOperations === 1) {
+            $this->driver->reference($this->busy);
+        }
+
+        $promise->onResolve(function () {
+            $this->pendingOperations--;
+
+            if ($this->pendingOperations === 0) {
+                $this->driver->unreference($this->busy);
+            }
+        });
     }
 }