| /* |
| * Thread pooling |
| * |
| * Copyright (c) 2006 Robert Shearman |
| * |
| * This library is free software; you can redistribute it and/or |
| * modify it under the terms of the GNU Lesser General Public |
| * License as published by the Free Software Foundation; either |
| * version 2.1 of the License, or (at your option) any later version. |
| * |
| * This library is distributed in the hope that it will be useful, |
| * but WITHOUT ANY WARRANTY; without even the implied warranty of |
| * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU |
| * Lesser General Public License for more details. |
| * |
| * You should have received a copy of the GNU Lesser General Public |
| * License along with this library; if not, write to the Free Software |
| * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA |
| */ |
| |
| #include "config.h" |
| #include "wine/port.h" |
| |
| #include <stdarg.h> |
| #include <limits.h> |
| |
| #define NONAMELESSUNION |
| #include "ntstatus.h" |
| #define WIN32_NO_STATUS |
| #include "winternl.h" |
| |
| #include "wine/debug.h" |
| #include "wine/list.h" |
| |
| #include "ntdll_misc.h" |
| |
| WINE_DEFAULT_DEBUG_CHANNEL(threadpool); |
| |
| #define WORKER_TIMEOUT 30000 /* 30 seconds */ |
| |
| static LONG num_workers; |
| static LONG num_work_items; |
| static LONG num_busy_workers; |
| |
| static struct list work_item_list = LIST_INIT(work_item_list); |
| static HANDLE work_item_event; |
| |
| static RTL_CRITICAL_SECTION threadpool_cs; |
| static RTL_CRITICAL_SECTION_DEBUG critsect_debug = |
| { |
| 0, 0, &threadpool_cs, |
| { &critsect_debug.ProcessLocksList, &critsect_debug.ProcessLocksList }, |
| 0, 0, { (DWORD_PTR)(__FILE__ ": threadpool_cs") } |
| }; |
| static RTL_CRITICAL_SECTION threadpool_cs = { &critsect_debug, -1, 0, 0, 0, 0 }; |
| |
| static HANDLE compl_port = NULL; |
| static RTL_CRITICAL_SECTION threadpool_compl_cs; |
| static RTL_CRITICAL_SECTION_DEBUG critsect_compl_debug = |
| { |
| 0, 0, &threadpool_compl_cs, |
| { &critsect_compl_debug.ProcessLocksList, &critsect_compl_debug.ProcessLocksList }, |
| 0, 0, { (DWORD_PTR)(__FILE__ ": threadpool_compl_cs") } |
| }; |
| static RTL_CRITICAL_SECTION threadpool_compl_cs = { &critsect_compl_debug, -1, 0, 0, 0, 0 }; |
| |
| struct work_item |
| { |
| struct list entry; |
| PRTL_WORK_ITEM_ROUTINE function; |
| PVOID context; |
| }; |
| |
| static inline LONG interlocked_inc( PLONG dest ) |
| { |
| return interlocked_xchg_add( dest, 1 ) + 1; |
| } |
| |
| static inline LONG interlocked_dec( PLONG dest ) |
| { |
| return interlocked_xchg_add( dest, -1 ) - 1; |
| } |
| |
| static void WINAPI worker_thread_proc(void * param) |
| { |
| interlocked_inc(&num_workers); |
| |
| /* free the work item memory sooner to reduce memory usage */ |
| while (TRUE) |
| { |
| if (num_work_items > 0) |
| { |
| struct list *item; |
| RtlEnterCriticalSection(&threadpool_cs); |
| item = list_head(&work_item_list); |
| if (item) |
| { |
| struct work_item *work_item_ptr = LIST_ENTRY(item, struct work_item, entry); |
| struct work_item work_item; |
| list_remove(&work_item_ptr->entry); |
| interlocked_dec(&num_work_items); |
| |
| RtlLeaveCriticalSection(&threadpool_cs); |
| |
| work_item = *work_item_ptr; |
| RtlFreeHeap(GetProcessHeap(), 0, work_item_ptr); |
| |
| TRACE("executing %p(%p)\n", work_item.function, work_item.context); |
| |
| interlocked_inc(&num_busy_workers); |
| |
| /* do the work */ |
| work_item.function(work_item.context); |
| |
| interlocked_dec(&num_busy_workers); |
| } |
| else |
| RtlLeaveCriticalSection(&threadpool_cs); |
| } |
| else |
| { |
| NTSTATUS status; |
| LARGE_INTEGER timeout; |
| timeout.QuadPart = -(WORKER_TIMEOUT * (ULONGLONG)10000); |
| status = NtWaitForSingleObject(work_item_event, FALSE, &timeout); |
| if (status != STATUS_WAIT_0) |
| break; |
| } |
| } |
| |
| interlocked_dec(&num_workers); |
| |
| RtlExitUserThread(0); |
| |
| /* never reached */ |
| } |
| |
| static NTSTATUS add_work_item_to_queue(struct work_item *work_item) |
| { |
| NTSTATUS status; |
| |
| RtlEnterCriticalSection(&threadpool_cs); |
| list_add_tail(&work_item_list, &work_item->entry); |
| num_work_items++; |
| RtlLeaveCriticalSection(&threadpool_cs); |
| |
| if (!work_item_event) |
| { |
| HANDLE sem; |
| status = NtCreateSemaphore(&sem, SEMAPHORE_ALL_ACCESS, NULL, 1, LONG_MAX); |
| if (interlocked_cmpxchg_ptr( &work_item_event, sem, 0 )) |
| NtClose(sem); /* somebody beat us to it */ |
| } |
| else |
| status = NtReleaseSemaphore(work_item_event, 1, NULL); |
| |
| return status; |
| } |
| |
| /*********************************************************************** |
| * RtlQueueWorkItem (NTDLL.@) |
| * |
| * Queues a work item into a thread in the thread pool. |
| * |
| * PARAMS |
| * Function [I] Work function to execute. |
| * Context [I] Context to pass to the work function when it is executed. |
| * Flags [I] Flags. See notes. |
| * |
| * RETURNS |
| * Success: STATUS_SUCCESS. |
| * Failure: Any NTSTATUS code. |
| * |
| * NOTES |
| * Flags can be one or more of the following: |
| *|WT_EXECUTEDEFAULT - Executes the work item in a non-I/O worker thread. |
| *|WT_EXECUTEINIOTHREAD - Executes the work item in an I/O worker thread. |
| *|WT_EXECUTEINPERSISTENTTHREAD - Executes the work item in a thread that is persistent. |
| *|WT_EXECUTELONGFUNCTION - Hints that the execution can take a long time. |
| *|WT_TRANSFER_IMPERSONATION - Executes the function with the current access token. |
| */ |
| NTSTATUS WINAPI RtlQueueWorkItem(PRTL_WORK_ITEM_ROUTINE Function, PVOID Context, ULONG Flags) |
| { |
| HANDLE thread; |
| NTSTATUS status; |
| struct work_item *work_item = RtlAllocateHeap(GetProcessHeap(), 0, sizeof(struct work_item)); |
| |
| if (!work_item) |
| return STATUS_NO_MEMORY; |
| |
| work_item->function = Function; |
| work_item->context = Context; |
| |
| if (Flags & ~WT_EXECUTELONGFUNCTION) |
| FIXME("Flags 0x%x not supported\n", Flags); |
| |
| status = add_work_item_to_queue(work_item); |
| |
| /* FIXME: tune this algorithm to not be as aggressive with creating threads |
| * if WT_EXECUTELONGFUNCTION isn't specified */ |
| if ((status == STATUS_SUCCESS) && |
| ((num_workers == 0) || (num_workers == num_busy_workers))) |
| { |
| status = RtlCreateUserThread( GetCurrentProcess(), NULL, FALSE, |
| NULL, 0, 0, |
| worker_thread_proc, NULL, &thread, NULL ); |
| if (status == STATUS_SUCCESS) |
| NtClose( thread ); |
| |
| /* NOTE: we don't care if we couldn't create the thread if there is at |
| * least one other available to process the request */ |
| if ((num_workers > 0) && (status != STATUS_SUCCESS)) |
| status = STATUS_SUCCESS; |
| } |
| |
| if (status != STATUS_SUCCESS) |
| { |
| RtlEnterCriticalSection(&threadpool_cs); |
| |
| interlocked_dec(&num_work_items); |
| list_remove(&work_item->entry); |
| RtlFreeHeap(GetProcessHeap(), 0, work_item); |
| |
| RtlLeaveCriticalSection(&threadpool_cs); |
| |
| return status; |
| } |
| |
| return STATUS_SUCCESS; |
| } |
| |
| /*********************************************************************** |
| * iocp_poller - get completion events and run callbacks |
| */ |
| static DWORD CALLBACK iocp_poller(LPVOID Arg) |
| { |
| while( TRUE ) |
| { |
| PRTL_OVERLAPPED_COMPLETION_ROUTINE callback; |
| LPVOID overlapped; |
| IO_STATUS_BLOCK iosb; |
| NTSTATUS res = NtRemoveIoCompletion( compl_port, (PULONG_PTR)&callback, (PULONG_PTR)&overlapped, &iosb, NULL ); |
| if (res) |
| { |
| ERR("NtRemoveIoCompletion failed: 0x%x\n", res); |
| } |
| else |
| { |
| DWORD transferred = 0; |
| DWORD err = 0; |
| |
| if (iosb.u.Status == STATUS_SUCCESS) |
| transferred = iosb.Information; |
| else |
| err = RtlNtStatusToDosError(iosb.u.Status); |
| |
| callback( err, transferred, overlapped ); |
| } |
| } |
| } |
| |
| /*********************************************************************** |
| * RtlSetIoCompletionCallback (NTDLL.@) |
| * |
| * Binds a handle to a thread pool's completion port, and possibly |
| * starts a non-I/O thread to monitor this port and call functions back. |
| * |
| * PARAMS |
| * FileHandle [I] Handle to bind to a completion port. |
| * Function [I] Callback function to call on I/O completions. |
| * Flags [I] Not used. |
| * |
| * RETURNS |
| * Success: STATUS_SUCCESS. |
| * Failure: Any NTSTATUS code. |
| * |
| */ |
| NTSTATUS WINAPI RtlSetIoCompletionCallback(HANDLE FileHandle, PRTL_OVERLAPPED_COMPLETION_ROUTINE Function, ULONG Flags) |
| { |
| IO_STATUS_BLOCK iosb; |
| FILE_COMPLETION_INFORMATION info; |
| |
| if (Flags) FIXME("Unknown value Flags=0x%x\n", Flags); |
| |
| if (!compl_port) |
| { |
| NTSTATUS res = STATUS_SUCCESS; |
| |
| RtlEnterCriticalSection(&threadpool_compl_cs); |
| if (!compl_port) |
| { |
| HANDLE cport; |
| |
| res = NtCreateIoCompletion( &cport, IO_COMPLETION_ALL_ACCESS, NULL, 0 ); |
| if (!res) |
| { |
| /* FIXME native can start additional threads in case of e.g. hung callback function. */ |
| res = RtlQueueWorkItem( iocp_poller, NULL, WT_EXECUTEDEFAULT ); |
| if (!res) |
| compl_port = cport; |
| else |
| NtClose( cport ); |
| } |
| } |
| RtlLeaveCriticalSection(&threadpool_compl_cs); |
| if (res) return res; |
| } |
| |
| info.CompletionPort = compl_port; |
| info.CompletionKey = (ULONG_PTR)Function; |
| |
| return NtSetInformationFile( FileHandle, &iosb, &info, sizeof(info), FileCompletionInformation ); |
| } |
| |
| static inline PLARGE_INTEGER get_nt_timeout( PLARGE_INTEGER pTime, ULONG timeout ) |
| { |
| if (timeout == INFINITE) return NULL; |
| pTime->QuadPart = (ULONGLONG)timeout * -10000; |
| return pTime; |
| } |
| |
| struct wait_work_item |
| { |
| HANDLE Object; |
| HANDLE CancelEvent; |
| WAITORTIMERCALLBACK Callback; |
| PVOID Context; |
| ULONG Milliseconds; |
| ULONG Flags; |
| HANDLE CompletionEvent; |
| LONG DeleteCount; |
| BOOLEAN CallbackInProgress; |
| }; |
| |
| static void delete_wait_work_item(struct wait_work_item *wait_work_item) |
| { |
| NtClose( wait_work_item->CancelEvent ); |
| RtlFreeHeap( GetProcessHeap(), 0, wait_work_item ); |
| } |
| |
| static DWORD CALLBACK wait_thread_proc(LPVOID Arg) |
| { |
| struct wait_work_item *wait_work_item = Arg; |
| NTSTATUS status; |
| BOOLEAN alertable = (wait_work_item->Flags & WT_EXECUTEINIOTHREAD) ? TRUE : FALSE; |
| HANDLE handles[2] = { wait_work_item->Object, wait_work_item->CancelEvent }; |
| LARGE_INTEGER timeout; |
| HANDLE completion_event; |
| |
| TRACE("\n"); |
| |
| while (TRUE) |
| { |
| status = NtWaitForMultipleObjects( 2, handles, FALSE, alertable, |
| get_nt_timeout( &timeout, wait_work_item->Milliseconds ) ); |
| if (status == STATUS_WAIT_0 || status == STATUS_TIMEOUT) |
| { |
| BOOLEAN TimerOrWaitFired; |
| |
| if (status == STATUS_WAIT_0) |
| { |
| TRACE( "object %p signaled, calling callback %p with context %p\n", |
| wait_work_item->Object, wait_work_item->Callback, |
| wait_work_item->Context ); |
| TimerOrWaitFired = FALSE; |
| } |
| else |
| { |
| TRACE( "wait for object %p timed out, calling callback %p with context %p\n", |
| wait_work_item->Object, wait_work_item->Callback, |
| wait_work_item->Context ); |
| TimerOrWaitFired = TRUE; |
| } |
| wait_work_item->CallbackInProgress = TRUE; |
| wait_work_item->Callback( wait_work_item->Context, TimerOrWaitFired ); |
| wait_work_item->CallbackInProgress = FALSE; |
| |
| if (wait_work_item->Flags & WT_EXECUTEONLYONCE) |
| break; |
| } |
| else |
| break; |
| } |
| |
| completion_event = wait_work_item->CompletionEvent; |
| if (completion_event) NtSetEvent( completion_event, NULL ); |
| |
| if (interlocked_inc( &wait_work_item->DeleteCount ) == 2 ) |
| delete_wait_work_item( wait_work_item ); |
| |
| return 0; |
| } |
| |
| /*********************************************************************** |
| * RtlRegisterWait (NTDLL.@) |
| * |
| * Registers a wait for a handle to become signaled. |
| * |
| * PARAMS |
| * NewWaitObject [I] Handle to the new wait object. Use RtlDeregisterWait() to free it. |
| * Object [I] Object to wait to become signaled. |
| * Callback [I] Callback function to execute when the wait times out or the handle is signaled. |
| * Context [I] Context to pass to the callback function when it is executed. |
| * Milliseconds [I] Number of milliseconds to wait before timing out. |
| * Flags [I] Flags. See notes. |
| * |
| * RETURNS |
| * Success: STATUS_SUCCESS. |
| * Failure: Any NTSTATUS code. |
| * |
| * NOTES |
| * Flags can be one or more of the following: |
| *|WT_EXECUTEDEFAULT - Executes the work item in a non-I/O worker thread. |
| *|WT_EXECUTEINIOTHREAD - Executes the work item in an I/O worker thread. |
| *|WT_EXECUTEINPERSISTENTTHREAD - Executes the work item in a thread that is persistent. |
| *|WT_EXECUTELONGFUNCTION - Hints that the execution can take a long time. |
| *|WT_TRANSFER_IMPERSONATION - Executes the function with the current access token. |
| */ |
| NTSTATUS WINAPI RtlRegisterWait(PHANDLE NewWaitObject, HANDLE Object, |
| RTL_WAITORTIMERCALLBACKFUNC Callback, |
| PVOID Context, ULONG Milliseconds, ULONG Flags) |
| { |
| struct wait_work_item *wait_work_item; |
| NTSTATUS status; |
| |
| TRACE( "(%p, %p, %p, %p, %d, 0x%x)\n", NewWaitObject, Object, Callback, Context, Milliseconds, Flags ); |
| |
| wait_work_item = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*wait_work_item) ); |
| if (!wait_work_item) |
| return STATUS_NO_MEMORY; |
| |
| wait_work_item->Object = Object; |
| wait_work_item->Callback = Callback; |
| wait_work_item->Context = Context; |
| wait_work_item->Milliseconds = Milliseconds; |
| wait_work_item->Flags = Flags; |
| wait_work_item->CallbackInProgress = FALSE; |
| wait_work_item->DeleteCount = 0; |
| wait_work_item->CompletionEvent = NULL; |
| |
| status = NtCreateEvent( &wait_work_item->CancelEvent, EVENT_ALL_ACCESS, NULL, TRUE, FALSE ); |
| if (status != STATUS_SUCCESS) |
| { |
| RtlFreeHeap( GetProcessHeap(), 0, wait_work_item ); |
| return status; |
| } |
| |
| status = RtlQueueWorkItem( wait_thread_proc, wait_work_item, Flags & ~WT_EXECUTEONLYONCE ); |
| if (status != STATUS_SUCCESS) |
| { |
| delete_wait_work_item( wait_work_item ); |
| return status; |
| } |
| |
| *NewWaitObject = wait_work_item; |
| return status; |
| } |
| |
| /*********************************************************************** |
| * RtlDeregisterWaitEx (NTDLL.@) |
| * |
| * Cancels a wait operation and frees the resources associated with calling |
| * RtlRegisterWait(). |
| * |
| * PARAMS |
| * WaitObject [I] Handle to the wait object to free. |
| * |
| * RETURNS |
| * Success: STATUS_SUCCESS. |
| * Failure: Any NTSTATUS code. |
| */ |
| NTSTATUS WINAPI RtlDeregisterWaitEx(HANDLE WaitHandle, HANDLE CompletionEvent) |
| { |
| struct wait_work_item *wait_work_item = WaitHandle; |
| NTSTATUS status = STATUS_SUCCESS; |
| |
| TRACE( "(%p)\n", WaitHandle ); |
| |
| NtSetEvent( wait_work_item->CancelEvent, NULL ); |
| if (wait_work_item->CallbackInProgress) |
| { |
| if (CompletionEvent != NULL) |
| { |
| if (CompletionEvent == INVALID_HANDLE_VALUE) |
| { |
| status = NtCreateEvent( &CompletionEvent, EVENT_ALL_ACCESS, NULL, TRUE, FALSE ); |
| if (status != STATUS_SUCCESS) |
| return status; |
| interlocked_xchg_ptr( &wait_work_item->CompletionEvent, CompletionEvent ); |
| if (wait_work_item->CallbackInProgress) |
| NtWaitForSingleObject( CompletionEvent, FALSE, NULL ); |
| NtClose( CompletionEvent ); |
| } |
| else |
| { |
| interlocked_xchg_ptr( &wait_work_item->CompletionEvent, CompletionEvent ); |
| if (wait_work_item->CallbackInProgress) |
| status = STATUS_PENDING; |
| } |
| } |
| else |
| status = STATUS_PENDING; |
| } |
| |
| if (interlocked_inc( &wait_work_item->DeleteCount ) == 2 ) |
| { |
| status = STATUS_SUCCESS; |
| delete_wait_work_item( wait_work_item ); |
| } |
| |
| return status; |
| } |
| |
| /*********************************************************************** |
| * RtlDeregisterWait (NTDLL.@) |
| * |
| * Cancels a wait operation and frees the resources associated with calling |
| * RtlRegisterWait(). |
| * |
| * PARAMS |
| * WaitObject [I] Handle to the wait object to free. |
| * |
| * RETURNS |
| * Success: STATUS_SUCCESS. |
| * Failure: Any NTSTATUS code. |
| */ |
| NTSTATUS WINAPI RtlDeregisterWait(HANDLE WaitHandle) |
| { |
| return RtlDeregisterWaitEx(WaitHandle, NULL); |
| } |
