Synchronization for parcels

src/Sync/Parcel.php

@@ -39,6 +39,11 @@ class Parcel implements ParcelInterface, \Serializable
      */
     private $key;
 
+    /**
+     * @var PosixSemaphore A semaphore for synchronizing on the parcel.
+     */
+    private $semaphore;
+
     /**
      * @var int An open handle to the shared memory segment.
      */
@@ -62,6 +67,8 @@ class Parcel implements ParcelInterface, \Serializable
         $this->memOpen($this->key, 'n', $permissions, $size + self::MEM_DATA_OFFSET);
         $this->setHeader(self::STATE_ALLOCATED, 0, $permissions);
         $this->wrap($value);
+
+        $this->semaphore = new PosixSemaphore(1);
     }
 
     /**
@@ -119,6 +126,20 @@ class Parcel implements ParcelInterface, \Serializable
         $this->memSet(self::MEM_DATA_OFFSET, $serialized);
     }
 
+    /**
+     * {@inheritdoc}
+     */
+    public function synchronized(callable $callback)
+    {
+        $lock = (yield $this->semaphore->acquire());
+
+        try {
+            yield $callback($this);
+        } finally {
+            $lock->release();
+        }
+    }
+
     /**
      * Frees the shared object from memory.
      *
@@ -138,6 +159,8 @@ class Parcel implements ParcelInterface, \Serializable
             $this->memDelete();
             shmop_close($this->handle);
             $this->handle = null;
+
+            $this->semaphore->free();
         }
     }
 
@@ -172,7 +195,7 @@ class Parcel implements ParcelInterface, \Serializable
      */
     public function serialize()
     {
-        return serialize($this->key);
+        return serialize([$this->key, $this->semaphore]);
     }
 
     /**
@@ -182,7 +205,7 @@ class Parcel implements ParcelInterface, \Serializable
      */
     public function unserialize($serialized)
     {
-        $this->key = unserialize($serialized);
+        list($this->key, $this->semaphore) = unserialize($serialized);
         $this->memOpen($this->key, 'w', 0, 0);
     }
 

src/SynchronizableInterface.php

@@ -9,11 +9,12 @@ interface SynchronizableInterface
     /**
      * @coroutine
      *
-     * Invokes a function while maintaining a lock on the object.
+     * Asynchronously invokes a callback while maintaining an exclusive lock on the object.
      *
-     * The given callback will be passed the object being synchronized on as the first argument.
+     * The given callback will be passed the object being synchronized on as the first argument. If the callback throws
+     * an exception, the lock on the object will be immediately released.
      *
-     * @param callable<self> $callback The synchronized function to invoke.
+     * @param callable<self> $callback The synchronized callback to invoke.
      *
      * @return \Generator
      *