|
12
|
1 using System;
|
|
|
2 using System.Collections.Generic;
|
|
|
3 using System.Linq;
|
|
|
4 using System.Text;
|
|
|
5 using System.Threading;
|
|
|
6 using System.Diagnostics;
|
|
|
7
|
|
|
8 namespace Implab.Parallels {
|
|
15
|
9 public class WorkerPool : DispatchPool<Action> {
|
|
12
|
10
|
|
15
|
11 MTQueue<Action> m_queue = new MTQueue<Action>();
|
|
|
12 int m_queueLength = 0;
|
|
16
|
13 readonly int m_threshold = 1;
|
|
13
|
14
|
|
16
|
15 public WorkerPool(int minThreads, int maxThreads, int threshold)
|
|
15
|
16 : base(minThreads, maxThreads) {
|
|
16
|
17 m_threshold = threshold;
|
|
|
18 InitPool();
|
|
|
19 }
|
|
|
20
|
|
|
21 public WorkerPool(int minThreads, int maxThreads) :
|
|
|
22 base(minThreads, maxThreads) {
|
|
|
23 InitPool();
|
|
13
|
24 }
|
|
|
25
|
|
15
|
26 public WorkerPool(int threads)
|
|
|
27 : base(threads) {
|
|
16
|
28 InitPool();
|
|
13
|
29 }
|
|
|
30
|
|
15
|
31 public WorkerPool()
|
|
|
32 : base() {
|
|
16
|
33 InitPool();
|
|
13
|
34 }
|
|
|
35
|
|
12
|
36 public Promise<T> Invoke<T>(Func<T> task) {
|
|
|
37 if (task == null)
|
|
|
38 throw new ArgumentNullException("task");
|
|
15
|
39 if (IsDisposed)
|
|
|
40 throw new ObjectDisposedException(ToString());
|
|
12
|
41
|
|
|
42 var promise = new Promise<T>();
|
|
|
43
|
|
15
|
44 EnqueueTask(delegate() {
|
|
13
|
45 try {
|
|
|
46 promise.Resolve(task());
|
|
|
47 } catch (Exception e) {
|
|
|
48 promise.Reject(e);
|
|
|
49 }
|
|
|
50 });
|
|
12
|
51
|
|
|
52 return promise;
|
|
|
53 }
|
|
|
54
|
|
15
|
55 protected void EnqueueTask(Action unit) {
|
|
|
56 Debug.Assert(unit != null);
|
|
16
|
57 var len = Interlocked.Increment(ref m_queueLength);
|
|
15
|
58 m_queue.Enqueue(unit);
|
|
16
|
59
|
|
20
|
60 ExtendPool();
|
|
16
|
61 }
|
|
|
62
|
|
17
|
63 protected override bool ExtendPool() {
|
|
20
|
64 if (m_queueLength <= m_threshold*ActiveThreads)
|
|
16
|
65 // in this case we are in active thread and it request for additional workers
|
|
|
66 // satisfy it only when queue is longer than threshold
|
|
17
|
67 return false;
|
|
|
68 return base.ExtendPool();
|
|
12
|
69 }
|
|
|
70
|
|
15
|
71 protected override bool TryDequeue(out Action unit) {
|
|
|
72 if (m_queue.TryDequeue(out unit)) {
|
|
|
73 Interlocked.Decrement(ref m_queueLength);
|
|
|
74 return true;
|
|
12
|
75 }
|
|
15
|
76 return false;
|
|
12
|
77 }
|
|
|
78
|
|
15
|
79 protected override void InvokeUnit(Action unit) {
|
|
|
80 unit();
|
|
12
|
81 }
|
|
16
|
82
|
|
20
|
83 protected override bool Suspend() {
|
|
16
|
84 if (m_queueLength == 0)
|
|
20
|
85 return base.Suspend();
|
|
|
86 else
|
|
|
87 return true; // spin again without locks...
|
|
16
|
88 }
|
|
12
|
89 }
|
|
|
90 }
|
