implab/ImplabNet Commit - r20:1c3b3d518480

refactoring, sync

cin -

r20:1c3b3d518480 promises

parent child

Context file:

r20:1c3b3d518480

Collapse all files

Implab.Test/AsyncTests.cs +7 -7

                      public void WorkerPoolSizeTest() {
                          var pool = new WorkerPool(5, 10, 0);
-                         Assert.AreEqual(5, pool.ThreadCount);
+                         Assert.AreEqual(5, pool.PoolSize);
                          pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
                          pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
                          pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
-                         Assert.AreEqual(5, pool.ThreadCount);
+                         Assert.AreEqual(5, pool.PoolSize);
                          for (int i = 0; i < 100; i++)
                              pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
                          Thread.Sleep(100);
-                         Assert.AreEqual(10, pool.ThreadCount);
+                         Assert.AreEqual(10, pool.PoolSize);
                          pool.Dispose();
                      }
                      [TestMethod]
                      public void WorkerPoolDisposeTest() {
                          var pool = new WorkerPool(5, 20);
-                         Assert.AreEqual(5, pool.ThreadCount);
+                         Assert.AreEqual(5, pool.PoolSize);
                          pool.Dispose();
-                         Thread.Sleep(100);
-                         Assert.AreEqual(0, pool.ThreadCount);
+                         Thread.Sleep(200);
+                         Assert.AreEqual(0, pool.PoolSize);
                          pool.Dispose();
                      }
                      [TestMethod]
                      public void ChainedMapTest() {
-                         using (var pool = new WorkerPool(4,4,0)) {
+                         using (var pool = new WorkerPool(0,100,0)) {
                              int count = 10000;
                              double[] args = new double[count];

Implab.v11.suo binary modified 0 0

NO CONTENT: modified file, binary diff hidden

Implab/Parallels/DispatchPool.cs +74 -33

                  public abstract class DispatchPool<TUnit> : IDisposable {
                      readonly int m_minThreads;
                      readonly int m_maxThreads;
-                     int m_runningThreads = 0;
+                     int m_createdThreads = 0;
+                     int m_activeThreads = 0;
+                     int m_sleepingThreads = 0;
                      int m_maxRunningThreads = 0;
-                     int m_suspended = 0;
                      int m_exitRequired = 0;
+                     int m_releaseTimeout = 100; // timeout while the working thread will wait for the new tasks before exit
                      AutoResetEvent m_hasTasks = new AutoResetEvent(false);
                      protected DispatchPool(int min, int max) {
                              StartWorker();
                      }
-                     public int ThreadCount {
+                     public int PoolSize {
                          get {
-                             return m_runningThreads;
+                             return m_createdThreads;
+                         }
+                     }
+                     public int ActiveThreads {
+                         get {
+                             return m_activeThreads;
                          }
                      }
                      protected abstract bool TryDequeue(out TUnit unit);
                      protected virtual bool ExtendPool() {
-                         if (m_suspended > 0) {
-                             m_hasTasks.Set();
+                         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;
-                         } else
-                             return StartWorker();
+                         }
                      }
                      /// <summary>
                          m_hasTasks.Set(); // wake sleeping thread;
                          if (AllocateThreadSlot(1)) {
+                             // if there were no threads in the pool
                              var worker = new Thread(this.Worker);
                              worker.IsBackground = true;
                              worker.Start();
                          }
                      }
-                     protected virtual void Suspend() {
-                         m_hasTasks.WaitOne();
+                     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() {
+                         //no tasks left, exit if the thread is no longer needed
+                         bool last;
+                         bool requestExit;
+                         if (m_releaseTimeout > 0)
+                             requestExit = !Sleep(m_releaseTimeout);
+                         else
+                             requestExit = true;
+                         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 (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();
+                             }
+                             return false;
+                         }
+                         Sleep(-1);
+                         return true;
                      }
                      #region thread slots traits
                          int current;
                          // use spins to allocate slot for the new thread
                          do {
-                             current = m_runningThreads;
+                             current = m_createdThreads;
                              if (current >= m_maxThreads || m_exitRequired != 0)
                                  // no more slots left or the pool has been disposed
                                  return false;
-                         } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
+                         } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current));
                          UpdateMaxThreads(current + 1);
                      }
                      bool AllocateThreadSlot(int desired) {
-                         if (desired - 1 != Interlocked.CompareExchange(ref m_runningThreads, desired, desired - 1))
+                         if (desired - 1 != Interlocked.CompareExchange(ref m_createdThreads, desired, desired - 1))
                              return false;
                          UpdateMaxThreads(desired);
                          int current;
                          // use spins to release slot for the new thread
                          do {
-                             current = m_runningThreads;
+                             current = m_createdThreads;
                              if (current <= m_minThreads && m_exitRequired == 0)
                                  // the thread is reserved
                                  return false;
-                         } while (current != Interlocked.CompareExchange(ref m_runningThreads, current - 1, current));
+                         } while (current != Interlocked.CompareExchange(ref m_createdThreads, current - 1, current));
                          last = (current == 1);
                      /// </summary>
                      /// <returns>true - no more threads left</returns>
                      bool ReleaseThreadSlotAnyway() {
-                         var left = Interlocked.Decrement(ref m_runningThreads);
+                         var left = Interlocked.Decrement(ref m_createdThreads);
                          return left == 0;
                      }
                              // exit if requested
                              if (m_exitRequired != 0) {
                                  // release the thread slot
+                                 Interlocked.Decrement(ref m_activeThreads);
                                  if (ReleaseThreadSlotAnyway()) // it was the last worker
                                      m_hasTasks.Dispose();
                                  else
                                  return true;
                              }
-                             //no tasks left, exit if the thread is no longer needed
-                             bool last;
-                             if (ReleaseThreadSlot(out last)) {
-                                 if (last && m_hasTasks.WaitOne(0)) {
-                                     if (AllocateThreadSlot(1))
-                                         continue; // spin again...
-                                     else
-                                         // we failed to reallocate slot for this thread
-                                         // therefore we need to release the event
-                                         m_hasTasks.Set();
+                                 }
-                                 return false;
+                             }
+                             Interlocked.Decrement(ref m_activeThreads);
                              // entering suspend state
-                             Interlocked.Increment(ref m_suspended);
                              // keep this thread and wait
-                             Suspend();
-                             Interlocked.Decrement(ref m_suspended);
+                             if (!Suspend())
+                                 return false;
+                             Interlocked.Increment(ref m_activeThreads);
                          } while (true);
                      }
                      void Worker() {
                          TUnit unit;
+                         Interlocked.Increment(ref m_activeThreads);
                          while (FetchTask(out unit))
                              InvokeUnit(unit);
                      }

Implab/Parallels/WorkerPool.cs +6 -5

                          var len = Interlocked.Increment(ref m_queueLength);
                          m_queue.Enqueue(unit);
-                         if(!ExtendPool())
-                             WakePool();
+                         ExtendPool();
                      }
                      protected override bool ExtendPool() {
-                         if (m_queueLength <= m_threshold*ThreadCount)
+                         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;
                          unit();
                      }
-                     protected override void Suspend() {
+                     protected override bool Suspend() {
                          if (m_queueLength == 0)
-                             base.Suspend();
+                             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