implab/ImplabNet Files · Implab/Parallels/WorkerPool.cs

removed the reference to the parent from the promise object this allows...

removed the reference to the parent from the promise object this allows resolved promises to release parents and results they are holding. Added complete set of operations to IPromiseBase interface Subscribing to the cancellation event of the promise should not affect it's IsExclusive property More tests.

cin - - Load All Authors

File last commit:

r21:6a56df4ec59e promises


                r33:b255e4aeef17

default

Download file

             WorkerPool.cs
        
                    77 lines
            
             | 2.1 KiB
            
                | text/x-csharp
            
             |
                CSharpLexer
            
             / Implab / Parallels / WorkerPool.cs
          
                    History
                
                 |
                  Annotation
                 | Raw
                 |Copy content
                 |Copy permalink

      using System;

      using System.Collections.Generic;

      using System.Linq;

      using System.Text;

      using System.Threading;

      using System.Diagnostics;

      namespace Implab.Parallels {

          public class WorkerPool : DispatchPool<Action> {

              MTQueue<Action> m_queue = new MTQueue<Action>();

              int m_queueLength = 0;

              readonly int m_threshold = 1;

              public WorkerPool(int minThreads, int maxThreads, int threshold)

                  : base(minThreads, maxThreads) {

                  m_threshold = threshold;

                  InitPool();

              }

              public WorkerPool(int minThreads, int maxThreads) :

                  base(minThreads, maxThreads) {

                  InitPool();

              }

              public WorkerPool(int threads)

                  : base(threads) {

                  InitPool();

              }

              public WorkerPool()

                  : base() {

                  InitPool();

              }

              public Promise<T> Invoke<T>(Func<T> task) {

                  if (task == null)

                      throw new ArgumentNullException("task");

                  if (IsDisposed)

                      throw new ObjectDisposedException(ToString());

                  var promise = new Promise<T>();

                  EnqueueTask(delegate() {

                      try {

                          promise.Resolve(task());

                      } catch (Exception e) {

                          promise.Reject(e);

                      }

                  });

                  return promise;

              }

              protected void EnqueueTask(Action unit) {

                  Debug.Assert(unit != null);

                  var len = Interlocked.Increment(ref m_queueLength);

                  m_queue.Enqueue(unit);

                  if (len > m_threshold*ActiveThreads)

                      GrowPool();

              }

              protected override bool TryDequeue(out Action unit) {

                  if (m_queue.TryDequeue(out unit)) {

                      Interlocked.Decrement(ref m_queueLength);

                      return true;

                  }

                  return false;

              }

              protected override void InvokeUnit(Action unit) {

                  unit();

              }

          }

      }

	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

				using System;
				using System.Collections.Generic;
				using System.Linq;
				using System.Text;
				using System.Threading;
				using System.Diagnostics;

				namespace Implab.Parallels {
				public class WorkerPool : DispatchPool<Action> {

				MTQueue<Action> m_queue = new MTQueue<Action>();
				int m_queueLength = 0;
				readonly int m_threshold = 1;

				public WorkerPool(int minThreads, int maxThreads, int threshold)
				: base(minThreads, maxThreads) {
				m_threshold = threshold;
				InitPool();
				}

				public WorkerPool(int minThreads, int maxThreads) :
				base(minThreads, maxThreads) {
				InitPool();
				}

				public WorkerPool(int threads)
				: base(threads) {
				InitPool();
				}

				public WorkerPool()
				: base() {
				InitPool();
				}

				public Promise<T> Invoke<T>(Func<T> task) {
				if (task == null)
				throw new ArgumentNullException("task");
				if (IsDisposed)
				throw new ObjectDisposedException(ToString());

				var promise = new Promise<T>();

				EnqueueTask(delegate() {
				try {
				promise.Resolve(task());
				} catch (Exception e) {
				promise.Reject(e);
				}
				});

				return promise;
				}

				protected void EnqueueTask(Action unit) {
				Debug.Assert(unit != null);
				var len = Interlocked.Increment(ref m_queueLength);
				m_queue.Enqueue(unit);

				if (len > m_threshold*ActiveThreads)
				GrowPool();
				}

				protected override bool TryDequeue(out Action unit) {
				if (m_queue.TryDequeue(out unit)) {
				Interlocked.Decrement(ref m_queueLength);
				return true;
				}
				return false;
				}

				protected override void InvokeUnit(Action unit) {
				unit();
				}

				}
				}