perf(Subject): subscription, unsubscription, and nexting improved ~4x (#7240)

* fix(subject): Faster subscriptions management in Subject

The current implementation of the Subject was backing itself on an array of observers. When one observer unsubscribed itself, it was performing a linear scan on the array of observers and dropping the item from the array of observers. The algorithm complexity of this operation was O(n) with n being the number of observers connected to the Subject.

It caused code like:

```js
const allSubscriptions = [];
const source = new Subject();
for (let index = 0 ; index !== 1_000_000 ; ++index) {
  allSubscriptions.push(source.subscribe()); // rather quick
}
for (const subscription of allSubscriptions) {
  subscription.unsubscribe(); // taking very long
}
```

The proposed approach consists into changing our backing collection (an array) into a Map. But we lose somehow the set capability on subject.observers (might possibly be patched in some way).

Following that change the command `cross-env TS_NODE_PROJECT=tsconfig.mocha.json mocha --config spec/support/.mocharc.js "spec/**/Subje*-spec.ts"` passed to 452ms from 10s.

* refactor(Subject): some perf optimizations

+ Loops over indices of the observers array rather than creating a new iterator and using for-of for each emission
+ Removes unnecessary dirty boolean in favor of clearing the snapshot, which frees up memory a little
+ Centralizes observers list teardown and ensures it's called during `unsubscribe()`.

---------

Co-authored-by: Ben Lesh <ben@benlesh.com>

Author: dubzzz

spec/Subject-spec.ts

@@ -1,5 +1,5 @@
 import { expect } from 'chai';
-import { Subject, ObjectUnsubscribedError, Observable, AsyncSubject, Observer, of, config, throwError, concat } from 'rxjs';
+import { Subject, ObjectUnsubscribedError, Observable, AsyncSubject, Observer, of, config, throwError, concat, Subscription } from 'rxjs';
 import { AnonymousSubject } from 'rxjs/internal/Subject';
 import { catchError, delay, map, mergeMap } from 'rxjs/operators';
 import { TestScheduler } from 'rxjs/testing';
@@ -683,6 +683,47 @@ describe('Subject', () => {
       expect(true).to.be.true;
     });
   });
