Mutex and RWMutex for Swoole

swoole-mutex.php

<?php declare(strict_types=1);

require __DIR__ . '/../src/bootstrap.php';
require_once __DIR__ . '/../src/Util/Swoole/Utils.php';

use Swoole\Coroutine;
use SneakyStu\Util\Swoole;

// TODO: will this help speed up concurrency servicing (via better scheduler?) not sure if it causes new userland concurrency coding issues
ini_set("swoole.enable_preemptive_scheduler", "1");

\Swoole\Runtime::enableCoroutine(SWOOLE_HOOK_ALL);

Interface IMutex {
  public function lock():void;
  public function unlock():void;
}

class FakeMutex implements IMutex {
  public function lock():void {}
  public function unlock():void {}
}

class Mutex implements IMutex {
  private \chan $lock;
  public function __construct() {
    $this->lock = new \chan();
  }
  public function lock():void {
    $this->lock->push(true);
  }
  public function unlock():void {
    $this->lock->pop();
  }
}

class RWMutex implements IMutex {
  private Mutex $rwlock;
  private int $rlockCount = 0;
  private ?\chan $rwlockAvailable;

  public function __construct() {
    $this->rwlockAvailable = NULL;
    $this->rwlock = new Mutex();
  }
  // rwlock
  public function lock():void {
    $this->rwlock->lock();
    if ($this->rlockCount === 0) {
      return;
    } else {
      $this->rwlockAvailable = new \chan();
      $this->rwlock->unlock();
      $this->rwlockAvailable->pop();
      $this->rwlock->lock();
      $this->rwlockAvailable = NULL;
      return;
    }
  }
  // rwlock
  public function unlock():void {
    $this->rwlock->unlock();
  }

  // rlock
  public function rlock():void {
    $this->rwlock->lock();
    $this->rlockCount++;
    $this->rwlock->unlock();
  }
  // rlock
  public function runlock():void {
    $this->rwlock->lock();
    $this->rlockCount--;
    if ($this->rwlockAvailable) {
      $this->rwlockAvailable->push(true);
    }
    $this->rwlock->unlock();
  }
}

class SharedData {
  public array $data = [];
  public IMutex $lock;

  public function __construct() {
    $this->lock = new RWMutex;
  }
}
Co\run(function() {
  $sharedData = new SharedData;

  // one function is constantly adding data
  go(function() use ($sharedData) {
    while (true) {
      $sharedData->lock->lock();
      $sharedData->data[] = rand(1,1000);
      print '+';
      \Swoole\Coroutine\System::sleep(Swoole\Utils::TIMEOUT_MIN);
      $sharedData->lock->unlock();
    }
  });

  // one function is constantly wiping all data
  go(function() use ($sharedData) {
    while (true) {
      $sharedData->lock->lock();
      $sharedData->data = [];
      print '+';
      \Swoole\Coroutine\System::sleep(Swoole\Utils::TIMEOUT_MIN);
      $sharedData->lock->unlock();
    }
  });

  // one function is constantly using the data and subject to the races
  go(function() use ($sharedData) {
    $t0 = microtime(true);
    $counter = 1;

    while (true) {
      $counter++;
      if ($counter % 1000 == 0) {
        $t = microtime(true) - $t0;
        $tps = $counter/$t;
        print "\nTPS: {$tps}/s\n";
      }

      $sharedData->lock->rlock();
      print '-';
      $sharedDataLength = count($sharedData->data);
      \Swoole\Coroutine\System::sleep(Swoole\Utils::TIMEOUT_MIN);
      $sharedDataLength2 = count($sharedData->data);
      if ($sharedDataLength != $sharedDataLength2) {
        print "race detected\n";
      }
      $sharedData->lock->runlock();
    }
  });

});