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

   1 // BoundedCapacityTest.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.Threading;
  26 using System.Threading.Tasks;
  27 using System.Threading.Tasks.Dataflow;
  28 using NUnit.Framework;
  29
  30 namespace MonoTests.System.Threading.Tasks.Dataflow {
  31         [TestFixture]
  32         public class BoundedCapacityTest {
  33                 [Test]
  34                 public void PostTest ()
  35                 {
  36                         var block =
  37                                 new BufferBlock<int> (new DataflowBlockOptions { BoundedCapacity = 1 });
  38                         Assert.IsTrue (block.Post (1));
  39                         Assert.IsFalse (block.Post (2));
  40                         Assert.AreEqual (1, block.Receive ());
  41                         Assert.IsTrue (block.Post (3));
  42                         Assert.AreEqual (3, block.Receive ());
  43                 }
  44
  45                 [Test]
  46                 public void OfferMessageTest ()
  47                 {
  48                         var block =
  49                                 new BufferBlock<int> (new DataflowBlockOptions { BoundedCapacity = 1 });
  50                         ITargetBlock<int> target = block;
  51
  52                         Assert.AreEqual (DataflowMessageStatus.Accepted,
  53                                 target.OfferMessage (new DataflowMessageHeader (1), 42, null, false));
  54                         Assert.AreEqual (DataflowMessageStatus.Declined,
  55                                 target.OfferMessage (new DataflowMessageHeader (2), 43, null, false));
  56
  57                         Assert.AreEqual (42, block.Receive ());
  58
  59                         Assert.AreEqual (DataflowMessageStatus.Accepted,
  60                                 target.OfferMessage (new DataflowMessageHeader (3), 44, null, false));
  61
  62                         Assert.AreEqual (44, block.Receive ());
  63                 }
  64
  65                 [Test]
  66                 public void OfferMessageWithSourceTest ()
  67                 {
  68                         var block =
  69                                 new BufferBlock<int> (new DataflowBlockOptions { BoundedCapacity = 1 });
  70                         ITargetBlock<int> target = block;
  71                         var source = new TestSourceBlock<int> ();
  72
  73                         Assert.AreEqual (DataflowMessageStatus.Accepted,
  74                                 target.OfferMessage (new DataflowMessageHeader (1), 42, source, false));
  75                         var header = new DataflowMessageHeader (2);
  76                         source.AddMessage (header, 43);
  77                         Assert.AreEqual (DataflowMessageStatus.Postponed,
  78                                 target.OfferMessage (header, 43, source, false));
  79
  80                         Assert.AreEqual (42, block.Receive (TimeSpan.FromMilliseconds (1000)));
  81
  82                         Assert.IsFalse (block.Completion.Wait (100));
  83
  84                         Assert.IsTrue (source.WasConsumed (header));
  85
  86                         Assert.AreEqual (43, block.Receive (TimeSpan.FromMilliseconds (1000)));
  87
  88                         Assert.AreEqual (DataflowMessageStatus.Accepted,
  89                                 target.OfferMessage (new DataflowMessageHeader (3), 44, source, false));
  90
  91                         Assert.AreEqual (44, block.Receive ());
  92                 }
  93
  94                 [Test]
  95                 public void TransformManyBlockTest ()
  96                 {
  97                         var block = new TransformManyBlock<int, int> (
  98                                 i => new[] { -i, i },
  99                                 new ExecutionDataflowBlockOptions { BoundedCapacity = 1 });
 100
 101                         Assert.IsTrue (block.Post (1));
 102                         Assert.IsFalse (block.Post (2));
 103
 104                         Assert.IsFalse (block.Completion.Wait (100));
 105
 106                         Assert.IsFalse (block.Post (3));
 107
 108                         Assert.AreEqual (-1, block.Receive ());
 109
 110                         Assert.IsFalse (block.Post (4));
 111
 112                         Assert.AreEqual (1, block.Receive ());
 113
 114                         Assert.IsTrue (block.Post (5));
 115
 116                         Assert.AreEqual (-5, block.Receive ());
 117                         Assert.AreEqual (5, block.Receive ());
 118                 }
 119
 120                 [Test]
 121                 public void TransformFullTest ()
 122                 {
 123                         var scheduler = new TestScheduler ();
 124
 125                         int n = 0;
 126                         var transform = new TransformBlock<int, int> (
 127                                 i => Interlocked.Increment (ref n),
 128                                 new ExecutionDataflowBlockOptions
 129                                 { BoundedCapacity = 2, TaskScheduler = scheduler });
 130
 131                         Assert.IsTrue (transform.Post (1));
 132                         Assert.IsTrue (transform.Post (2));
 133
 134                         AssertHelper.GreaterOrEqual (scheduler.ExecuteAll (), 1);
 135
 136                         Assert.AreEqual (2, Volatile.Read (ref n));
 137                 }
 138
 139                 [Test]
 140                 public void TransformManyOverfullTest ()
 141                 {
 142                         var scheduler = new TestScheduler ();
 143
 144                         int n = 0;
 145                         var transform = new TransformManyBlock<int, int> (
 146                                 i =>
 147                                 {
 148                                         Interlocked.Increment (ref n);
 149                                         return new[] { -i, i };
 150                                 },
 151                                 new ExecutionDataflowBlockOptions
 152                                 { BoundedCapacity = 2, TaskScheduler = scheduler });
 153
 154                         Assert.IsTrue (transform.Post (1));
 155                         Assert.IsTrue (transform.Post (2));
 156
 157                         AssertHelper.GreaterOrEqual (scheduler.ExecuteAll (), 1);
 158
 159                         Assert.AreEqual (2, Volatile.Read (ref n));
 160                 }
 161
 162                 int n;
 163
 164                 [Test]
 165                 public void TransformManyOverfullTest2 ()
 166                 {
 167                         var scheduler = new TestScheduler ();
 168
 169                         n = 0;
 170                         var transform = new TransformManyBlock<int, int> (
 171                                 i => ComputeResults (),
 172                                 new ExecutionDataflowBlockOptions
 173                                 { BoundedCapacity = 100, TaskScheduler = scheduler });
 174
 175                         for (int i = 0; i < 100; i++)
 176                                 Assert.IsTrue (transform.Post (i));
 177
 178                         Assert.IsFalse (transform.Post (101));
 179
 180                         AssertHelper.GreaterOrEqual (scheduler.ExecuteAll (), 1);
 181
 182                         Assert.IsFalse (transform.Post (102));
 183
 184                         Assert.AreEqual (10000, Volatile.Read (ref n));
 185                 }
 186
 187                 IEnumerable<int> ComputeResults ()
 188                 {
 189                         for (int j = 0; j < 100; j++)
 190                                 yield return Interlocked.Increment (ref n);
 191                 }
 192
 193                 [Test]
 194                 public void MultipleOffersTest ()
 195                 {
 196                         var scheduler = new TestScheduler ();
 197                         var block = new BufferBlock<int> (
 198                                 new DataflowBlockOptions { BoundedCapacity = 1, TaskScheduler = scheduler });
 199                         var target = (ITargetBlock<int>)block;
 200                         var source = new TestSourceBlock<int> ();
 201
 202                         var header1 = new DataflowMessageHeader (1);
 203                         Assert.AreEqual (DataflowMessageStatus.Accepted,
 204                                 target.OfferMessage (header1, 41, source, false));
 205
 206                         var header2 = new DataflowMessageHeader (2);
 207                         source.AddMessage (header2, 42);
 208                         Assert.AreEqual (DataflowMessageStatus.Postponed,
 209                                 target.OfferMessage (header2, 42, source, false));
 210
 211                         var header3 = new DataflowMessageHeader (3);
 212                         source.AddMessage (header3, 43);
 213                         Assert.AreEqual (DataflowMessageStatus.Postponed,
 214                                 target.OfferMessage (header3, 43, source, false));
 215
 216                         Assert.AreEqual (41, block.Receive ());
 217                         scheduler.ExecuteAll ();
 218                         Assert.IsTrue (source.WasConsumed (header3));
 219                         Assert.IsFalse (source.WasConsumed (header2));
 220                 }
 221
 222                 [Test]
 223                 public void DontConsumePostponedAfterCompleteTest ()
 224                 {
 225                         var scheduler = new TestScheduler ();
 226                         var block = new BufferBlock<int> (
 227                                 new DataflowBlockOptions { BoundedCapacity = 1, TaskScheduler = scheduler });
 228                         var target = (ITargetBlock<int>)block;
 229                         var source = new TestSourceBlock<int> ();
 230
 231                         Assert.IsTrue (block.Post (11));
 232
 233                         var header = new DataflowMessageHeader (1);
 234                         source.AddMessage (header, 12);
 235                         Assert.AreEqual (DataflowMessageStatus.Postponed,
 236                                 target.OfferMessage (header, 12, source, false));
 237
 238                         block.Complete ();
 239
 240                         Assert.AreEqual (11, block.Receive ());
 241
 242                         scheduler.ExecuteAll ();
 243
 244                         Assert.IsFalse (source.WasConsumed (header));
 245                 }
 246         }
 247
 248         class TestSourceBlock<T> : ISourceBlock<T> {
 249                 readonly Dictionary<DataflowMessageHeader, T> messages =
 250                         new Dictionary<DataflowMessageHeader, T> ();
 251
 252                 readonly HashSet<DataflowMessageHeader> consumed =
 253                         new HashSet<DataflowMessageHeader> ();
 254                 readonly HashSet<DataflowMessageHeader> reserved =
 255                         new HashSet<DataflowMessageHeader> ();
 256
 257                 public void Complete ()
 258                 {
 259                         throw new NotImplementedException ();
 260                 }
 261
 262                 public void Fault (Exception exception)
 263                 {
 264                         throw new NotImplementedException ();
 265                 }
 266
 267                 public Task Completion { get; private set; }
 268
 269                 public void AddMessage (DataflowMessageHeader header, T item)
 270                 {
 271                         messages.Add (header, item);
 272                 }
 273
 274                 public bool WasConsumed (DataflowMessageHeader header)
 275                 {
 276                         return consumed.Contains (header);
 277                 }
 278
 279                 public bool WasReserved (DataflowMessageHeader header)
 280                 {
 281                         return reserved.Contains (header);
 282                 }
 283
 284                 public Action ConsumeWaiter { get; set; }
 285
 286                 public T ConsumeMessage (DataflowMessageHeader messageHeader,
 287                                          ITargetBlock<T> target, out bool messageConsumed)
 288                 {
 289                         T item;
 290                         if (messages.TryGetValue (messageHeader, out item)) {
 291                                 if (ConsumeWaiter != null)
 292                                         ConsumeWaiter ();
 293                                 messages.Remove (messageHeader);
 294                                 consumed.Add (messageHeader);
 295                                 messageConsumed = true;
 296                                 return item;
 297                         }
 298                         messageConsumed = false;
 299                         return default(T);
 300                 }
 301
 302                 public IDisposable LinkTo (ITargetBlock<T> target,
 303                                            DataflowLinkOptions linkOptions)
 304                 {
 305                         throw new NotImplementedException ();
 306                 }
 307
 308                 public bool ReserveMessage (DataflowMessageHeader messageHeader,
 309                                             ITargetBlock<T> target)
 310                 {
 311                         reserved.Add (messageHeader);
 312                         return messages.ContainsKey (messageHeader);
 313                 }
 314
 315                 public void ReleaseReservation (DataflowMessageHeader messageHeader,
 316                                                 ITargetBlock<T> target)
 317                 {
 318                         throw new NotImplementedException ();
 319                 }
 320         }
 321 }