fix: use cheaper async mutex in place of 1-queue

Author: Nerivec

src/adapter/ember/ezsp/ezsp.ts

@@ -1,8 +1,7 @@
 /* v8 ignore start */
 
 import EventEmitter from "node:events";
-
-import {Queue} from "../../../utils";
+import {AsyncMutex} from "../../../utils/async-mutex";
 import {logger} from "../../../utils/logger";
 import * as ZSpec from "../../../zspec";
 import type {Eui64, ExtendedPanId, NodeId, PanId} from "../../../zspec/tstypes";
@@ -239,7 +238,7 @@ export class Ezsp extends EventEmitter<EmberEzspEventMap> {
     private frameSequence: number;
     /** Sequence used for EZSP send() tagging. static uint8_t */
     private sendSequence: number;
-    private readonly queue: Queue;
+    private readonly queue: AsyncMutex;
     /** Awaiting response resolve/timer struct. undefined if not waiting for response. */
     private responseWaiter?: EzspWaiter;
 
@@ -254,7 +253,7 @@ export class Ezsp extends EventEmitter<EmberEzspEventMap> {
         this.callbackFrameContents = Buffer.alloc(EZSP_MAX_FRAME_LENGTH);
         this.callbackBuffalo = new EzspBuffalo(this.callbackFrameContents, 0);
 
-        this.queue = new Queue(1);
+        this.queue = new AsyncMutex();
         this.ash = new UartAsh(options);
 
         this.version = 0;
@@ -535,7 +534,7 @@ export class Ezsp extends EventEmitter<EmberEzspEventMap> {
             return EzspStatus.ERROR_COMMAND_TOO_LONG;
         }
 
-        return await this.queue.execute<EzspStatus>(async () => {
+        return await this.queue.run<EzspStatus>(async () => {
             let status: EzspStatus = EzspStatus.ASH_ERROR_TIMEOUTS; // will be overwritten below as necessary
             const frameId = sendBuffalo.getFrameId();
             const frameString = `[FRAME: ID=${frameId}:"${EzspFrameID[frameId]}" Seq=${sequence} Len=${length}]`;

src/adapter/ezsp/driver/uart.ts

@@ -2,8 +2,8 @@
 
 import {EventEmitter} from "node:events";
 import net from "node:net";
-
-import {Queue, Waitress, wait} from "../../../utils";
+import {Waitress, wait} from "../../../utils";
+import {AsyncMutex} from "../../../utils/async-mutex";
 import {logger} from "../../../utils/logger";
 import {SerialPort} from "../../serialPort";
 import type {SerialPortOptions} from "../../tstype";
@@ -45,12 +45,12 @@ export class SerialDriver extends EventEmitter {
     private ackSeq = 0; // next number after the last accepted frame
     private rejectCondition = false;
     private waitress: Waitress<EZSPPacket, EZSPPacketMatcher>;
-    private queue: Queue;
+    private queue: AsyncMutex;
 
     constructor() {
         super();
         this.initialized = false;
-        this.queue = new Queue(1);
+        this.queue = new AsyncMutex();
         this.waitress = new Waitress<EZSPPacket, EZSPPacketMatcher>(this.waitressValidator, this.waitressTimeoutFormatter);
         this.writer = new Writer();
         this.parser = new Parser();
@@ -332,7 +332,7 @@ export class SerialDriver extends EventEmitter {
         this.sendSeq = 0;
         this.recvSeq = 0;
 
-        return await this.queue.execute<void>(async (): Promise<void> => {
+        return await this.queue.run<void>(async (): Promise<void> => {
             try {
                 logger.debug("--> Write reset", NS);
                 const waiter = this.waitFor(-1, 10000);
@@ -410,7 +410,7 @@ export class SerialDriver extends EventEmitter {
         const nextSeq = this.sendSeq;
         const ackSeq = this.recvSeq;
 
-        return await this.queue.execute<void>(async (): Promise<void> => {
+        return await this.queue.run<void>(async (): Promise<void> => {
             const randData = NpiFrame.makeRandomizedBuffer(data);
 
             try {

src/adapter/zboss/uart.ts

@@ -2,8 +2,8 @@
 
 import EventEmitter from "node:events";
 import {Socket} from "node:net";
-
-import {Queue, Waitress, wait} from "../../utils";
+import {AsyncMutex} from "src/utils/async-mutex";
+import {Waitress, wait} from "../../utils";
 import {logger} from "../../utils/logger";
 import {SerialPort} from "../serialPort";
 import type {SerialPortOptions} from "../tstype";
@@ -27,7 +27,7 @@ export class ZBOSSUart extends EventEmitter {
     private recvSeq = 0; // next frame number to receive
     private ackSeq = 0; // next number after the last accepted frame
     private waitress: Waitress<number, number>;
-    private queue: Queue;
+    private queue: AsyncMutex;
     public inReset = false;
 
     constructor(options: SerialPortOptions) {
@@ -39,7 +39,7 @@ export class ZBOSSUart extends EventEmitter {
         this.writer = new ZBOSSWriter();
         this.reader = new ZBOSSReader();
         this.reader.on("data", this.onPackage.bind(this));
-        this.queue = new Queue(1);
+        this.queue = new AsyncMutex();
         this.waitress = new Waitress<number, number>(this.waitressValidator, this.waitressTimeoutFormatter);
     }
 
@@ -307,7 +307,7 @@ export class ZBOSSUart extends EventEmitter {
         const nextSeq = this.sendSeq;
         const ackSeq = this.recvSeq;
 
-        return await this.queue.execute<void>(async (): Promise<void> => {
+        return await this.queue.run<void>(async (): Promise<void> => {
             try {
                 logger.debug(`--> DATA (${seq},${ackSeq},0): ${data.toString("hex")}`, NS);
                 if (!isACK) {

src/adapter/zigate/driver/zigate.ts

@@ -3,10 +3,8 @@
 import assert from "node:assert";
 import {EventEmitter} from "node:events";
 import net from "node:net";
-
 import {DelimiterParser} from "@serialport/parser-delimiter";
-
-import {Queue} from "../../../utils";
+import {AsyncMutex} from "../../../utils/async-mutex";
 import {logger} from "../../../utils/logger";
 import {Waitress} from "../../../utils/waitress";
 import * as ZSpec from "../../../zspec";
@@ -79,7 +77,7 @@ export default class ZiGate extends EventEmitter<ZiGateEventMap> {
     private parser?: EventEmitter;
     private serialPort?: SerialPort;
     private socketPort?: net.Socket;
-    private queue: Queue;
+    private queue: AsyncMutex;
 
     public portWrite?: SerialPort | net.Socket;
     private waitress: Waitress<ZiGateObject, WaitressMatcher>;
@@ -92,7 +90,7 @@ export default class ZiGate extends EventEmitter<ZiGateEventMap> {
         // XXX: not used?
         // this.rtscts = typeof serialPortOptions.rtscts === 'boolean' ? serialPortOptions.rtscts : false;
         this.initialized = false;
-        this.queue = new Queue(1);
+        this.queue = new AsyncMutex();
 
         this.waitress = new Waitress<ZiGateObject, WaitressMatcher>(this.waitressValidator, this.waitressTimeoutFormatter);
         this.zdoWaitress = new Waitress<ZdoWaitressPayload, ZdoWaitressMatcher>(this.zdoWaitressValidator, this.waitressTimeoutFormatter);
@@ -107,7 +105,7 @@ export default class ZiGate extends EventEmitter<ZiGateEventMap> {
     ): Promise<ZiGateObject> {
         const waiters: Promise<ZiGateObject>[] = [];
         const waitersId: number[] = [];
-        return await this.queue.execute(async () => {
+        return await this.queue.run(async () => {
             try {
                 logger.debug(
                     () => `Send command \x1b[32m>>>> ${ZiGateCommandCode[code]} 0x${zeroPad(code)} <<<<\x1b[0m \nPayload: ${JSON.stringify(payload)}`,
@@ -163,7 +161,7 @@ export default class ZiGate extends EventEmitter<ZiGateEventMap> {
     }
 
     public async requestZdo(clusterId: Zdo.ClusterId, payload: Buffer): Promise<boolean> {
-        return await this.queue.execute(async () => {
+        return await this.queue.run(async () => {
             const commandCode = ZDO_REQ_CLUSTER_ID_TO_ZIGATE_COMMAND_ID[clusterId];
             assert(commandCode !== undefined, `ZDO cluster ID '${clusterId}' not supported.`);
             const ruleStatus: ZiGateResponseMatcher = [

src/utils/async-mutex.ts

@@ -0,0 +1,31 @@
+export class AsyncMutex {
+    #locked = false;
+    readonly #queue: Array<() => void> = [];
+
+    get count() {
+        return this.#queue.length;
+    }
+
+    async run<T>(fn: () => Promise<T>): Promise<T> {
+        if (this.#locked) {
+            await new Promise<void>((resolve) => this.#queue.push(resolve));
+        }
+
+        this.#locked = true;
+
+        try {
+            return await fn();
+        } finally {
+            this.#locked = false;
+            const next = this.#queue.shift();
+
+            if (next) {
+                next();
+            }
+        }
+    }
+
+    clear() {
+        this.#queue.length = 0;
+    }
+}

test/utils.test.ts

@@ -1,6 +1,7 @@
 import {describe, expect, it, vi} from "vitest";
 import {checkInstallCode} from "../src/controller/helpers/installCodes";
 import {Queue, Utils, Waitress, wait} from "../src/utils";
+import {AsyncMutex} from "../src/utils/async-mutex";
 import {logger, setLogger} from "../src/utils/logger";
 
 const mockLogger = {
@@ -201,6 +202,52 @@ describe("Utils", () => {
         expect(queue.count()).toBe(5);
     });
 
+    it("Test async mutex", async () => {
+        vi.useFakeTimers();
+
+        const queue = new AsyncMutex();
+
+        void queue.run(async () => {
+            await new Promise((resolve) => setTimeout(resolve, 1000));
+        });
+
+        void queue.run(async () => {
+            await new Promise((resolve) => setTimeout(resolve, 1000));
+        });
+
+        await vi.advanceTimersByTimeAsync(500);
+        expect(queue.count).toStrictEqual(1); // first has ran but still pending return, second is queued
+
+        await vi.advanceTimersByTimeAsync(1000);
+        expect(queue.count).toStrictEqual(0);
+
+        void queue.run(async () => {
+            await new Promise((resolve) => setTimeout(resolve, 1000));
+        });
+
+        expect(queue.count).toStrictEqual(1); // second has ran but still pending return, third is queued
+        await vi.advanceTimersByTimeAsync(1600);
+        expect(queue.count).toStrictEqual(0);
+
+        //-- clear
+
+        void queue.run(async () => {
+            await new Promise((resolve) => setTimeout(resolve, 1000));
+        });
+        void queue.run(async () => {
+            await new Promise((resolve) => setTimeout(resolve, 1000));
+        });
+
+        expect(queue.count).toStrictEqual(1);
+
+        queue.clear();
+
+        expect(queue.count).toStrictEqual(0);
+        await vi.runOnlyPendingTimersAsync(); // cleanup
+
+        vi.useRealTimers();
+    });
+
     it("Logs", () => {
         const debugSpy = vi.spyOn(console, "debug");
         const infoSpy = vi.spyOn(console, "info");