mcs/class/System.Threading.Tasks.Dataflow/Test/System.Threading.Tasks.Dataflow/ExecutionBlocksTest.cs

   1 // ExecutionBlocksTest.cs
   2 //
   3 // Copyright (c) 2012 Petr Onderka
   4 //
   5 // Permission is hereby granted, free of charge, to any person obtaining a copy
   6 // of this software and associated documentation files (the "Software"), to deal
   7 // in the Software without restriction, including without limitation the rights
   8 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   9 // copies of the Software, and to permit persons to whom the Software is
  10 // furnished to do so, subject to the following conditions:
  11 //
  12 // The above copyright notice and this permission notice shall be included in
  13 // all copies or substantial portions of the Software.
  14 //
  15 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  20 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  21 // THE SOFTWARE.
  22
  23 using System;
  24 using System.Collections.Generic;
  25 using System.Linq;
  26 using System.Threading;
  27 using System.Threading.Tasks;
  28 using System.Threading.Tasks.Dataflow;
  29 using NUnit.Framework;
  30
  31 namespace MonoTests.System.Threading.Tasks.Dataflow {
  32         [TestFixture]
  33         public class ExecutionBlocksTest {
  34                 static IEnumerable<ITargetBlock<int>> GetExecutionBlocksWithAction(Action action)
  35                 {
  36                         yield return new ActionBlock<int> (i => action());
  37                         yield return new TransformBlock<int, int> (i =>
  38                         {
  39                                 action ();
  40                                 return i;
  41                         });
  42                         yield return new TransformManyBlock<int, int> (i =>
  43                         {
  44                                 action ();
  45                                 return Enumerable.Empty<int> ();
  46                         });
  47                 }
  48
  49                 static IEnumerable<ITargetBlock<int>> GetExecutionBlocksWithAsyncAction (
  50                         Func<int, Task> action, ExecutionDataflowBlockOptions options)
  51                 {
  52                         yield return new ActionBlock<int> (action, options);
  53                         yield return new TransformBlock<int, int> (
  54                                 i => action (i).ContinueWith (
  55                                         t =>
  56                                         {
  57                                                 t.Wait ();
  58                                                 return i;
  59                                         }), options);
  60                         yield return new TransformManyBlock<int, int> (
  61                                 i => action (i).ContinueWith (
  62                                         t =>
  63                                         {
  64                                                 t.Wait ();
  65                                                 return Enumerable.Empty<int> ();
  66                                         }), options);
  67                 }
  68
  69                 [Test]
  70                 public void ExceptionTest ()
  71                 {
  72                         var exception = new Exception ();
  73                         var blocks = GetExecutionBlocksWithAction (() => { throw exception; });
  74                         foreach (var block in blocks) {
  75                                 Assert.IsFalse (block.Completion.Wait (100));
  76
  77                                 block.Post (1);
  78
  79                                 var ae =
  80                                         AssertEx.Throws<AggregateException> (() => block.Completion.Wait (100));
  81                                 Assert.AreEqual (1, ae.InnerExceptions.Count);
  82                                 Assert.AreSame (exception, ae.InnerException);
  83                         }
  84                 }
  85
  86                 [Test]
  87                 public void NoProcessingAfterFaultTest ()
  88                 {
  89                         int shouldRun = 1;
  90                         int ranAfterFault = 0;
  91                         var evt = new ManualResetEventSlim ();
  92
  93                         var blocks = GetExecutionBlocksWithAction (() =>
  94                         {
  95                                 if (Thread.VolatileRead (ref shouldRun) == 0) {
  96                                         ranAfterFault++;
  97                                         return;
  98                                 }
  99
 100                                 evt.Wait ();
 101                         });
 102
 103                         foreach (var block in blocks) {
 104                                 shouldRun = 1;
 105                                 ranAfterFault = 0;
 106                                 evt.Reset ();
 107
 108                                 Assert.IsTrue (block.Post (1));
 109                                 Assert.IsTrue (block.Post (2));
 110
 111                                 Assert.IsFalse (block.Completion.Wait (100));
 112                                 Assert.AreEqual (0, ranAfterFault);
 113
 114                                 block.Fault (new Exception ());
 115
 116                                 Assert.IsFalse (block.Completion.Wait (100));
 117
 118                                 shouldRun = 0;
 119                                 evt.Set ();
 120
 121                                 AssertEx.Throws<AggregateException> (() => block.Completion.Wait (100));
 122
 123                                 Thread.Sleep (100);
 124
 125                                 Assert.AreEqual (0, Thread.VolatileRead (ref ranAfterFault));
 126                         }
 127                 }
 128
 129                 [Test]
 130                 public void AsyncTest ()
 131                 {
 132                         var tcs = new TaskCompletionSource<int> ();
 133                         int result = 0;
 134
 135                         var scheduler = new TestScheduler ();
 136
 137                         var blocks = GetExecutionBlocksWithAsyncAction (
 138                                 i =>
 139                                 tcs.Task.ContinueWith (t => Thread.VolatileWrite (ref result, i + t.Result)),
 140                                 new ExecutionDataflowBlockOptions { TaskScheduler = scheduler });
 141
 142                         foreach (var block in blocks) {
 143                                 Assert.IsTrue (block.Post (1));
 144
 145                                 scheduler.ExecuteAll ();
 146                                 Thread.Sleep (100);
 147                                 Thread.MemoryBarrier ();
 148
 149                                 Assert.AreEqual (0, result);
 150
 151                                 tcs.SetResult (10);
 152
 153                                 Thread.Sleep (100);
 154
 155                                 // the continuation should be executed on the configured TaskScheduler
 156                                 Assert.AreEqual (0, result);
 157
 158                                 scheduler.ExecuteAll ();
 159
 160                                 Assert.AreEqual (11, result);
 161
 162                                 tcs = new TaskCompletionSource<int> ();
 163                                 Thread.VolatileWrite (ref result, 0);
 164                         }
 165                 }
 166
 167                 [Test]
 168                 public void AsyncExceptionTest ()
 169                 {
 170                         var scheduler = new TestScheduler ();
 171                         var exception = new Exception ();
 172
 173                         var blocks = GetExecutionBlocksWithAsyncAction (
 174                                 i =>
 175                                 {
 176                                         var tcs = new TaskCompletionSource<int> ();
 177                                         tcs.SetException (exception);
 178                                         return tcs.Task;
 179                                 },
 180                                 new ExecutionDataflowBlockOptions { TaskScheduler = scheduler });
 181
 182                         foreach (var block in blocks) {
 183                                 Assert.IsTrue (block.Post (1));
 184
 185                                 // the task should be executed on the configured TaskScheduler
 186                                 Assert.IsFalse (block.Completion.Wait (100));
 187
 188                                 scheduler.ExecuteAll ();
 189
 190                                 var ae =
 191                                         AssertEx.Throws<AggregateException> (() => block.Completion.Wait (100)).
 192                                                 Flatten ();
 193
 194                                 Assert.AreEqual (1, ae.InnerExceptions.Count);
 195                                 Assert.AreSame (exception, ae.InnerException);
 196                         }
 197                 }
 198         }
 199 }