Merge pull request #2305 from Clockwork-Muse/PLINQ_2239

stephentoub · stephentoub · commit d9ad8dc965d5 · 2015-07-20T15:17:36.000-04:00
PLINQ - Minor fixes
diff --git a/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableDecimalAverageAggregationOperator.cs b/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableDecimalAverageAggregationOperator.cs
@@ -118,8 +118,11 @@ protected override bool MoveNextCore(ref Pair<decimal, long> currentElement)
 
                     if (current.HasValue)
                     {
-                        sum += current.GetValueOrDefault();
-                        count++;
+                        checked
+                        {
+                            sum += current.GetValueOrDefault();
+                            count++;
+                        }
                     }
                 }
 
diff --git a/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableDoubleAverageAggregationOperator.cs b/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableDoubleAverageAggregationOperator.cs
@@ -118,8 +118,11 @@ protected override bool MoveNextCore(ref Pair<double, long> currentElement)
                         if ((i++ & CancellationState.POLL_INTERVAL) == 0)
                             CancellationState.ThrowIfCanceled(_cancellationToken);
 
-                        sum += current.GetValueOrDefault();
-                        count++;
+                        checked
+                        {
+                            sum += current.GetValueOrDefault();
+                            count++;
+                        }
                     }
                 }
 
diff --git a/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableFloatAverageAggregationOperator.cs b/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableFloatAverageAggregationOperator.cs
@@ -119,8 +119,11 @@ protected override bool MoveNextCore(ref Pair<double, long> currentElement)
 
                     if (current.HasValue)
                     {
-                        sum += current.GetValueOrDefault();
-                        count++;
+                        checked
+                        {
+                            sum += current.GetValueOrDefault();
+                            count++;
+                        }
                     }
                 }
 
diff --git a/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableIntAverageAggregationOperator.cs b/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableIntAverageAggregationOperator.cs
@@ -119,8 +119,11 @@ protected override bool MoveNextCore(ref Pair<long, long> currentElement)
 
                     if (current.HasValue)
                     {
-                        sum += current.GetValueOrDefault();
-                        count++;
+                        checked
+                        {
+                            sum += current.GetValueOrDefault();
+                            count++;
+                        }
                     }
                 }
 
diff --git a/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableLongAverageAggregationOperator.cs b/src/System.Linq.Parallel/src/System/Linq/Parallel/QueryOperators/Inlined/NullableLongAverageAggregationOperator.cs
@@ -119,8 +119,11 @@ protected override bool MoveNextCore(ref Pair<long, long> currentElement)
                         CancellationState.ThrowIfCanceled(_cancellationToken);
                     if (current.HasValue)
                     {
-                        sum += current.GetValueOrDefault();
-                        count++;
+                        checked
+                        {
+                            sum += current.GetValueOrDefault();
+                            count++;
+                        }
                     }
                 }
 
diff --git a/src/System.Linq.Parallel/src/System/Linq/Parallel/Utils/ReverseComparer.cs b/src/System.Linq.Parallel/src/System/Linq/Parallel/Utils/ReverseComparer.cs
@@ -27,7 +27,7 @@ internal ReverseComparer(IComparer<T> comparer)
 
         public int Compare(T x, T y)
         {
-            return -_comparer.Compare(x, y);
+            return _comparer.Compare(y, x);
         }
     }
 }
