Merge pull request #57 from aesy-engineering/weighted_semaphore

Weighted semaphore

src/Mutex.ts

@@ -6,6 +6,10 @@ class Mutex implements MutexInterface {
         this._semaphore = new Semaphore(1, cancelError);
     }
 
+    async weightedAcquire(): Promise<MutexInterface.Releaser> {
+        return this.acquire()
+    }
+
     async acquire(): Promise<MutexInterface.Releaser> {
         const [, releaser] = await this._semaphore.acquire();
 

src/MutexInterface.ts

@@ -1,6 +1,8 @@
 interface MutexInterface {
     acquire(): Promise<MutexInterface.Releaser>;
 
+    weightedAcquire(): Promise<MutexInterface.Releaser>;
+
     runExclusive<T>(callback: MutexInterface.Worker<T>): Promise<T>;
 
     waitForUnlock(): Promise<void>;

src/Semaphore.ts

@@ -20,12 +20,16 @@ class Semaphore implements SemaphoreInterface {
     }
 
     acquire(): Promise<[number, SemaphoreInterface.Releaser]> {
+        return this.weightedAcquire(1)
+    }
+
+    weightedAcquire(weight: number): Promise<[number, SemaphoreInterface.Releaser]> {
         const locked = this.isLocked();
         const ticketPromise = new Promise<[number, SemaphoreInterface.Releaser]>((resolve, reject) =>
             this._queue.push({ resolve, reject })
         );
 
-        if (!locked) this._dispatch();
+        if (!locked) this._dispatch(weight);
 
         return ticketPromise;
     }
@@ -75,7 +79,7 @@ class Semaphore implements SemaphoreInterface {
         this._queue = [];
     }
 
-    private _dispatch(): void {
+    private _dispatch(weight: number): void {
         const nextTicket = this._queue.shift();
 
         if (!nextTicket) return;
@@ -85,13 +89,19 @@ class Semaphore implements SemaphoreInterface {
             if (released) return;
 
             released = true;
-            this._value++;
+            this._value = this._value + weight
             this._resolveWaiters();
 
-            this._dispatch();
+            this._dispatch(weight);
         };
 
-        nextTicket.resolve([this._value--, this._currentReleaser]);
+        nextTicket.resolve([this._valueMinusMinusWeight(weight), this._currentReleaser]);
+    }
+
+    private _valueMinusMinusWeight(weight: number): number {
+        const oldValue = this._value
+        this._value = this._value - weight
+        return oldValue
     }
 
     private _resolveWaiters() {

src/SemaphoreInterface.ts

@@ -1,6 +1,8 @@
 interface SemaphoreInterface {
     acquire(): Promise<[number, SemaphoreInterface.Releaser]>;
 
+    weightedAcquire(weight: number): Promise<[number, SemaphoreInterface.Releaser]>;  
+
     runExclusive<T>(callback: SemaphoreInterface.Worker<T>): Promise<T>;
 
     waitForUnlock(): Promise<void>;

src/withTimeout.ts

@@ -7,7 +7,7 @@ export function withTimeout(semaphore: SemaphoreInterface, timeout: number, time
 // eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types
 export function withTimeout(sync: MutexInterface | SemaphoreInterface, timeout: number, timeoutError = E_TIMEOUT) {
     return {
-        acquire: (): Promise<MutexInterface.Releaser | [number, SemaphoreInterface.Releaser]> =>
+        weightedAcquire: (weight: number): Promise<MutexInterface.Releaser | [number, SemaphoreInterface.Releaser]> =>
             new Promise(async (resolve, reject) => {
                 let isTimeout = false;
 
@@ -17,7 +17,7 @@ export function withTimeout(sync: MutexInterface | SemaphoreInterface, timeout:
                 }, timeout);
 
                 try {
-                    const ticket = await sync.acquire();
+                    const ticket = await sync.weightedAcquire(weight);
 
                     if (isTimeout) {
                         const release = Array.isArray(ticket) ? ticket[1] : ticket;
@@ -36,6 +36,10 @@ export function withTimeout(sync: MutexInterface | SemaphoreInterface, timeout:
                 }
             }),
 
+        acquire(): Promise<MutexInterface.Releaser | [number, SemaphoreInterface.Releaser]> {
+            return this.weightedAcquire(1)
+        },
+
         async runExclusive<T>(callback: (value?: number) => Promise<T> | T): Promise<T> {
             let release: () => void = () => undefined;
 

test/semaphore.ts

@@ -29,6 +29,28 @@ export const semaphoreSuite = (factory: (maxConcurrency: number, err?: Error) =>
         assert.deepStrictEqual(values.sort(), [1, 2]);
     });
 
+    test('weightedAcquire does not block while the semaphore has not reached zero', async () => {
+        const values: Array<number> = [];
+
+        semaphore.weightedAcquire(1).then(([value]) => values.push(value));
+        semaphore.weightedAcquire(1).then(([value]) => values.push(value));
+
+        await clock.tickAsync(0);
+
+        assert.deepStrictEqual(values.sort(), [1, 2]);
+    });
+
+    test('weightedAcquire does block while the semaphore has reached zero', async () => {
+        const values: Array<number> = [];
+
+        semaphore.weightedAcquire(2).then(([value]) => values.push(value));
+        semaphore.weightedAcquire(1).then(([value]) => values.push(value));
+
+        await clock.tickAsync(0);
+
+        assert.deepStrictEqual(values.sort(), [2]);
+    });
+
     test('acquire blocks when the semaphore has reached zero', async () => {
         const values: Array<number> = [];
 
@@ -48,6 +70,33 @@ export const semaphoreSuite = (factory: (maxConcurrency: number, err?: Error) =>
         assert.deepStrictEqual(values.sort(), [1, 1, 2]);
     });
 
+    test('weightedAcquire blocks when the semaphore has reached zero and unblocked on release', async () => {
+        const values: Array<number> = [];
+
+        semaphore.weightedAcquire(1).then(([value, release]) => { 
+            values.push(value)
+            release() 
+        });
+        semaphore.weightedAcquire(3).then(([value, release]) => {
+            values.push(value);
+            setTimeout(() => {
+                release();
+            }, 100);
+        });
+        semaphore.weightedAcquire(1).then(([value, release]) => {
+            values.push(value)
+            release();
+        });
+
+        await clock.tickAsync(0);
+
+        assert.deepStrictEqual(values.sort(), [-1, 1, 2]);
+
+        await clock.runAllAsync();
+
+        assert.deepStrictEqual(values.sort(), [-1, 1, 2]);
+    });
+
     test('the semaphore increments again after a release', async () => {
         semaphore.acquire().then(([, release]) => setTimeout(release, 100));
         semaphore.acquire().then(([, release]) => setTimeout(release, 200));