Replace async queue with mutex lock for mute operations (#632)

* Replace async queue with mutex lock for mute operations * decouple pauseUpstream and mute lock * Create .changeset/twelve-maps-double.md
livekit · Mar 31, 2023 · 111971d · 111971d
1 parent f085fbe
commit 111971d
Show file tree

Hide file tree

Showing 5 changed files with 56 additions and 34 deletions.
diff --git a/.changeset/twelve-maps-double.md b/.changeset/twelve-maps-double.md
@@ -0,0 +1,5 @@
+---
+"livekit-client": patch
+---
+
+Replace async queue with mutex lock for mute operations
diff --git a/src/room/track/LocalAudioTrack.ts b/src/room/track/LocalAudioTrack.ts
@@ -39,20 +39,24 @@ export default class LocalAudioTrack extends LocalTrack {
   }
 
   async mute(): Promise<LocalAudioTrack> {
-    await this.muteQueue.run(async () => {
+    const unlock = await this.muteLock.lock();
+    try {
       // disabled special handling as it will cause BT headsets to switch communication modes
       if (this.source === Track.Source.Microphone && this.stopOnMute && !this.isUserProvided) {
         log.debug('stopping mic track');
         // also stop the track, so that microphone indicator is turned off
         this._mediaStreamTrack.stop();
       }
       await super.mute();
-    });
-    return this;
+      return this;
+    } finally {
+      unlock();
+    }
   }
 
   async unmute(): Promise<LocalAudioTrack> {
-    await this.muteQueue.run(async () => {
+    const unlock = await this.muteLock.lock();
+    try {
       if (
         this.source === Track.Source.Microphone &&
         (this.stopOnMute || this._mediaStreamTrack.readyState === 'ended') &&
@@ -62,8 +66,11 @@ export default class LocalAudioTrack extends LocalTrack {
         await this.restartTrack();
       }
       await super.unmute();
-    });
-    return this;
+
+      return this;
+    } finally {
+      unlock();
+    }
   }
 
   async restartTrack(options?: AudioCaptureOptions) {

diff --git a/src/room/track/LocalTrack.ts b/src/room/track/LocalTrack.ts
@@ -1,9 +1,14 @@
-import Queue from 'async-await-queue';
 import log from '../../logger';
 import DeviceManager from '../DeviceManager';
 import { TrackInvalidError } from '../errors';
 import { TrackEvent } from '../events';
-import { getEmptyAudioStreamTrack, getEmptyVideoStreamTrack, isMobile, sleep } from '../utils';
+import {
+  getEmptyAudioStreamTrack,
+  getEmptyVideoStreamTrack,
+  isMobile,
+  Mutex,
+  sleep,
+} from '../utils';
 import type { VideoCodec } from './options';
 import { attachToElement, detachTrack, Track } from './Track';
 
@@ -22,7 +27,9 @@ export default abstract class LocalTrack extends Track {
 
   protected providedByUser: boolean;
 
-  protected muteQueue: Queue;
+  protected muteLock: Mutex;
+
+  protected pauseUpstreamLock: Mutex;
 
   /**
    *
@@ -42,7 +49,8 @@ export default abstract class LocalTrack extends Track {
     this.constraints = constraints ?? mediaTrack.getConstraints();
     this.reacquireTrack = false;
     this.providedByUser = userProvidedTrack;
-    this.muteQueue = new Queue();
+    this.muteLock = new Mutex();
+    this.pauseUpstreamLock = new Mutex();
   }
 
   get id(): string {
@@ -246,7 +254,8 @@ export default abstract class LocalTrack extends Track {
   };
 
   async pauseUpstream() {
-    this.muteQueue.run(async () => {
+    const unlock = await this.pauseUpstreamLock.lock();
+    try {
       if (this._isUpstreamPaused === true) {
         return;
       }
@@ -260,11 +269,14 @@ export default abstract class LocalTrack extends Track {
       const emptyTrack =
         this.kind === Track.Kind.Audio ? getEmptyAudioStreamTrack() : getEmptyVideoStreamTrack();
       await this.sender.replaceTrack(emptyTrack);
-    });
+    } finally {
+      unlock();
+    }
   }
 
   async resumeUpstream() {
-    this.muteQueue.run(async () => {
+    const unlock = await this.pauseUpstreamLock.lock();
+    try {
       if (this._isUpstreamPaused === false) {
         return;
       }
@@ -276,7 +288,9 @@ export default abstract class LocalTrack extends Track {
       this.emit(TrackEvent.UpstreamResumed, this);
 
       await this.sender.replaceTrack(this._mediaStreamTrack);
-    });
+    } finally {
+      unlock();
+    }
   }
 
   protected abstract monitorSender(): void;

diff --git a/src/room/track/LocalVideoTrack.ts b/src/room/track/LocalVideoTrack.ts
@@ -97,26 +97,32 @@ export default class LocalVideoTrack extends LocalTrack {
   }
 
   async mute(): Promise<LocalVideoTrack> {
-    await this.muteQueue.run(async () => {
+    const unlock = await this.muteLock.lock();
+    try {
       if (this.source === Track.Source.Camera && !this.isUserProvided) {
         log.debug('stopping camera track');
         // also stop the track, so that camera indicator is turned off
         this._mediaStreamTrack.stop();
       }
       await super.mute();
-    });
-    return this;
+      return this;
+    } finally {
+      unlock();
+    }
   }
 
   async unmute(): Promise<LocalVideoTrack> {
-    await this.muteQueue.run(async () => {
+    const unlock = await this.muteLock.lock();
+    try {
       if (this.source === Track.Source.Camera && !this.isUserProvided) {
         log.debug('reacquiring camera track');
         await this.restartTrack();
       }
       await super.unmute();
-    });
-    return this;
+      return this;
+    } finally {
+      unlock();
+    }
   }
 
   async getSenderStats(): Promise<VideoSenderStats[]> {

diff --git a/yarn.lock b/yarn.lock
@@ -3228,20 +3228,10 @@ camelcase@^6.2.0:
   resolved "https://registry.yarnpkg.com/camelcase/-/camelcase-6.2.1.tgz#250fd350cfd555d0d2160b1d51510eaf8326e86e"
   integrity sha512-tVI4q5jjFV5CavAU8DXfza/TJcZutVKo/5Foskmsqcm0MsL91moHvwiGNnqaa2o6PF/7yT5ikDRcVcl8Rj6LCA==
 
-caniuse-lite@^1.0.30001280:
-  version "1.0.30001361"
-  resolved "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001361.tgz"
-  integrity sha512-ybhCrjNtkFji1/Wto6SSJKkWk6kZgVQsDq5QI83SafsF6FXv2JB4df9eEdH6g8sdGgqTXrFLjAxqBGgYoU3azQ==
-
-caniuse-lite@^1.0.30001366:
-  version "1.0.30001368"
-  resolved "https://registry.yarnpkg.com/caniuse-lite/-/caniuse-lite-1.0.30001368.tgz#c5c06381c6051cd863c45021475434e81936f713"
-  integrity sha512-wgfRYa9DenEomLG/SdWgQxpIyvdtH3NW8Vq+tB6AwR9e56iOIcu1im5F/wNdDf04XlKHXqIx4N8Jo0PemeBenQ==
-
-caniuse-lite@^1.0.30001400:
-  version "1.0.30001414"
-  resolved "https://registry.yarnpkg.com/caniuse-lite/-/caniuse-lite-1.0.30001414.tgz#5f1715e506e71860b4b07c50060ea6462217611e"
-  integrity sha512-t55jfSaWjCdocnFdKQoO+d2ct9C59UZg4dY3OnUlSZ447r8pUtIKdp0hpAzrGFultmTC+Us+KpKi4GZl/LXlFg==
+caniuse-lite@^1.0.30001280, caniuse-lite@^1.0.30001366, caniuse-lite@^1.0.30001400:
+  version "1.0.30001472"
+  resolved "https://registry.npmjs.org/caniuse-lite/-/caniuse-lite-1.0.30001472.tgz"
+  integrity sha512-xWC/0+hHHQgj3/vrKYY0AAzeIUgr7L9wlELIcAvZdDUHlhL/kNxMdnQLOSOQfP8R51ZzPhmHdyMkI0MMpmxCfg==
 
 case-anything@^2.1.10:
   version "2.1.10"