diff --git a/src/System.Linq.Parallel/src/System/Linq/ParallelEnumerable.cs b/src/System.Linq.Parallel/src/System/Linq/ParallelEnumerable.cs
@@ -352,29 +352,10 @@ public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource>(this Paral
         /// <exception cref="T:System.InvalidOperationException">
         /// WithCancellation is used multiple times in the query.
         /// </exception>
-        /// <exception cref="T:System.ObjectDisposedException">
-        /// The <see cref="T:System.Threading.CancellationTokenSource"/> associated with the <paramref name="cancellationToken"/> has been disposed.
-        /// </exception>
         public static ParallelQuery<TSource> WithCancellation<TSource>(this ParallelQuery<TSource> source, CancellationToken cancellationToken)
         {
             if (source == null) throw new ArgumentNullException("source");
 
-            // also a convenience check whether the cancellationTokenSource backing the token is already disposed.
-            // do this via a dummy registration as there is no public IsDipsosed property on CT.
-            CancellationTokenRegistration dummyRegistration = new CancellationTokenRegistration();
-            try
-            {
-                dummyRegistration = cancellationToken.Register(() => { });
-            }
-            catch (ObjectDisposedException)
-            {
-                throw new ArgumentException(SR.ParallelEnumerable_WithCancellation_TokenSourceDisposed, "cancellationToken");
-            }
-            finally
-            {
-                dummyRegistration.Dispose();
-            }
-
             QuerySettings settings = QuerySettings.Empty;
             settings.CancellationState = new CancellationState(cancellationToken);
 
diff --git a/src/System.Linq.Parallel/tests/ExchangeTests.cs b/src/System.Linq.Parallel/tests/ExchangeTests.cs
@@ -5,6 +5,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using Xunit;
 
 namespace Test
@@ -131,22 +132,26 @@ public static void Merge_Ordered_Longrunning(Labeled<ParallelQuery<int>> labeled
         }
 
         [Theory]
-        [MemberData("MergeData", (object)(new int[] { 16, 1024 }))]
-        [MemberData("ThrowOnCount_AllMergeOptions_MemberData", (object)(new int[] { 16, 1024 }))]
+        [MemberData("ThrowOnCount_AllMergeOptions_MemberData", (object)(new int[] { 4, 8 }))]
         // FailingMergeData has enumerables that throw errors when attempting to perform the nth enumeration.
         // This test checks whether the query runs in a pipelined or buffered fashion.
         public static void Merge_Ordered_Pipelining(Labeled<ParallelQuery<int>> labeled, int count, ParallelMergeOptions options)
         {
-            Assert.Equal(0, labeled.Item.WithMergeOptions(options).Select(x => x).First());
+            Assert.Equal(0, labeled.Item.WithDegreeOfParallelism(count - 1).WithMergeOptions(options).First());
         }
 
         [Theory]
-        [OuterLoop]
-        [MemberData("MergeData", (object)(new int[] { 1024 * 4, 1024 * 1024 }))]
-        [MemberData("ThrowOnCount_AllMergeOptions_MemberData", (object)(new int[] { 1024 * 4, 1024 * 1024 }))]
-        public static void Merge_Ordered_Pipelining_Longrunning(Labeled<ParallelQuery<int>> labeled, int count, ParallelMergeOptions options)
+        [MemberData("MergeData", (object)(new int[] { 4, 8 }))]
+        // This test checks whether the query runs in a pipelined or buffered fashion.
+        public static void Merge_Ordered_Pipelining_Select(Labeled<ParallelQuery<int>> labeled, int count, ParallelMergeOptions options)
         {
-            Merge_Ordered_Pipelining(labeled, count, options);
+            int countdown = count;
+            Func<int, int> down = i =>
+            {
+                if (Interlocked.Decrement(ref countdown) == 0) throw new DeliberateTestException();
+                return i;
+            };
+            Assert.Equal(0, labeled.Item.WithDegreeOfParallelism(count - 1).WithMergeOptions(options).Select(down).First());
         }
 
         [Theory]
diff --git a/src/System.Linq.Parallel/tests/Helpers/Comparers.cs b/src/System.Linq.Parallel/tests/Helpers/Comparers.cs
@@ -72,6 +72,24 @@ public int Compare(int x, int y)
         }
     }
 
+    /// <summary>
+    /// Returns an extreme value from non-equal comparisons.
+    /// </summary>
+    /// <remarks>Helper for regression test against PLINQ's version of #2239 .</remarks>
+    /// <typeparam name="T">The type being compared.</typeparam>
+    internal class ExtremeComparer<T> : IComparer<T>
+    {
+        private IComparer<T> _def = Comparer<T>.Default;
+
+        public int Compare(T x, T y)
+        {
+            int direction = _def.Compare(x, y);
+            return direction == 0 ? 0 :
+                direction > 0 ? int.MaxValue :
+                int.MinValue;
+        }
+    }
+
     internal static class DelgatedComparable
     {
         public static DelegatedComparable<T> Delegate<T>(T value, IComparer<T> comparer) where T : IComparable<T>
diff --git a/src/System.Linq.Parallel/tests/QueryOperators/AverageTests.cs b/src/System.Linq.Parallel/tests/QueryOperators/AverageTests.cs
@@ -14,6 +14,8 @@ public class AverageTests
         //
         // Average
         //
+
+        // Get a set of ranges from 0 to each count, with an extra parameter containing the expected average.
         public static IEnumerable<object[]> AverageData(object[] counts)
         {
             Func<int, double> average = x => (x - 1) / 2.0;
@@ -76,6 +78,16 @@ public static void Average_Long_Longrunning(Labeled<ParallelQuery<int>> labeled,
             Average_Long(labeled, count, average);
         }
 
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 2 }), MemberType = typeof(UnorderedSources))]
+        public static void Average_Long_Overflow(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            Functions.AssertThrowsWrapped<OverflowException>(() => labeled.Item.Select(x => x == 0 ? 1 : long.MaxValue).Average());
+            Functions.AssertThrowsWrapped<OverflowException>(() => labeled.Item.Select(x => x == 0 ? (long?)1 : long.MaxValue).Average());
+            Functions.AssertThrowsWrapped<OverflowException>(() => labeled.Item.Average(x => x == 0 ? -1 : long.MinValue));
+            Functions.AssertThrowsWrapped<OverflowException>(() => labeled.Item.Average(x => x == 0 ? (long?)-1 : long.MinValue));
+        }
+
         [Theory]
         [MemberData("AverageData", (object)(new int[] { 1, 2, 16 }))]
         public static void Average_Long_SomeNull(Labeled<ParallelQuery<int>> labeled, int count, double average)
