4 * Copyright (c) 2006 Robert Shearman
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, write to the Free Software
18 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
22 #include "wine/port.h"
27 #define NONAMELESSUNION
29 #define WIN32_NO_STATUS
32 #include "wine/debug.h"
33 #include "wine/list.h"
35 #include "ntdll_misc.h"
37 WINE_DEFAULT_DEBUG_CHANNEL(threadpool);
39 #define WORKER_TIMEOUT 30000 /* 30 seconds */
41 static LONG num_workers;
42 static LONG num_work_items;
43 static LONG num_busy_workers;
45 static struct list work_item_list = LIST_INIT(work_item_list);
46 static HANDLE work_item_event;
48 static RTL_CRITICAL_SECTION threadpool_cs;
49 static RTL_CRITICAL_SECTION_DEBUG critsect_debug =
52 { &critsect_debug.ProcessLocksList, &critsect_debug.ProcessLocksList },
53 0, 0, { (DWORD_PTR)(__FILE__ ": threadpool_cs") }
55 static RTL_CRITICAL_SECTION threadpool_cs = { &critsect_debug, -1, 0, 0, 0, 0 };
57 static HANDLE compl_port = NULL;
58 static RTL_CRITICAL_SECTION threadpool_compl_cs;
59 static RTL_CRITICAL_SECTION_DEBUG critsect_compl_debug =
61 0, 0, &threadpool_compl_cs,
62 { &critsect_compl_debug.ProcessLocksList, &critsect_compl_debug.ProcessLocksList },
63 0, 0, { (DWORD_PTR)(__FILE__ ": threadpool_compl_cs") }
65 static RTL_CRITICAL_SECTION threadpool_compl_cs = { &critsect_compl_debug, -1, 0, 0, 0, 0 };
70 PRTL_WORK_ITEM_ROUTINE function;
74 static inline LONG interlocked_inc( PLONG dest )
76 return interlocked_xchg_add( (int *)dest, 1 ) + 1;
79 static inline LONG interlocked_dec( PLONG dest )
81 return interlocked_xchg_add( (int *)dest, -1 ) - 1;
84 static void WINAPI worker_thread_proc(void * param)
86 interlocked_inc(&num_workers);
88 /* free the work item memory sooner to reduce memory usage */
91 if (num_work_items > 0)
94 RtlEnterCriticalSection(&threadpool_cs);
95 item = list_head(&work_item_list);
98 struct work_item *work_item_ptr = LIST_ENTRY(item, struct work_item, entry);
99 struct work_item work_item;
100 list_remove(&work_item_ptr->entry);
101 interlocked_dec(&num_work_items);
103 RtlLeaveCriticalSection(&threadpool_cs);
105 work_item = *work_item_ptr;
106 RtlFreeHeap(GetProcessHeap(), 0, work_item_ptr);
108 TRACE("executing %p(%p)\n", work_item.function, work_item.context);
110 interlocked_inc(&num_busy_workers);
113 work_item.function(work_item.context);
115 interlocked_dec(&num_busy_workers);
118 RtlLeaveCriticalSection(&threadpool_cs);
123 LARGE_INTEGER timeout;
124 timeout.QuadPart = -(WORKER_TIMEOUT * (ULONGLONG)10000);
125 status = NtWaitForSingleObject(work_item_event, FALSE, &timeout);
126 if (status != STATUS_WAIT_0)
131 interlocked_dec(&num_workers);
133 RtlExitUserThread(0);
138 static NTSTATUS add_work_item_to_queue(struct work_item *work_item)
142 RtlEnterCriticalSection(&threadpool_cs);
143 list_add_tail(&work_item_list, &work_item->entry);
145 RtlLeaveCriticalSection(&threadpool_cs);
147 if (!work_item_event)
150 status = NtCreateSemaphore(&sem, SEMAPHORE_ALL_ACCESS, NULL, 1, LONG_MAX);
151 if (interlocked_cmpxchg_ptr( (PVOID *)&work_item_event, sem, 0 ))
152 NtClose(sem); /* somebody beat us to it */
155 status = NtReleaseSemaphore(work_item_event, 1, NULL);
160 /***********************************************************************
161 * RtlQueueWorkItem (NTDLL.@)
163 * Queues a work item into a thread in the thread pool.
166 * Function [I] Work function to execute.
167 * Context [I] Context to pass to the work function when it is executed.
168 * Flags [I] Flags. See notes.
171 * Success: STATUS_SUCCESS.
172 * Failure: Any NTSTATUS code.
175 * Flags can be one or more of the following:
176 *|WT_EXECUTEDEFAULT - Executes the work item in a non-I/O worker thread.
177 *|WT_EXECUTEINIOTHREAD - Executes the work item in an I/O worker thread.
178 *|WT_EXECUTEINPERSISTENTTHREAD - Executes the work item in a thread that is persistent.
179 *|WT_EXECUTELONGFUNCTION - Hints that the execution can take a long time.
180 *|WT_TRANSFER_IMPERSONATION - Executes the function with the current access token.
182 NTSTATUS WINAPI RtlQueueWorkItem(PRTL_WORK_ITEM_ROUTINE Function, PVOID Context, ULONG Flags)
186 struct work_item *work_item = RtlAllocateHeap(GetProcessHeap(), 0, sizeof(struct work_item));
189 return STATUS_NO_MEMORY;
191 work_item->function = Function;
192 work_item->context = Context;
194 if (Flags & ~WT_EXECUTELONGFUNCTION)
195 FIXME("Flags 0x%x not supported\n", Flags);
197 status = add_work_item_to_queue(work_item);
199 /* FIXME: tune this algorithm to not be as aggressive with creating threads
200 * if WT_EXECUTELONGFUNCTION isn't specified */
201 if ((status == STATUS_SUCCESS) &&
202 ((num_workers == 0) || (num_workers == num_busy_workers)))
204 status = RtlCreateUserThread( GetCurrentProcess(), NULL, FALSE,
206 worker_thread_proc, NULL, &thread, NULL );
207 if (status == STATUS_SUCCESS)
210 /* NOTE: we don't care if we couldn't create the thread if there is at
211 * least one other available to process the request */
212 if ((num_workers > 0) && (status != STATUS_SUCCESS))
213 status = STATUS_SUCCESS;
216 if (status != STATUS_SUCCESS)
218 RtlEnterCriticalSection(&threadpool_cs);
220 interlocked_dec(&num_work_items);
221 list_remove(&work_item->entry);
222 RtlFreeHeap(GetProcessHeap(), 0, work_item);
224 RtlLeaveCriticalSection(&threadpool_cs);
229 return STATUS_SUCCESS;
232 /***********************************************************************
233 * iocp_poller - get completion events and run callbacks
235 static DWORD CALLBACK iocp_poller(LPVOID Arg)
239 PRTL_OVERLAPPED_COMPLETION_ROUTINE callback;
241 IO_STATUS_BLOCK iosb;
242 NTSTATUS res = NtRemoveIoCompletion( compl_port, (PULONG_PTR)&callback, (PULONG_PTR)&overlapped, &iosb, NULL );
245 ERR("NtRemoveIoCompletion failed: 0x%x\n", res);
249 DWORD transferred = 0;
252 if (iosb.u.Status == STATUS_SUCCESS)
253 transferred = iosb.Information;
255 err = RtlNtStatusToDosError(iosb.u.Status);
257 callback( err, transferred, overlapped );
262 /***********************************************************************
263 * RtlSetIoCompletionCallback (NTDLL.@)
265 * Binds a handle to a thread pool's completion port, and possibly
266 * starts a non-I/O thread to monitor this port and call functions back.
269 * FileHandle [I] Handle to bind to a completion port.
270 * Function [I] Callback function to call on I/O completions.
271 * Flags [I] Not used.
274 * Success: STATUS_SUCCESS.
275 * Failure: Any NTSTATUS code.
278 NTSTATUS WINAPI RtlSetIoCompletionCallback(HANDLE FileHandle, PRTL_OVERLAPPED_COMPLETION_ROUTINE Function, ULONG Flags)
280 IO_STATUS_BLOCK iosb;
281 FILE_COMPLETION_INFORMATION info;
283 if (Flags) FIXME("Unknown value Flags=0x%x\n", Flags);
287 NTSTATUS res = STATUS_SUCCESS;
289 RtlEnterCriticalSection(&threadpool_compl_cs);
294 res = NtCreateIoCompletion( &cport, IO_COMPLETION_ALL_ACCESS, NULL, 0 );
297 /* FIXME native can start additional threads in case of e.g. hung callback function. */
298 res = RtlQueueWorkItem( iocp_poller, NULL, WT_EXECUTEDEFAULT );
305 RtlLeaveCriticalSection(&threadpool_compl_cs);
309 info.CompletionPort = compl_port;
310 info.CompletionKey = (ULONG_PTR)Function;
312 return NtSetInformationFile( FileHandle, &iosb, &info, sizeof(info), FileCompletionInformation );