Save an allocation during inner observer creation in Switch.

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/Program.cs

@@ -13,7 +13,8 @@ static void Main()
         {
             var switcher = new BenchmarkSwitcher(new[] {
                 typeof(ZipBenchmark),
-                typeof(CombineLatestBenchmark)
+                typeof(CombineLatestBenchmark),
+                typeof(SwitchBenchmark)
             });
 
             switcher.Run();

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/SwitchBenchmark.cs

@@ -0,0 +1,25 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+using System.Reactive.Linq;
+using BenchmarkDotNet.Attributes;
+using System.Reactive.Threading.Tasks;
+using System.Threading.Tasks;
+
+namespace Benchmarks.System.Reactive
+{
+    [MemoryDiagnoser]
+    public class SwitchBenchmark
+    {
+        [Benchmark]
+        public async Task Switch_10000_Sources()
+        {
+            await Observable
+                .Range(1, 10000)
+                .Select(x => Observable.Return(x))
+                .Switch()
+                .ToTask();
+        }
+    }
+}

Rx.NET/Source/benchmarks/Benchmarks.System.Reactive/ZipBenchmark.cs

@@ -77,7 +77,7 @@ public void Zip_AllCompleted7()
         {
             _zipTest.Zip_AllCompleted7();
         }
-        
+
         [Benchmark]
         public void Zip_AllCompleted8()
         {

Rx.NET/Source/src/System.Reactive/Linq/Observable/Switch.cs

@@ -50,9 +50,10 @@ public override void OnNext(IObservable<TSource> value)
                     _hasLatest = true;
                 }
 
-                var d = new SingleAssignmentDisposable();
-                Disposable.TrySetSerial(ref _innerSerialDisposable, d);
-                d.Disposable = value.SubscribeSafe(new InnerObserver(this, id, d));
+                var innerObserver = new InnerObserver(this, id);
+
+                Disposable.TrySetSerial(ref _innerSerialDisposable, innerObserver);
+                innerObserver.SetResource(value.SubscribeSafe(innerObserver));
             }
 
             public override void OnError(Exception error)
@@ -75,20 +76,18 @@ public override void OnCompleted()
                 }
             }
 
-            private sealed class InnerObserver : IObserver<TSource>
+            private sealed class InnerObserver : SafeObserver<TSource>
             {
                 private readonly _ _parent;
                 private readonly ulong _id;
-                private readonly IDisposable _self;
 
-                public InnerObserver(_ parent, ulong id, IDisposable self)
+                public InnerObserver(_ parent, ulong id)
                 {
                     _parent = parent;
                     _id = id;
-                    _self = self;
                 }
 
-                public void OnNext(TSource value)
+                public override void OnNext(TSource value)
                 {
                     lock (_parent._gate)
                     {
@@ -99,11 +98,11 @@ public void OnNext(TSource value)
                     }
                 }
 
-                public void OnError(Exception error)
+                public override void OnError(Exception error)
                 {
                     lock (_parent._gate)
                     {
-                        _self.Dispose();
+                        Dispose();
 
                         if (_parent._latest == _id)
                         {
@@ -112,11 +111,11 @@ public void OnError(Exception error)
                     }
                 }
 
-                public void OnCompleted()
+                public override void OnCompleted()
                 {
                     lock (_parent._gate)
                     {
-                        _self.Dispose();
+                        Dispose();
 
                         if (_parent._latest == _id)
                         {