+
+  describe('many subscribers', () => {
+    it('should be able to subscribe and unsubscribe huge amounts of subscribers', () => {
+      let numResultsReceived = 0;
+      const allSubscriptions: Subscription[] = [];
+      const source = new Subject<number>();
+      const numSubscribers = 100000;
+      for (let index = 0 ; index !== numSubscribers ; ++index) {
+        allSubscriptions.push(source.subscribe(() => {
+          ++numResultsReceived;
+        }));
+      }
+      expect(numResultsReceived).to.eq(0);
+      expect(source.observed).to.be.true;
+      source.next(42);
+      expect(numResultsReceived).to.eq(numSubscribers);
+      expect(source.observed).to.be.true;
+      for (const subscription of allSubscriptions) {
+        subscription.unsubscribe();
+      }
+      expect(numResultsReceived).to.eq(numSubscribers);
+      expect(source.observed).to.be.false;
+    });
+  });
+
+  describe('re-rentrant subscribers', () => {
+    it('should handle re-entrant subscribers', () => {
+      const seenValues: number[] = [];
+      const source = new Subject<number>();
+      source.subscribe((value) => {
+        seenValues.push(value);
+        source.subscribe(nestedValue => {
+          seenValues.push(nestedValue);
+        });
+      });
+      source.next(1);
+      source.next(2);
+      source.next(3);
+      expect(seenValues).to.deep.eq([1, 2, 2, 3, 3, 3]);
+    });
+  });
 });
 
 describe('AnonymousSubject', () => {

src/internal/Subject.ts

@@ -3,7 +3,6 @@ import { Subscriber } from './Subscriber';
 import { Subscription } from './Subscription';
 import { Observer, SubscriptionLike, TeardownLogic } from './types';
 import { ObjectUnsubscribedError } from './util/ObjectUnsubscribedError';
-import { arrRemove } from './util/arrRemove';
 
 /**
  * A Subject is a special type of Observable that allows values to be
@@ -15,14 +14,26 @@ import { arrRemove } from './util/arrRemove';
 export class Subject<T> extends Observable<T> implements SubscriptionLike {
   closed = false;
 
-  private currentObservers: Observer<T>[] | null = null;
+  private currentObservers = new Map<Subscription, Observer<T>>();
+
+  /**
+   * This is used to track a known array of observers, so we don't have to
+   * clone them while iterating to prevent reentrant behaviors.
+   * (for example, what if the subject is subscribed to when nexting to an observer)
+   */
+  private observerSnapshot: Observer<T>[] | undefined;
+
+  /** @internal */
+  get observers(): Observer<T>[] {
+    return (this.observerSnapshot ??= Array.from(this.currentObservers.values()));
+  }
 
-  /** @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. */
-  observers: Observer<T>[] = [];
   /** @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. */
   isStopped = false;
+
   /** @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. */
   hasError = false;
+
   /** @deprecated Internal implementation detail, do not use directly. Will be made internal in v8. */
   thrownError: any = null;
 
@@ -48,14 +59,18 @@ export class Subject<T> extends Observable<T> implements SubscriptionLike {
     }
   }
 
+  protected _clearObservers() {
+    this.currentObservers.clear();
+    this.observerSnapshot = undefined;
+  }
+
   next(value: T) {
     this._throwIfClosed();
     if (!this.isStopped) {
-      if (!this.currentObservers) {
-        this.currentObservers = Array.from(this.observers);
-      }
-      for (const observer of this.currentObservers) {
-        observer.next(value);
+      const { observers } = this;
+      const len = observers.length;
+      for (let i = 0; i < len; i++) {
+        observers[i].next(value);
       }
     }
   }
@@ -66,9 +81,11 @@ export class Subject<T> extends Observable<T> implements SubscriptionLike {
       this.hasError = this.isStopped = true;
       this.thrownError = err;
       const { observers } = this;
-      while (observers.length) {
-        observers.shift()!.error(err);
+      const len = observers.length;
+      for (let i = 0; i < len; i++) {
+        observers[i].error(err);
       }
+      this._clearObservers();
     }
   }
 
@@ -77,19 +94,21 @@ export class Subject<T> extends Observable<T> implements SubscriptionLike {
     if (!this.isStopped) {
       this.isStopped = true;
       const { observers } = this;
-      while (observers.length) {
-        observers.shift()!.complete();
+      const len = observers.length;
+      for (let i = 0; i < len; i++) {
+        observers[i].complete();
       }
+      this._clearObservers();
     }
   }
 
   unsubscribe() {
     this.isStopped = this.closed = true;
-    this.observers = this.currentObservers = null!;
+    this._clearObservers();
   }
 
   get observed() {
-    return this.observers?.length > 0;
+    return this.currentObservers.size > 0;
   }
 
   /** @internal */
@@ -107,16 +126,17 @@ export class Subject<T> extends Observable<T> implements SubscriptionLike {
 
   /** @internal */
   protected _innerSubscribe(subscriber: Subscriber<any>) {
-    const { hasError, isStopped, observers } = this;
+    const { hasError, isStopped, currentObservers } = this;
     if (hasError || isStopped) {
       return Subscription.EMPTY;
     }
-    this.currentObservers = null;
-    observers.push(subscriber);
-    return new Subscription(() => {
-      this.currentObservers = null;
-      arrRemove(observers, subscriber);
+    const subscription = new Subscription(() => {
+      currentObservers.delete(subscription);
+      this.observerSnapshot = undefined;
     });
+    currentObservers.set(subscription, subscriber);
+    this.observerSnapshot = undefined;
+    return subscription;
   }
 
   /** @internal */