diff --git a/src/System.Linq.Parallel/tests/QueryOperators/OrderByThenByTests.cs b/src/System.Linq.Parallel/tests/QueryOperators/OrderByThenByTests.cs
@@ -326,6 +326,55 @@ public static void OrderByDescending_NotPipelined_CustomComparator_Longrunning(L
             OrderByDescending_NotPipelined_CustomComparer(labeled, count);
         }
 
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("OrderByRandomData", (object)(new[] { 0, 1, 2, 16 }))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        // Regression test for the PLINQ version of #2239 - comparer returning max/min value.
+        public static void OrderBy_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MinValue;
+            foreach (int i in labeled.Item.OrderBy(x => x, new ExtremeComparer<int>()))
+            {
+                Assert.InRange(i, prev, int.MaxValue);
+                prev = i;
+            }
+        }
+
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("OrderByRandomData", (object)(new[] { 0, 1, 2, 16 }))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        // Regression test for the PLINQ version of #2239 - comparer returning max/min value.
+        public static void OrderByDescending_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MaxValue;
+            foreach (int i in labeled.Item.OrderByDescending(x => x, new ExtremeComparer<int>()))
+            {
+                Assert.InRange(i, int.MinValue, prev);
+                prev = i;
+            }
+        }
+
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        public static void OrderBy_NotPipelined_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MinValue;
+            Assert.All(labeled.Item.OrderBy(x => x, new ExtremeComparer<int>()).ToList(), x => { Assert.InRange(x, prev, int.MaxValue); prev = x; });
+        }
+
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("OrderByRandomData", (object)(new[] { 0, 1, 2, 16 }))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        public static void OrderByDescending_NotPipelined_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MaxValue;
+            Assert.All(labeled.Item.OrderByDescending(x => x, new ExtremeComparer<int>()).ToList(), x => { Assert.InRange(x, int.MinValue, prev); prev = x; });
+        }
+
         [Fact]
         public static void OrderBy_ArgumentNullException()
         {
@@ -753,6 +802,55 @@ public static void ThenByDescending_NotPipelined_CustomComparator_Longrunning(La
             ThenByDescending_NotPipelined_CustomComparer(labeled, count);
         }
 
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("OrderByRandomData", (object)(new[] { 0, 1, 2, 16 }))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        // Regression test for the PLINQ version of #2239 - comparer returning max/min value.
+        public static void ThenBy_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MinValue;
+            foreach (int i in labeled.Item.OrderBy(x => 0).ThenBy(x => x, new ExtremeComparer<int>()))
+            {
+                Assert.InRange(i, prev, int.MaxValue);
+                prev = i;
+            }
+        }
+
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("OrderByRandomData", (object)(new[] { 0, 1, 2, 16 }))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        // Regression test for the PLINQ version of #2239 - comparer returning max/min value.
+        public static void ThenByDescending_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MaxValue;
+            foreach (int i in labeled.Item.OrderBy(x => 0).ThenByDescending(x => x, new ExtremeComparer<int>()))
+            {
+                Assert.InRange(i, int.MinValue, prev);
+                prev = i;
+            }
+        }
+
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        public static void ThenBy_NotPipelined_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MinValue;
+            Assert.All(labeled.Item.OrderBy(x => 0).ThenBy(x => x, new ExtremeComparer<int>()).ToList(), x => { Assert.InRange(x, prev, int.MaxValue); prev = x; });
+        }
+
+        [Theory]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(Sources))]
+        [MemberData("OrderByRandomData", (object)(new[] { 0, 1, 2, 16 }))]
+        [MemberData("Ranges", (object)(new int[] { 0, 1, 2, 16 }), MemberType = typeof(UnorderedSources))]
+        public static void ThenByDescending_NotPipelined_ExtremeComparer(Labeled<ParallelQuery<int>> labeled, int count)
+        {
+            int prev = int.MaxValue;
+            Assert.All(labeled.Item.OrderBy(x => 0).ThenByDescending(x => x, new ExtremeComparer<int>()).ToList(), x => { Assert.InRange(x, int.MinValue, prev); prev = x; });
+        }
+
         // Recursive sort with nested ThenBy...s
         // Due to the use of randomized input, cycles will not start with a known input (and may skip values, etc).
 
