(Question) The performance of ForkJoin.

[Warpten]

Question

What is the context of your question or problem?

This is more of a performance-related question, if anything.

What is the question or problem you try to solve?

Would it be worth it to move away from locks in the implementation of ForkJoin, seeing as that would only benefit large quantities of observables (> 1000) ?

What were the (original) requirements you tried to solve?

I have an application that is producing dozens of thousands (usually hundreds) of IObservable and need to execute these observables in parallel. I however have to wait for all of them to complete before moving on.

What have you tried so far, what code have you written so far?

I have written a (possibly crude) lock-free implementation of ForkJoinObservable.

    sealed class ForkJoinObservable<TSource> : ObservableBase<TSource[]>
    {
        private readonly IObservable<TSource>[] _sources;

        public ForkJoinObservable(IObservable<TSource>[] sources)
        {
            _sources = sources;
        }

        protected override IDisposable SubscribeCore(IObserver<TSource[]> observer)
        {
            var count = _sources.Length;
            if (count == 0)
            {
                observer.OnCompleted();
                return Disposable.Empty;
            }

            var group = new CompositeDisposable(count);

            var finished = false;
            var hasResults = new bool[count];
            var atomicRequiredCompletedCount = count;
            var results = new TSource[count];

            Parallel.For(0, count, (index, state) =>
            {
                var currentIndex = index;
                var source = _sources[index];

                group.Add(source.Subscribe(
                    value => {
                        if (Volatile.Read(ref finished))
                            return;

                        hasResults[currentIndex] = true;
                        results[currentIndex] = value;
                    },
                    error => {
                        Volatile.Write(ref finished, true);
                        observer.OnError(error);
                        group.Dispose();
                    },
                    () => {
                        if (Volatile.Read(ref finished))
                            return;

                        if (!hasResults[currentIndex])
                        {
                            observer.OnCompleted();
                            return;
                        }

                        if (Interlocked.Decrement(ref atomicRequiredCompletedCount) > 0)
                            return;

                        Volatile.Write(ref finished, true);
                        observer.OnNext(results);
                        observer.OnCompleted();
                    }));
            });
            return group;
        }
    }

I have also benchmarked it:

BenchmarkDotNet=v0.12.1, OS=Windows 10.0.19042
Intel Core i7-7700HQ CPU 2.80GHz (Kaby Lake), 1 CPU, 8 logical and 4 physical cores
.NET Core SDK=5.0.201
  [Host]     : .NET Core 5.0.4 (CoreCLR 5.0.421.11614, CoreFX 5.0.421.11614), X64 RyuJIT
  DefaultJob : .NET Core 5.0.4 (CoreCLR 5.0.421.11614, CoreFX 5.0.421.11614), X64 RyuJIT


|              Method | ObservableCount |            Mean |         Error |        StdDev |          Median |
|-------------------- |---------------- |----------------:|--------------:|--------------:|----------------:|
| 'Rx.NET ForkJoin()' |              10 |        15.70 us |      0.300 us |      0.321 us |        15.71 us |
| 'Custom ForkJoin()' |              10 |        18.65 us |      0.361 us |      0.494 us |        18.68 us |
| 'Rx.NET ForkJoin()' |             100 |       157.87 us |      3.004 us |      3.214 us |       157.94 us |
| 'Custom ForkJoin()' |             100 |       132.38 us |      1.227 us |      1.148 us |       132.64 us |
| 'Rx.NET ForkJoin()' |            1000 |     1,625.10 us |     31.486 us |     72.344 us |     1,604.59 us |
| 'Custom ForkJoin()' |            1000 |     1,262.97 us |     23.834 us |     25.502 us |     1,262.70 us |
| 'Rx.NET ForkJoin()' |           10000 |    53,397.78 us |  1,164.106 us |  3,414.120 us |    53,030.44 us |
| 'Custom ForkJoin()' |           10000 |    17,382.86 us |    604.506 us |  1,734.441 us |    16,901.41 us |
| 'Rx.NET ForkJoin()' |          100000 | 3,517,157.55 us | 62,678.544 us | 52,339.423 us | 3,531,034.90 us |
| 'Custom ForkJoin()' |          100000 |   214,621.17 us |  3,792.702 us |  3,547.695 us |   215,474.57 us |

    public class ForkJoinBenchmark
    {
        [Params(1, 10, 100, 1000, 10000)]
        public int ObservableCount { get; set; }
        
        [Benchmark(Description = "Rx.NET ForkJoin()")]
        public async Task<long> RxForkJoin()
        {
            var value = await Enumerable.Range(1, ObservableCount)
                .Select(v => Observable.Start(() => (long) v, TaskPoolScheduler.Default))
                .ForkJoin()
                .Select(values => values.Sum());

            Debug.Assert(value == ObservableCount * (ObservableCount + 1) / 2);
            return value;
        }

        [Benchmark(Description = "Custom ForkJoin()")]
        public async Task<long> CustomForkJoin()
        {
            var value = await new ForkJoinObservable<long>(
                Enumerable.Range(1, ObservableCount)
                    .Select(v => Observable.Start(() => (long) v, TaskPoolScheduler.Default)).ToArray()
                ).Select(values => values.Sum());

            Debug.Assert(value == ObservableCount * (ObservableCount + 1) / 2);

            return value;
        }
    }

