/** * WinPR: Windows Portable Runtime * Thread Pool API (Pool) * * Copyright 2012 Marc-Andre Moreau * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #include #include #include #include #include #include "pool.h" #ifdef WINPR_THREAD_POOL #ifdef _WIN32 static INIT_ONCE init_once_module = INIT_ONCE_STATIC_INIT; static PTP_POOL(WINAPI* pCreateThreadpool)(PVOID reserved); static VOID(WINAPI* pCloseThreadpool)(PTP_POOL ptpp); static BOOL(WINAPI* pSetThreadpoolThreadMinimum)(PTP_POOL ptpp, DWORD cthrdMic); static VOID(WINAPI* pSetThreadpoolThreadMaximum)(PTP_POOL ptpp, DWORD cthrdMost); static BOOL CALLBACK init_module(PINIT_ONCE once, PVOID param, PVOID* context) { HMODULE kernel32 = LoadLibraryA("kernel32.dll"); if (kernel32) { pCreateThreadpool = GetProcAddressAs(kernel32, "CreateThreadpool", void*); pCloseThreadpool = GetProcAddressAs(kernel32, "CloseThreadpool", void*); pSetThreadpoolThreadMinimum = GetProcAddressAs(kernel32, "SetThreadpoolThreadMinimum", void*); pSetThreadpoolThreadMaximum = GetProcAddressAs(kernel32, "SetThreadpoolThreadMaximum", void*); } return TRUE; } #endif static TP_POOL DEFAULT_POOL = { 0, /* DWORD Minimum */ 500, /* DWORD Maximum */ nullptr, /* wArrayList* Threads */ nullptr, /* wQueue* PendingQueue */ nullptr, /* HANDLE TerminateEvent */ nullptr, /* wCountdownEvent* WorkComplete */ }; static DWORD WINAPI thread_pool_work_func(LPVOID arg) { DWORD status = 0; PTP_POOL pool = nullptr; PTP_WORK work = nullptr; HANDLE events[2]; PTP_CALLBACK_INSTANCE callbackInstance = nullptr; pool = (PTP_POOL)arg; events[0] = pool->TerminateEvent; events[1] = Queue_Event(pool->PendingQueue); while (1) { status = WaitForMultipleObjects(2, events, FALSE, INFINITE); if (status == WAIT_OBJECT_0) break; if (status != (WAIT_OBJECT_0 + 1)) break; callbackInstance = (PTP_CALLBACK_INSTANCE)Queue_Dequeue(pool->PendingQueue); if (callbackInstance) { work = callbackInstance->Work; work->WorkCallback(callbackInstance, work->CallbackParameter, work); CountdownEvent_Signal(pool->WorkComplete, 1); free(callbackInstance); } } ExitThread(0); return 0; } static void threads_close(void* thread) { (void)WaitForSingleObject(thread, INFINITE); (void)CloseHandle(thread); } static BOOL InitializeThreadpool(PTP_POOL pool) { BOOL rc = FALSE; wObject* obj = nullptr; if (pool->Threads) return TRUE; if (!(pool->PendingQueue = Queue_New(TRUE, -1, -1))) goto fail; if (!(pool->WorkComplete = CountdownEvent_New(0))) goto fail; if (!(pool->TerminateEvent = CreateEvent(nullptr, TRUE, FALSE, nullptr))) goto fail; if (!(pool->Threads = ArrayList_New(TRUE))) goto fail; obj = ArrayList_Object(pool->Threads); obj->fnObjectFree = threads_close; #if !defined(WINPR_THREADPOOL_DEFAULT_MIN_COUNT) #error "WINPR_THREADPOOL_DEFAULT_MIN_COUNT must be defined" #endif #if !defined(WINPR_THREADPOOL_DEFAULT_MAX_COUNT) #error "WINPR_THREADPOOL_DEFAULT_MAX_COUNT must be defined" #endif { SYSTEM_INFO info = WINPR_C_ARRAY_INIT; GetSystemInfo(&info); DWORD min = info.dwNumberOfProcessors; DWORD max = info.dwNumberOfProcessors; if (info.dwNumberOfProcessors < WINPR_THREADPOOL_DEFAULT_MIN_COUNT) min = WINPR_THREADPOOL_DEFAULT_MIN_COUNT; if (info.dwNumberOfProcessors > WINPR_THREADPOOL_DEFAULT_MAX_COUNT) max = WINPR_THREADPOOL_DEFAULT_MAX_COUNT; if (min > max) min = max; if (!SetThreadpoolThreadMinimum(pool, min)) goto fail; SetThreadpoolThreadMaximum(pool, max); } rc = TRUE; fail: return rc; } PTP_POOL GetDefaultThreadpool(void) { PTP_POOL pool = &DEFAULT_POOL; if (!InitializeThreadpool(pool)) return nullptr; return pool; } PTP_POOL winpr_CreateThreadpool(PVOID reserved) { PTP_POOL pool = nullptr; #ifdef _WIN32 if (!InitOnceExecuteOnce(&init_once_module, init_module, nullptr, nullptr)) return nullptr; if (pCreateThreadpool) return pCreateThreadpool(reserved); #else WINPR_UNUSED(reserved); #endif if (!(pool = (PTP_POOL)calloc(1, sizeof(TP_POOL)))) return nullptr; if (!InitializeThreadpool(pool)) { winpr_CloseThreadpool(pool); return nullptr; } return pool; } VOID winpr_CloseThreadpool(PTP_POOL ptpp) { #ifdef _WIN32 if (!InitOnceExecuteOnce(&init_once_module, init_module, nullptr, nullptr)) return; if (pCloseThreadpool) { pCloseThreadpool(ptpp); return; } #endif (void)SetEvent(ptpp->TerminateEvent); ArrayList_Free(ptpp->Threads); Queue_Free(ptpp->PendingQueue); CountdownEvent_Free(ptpp->WorkComplete); (void)CloseHandle(ptpp->TerminateEvent); { TP_POOL empty = WINPR_C_ARRAY_INIT; *ptpp = empty; } if (ptpp != &DEFAULT_POOL) free(ptpp); } BOOL winpr_SetThreadpoolThreadMinimum(PTP_POOL ptpp, DWORD cthrdMic) { BOOL rc = FALSE; #ifdef _WIN32 if (!InitOnceExecuteOnce(&init_once_module, init_module, nullptr, nullptr)) return FALSE; if (pSetThreadpoolThreadMinimum) return pSetThreadpoolThreadMinimum(ptpp, cthrdMic); #endif ptpp->Minimum = cthrdMic; ArrayList_Lock(ptpp->Threads); while (ArrayList_Count(ptpp->Threads) < ptpp->Minimum) { HANDLE thread = CreateThread(nullptr, 0, thread_pool_work_func, (void*)ptpp, 0, nullptr); if (!thread) goto fail; if (!ArrayList_Append(ptpp->Threads, thread)) { (void)CloseHandle(thread); goto fail; } } rc = TRUE; fail: ArrayList_Unlock(ptpp->Threads); return rc; } VOID winpr_SetThreadpoolThreadMaximum(PTP_POOL ptpp, DWORD cthrdMost) { #ifdef _WIN32 if (!InitOnceExecuteOnce(&init_once_module, init_module, nullptr, nullptr)) return; if (pSetThreadpoolThreadMaximum) { pSetThreadpoolThreadMaximum(ptpp, cthrdMost); return; } #endif ptpp->Maximum = cthrdMost; ArrayList_Lock(ptpp->Threads); if (ArrayList_Count(ptpp->Threads) > ptpp->Maximum) { (void)SetEvent(ptpp->TerminateEvent); ArrayList_Clear(ptpp->Threads); (void)ResetEvent(ptpp->TerminateEvent); } ArrayList_Unlock(ptpp->Threads); winpr_SetThreadpoolThreadMinimum(ptpp, ptpp->Minimum); } #endif /* WINPR_THREAD_POOL defined */