Mercurial > pub > ImplabNet

comparison Implab/Parallels/DispatchPool.cs @ 16:5a4b735ba669 promises

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
sync
author cin
date Thu, 07 Nov 2013 20:20:26 +0400
parents 0f982f9b7d4d
children 7cd4a843b4e4
comparison
equal deleted inserted replaced
15:0f982f9b7d4d 16:5a4b735ba669
61 return m_exitRequired != 0; 61 return m_exitRequired != 0;
62 } 62 }
63 } 63 }
64 64
65 bool StartWorker() { 65 bool StartWorker() {
66 var current = m_runningThreads; 66 int current;
67 // use spins to allocate slot for the new thread 67 // use spins to allocate slot for the new thread
68 do { 68 do {
69 current = m_runningThreads;
69 if (current >= m_maxThreads || m_exitRequired != 0) 70 if (current >= m_maxThreads || m_exitRequired != 0)
70 // no more slots left or the pool has been disposed 71 // no more slots left or the pool has been disposed
71 return false; 72 return false;
72 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current)); 73 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
73 74
82 return true; 83 return true;
83 } 84 }
84 85
85 protected abstract bool TryDequeue(out TUnit unit); 86 protected abstract bool TryDequeue(out TUnit unit);
86 87
87 protected virtual void WakeNewWorker() { 88 protected virtual void WakeNewWorker(bool extend) {
88 if (m_suspended > 0) 89 if (m_suspended > 0)
89 m_hasTasks.Set(); 90 m_hasTasks.Set();
90 else 91 else
91 StartWorker(); 92 StartWorker();
93 }
94
95 /// <summary>
96 /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока
97 /// </summary>
98 protected void StartIfIdle() {
99 int threads;
100 do {
101
102 }
103 }
104
105 protected virtual void Suspend() {
106 m_hasTasks.WaitOne();
92 } 107 }
93 108
94 bool FetchTask(out TUnit unit) { 109 bool FetchTask(out TUnit unit) {
95 do { 110 do {
96 // exit if requested 111 // exit if requested
97 if (m_exitRequired != 0) { 112 if (m_exitRequired != 0) {
98 // release the thread slot 113 // release the thread slot
99 int running; 114 var running = Interlocked.Decrement(ref m_runningThreads);
100 do {
101 running = m_runningThreads;
102 } while (running != Interlocked.CompareExchange(ref m_runningThreads, running - 1, running));
103 running--;
104
105 if (running == 0) // it was the last worker 115 if (running == 0) // it was the last worker
106 m_hasTasks.Dispose(); 116 m_hasTasks.Dispose();
107 else 117 else
108 m_hasTasks.Set(); // release next worker 118 m_hasTasks.Set(); // release next worker
109 unit = default(TUnit); 119 unit = default(TUnit);
110 return false; 120 return false;
111 } 121 }
112 122
113 // fetch task 123 // fetch task
114 if (TryDequeue(out unit)) { 124 if (TryDequeue(out unit)) {
115 WakeNewWorker(); 125 WakeNewWorker(true);
116 return true; 126 return true;
117 } 127 }
118 128
119 //no tasks left, exit if the thread is no longer needed 129 //no tasks left, exit if the thread is no longer needed
120 int runningThreads; 130 int runningThreads;
121 bool exit = true; 131 bool exit = true;
122 do { 132 do {
123 runningThreads = m_runningThreads; 133 runningThreads = m_runningThreads;
124 if (runningThreads <= m_minThreads) { 134 if (runningThreads <= m_minThreads) {
135 // check wheather this is the last thread and we have tasks
136
125 exit = false; 137 exit = false;
126 break; 138 break;
127 } 139 }
128 } while (runningThreads != Interlocked.CompareExchange(ref m_runningThreads, runningThreads - 1, runningThreads)); 140 } while (runningThreads != Interlocked.CompareExchange(ref m_runningThreads, runningThreads - 1, runningThreads));
129 141
130 if (exit) { 142 if (exit) {
131 Interlocked.Decrement(ref m_runningThreads);
132 return false; 143 return false;
133 } 144 }
134 145
135 // keep this thread and wait 146 // entering suspend state
136 Interlocked.Increment(ref m_suspended); 147 Interlocked.Increment(ref m_suspended);
137 m_hasTasks.WaitOne(); 148 // keep this thread and wait
149 Suspend();
138 Interlocked.Decrement(ref m_suspended); 150 Interlocked.Decrement(ref m_suspended);
139 } while (true); 151 } while (true);
140 } 152 }
141 153
142 protected abstract void InvokeUnit(TUnit unit); 154 protected abstract void InvokeUnit(TUnit unit);