Added Hyperf\Coroutine\Mutex. (#7224)

* Added `Hyperf\Coroutine\Mutex`.

* Update CHANGELOG-3.1.md

src/Mutex.php

@@ -0,0 +1,61 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\Coroutine;
+
+use Hyperf\Engine\Channel;
+
+class Mutex
+{
+    /**
+     * @var Channel[]
+     */
+    protected static array $channels = [];
+
+    public static function lock(string $key, float $timeout = -1): bool
+    {
+        if (! isset(static::$channels[$key])) {
+            static::$channels[$key] = new Channel(1);
+        }
+
+        $channel = static::$channels[$key];
+        $channel->push(1, $timeout);
+        if ($channel->isTimeout() || $channel->isClosing()) {
+            return false;
+        }
+
+        return true;
+    }
+
+    public static function unlock(string $key, float $timeout = 5): bool
+    {
+        if (isset(static::$channels[$key])) {
+            $channel = static::$channels[$key];
+            $channel->pop($timeout);
+            if ($channel->isTimeout()) {
+                // unlock more than once
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    public static function clear(string $key): void
+    {
+        if (isset(static::$channels[$key])) {
+            $channel = static::$channels[$key];
+            static::$channels[$key] = null;
+            $channel->close();
+        }
+    }
+}

tests/MutexTest.php

@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace HyperfTest\Coroutine;
+
+use Hyperf\Coroutine\Mutex;
+use Hyperf\Coroutine\WaitGroup;
+use Hyperf\Engine\Channel;
+use PHPUnit\Framework\Attributes\CoversNothing;
+use PHPUnit\Framework\TestCase;
+
+use function Hyperf\Coroutine\go;
+
+/**
+ * @internal
+ * @coversNothing
+ */
+#[CoversNothing]
+class MutexTest extends TestCase
+{
+    public function testMutexLock()
+    {
+        $chan = new Channel(5);
+        $func = function (string $value) use ($chan) {
+            if (Mutex::lock('test')) {
+                try {
+                    usleep(1000);
+                    $chan->push($value);
+                } finally {
+                    Mutex::unlock('test');
+                }
+            }
+        };
+
+        $wg = new WaitGroup(5);
+        foreach (['h', 'e', 'l', 'l', 'o'] as $value) {
+            go(function () use ($func, $value, $wg) {
+                $func($value);
+                $wg->done();
+            });
+        }
+
+        $res = '';
+        $wg->wait(1);
+        for ($i = 0; $i < 5; ++$i) {
+            $res .= $chan->pop(1);
+        }
+
+        $this->assertSame('hello', $res);
+    }
+}