History

add basic threadpool functions https://rbcommons.com/s/OpenH264/r/1294/

Sijia Chen committed 11 years ago

Commit 757a5961 parent af6a9a814 files · +1043 −4Browse files

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

Unified · Split

codec/common/src/WelsThreadPool.cpp+305−0
@@ -0,0 +1,305 @@
1+/*!
2+ * \copy
3+ * Copyright (c) 2009-2015, Cisco Systems
4+ * All rights reserved.
5+ *
6+ * Redistribution and use in source and binary forms, with or without
7+ * modification, are permitted provided that the following conditions
8+ * are met:
9+ *
10+ * * Redistributions of source code must retain the above copyright
11+ * notice, this list of conditions and the following disclaimer.
12+ *
13+ * * Redistributions in binary form must reproduce the above copyright
14+ * notice, this list of conditions and the following disclaimer in
15+ * the documentation and/or other materials provided with the
16+ * distribution.
17+ *
18+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
21+ * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
22+ * COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
23+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
24+ * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
25+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
26+ * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
27+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
28+ * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
29+ * POSSIBILITY OF SUCH DAMAGE.
30+ *
31+ *
32+ * \file WelsThreadPool.cpp
33+ *
34+ * \brief functions for Thread Pool
35+ *
36+ * \date 5/09/2012 Created
37+ *
38+ *************************************************************************************
39+ */
40+
41+
42+#include <list>
43+#include <map>
44+
45+#include "typedefs.h"
46+#include "WelsThreadPool.h"
47+
48+namespace WelsCommon {
49+
50+
51+CWelsThreadPool::CWelsThreadPool (IWelsThreadPoolSink* pSink, int32_t iMaxThreadNum) :
52+ m_pSink (pSink) {
53+ m_iMaxThreadNum = 0;
54+
55+ Init (iMaxThreadNum);
56+}
57+
58+
59+CWelsThreadPool::~CWelsThreadPool() {
60+ Uninit();
61+}
62+
63+WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStart (CWelsTaskThread* pThread, IWelsTask* pTask) {
64+ AddThreadToBusyMap (pThread);
65+
66+ return WELS_THREAD_ERROR_OK;
67+}
68+
69+WELS_THREAD_ERROR_CODE CWelsThreadPool::OnTaskStop (CWelsTaskThread* pThread, IWelsTask* pTask) {
70+ RemoveThreadFromBusyMap (pThread);
71+ AddThreadToIdleMap (pThread);
72+
73+ if (m_pSink) {
74+ m_pSink->OnTaskExecuted (pTask);
75+ }
76+
77+ //WELS_INFO_TRACE("ThreadPool: Task "<<(uint32_t)pTask<<" Finished, Thread "<<(uint32_t)pThread<<" put to idle list");
78+
79+ SignalThread();
80+ return WELS_THREAD_ERROR_OK;
81+}
82+
83+WELS_THREAD_ERROR_CODE CWelsThreadPool::Init (int32_t iMaxThreadNum) {
84+ CWelsAutoLock cLock (m_cLockPool);
85+ //WELS_INFO_TRACE("Enter WelsThreadPool Init");
86+
87+ int32_t i;
88+
89+ if (iMaxThreadNum <= 0) iMaxThreadNum = 1;
90+ m_iMaxThreadNum = iMaxThreadNum;
91+
92+ for (i = 0; i < m_iMaxThreadNum; i++) {
93+ if (WELS_THREAD_ERROR_OK != CreateIdleThread()) {
94+ return WELS_THREAD_ERROR_GENERAL;
95+ }
96+ }
97+
98+ if (WELS_THREAD_ERROR_OK != Start()) {
99+ return WELS_THREAD_ERROR_GENERAL;
100+ }
101+
102+ return WELS_THREAD_ERROR_OK;
103+}
104+
105+WELS_THREAD_ERROR_CODE CWelsThreadPool::Uninit() {
106+ WELS_THREAD_ERROR_CODE iReturn = WELS_THREAD_ERROR_OK;
107+ CWelsAutoLock cLock (m_cLockPool);
108+
109+ ClearWaitedTasks();
110+
111+ while (GetBusyThreadNum() > 0) {
112+ //WELS_INFO_TRACE ("CWelsThreadPool::Uninit - Waiting all thread to exit");
113+ WelsSleep (10);
114+ }
115+
116+ if (GetIdleThreadNum() != m_iMaxThreadNum) {
117+ iReturn = WELS_THREAD_ERROR_GENERAL;
118+ }
119+
120+ m_cLockIdleTasks.Lock();
121+ std::map<uintptr_t, CWelsTaskThread*>::iterator iter = m_cIdleThreads.begin();
122+ while (iter != m_cIdleThreads.end()) {
123+ DestroyThread (iter->second);
124+ ++ iter;
125+ }
126+ m_cLockIdleTasks.Unlock();
127+
128+ m_iMaxThreadNum = 0;
129+ Kill();
130+
131+ return iReturn;
132+}
133+
134+void CWelsThreadPool::ExecuteTask() {
135+ //WELS_INFO_TRACE("ThreadPool: schedule tasks");
136+ CWelsTaskThread* pThread = NULL;
137+ IWelsTask* pTask = NULL;
138+ while (GetWaitedTaskNum() > 0) {
139+ pThread = GetIdleThread();
140+ if (pThread == NULL) {
141+ break;
142+ }
143+ pTask = GetWaitedTask();
144+ //WELS_INFO_TRACE("ThreadPool: ExecuteTask = "<<(uint32_t)(pTask)<<" at thread = "<<(uint32_t)(pThread));
145+ pThread->SetTask (pTask);
146+ }
147+}
148+
149+WELS_THREAD_ERROR_CODE CWelsThreadPool::QueueTask (IWelsTask* pTask) {
150+ CWelsAutoLock cLock (m_cLockPool);
151+
152+ //WELS_INFO_TRACE("ThreadPool: QueueTask = "<<(uint32_t)(pTask));
153+ if (GetWaitedTaskNum() == 0) {
154+ CWelsTaskThread* pThread = GetIdleThread();
155+
156+ if (pThread != NULL) {
157+ //WELS_INFO_TRACE("ThreadPool: ExecuteTask = "<<(uint32_t)(pTask));
158+ pThread->SetTask (pTask);
159+
160+ return WELS_THREAD_ERROR_OK;
161+ }
162+ }
163+
164+ AddTaskToWaitedList (pTask);
165+
166+ SignalThread();
167+
168+ return WELS_THREAD_ERROR_OK;
169+}
170+
171+WELS_THREAD_ERROR_CODE CWelsThreadPool::CreateIdleThread() {
172+ CWelsTaskThread* pThread = new CWelsTaskThread (this);
173+
174+ if (NULL == pThread) {
175+ return WELS_THREAD_ERROR_GENERAL;
176+ }
177+
178+ pThread->Start();
179+ AddThreadToIdleMap (pThread);
180+
181+ return WELS_THREAD_ERROR_OK;
182+}
183+
184+void CWelsThreadPool::DestroyThread (CWelsTaskThread* pThread) {
185+ pThread->Kill();
186+ delete pThread;
187+
188+ return;
189+}
190+
191+WELS_THREAD_ERROR_CODE CWelsThreadPool::AddThreadToIdleMap (CWelsTaskThread* pThread) {
192+ CWelsAutoLock cLock (m_cLockIdleTasks);
193+
194+ uintptr_t id = pThread->GetID();
195+
196+ std::map<uintptr_t, CWelsTaskThread*>::iterator iter = m_cIdleThreads.find (id);
197+
198+ if (iter != m_cIdleThreads.end()) {
199+ return WELS_THREAD_ERROR_GENERAL;
200+ }
201+
202+ m_cIdleThreads[id] = pThread;
203+
204+ return WELS_THREAD_ERROR_OK;
205+}
206+
207+WELS_THREAD_ERROR_CODE CWelsThreadPool::AddThreadToBusyMap (CWelsTaskThread* pThread) {
208+ CWelsAutoLock cLock (m_cLockBusyTasks);
209+
210+ uintptr_t id = pThread->GetID();
211+
212+ std::map<uintptr_t, CWelsTaskThread*>::iterator iter = m_cBusyThreads.find (id);
213+
214+ if (iter != m_cBusyThreads.end()) {
215+ return WELS_THREAD_ERROR_GENERAL;
216+ }
217+
218+ m_cBusyThreads[id] = pThread;
219+
220+ return WELS_THREAD_ERROR_OK;
221+}
222+
223+WELS_THREAD_ERROR_CODE CWelsThreadPool::RemoveThreadFromBusyMap (CWelsTaskThread* pThread) {
224+ CWelsAutoLock cLock (m_cLockBusyTasks);
225+
226+ uintptr_t id = pThread->GetID();
227+
228+ std::map<uintptr_t, CWelsTaskThread*>::iterator iter = m_cBusyThreads.find (id);
229+
230+ if (iter != m_cBusyThreads.end()) {
231+ m_cBusyThreads.erase (iter);
232+ } else {
233+ return WELS_THREAD_ERROR_GENERAL;
234+ }
235+
236+ return WELS_THREAD_ERROR_OK;
237+}
238+
239+void CWelsThreadPool::AddTaskToWaitedList (IWelsTask* pTask) {
240+ CWelsAutoLock cLock (m_cLockWaitedTasks);
241+
242+ m_cWaitedTasks.push_back (pTask);
243+ return;
244+}
245+
246+CWelsTaskThread* CWelsThreadPool::GetIdleThread() {
247+ CWelsAutoLock cLock (m_cLockIdleTasks);
248+
249+ if (m_cIdleThreads.size() == 0) {
250+ return NULL;
251+ }
252+
253+ std::map<uintptr_t, CWelsTaskThread*>::iterator it = m_cIdleThreads.begin();
254+ CWelsTaskThread* pThread = it->second;
255+
256+ m_cIdleThreads.erase (it);
257+
258+ return pThread;
259+}
260+
261+int32_t CWelsThreadPool::GetBusyThreadNum() {
262+ return static_cast<int32_t> (m_cBusyThreads.size());
263+}
264+
265+int32_t CWelsThreadPool::GetIdleThreadNum() {
266+ return static_cast<int32_t> (m_cIdleThreads.size());
267+}
268+
269+int32_t CWelsThreadPool::GetWaitedTaskNum() {
270+ return static_cast<int32_t> (m_cWaitedTasks.size());
271+}
272+
273+IWelsTask* CWelsThreadPool::GetWaitedTask() {
274+ CWelsAutoLock lock (m_cLockWaitedTasks);
275+
276+ if (m_cWaitedTasks.size() == 0) {
277+ return NULL;
278+ }
279+
280+ std::list<IWelsTask*>::iterator it = m_cWaitedTasks.begin();
281+ IWelsTask* pTask = *it;
282+
283+ m_cWaitedTasks.pop_front();
284+
285+ return pTask;
286+}
287+
288+void CWelsThreadPool::ClearWaitedTasks() {
289+ CWelsAutoLock cLock (m_cLockWaitedTasks);
290+
291+ std::list<IWelsTask*>::iterator iter = m_cWaitedTasks.begin();
292+
293+ if (m_pSink) {
294+ while (iter != m_cWaitedTasks.end()) {
295+ m_pSink->OnTaskCancelled (*iter);
296+ ++ iter;
297+ }
298+ }
299+
300+ m_cWaitedTasks.clear();
301+}
302+
303+}
304+
305+

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