Source Code for Me (s-c.me)

Copy Source | Copy HTML

1. using System;
2. using System.Collections.Generic;
3. using System.Diagnostics;
4. using System.Threading;
5. using Wintellect.Threading.AsyncProgModel;
6. using AsyncMethod = System.Func<Wintellect.Threading.AsyncProgModel.AsyncEnumerator,
7.                                 System.Collections.Generic.IEnumerator<System.Int32>>;
8.  
9. namespace FunWithLambdas
10. {
11.   public class DependencyManager
12.   {
13.     private Dictionary<int, List<int>> dependenciesFromTo;
14.     private ManualResetEvent doneEvent;
15.     private readonly Dictionary<int, OperationData> operations = new Dictionary<int, OperationData>();
16.     private int remainingCount;
17.     private readonly object stateLock = new object();
18.     public event EventHandler<OperationCompletedEventArgs> OperationCompleted;
19.  
20.     public void AddOperation(int id, AsyncMethod operation, params int[] dependencies)
21.     {
22.       if (operation == null)
23.         throw new ArgumentNullException("operation");
24.       if (dependencies == null)
25.         throw new ArgumentNullException("dependencies");
26.  
27.       var data = new OperationData
28.       {
29.         Context = ExecutionContext.Capture(),
30.         Id = id,
31.         Operation = operation,
32.         Dependencies = dependencies
33.       };
34.       operations.Add(id, data);
35.     }
36.  
37.     public void Execute()
38.     {
39.       VerifyThatAllOperationsHaveBeenRegistered();
40.       VerifyThereAreNoCycles();
41.       // Fill dependency data structures
42.       dependenciesFromTo = new Dictionary<int, List<int>>();
43.       foreach (var op in operations.Values)
44.       {
45.         op.NumRemainingDependencies = op.Dependencies.Length;
46.  
47.         foreach (var from in op.Dependencies)
48.         {
49.           List<int> toList;
50.           if (!dependenciesFromTo.TryGetValue(from, out toList))
51.           {
52.             toList = new List<int>();
53.             dependenciesFromTo.Add(from, toList);
54.           }
55.           toList.Add(op.Id);
56.         }
57.       }
58.  
59.       // Launch and wait
60.       remainingCount = operations.Count;
61.       using (doneEvent = new ManualResetEvent(false))
62.       {
63.         lock (stateLock)
64.         {
65.           foreach (var op in operations.Values)
66.           {
67.             if (op.NumRemainingDependencies ==  0)
68.               QueueOperation(op);
69.           }
70.         }
71.         doneEvent.WaitOne();
72.       }
73.     }
74.  
75.  
76.     private List<int> CreateTopologicalSort()
77.     {
78.       // Build up the dependencies graph
79.       var dependenciesToFrom = new Dictionary<int, List<int>>();
80.       var dependenciesFromTo = new Dictionary<int, List<int>>();
81.       foreach (var op in operations.Values)
82.       {
83.         dependenciesToFrom.Add(op.Id, new List<int>(op.Dependencies));
84.         foreach (var depId in op.Dependencies)
85.         {
86.           List<int> ids;
87.           if (!dependenciesFromTo.TryGetValue(depId, out ids))
88.           {
89.             ids = new List<int>();
90.             dependenciesFromTo.Add(depId, ids);
91.           }
92.           ids.Add(op.Id);
93.         }
94.       }
95.  
96.       // Create the sorted list
97.       var overallPartialOrderingIds = new List<int>(dependenciesToFrom.Count);
98.       var thisIterationIds = new List<int>(dependenciesToFrom.Count);
99.       while (dependenciesToFrom.Count >  0)
100.       {
101.         thisIterationIds.Clear();
102.         foreach (var item in dependenciesToFrom)
103.         {
104.           // If an item has zero input operations, remove it.
105.           if (item.Value.Count ==  0)
106.           {
107.             thisIterationIds.Add(item.Key);
108.  
109.             // Remove all outbound edges
110.             List<int> depIds;
111.             if (dependenciesFromTo.TryGetValue(item.Key, out depIds))
112.             {
113.               foreach (var depId in depIds)
114.               {
115.                 dependenciesToFrom[depId].Remove(item.Key);
116.               }
117.             }
118.           }
119.         }
120.  
121.         // If nothing was found to remove, there's no valid sort.
122.         if (thisIterationIds.Count ==  0) return null;
123.  
124.         // Remove the found items from the dictionary and 
125.         // add them to the overall ordering
126.         foreach (var id in thisIterationIds) dependenciesToFrom.Remove(id);
127.         overallPartialOrderingIds.AddRange(thisIterationIds);
128.       }
129.  
130.       return overallPartialOrderingIds;
131.     }
132.  
133.     private void OnOperationCompleted(OperationData data)
134.     {
135.       var handler = OperationCompleted;
136.       if (handler != null)
137.         handler(this, new OperationCompletedEventArgs(
138.           data.Id, data.Start, data.End));
139.     }
140.  
141.     private void ProcessOperation(OperationData data)
142.     {
143.       //// Time and run the operation's delegate
144.       data.Start = DateTimeOffset.Now;
145.  
146.       AsyncEnumerator ae = new AsyncEnumerator();
147.       if (data.Context != null)
148.       {
149.         ExecutionContext.Run(data.Context.CreateCopy(),
150.           op => ae.Execute(
151.             ((OperationData)op).Operation(ae)),
152.             data);
153.       }
154.       else data.Operation(ae);
155.  
156.       data.End = DateTimeOffset.Now;
157.  
158.  
159.       // Raise the operation completed event
160.       OnOperationCompleted(data);
161.  
162.       // Signal to all that depend on this operation of its
163.       // completion, and potentially launch newly available
164.       lock (stateLock)
165.       {
166.         List<int> toList;
167.         if (dependenciesFromTo.TryGetValue(data.Id, out toList))
168.         {
169.           foreach (var targetId in toList)
170.           {
171.             OperationData targetData = operations[targetId];
172.             if (--targetData.NumRemainingDependencies ==  0)
173.               QueueOperation(targetData);
174.           }
175.         }
176.         dependenciesFromTo.Remove(data.Id);
177.  
178.  
179.         if (--remainingCount ==  0) doneEvent.Set();
180.       }
181.     }
182.  
183.     private void QueueOperation(OperationData data)
184.     {
185.       ThreadPool.UnsafeQueueUserWorkItem(state =>
186.         ProcessOperation((OperationData)state), data);
187.     }
188.  
189.     private void VerifyThatAllOperationsHaveBeenRegistered()
190.     {
191.       foreach (var op in operations.Values)
192.         foreach (var dependency in op.Dependencies)
193.           if (!operations.ContainsKey(dependency))
194.           {
195.             throw new InvalidOperationException(
196.               "Missing operation: " + dependency);
197.           }
198.     }
199.  
200.     private void VerifyThereAreNoCycles()
201.     {
202.       if (CreateTopologicalSort() == null)
203.         throw new InvalidOperationException("Cycle detected");
204.     }
205.  
206.  
207.  
208.  
209.  
210.     public class OperationCompletedEventArgs : EventArgs
211.     {
212.       internal OperationCompletedEventArgs(
213.         int id, DateTimeOffset start, DateTimeOffset end)
214.       {
215.         Id = id; Start = start; End = end;
216.       }
217.       public DateTimeOffset End { get; private set; }
218.       public int Id { get; private set; }
219.       public DateTimeOffset Start { get; private set; }
220.     }
221.  
222.     private class OperationData
223.     {
224.       internal ExecutionContext Context;
225.       internal int[] Dependencies;
226.       internal int Id;
227.       internal int NumRemainingDependencies;
228.       internal AsyncMethod Operation;
229.  
230.       internal DateTimeOffset Start, End;
231.     }
232.   }
233.  
234.   class Program
235.   {
236.     static IEnumerator<Int32> WasteTime(AsyncEnumerator ae)
237.     {
238.       Console.WriteLine("Wasting one second on thread " + Thread.CurrentThread.ManagedThreadId);
239.       Thread.Sleep(1000);
240.       yield break;
241.     }
242.  
243.     static void Main()
244.     {
245.       var dm = new DependencyManager();
246.       AsyncMethod op = WasteTime;
247.       dm.AddOperation(1, op);
248.       dm.AddOperation(2, op);
249.       dm.AddOperation(3, op);
250.       dm.AddOperation(4, op, 1);
251.       dm.AddOperation(5, op, 1, 2, 3);
252.       dm.AddOperation(6, op, 3, 4);
253.       dm.AddOperation(7, op, 5, 6);
254.       dm.AddOperation(8, op, 5);
255.  
256.       Stopwatch st = new Stopwatch();
257.       st.Start();
258.       dm.Execute();
259.       st.Stop();
260.       Console.WriteLine(st.Elapsed);
261.       Console.ReadKey();
262.     }
263.   }
264. }