Merge branch 'koush:main' into main

.github/workflows/docker.yml

@@ -19,7 +19,14 @@ jobs:
 #     runs-on: ubuntu-latest
     strategy:
       matrix:
-        BASE: ["18-jammy-full", "18-jammy-lite", "18-jammy-thin", "20-jammy-full", "20-jammy-lite", "20-jammy-thin"]
+        BASE: [
+          "18-jammy-full",
+          "18-jammy-lite",
+          # "18-jammy-thin",
+          # "20-jammy-full",
+          # "20-jammy-lite",
+          # "20-jammy-thin",
+        ]
         SUPERVISOR: ["", ".s6"]
     steps:
       - name: Check out the repo

common/src/async-queue.ts

@@ -45,36 +45,64 @@ export function createAsyncQueue<T>() {
             return true;
         }
 
+        if (signal)
+            dequeued ||= new Deferred();
+
         const qi = {
             item,
             dequeued,
         };
         queued!.push(qi);
 
-        signal?.addEventListener('abort', () => {
+        if (!signal)
+            return true;
+
+        const h = () => {
             const index = queued.indexOf(qi);
             if (index === -1)
                 return;
             queued.splice(index, 1);
             dequeued?.reject(new Error('abort'));
-        });
+        };
+
+        dequeued.promise.catch(() => {}).finally(() => signal.removeEventListener('abort', h));
+        signal.addEventListener('abort', h);
 
         return true;
     }
 
+    function end(e?: Error) {
+        if (ended)
+            return false;
+        // catch to prevent unhandled rejection.
+        ended = e || new EndError()
+        while (waiting.length) {
+            waiting.shift().reject(ended);
+        }
+        return true;
+    }
+
     function queue() {
         return (async function* () {
-            while (true) {
-                try {
-                    const item = await dequeue();
-                    yield item;
-                }
-                catch (e) {
-                    if (e instanceof EndError)
-                        return;
-                    throw e;
+            try {
+                while (true) {
+                    try {
+                        const item = await dequeue();
+                        yield item;
+                    }
+                    catch (e) {
+                        // the yield above may raise an error, and the queue should be ended.
+                        end(e);
+                        if (e instanceof EndError)
+                            return;
+                        throw e;
+                    }
                 }
             }
+            finally {
+                // the yield above may cause an iterator return, and the queue should be ended.
+                end();
+            }
         })();
     }
 
@@ -93,6 +121,9 @@ export function createAsyncQueue<T>() {
     }
 
     return {
+        get ended() {
+            return ended;
+        },
         take,
         clear() {
             return clear();
@@ -106,14 +137,7 @@ export function createAsyncQueue<T>() {
         submit(item: T, signal?: AbortSignal) {
             return submit(item, undefined, signal);
         },
-        end(e?: Error) {
-            if (ended)
-                return false;
-            // catch to prevent unhandled rejection.
-            ended = e || new EndError()
-            clear(e);
-            return true;
-        },
+        end,
         async enqueue(item: T, signal?: AbortSignal) {
             const dequeued = new Deferred<void>();
             if (!submit(item, dequeued, signal))

common/src/zygote.ts

@@ -0,0 +1,77 @@
+import sdk, { PluginFork } from '@scrypted/sdk';
+import worker_threads from 'worker_threads';
+import { createAsyncQueue } from './async-queue';
+import os from 'os';
+
+export type Zygote<T> = () => PluginFork<T>;
+
+export function createZygote<T>(): Zygote<T> {
+    if (!worker_threads.isMainThread)
+        return;
+
+    let zygote = sdk.fork<T>();
+    function* next() {
+        while (true) {
+            const cur = zygote;
+            zygote = sdk.fork<T>();
+            yield cur;
+        }
+    }
+
+    const gen = next();
+    return () => gen.next().value as PluginFork<T>;
+}
+
+
+export function createZygoteWorkQueue<T>(maxWorkers: number = os.cpus().length >> 1) {
+    const queue = createAsyncQueue<(doWork: (fork: PluginFork<T>) => Promise<any>) => Promise<any>>();
+    let forks = 0;
+
+    return async <R>(doWork: (fork: PluginFork<T>) => Promise<R>): Promise<R> => {
+        const check = queue.take();
+        if (check)
+            return check(doWork);
+
+        if (maxWorkers && forks < maxWorkers) {
+            let exited = false;
+            const controller = new AbortController();
+            // necessary to prevent unhandledrejection errors
+            controller.signal.addEventListener('abort', () => { });
+            const fork = sdk.fork<T>();
+            forks++;
+            fork.worker.on('exit', () => {
+                forks--;
+                exited = true;
+                controller.abort();
+            });
+
+            let timeout: NodeJS.Timeout;
+            const queueFork = () => {
+                clearTimeout(timeout);
+                timeout = setTimeout(() => {
+                    // keep one alive.
+                    if (forks === 1)
+                        return;
+                    fork.worker.terminate();
+                }, 30000);
+
+                queue.submit(async v2 => {
+                    clearTimeout(timeout);
+                    try {
+                        return await v2(fork);
+                    }
+                    finally {
+                        if (!exited) {
+                            queueFork();
+                        }
+                    }
+                }, controller.signal);
+            }
+
+            queueFork();
+        }
+
+        const d = await queue.dequeue();
+        return d(doWork);
+    };
+}

install/config.yaml

@@ -1,6 +1,6 @@
 # Home Assistant Addon Configuration
 name: Scrypted
-version: "18-jammy-full.s6-v0.57.0"
+version: "18-jammy-full.s6-v0.68.0"
 slug: scrypted
 description: Scrypted is a high performance home video integration and automation platform
 url: "https://github.com/koush/scrypted"

packages/cli/.vscode/launch.json

@@ -21,9 +21,7 @@
             ],
             "preLaunchTask": "npm: build",
             "args": [
-                "ffplay",
-                "Baby [email protected]",
-                "getVideoStream",
+                "shell",
             ],
             "sourceMaps": true,
             "resolveSourceMapLocations": [