1 /**
2 * WinPR: Windows Portable Runtime
3 * Thread Pool API (Pool)
4 *
5 * Copyright 2012 Marc-Andre Moreau <marcandre.moreau@gmail.com>
6 *
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
10 *
11 * http://www.apache.org/licenses/LICENSE-2.0
12 *
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 */
19
20 #ifdef HAVE_CONFIG_H
21 #include "config.h"
22 #endif
23
24 #include <winpr/crt.h>
25 #include <winpr/pool.h>
26 #include <winpr/library.h>
27
28 #include "pool.h"
29
30 #ifdef WINPR_THREAD_POOL
31
32 #ifdef _WIN32
33 static INIT_ONCE init_once_module = INIT_ONCE_STATIC_INIT;
34 static PTP_POOL(WINAPI* pCreateThreadpool)(PVOID reserved);
35 static VOID(WINAPI* pCloseThreadpool)(PTP_POOL ptpp);
36 static BOOL(WINAPI* pSetThreadpoolThreadMinimum)(PTP_POOL ptpp, DWORD cthrdMic);
37 static VOID(WINAPI* pSetThreadpoolThreadMaximum)(PTP_POOL ptpp, DWORD cthrdMost);
38
init_module(PINIT_ONCE once,PVOID param,PVOID * context)39 static BOOL CALLBACK init_module(PINIT_ONCE once, PVOID param, PVOID* context)
40 {
41 HMODULE kernel32 = LoadLibraryA("kernel32.dll");
42 if (kernel32)
43 {
44 pCreateThreadpool = (void*)GetProcAddress(kernel32, "CreateThreadpool");
45 pCloseThreadpool = (void*)GetProcAddress(kernel32, "CloseThreadpool");
46 pSetThreadpoolThreadMinimum = (void*)GetProcAddress(kernel32, "SetThreadpoolThreadMinimum");
47 pSetThreadpoolThreadMaximum = (void*)GetProcAddress(kernel32, "SetThreadpoolThreadMaximum");
48 }
49 return TRUE;
50 }
51 #endif
52
53 static TP_POOL DEFAULT_POOL = {
54 0, /* DWORD Minimum */
55 500, /* DWORD Maximum */
56 NULL, /* wArrayList* Threads */
57 NULL, /* wQueue* PendingQueue */
58 NULL, /* HANDLE TerminateEvent */
59 NULL, /* wCountdownEvent* WorkComplete */
60 };
61
thread_pool_work_func(LPVOID arg)62 static DWORD WINAPI thread_pool_work_func(LPVOID arg)
63 {
64 DWORD status;
65 PTP_POOL pool;
66 PTP_WORK work;
67 HANDLE events[2];
68 PTP_CALLBACK_INSTANCE callbackInstance;
69
70 pool = (PTP_POOL)arg;
71
72 events[0] = pool->TerminateEvent;
73 events[1] = Queue_Event(pool->PendingQueue);
74
75 while (1)
76 {
77 status = WaitForMultipleObjects(2, events, FALSE, INFINITE);
78
79 if (status == WAIT_OBJECT_0)
80 break;
81
82 if (status != (WAIT_OBJECT_0 + 1))
83 break;
84
85 callbackInstance = (PTP_CALLBACK_INSTANCE)Queue_Dequeue(pool->PendingQueue);
86
87 if (callbackInstance)
88 {
89 work = callbackInstance->Work;
90 work->WorkCallback(callbackInstance, work->CallbackParameter, work);
91 CountdownEvent_Signal(pool->WorkComplete, 1);
92 free(callbackInstance);
93 }
94 }
95
96 ExitThread(0);
97 return 0;
98 }
99
threads_close(void * thread)100 static void threads_close(void* thread)
101 {
102 WaitForSingleObject(thread, INFINITE);
103 CloseHandle(thread);
104 }
105
InitializeThreadpool(PTP_POOL pool)106 static BOOL InitializeThreadpool(PTP_POOL pool)
107 {
108 int index;
109 HANDLE thread;
110
111 if (pool->Threads)
112 return TRUE;
113
114 pool->Minimum = 0;
115 pool->Maximum = 500;
116
117 if (!(pool->PendingQueue = Queue_New(TRUE, -1, -1)))
118 goto fail_queue_new;
119
120 if (!(pool->WorkComplete = CountdownEvent_New(0)))
121 goto fail_countdown_event;
122
123 if (!(pool->TerminateEvent = CreateEvent(NULL, TRUE, FALSE, NULL)))
124 goto fail_terminate_event;
125
126 if (!(pool->Threads = ArrayList_New(TRUE)))
127 goto fail_thread_array;
128
129 pool->Threads->object.fnObjectFree = threads_close;
130
131 for (index = 0; index < 4; index++)
132 {
133 if (!(thread = CreateThread(NULL, 0, thread_pool_work_func, (void*)pool, 0, NULL)))
134 {
135 goto fail_create_threads;
136 }
137
138 if (ArrayList_Add(pool->Threads, thread) < 0)
139 goto fail_create_threads;
140 }
141
142 return TRUE;
143
144 fail_create_threads:
145 SetEvent(pool->TerminateEvent);
146 ArrayList_Free(pool->Threads);
147 pool->Threads = NULL;
148 fail_thread_array:
149 CloseHandle(pool->TerminateEvent);
150 pool->TerminateEvent = NULL;
151 fail_terminate_event:
152 CountdownEvent_Free(pool->WorkComplete);
153 pool->WorkComplete = NULL;
154 fail_countdown_event:
155 Queue_Free(pool->PendingQueue);
156 pool->WorkComplete = NULL;
157 fail_queue_new:
158
159 return FALSE;
160 }
161
GetDefaultThreadpool()162 PTP_POOL GetDefaultThreadpool()
163 {
164 PTP_POOL pool = NULL;
165
166 pool = &DEFAULT_POOL;
167
168 if (!InitializeThreadpool(pool))
169 return NULL;
170
171 return pool;
172 }
173
winpr_CreateThreadpool(PVOID reserved)174 PTP_POOL winpr_CreateThreadpool(PVOID reserved)
175 {
176 PTP_POOL pool = NULL;
177 #ifdef _WIN32
178 InitOnceExecuteOnce(&init_once_module, init_module, NULL, NULL);
179 if (pCreateThreadpool)
180 return pCreateThreadpool(reserved);
181 #else
182 WINPR_UNUSED(reserved);
183 #endif
184 if (!(pool = (PTP_POOL)calloc(1, sizeof(TP_POOL))))
185 return NULL;
186
187 if (!InitializeThreadpool(pool))
188 {
189 free(pool);
190 return NULL;
191 }
192
193 return pool;
194 }
195
winpr_CloseThreadpool(PTP_POOL ptpp)196 VOID winpr_CloseThreadpool(PTP_POOL ptpp)
197 {
198 #ifdef _WIN32
199 InitOnceExecuteOnce(&init_once_module, init_module, NULL, NULL);
200 if (pCloseThreadpool)
201 {
202 pCloseThreadpool(ptpp);
203 return;
204 }
205 #endif
206 SetEvent(ptpp->TerminateEvent);
207
208 ArrayList_Free(ptpp->Threads);
209 Queue_Free(ptpp->PendingQueue);
210 CountdownEvent_Free(ptpp->WorkComplete);
211 CloseHandle(ptpp->TerminateEvent);
212
213 if (ptpp == &DEFAULT_POOL)
214 {
215 ptpp->Threads = NULL;
216 ptpp->PendingQueue = NULL;
217 ptpp->WorkComplete = NULL;
218 ptpp->TerminateEvent = NULL;
219 }
220 else
221 {
222 free(ptpp);
223 }
224 }
225
winpr_SetThreadpoolThreadMinimum(PTP_POOL ptpp,DWORD cthrdMic)226 BOOL winpr_SetThreadpoolThreadMinimum(PTP_POOL ptpp, DWORD cthrdMic)
227 {
228 HANDLE thread;
229 #ifdef _WIN32
230 InitOnceExecuteOnce(&init_once_module, init_module, NULL, NULL);
231 if (pSetThreadpoolThreadMinimum)
232 return pSetThreadpoolThreadMinimum(ptpp, cthrdMic);
233 #endif
234 ptpp->Minimum = cthrdMic;
235
236 while (ArrayList_Count(ptpp->Threads) < (INT64)ptpp->Minimum)
237 {
238 if (!(thread = CreateThread(NULL, 0, thread_pool_work_func, (void*)ptpp, 0, NULL)))
239 {
240 return FALSE;
241 }
242
243 if (ArrayList_Add(ptpp->Threads, thread) < 0)
244 return FALSE;
245 }
246
247 return TRUE;
248 }
249
winpr_SetThreadpoolThreadMaximum(PTP_POOL ptpp,DWORD cthrdMost)250 VOID winpr_SetThreadpoolThreadMaximum(PTP_POOL ptpp, DWORD cthrdMost)
251 {
252 #ifdef _WIN32
253 InitOnceExecuteOnce(&init_once_module, init_module, NULL, NULL);
254 if (pSetThreadpoolThreadMaximum)
255 {
256 pSetThreadpoolThreadMaximum(ptpp, cthrdMost);
257 return;
258 }
259 #endif
260 ptpp->Maximum = cthrdMost;
261 }
262
263 #endif /* WINPR_THREAD_POOL defined */
264