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

   1 #pragma warning disable 1634, 1691
   2 using System;
   3 using System.Diagnostics;
   4 using System.Transactions;
   5 using System.Collections;
   6 using System.Collections.Generic;
   7 using System.Workflow.Runtime.Hosting;
   8
   9 namespace System.Workflow.Runtime
  10 {
  11     #region Runtime Batch Implementation
  12
  13     #region WorkBatch
  14
  15     internal enum WorkBatchState
  16     {
  17         Usable,
  18         Merged,
  19         Completed
  20     }
  21
  22     /// <summary>
  23     /// Summary description for Work Batching.
  24     /// </summary>
  25     internal sealed class WorkBatch : IWorkBatch, IDisposable
  26     {
  27         private PendingWorkCollection _pendingWorkCollection;
  28         private object mutex = new object();
  29         private WorkBatchState _state;
  30         private WorkBatchCollection _collection = null;
  31
  32         private WorkBatch()
  33         {
  34         }
  35
  36         internal WorkBatch(WorkBatchCollection workBackCollection)
  37         {
  38             _pendingWorkCollection = new PendingWorkCollection();
  39             _state = WorkBatchState.Usable;
  40             _collection = workBackCollection;
  41         }
  42
  43         internal int Count
  44         {
  45             get { return _pendingWorkCollection.WorkItems.Count; }
  46         }
  47
  48         internal void SetWorkBatchCollection(WorkBatchCollection workBatchCollection)
  49         {
  50             _collection = workBatchCollection;
  51         }
  52
  53         #region IWorkBatch Implementation
  54         /// <summary>
  55         /// Add Work to Batch
  56         /// </summary>
  57         /// <param name="work"></param>
  58         /// <param name="workItem"></param>
  59         void IWorkBatch.Add(IPendingWork work, object workItem)
  60         {
  61             if (_pendingWorkCollection == null)
  62                 throw new ObjectDisposedException("WorkBatch");
  63
  64             lock (this.mutex)
  65             {
  66                 System.Diagnostics.Debug.Assert(this._state == WorkBatchState.Usable, "Trying to add to unusable batch.");
  67
  68                 _pendingWorkCollection.Add(work, _collection.GetNextWorkItemOrderId(work), workItem);
  69             }
  70         }
  71         #endregion
  72
  73         #region Internal Implementation
  74
  75         internal bool IsDirty
  76         {
  77             get
  78             {
  79                 return this._pendingWorkCollection.IsDirty;
  80             }
  81         }
  82         /// <summary>
  83         /// This one commits all the pending work and its items
  84         /// added so far in this batch.
  85         /// </summary>
  86         /// <param name="transaction"></param>
  87         internal void Commit(Transaction transaction)
  88         {
  89             lock (this.mutex)
  90             {
  91                 _pendingWorkCollection.Commit(transaction);
  92             }
  93         }
  94
  95
  96         /// <summary>
  97         /// This one completes the pending work
  98         /// </summary>
  99         /// <param name="succeeded"></param>
 100         internal void Complete(bool succeeded)
 101         {
 102             lock (this.mutex)
 103             {
 104                 if (this._pendingWorkCollection.IsUsable)
 105                 {
 106                     _pendingWorkCollection.Complete(succeeded);
 107                     _pendingWorkCollection.Dispose();
 108                     this._state = WorkBatchState.Completed;
 109                 }
 110             }
 111         }
 112
 113         /// <summary>
 114         /// API for Runtime to call to do Merge Operation: Right now
 115         /// we dont use this because we dont support incoming work collection.
 116         /// </summary>
 117         /// <param name="batch"></param>
 118         internal void Merge(WorkBatch batch)
 119         {
 120             if (batch == null)
 121                 return; //nothing to merge
 122
 123             if (_pendingWorkCollection == null)
 124                 throw new ObjectDisposedException("WorkBatch");
 125
 126             lock (this.mutex)
 127             {
 128                 lock (batch.mutex)
 129                 {
 130                     foreach (KeyValuePair<IPendingWork, SortedList<long, object>> item in batch._pendingWorkCollection.WorkItems)
 131                     {
 132                         //_pendingWorkCollection.AddRange(item.Key, item.Value);
 133                         SortedList<long, object> newItems = item.Value;
 134                         foreach (KeyValuePair<long, object> kvp in newItems)
 135                             _pendingWorkCollection.Add(item.Key, kvp.Key, kvp.Value);
 136                     }
 137                 }
 138
 139                 this._state = WorkBatchState.Merged;
 140             }
 141         }
 142         #endregion
 143
 144         #region IDisposable Members
 145         public void Dispose()
 146         {
 147             Dispose(true);
 148             GC.SuppressFinalize(this);
 149         }
 150
 151         private void Dispose(bool disposing)
 152         {
 153             if (disposing)
 154             {
 155                 _pendingWorkCollection.Dispose();
 156                 _pendingWorkCollection = null;
 157             }
 158         }
 159         #endregion
 160
 161         #region PendingWorkCollection implementation
 162
 163         /// <summary>
 164         /// Pending Work Implementation
 165         /// </summary>
 166         internal sealed class PendingWorkCollection : IDisposable
 167         {
 168             Dictionary<IPendingWork, SortedList<long, object>> Items;
 169
 170             #region Internal Implementation
 171             internal PendingWorkCollection()
 172             {
 173                 Items = new Dictionary<IPendingWork, SortedList<long, object>>();
 174             }
 175
 176             internal Dictionary<IPendingWork, SortedList<long, object>> WorkItems
 177             {
 178                 get
 179                 {
 180                     return Items;
 181                 }
 182             }
 183
 184             internal bool IsUsable
 185             {
 186                 get
 187                 {
 188                     return this.Items != null;
 189                 }
 190             }
 191
 192             internal bool IsDirty
 193             {
 194                 get
 195                 {
 196                     if (!this.IsUsable)
 197                         return false;
 198
 199                     //
 200                     // Loop through all pending work items in the collection
 201                     // If any of them assert that they need to commit than the batch is dirty
 202                     foreach (KeyValuePair<IPendingWork, SortedList<long, object>> workItem in this.WorkItems)
 203                     {
 204                         try
 205                         {
 206                             IPendingWork work = workItem.Key;
 207                             if (work.MustCommit(workItem.Value))
 208                                 return true;
 209                         }
 210                         catch (Exception e)
 211                         {
 212                             if (WorkflowExecutor.IsIrrecoverableException(e))
 213                             {
 214 #pragma warning disable 56503
 215                                 throw;
 216 #pragma warning restore 56503
 217                             }
 218                             else
 219                             {
 220                                 // Ignore exceptions and treat condition as false return value;
 221                             }
 222                         }
 223                     }
 224                     //
 225                     // If no one asserted that they need to commit we're not dirty
 226                     return false;
 227                 }
 228             }
 229
 230             internal void Add(IPendingWork work, long orderId, object workItem)
 231             {
 232                 SortedList<long, object> workItems = null;
 233
 234                 if (!Items.TryGetValue(work, out workItems))
 235                 {
 236                     workItems = new SortedList<long, object>();
 237                     Items.Add(work, workItems);
 238                 }
 239                 Debug.Assert(!workItems.ContainsKey(orderId), string.Format(System.Globalization.CultureInfo.InvariantCulture, "List already contains key {0}", orderId));
 240                 workItems.Add(orderId, workItem);
 241                 WorkflowTrace.Runtime.TraceEvent(TraceEventType.Information, 0, "pending work hc {0} added workItem hc {1}", work.GetHashCode(), workItem.GetHashCode());
 242             }
 243
 244             //Commit All Pending Work
 245             internal void Commit(Transaction transaction)
 246             {
 247                 //ignore items param
 248                 foreach (KeyValuePair<IPendingWork, SortedList<long, object>> workItem in Items)
 249                 {
 250                     IPendingWork work = workItem.Key;
 251                     List<object> values = new List<object>(workItem.Value.Values);
 252                     work.Commit(transaction, values);
 253                 }
 254             }
 255
 256             //Complete All Pending Work
 257             internal void Complete(bool succeeded)
 258             {
 259                 foreach (KeyValuePair<IPendingWork, SortedList<long, object>> workItem in Items)
 260                 {
 261                     IPendingWork work = workItem.Key;
 262                     List<object> values = new List<object>(workItem.Value.Values);
 263                     try
 264                     {
 265                         work.Complete(succeeded, values);
 266                     }
 267                     catch (Exception e)
 268                     {
 269                         if (WorkflowExecutor.IsIrrecoverableException(e))
 270                         {
 271                             throw;
 272                         }
 273                         else
 274                         {
 275                             WorkflowTrace.Runtime.TraceEvent(TraceEventType.Warning, 0, "Work Item {0} threw exception on complete notification", workItem.GetType());
 276                         }
 277                     }
 278                 }
 279             }
 280
 281             #endregion
 282
 283             #region IDisposable Members
 284             public void Dispose()
 285             {
 286                 Dispose(true);
 287                 GC.SuppressFinalize(this);
 288             }
 289
 290             private void Dispose(bool disposing)
 291             {
 292                 if (disposing && Items != null)
 293                 {
 294                     Items.Clear();
 295                     Items = null;
 296                 }
 297             }
 298
 299             #endregion
 300         }
 301         #endregion
 302     }
 303     #endregion
 304
 305     #region WorkBatchCollection
 306     /// <summary>
 307     /// collection of name to Batch
 308     /// </summary>
 309     internal sealed class WorkBatchCollection : Dictionary<object, WorkBatch>
 310     {
 311         object transientBatchID = new object();
 312         private object mutex = new object();
 313         //
 314         // All access must be through Interlocked.* methods
 315         private long _workItemOrderId = 0;
 316
 317         internal long WorkItemOrderId
 318         {
 319             get
 320             {
 321                 return Threading.Interlocked.Read(ref _workItemOrderId);
 322             }
 323             set
 324             {
 325                 Debug.Assert(value >= _workItemOrderId, "New value for WorkItemOrderId must be greater than the current value");
 326                 lock (mutex)
 327                 {
 328                     _workItemOrderId = value;
 329                 }
 330             }
 331         }
 332
 333         internal long GetNextWorkItemOrderId(IPendingWork pendingWork)
 334         {
 335             return Threading.Interlocked.Increment(ref _workItemOrderId);
 336         }
 337         /// <summary>
 338         /// A new batch is created per atomic scope or any
 339         /// required sub batches. An example of an optional sub batch
 340         /// could be a batch created for Send activities
 341         /// </summary>
 342         /// <param name="id"></param>
 343         /// <returns></returns>
 344         internal IWorkBatch GetBatch(object id)
 345         {
 346             WorkBatch batch = null;
 347
 348             lock (mutex)
 349             {
 350                 if (this.TryGetValue(id, out batch))
 351                     return batch;
 352
 353                 batch = new WorkBatch(this);
 354                 Add(id, batch);
 355             }
 356
 357             return batch;
 358         }
 359
 360         /// <summary>
 361         /// Find a batch for a given id without creating it.
 362         /// </summary>
 363         /// <param name="id">batch key</param>
 364         /// <returns>batch or null if not found</returns>
 365         private WorkBatch FindBatch(object id)
 366         {
 367             WorkBatch batch = null;
 368             lock (mutex)
 369             {
 370                 TryGetValue(id, out batch);
 371             }
 372
 373             return batch;
 374         }
 375
 376         internal IWorkBatch GetTransientBatch()
 377         {
 378             return GetBatch(transientBatchID);
 379         }
 380
 381         internal WorkBatch GetMergedBatch()
 382         {
 383             lock (mutex)
 384             {
 385                 WorkBatch batch = new WorkBatch(this);
 386
 387                 foreach (WorkBatch existingBatch in this.Values)
 388                 {
 389                     batch.Merge(existingBatch);
 390                 }
 391                 //Copy of all the items merged in one batch.
 392                 //Order is preserved in the same way batches are created.
 393                 return batch;
 394             }
 395         }
 396
 397         internal void RollbackBatch(object id)
 398         {
 399             lock (mutex)
 400             {
 401                 WorkBatch batch = FindBatch(id);
 402                 if (batch != null)
 403                 {
 404                     batch.Complete(false);
 405                     batch.Dispose();
 406                     Remove(id);
 407                 }
 408             }
 409         }
 410
 411         // Rollback all sub batches, calling "complete(false)" on all entries.
 412         internal void RollbackAllBatchedWork()
 413         {
 414             lock (mutex)
 415             {
 416                 foreach (WorkBatch batch in this.Values)
 417                 {
 418                     batch.Complete(false);
 419                     batch.Dispose();
 420                 }
 421                 Clear(); // clear the collection
 422             }
 423         }
 424
 425         // Clear sub batches after successful commit/complete.
 426         internal void ClearSubBatches()
 427         {
 428             lock (mutex)
 429             {
 430                 foreach (WorkBatch existingBatch in this.Values)
 431                 {
 432                     existingBatch.Dispose();
 433                 }
 434                 Clear(); // clear the collection
 435             }
 436         }
 437
 438         internal void ClearTransientBatch()
 439         {
 440             RollbackBatch(transientBatchID);
 441         }
 442     }
 443     #endregion
 444
 445     #endregion
 446 }