History

put CWelsThreadPool to singleTon for future usage (including add sink for IWelsTask)

sijchen committed 10 years ago

Commit d4f09d91 parent 52d25f510 files · +273 −85Browse files

Showing the changes to codec/common/src/WelsThreadPool.cpp (1 of 10 files). Show all

Unified · Split

codec/common/src/WelsThreadPool.cpp+112−44
@@ -43,70 +43,119 @@ · 42 unmodified lines
43 43
44namespace WelsCommon {44namespace WelsCommon {
45 45
46int32_t CWelsThreadPool::m_iRefCount = 0;
47CWelsLock CWelsThreadPool::m_cInitLock;
48int32_t CWelsThreadPool::m_iMaxThreadNum = DEFAULT_THREAD_NUM;
46 49
47CWelsThreadPool::CWelsThreadPool (IWelsThreadPoolSink* pSink, int32_t iMaxThreadNum) :50CWelsThreadPool::CWelsThreadPool (IWelsThreadPoolSink* pSink) :
48 m_pSink (pSink) {51 m_cWaitedTasks (NULL), m_cIdleThreads (NULL), m_cBusyThreads (NULL), m_pSink (pSink) {
49 m_cWaitedTasks = new CWelsCircleQueue<IWelsTask>();52}
50 m_cIdleThreads = new CWelsCircleQueue<CWelsTaskThread>();
51 m_cBusyThreads = new CWelsList<CWelsTaskThread>();
52 m_iMaxThreadNum = 0;
53 53
54 if (NULL == m_cWaitedTasks || NULL == m_cIdleThreads || NULL == m_cBusyThreads) {
55 WELS_DELETE_OP(m_cWaitedTasks);
56 WELS_DELETE_OP(m_cIdleThreads);
57 WELS_DELETE_OP(m_cBusyThreads);
58 54
59 return;55CWelsThreadPool::~CWelsThreadPool() {
56 //fprintf(stdout, "CWelsThreadPool::~CWelsThreadPool: delete %x, %x, %x\n", m_cWaitedTasks, m_cIdleThreads, m_cBusyThreads);
57 if (0 != m_iRefCount) {
58 m_iRefCount = 0;
59 Uninit();
60 }
61}
62
63WELS_THREAD_ERROR_CODE CWelsThreadPool::SetThreadNum (int32_t iMaxThreadNum) {
64 CWelsAutoLock cLock (m_cInitLock);
65
66 if (m_iRefCount != 0) {
67 return WELS_THREAD_ERROR_GENERAL;
60 }68 }
61 69
62 if (WELS_THREAD_ERROR_OK != Init (iMaxThreadNum)) {70 if (iMaxThreadNum <= 0) {
63 Uninit();71 iMaxThreadNum = 1;
72 }
73 m_iMaxThreadNum = iMaxThreadNum;
74 return WELS_THREAD_ERROR_OK;
75}
64 76
65 WELS_DELETE_OP(m_cWaitedTasks);77CWelsThreadPool& CWelsThreadPool::AddReference (IWelsThreadPoolSink* pSink) {
66 WELS_DELETE_OP(m_cIdleThreads);78 CWelsAutoLock cLock (m_cInitLock);
67 WELS_DELETE_OP(m_cBusyThreads);79 static CWelsThreadPool m_cThreadPoolSelf (pSink);
80 if (m_iRefCount == 0) {
81 //TODO: will remove this afterwards
82 if (WELS_THREAD_ERROR_OK != m_cThreadPoolSelf.Init(pSink)) {
83 m_cThreadPoolSelf.Uninit();
84 }
85 m_cThreadPoolSelf.UpdateSink (pSink);
68 }86 }
87
88 //fprintf(stdout, "m_iRefCount=%d, pSink=%x, iMaxThreadNum=%d\n", m_iRefCount, pSink, iMaxThreadNum);
89
90 ++ m_iRefCount;
91 //fprintf(stdout, "m_iRefCount2=%d\n", m_iRefCount);
92 return m_cThreadPoolSelf;
69}93}
70 94
95void CWelsThreadPool::RemoveInstance() {
96 CWelsAutoLock cLock (m_cInitLock);
97 //fprintf(stdout, "m_iRefCount=%d\n", m_iRefCount);
98 -- m_iRefCount;
99 if (0 == m_iRefCount) {
100 StopAllRunning();
101 m_pSink = NULL;
102 Uninit();
103 //fprintf(stdout, "m_iRefCount=%d, IdleThreadNum=%d, BusyThreadNum=%d, WaitedTask=%d\n", m_iRefCount, GetIdleThreadNum(), GetBusyThreadNum(), GetWaitedTaskNum());
104 }
105}
71 106
72CWelsThreadPool::~CWelsThreadPool() {107bool CWelsThreadPool::IsReferenced() {
73 Uninit();108 CWelsAutoLock cLock (m_cInitLock);
109 return (m_iRefCount>0);
110}
74 111
75 WELS_DELETE_OP(m_cWaitedTasks);112void CWelsThreadPool::UpdateSink (IWelsThreadPoolSink* pSink) {
76 WELS_DELETE_OP(m_cIdleThreads);113 m_pSink = pSink;
77 WELS_DELETE_OP(m_cBusyThreads);114 //fprintf(stdout, "UpdateSink: m_pSink=%x\n", m_pSink);
115 //fprintf(stdout, "m_iRefCount=%d, IdleThreadNum=%d, BusyThreadNum=%d, WaitedTask=%d\n", m_iRefCount, GetIdleThreadNum(), GetBusyThreadNum(), GetWaitedTaskNum());
78}116}
79 117
118
80WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStart (CWelsTaskThread* pThread, IWelsTask* pTask) {119WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStart (CWelsTaskThread* pThread, IWelsTask* pTask) {
81 AddThreadToBusyList (pThread);120 AddThreadToBusyList (pThread);
82 121 //fprintf(stdout, "CWelsThreadPool::AddThreadToBusyList: Task %x at Thread %x\n", pTask, pThread);
83 return WELS_THREAD_ERROR_OK;122 return WELS_THREAD_ERROR_OK;
84}123}
85 124
86WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStop (CWelsTaskThread* pThread, IWelsTask* pTask) {125WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStop (CWelsTaskThread* pThread, IWelsTask* pTask) {
126 //fprintf(stdout, "CWelsThreadPool::OnTaskStop 0: Task %x at Thread %x Finished\n", pTask, pThread);
127
87 RemoveThreadFromBusyList (pThread);128 RemoveThreadFromBusyList (pThread);
88 AddThreadToIdleQueue (pThread);129 AddThreadToIdleQueue (pThread);
130 //fprintf(stdout, "CWelsThreadPool::OnTaskStop 1: Task %x at Thread %x Finished, m_pSink=%x\n", pTask, pThread, m_pSink);
89 131
90 if (m_pSink) {132 if (pTask->GetSink()) {
91 m_pSink->OnTaskExecuted (pTask);133 pTask->GetSink()->OnTaskExecuted();
92 }134 }
93 135 //if (m_pSink) {
94 //WELS_INFO_TRACE("ThreadPool: Task "<<(uint32_t)pTask<<" Finished, Thread "<<(uint32_t)pThread<<" put to idle list");136 // m_pSink->OnTaskExecuted (pTask);
137 //}
138 //fprintf(stdout, "CWelsThreadPool::OnTaskStop 2: Task %x at Thread %x Finished\n", pTask, pThread);
95 139
96 SignalThread();140 SignalThread();
141
142 //fprintf(stdout, "ThreadPool: Task %x at Thread %x Finished\n", pTask, pThread);
97 return WELS_THREAD_ERROR_OK;143 return WELS_THREAD_ERROR_OK;
98}144}
99 145
100WELS_THREAD_ERROR_CODE CWelsThreadPool::Init (int32_t iMaxThreadNum) {146WELS_THREAD_ERROR_CODE CWelsThreadPool::Init (IWelsThreadPoolSink* pSink) {
101 CWelsAutoLock cLock (m_cLockPool);147 //fprintf(stdout, "Enter WelsThreadPool Init\n");
102 //WELS_INFO_TRACE("Enter WelsThreadPool Init");
103 148
104 int32_t i;149 CWelsAutoLock cLock (m_cLockPool);
105 150
106 if (iMaxThreadNum <= 0) iMaxThreadNum = 1;151 m_cWaitedTasks = new CWelsCircleQueue<IWelsTask>();
107 m_iMaxThreadNum = iMaxThreadNum;152 m_cIdleThreads = new CWelsCircleQueue<CWelsTaskThread>();
153 m_cBusyThreads = new CWelsList<CWelsTaskThread>();
154 if (NULL == m_cWaitedTasks || NULL == m_cIdleThreads || NULL == m_cBusyThreads) {
155 return WELS_THREAD_ERROR_GENERAL;
156 }
108 157
109 for (i = 0; i < m_iMaxThreadNum; i++) {158 for (int32_t i = 0; i < m_iMaxThreadNum; i++) {
110 if (WELS_THREAD_ERROR_OK != CreateIdleThread()) {159 if (WELS_THREAD_ERROR_OK != CreateIdleThread()) {
111 return WELS_THREAD_ERROR_GENERAL;160 return WELS_THREAD_ERROR_GENERAL;
112 }161 }
@@ -119,9 +168,8 @@ WELS_THREAD_ERROR_CODE CWelsThreadPool::Init (int32_t iMaxThreadNum) { · 6 unmodified lines
119 return WELS_THREAD_ERROR_OK;168 return WELS_THREAD_ERROR_OK;
120}169}
121 170
122WELS_THREAD_ERROR_CODE CWelsThreadPool::Uninit() {171WELS_THREAD_ERROR_CODE CWelsThreadPool::StopAllRunning() {
123 WELS_THREAD_ERROR_CODE iReturn = WELS_THREAD_ERROR_OK;172 WELS_THREAD_ERROR_CODE iReturn = WELS_THREAD_ERROR_OK;
124 CWelsAutoLock cLock (m_cLockPool);
125 173
126 ClearWaitedTasks();174 ClearWaitedTasks();
127 175
@@ -134,6 +182,18 @@ WELS_THREAD_ERROR_CODE CWelsThreadPool::Uninit() { · 6 unmodified lines
134 iReturn = WELS_THREAD_ERROR_GENERAL;182 iReturn = WELS_THREAD_ERROR_GENERAL;
135 }183 }
136 184
185 return iReturn;
186}
187
188WELS_THREAD_ERROR_CODE CWelsThreadPool::Uninit() {
189 WELS_THREAD_ERROR_CODE iReturn = WELS_THREAD_ERROR_OK;
190 CWelsAutoLock cLock (m_cLockPool);
191
192 iReturn = StopAllRunning();
193 if (WELS_THREAD_ERROR_OK != iReturn) {
194 return iReturn;
195 }
196
137 m_cLockIdleTasks.Lock();197 m_cLockIdleTasks.Lock();
138 while (m_cIdleThreads->size() > 0) {198 while (m_cIdleThreads->size() > 0) {
139 DestroyThread (m_cIdleThreads->begin());199 DestroyThread (m_cIdleThreads->begin());
@@ -141,14 +201,17 @@ WELS_THREAD_ERROR_CODE CWelsThreadPool::Uninit() { · 1 unmodified line
141 }201 }
142 m_cLockIdleTasks.Unlock();202 m_cLockIdleTasks.Unlock();
143 203
144 m_iMaxThreadNum = 0;
145 Kill();204 Kill();
146 205
206 WELS_DELETE_OP(m_cWaitedTasks);
207 WELS_DELETE_OP(m_cIdleThreads);
208 WELS_DELETE_OP(m_cBusyThreads);
209
147 return iReturn;210 return iReturn;
148}211}
149 212
150void CWelsThreadPool::ExecuteTask() {213void CWelsThreadPool::ExecuteTask() {
151 //fprintf(stdout, "ThreadPool: schedule tasks\n");214 //fprintf(stdout, "ThreadPool: scheduled tasks: ExecuteTask\n");
152 CWelsTaskThread* pThread = NULL;215 CWelsTaskThread* pThread = NULL;
153 IWelsTask* pTask = NULL;216 IWelsTask* pTask = NULL;
154 while (GetWaitedTaskNum() > 0) {217 while (GetWaitedTaskNum() > 0) {
@@ -165,20 +228,21 @@ void CWelsThreadPool::ExecuteTask() { · 10 unmodified lines
165WELS_THREAD_ERROR_CODE CWelsThreadPool::QueueTask (IWelsTask* pTask) {228WELS_THREAD_ERROR_CODE CWelsThreadPool::QueueTask (IWelsTask* pTask) {
166 CWelsAutoLock cLock (m_cLockPool);229 CWelsAutoLock cLock (m_cLockPool);
167 230
168 //fprintf(stdout, "ThreadPool: QueueTask = %x\n", pTask);231 //fprintf(stdout, "CWelsThreadPool::QueueTask: %d, pTask=%x\n", m_iRefCount, pTask);
169 if (GetWaitedTaskNum() == 0) {232 if (GetWaitedTaskNum() == 0) {
170 CWelsTaskThread* pThread = GetIdleThread();233 CWelsTaskThread* pThread = GetIdleThread();
171 234
172 if (pThread != NULL) {235 if (pThread != NULL) {
173 //fprintf(stdout, "ThreadPool: ExecuteTask = %x\n", pTask);236 //fprintf(stdout, "ThreadPool: ExecuteTask = %x at thread %x\n", pTask, pThread);
174 pThread->SetTask (pTask);237 pThread->SetTask (pTask);
175 238
176 return WELS_THREAD_ERROR_OK;239 return WELS_THREAD_ERROR_OK;
177 }240 }
178 }241 }
179
180 AddTaskToWaitedList (pTask);
181 //fprintf(stdout, "ThreadPool: AddTaskToWaitedList: %x\n", pTask);242 //fprintf(stdout, "ThreadPool: AddTaskToWaitedList: %x\n", pTask);
243 AddTaskToWaitedList (pTask);
244
245 //fprintf(stdout, "ThreadPool: SignalThread: %x\n", pTask);
182 SignalThread();246 SignalThread();
183 return WELS_THREAD_ERROR_OK;247 return WELS_THREAD_ERROR_OK;
184}248}
@@ -193,6 +257,7 @@ WELS_THREAD_ERROR_CODE CWelsThreadPool::CreateIdleThread() { · 8 unmodified lines
193 if (WELS_THREAD_ERROR_OK != pThread->Start()) {257 if (WELS_THREAD_ERROR_OK != pThread->Start()) {
194 return WELS_THREAD_ERROR_GENERAL;258 return WELS_THREAD_ERROR_GENERAL;
195 }259 }
260 //fprintf(stdout, "ThreadPool: AddThreadToIdleQueue: %x\n", pThread);
196 AddThreadToIdleQueue (pThread);261 AddThreadToIdleQueue (pThread);
197 262
198 return WELS_THREAD_ERROR_OK;263 return WELS_THREAD_ERROR_OK;
@@ -219,10 +284,10 @@ WELS_THREAD_ERROR_CODE CWelsThreadPool::AddThreadToBusyList (CWelsTaskThread* pT · 20 unmodified lines
219 284
220WELS_THREAD_ERROR_CODE CWelsThreadPool::RemoveThreadFromBusyList (CWelsTaskThread* pThread) {285WELS_THREAD_ERROR_CODE CWelsThreadPool::RemoveThreadFromBusyList (CWelsTaskThread* pThread) {
221 CWelsAutoLock cLock (m_cLockBusyTasks);286 CWelsAutoLock cLock (m_cLockBusyTasks);
222 if (m_cBusyThreads->erase(pThread)) {287 if (m_cBusyThreads->erase (pThread)) {
223 return WELS_THREAD_ERROR_OK;288 return WELS_THREAD_ERROR_OK;
224 } else {289 } else {
225 return WELS_THREAD_ERROR_GENERAL;290 return WELS_THREAD_ERROR_GENERAL;
226 }291 }
227}292}
228 293
@@ -230,12 +295,14 @@ void CWelsThreadPool::AddTaskToWaitedList (IWelsTask* pTask) { · 1 unmodified line
230 CWelsAutoLock cLock (m_cLockWaitedTasks);295 CWelsAutoLock cLock (m_cLockWaitedTasks);
231 296
232 m_cWaitedTasks->push_back (pTask);297 m_cWaitedTasks->push_back (pTask);
298 //fprintf(stdout, "CWelsThreadPool::AddTaskToWaitedList=%d, pTask=%x\n", m_cWaitedTasks->size(), pTask);
233 return;299 return;
234}300}
235 301
236CWelsTaskThread* CWelsThreadPool::GetIdleThread() {302CWelsTaskThread* CWelsThreadPool::GetIdleThread() {
237 CWelsAutoLock cLock (m_cLockIdleTasks);303 CWelsAutoLock cLock (m_cLockIdleTasks);
238 304
305 //fprintf(stdout, "CWelsThreadPool::GetIdleThread=%d\n", m_cIdleThreads->size());
239 if (m_cIdleThreads->size() == 0) {306 if (m_cIdleThreads->size() == 0) {
240 return NULL;307 return NULL;
241 }308 }
@@ -254,6 +321,7 @@ int32_t CWelsThreadPool::GetIdleThreadNum() { · 12 unmodified lines
254}321}
255 322
256int32_t CWelsThreadPool::GetWaitedTaskNum() {323int32_t CWelsThreadPool::GetWaitedTaskNum() {
324 //fprintf(stdout, "CWelsThreadPool::m_cWaitedTasks=%d\n", m_cWaitedTasks->size());
257 return m_cWaitedTasks->size();325 return m_cWaitedTasks->size();
258}326}
259 327

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t