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

   1 //
   2 // BatchBlockTest.cs
   3 //
   4 // Author:
   5 //       Jérémie "garuma" Laval <jeremie.laval@gmail.com>
   6 //       Petr Onderka <gsvick@gmail.com>
   7 //
   8 // Copyright (c) 2011 Jérémie "garuma" Laval
   9 // Copyright (c) 2012 Petr Onderka
  10 //
  11 // Permission is hereby granted, free of charge, to any person obtaining a copy
  12 // of this software and associated documentation files (the "Software"), to deal
  13 // in the Software without restriction, including without limitation the rights
  14 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  15 // copies of the Software, and to permit persons to whom the Software is
  16 // furnished to do so, subject to the following conditions:
  17 //
  18 // The above copyright notice and this permission notice shall be included in
  19 // all copies or substantial portions of the Software.
  20 //
  21 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  22 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  23 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  24 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  25 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  26 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  27 // THE SOFTWARE.
  28
  29 using System;
  30 using System.Threading;
  31 using System.Threading.Tasks.Dataflow;
  32 using NUnit.Framework;
  33
  34 namespace MonoTests.System.Threading.Tasks.Dataflow {
  35         [TestFixture]
  36         public class BatchBlockTest {
  37                 [Test]
  38                 public void BasicUsageTest ()
  39                 {
  40                         int[] array = null;
  41                         var evt = new ManualResetEventSlim (false);
  42
  43                         var buffer = new BatchBlock<int> (10);
  44                         var block = new ActionBlock<int[]> (i =>
  45                         {
  46                                 array = i;
  47                                 evt.Set ();
  48                         });
  49                         buffer.LinkTo<int[]> (block);
  50
  51                         for (int i = 0; i < 9; i++)
  52                                 Assert.IsTrue (buffer.Post (i));
  53
  54                         Assert.IsFalse (evt.Wait (100));
  55
  56                         Assert.IsNull (array);
  57
  58                         Assert.IsTrue (buffer.Post (42));
  59                         Assert.IsTrue (evt.Wait (1000));
  60
  61                         Assert.IsNotNull (array);
  62                         CollectionAssert.AreEqual (new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8, 42 }, array);
  63                 }
  64
  65                 [Test]
  66                 public void TriggerBatchTest ()
  67                 {
  68                         int[] array = null;
  69                         var evt = new ManualResetEventSlim (false);
  70
  71                         var buffer = new BatchBlock<int> (10);
  72                         var block = new ActionBlock<int[]> (i =>
  73                         {
  74                                 array = i;
  75                                 evt.Set ();
  76                         });
  77                         buffer.LinkTo (block);
  78
  79                         for (int i = 0; i < 9; i++)
  80                                 Assert.IsTrue (buffer.Post (i));
  81
  82                         buffer.TriggerBatch ();
  83                         evt.Wait ();
  84
  85                         Assert.IsNotNull (array);
  86                         Assert.IsTrue (buffer.Post (42));
  87                         evt.Wait (1600);
  88
  89                         CollectionAssert.AreEquivalent (new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8 },
  90                                 array);
  91                 }
  92
  93                 [Test]
  94                 public void TriggerBatchLateBinding ()
  95                 {
  96                         int[] array = null;
  97                         var evt = new ManualResetEventSlim (false);
  98
  99                         var buffer = new BatchBlock<int> (10);
 100                         var block = new ActionBlock<int[]> (i =>
 101                         {
 102                                 array = i;
 103                                 evt.Set ();
 104                         });
 105
 106                         for (int i = 0; i < 9; i++)
 107                                 Assert.IsTrue (buffer.Post (i));
 108
 109                         buffer.TriggerBatch ();
 110                         buffer.LinkTo (block);
 111
 112                         evt.Wait ();
 113                         Assert.IsNotNull (array);
 114
 115                         CollectionAssert.AreEquivalent (new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8 },
 116                                 array);
 117                 }
 118
 119                 [Test]
 120                 public void LateTriggerBatchKeepCountTest ()
 121                 {
 122                         int[] array = null;
 123                         var evt = new ManualResetEventSlim (false);
 124
 125                         var buffer = new BatchBlock<int> (15);
 126                         var block = new ActionBlock<int[]> (i =>
 127                         {
 128                                 array = i;
 129                                 evt.Set ();
 130                         });
 131
 132                         for (int i = 0; i < 9; i++)
 133                                 Assert.IsTrue (buffer.Post (i));
 134                         buffer.TriggerBatch ();
 135                         Assert.IsTrue (buffer.Post (42));
 136                         buffer.LinkTo (block);
 137
 138                         evt.Wait ();
 139
 140                         Assert.IsNotNull (array);
 141                         CollectionAssert.AreEquivalent (new int[] { 0, 1, 2, 3, 4, 5, 6, 7, 8 },
 142                                 array);
 143                 }
 144
 145                 [Test]
 146                 public void TriggerBatchWhenEmpty ()
 147                 {
 148                         var scheduler = new TestScheduler ();
 149                         var block = new BatchBlock<int> (5,
 150                                 new GroupingDataflowBlockOptions { TaskScheduler = scheduler });
 151                         block.TriggerBatch ();
 152
 153                         scheduler.ExecuteAll ();
 154
 155                         int[] batch;
 156                         Assert.IsFalse (block.TryReceive (out batch));
 157                         Assert.IsNull (batch);
 158                 }
 159
 160                 [Test]
 161                 public void NonGreedyBatchWithBoundedCapacityTest ()
 162                 {
 163                         var scheduler = new TestScheduler ();
 164                         var block = new BatchBlock<int> (2,
 165                                 new GroupingDataflowBlockOptions
 166                                 { Greedy = false, BoundedCapacity = 2, TaskScheduler = scheduler });
 167                         var source1 =
 168                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 169                         var source2 =
 170                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 171                         Assert.IsNotNull (source1.LinkTo (block));
 172                         Assert.IsNotNull (source2.LinkTo (block));
 173
 174                         Assert.IsTrue (source1.Post (11));
 175                         Assert.IsTrue (source2.Post (21));
 176
 177                         scheduler.ExecuteAll ();
 178
 179                         Assert.IsTrue (source1.Post (12));
 180                         Assert.IsTrue (source2.Post (22));
 181
 182                         scheduler.ExecuteAll ();
 183
 184                         int i;
 185                         Assert.IsTrue (source1.TryReceive (out i));
 186                         Assert.AreEqual (12, i);
 187
 188                         Assert.IsTrue (source1.Post (13));
 189
 190                         int[] batch;
 191                         Assert.IsTrue (block.TryReceive (out batch));
 192                         CollectionAssert.AreEquivalent (new[] { 11, 21 }, batch);
 193
 194                         scheduler.ExecuteAll ();
 195
 196                         Assert.IsTrue (block.TryReceive (out batch));
 197                         CollectionAssert.AreEquivalent (new[] { 13, 22 }, batch);
 198                 }
 199
 200                 [Test]
 201                 public void GreedyBatchWithBoundedCapacityTest ()
 202                 {
 203                         var scheduler = new TestScheduler ();
 204                         var block = new BatchBlock<int> (3,
 205                                 new GroupingDataflowBlockOptions
 206                                 { Greedy = true, BoundedCapacity = 3, TaskScheduler = scheduler });
 207
 208                         Assert.IsTrue (block.Post (1));
 209                         Assert.IsTrue (block.Post (2));
 210
 211                         block.TriggerBatch ();
 212
 213                         scheduler.ExecuteAll ();
 214
 215                         Assert.IsTrue (block.Post (3));
 216                         Assert.IsFalse (block.Post (4));
 217
 218                         int[] batch;
 219                         Assert.IsTrue (block.TryReceive (out batch));
 220                         CollectionAssert.AreEqual (new[] { 1, 2 }, batch);
 221
 222                         Assert.IsTrue (block.Post (5));
 223                         Assert.IsTrue (block.Post (6));
 224                 }
 225
 226                 [Test]
 227                 public void NonGreedyBatchWithBoundedCapacityTriggerTest ()
 228                 {
 229                         var scheduler = new TestScheduler ();
 230                         var block = new BatchBlock<int> (3,
 231                                 new GroupingDataflowBlockOptions
 232                                 { Greedy = false, BoundedCapacity = 3, TaskScheduler = scheduler });
 233                         var source1 =
 234                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 235                         var source2 =
 236                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 237                         Assert.IsNotNull (source1.LinkTo (block));
 238                         Assert.IsNotNull (source2.LinkTo (block));
 239
 240                         // trigger 2 and then trigger 1 with capacity of 3
 241
 242                         Assert.IsTrue (source1.Post (11));
 243                         Assert.IsTrue (source2.Post (21));
 244                         block.TriggerBatch ();
 245
 246                         scheduler.ExecuteAll ();
 247
 248                         Assert.IsTrue (source1.Post (12));
 249                         block.TriggerBatch ();
 250
 251                         scheduler.ExecuteAll ();
 252
 253                         int i;
 254                         Assert.IsFalse (source1.TryReceive (out i));
 255
 256                         int[] batch;
 257                         Assert.IsTrue (block.TryReceive (out batch));
 258                         CollectionAssert.AreEquivalent (new[] { 11, 21 }, batch);
 259
 260                         Assert.IsTrue (block.TryReceive (out batch));
 261                         CollectionAssert.AreEquivalent (new[] { 12 }, batch);
 262                 }
 263
 264                 [Test]
 265                 public void NonGreedyBatchWithBoundedCapacityTriggerTest2 ()
 266                 {
 267                         var scheduler = new TestScheduler ();
 268                         var block = new BatchBlock<int> (3,
 269                                 new GroupingDataflowBlockOptions
 270                                 { Greedy = false, BoundedCapacity = 3, TaskScheduler = scheduler });
 271                         var source1 =
 272                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 273                         var source2 =
 274                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 275                         Assert.IsNotNull (source1.LinkTo (block));
 276                         Assert.IsNotNull (source2.LinkTo (block));
 277
 278                         // trigger 2, then trigger another 2 and then trigger 2 once more
 279                         // while havaing capacity of 3
 280
 281                         Assert.IsTrue (source1.Post (11));
 282                         Assert.IsTrue (source2.Post (21));
 283                         block.TriggerBatch ();
 284
 285                         scheduler.ExecuteAll ();
 286
 287                         Assert.IsTrue (source1.Post (12));
 288                         Assert.IsTrue (source2.Post (22));
 289                         block.TriggerBatch ();
 290
 291                         scheduler.ExecuteAll ();
 292
 293                         Assert.IsTrue (source1.Post (13));
 294                         Assert.IsTrue (source2.Post (23));
 295                         block.TriggerBatch ();
 296
 297                         scheduler.ExecuteAll ();
 298
 299                         int i;
 300                         Assert.IsTrue (source1.TryReceive (out i));
 301                         Assert.AreEqual (13, i);
 302                         Assert.IsTrue (source2.TryReceive (out i));
 303                         Assert.AreEqual (23, i);
 304
 305                         int[] batch;
 306                         Assert.IsTrue (block.TryReceive (out batch));
 307                         CollectionAssert.AreEquivalent (new[] { 11, 21 }, batch);
 308
 309                         Assert.IsTrue (block.TryReceive (out batch));
 310                         CollectionAssert.AreEquivalent (new[] { 12, 22 }, batch);
 311
 312                         Assert.IsFalse (block.TryReceive (out batch));
 313                 }
 314
 315                 [Test]
 316                 public void MaxNumberOfGroupsTest ()
 317                 {
 318                         var scheduler = new TestScheduler ();
 319
 320                         var block = new BatchBlock<int> (2,
 321                                 new GroupingDataflowBlockOptions
 322                                 { MaxNumberOfGroups = 2, TaskScheduler = scheduler });
 323
 324                         Assert.IsTrue (block.Post (1));
 325                         Assert.IsTrue (block.Post (2));
 326
 327                         Assert.IsTrue (block.Post (3));
 328                         Assert.IsTrue (block.Post (4));
 329
 330                         Assert.IsFalse (block.Post (5));
 331
 332                         scheduler.ExecuteAll ();
 333
 334                         int[] batch;
 335                         Assert.IsTrue (block.TryReceive (out batch));
 336                         CollectionAssert.AreEqual (new[] { 1, 2 }, batch);
 337
 338                         Assert.IsTrue (block.TryReceive (out batch));
 339                         CollectionAssert.AreEqual (new[] { 3, 4 }, batch);
 340
 341                         scheduler.ExecuteAll ();
 342
 343                         Assert.IsTrue (block.Completion.Wait (1000));
 344                 }
 345
 346                 [Test]
 347                 public void CompletionWithTriggerTest ()
 348                 {
 349                         var block = new BatchBlock<int> (2);
 350
 351                         Assert.IsTrue (block.Post (1));
 352
 353                         block.TriggerBatch ();
 354
 355                         block.Complete ();
 356
 357                         CollectionAssert.AreEqual (new[] { 1 },
 358                                 block.Receive (TimeSpan.FromMilliseconds (2000)));
 359
 360                         Assert.IsTrue (block.Completion.Wait (1000));
 361                 }
 362
 363                 [Test]
 364                 public void CompletionWithoutTriggerTest ()
 365                 {
 366                         var block = new BatchBlock<int> (2);
 367
 368                         Assert.IsTrue (block.Post (1));
 369                         Assert.IsTrue (block.Post (2));
 370
 371                         block.Complete ();
 372
 373                         CollectionAssert.AreEqual (new[] { 1, 2 },
 374                                 block.Receive (TimeSpan.FromMilliseconds (2000)));
 375
 376                         Assert.IsTrue (block.Completion.Wait (1000));
 377                 }
 378
 379                 [Test]
 380                 public void CompleteTriggersBatchTest ()
 381                 {
 382                         var block = new BatchBlock<int> (2);
 383
 384                         Assert.IsTrue (block.Post (1));
 385
 386                         block.Complete ();
 387
 388                         CollectionAssert.AreEqual (new[] { 1 },
 389                                 block.Receive (TimeSpan.FromMilliseconds (2000)));
 390
 391                         Assert.IsTrue (block.Completion.Wait (1000));
 392                 }
 393
 394                 [Test]
 395                 public void NonGreedyCompleteDoesnTriggerBatchTest ()
 396                 {
 397                         var scheduler = new TestScheduler ();
 398                         var block = new BatchBlock<int> (2,
 399                                 new GroupingDataflowBlockOptions
 400                                 { Greedy = false, TaskScheduler = scheduler });
 401                         var source =
 402                                 new BufferBlock<int> (new DataflowBlockOptions { TaskScheduler = scheduler });
 403
 404                         Assert.IsNotNull (source.LinkTo (block));
 405
 406                         Assert.IsTrue (source.Post (1));
 407
 408                         block.Complete ();
 409
 410                         int[] batch;
 411                         Assert.IsFalse (block.TryReceive (out batch));
 412
 413                         Assert.IsTrue (block.Completion.Wait (1000));
 414                 }
 415
 416                 [Test]
 417                 public void NonGreedyMaxNumberOfGroupsTest ()
 418                 {
 419                         var scheduler = new TestScheduler ();
 420                         var block = new BatchBlock<int> (2,
 421                                 new GroupingDataflowBlockOptions
 422                                 { MaxNumberOfGroups = 1, Greedy = false, TaskScheduler = scheduler });
 423                         ITargetBlock<int> target = block;
 424                         var source1 = new TestSourceBlock<int> ();
 425                         var source2 = new TestSourceBlock<int> ();
 426
 427                         var header1 = new DataflowMessageHeader (1);
 428                         source1.AddMessage (header1, 11);
 429                         source2.AddMessage (header1, 21);
 430
 431                         Assert.AreEqual (DataflowMessageStatus.Postponed,
 432                                 target.OfferMessage (header1, 11, source1, false));
 433                         Assert.AreEqual (DataflowMessageStatus.Postponed,
 434                                 target.OfferMessage (header1, 21, source2, false));
 435
 436                         scheduler.ExecuteAll ();
 437
 438                         Assert.IsTrue (source1.WasConsumed (header1));
 439                         Assert.IsTrue (source2.WasConsumed (header1));
 440
 441                         var header2 = new DataflowMessageHeader (2);
 442                         Assert.AreEqual (DataflowMessageStatus.DecliningPermanently,
 443                                 target.OfferMessage (header2, 21, source1, false));
 444
 445                         int[] batch;
 446                         Assert.IsTrue (block.TryReceive (out batch));
 447                         CollectionAssert.AreEquivalent (new[] { 11, 21 }, batch);
 448
 449                         Assert.IsTrue (block.Completion.Wait (1000));
 450                 }
 451         }
 452 }