Debug assertions are included to make sure I don't run into parallelism issues and miscount. This is in line with what I noticed in a real-world test, where I have ~725.000 observables that need to be waited for, where it actually takes a few minutes for ForkJoin() to process.

An argument could be made that observables are not the best solution for this specific problem .... but it is quite convenient.

[Warpten]

Some more metrics from BDN diagnosers:

BenchmarkDotNet=v0.12.1, OS=Windows 10.0.19042
Intel Core i7-7700HQ CPU 2.80GHz (Kaby Lake), 1 CPU, 8 logical and 4 physical cores
.NET Core SDK=5.0.201
  [Host]   : .NET Core 5.0.4 (CoreCLR 5.0.421.11614, CoreFX 5.0.421.11614), X64 RyuJIT
  ShortRun : .NET Core 5.0.4 (CoreCLR 5.0.421.11614, CoreFX 5.0.421.11614), X64 RyuJIT

Job=ShortRun  IterationCount=3  LaunchCount=1
WarmupCount=3

|              Method | ObservableCount |            Mean |          Error |       StdDev | Completed Work Items | Lock Contentions |      Gen 0 |     Gen 1 |     Gen 2 |   Allocated |
|-------------------- |---------------- |----------------:|---------------:|-------------:|---------------------:|-----------------:|-----------:|----------:|----------:|------------:|
| 'Rx.NET ForkJoin()' |              10 |        12.69 us |       1.573 us |     0.086 us |              10.0000 |                - |     3.0670 |         - |         - |     9.42 KB |
| 'Custom ForkJoin()' |              10 |        15.56 us |       1.378 us |     0.076 us |              14.4305 |           0.0002 |     3.9063 |         - |         - |    11.88 KB |
| 'Rx.NET ForkJoin()' |             100 |       121.61 us |      20.867 us |     1.144 us |             100.0002 |           0.0001 |    28.3203 |         - |         - |    86.94 KB |
| 'Custom ForkJoin()' |             100 |       116.44 us |       9.139 us |     0.501 us |             107.7897 |           0.0925 |    29.4189 |         - |         - |    89.21 KB |
| 'Rx.NET ForkJoin()' |            1000 |     1,447.45 us |     137.254 us |     7.523 us |            1000.0039 |                - |   281.2500 |    3.9063 |         - |   862.13 KB |
| 'Custom ForkJoin()' |            1000 |     1,165.55 us |     198.803 us |    10.897 us |            1010.8203 |           3.1367 |   285.1563 |         - |         - |   857.01 KB |
| 'Rx.NET ForkJoin()' |           10000 |    44,292.18 us |   4,466.019 us |   244.798 us |           10000.1667 |                - |  1500.0000 |  583.3333 |   83.3333 |  8614.08 KB |
| 'Custom ForkJoin()' |           10000 |    14,057.18 us |   5,357.905 us |   293.685 us |           10011.1250 |          15.9375 |  1453.1250 |  625.0000 |  218.7500 |  8529.98 KB |
| 'Rx.NET ForkJoin()' |          100000 | 3,197,271.50 us | 174,432.102 us | 9,561.212 us |          100002.0000 |                - | 15000.0000 | 4000.0000 | 2000.0000 | 86138.16 KB |
| 'Custom ForkJoin()' |          100000 |   180,988.18 us |  25,987.170 us | 1,424.444 us |          100009.6667 |         174.3333 | 14666.6667 | 4333.3333 | 1666.6667 | 85270.72 KB |

[akarnokd]

Yes, this operator can be made lock-free. For correctness though, you'll have to do a bit more with those atomics:

error handler

If multiple sources fail/complete empty, this could invoke OnError concurrently.

The trick is to use an atomic exchange so that an N->0 successful transition will be only on a single thread:

if (Interlocked.Exchange(ref atomicRequiredCompletedCount, 0) > 0) {
   observer.OnError(ex);
   group.Dispose();
}

complete handler

There is a similar problem when two or more sources complete without items and both would invoke OnCompleted.

Again, use the N -> 0 transition to make this happen at most once.

if (!hasResults[currentIndex])
{
    if (Interlocked.Exchange(ref atomicRequiredCompletedCount, 0) > 0) {
        observer.OnCompleted();
    }
    return;
}

Parallel.For

Could you measure without using Parallel.For ? This introduces unexpected threading into the subscription process.

