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
, (PVOID
)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
);