mcs/class/System.Threading.Tasks.Dataflow/System.Threading.Tasks.Dataflow/AsyncExecutingMessageBox.cs

   1 // AsyncExecutingMessageBox.cs
   2 //
   3 // Copyright (c) 2011 Jérémie "garuma" Laval
   4 // Copyright (c) 2012 Petr Onderka
   5 //
   6 // Permission is hereby granted, free of charge, to any person obtaining a copy
   7 // of this software and associated documentation files (the "Software"), to deal
   8 // in the Software without restriction, including without limitation the rights
   9 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  10 // copies of the Software, and to permit persons to whom the Software is
  11 // furnished to do so, subject to the following conditions:
  12 //
  13 // The above copyright notice and this permission notice shall be included in
  14 // all copies or substantial portions of the Software.
  15 //
  16 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  21 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  22 // THE SOFTWARE.
  23
  24 using System.Collections.Concurrent;
  25
  26 namespace System.Threading.Tasks.Dataflow {
  27         /// <summary>
  28         /// Message box for executing blocks with asynchrnous
  29         /// (<see cref="Task"/>-returning) actions.
  30         /// </summary>
  31         /// <typeparam name="TInput">Type of the item the block is processing.</typeparam>
  32         /// <typeparam name="TTask">Type of the Task the action is returning.</typeparam>
  33         class AsyncExecutingMessageBox<TInput, TTask>
  34                 : ExecutingMessageBoxBase<TInput>
  35                 where TTask : Task {
  36                 /// <summary>
  37                 /// Represents executing synchrnous part of the action.
  38                 /// </summary>
  39                 /// <param name="task">The Task that was returned by the synchronous part of the action.</param>
  40                 /// <returns>Returns whether an item was processed. Returns <c>false</c> if the queue was empty.</returns>
  41                 public delegate bool AsyncProcessItem (out TTask task);
  42
  43                 readonly AsyncProcessItem processItem;
  44                 readonly Action<TTask> processFinishedTask;
  45
  46                 public AsyncExecutingMessageBox (
  47                         ITargetBlock<TInput> target, BlockingCollection<TInput> messageQueue,
  48                         CompletionHelper compHelper, Func<bool> externalCompleteTester,
  49                         AsyncProcessItem processItem, Action<TTask> processFinishedTask,
  50                         Action outgoingQueueComplete, ExecutionDataflowBlockOptions options)
  51                         : base (
  52                                 target, messageQueue, compHelper, externalCompleteTester,
  53                                 outgoingQueueComplete, options)
  54                 {
  55                         this.processItem = processItem;
  56                         this.processFinishedTask = processFinishedTask;
  57                 }
  58
  59                 /// <summary>
  60                 /// Processes the input queue of the block.
  61                 /// </summary>
  62                 protected override void ProcessQueue ()
  63                 {
  64                         StartProcessQueue ();
  65
  66                         ProcessQueueWithoutStart ();
  67                 }
  68
  69                 /// <summary>
  70                 /// The part of <see cref="ProcessQueue"/> specific to asynchronous execution.
  71                 /// Handles scheduling continuation on the Task returned by the block's action
  72                 /// (or continuing synchrnously if possible).
  73                 /// </summary>
  74                 void ProcessQueueWithoutStart ()
  75                 {
  76                         // catch is needed here, if the Task-returning delegate throws exception itself
  77                         try {
  78                                 int i = 0;
  79                                 while (CanRun (i)) {
  80                                         TTask task;
  81                                         if (!processItem (out task))
  82                                                 break;
  83                                         if (task == null || task.IsCanceled
  84                                             || (task.IsCompleted && !task.IsFaulted)) {
  85                                                 if (processFinishedTask != null)
  86                                                         processFinishedTask (task);
  87                                         } else if (task.IsFaulted) {
  88                                                 CompHelper.RequestFault (task.Exception, false);
  89                                                 break;
  90                                         } else {
  91                                                 task.ContinueWith (
  92                                                         t => TaskFinished ((TTask)t), Options.TaskScheduler);
  93                                                 return;
  94                                         }
  95                                         i++;
  96                                 }
  97                         } catch (Exception e) {
  98                                 CompHelper.RequestFault (e, false);
  99                         }
 100
 101                         FinishProcessQueue ();
 102                 }
 103
 104                 /// <summary>
 105                 /// Handles asynchronously finished Task, continues processing the queue.
 106                 /// </summary>
 107                 void TaskFinished (TTask task)
 108                 {
 109                         if (task.IsFaulted) {
 110                                 CompHelper.RequestFault (task.Exception, false);
 111                                 FinishProcessQueue ();
 112                                 return;
 113                         }
 114
 115                         if (processFinishedTask != null)
 116                                 processFinishedTask (task);
 117
 118                         ProcessQueueWithoutStart ();
 119                 }
 120         }
 121 }