In addition CompositeDisposable still uses a lock so group.Add could still have a lot of contention due to Parallel.For, possibly limiting the benefits.

[Warpten]

Good point about using atomic exchanges, I have to admit I rarely consider error cases when dealing with observables (still fairly new to this stuff!)

This is with your suggestions, and lock contention eliminated from CompositeDisposable.Add. I also used ParallelLoopState instead of a local variable to track finished state. This got rid of lock contentation but still introduces a bit of overhead in the number of work units executed (and is more allocatey).

|              Method | ObservableCount |         Mean |         Error |     StdDev | Completed Work Items | Lock Contentions |     Gen 0 |    Gen 1 |    Gen 2 |  Allocated |
|-------------------- |---------------- |-------------:|--------------:|-----------:|---------------------:|-----------------:|----------:|---------:|---------:|-----------:|
| 'Rx.NET ForkJoin()' |              10 |     12.86 us |      0.714 us |   0.039 us |              10.0000 |           0.0000 |    3.0823 |        - |        - |    9.42 KB |
| 'Custom ForkJoin()' |              10 |     16.07 us |      2.968 us |   0.163 us |              14.1982 |           0.0000 |    4.1504 |        - |        - |    12.6 KB |
| 'Rx.NET ForkJoin()' |             100 |    123.27 us |     11.209 us |   0.614 us |             100.0002 |           0.0001 |   28.3203 |        - |        - |   86.94 KB |
| 'Custom ForkJoin()' |             100 |    112.39 us |     16.558 us |   0.908 us |             107.5356 |           0.0001 |   31.9824 |        - |        - |   96.96 KB |
| 'Rx.NET ForkJoin()' |            1000 |  1,457.70 us |    170.058 us |   9.321 us |            1000.0039 |                - |  281.2500 |   3.9063 |        - |  862.13 KB |
| 'Custom ForkJoin()' |            1000 |  1,094.01 us |    382.156 us |  20.947 us |            1008.0410 |           0.0059 |  308.5938 |   1.9531 |        - |   934.6 KB |
| 'Rx.NET ForkJoin()' |           10000 | 48,011.63 us | 12,060.692 us | 661.087 us |           10000.1818 |                - | 1545.4545 | 545.4545 |        - |    8615 KB |
| 'Custom ForkJoin()' |           10000 | 14,123.81 us |  9,962.026 us | 546.052 us |           10008.8906 |                - | 1609.3750 | 640.6250 | 187.5000 | 9310.78 KB |

Implementation

   sealed class ForkJoinObservable<TSource> : ObservableBase<TSource[]>
    {
        private readonly IObservable<TSource>[] _sources;

        public ForkJoinObservable(IObservable<TSource>[] sources)
        {
            _sources = sources;
        }

        protected override IDisposable SubscribeCore(IObserver<TSource[]> observer)
        {
            var count = _sources.Length;
            if (count == 0)
            {
                observer.OnCompleted();
                return Disposable.Empty;
            }

            var group = new IDisposable[count];

            var hasResults = new bool[count];
            var atomicRequiredCompletedCount = count;
            var results = new TSource[count];

            Parallel.For(0, count, (index, state) =>
            {
                var currentIndex = index;
                var source = _sources[index];

                var subscription = source.Subscribe(
                    value => {
                        if (state.IsStopped)
                            return;

                        hasResults[currentIndex] = true;
                        results[currentIndex] = value;
                    },
                    error => {
                        state.Stop();

                        if (Interlocked.Exchange(ref atomicRequiredCompletedCount, 0) > 0)
                            observer.OnError(error);
                    },
                    () => {
                        if (state.IsStopped)
                            return;

                        if (!hasResults[currentIndex])
                        {
                            if (Interlocked.Exchange(ref atomicRequiredCompletedCount, 0) > 0)
                                observer.OnCompleted();

                            return;
                        }

                        if (Interlocked.Decrement(ref atomicRequiredCompletedCount) > 0)
                            return;

                        state.Stop();

                        observer.OnNext(results);
                        observer.OnCompleted();
                    });

                group[index] = subscription;
            });

            return new CompositeDisposable(group);
        }
    }

---

And this is when transitioning back to regular old loops instead of Parallel.For:

