00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022 #ifndef moses_ThreadPool_h
00023 #define moses_ThreadPool_h
00024
00025 #include <iostream>
00026 #include <queue>
00027 #include <vector>
00028
00029 #include <boost/shared_ptr.hpp>
00030
00031 #ifdef WITH_THREADS
00032 #include <boost/bind.hpp>
00033 #include <boost/thread.hpp>
00034 #endif
00035
00036 #ifdef BOOST_HAS_PTHREADS
00037 #include <pthread.h>
00038 #endif
00039
00040
00041
00042
00043
00047 namespace Moses
00048 {
00049
00052 class Task
00053 {
00054 public:
00055 virtual void Run() = 0;
00056 virtual ~Task() {}
00057 };
00058
00059 #ifdef WITH_THREADS
00060
00061 class ThreadPool
00062 {
00063 public:
00067 explicit ThreadPool(size_t numThreads);
00068
00069 ~ThreadPool() {
00070 Stop();
00071 }
00072
00076 void Submit(boost::shared_ptr<Task> task);
00077
00082 void Stop(bool processRemainingJobs = false);
00083
00087 void SetQueueLimit( size_t limit ) {
00088 m_queueLimit = limit;
00089 }
00090
00091 private:
00095 void Execute();
00096
00097 std::queue<boost::shared_ptr<Task> > m_tasks;
00098 boost::thread_group m_threads;
00099 boost::mutex m_mutex;
00100 boost::condition_variable m_threadNeeded;
00101 boost::condition_variable m_threadAvailable;
00102 bool m_stopped;
00103 bool m_stopping;
00104 size_t m_queueLimit;
00105 };
00106
00107 class TestTask : public Task
00108 {
00109 public:
00110 TestTask(int id) : m_id(id) {}
00111
00112 virtual void Run() {
00113 #ifdef BOOST_HAS_PTHREADS
00114 pthread_t tid = pthread_self();
00115 #else
00116 typedef void * pthread_t;
00117 pthread_t tid = 0;
00118 #endif
00119 std::cerr << "Executing " << m_id << " in thread id " << tid << std::endl;
00120 }
00121
00122 virtual ~TestTask() {}
00123
00124 private:
00125 int m_id;
00126 };
00127
00128 #endif //WITH_THREADS
00129
00130 }
00131 #endif // moses_ThreadPool_h