implab/ImplabNet Commit - r21:6a56df4ec59e

DispatchPool works again, but performance is poor in some cases

cin -

r21:6a56df4ec59e promises

parent child

Context file:

r21:6a56df4ec59e

Collapse all files

Implab.Test/AsyncTests.cs +2 -2

                          var pool = new WorkerPool(5, 20);
                          Assert.AreEqual(5, pool.PoolSize);
                          pool.Dispose();
-                         Thread.Sleep(200);
+                         Thread.Sleep(500);
                          Assert.AreEqual(0, pool.PoolSize);
                          pool.Dispose();
                      }
                      [TestMethod]
                      public void ChainedMapTest() {
-                         using (var pool = new WorkerPool(0,100,0)) {
+                         using (var pool = new WorkerPool(0,100,1)) {
                              int count = 10000;
                              double[] args = new double[count];

Implab.suo binary modified 0 0

NO CONTENT: modified file, binary diff hidden

Implab/Parallels/DispatchPool.cs +69 -46

                  public abstract class DispatchPool<TUnit> : IDisposable {
                      readonly int m_minThreads;
                      readonly int m_maxThreads;
-                     int m_createdThreads = 0;
-                     int m_activeThreads = 0;
-                     int m_sleepingThreads = 0;
-                     int m_maxRunningThreads = 0;
-                     int m_exitRequired = 0;
-                     int m_releaseTimeout = 100; // timeout while the working thread will wait for the new tasks before exit
+                     int m_createdThreads = 0; // the current size of the pool
+                     int m_activeThreads = 0; // the count of threads which are active
+                     int m_sleepingThreads = 0; // the count of currently inactive threads
+                     int m_maxRunningThreads = 0; // the meximum reached size of the pool
+                     int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released
+                     int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit
+                     int m_wakeEvents = 0; // the count of wake events
                      AutoResetEvent m_hasTasks = new AutoResetEvent(false);
                      protected DispatchPool(int min, int max) {
                      protected abstract bool TryDequeue(out TUnit unit);
-                     protected virtual bool ExtendPool() {
-                         if (m_sleepingThreads == 0)
-                             // no sleeping workers are available
-                             // try create one
-                             return StartWorker();
-                         else {
-                             // we can get here a race condition when several threads asks to extend pool
-                             // and some sleaping threads are exited due timeout but they are still counted as sleeping
-                             // in that case all of this threads could exit except one
-                             WakePool();
-                             return true;
+                     #region thread execution traits
+                     int SignalThread() {
+                         var signals = Interlocked.Increment(ref m_wakeEvents);
+                         if(signals == 1)
+                             m_hasTasks.Set();
+                         return signals;
-                         }
+                     bool Sleep(int timeout) {
+                         Interlocked.Increment(ref m_sleepingThreads);
+                         if (m_hasTasks.WaitOne(timeout)) {
+                             // this is autoreset event, only one thread can run this block simultaneously
+                             var sleeping = Interlocked.Decrement(ref m_sleepingThreads);
+                             if (Interlocked.Decrement(ref m_wakeEvents) > 0)
+                                 m_hasTasks.Set(); // wake next worker
+                             return true;
+                         } else {
+                             Interlocked.Decrement(ref m_sleepingThreads);
+                             return false;
                      }
+                     }
+                     #endregion
                      /// <summary>
                      /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока
                      /// </summary>
-                     protected void WakePool() {
-                         m_hasTasks.Set(); // wake sleeping thread;
+                     protected void GrowPool() {
+                         if (m_exitRequired != 0)
+                             return;
+                         if (m_sleepingThreads > m_wakeEvents) {
+                             // all sleeping threads may gone
+                             SignalThread(); // wake a sleeping thread;
+                             // we can't check whether signal has been processed
+                             // anyway it may take some time for the thread to start
+                             // we will ensure that at least one thread is running
                          if (AllocateThreadSlot(1)) {
                              // if there were no threads in the pool
                              worker.IsBackground = true;
                              worker.Start();
                          }
+                         } else {
+                             // if there is no sleeping threads in the pool
+                             StartWorker();
+                         }
                      }
-                     bool Sleep(int timeout) {
-                         Interlocked.Increment(ref m_sleepingThreads);
-                         var result = m_hasTasks.WaitOne(timeout);
-                         Interlocked.Decrement(ref m_sleepingThreads);
-                         return result;
+                     }
-                     protected virtual bool Suspend() {
+                     private bool Suspend() {
                          //no tasks left, exit if the thread is no longer needed
                          bool last;
                          bool requestExit;
+                         // if threads have a timeout before releasing
                          if (m_releaseTimeout > 0)
                              requestExit = !Sleep(m_releaseTimeout);
                          else
                              requestExit = true;
+                         if (!requestExit)
+                             return true;
+                         // release unsused thread
                          if (requestExit && ReleaseThreadSlot(out last)) {
                              // in case at the moment the last thread was being released
                              // a new task was added to the queue, we need to try
                              // to revoke the thread to avoid the situation when the task is left unprocessed
-                             if (last && m_hasTasks.WaitOne(0)) {
+                             if (last && Sleep(0)) { // Sleep(0) will fetch pending task or will return false
                                  if (AllocateThreadSlot(1))
                                      return true; // spin again...
                                  else
-                                     // we failed to reallocate the first slot for this thread
-                                     // therefore we need to release the event
-                                     m_hasTasks.Set();
+                                     SignalThread(); // since Sleep(0) has fetched the signal we neet to reschedule it
                              }
                              return false;
                          }
+                         // wait till infinity
                          Sleep(-1);
                          return true;
                          }
                      }
-                     bool FetchTask(out TUnit unit) {
+                     protected abstract void InvokeUnit(TUnit unit);
+                     void Worker() {
+                         TUnit unit;
+                         Interlocked.Increment(ref m_activeThreads);
+                         Sleep(0); // remove wake request if the new thread is started
                          do {
                              // exit if requested
                              if (m_exitRequired != 0) {
                                  if (ReleaseThreadSlotAnyway()) // it was the last worker
                                      m_hasTasks.Dispose();
                                  else
-                                     m_hasTasks.Set(); // wake next worker
+                                     SignalThread(); // wake next worker
                                  unit = default(TUnit);
-                                 return false;
+                                 break;
                              }
                              // fetch task
                              if (TryDequeue(out unit)) {
-                                 ExtendPool();
-                                 return true;
+                                 InvokeUnit(unit);
+                                 continue;
                              }
                              Interlocked.Decrement(ref m_activeThreads);
                              // entering suspend state
                              // keep this thread and wait
                              if (!Suspend())
-                                 return false;
+                                 break;
                              Interlocked.Increment(ref m_activeThreads);
                          } while (true);
+                     }
-                     protected abstract void InvokeUnit(TUnit unit);
-                     void Worker() {
-                         TUnit unit;
-                         Interlocked.Increment(ref m_activeThreads);
-                         while (FetchTask(out unit))
-                             InvokeUnit(unit);
                      }
                      protected virtual void Dispose(bool disposing) {
                                      return;
                                  // wake sleeping threads
-                                 m_hasTasks.Set();
+                                 if (m_createdThreads > 0)
+                                     SignalThread();
+                                 else
+                                     m_hasTasks.Dispose();
                                  GC.SuppressFinalize(this);
                              }
                          }

Implab/Parallels/WorkerPool.cs +2 -15

                          var len = Interlocked.Increment(ref m_queueLength);
                          m_queue.Enqueue(unit);
-                         ExtendPool();
+                     }
-                     protected override bool ExtendPool() {
-                         if (m_queueLength <= m_threshold*ActiveThreads)
-                             // in this case we are in active thread and it request for additional workers
-                             // satisfy it only when queue is longer than threshold
-                             return false;
-                         return base.ExtendPool();
+                         if (len > m_threshold*ActiveThreads)
+                             GrowPool();
                      }
                      protected override bool TryDequeue(out Action unit) {
                          unit();
                      }
-                     protected override bool Suspend() {
-                         if (m_queueLength == 0)
-                             return base.Suspend();
-                         else
-                             return true; // spin again without locks...
-                     }
                  }
+             }

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages