mcs/class/referencesource/System.Workflow.Runtime/Scheduler.cs

   1 using System;
   2 using System.Globalization;
   3 using System.Collections.Generic;
   4 using System.Diagnostics;
   5 using System.IO;
   6 using System.Transactions;
   7 using System.Workflow.ComponentModel;
   8
   9 namespace System.Workflow.Runtime
  10 {
  11     #region Scheduler
  12
  13     // Only one instance of this type is used for a workflow instance.
  14     //
  15     class Scheduler
  16     {
  17         #region data
  18
  19         // state to be persisted for the scheduler
  20         internal static DependencyProperty HighPriorityEntriesQueueProperty = DependencyProperty.RegisterAttached("HighPriorityEntriesQueue", typeof(Queue<SchedulableItem>), typeof(Scheduler));
  21         internal static DependencyProperty NormalPriorityEntriesQueueProperty = DependencyProperty.RegisterAttached("NormalPriorityEntriesQueue", typeof(Queue<SchedulableItem>), typeof(Scheduler));
  22         Queue<SchedulableItem> highPriorityEntriesQueue;
  23         Queue<SchedulableItem> normalPriorityEntriesQueue;
  24
  25         // non-persisted state for the scheduler
  26         WorkflowExecutor rootWorkflowExecutor;
  27         bool empty;
  28         bool canRun;
  29         bool threadRequested;
  30         bool abortOrTerminateRequested;
  31         Queue<SchedulableItem> transactedEntries;
  32         object syncObject = new object();
  33
  34         #endregion data
  35
  36         #region ctors
  37
  38         // loading with some state
  39         public Scheduler(WorkflowExecutor rootExec, bool canRun)
  40         {
  41             this.rootWorkflowExecutor = rootExec;
  42             this.threadRequested = false;
  43
  44             // canRun is true if normal creation
  45             // false if loading from a persisted state. Will be set to true later at ResumeOnIdle
  46             this.canRun = canRun;
  47
  48             this.highPriorityEntriesQueue = (Queue<SchedulableItem>)rootExec.RootActivity.GetValue(Scheduler.HighPriorityEntriesQueueProperty);
  49             this.normalPriorityEntriesQueue = (Queue<SchedulableItem>)rootExec.RootActivity.GetValue(Scheduler.NormalPriorityEntriesQueueProperty);
  50             if (this.highPriorityEntriesQueue == null)
  51             {
  52                 this.highPriorityEntriesQueue = new Queue<SchedulableItem>();
  53                 rootExec.RootActivity.SetValue(Scheduler.HighPriorityEntriesQueueProperty, this.highPriorityEntriesQueue);
  54             }
  55             if (this.normalPriorityEntriesQueue == null)
  56             {
  57                 this.normalPriorityEntriesQueue = new Queue<SchedulableItem>();
  58                 rootExec.RootActivity.SetValue(Scheduler.NormalPriorityEntriesQueueProperty, this.normalPriorityEntriesQueue);
  59             }
  60
  61             this.empty = ((this.normalPriorityEntriesQueue.Count == 0) && (this.highPriorityEntriesQueue.Count == 0));
  62         }
  63
  64         #endregion ctors
  65
  66         #region Misc properties
  67
  68         public override string ToString()
  69         {
  70             return "Scheduler('" + ((Activity)this.RootWorkflowExecutor.WorkflowDefinition).QualifiedName + "')";
  71         }
  72
  73         protected WorkflowExecutor RootWorkflowExecutor
  74         {
  75             get { return this.rootWorkflowExecutor; }
  76         }
  77
  78         public bool IsStalledNow
  79         {
  80             get
  81             {
  82                 return empty;
  83             }
  84         }
  85
  86         public bool CanRun
  87         {
  88             get
  89             {
  90                 return canRun;
  91             }
  92
  93             set
  94             {
  95                 canRun = value;
  96             }
  97         }
  98
  99         internal bool AbortOrTerminateRequested
 100         {
 101             get
 102             {
 103                 return abortOrTerminateRequested;
 104             }
 105             set
 106             {
 107                 abortOrTerminateRequested = value;
 108             }
 109         }
 110
 111         #endregion Misc properties
 112
 113         #region Run work
 114
 115         public void Run()
 116         {
 117             do
 118             {
 119                 this.RootWorkflowExecutor.ProcessQueuedEvents();
 120                 // Get item to run
 121                 SchedulableItem item = GetItemToRun();
 122                 bool runningItem = false;
 123
 124                 // no ready work to run... go away
 125                 if (item == null)
 126                     break;
 127
 128                 Activity itemActivity = null;
 129                 Exception exp = null;
 130
 131                 TransactionalProperties transactionalProperties = null;
 132                 int contextId = item.ContextId;
 133
 134                 // This function gets the root or enclosing while-loop activity
 135                 Activity contextActivity = this.RootWorkflowExecutor.GetContextActivityForId(contextId);
 136                 if (contextActivity == null)
 137                     throw new InvalidOperationException(ExecutionStringManager.InvalidExecutionContext);
 138
 139                 // This is the activity corresponding to the item's ActivityId
 140                 itemActivity = contextActivity.GetActivityByName(item.ActivityId);
 141                 using (new ServiceEnvironment(itemActivity))
 142                 {
 143                     exp = null;
 144                     bool ignoreFinallyBlock = false;
 145
 146                     try
 147                     {
 148                         // item preamble
 149                         // set up the item transactional context if necessary
 150                         //
 151                         Debug.Assert(itemActivity != null, "null itemActivity");
 152                         if (itemActivity == null)
 153                             throw new InvalidOperationException(String.Format(CultureInfo.CurrentCulture, ExecutionStringManager.InvalidActivityName, item.ActivityId));
 154
 155                         Activity atomicActivity = null;
 156                         if (this.RootWorkflowExecutor.IsActivityInAtomicContext(itemActivity, out atomicActivity))
 157                         {
 158                             transactionalProperties = (TransactionalProperties)atomicActivity.GetValue(WorkflowExecutor.TransactionalPropertiesProperty);
 159                             // If we've aborted for any reason stop now!
 160                             // If we attempt to enter a new TransactionScope the com+ context will get corrupted
 161                             // See windows se
 162                             if (!WorkflowExecutor.CheckAndProcessTransactionAborted(transactionalProperties))
 163                             {
 164                                 if (transactionalProperties.TransactionScope == null)
 165                                 {
 166                                     // Use TimeSpan.Zero so scope will not create timeout independent of the transaction
 167                                     // Use EnterpriseServicesInteropOption.Full to flow transaction to COM+
 168                                     transactionalProperties.TransactionScope =
 169                                         new TransactionScope(transactionalProperties.Transaction, TimeSpan.Zero, EnterpriseServicesInteropOption.Full);
 170
 171                                     WorkflowTrace.Runtime.TraceEvent(TraceEventType.Information, 0,
 172                                         "Workflow Runtime: Scheduler: instanceId: " + this.RootWorkflowExecutor.InstanceIdString +
 173                                         "Entered into TransactionScope, Current atomic acitivity " + atomicActivity.Name);
 174                                 }
 175                             }
 176                         }
 177
 178                         // Run the item
 179                         //
 180                         runningItem = true;
 181                         WorkflowTrace.Runtime.TraceEvent(TraceEventType.Information, 1, "Workflow Runtime: Scheduler: InstanceId: {0} : Running scheduled entry: {1}", this.RootWorkflowExecutor.InstanceIdString, item.ToString());
 182
 183                         // running any entry implicitly changes some state of the workflow instance
 184                         this.RootWorkflowExecutor.stateChangedSincePersistence = true;
 185
 186                         item.Run(this.RootWorkflowExecutor);
 187                     }
 188                     catch (Exception e)
 189                     {
 190                         if (WorkflowExecutor.IsIrrecoverableException(e))
 191                         {
 192                             ignoreFinallyBlock = true;
 193                             throw;
 194                         }
 195                         else
 196                         {
 197                             if (transactionalProperties != null)
 198                                 transactionalProperties.TransactionState = TransactionProcessState.AbortProcessed;
 199                             exp = e;
 200                         }
 201                     }
 202                     finally
 203                     {
 204                         if (!ignoreFinallyBlock)
 205                         {
 206                             if (runningItem)
 207                                 WorkflowTrace.Runtime.TraceEvent(TraceEventType.Information, 1, "Workflow Runtime: Scheduler: InstanceId: {0} : Done with running scheduled entry: {1}", this.RootWorkflowExecutor.InstanceIdString, item.ToString());
 208
 209                             // Process exception
 210                             //
 211                             if (exp != null)
 212                             {
 213                                 //
 214                                 this.RootWorkflowExecutor.ExceptionOccured(exp, itemActivity == null ? contextActivity : itemActivity, null);
 215                                 exp = null;
 216                             }
 217                         }
 218                     }
 219                 }
 220             } while (true);
 221         }
 222
 223         private SchedulableItem GetItemToRun()
 224         {
 225             SchedulableItem ret = null;
 226
 227             lock (this.syncObject)
 228             {
 229                 bool workToDo = false;
 230                 if ((this.highPriorityEntriesQueue.Count > 0) || (this.normalPriorityEntriesQueue.Count > 0))
 231                 {
 232                     workToDo = true;
 233
 234                     // If an abort or termination of the workflow has been requested,
 235                     // then the workflow should try to terminate ASAP. Even transaction scopes
 236                     // in progress shouldn't be executed to completion. (Ref: 16534)
 237                     if (this.AbortOrTerminateRequested)
 238                     {
 239                         ret = null;
 240                     }
 241                     // got work to do in the scheduler
 242                     else if ((this.highPriorityEntriesQueue.Count > 0))
 243                     {
 244                         ret = this.highPriorityEntriesQueue.Dequeue();
 245                     }
 246                     else if (this.CanRun)
 247                     {
 248                         // the scheduler can run right now
 249                         //
 250
 251                         // pick an entry to run
 252                         //
 253                         if (((IWorkflowCoreRuntime)this.RootWorkflowExecutor).CurrentAtomicActivity == null &&
 254                             (this.normalPriorityEntriesQueue.Count > 0))
 255                             ret = this.normalPriorityEntriesQueue.Dequeue();
 256                     }
 257                     else
 258                     {
 259                         // scheduler can't run right now.. even though there is ready work
 260                         // do nothing in the scheduler
 261                         ret = null;
 262                     }
 263                 }
 264
 265                 if (!workToDo)
 266                 {
 267                     // no ready work to do in the scheduler...
 268                     // we are gonna return the thread back
 269                     this.empty = true;
 270                 }
 271
 272                 // set it to true only iff there is something to run
 273                 this.threadRequested = (ret != null);
 274             }
 275             return ret;
 276         }
 277
 278         // This method should be called only after we have determined that
 279         // this instance can start running now
 280         public void Resume()
 281         {
 282             canRun = true;
 283
 284             if (!empty)
 285             {
 286                 // There is scheduled work
 287                 // ask the threadprovider for a thread
 288                 this.RootWorkflowExecutor.ScheduleForWork();
 289             }
 290         }
 291
 292         // This method should be called only after we have determined that
 293         // this instance can start running now
 294         public void ResumeIfRunnable()
 295         {
 296             if (!canRun)
 297                 return;
 298
 299             if (!empty)
 300             {
 301                 // There is scheduled work
 302                 // ask the threadprovider for a thread
 303                 this.RootWorkflowExecutor.ScheduleForWork();
 304             }
 305         }
 306         #endregion Run work
 307
 308         #region Schedule work
 309
 310         public void ScheduleItem(SchedulableItem s, bool isInAtomicTransaction, bool transacted)
 311         {
 312             lock (this.syncObject)
 313             {
 314                 WorkflowTrace.Runtime.TraceEvent(TraceEventType.Information, 1, "Workflow Runtime: Scheduler: InstanceId: {0} : Scheduling entry: {1}", this.RootWorkflowExecutor.InstanceIdString, s.ToString());
 315                 // SchedulableItems in AtomicTransaction has higher priority
 316                 Queue<SchedulableItem> q = isInAtomicTransaction ? this.highPriorityEntriesQueue : this.normalPriorityEntriesQueue;
 317                 q.Enqueue(s);
 318
 319                 if (transacted)
 320                 {
 321                     if (transactedEntries == null)
 322                         transactedEntries = new Queue<SchedulableItem>();
 323                     transactedEntries.Enqueue(s);
 324                 }
 325
 326                 if (!this.threadRequested)
 327                 {
 328                     if (this.CanRun)
 329                     {
 330                         this.RootWorkflowExecutor.ScheduleForWork();
 331                         this.threadRequested = true;
 332                     }
 333                 }
 334                 this.empty = false;
 335             }
 336         }
 337
 338         #endregion Schedule work
 339
 340         #region psuedo-transacted support
 341
 342         public void PostPersist()
 343         {
 344             transactedEntries = null;
 345         }
 346
 347         public void Rollback()
 348         {
 349             if (transactedEntries != null && transactedEntries.Count > 0)
 350             {
 351                 // make a list of non-transacted entries
 352                 // @undone: bmalhi: transacted entries only on priority-0
 353
 354                 IEnumerator<SchedulableItem> e = this.normalPriorityEntriesQueue.GetEnumerator();
 355                 Queue<SchedulableItem> newScheduled = new Queue<SchedulableItem>();
 356                 while (e.MoveNext())
 357                 {
 358                     if (!transactedEntries.Contains(e.Current))
 359                         newScheduled.Enqueue(e.Current);
 360                 }
 361
 362                 // clear the scheduled items
 363                 this.normalPriorityEntriesQueue.Clear();
 364
 365                 // schedule the non-transacted items back
 366                 e = newScheduled.GetEnumerator();
 367                 while (e.MoveNext())
 368                     this.normalPriorityEntriesQueue.Enqueue(e.Current);
 369
 370                 transactedEntries = null;
 371             }
 372         }
 373
 374         #endregion psuedo-transacted support
 375     }
 376
 377     #endregion Scheduler
 378 }