Add thread affinity support to taskmanager

Foundation/include/Poco/TaskManager.h

@@ -26,6 +26,7 @@
 #include "Poco/AutoPtr.h"
 #include "Poco/NotificationCenter.h"
 #include "Poco/Timestamp.h"
+#include "Poco/ThreadPool.h"
 #include <list>
 
 
@@ -33,7 +34,6 @@ namespace Poco {
 
 
 class Notification;
-class ThreadPool;
 class Exception;
 
 
@@ -52,7 +52,7 @@ public:
 	typedef AutoPtr<Task>      TaskPtr;
 	typedef std::list<TaskPtr> TaskList;
 
-	TaskManager();
+	TaskManager(ThreadPool::ThreadAffinityPolicy affinityPolicy = ThreadPool::OS_DEFAULT);
 		/// Creates the TaskManager, using the
 		/// default ThreadPool.
 
@@ -63,10 +63,10 @@ public:
 	~TaskManager();
 		/// Destroys the TaskManager.
 
-	void start(Task* pTask);
+	void start(Task* pTask, int cpu = -1);
 		/// Starts the given task in a thread obtained
-		/// from the thread pool.
+		/// from the thread pool,
-		///
+		/// on specified cpu.
 		/// The TaskManager takes ownership of the Task object
 		/// and deletes it when it it finished.
 

Foundation/include/Poco/ThreadPool.h

@@ -100,6 +100,12 @@ public:
 	int getStackSize() const;
 		/// Returns the stack size used to create new threads.
 
+	void setAffinityPolicy(ThreadAffinityPolicy affinityPolicy);
+		/// Sets the thread affinity policy for newly created threads
+
+	ThreadAffinityPolicy getAffinityPolicy();
+		/// Returns the thread affinity policy used to create new thread
+
 	int used() const;
 		/// Returns the number of currently used threads.
 
@@ -165,7 +171,7 @@ public:
 		/// or an empty string if no name has been
 		/// specified in the constructor.
 
-	static ThreadPool& defaultPool();
+	static ThreadPool& defaultPool(ThreadAffinityPolicy affinityPolicy = OS_DEFAULT);
 		/// Returns a reference to the default
 		/// thread pool.
 
@@ -209,6 +215,15 @@ inline int ThreadPool::getStackSize() const
 	return _stackSize;
 }
 