|              Method | ObservableCount |         Mean |         Error |     StdDev | Completed Work Items | Lock Contentions |     Gen 0 |    Gen 1 |    Gen 2 |  Allocated |
|-------------------- |---------------- |-------------:|--------------:|-----------:|---------------------:|-----------------:|----------:|---------:|---------:|-----------:|
| 'Rx.NET ForkJoin()' |              10 |     12.77 us |      2.715 us |   0.149 us |              10.0000 |                - |    3.0823 |        - |        - |    9.42 KB |
| 'Custom ForkJoin()' |              10 |     12.07 us |      4.078 us |   0.224 us |              10.0000 |           0.0000 |    3.0518 |        - |        - |    9.34 KB |
| 'Rx.NET ForkJoin()' |             100 |    123.58 us |     32.270 us |   1.769 us |             100.0005 |           0.0002 |   28.3203 |        - |        - |   86.94 KB |
| 'Custom ForkJoin()' |             100 |    113.28 us |     10.795 us |   0.592 us |             100.0004 |                - |   28.3203 |        - |        - |   86.77 KB |
| 'Rx.NET ForkJoin()' |            1000 |  1,475.52 us |    318.147 us |  17.439 us |            1000.0039 |                - |  281.2500 |   3.9063 |        - |  862.13 KB |
| 'Custom ForkJoin()' |            1000 |  1,112.08 us |    195.209 us |  10.700 us |            1000.0039 |                - |  279.2969 |        - |        - |  861.08 KB |
| 'Rx.NET ForkJoin()' |           10000 | 47,835.33 us | 16,927.498 us | 927.853 us |           10000.1818 |                - | 1545.4545 | 545.4545 |        - |    8615 KB |
| 'Custom ForkJoin()' |           10000 | 14,897.03 us |  2,685.349 us | 147.193 us |           10000.0313 |                - | 1468.7500 | 531.2500 | 156.2500 | 8604.24 KB |

Implementation

    sealed class ForkJoinObservable<TSource> : ObservableBase<TSource[]>
    {
        private readonly IObservable<TSource>[] _sources;

        public ForkJoinObservable(IObservable<TSource>[] sources)
        {
            _sources = sources;
        }

        protected override IDisposable SubscribeCore(IObserver<TSource[]> observer)
        {
            var count = _sources.Length;
            if (count == 0)
            {
                observer.OnCompleted();
                return Disposable.Empty;
            }

            var group = new IDisposable[count];

            var hasResults = new bool[count];
            var atomicRequiredCompletedCount = count;
            var results = new TSource[count];

            var finished = false;
            
            for (var index = 0; index < count; ++index)
            {
                var currentIndex = index;
                var source = _sources[index];

                var subscription = source.Subscribe(
                    value => {
                        if (Volatile.Read(ref finished))
                            return;

                        hasResults[currentIndex] = true;
                        results[currentIndex] = value;
                    },
                    error =>
                    {
                        Volatile.Write(ref finished, true);

                        if (Interlocked.Exchange(ref atomicRequiredCompletedCount, 0) > 0)
                            observer.OnError(error);
                    },
                    () => {
                        if (Volatile.Read(ref finished))
                            return;

                        if (!hasResults[currentIndex])
                        {
                            if (Interlocked.Exchange(ref atomicRequiredCompletedCount, 0) > 0)
                                observer.OnCompleted();

                            return;
                        }

                        if (Interlocked.Decrement(ref atomicRequiredCompletedCount) > 0)
                            return;

                        Volatile.Write(ref finished, true);

                        observer.OnNext(results);
                        observer.OnCompleted();
                    });

                group[index] = subscription;
            }

            return new CompositeDisposable(group);
        }
    }

Less allocations (no more parallel loop state tracking). The timings are not perfect (those are short runs - IterationCount=3 LaunchCount=1 WarmupCount=3) but they should be indicative enough.

[akarnokd]

My concern with Parallel.For is still the trouble it runs the Subscribes not on the current thread whereas the existing implementation does it implicitly. I'd defer to @bartdesmet to indicate if this type of optimization is allowed or not.

With the IDisposable array version, the current implementation calls group.Dispose() in OnError to eagerly stop the other sources. Removing this behavior could be considered an unexpected behavior change.

[Warpten]

Right, I missed that part. In that case, won't manually calling Dispose on each IDisposable whenever an error is received work? I can try to benchmark it as well but that would be a little tricky since I guess the benchmark would depend on when exactly one observable fails. I'm down to get rid of the parallel loops (after all, they are not that much beneficial after all).

If I revert to using an instance of CompositeDisposable constructed with a capacity, I get a bit of lock contention according to BDN. It's not immediately obvious to me as to why, since all Add calls would occur on the same thread.

[akarnokd]

In that case, won't manually calling Dispose on each IDisposable whenever an error is received work?

No. The failure could come at any moment, even while the array is being populated by IDisposables. Some entries would be null, others would not be properly visible to other threads due to lack of memory barriers. CompositeDisposable.Add did handle this case.

The workaround I can think of is using an array of SingleAssignmentDisposables populated before the subscriptions so assignment/dispose would be thread safe again.

[Warpten]

In any case, simply replacing _gate by atomic operations is already improving the operator quite a bit. I'm fine with a bit of lock contention; it should, realisitically, only happen for the first few hundred - or thousands (educated guess) - iterations, where individual observables are firing but ForkJoin is still subscribing to subsequent observables.