From 71ddfa1ebec36dd90d444322608581f964c4b3b0 Mon Sep 17 00:00:00 2001
From: Fabian Gruber <fabian@appwrite.io>
Date: Thu, 24 Apr 2025 16:10:01 +0200
Subject: [PATCH] feat(dispatch): add contextual dispatch logic

---
 src/Swoole/Dispatch/ContextualDispatch.php | 52 ++++++++++++
 src/Swoole/Dispatch/DispatchInterface.php  | 29 +++++++
 src/Swoole/Dispatch/RiskyRequest.php       | 94 ++++++++++++++++++++++
 3 files changed, 175 insertions(+)
 create mode 100644 src/Swoole/Dispatch/ContextualDispatch.php
 create mode 100644 src/Swoole/Dispatch/DispatchInterface.php
 create mode 100644 src/Swoole/Dispatch/RiskyRequest.php

diff --git a/src/Swoole/Dispatch/ContextualDispatch.php b/src/Swoole/Dispatch/ContextualDispatch.php
new file mode 100644
index 0000000..872a02e
--- /dev/null
+++ b/src/Swoole/Dispatch/ContextualDispatch.php
@@ -0,0 +1,52 @@
+<?php
+
+namespace Utopia\Swoole\Dispatch;
+
+use Swoole\Server;
+use Swoole\Table;
+
+/**
+ * Dispatch requests to workers according to the HTTP request message
+ */
+abstract class ContextualDispatch implements DispatchInterface
+{
+    private Table $dispatchMap;
+
+    /**
+     * @param int $dispatcherMapSize This should correspond to the max_connection paramter
+     *                               https://wiki.swoole.com/en/#/server/setting?id=max_conn-max_connection
+     */
+    public function __construct(readonly int $dispatcherMapSize)
+    {
+        $this->dispatchMap = new Table($dispatcherMapSize);
+        $this->dispatchMap->column('workerId', Table::TYPE_INT);
+        $this->dispatchMap->create();
+    }
+
+    /**
+     * {@inheritdoc}
+     */
+    public function __invoke(Server $server, int $fd, int $type, ?string $data = null): int
+    {
+        if ($this->dispatchMap->exists($fd)) {
+            $workerId = $this->dispatchMap->get($fd, 'workerId');
+        } else {
+            $workerId = $this->resolveWorkerId($server, $data);
+            $this->dispatchMap->set($fd, ['workerId' => $workerId]);
+        }
+        if ($type == self::CONNECTION_CLOSE) {
+            $this->dispatchMap->delete($fd);
+        }
+        return $workerId;
+    }
+
+    /**
+     * Extract request identifying information from a request message
+     *
+     * @param Server $server
+     * @param ?string $data
+     *
+     * @return int
+     */
+    abstract protected function resolveWorkerId(Server $server, ?string $data): int;
+}
diff --git a/src/Swoole/Dispatch/DispatchInterface.php b/src/Swoole/Dispatch/DispatchInterface.php
new file mode 100644
index 0000000..62103bc
--- /dev/null
+++ b/src/Swoole/Dispatch/DispatchInterface.php
@@ -0,0 +1,29 @@
+<?php
+
+namespace Utopia\Swoole\Dispatch;
+
+use Swoole\Server;
+
+interface DispatchInterface
+{
+    /**
+     * Connection dispatch type
+     *
+     * @link https://www.swoole.co.uk/docs/modules/swoole-server/configuration#dispatch_func
+     */
+    public const CONNECTION_FETCH  = 10;
+    public const CONNECTION_START  = 5;
+    public const CONNECTION_CLOSE  = 4;
+
+    /**
+     * Resolve requests to corresponding worker processes
+     *
+     * @param Server $server
+     * @param int $fd Client ID number
+     * @param int $type Dispatch type
+     * @param ?string $data Request packet data (0-8180 bytes)
+     *
+     * @return int Worker ID number
+     */
+    public function __invoke(Server $server, int $fd, int $type, ?string $data = null): int;
+}
diff --git a/src/Swoole/Dispatch/RiskyRequest.php b/src/Swoole/Dispatch/RiskyRequest.php
new file mode 100644
index 0000000..86622d3
--- /dev/null
+++ b/src/Swoole/Dispatch/RiskyRequest.php
@@ -0,0 +1,94 @@
+<?php
+
+namespace Utopia\Swoole\Dispatch;
+
+use Swoole\Server;
+
+abstract class RiskyRequest extends ContextualDispatch
+{
+    /**
+     * @param float $riskyWorkersPercent Decimal form 0 to 1
+     */
+    public function __construct(
+        int $dispatcherMapSize,
+        private readonly float $riskyWorkersPercent
+    ) {
+        parent::__construct($dispatcherMapSize);
+        if ($this->riskyWorkersPercent < 0 || $this->riskyWorkersPercent > 1) {
+            throw new \InvalidArgumentException('riskyWorkersPercent must be >=0 && <=1');
+        }
+    }
+
+    /**
+     * @param string $request
+     * @param string $domain
+     * @return bool
+     */
+    abstract protected function isRisky(string $request, string $domain): bool;
+
+    protected function randomRiskyWorker(int $riskyWorkers, int $totalWorkers): int
+    {
+        return rand($riskyWorkers, $totalWorkers - 1);
+    }
+
+    protected function randomSafeWorker(int $riskyWorkers): int
+    {
+        return rand(0, $riskyWorkers - 1);
+    }
+
+    protected function resolveWorkerId(Server $server, ?string $data): int
+    {
+        $totalWorkers = $server->setting['worker_num'];
+
+        // If data is not set, we can send the request to any worker.
+        // First we try to pick an idle worker, otherwise we randomly pick a worker.
+        if (empty($data)) {
+            for ($i = 0; $i < $totalWorkers; $i++) {
+                if ($server->getWorkerStatus($i) === SWOOLE_WORKER_IDLE) {
+                    return $i;
+                }
+            }
+            return rand(0, $totalWorkers - 1);
+        }
+
+        // Each worker has a numeric ID, starting from 0 and incrementing
+        // From 0 to $riskyWorkers, we consider safe workers
+        // From $riskyWorkers to $totalWorkers, we consider risky workers
+        $riskyWorkers = (int) floor($totalWorkers * $this->riskyWorkersPercent); // Absolute number of risky workers
+
+        $headers = explode("\n", strstr($data, "\r\n", true));
+        $request = $headers[0];
+        $domain = '';
+        if (count($headers) > 1) {
+            $domain = trim(explode('Host: ', $headers[1])[1]);
+        }
+
+        $risky = $this->isRisky($request, $domain);
+
+        if ($risky) {
+            // If risky request, only consider risky workers
+            for ($j = $riskyWorkers; $j < $totalWorkers; $j++) {
+                /** Reference https://openswoole.com/docs/modules/swoole-server-getWorkerStatus#description */
+                if ($server->getWorkerStatus($j) === SWOOLE_WORKER_IDLE) {
+                    // If idle worker found, give to him
+                    return $j;
+                }
+            }
+
+            // If no idle workers, give to random risky worker
+            return $this->randomRiskyWorker($riskyWorkers, $totalWorkers);
+        }
+
+        // If safe request, give to any idle worker
+        // It's fine to pick a risky worker here because it's idle. Idle is never actually risky
+        for ($i = 0; $i < $totalWorkers; $i++) {
+            if ($server->getWorkerStatus($i) === SWOOLE_WORKER_IDLE) {
+                return $i;
+            }
+        }
+
+        // If no idle worker found, give to a random safe worker
+        // We avoid risky workers here, as it could be in work - not idle. That's exactly when they are risky.
+        return $this->randomSafeWorker($riskyWorkers);
+    }
+}