diff --git a/src/System.Linq.Parallel/tests/WithCancellationTests.cs b/src/System.Linq.Parallel/tests/WithCancellationTests.cs
@@ -617,71 +617,5 @@ public static void DontDoWorkIfTokenAlreadyCanceled()
 
             Assert.NotNull(oce);
         }
-
-        // To help the user, we will check if a cancellation token passed to WithCancellation() is
-        // not backed by a disposed CTS.  This will help them identify incorrect cts.Dispose calls, but
-        // doesn't solve all their problems if they don't manage CTS lifetime correctly.
-        // We test via a few random queries that have shown inconsistent behavior in the past.
-        [Fact]
-        public static void PreDisposedCTSPassedToPlinq()
-        {
-            ArgumentException ae1 = null;
-            ArgumentException ae2 = null;
-            ArgumentException ae3 = null;
-
-            CancellationTokenSource cts = new CancellationTokenSource();
-            CancellationToken ct = cts.Token;
-            cts.Dispose();  // Early dispose
-            try
-            {
-                Enumerable.Range(1, 10).AsParallel()
-                    .WithCancellation(ct)
-                    .OrderBy(x => x)
-                    .ToArray();
-            }
-            catch (Exception ex)
-            {
-                // This is not going to be the case since we changed the behavior of WithCancellation
-                // to not throw when called on disposed cancellationtokens.
-                ae1 = (ArgumentException)ex;
-            }
-
-            try
-            {
-                Enumerable.Range(1, 10).AsParallel()
-                    .WithCancellation(ct)
-                    .Last();
-            }
-            catch (Exception ex)
-            {
-                // This is not going to be the case since we changed the behavior of WithCancellation
-                // to not throw when called on disposed cancellationtokens.
-                ae2 = (ArgumentException)ex;
-            }
-
-            try
-            {
-                Enumerable.Range(1, 10).AsParallel()
-                    .WithCancellation(ct)
-                    .OrderBy(x => x)
-                    .Last();
-            }
-            catch (Exception ex)
-            {
-                // This is not going to be the case since we changed the behavior of WithCancellation
-                // to not throw when called on disposed cancellationtokens.
-                ae3 = (ArgumentException)ex;
-            }
-
-            Assert.Null(ae1);
-            Assert.Null(ae2);
-            Assert.Null(ae3);
-        }
-
-        public static void SimulateThreadSleep(int milliseconds)
-        {
-            ManualResetEvent mre = new ManualResetEvent(false);
-            mre.WaitOne(milliseconds);
-        }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -118,8 +118,11 @@ protected override bool MoveNextCore(ref Pair<decimal, long> currentElement)`
`118`	`118`
`119`	`119`	`if (current.HasValue)`
`120`	`120`	`{`
`121`		`- sum += current.GetValueOrDefault();`
`122`		`- count++;`
	`121`	`+ checked`
	`122`	`+ {`
	`123`	`+ sum += current.GetValueOrDefault();`
	`124`	`+ count++;`
	`125`	`+ }`
`123`	`126`	`}`
`124`	`127`	`}`
`125`	`128`
Original file line number	Diff line number	Diff line change
`@@ -119,8 +119,11 @@ protected override bool MoveNextCore(ref Pair<double, long> currentElement)`
`119`	`119`
`120`	`120`	`if (current.HasValue)`
`121`	`121`	`{`
`122`		`- sum += current.GetValueOrDefault();`
`123`		`- count++;`
	`122`	`+ checked`
	`123`	`+ {`
	`124`	`+ sum += current.GetValueOrDefault();`
	`125`	`+ count++;`
	`126`	`+ }`
`124`	`127`	`}`
`125`	`128`	`}`
`126`	`129`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ internal ReverseComparer(IComparer<T> comparer)`
`27`	`27`
`28`	`28`	`public int Compare(T x, T y)`
`29`	`29`	`{`
`30`		`- return -_comparer.Compare(x, y);`
	`30`	`+ return _comparer.Compare(y, x);`
`31`	`31`	`}`
`32`	`32`	`}`
`33`	`33`	`}`