+inline void ThreadPool::setAffinityPolicy(ThreadPool::ThreadAffinityPolicy affinityPolicy)
+{
+	_affinityPolicy = affinityPolicy;
+}
+
+inline ThreadPool::ThreadAffinityPolicy ThreadPool::getAffinityPolicy()
+{
+	return _affinityPolicy;
+}
 
 inline const std::string& ThreadPool::name() const
 {

Foundation/src/TaskManager.cpp

@@ -16,7 +16,6 @@
 
 #include "Poco/TaskManager.h"
 #include "Poco/TaskNotification.h"
-#include "Poco/ThreadPool.h"
 
 
 namespace Poco {
@@ -25,8 +24,8 @@ namespace Poco {
 const int TaskManager::MIN_PROGRESS_NOTIFICATION_INTERVAL = 100000; // 100 milliseconds
 
 
-TaskManager::TaskManager():
+TaskManager::TaskManager(ThreadPool::ThreadAffinityPolicy affinityPolicy):
-	_threadPool(ThreadPool::defaultPool())
+	_threadPool(ThreadPool::defaultPool(affinityPolicy))
 {
 }
 
@@ -42,7 +41,7 @@ TaskManager::~TaskManager()
 }
 
 
-void TaskManager::start(Task* pTask)
+void TaskManager::start(Task* pTask, int cpu)
 {
 	TaskPtr pAutoTask(pTask); // take ownership immediately
 	FastMutex::ScopedLock lock(_mutex);
@@ -52,7 +51,7 @@ void TaskManager::start(Task* pTask)
 	_taskList.push_back(pAutoTask);
 	try
 	{
-		_threadPool.start(*pAutoTask, pAutoTask->name());
+		_threadPool.start(*pAutoTask, pAutoTask->name(), cpu);
 	}
 	catch (...)
 	{

Foundation/src/ThreadPool.cpp

@@ -550,13 +550,14 @@ public:
 	{
 		delete _pPool;
 	}
-	ThreadPool* pool()
+	ThreadPool* pool(ThreadPool::ThreadAffinityPolicy affinityPolicy = ThreadPool::OS_DEFAULT)
 	{
 		FastMutex::ScopedLock lock(_mutex);
 		
 		if (!_pPool)
 		{
 			_pPool = new ThreadPool("default");
+			_pPool->setAffinityPolicy(affinityPolicy);
 			if (POCO_THREAD_STACK_SIZE > 0)
 				_pPool->setStackSize(POCO_THREAD_STACK_SIZE);
 		}
@@ -575,9 +576,9 @@ namespace
 }
 
 
-ThreadPool& ThreadPool::defaultPool()
+ThreadPool& ThreadPool::defaultPool(ThreadAffinityPolicy affinityPolicy)
 {
-	return *sh.pool();
+	return *sh.pool(affinityPolicy);
 }
 
 

Foundation/testsuite/src/TaskManagerTest.cpp

@@ -246,7 +246,7 @@ TaskManagerTest::~TaskManagerTest()
 
 void TaskManagerTest::testFinish()
 {
-	TaskManager tm;
+	TaskManager tm(ThreadPool::UNIFORM_DISTRIBUTION);
 	TaskObserver to;
 	tm.addObserver(Observer<TaskObserver, TaskStartedNotification>(to, &TaskObserver::taskStarted));
 	tm.addObserver(Observer<TaskObserver, TaskCancelledNotification>(to, &TaskObserver::taskCancelled));
@@ -281,7 +281,7 @@ void TaskManagerTest::testFinish()
 
 void TaskManagerTest::testCancel()
 {
-	TaskManager tm;
+	TaskManager tm(ThreadPool::UNIFORM_DISTRIBUTION);
 	TaskObserver to;
 	tm.addObserver(Observer<TaskObserver, TaskStartedNotification>(to, &TaskObserver::taskStarted));
 	tm.addObserver(Observer<TaskObserver, TaskCancelledNotification>(to, &TaskObserver::taskCancelled));
@@ -315,7 +315,7 @@ void TaskManagerTest::testCancel()
 
 void TaskManagerTest::testError()
 {
-	TaskManager tm;
+	TaskManager tm(ThreadPool::UNIFORM_DISTRIBUTION);
 	TaskObserver to;
 	tm.addObserver(Observer<TaskObserver, TaskStartedNotification>(to, &TaskObserver::taskStarted));
 	tm.addObserver(Observer<TaskObserver, TaskCancelledNotification>(to, &TaskObserver::taskCancelled));
@@ -348,7 +348,7 @@ void TaskManagerTest::testError()
 
 void TaskManagerTest::testCustom()
 {
-	TaskManager tm;
+	TaskManager tm(ThreadPool::UNIFORM_DISTRIBUTION);
 	
 	CustomTaskObserver<int> ti(0);
 	tm.addObserver(
@@ -431,7 +431,7 @@ void TaskManagerTest::testCustom()
 
 void TaskManagerTest::testMultiTasks()
 {
-	TaskManager tm;
+	TaskManager tm(ThreadPool::UNIFORM_DISTRIBUTION);
 	tm.start(new SimpleTask);
 	tm.start(new SimpleTask);
 	tm.start(new SimpleTask);
@@ -447,7 +447,7 @@ void TaskManagerTest::testMultiTasks()
 
 void TaskManagerTest::testCustomThreadPool()
 {
-	ThreadPool  tp(2, 5, 120);
+	ThreadPool  tp(2, 5, 120, POCO_THREAD_STACK_SIZE, ThreadPool::UNIFORM_DISTRIBUTION);
 	TaskManager tm(tp);
 
 	// fill up the thread pool