27 Jan
2025
27 Jan
'25
7:31 a.m.
https://git.reactos.org/?p=reactos.git;a=commitdiff;h=405ed2b4ceeff7e011e35…
commit 405ed2b4ceeff7e011e35e5308abdfc582f29396
Author: Denis Malikov <filedem(a)gmail.com>
AuthorDate: Thu Apr 25 21:08:41 2024 -0700
Commit: Justin Miller <justinmiller100(a)gmail.com>
CommitDate: Sun Jan 26 23:30:59 2025 -0800
[KERNEL32_VISTA][SDK] Import Threadpool.c from wine-9.7
---
dll/ntdll/nt_0600/ntdll_vista.spec | 29 +
media/doc/WINESYNC.txt | 3 +-
sdk/include/ndk/rtlfuncs.h | 48 +
sdk/include/psdk/winbase.h | 1 +
sdk/include/xdk/winnt_old.h | 7 +
sdk/lib/rtl/CMakeLists.txt | 1 +
sdk/lib/rtl/threadpool.c | 3357 ++++++++++++++++++++++++++++++++++++
7 files changed, 3444 insertions(+), 2 deletions(-)
diff --git a/dll/ntdll/nt_0600/ntdll_vista.spec b/dll/ntdll/nt_0600/ntdll_vista.spec
index d36c6449d8e..23fd88e02c4 100644
--- a/dll/ntdll/nt_0600/ntdll_vista.spec
+++ b/dll/ntdll/nt_0600/ntdll_vista.spec
@@ -13,5 +13,34 @@
@ stdcall RtlRunOnceComplete(ptr long ptr)
@ stdcall RtlRunOnceExecuteOnce(ptr ptr ptr ptr)
+@ stdcall TpAllocCleanupGroup(ptr)
+@ stdcall TpAllocPool(ptr ptr)
+@ stdcall TpAllocTimer(ptr ptr ptr ptr)
+@ stdcall TpAllocWait(ptr ptr ptr ptr)
+@ stdcall TpAllocWork(ptr ptr ptr ptr)
+@ stdcall TpCallbackLeaveCriticalSectionOnCompletion(ptr ptr)
+@ stdcall TpCallbackMayRunLong(ptr)
+@ stdcall TpCallbackReleaseMutexOnCompletion(ptr ptr)
+@ stdcall TpCallbackReleaseSemaphoreOnCompletion(ptr ptr long)
+@ stdcall TpCallbackSetEventOnCompletion(ptr ptr)
+@ stdcall TpCallbackUnloadDllOnCompletion(ptr ptr)
+@ stdcall TpDisassociateCallback(ptr)
+@ stdcall TpIsTimerSet(ptr)
+@ stdcall TpPostWork(ptr)
+@ stdcall TpReleaseCleanupGroup(ptr)
+@ stdcall TpReleaseCleanupGroupMembers(ptr long ptr)
+@ stdcall TpReleasePool(ptr)
+@ stdcall TpReleaseTimer(ptr)
+@ stdcall TpReleaseWait(ptr)
+@ stdcall TpReleaseWork(ptr)
+@ stdcall TpSetPoolMaxThreads(ptr long)
+@ stdcall TpSetPoolMinThreads(ptr long)
+@ stdcall TpSetTimer(ptr ptr long long)
+@ stdcall TpSetWait(ptr long ptr)
+@ stdcall TpSimpleTryPost(ptr ptr ptr)
+@ stdcall TpWaitForTimer(ptr long)
+@ stdcall TpWaitForWait(ptr long)
+@ stdcall TpWaitForWork(ptr long)
+
@ stdcall RtlConnectToSm(ptr ptr long ptr) SmConnectToSm
@ stdcall RtlSendMsgToSm(ptr ptr) SmSendMsgToSm
diff --git a/media/doc/WINESYNC.txt b/media/doc/WINESYNC.txt
index bf5b76bfb55..61e11c67f15 100644
--- a/media/doc/WINESYNC.txt
+++ b/media/doc/WINESYNC.txt
@@ -268,8 +268,7 @@ check Wine current sources first as it may already be fixed.
sdk/lib/3rdparty/strmbase # Synced to WineStaging-3.3
sdk/lib/rtl/actctx.c # Synced to wine-5.18
-sdk/lib/rtl/timerqueue.c # Synced to wine-5.18
-sdk/lib/rtl/wait.c # Synced to wine-5.18
+sdk/lib/rtl/threadpool.c # Synced with wine-9.7
advapi32 -
dll/win32/advapi32/wine/cred.c # Synced to WineStaging-3.3
diff --git a/sdk/include/ndk/rtlfuncs.h b/sdk/include/ndk/rtlfuncs.h
index 3a00f490935..7b128cd440b 100644
--- a/sdk/include/ndk/rtlfuncs.h
+++ b/sdk/include/ndk/rtlfuncs.h
@@ -4745,6 +4745,54 @@ RtlSleepConditionVariableSRW(
_In_ ULONG Flags);
#endif
+//
+// Synchronization functions
+//
+VOID
+NTAPI
+RtlInitializeConditionVariable(OUT PRTL_CONDITION_VARIABLE ConditionVariable);
+
+VOID
+NTAPI
+RtlWakeConditionVariable(IN OUT PRTL_CONDITION_VARIABLE ConditionVariable);
+
+VOID
+NTAPI
+RtlWakeAllConditionVariable(IN OUT PRTL_CONDITION_VARIABLE ConditionVariable);
+
+NTSTATUS
+NTAPI
+RtlSleepConditionVariableCS(IN OUT PRTL_CONDITION_VARIABLE ConditionVariable,
+ IN OUT PRTL_CRITICAL_SECTION CriticalSection,
+ IN const PLARGE_INTEGER TimeOut OPTIONAL);
+
+NTSTATUS
+NTAPI
+RtlSleepConditionVariableSRW(IN OUT PRTL_CONDITION_VARIABLE ConditionVariable,
+ IN OUT PRTL_SRWLOCK SRWLock,
+ IN PLARGE_INTEGER TimeOut OPTIONAL,
+ IN ULONG Flags);
+
+VOID
+NTAPI
+RtlInitializeSRWLock(OUT PRTL_SRWLOCK SRWLock);
+
+VOID
+NTAPI
+RtlAcquireSRWLockShared(IN OUT PRTL_SRWLOCK SRWLock);
+
+VOID
+NTAPI
+RtlReleaseSRWLockShared(IN OUT PRTL_SRWLOCK SRWLock);
+
+VOID
+NTAPI
+RtlAcquireSRWLockExclusive(IN OUT PRTL_SRWLOCK SRWLock);
+
+VOID
+NTAPI
+RtlReleaseSRWLockExclusive(IN OUT PRTL_SRWLOCK SRWLock);
+
//
// Secure Memory Functions
//
diff --git a/sdk/include/psdk/winbase.h b/sdk/include/psdk/winbase.h
index f5eed964654..4ff919b83f1 100644
--- a/sdk/include/psdk/winbase.h
+++ b/sdk/include/psdk/winbase.h
@@ -4123,6 +4123,7 @@ InitOnceExecuteOnce(
_Inout_opt_ PVOID Parameter,
_Outptr_opt_result_maybenull_ LPVOID *Context);
+typedef VOID (NTAPI
*PTP_WIN32_IO_CALLBACK)(PTP_CALLBACK_INSTANCE,PVOID,PVOID,ULONG,ULONG_PTR,PTP_IO);
#if defined(_SLIST_HEADER_) && !defined(_NTOS_) && !defined(_NTOSP_)
diff --git a/sdk/include/xdk/winnt_old.h b/sdk/include/xdk/winnt_old.h
index 157c3724e80..ee28a933347 100644
--- a/sdk/include/xdk/winnt_old.h
+++ b/sdk/include/xdk/winnt_old.h
@@ -4480,7 +4480,11 @@ DbgRaiseAssertionFailure(VOID)
typedef struct _TP_POOL TP_POOL, *PTP_POOL;
typedef struct _TP_WORK TP_WORK, *PTP_WORK;
typedef struct _TP_CALLBACK_INSTANCE TP_CALLBACK_INSTANCE, *PTP_CALLBACK_INSTANCE;
+typedef struct _TP_TIMER TP_TIMER, *PTP_TIMER;
+typedef struct _TP_WAIT TP_WAIT, *PTP_WAIT;
+typedef struct _TP_IO TP_IO, *PTP_IO;
+typedef DWORD TP_WAIT_RESULT;
typedef DWORD TP_VERSION, *PTP_VERSION;
typedef enum _TP_CALLBACK_PRIORITY {
@@ -4509,6 +4513,9 @@ typedef VOID
_Inout_opt_ PVOID ObjectContext,
_Inout_opt_ PVOID CleanupContext);
+typedef VOID (NTAPI *PTP_TIMER_CALLBACK)(PTP_CALLBACK_INSTANCE,PVOID,PTP_TIMER);
+typedef VOID (NTAPI
*PTP_WAIT_CALLBACK)(PTP_CALLBACK_INSTANCE,PVOID,PTP_WAIT,TP_WAIT_RESULT);
+
#if (_WIN32_WINNT >= _WIN32_WINNT_WIN7)
typedef struct _TP_CALLBACK_ENVIRON_V3 {
TP_VERSION Version;
diff --git a/sdk/lib/rtl/CMakeLists.txt b/sdk/lib/rtl/CMakeLists.txt
index efe2c90a40b..78077605a91 100644
--- a/sdk/lib/rtl/CMakeLists.txt
+++ b/sdk/lib/rtl/CMakeLists.txt
@@ -127,6 +127,7 @@ list(APPEND SOURCE_VISTA
condvar.c
runonce.c
srw.c
+ threadpool.c
utf8.c)
add_library(rtl_vista ${SOURCE_VISTA})
diff --git a/sdk/lib/rtl/threadpool.c b/sdk/lib/rtl/threadpool.c
new file mode 100644
index 00000000000..c6e40fc78f4
--- /dev/null
+++ b/sdk/lib/rtl/threadpool.c
@@ -0,0 +1,3357 @@
+/*
+ * Thread pooling
+ *
+ * Copyright (c) 2006 Robert Shearman
+ * Copyright (c) 2014-2016 Sebastian Lackner
+ *
+ * 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 <assert.h>
+#include <stdarg.h>
+#include <limits.h>
+
+#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);
+
+/*
+ * Old thread pooling API
+ */
+
+struct rtl_work_item
+{
+ PRTL_WORK_ITEM_ROUTINE function;
+ PVOID context;
+};
+
+#define EXPIRE_NEVER (~(ULONGLONG)0)
+#define TIMER_QUEUE_MAGIC 0x516d6954 /* TimQ */
+
+static RTL_CRITICAL_SECTION_DEBUG critsect_compl_debug;
+
+static struct
+{
+ HANDLE compl_port;
+ RTL_CRITICAL_SECTION threadpool_compl_cs;
+}
+old_threadpool =
+{
+ NULL, /* compl_port */
+ { &critsect_compl_debug, -1, 0, 0, 0, 0 }, /* threadpool_compl_cs */
+};
+
+static RTL_CRITICAL_SECTION_DEBUG critsect_compl_debug =
+{
+ 0, 0, &old_threadpool.threadpool_compl_cs,
+ { &critsect_compl_debug.ProcessLocksList,
&critsect_compl_debug.ProcessLocksList },
+ 0, 0, { (DWORD_PTR)(__FILE__ ": threadpool_compl_cs") }
+};
+
+struct timer_queue;
+struct queue_timer
+{
+ struct timer_queue *q;
+ struct list entry;
+ ULONG runcount; /* number of callbacks pending execution */
+ RTL_WAITORTIMERCALLBACKFUNC callback;
+ PVOID param;
+ DWORD period;
+ ULONG flags;
+ ULONGLONG expire;
+ BOOL destroy; /* timer should be deleted; once set, never unset */
+ HANDLE event; /* removal event */
+};
+
+struct timer_queue
+{
+ DWORD magic;
+ RTL_CRITICAL_SECTION cs;
+ struct list timers; /* sorted by expiration time */
+ BOOL quit; /* queue should be deleted; once set, never unset */
+ HANDLE event;
+ HANDLE thread;
+};
+
+/*
+ * Object-oriented thread pooling API
+ */
+
+#define THREADPOOL_WORKER_TIMEOUT 5000
+#define MAXIMUM_WAITQUEUE_OBJECTS (MAXIMUM_WAIT_OBJECTS - 1)
+
+/* internal threadpool representation */
+struct threadpool
+{
+ LONG refcount;
+ LONG objcount;
+ BOOL shutdown;
+ CRITICAL_SECTION cs;
+ /* Pools of work items, locked via .cs, order matches TP_CALLBACK_PRIORITY - high,
normal, low. */
+ struct list pools[3];
+ RTL_CONDITION_VARIABLE update_event;
+ /* information about worker threads, locked via .cs */
+ int max_workers;
+ int min_workers;
+ int num_workers;
+ int num_busy_workers;
+ HANDLE compl_port;
+ TP_POOL_STACK_INFORMATION stack_info;
+};
+
+enum threadpool_objtype
+{
+ TP_OBJECT_TYPE_SIMPLE,
+ TP_OBJECT_TYPE_WORK,
+ TP_OBJECT_TYPE_TIMER,
+ TP_OBJECT_TYPE_WAIT,
+ TP_OBJECT_TYPE_IO,
+};
+
+struct io_completion
+{
+ IO_STATUS_BLOCK iosb;
+ ULONG_PTR cvalue;
+};
+
+/* internal threadpool object representation */
+struct threadpool_object
+{
+ void *win32_callback; /* leave space for kernelbase to store win32
callback */
+ LONG refcount;
+ BOOL shutdown;
+ /* read-only information */
+ enum threadpool_objtype type;
+ struct threadpool *pool;
+ struct threadpool_group *group;
+ PVOID userdata;
+ PTP_CLEANUP_GROUP_CANCEL_CALLBACK group_cancel_callback;
+ PTP_SIMPLE_CALLBACK finalization_callback;
+ BOOL may_run_long;
+ HMODULE race_dll;
+ TP_CALLBACK_PRIORITY priority;
+ /* information about the group, locked via .group->cs */
+ struct list group_entry;
+ BOOL is_group_member;
+ /* information about the pool, locked via .pool->cs */
+ struct list pool_entry;
+ RTL_CONDITION_VARIABLE finished_event;
+ RTL_CONDITION_VARIABLE group_finished_event;
+ HANDLE completed_event;
+ LONG num_pending_callbacks;
+ LONG num_running_callbacks;
+ LONG num_associated_callbacks;
+ /* arguments for callback */
+ union
+ {
+ struct
+ {
+ PTP_SIMPLE_CALLBACK callback;
+ } simple;
+ struct
+ {
+ PTP_WORK_CALLBACK callback;
+ } work;
+ struct
+ {
+ PTP_TIMER_CALLBACK callback;
+ /* information about the timer, locked via timerqueue.cs */
+ BOOL timer_initialized;
+ BOOL timer_pending;
+ struct list timer_entry;
+ BOOL timer_set;
+ ULONGLONG timeout;
+ LONG period;
+ LONG window_length;
+ } timer;
+ struct
+ {
+ PTP_WAIT_CALLBACK callback;
+ LONG signaled;
+ /* information about the wait object, locked via waitqueue.cs */
+ struct waitqueue_bucket *bucket;
+ BOOL wait_pending;
+ struct list wait_entry;
+ ULONGLONG timeout;
+ HANDLE handle;
+ DWORD flags;
+ RTL_WAITORTIMERCALLBACKFUNC rtl_callback;
+ } wait;
+ struct
+ {
+ PTP_IO_CALLBACK callback;
+ /* locked via .pool->cs */
+ unsigned int pending_count, skipped_count, completion_count,
completion_max;
+ BOOL shutting_down;
+ struct io_completion *completions;
+ } io;
+ } u;
+};
+
+/* internal threadpool instance representation */
+struct threadpool_instance
+{
+ struct threadpool_object *object;
+ DWORD threadid;
+ BOOL associated;
+ BOOL may_run_long;
+ struct
+ {
+ CRITICAL_SECTION *critical_section;
+ HANDLE mutex;
+ HANDLE semaphore;
+ LONG semaphore_count;
+ HANDLE event;
+ HMODULE library;
+ } cleanup;
+};
+
+/* internal threadpool group representation */
+struct threadpool_group
+{
+ LONG refcount;
+ BOOL shutdown;
+ CRITICAL_SECTION cs;
+ /* list of group members, locked via .cs */
+ struct list members;
+};
+
+/* global timerqueue object */
+static RTL_CRITICAL_SECTION_DEBUG timerqueue_debug;
+
+static struct
+{
+ CRITICAL_SECTION cs;
+ LONG objcount;
+ BOOL thread_running;
+ struct list pending_timers;
+ RTL_CONDITION_VARIABLE update_event;
+}
+timerqueue =
+{
+ { &timerqueue_debug, -1, 0, 0, 0, 0 }, /* cs */
+ 0, /* objcount */
+ FALSE, /* thread_running */
+ LIST_INIT( timerqueue.pending_timers ), /* pending_timers */
+ RTL_CONDITION_VARIABLE_INIT /* update_event */
+};
+
+static RTL_CRITICAL_SECTION_DEBUG timerqueue_debug =
+{
+ 0, 0, &timerqueue.cs,
+ { &timerqueue_debug.ProcessLocksList, &timerqueue_debug.ProcessLocksList },
+ 0, 0, { (DWORD_PTR)(__FILE__ ": timerqueue.cs") }
+};
+
+/* global waitqueue object */
+static RTL_CRITICAL_SECTION_DEBUG waitqueue_debug;
+
+static struct
+{
+ CRITICAL_SECTION cs;
+ LONG num_buckets;
+ struct list buckets;
+}
+waitqueue =
+{
+ { &waitqueue_debug, -1, 0, 0, 0, 0 }, /* cs */
+ 0, /* num_buckets */
+ LIST_INIT( waitqueue.buckets ) /* buckets */
+};
+
+static RTL_CRITICAL_SECTION_DEBUG waitqueue_debug =
+{
+ 0, 0, &waitqueue.cs,
+ { &waitqueue_debug.ProcessLocksList, &waitqueue_debug.ProcessLocksList },
+ 0, 0, { (DWORD_PTR)(__FILE__ ": waitqueue.cs") }
+};
+
+struct waitqueue_bucket
+{
+ struct list bucket_entry;
+ LONG objcount;
+ struct list reserved;
+ struct list waiting;
+ HANDLE update_event;
+ BOOL alertable;
+};
+
+/* global I/O completion queue object */
+static RTL_CRITICAL_SECTION_DEBUG ioqueue_debug;
+
+static struct
+{
+ CRITICAL_SECTION cs;
+ LONG objcount;
+ BOOL thread_running;
+ HANDLE port;
+ RTL_CONDITION_VARIABLE update_event;
+}
+ioqueue =
+{
+ .cs = { &ioqueue_debug, -1, 0, 0, 0, 0 },
+};
+
+static RTL_CRITICAL_SECTION_DEBUG ioqueue_debug =
+{
+ 0, 0, &ioqueue.cs,
+ { &ioqueue_debug.ProcessLocksList, &ioqueue_debug.ProcessLocksList },
+ 0, 0, { (DWORD_PTR)(__FILE__ ": ioqueue.cs") }
+};
+
+static inline struct threadpool *impl_from_TP_POOL( TP_POOL *pool )
+{
+ return (struct threadpool *)pool;
+}
+
+static inline struct threadpool_object *impl_from_TP_WORK( TP_WORK *work )
+{
+ struct threadpool_object *object = (struct threadpool_object *)work;
+ assert( object->type == TP_OBJECT_TYPE_WORK );
+ return object;
+}
+
+static inline struct threadpool_object *impl_from_TP_TIMER( TP_TIMER *timer )
+{
+ struct threadpool_object *object = (struct threadpool_object *)timer;
+ assert( object->type == TP_OBJECT_TYPE_TIMER );
+ return object;
+}
+
+static inline struct threadpool_object *impl_from_TP_WAIT( TP_WAIT *wait )
+{
+ struct threadpool_object *object = (struct threadpool_object *)wait;
+ assert( object->type == TP_OBJECT_TYPE_WAIT );
+ return object;
+}
+
+static inline struct threadpool_object *impl_from_TP_IO( TP_IO *io )
+{
+ struct threadpool_object *object = (struct threadpool_object *)io;
+ assert( object->type == TP_OBJECT_TYPE_IO );
+ return object;
+}
+
+static inline struct threadpool_group *impl_from_TP_CLEANUP_GROUP( TP_CLEANUP_GROUP
*group )
+{
+ return (struct threadpool_group *)group;
+}
+
+static inline struct threadpool_instance *impl_from_TP_CALLBACK_INSTANCE(
TP_CALLBACK_INSTANCE *instance )
+{
+ return (struct threadpool_instance *)instance;
+}
+
+static void CALLBACK threadpool_worker_proc( void *param );
+static void tp_object_submit( struct threadpool_object *object, BOOL signaled );
+static void tp_object_execute( struct threadpool_object *object, BOOL wait_thread );
+static void tp_object_prepare_shutdown( struct threadpool_object *object );
+static BOOL tp_object_release( struct threadpool_object *object );
+static struct threadpool *default_threadpool = NULL;
+
+static BOOL array_reserve(void **elements, unsigned int *capacity, unsigned int count,
unsigned int size)
+{
+ unsigned int new_capacity, max_capacity;
+ void *new_elements;
+
+ if (count <= *capacity)
+ return TRUE;
+
+ max_capacity = ~(SIZE_T)0 / size;
+ if (count > max_capacity)
+ return FALSE;
+
+ new_capacity = max(4, *capacity);
+ while (new_capacity < count && new_capacity <= max_capacity / 2)
+ new_capacity *= 2;
+ if (new_capacity < count)
+ new_capacity = max_capacity;
+
+ if (!(new_elements = RtlReAllocateHeap( GetProcessHeap(), 0, *elements, new_capacity
* size )))
+ return FALSE;
+
+ *elements = new_elements;
+ *capacity = new_capacity;
+
+ return TRUE;
+}
+
+static void set_thread_name(const WCHAR *name)
+{
+ THREAD_NAME_INFORMATION info;
+
+ RtlInitUnicodeString(&info.ThreadName, name);
+ NtSetInformationThread(GetCurrentThread(), ThreadNameInformation, &info,
sizeof(info));
+}
+
+static void CALLBACK process_rtl_work_item( TP_CALLBACK_INSTANCE *instance, void
*userdata )
+{
+ struct rtl_work_item *item = userdata;
+
+ TRACE("executing %p(%p)\n", item->function, item->context);
+ item->function( item->context );
+
+ RtlFreeHeap( GetProcessHeap(), 0, item );
+}
+
+/***********************************************************************
+ * 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 )
+{
+ TP_CALLBACK_ENVIRON environment;
+ struct rtl_work_item *item;
+ NTSTATUS status;
+
+ TRACE( "%p %p %lu\n", function, context, flags );
+
+ item = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*item) );
+ if (!item)
+ return STATUS_NO_MEMORY;
+
+ memset( &environment, 0, sizeof(environment) );
+ environment.Version = 1;
+ environment.u.s.LongFunction = (flags & WT_EXECUTELONGFUNCTION) != 0;
+ environment.u.s.Persistent = (flags & WT_EXECUTEINPERSISTENTTHREAD) != 0;
+
+ item->function = function;
+ item->context = context;
+
+ status = TpSimpleTryPost( process_rtl_work_item, item, &environment );
+ if (status) RtlFreeHeap( GetProcessHeap(), 0, item );
+ return status;
+}
+
+/***********************************************************************
+ * iocp_poller - get completion events and run callbacks
+ */
+static DWORD CALLBACK iocp_poller(LPVOID Arg)
+{
+ HANDLE cport = Arg;
+
+ while( TRUE )
+ {
+ PRTL_OVERLAPPED_COMPLETION_ROUTINE callback;
+ LPVOID overlapped;
+ IO_STATUS_BLOCK iosb;
+ NTSTATUS res = NtRemoveIoCompletion( cport, (PULONG_PTR)&callback,
(PULONG_PTR)&overlapped, &iosb, NULL );
+ if (res)
+ {
+ ERR("NtRemoveIoCompletion failed: 0x%lx\n", res);
+ }
+ else
+ {
+ DWORD transferred = 0;
+ DWORD err = 0;
+
+ if (iosb.Status == STATUS_SUCCESS)
+ transferred = iosb.Information;
+ else
+ err = RtlNtStatusToDosError(iosb.Status);
+
+ callback( err, transferred, overlapped );
+ }
+ }
+ return 0;
+}
+
+/***********************************************************************
+ * 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%lx\n", Flags);
+
+ if (!old_threadpool.compl_port)
+ {
+ NTSTATUS res = STATUS_SUCCESS;
+
+ RtlEnterCriticalSection(&old_threadpool.threadpool_compl_cs);
+ if (!old_threadpool.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, cport, WT_EXECUTEDEFAULT );
+ if (!res)
+ old_threadpool.compl_port = cport;
+ else
+ NtClose( cport );
+ }
+ }
+ RtlLeaveCriticalSection(&old_threadpool.threadpool_compl_cs);
+ if (res) return res;
+ }
+
+ info.CompletionPort = old_threadpool.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;
+}
+
+
+/************************** Timer Queue Impl **************************/
+
+static void queue_remove_timer(struct queue_timer *t)
+{
+ /* We MUST hold the queue cs while calling this function. This ensures
+ that we cannot queue another callback for this timer. The runcount
+ being zero makes sure we don't have any already queued. */
+ struct timer_queue *q = t->q;
+
+ assert(t->runcount == 0);
+ assert(t->destroy);
+
+ list_remove(&t->entry);
+ if (t->event)
+ NtSetEvent(t->event, NULL);
+ RtlFreeHeap(GetProcessHeap(), 0, t);
+
+ if (q->quit && list_empty(&q->timers))
+ NtSetEvent(q->event, NULL);
+}
+
+static void timer_cleanup_callback(struct queue_timer *t)
+{
+ struct timer_queue *q = t->q;
+ RtlEnterCriticalSection(&q->cs);
+
+ assert(0 < t->runcount);
+ --t->runcount;
+
+ if (t->destroy && t->runcount == 0)
+ queue_remove_timer(t);
+
+ RtlLeaveCriticalSection(&q->cs);
+}
+
+static DWORD WINAPI timer_callback_wrapper(LPVOID p)
+{
+ struct queue_timer *t = p;
+ t->callback(t->param, TRUE);
+ timer_cleanup_callback(t);
+ return 0;
+}
+
+static inline ULONGLONG queue_current_time(void)
+{
+ LARGE_INTEGER now, freq;
+ NtQueryPerformanceCounter(&now, &freq);
+ return now.QuadPart * 1000 / freq.QuadPart;
+}
+
+static void queue_add_timer(struct queue_timer *t, ULONGLONG time,
+ BOOL set_event)
+{
+ /* We MUST hold the queue cs while calling this function. */
+ struct timer_queue *q = t->q;
+ struct list *ptr = &q->timers;
+
+ assert(!q->quit || (t->destroy && time == EXPIRE_NEVER));
+
+ if (time != EXPIRE_NEVER)
+ LIST_FOR_EACH(ptr, &q->timers)
+ {
+ struct queue_timer *cur = LIST_ENTRY(ptr, struct queue_timer, entry);
+ if (time < cur->expire)
+ break;
+ }
+ list_add_before(ptr, &t->entry);
+
+ t->expire = time;
+
+ /* If we insert at the head of the list, we need to expire sooner
+ than expected. */
+ if (set_event && &t->entry == list_head(&q->timers))
+ NtSetEvent(q->event, NULL);
+}
+
+static inline void queue_move_timer(struct queue_timer *t, ULONGLONG time,
+ BOOL set_event)
+{
+ /* We MUST hold the queue cs while calling this function. */
+ list_remove(&t->entry);
+ queue_add_timer(t, time, set_event);
+}
+
+static void queue_timer_expire(struct timer_queue *q)
+{
+ struct queue_timer *t = NULL;
+
+ RtlEnterCriticalSection(&q->cs);
+ if (list_head(&q->timers))
+ {
+ ULONGLONG now, next;
+ t = LIST_ENTRY(list_head(&q->timers), struct queue_timer, entry);
+ if (!t->destroy && t->expire <= ((now = queue_current_time())))
+ {
+ ++t->runcount;
+ if (t->period)
+ {
+ next = t->expire + t->period;
+ /* avoid trigger cascade if overloaded / hibernated */
+ if (next < now)
+ next = now + t->period;
+ }
+ else
+ next = EXPIRE_NEVER;
+ queue_move_timer(t, next, FALSE);
+ }
+ else
+ t = NULL;
+ }
+ RtlLeaveCriticalSection(&q->cs);
+
+ if (t)
+ {
+ if (t->flags & WT_EXECUTEINTIMERTHREAD)
+ timer_callback_wrapper(t);
+ else
+ {
+ ULONG flags
+ = (t->flags
+ & (WT_EXECUTEINIOTHREAD | WT_EXECUTEINPERSISTENTTHREAD
+ | WT_EXECUTELONGFUNCTION | WT_TRANSFER_IMPERSONATION));
+ NTSTATUS status = RtlQueueWorkItem(timer_callback_wrapper, t, flags);
+ if (status != STATUS_SUCCESS)
+ timer_cleanup_callback(t);
+ }
+ }
+}
+
+static ULONG queue_get_timeout(struct timer_queue *q)
+{
+ struct queue_timer *t;
+ ULONG timeout = INFINITE;
+
+ RtlEnterCriticalSection(&q->cs);
+ if (list_head(&q->timers))
+ {
+ t = LIST_ENTRY(list_head(&q->timers), struct queue_timer, entry);
+ assert(!t->destroy || t->expire == EXPIRE_NEVER);
+
+ if (t->expire != EXPIRE_NEVER)
+ {
+ ULONGLONG time = queue_current_time();
+ timeout = t->expire < time ? 0 : t->expire - time;
+ }
+ }
+ RtlLeaveCriticalSection(&q->cs);
+
+ return timeout;
+}
+
+static void WINAPI timer_queue_thread_proc(LPVOID p)
+{
+ struct timer_queue *q = p;
+ ULONG timeout_ms;
+
+ set_thread_name(L"wine_threadpool_timer_queue");
+ timeout_ms = INFINITE;
+ for (;;)
+ {
+ LARGE_INTEGER timeout;
+ NTSTATUS status;
+ BOOL done = FALSE;
+
+ status = NtWaitForSingleObject(
+ q->event, FALSE, get_nt_timeout(&timeout, timeout_ms));
+
+ if (status == STATUS_WAIT_0)
+ {
+ /* There are two possible ways to trigger the event. Either
+ we are quitting and the last timer got removed, or a new
+ timer got put at the head of the list so we need to adjust
+ our timeout. */
+ RtlEnterCriticalSection(&q->cs);
+ if (q->quit && list_empty(&q->timers))
+ done = TRUE;
+ RtlLeaveCriticalSection(&q->cs);
+ }
+ else if (status == STATUS_TIMEOUT)
+ queue_timer_expire(q);
+
+ if (done)
+ break;
+
+ timeout_ms = queue_get_timeout(q);
+ }
+
+ NtClose(q->event);
+ RtlDeleteCriticalSection(&q->cs);
+ q->magic = 0;
+ RtlFreeHeap(GetProcessHeap(), 0, q);
+ RtlExitUserThread( 0 );
+}
+
+static void queue_destroy_timer(struct queue_timer *t)
+{
+ /* We MUST hold the queue cs while calling this function. */
+ t->destroy = TRUE;
+ if (t->runcount == 0)
+ /* Ensure a timer is promptly removed. If callbacks are pending,
+ it will be removed after the last one finishes by the callback
+ cleanup wrapper. */
+ queue_remove_timer(t);
+ else
+ /* Make sure no destroyed timer masks an active timer at the head
+ of the sorted list. */
+ queue_move_timer(t, EXPIRE_NEVER, FALSE);
+}
+
+/***********************************************************************
+ * RtlCreateTimerQueue (NTDLL.@)
+ *
+ * Creates a timer queue object and returns a handle to it.
+ *
+ * PARAMS
+ * NewTimerQueue [O] The newly created queue.
+ *
+ * RETURNS
+ * Success: STATUS_SUCCESS.
+ * Failure: Any NTSTATUS code.
+ */
+NTSTATUS WINAPI RtlCreateTimerQueue(PHANDLE NewTimerQueue)
+{
+ NTSTATUS status;
+ struct timer_queue *q = RtlAllocateHeap(GetProcessHeap(), 0, sizeof *q);
+ if (!q)
+ return STATUS_NO_MEMORY;
+
+ RtlInitializeCriticalSection(&q->cs);
+ list_init(&q->timers);
+ q->quit = FALSE;
+ q->magic = TIMER_QUEUE_MAGIC;
+ status = NtCreateEvent(&q->event, EVENT_ALL_ACCESS, NULL,
SynchronizationEvent, FALSE);
+ if (status != STATUS_SUCCESS)
+ {
+ RtlFreeHeap(GetProcessHeap(), 0, q);
+ return status;
+ }
+ status = RtlCreateUserThread(GetCurrentProcess(), NULL, FALSE, 0, 0, 0,
+ timer_queue_thread_proc, q, &q->thread, NULL);
+ if (status != STATUS_SUCCESS)
+ {
+ NtClose(q->event);
+ RtlFreeHeap(GetProcessHeap(), 0, q);
+ return status;
+ }
+
+ *NewTimerQueue = q;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * RtlDeleteTimerQueueEx (NTDLL.@)
+ *
+ * Deletes a timer queue object.
+ *
+ * PARAMS
+ * TimerQueue [I] The timer queue to destroy.
+ * CompletionEvent [I] If NULL, return immediately. If INVALID_HANDLE_VALUE,
+ * wait until all timers are finished firing before
+ * returning. Otherwise, return immediately and set the
+ * event when all timers are done.
+ *
+ * RETURNS
+ * Success: STATUS_SUCCESS if synchronous, STATUS_PENDING if not.
+ * Failure: Any NTSTATUS code.
+ */
+NTSTATUS WINAPI RtlDeleteTimerQueueEx(HANDLE TimerQueue, HANDLE CompletionEvent)
+{
+ struct timer_queue *q = TimerQueue;
+ struct queue_timer *t, *temp;
+ HANDLE thread;
+ NTSTATUS status;
+
+ if (!q || q->magic != TIMER_QUEUE_MAGIC)
+ return STATUS_INVALID_HANDLE;
+
+ thread = q->thread;
+
+ RtlEnterCriticalSection(&q->cs);
+ q->quit = TRUE;
+ if (list_head(&q->timers))
+ /* When the last timer is removed, it will signal the timer thread to
+ exit... */
+ LIST_FOR_EACH_ENTRY_SAFE(t, temp, &q->timers, struct queue_timer, entry)
+ queue_destroy_timer(t);
+ else
+ /* However if we have none, we must do it ourselves. */
+ NtSetEvent(q->event, NULL);
+ RtlLeaveCriticalSection(&q->cs);
+
+ if (CompletionEvent == INVALID_HANDLE_VALUE)
+ {
+ NtWaitForSingleObject(thread, FALSE, NULL);
+ status = STATUS_SUCCESS;
+ }
+ else
+ {
+ if (CompletionEvent)
+ {
+ FIXME("asynchronous return on completion event unimplemented\n");
+ NtWaitForSingleObject(thread, FALSE, NULL);
+ NtSetEvent(CompletionEvent, NULL);
+ }
+ status = STATUS_PENDING;
+ }
+
+ NtClose(thread);
+ return status;
+}
+
+static struct timer_queue *get_timer_queue(HANDLE TimerQueue)
+{
+ static struct timer_queue *default_timer_queue;
+
+ if (TimerQueue)
+ return TimerQueue;
+ else
+ {
+ if (!default_timer_queue)
+ {
+ HANDLE q;
+ NTSTATUS status = RtlCreateTimerQueue(&q);
+ if (status == STATUS_SUCCESS)
+ {
+ PVOID p = InterlockedCompareExchangePointer( (void **)
&default_timer_queue, q, NULL );
+ if (p)
+ /* Got beat to the punch. */
+ RtlDeleteTimerQueueEx(q, NULL);
+ }
+ }
+ return default_timer_queue;
+ }
+}
+
+/***********************************************************************
+ * RtlCreateTimer (NTDLL.@)
+ *
+ * Creates a new timer associated with the given queue.
+ *
+ * PARAMS
+ * TimerQueue [I] The queue to hold the timer.
+ * NewTimer [O] The newly created timer.
+ * Callback [I] The callback to fire.
+ * Parameter [I] The argument for the callback.
+ * DueTime [I] The delay, in milliseconds, before first firing the
+ * timer.
+ * Period [I] The period, in milliseconds, at which to fire the timer
+ * after the first callback. If zero, the timer will only
+ * fire once. It still needs to be deleted with
+ * RtlDeleteTimer.
+ * Flags [I] Flags controlling the execution of the callback. In
+ * addition to the WT_* thread pool flags (see
+ * RtlQueueWorkItem), WT_EXECUTEINTIMERTHREAD and
+ * WT_EXECUTEONLYONCE are supported.
+ *
+ * RETURNS
+ * Success: STATUS_SUCCESS.
+ * Failure: Any NTSTATUS code.
+ */
+NTSTATUS WINAPI RtlCreateTimer(HANDLE TimerQueue, HANDLE *NewTimer,
+ RTL_WAITORTIMERCALLBACKFUNC Callback,
+ PVOID Parameter, DWORD DueTime, DWORD Period,
+ ULONG Flags)
+{
+ NTSTATUS status;
+ struct queue_timer *t;
+ struct timer_queue *q = get_timer_queue(TimerQueue);
+
+ if (!q) return STATUS_NO_MEMORY;
+ if (q->magic != TIMER_QUEUE_MAGIC) return STATUS_INVALID_HANDLE;
+
+ t = RtlAllocateHeap(GetProcessHeap(), 0, sizeof *t);
+ if (!t)
+ return STATUS_NO_MEMORY;
+
+ t->q = q;
+ t->runcount = 0;
+ t->callback = Callback;
+ t->param = Parameter;
+ t->period = Period;
+ t->flags = Flags;
+ t->destroy = FALSE;
+ t->event = NULL;
+
+ status = STATUS_SUCCESS;
+ RtlEnterCriticalSection(&q->cs);
+ if (q->quit)
+ status = STATUS_INVALID_HANDLE;
+ else
+ queue_add_timer(t, queue_current_time() + DueTime, TRUE);
+ RtlLeaveCriticalSection(&q->cs);
+
+ if (status == STATUS_SUCCESS)
+ *NewTimer = t;
+ else
+ RtlFreeHeap(GetProcessHeap(), 0, t);
+
+ return status;
+}
+
+/***********************************************************************
+ * RtlUpdateTimer (NTDLL.@)
+ *
+ * Changes the time at which a timer expires.
+ *
+ * PARAMS
+ * TimerQueue [I] The queue that holds the timer.
+ * Timer [I] The timer to update.
+ * DueTime [I] The delay, in milliseconds, before next firing the timer.
+ * Period [I] The period, in milliseconds, at which to fire the timer
+ * after the first callback. If zero, the timer will not
+ * refire once. It still needs to be deleted with
+ * RtlDeleteTimer.
+ *
+ * RETURNS
+ * Success: STATUS_SUCCESS.
+ * Failure: Any NTSTATUS code.
+ */
+NTSTATUS WINAPI RtlUpdateTimer(HANDLE TimerQueue, HANDLE Timer,
+ DWORD DueTime, DWORD Period)
+{
+ struct queue_timer *t = Timer;
+ struct timer_queue *q = t->q;
+
+ RtlEnterCriticalSection(&q->cs);
+ /* Can't change a timer if it was once-only or destroyed. */
+ if (t->expire != EXPIRE_NEVER)
+ {
+ t->period = Period;
+ queue_move_timer(t, queue_current_time() + DueTime, TRUE);
+ }
+ RtlLeaveCriticalSection(&q->cs);
+
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * RtlDeleteTimer (NTDLL.@)
+ *
+ * Cancels a timer-queue timer.
+ *
+ * PARAMS
+ * TimerQueue [I] The queue that holds the timer.
+ * Timer [I] The timer to update.
+ * CompletionEvent [I] If NULL, return immediately. If INVALID_HANDLE_VALUE,
+ * wait until the timer is finished firing all pending
+ * callbacks before returning. Otherwise, return
+ * immediately and set the timer is done.
+ *
+ * RETURNS
+ * Success: STATUS_SUCCESS if the timer is done, STATUS_PENDING if not,
+ or if the completion event is NULL.
+ * Failure: Any NTSTATUS code.
+ */
+NTSTATUS WINAPI RtlDeleteTimer(HANDLE TimerQueue, HANDLE Timer,
+ HANDLE CompletionEvent)
+{
+ struct queue_timer *t = Timer;
+ struct timer_queue *q;
+ NTSTATUS status = STATUS_PENDING;
+ HANDLE event = NULL;
+
+ if (!Timer)
+ return STATUS_INVALID_PARAMETER_1;
+ q = t->q;
+ if (CompletionEvent == INVALID_HANDLE_VALUE)
+ {
+ status = NtCreateEvent(&event, EVENT_ALL_ACCESS, NULL, SynchronizationEvent,
FALSE);
+ if (status == STATUS_SUCCESS)
+ status = STATUS_PENDING;
+ }
+ else if (CompletionEvent)
+ event = CompletionEvent;
+
+ RtlEnterCriticalSection(&q->cs);
+ t->event = event;
+ if (t->runcount == 0 && event)
+ status = STATUS_SUCCESS;
+ queue_destroy_timer(t);
+ RtlLeaveCriticalSection(&q->cs);
+
+ if (CompletionEvent == INVALID_HANDLE_VALUE && event)
+ {
+ if (status == STATUS_PENDING)
+ {
+ NtWaitForSingleObject(event, FALSE, NULL);
+ status = STATUS_SUCCESS;
+ }
+ NtClose(event);
+ }
+
+ return status;
+}
+
+/***********************************************************************
+ * timerqueue_thread_proc (internal)
+ */
+static void CALLBACK timerqueue_thread_proc( void *param )
+{
+ ULONGLONG timeout_lower, timeout_upper, new_timeout;
+ struct threadpool_object *other_timer;
+ LARGE_INTEGER now, timeout;
+ struct list *ptr;
+
+ TRACE( "starting timer queue thread\n" );
+ set_thread_name(L"wine_threadpool_timerqueue");
+
+ RtlEnterCriticalSection( &timerqueue.cs );
+ for (;;)
+ {
+ NtQuerySystemTime( &now );
+
+ /* Check for expired timers. */
+ while ((ptr = list_head( &timerqueue.pending_timers )))
+ {
+ struct threadpool_object *timer = LIST_ENTRY( ptr, struct threadpool_object,
u.timer.timer_entry );
+ assert( timer->type == TP_OBJECT_TYPE_TIMER );
+ assert( timer->u.timer.timer_pending );
+ if (timer->u.timer.timeout > now.QuadPart)
+ break;
+
+ /* Queue a new callback in one of the worker threads. */
+ list_remove( &timer->u.timer.timer_entry );
+ timer->u.timer.timer_pending = FALSE;
+ tp_object_submit( timer, FALSE );
+
+ /* Insert the timer back into the queue, except it's marked for shutdown.
*/
+ if (timer->u.timer.period && !timer->shutdown)
+ {
+ timer->u.timer.timeout += (ULONGLONG)timer->u.timer.period *
10000;
+ if (timer->u.timer.timeout <= now.QuadPart)
+ timer->u.timer.timeout = now.QuadPart + 1;
+
+ LIST_FOR_EACH_ENTRY( other_timer, &timerqueue.pending_timers,
+ struct threadpool_object, u.timer.timer_entry )
+ {
+ assert( other_timer->type == TP_OBJECT_TYPE_TIMER );
+ if (timer->u.timer.timeout < other_timer->u.timer.timeout)
+ break;
+ }
+ list_add_before( &other_timer->u.timer.timer_entry,
&timer->u.timer.timer_entry );
+ timer->u.timer.timer_pending = TRUE;
+ }
+ }
+
+ timeout_lower = timeout_upper = MAXLONGLONG;
+
+ /* Determine next timeout and use the window length to optimize wakeup times. */
+ LIST_FOR_EACH_ENTRY( other_timer, &timerqueue.pending_timers,
+ struct threadpool_object, u.timer.timer_entry )
+ {
+ assert( other_timer->type == TP_OBJECT_TYPE_TIMER );
+ if (other_timer->u.timer.timeout >= timeout_upper)
+ break;
+
+ timeout_lower = other_timer->u.timer.timeout;
+ new_timeout = timeout_lower +
(ULONGLONG)other_timer->u.timer.window_length * 10000;
+ if (new_timeout < timeout_upper)
+ timeout_upper = new_timeout;
+ }
+
+ /* Wait for timer update events or until the next timer expires. */
+ if (timerqueue.objcount)
+ {
+ timeout.QuadPart = timeout_lower;
+ RtlSleepConditionVariableCS( &timerqueue.update_event,
&timerqueue.cs, &timeout );
+ continue;
+ }
+
+ /* All timers have been destroyed, if no new timers are created
+ * within some amount of time, then we can shutdown this thread. */
+ timeout.QuadPart = (ULONGLONG)THREADPOOL_WORKER_TIMEOUT * -10000;
+ if (RtlSleepConditionVariableCS( &timerqueue.update_event,
&timerqueue.cs,
+ &timeout ) == STATUS_TIMEOUT && !timerqueue.objcount)
+ {
+ break;
+ }
+ }
+
+ timerqueue.thread_running = FALSE;
+ RtlLeaveCriticalSection( &timerqueue.cs );
+
+ TRACE( "terminating timer queue thread\n" );
+ RtlExitUserThread( 0 );
+}
+
+/***********************************************************************
+ * tp_new_worker_thread (internal)
+ *
+ * Create and account a new worker thread for the desired pool.
+ */
+static NTSTATUS tp_new_worker_thread( struct threadpool *pool )
+{
+ HANDLE thread;
+ NTSTATUS status;
+
+ status = RtlCreateUserThread( GetCurrentProcess(), NULL, FALSE, 0,
+ pool->stack_info.StackReserve,
pool->stack_info.StackCommit,
+ threadpool_worker_proc, pool, &thread, NULL );
+ if (status == STATUS_SUCCESS)
+ {
+ InterlockedIncrement( &pool->refcount );
+ pool->num_workers++;
+ NtClose( thread );
+ }
+ return status;
+}
+
+/***********************************************************************
+ * tp_timerqueue_lock (internal)
+ *
+ * Acquires a lock on the global timerqueue. When the lock is acquired
+ * successfully, it is guaranteed that the timer thread is running.
+ */
+static NTSTATUS tp_timerqueue_lock( struct threadpool_object *timer )
+{
+ NTSTATUS status = STATUS_SUCCESS;
+ assert( timer->type == TP_OBJECT_TYPE_TIMER );
+
+ timer->u.timer.timer_initialized = FALSE;
+ timer->u.timer.timer_pending = FALSE;
+ timer->u.timer.timer_set = FALSE;
+ timer->u.timer.timeout = 0;
+ timer->u.timer.period = 0;
+ timer->u.timer.window_length = 0;
+
+ RtlEnterCriticalSection( &timerqueue.cs );
+
+ /* Make sure that the timerqueue thread is running. */
+ if (!timerqueue.thread_running)
+ {
+ HANDLE thread;
+ status = RtlCreateUserThread( GetCurrentProcess(), NULL, FALSE, 0, 0, 0,
+ timerqueue_thread_proc, NULL, &thread, NULL );
+ if (status == STATUS_SUCCESS)
+ {
+ timerqueue.thread_running = TRUE;
+ NtClose( thread );
+ }
+ }
+
+ if (status == STATUS_SUCCESS)
+ {
+ timer->u.timer.timer_initialized = TRUE;
+ timerqueue.objcount++;
+ }
+
+ RtlLeaveCriticalSection( &timerqueue.cs );
+ return status;
+}
+
+/***********************************************************************
+ * tp_timerqueue_unlock (internal)
+ *
+ * Releases a lock on the global timerqueue.
+ */
+static void tp_timerqueue_unlock( struct threadpool_object *timer )
+{
+ assert( timer->type == TP_OBJECT_TYPE_TIMER );
+
+ RtlEnterCriticalSection( &timerqueue.cs );
+ if (timer->u.timer.timer_initialized)
+ {
+ /* If timer was pending, remove it. */
+ if (timer->u.timer.timer_pending)
+ {
+ list_remove( &timer->u.timer.timer_entry );
+ timer->u.timer.timer_pending = FALSE;
+ }
+
+ /* If the last timer object was destroyed, then wake up the thread. */
+ if (!--timerqueue.objcount)
+ {
+ assert( list_empty( &timerqueue.pending_timers ) );
+ RtlWakeAllConditionVariable( &timerqueue.update_event );
+ }
+
+ timer->u.timer.timer_initialized = FALSE;
+ }
+ RtlLeaveCriticalSection( &timerqueue.cs );
+}
+
+/***********************************************************************
+ * waitqueue_thread_proc (internal)
+ */
+static void CALLBACK waitqueue_thread_proc( void *param )
+{
+ struct threadpool_object *objects[MAXIMUM_WAITQUEUE_OBJECTS];
+ HANDLE handles[MAXIMUM_WAITQUEUE_OBJECTS + 1];
+ struct waitqueue_bucket *bucket = param;
+ struct threadpool_object *wait, *next;
+ LARGE_INTEGER now, timeout;
+ DWORD num_handles;
+ NTSTATUS status;
+
+ TRACE( "starting wait queue thread\n" );
+ set_thread_name(L"wine_threadpool_waitqueue");
+
+ RtlEnterCriticalSection( &waitqueue.cs );
+
+ for (;;)
+ {
+ NtQuerySystemTime( &now );
+ timeout.QuadPart = MAXLONGLONG;
+ num_handles = 0;
+
+ LIST_FOR_EACH_ENTRY_SAFE( wait, next, &bucket->waiting, struct
threadpool_object,
+ u.wait.wait_entry )
+ {
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+ if (wait->u.wait.timeout <= now.QuadPart)
+ {
+ /* Wait object timed out. */
+ if ((wait->u.wait.flags & WT_EXECUTEONLYONCE))
+ {
+ list_remove( &wait->u.wait.wait_entry );
+ list_add_tail( &bucket->reserved,
&wait->u.wait.wait_entry );
+ }
+ if ((wait->u.wait.flags & (WT_EXECUTEINWAITTHREAD |
WT_EXECUTEINIOTHREAD)))
+ {
+ InterlockedIncrement( &wait->refcount );
+ wait->num_pending_callbacks++;
+ RtlEnterCriticalSection( &wait->pool->cs );
+ tp_object_execute( wait, TRUE );
+ RtlLeaveCriticalSection( &wait->pool->cs );
+ tp_object_release( wait );
+ }
+ else tp_object_submit( wait, FALSE );
+ }
+ else
+ {
+ if (wait->u.wait.timeout < timeout.QuadPart)
+ timeout.QuadPart = wait->u.wait.timeout;
+
+ assert( num_handles < MAXIMUM_WAITQUEUE_OBJECTS );
+ InterlockedIncrement( &wait->refcount );
+ objects[num_handles] = wait;
+ handles[num_handles] = wait->u.wait.handle;
+ num_handles++;
+ }
+ }
+
+ if (!bucket->objcount)
+ {
+ /* All wait objects have been destroyed, if no new wait objects are created
+ * within some amount of time, then we can shutdown this thread. */
+ assert( num_handles == 0 );
+ RtlLeaveCriticalSection( &waitqueue.cs );
+ timeout.QuadPart = (ULONGLONG)THREADPOOL_WORKER_TIMEOUT * -10000;
+ status = NtWaitForMultipleObjects( 1, &bucket->update_event, TRUE,
bucket->alertable, &timeout );
+ RtlEnterCriticalSection( &waitqueue.cs );
+
+ if (status == STATUS_TIMEOUT && !bucket->objcount)
+ break;
+ }
+ else
+ {
+ handles[num_handles] = bucket->update_event;
+ RtlLeaveCriticalSection( &waitqueue.cs );
+ status = NtWaitForMultipleObjects( num_handles + 1, handles, TRUE,
bucket->alertable, &timeout );
+ RtlEnterCriticalSection( &waitqueue.cs );
+
+ if (status >= STATUS_WAIT_0 && status < STATUS_WAIT_0 +
num_handles)
+ {
+ wait = objects[status - STATUS_WAIT_0];
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+ if (wait->u.wait.bucket)
+ {
+ /* Wait object signaled. */
+ assert( wait->u.wait.bucket == bucket );
+ if ((wait->u.wait.flags & WT_EXECUTEONLYONCE))
+ {
+ list_remove( &wait->u.wait.wait_entry );
+ list_add_tail( &bucket->reserved,
&wait->u.wait.wait_entry );
+ }
+ if ((wait->u.wait.flags & (WT_EXECUTEINWAITTHREAD |
WT_EXECUTEINIOTHREAD)))
+ {
+ wait->u.wait.signaled++;
+ wait->num_pending_callbacks++;
+ RtlEnterCriticalSection( &wait->pool->cs );
+ tp_object_execute( wait, TRUE );
+ RtlLeaveCriticalSection( &wait->pool->cs );
+ }
+ else tp_object_submit( wait, TRUE );
+ }
+ else
+ WARN("wait object %p triggered while object was
destroyed\n", wait);
+ }
+
+ /* Release temporary references to wait objects. */
+ while (num_handles)
+ {
+ wait = objects[--num_handles];
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+ tp_object_release( wait );
+ }
+ }
+
+ /* Try to merge bucket with other threads. */
+ if (waitqueue.num_buckets > 1 && bucket->objcount &&
+ bucket->objcount <= MAXIMUM_WAITQUEUE_OBJECTS * 1 / 3)
+ {
+ struct waitqueue_bucket *other_bucket;
+ LIST_FOR_EACH_ENTRY( other_bucket, &waitqueue.buckets, struct
waitqueue_bucket, bucket_entry )
+ {
+ if (other_bucket != bucket && other_bucket->objcount
&& other_bucket->alertable == bucket->alertable &&
+ other_bucket->objcount + bucket->objcount <=
MAXIMUM_WAITQUEUE_OBJECTS * 2 / 3)
+ {
+ other_bucket->objcount += bucket->objcount;
+ bucket->objcount = 0;
+
+ /* Update reserved list. */
+ LIST_FOR_EACH_ENTRY( wait, &bucket->reserved, struct
threadpool_object, u.wait.wait_entry )
+ {
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+ wait->u.wait.bucket = other_bucket;
+ }
+ list_move_tail( &other_bucket->reserved,
&bucket->reserved );
+
+ /* Update waiting list. */
+ LIST_FOR_EACH_ENTRY( wait, &bucket->waiting, struct
threadpool_object, u.wait.wait_entry )
+ {
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+ wait->u.wait.bucket = other_bucket;
+ }
+ list_move_tail( &other_bucket->waiting,
&bucket->waiting );
+
+ /* Move bucket to the end, to keep the probability of
+ * newly added wait objects as small as possible. */
+ list_remove( &bucket->bucket_entry );
+ list_add_tail( &waitqueue.buckets, &bucket->bucket_entry
);
+
+ NtSetEvent( other_bucket->update_event, NULL );
+ break;
+ }
+ }
+ }
+ }
+
+ /* Remove this bucket from the list. */
+ list_remove( &bucket->bucket_entry );
+ if (!--waitqueue.num_buckets)
+ assert( list_empty( &waitqueue.buckets ) );
+
+ RtlLeaveCriticalSection( &waitqueue.cs );
+
+ TRACE( "terminating wait queue thread\n" );
+
+ assert( bucket->objcount == 0 );
+ assert( list_empty( &bucket->reserved ) );
+ assert( list_empty( &bucket->waiting ) );
+ NtClose( bucket->update_event );
+
+ RtlFreeHeap( GetProcessHeap(), 0, bucket );
+ RtlExitUserThread( 0 );
+}
+
+/***********************************************************************
+ * tp_waitqueue_lock (internal)
+ */
+static NTSTATUS tp_waitqueue_lock( struct threadpool_object *wait )
+{
+ struct waitqueue_bucket *bucket;
+ NTSTATUS status;
+ HANDLE thread;
+ BOOL alertable = (wait->u.wait.flags & WT_EXECUTEINIOTHREAD) != 0;
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+
+ wait->u.wait.signaled = 0;
+ wait->u.wait.bucket = NULL;
+ wait->u.wait.wait_pending = FALSE;
+ wait->u.wait.timeout = 0;
+ wait->u.wait.handle = INVALID_HANDLE_VALUE;
+
+ RtlEnterCriticalSection( &waitqueue.cs );
+
+ /* Try to assign to existing bucket if possible. */
+ LIST_FOR_EACH_ENTRY( bucket, &waitqueue.buckets, struct waitqueue_bucket,
bucket_entry )
+ {
+ if (bucket->objcount < MAXIMUM_WAITQUEUE_OBJECTS &&
bucket->alertable == alertable)
+ {
+ list_add_tail( &bucket->reserved, &wait->u.wait.wait_entry );
+ wait->u.wait.bucket = bucket;
+ bucket->objcount++;
+
+ status = STATUS_SUCCESS;
+ goto out;
+ }
+ }
+
+ /* Create a new bucket and corresponding worker thread. */
+ bucket = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*bucket) );
+ if (!bucket)
+ {
+ status = STATUS_NO_MEMORY;
+ goto out;
+ }
+
+ bucket->objcount = 0;
+ bucket->alertable = alertable;
+ list_init( &bucket->reserved );
+ list_init( &bucket->waiting );
+
+ status = NtCreateEvent( &bucket->update_event, EVENT_ALL_ACCESS,
+ NULL, SynchronizationEvent, FALSE );
+ if (status)
+ {
+ RtlFreeHeap( GetProcessHeap(), 0, bucket );
+ goto out;
+ }
+
+ status = RtlCreateUserThread( GetCurrentProcess(), NULL, FALSE, 0, 0, 0,
+ waitqueue_thread_proc, bucket, &thread, NULL );
+ if (status == STATUS_SUCCESS)
+ {
+ list_add_tail( &waitqueue.buckets, &bucket->bucket_entry );
+ waitqueue.num_buckets++;
+
+ list_add_tail( &bucket->reserved, &wait->u.wait.wait_entry );
+ wait->u.wait.bucket = bucket;
+ bucket->objcount++;
+
+ NtClose( thread );
+ }
+ else
+ {
+ NtClose( bucket->update_event );
+ RtlFreeHeap( GetProcessHeap(), 0, bucket );
+ }
+
+out:
+ RtlLeaveCriticalSection( &waitqueue.cs );
+ return status;
+}
+
+/***********************************************************************
+ * tp_waitqueue_unlock (internal)
+ */
+static void tp_waitqueue_unlock( struct threadpool_object *wait )
+{
+ assert( wait->type == TP_OBJECT_TYPE_WAIT );
+
+ RtlEnterCriticalSection( &waitqueue.cs );
+ if (wait->u.wait.bucket)
+ {
+ struct waitqueue_bucket *bucket = wait->u.wait.bucket;
+ assert( bucket->objcount > 0 );
+
+ list_remove( &wait->u.wait.wait_entry );
+ wait->u.wait.bucket = NULL;
+ bucket->objcount--;
+
+ NtSetEvent( bucket->update_event, NULL );
+ }
+ RtlLeaveCriticalSection( &waitqueue.cs );
+}
+
+static void CALLBACK ioqueue_thread_proc( void *param )
+{
+ struct io_completion *completion;
+ struct threadpool_object *io;
+ IO_STATUS_BLOCK iosb;
+ ULONG_PTR key, value;
+ BOOL destroy, skip;
+ NTSTATUS status;
+
+ TRACE( "starting I/O completion thread\n" );
+ set_thread_name(L"wine_threadpool_ioqueue");
+
+ RtlEnterCriticalSection( &ioqueue.cs );
+
+ for (;;)
+ {
+ RtlLeaveCriticalSection( &ioqueue.cs );
+ if ((status = NtRemoveIoCompletion( ioqueue.port, &key, &value,
&iosb, NULL )))
+ ERR("NtRemoveIoCompletion failed, status %#lx.\n", status);
+ RtlEnterCriticalSection( &ioqueue.cs );
+
+ destroy = skip = FALSE;
+ io = (struct threadpool_object *)key;
+
+ TRACE( "io %p, iosb.Status %#lx.\n", io, iosb.Status );
+
+ if (io && (io->shutdown || io->u.io.shutting_down))
+ {
+ RtlEnterCriticalSection( &io->pool->cs );
+ if (!io->u.io.pending_count)
+ {
+ if (io->u.io.skipped_count)
+ --io->u.io.skipped_count;
+
+ if (io->u.io.skipped_count)
+ skip = TRUE;
+ else
+ destroy = TRUE;
+ }
+ RtlLeaveCriticalSection( &io->pool->cs );
+ if (skip) continue;
+ }
+
+ if (destroy)
+ {
+ --ioqueue.objcount;
+ TRACE( "Releasing io %p.\n", io );
+ io->shutdown = TRUE;
+ tp_object_release( io );
+ }
+ else if (io)
+ {
+ RtlEnterCriticalSection( &io->pool->cs );
+
+ TRACE( "pending_count %u.\n", io->u.io.pending_count );
+
+ if (io->u.io.pending_count)
+ {
+ --io->u.io.pending_count;
+ if (!array_reserve((void **)&io->u.io.completions,
&io->u.io.completion_max,
+ io->u.io.completion_count + 1,
sizeof(*io->u.io.completions)))
+ {
+ ERR( "Failed to allocate memory.\n" );
+ RtlLeaveCriticalSection( &io->pool->cs );
+ continue;
+ }
+
+ completion =
&io->u.io.completions[io->u.io.completion_count++];
+ completion->iosb = iosb;
+ completion->cvalue = value;
+
+ tp_object_submit( io, FALSE );
+ }
+ RtlLeaveCriticalSection( &io->pool->cs );
+ }
+
+ if (!ioqueue.objcount)
+ {
+ /* All I/O objects have been destroyed; if no new objects are
+ * created within some amount of time, then we can shutdown this
+ * thread. */
+ LARGE_INTEGER timeout = {.QuadPart = (ULONGLONG)THREADPOOL_WORKER_TIMEOUT *
-10000};
+ if (RtlSleepConditionVariableCS( &ioqueue.update_event, &ioqueue.cs,
+ &timeout) == STATUS_TIMEOUT && !ioqueue.objcount)
+ break;
+ }
+ }
+
+ ioqueue.thread_running = FALSE;
+ RtlLeaveCriticalSection( &ioqueue.cs );
+
+ TRACE( "terminating I/O completion thread\n" );
+
+ RtlExitUserThread( 0 );
+}
+
+static NTSTATUS tp_ioqueue_lock( struct threadpool_object *io, HANDLE file )
+{
+ NTSTATUS status = STATUS_SUCCESS;
+
+ assert( io->type == TP_OBJECT_TYPE_IO );
+
+ RtlEnterCriticalSection( &ioqueue.cs );
+
+ if (!ioqueue.port && (status = NtCreateIoCompletion( &ioqueue.port,
+ IO_COMPLETION_ALL_ACCESS, NULL, 0 )))
+ {
+ RtlLeaveCriticalSection( &ioqueue.cs );
+ return status;
+ }
+
+ if (!ioqueue.thread_running)
+ {
+ HANDLE thread;
+
+ if (!(status = RtlCreateUserThread( GetCurrentProcess(), NULL, FALSE,
+ 0, 0, 0, ioqueue_thread_proc, NULL,
&thread, NULL )))
+ {
+ ioqueue.thread_running = TRUE;
+ NtClose( thread );
+ }
+ }
+
+ if (status == STATUS_SUCCESS)
+ {
+ FILE_COMPLETION_INFORMATION info;
+ IO_STATUS_BLOCK iosb;
+
+ info.CompletionPort = ioqueue.port;
+ info.CompletionKey = (ULONG_PTR)io;
+
+ status = NtSetInformationFile( file, &iosb, &info, sizeof(info),
FileCompletionInformation );
+ }
+
+ if (status == STATUS_SUCCESS)
+ {
+ if (!ioqueue.objcount++)
+ RtlWakeConditionVariable( &ioqueue.update_event );
+ }
+
+ RtlLeaveCriticalSection( &ioqueue.cs );
+ return status;
+}
+
+/***********************************************************************
+ * tp_threadpool_alloc (internal)
+ *
+ * Allocates a new threadpool object.
+ */
+static NTSTATUS tp_threadpool_alloc( struct threadpool **out )
+{
+ IMAGE_NT_HEADERS *nt = RtlImageNtHeader( NtCurrentTeb()->Peb->ImageBaseAddress
);
+ struct threadpool *pool;
+ unsigned int i;
+
+ pool = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*pool) );
+ if (!pool)
+ return STATUS_NO_MEMORY;
+
+ pool->refcount = 1;
+ pool->objcount = 0;
+ pool->shutdown = FALSE;
+
+ RtlInitializeCriticalSectionEx( &pool->cs, 0,
RTL_CRITICAL_SECTION_FLAG_FORCE_DEBUG_INFO );
+ pool->cs.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ":
threadpool.cs");
+
+ for (i = 0; i < ARRAY_SIZE(pool->pools); ++i)
+ list_init( &pool->pools[i] );
+ RtlInitializeConditionVariable( &pool->update_event );
+
+ pool->max_workers = 500;
+ pool->min_workers = 0;
+ pool->num_workers = 0;
+ pool->num_busy_workers = 0;
+ pool->stack_info.StackReserve = nt->OptionalHeader.SizeOfStackReserve;
+ pool->stack_info.StackCommit = nt->OptionalHeader.SizeOfStackCommit;
+
+ TRACE( "allocated threadpool %p\n", pool );
+
+ *out = pool;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * tp_threadpool_shutdown (internal)
+ *
+ * Prepares the shutdown of a threadpool object and notifies all worker
+ * threads to terminate (after all remaining work items have been
+ * processed).
+ */
+static void tp_threadpool_shutdown( struct threadpool *pool )
+{
+ assert( pool != default_threadpool );
+
+ pool->shutdown = TRUE;
+ RtlWakeAllConditionVariable( &pool->update_event );
+}
+
+/***********************************************************************
+ * tp_threadpool_release (internal)
+ *
+ * Releases a reference to a threadpool object.
+ */
+static BOOL tp_threadpool_release( struct threadpool *pool )
+{
+ unsigned int i;
+
+ if (InterlockedDecrement( &pool->refcount ))
+ return FALSE;
+
+ TRACE( "destroying threadpool %p\n", pool );
+
+ assert( pool->shutdown );
+ assert( !pool->objcount );
+ for (i = 0; i < ARRAY_SIZE(pool->pools); ++i)
+ assert( list_empty( &pool->pools[i] ) );
+
+ pool->cs.DebugInfo->Spare[0] = 0;
+ RtlDeleteCriticalSection( &pool->cs );
+
+ RtlFreeHeap( GetProcessHeap(), 0, pool );
+ return TRUE;
+}
+
+/***********************************************************************
+ * tp_threadpool_lock (internal)
+ *
+ * Acquires a lock on a threadpool, specified with an TP_CALLBACK_ENVIRON
+ * block. When the lock is acquired successfully, it is guaranteed that
+ * there is at least one worker thread to process tasks.
+ */
+static NTSTATUS tp_threadpool_lock( struct threadpool **out, TP_CALLBACK_ENVIRON
*environment )
+{
+ struct threadpool *pool = NULL;
+ NTSTATUS status = STATUS_SUCCESS;
+
+ if (environment)
+ {
+ /* Validate environment parameters. */
+ if (environment->Version == 3)
+ {
+ TP_CALLBACK_ENVIRON_V3 *environment3 = (TP_CALLBACK_ENVIRON_V3
*)environment;
+
+ switch (environment3->CallbackPriority)
+ {
+ case TP_CALLBACK_PRIORITY_HIGH:
+ case TP_CALLBACK_PRIORITY_NORMAL:
+ case TP_CALLBACK_PRIORITY_LOW:
+ break;
+ default:
+ return STATUS_INVALID_PARAMETER;
+ }
+ }
+
+ pool = (struct threadpool *)environment->Pool;
+ }
+
+ if (!pool)
+ {
+ if (!default_threadpool)
+ {
+ status = tp_threadpool_alloc( &pool );
+ if (status != STATUS_SUCCESS)
+ return status;
+
+ if (InterlockedCompareExchangePointer( (void *)&default_threadpool, pool,
NULL ) != NULL)
+ {
+ tp_threadpool_shutdown( pool );
+ tp_threadpool_release( pool );
+ }
+ }
+
+ pool = default_threadpool;
+ }
+
+ RtlEnterCriticalSection( &pool->cs );
+
+ /* Make sure that the threadpool has at least one thread. */
+ if (!pool->num_workers)
+ status = tp_new_worker_thread( pool );
+
+ /* Keep a reference, and increment objcount to ensure that the
+ * last thread doesn't terminate. */
+ if (status == STATUS_SUCCESS)
+ {
+ InterlockedIncrement( &pool->refcount );
+ pool->objcount++;
+ }
+
+ RtlLeaveCriticalSection( &pool->cs );
+
+ if (status != STATUS_SUCCESS)
+ return status;
+
+ *out = pool;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * tp_threadpool_unlock (internal)
+ *
+ * Releases a lock on a threadpool.
+ */
+static void tp_threadpool_unlock( struct threadpool *pool )
+{
+ RtlEnterCriticalSection( &pool->cs );
+ pool->objcount--;
+ RtlLeaveCriticalSection( &pool->cs );
+ tp_threadpool_release( pool );
+}
+
+/***********************************************************************
+ * tp_group_alloc (internal)
+ *
+ * Allocates a new threadpool group object.
+ */
+static NTSTATUS tp_group_alloc( struct threadpool_group **out )
+{
+ struct threadpool_group *group;
+
+ group = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*group) );
+ if (!group)
+ return STATUS_NO_MEMORY;
+
+ group->refcount = 1;
+ group->shutdown = FALSE;
+
+ RtlInitializeCriticalSectionEx( &group->cs, 0,
RTL_CRITICAL_SECTION_FLAG_FORCE_DEBUG_INFO );
+ group->cs.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ":
threadpool_group.cs");
+
+ list_init( &group->members );
+
+ TRACE( "allocated group %p\n", group );
+
+ *out = group;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * tp_group_shutdown (internal)
+ *
+ * Marks the group object for shutdown.
+ */
+static void tp_group_shutdown( struct threadpool_group *group )
+{
+ group->shutdown = TRUE;
+}
+
+/***********************************************************************
+ * tp_group_release (internal)
+ *
+ * Releases a reference to a group object.
+ */
+static BOOL tp_group_release( struct threadpool_group *group )
+{
+ if (InterlockedDecrement( &group->refcount ))
+ return FALSE;
+
+ TRACE( "destroying group %p\n", group );
+
+ assert( group->shutdown );
+ assert( list_empty( &group->members ) );
+
+ group->cs.DebugInfo->Spare[0] = 0;
+ RtlDeleteCriticalSection( &group->cs );
+
+ RtlFreeHeap( GetProcessHeap(), 0, group );
+ return TRUE;
+}
+
+/***********************************************************************
+ * tp_object_initialize (internal)
+ *
+ * Initializes members of a threadpool object.
+ */
+static void tp_object_initialize( struct threadpool_object *object, struct threadpool
*pool,
+ PVOID userdata, TP_CALLBACK_ENVIRON *environment )
+{
+ BOOL is_simple_callback = (object->type == TP_OBJECT_TYPE_SIMPLE);
+
+ object->refcount = 1;
+ object->shutdown = FALSE;
+
+ object->pool = pool;
+ object->group = NULL;
+ object->userdata = userdata;
+ object->group_cancel_callback = NULL;
+ object->finalization_callback = NULL;
+ object->may_run_long = 0;
+ object->race_dll = NULL;
+ object->priority = TP_CALLBACK_PRIORITY_NORMAL;
+
+ memset( &object->group_entry, 0, sizeof(object->group_entry) );
+ object->is_group_member = FALSE;
+
+ memset( &object->pool_entry, 0, sizeof(object->pool_entry) );
+ RtlInitializeConditionVariable( &object->finished_event );
+ RtlInitializeConditionVariable( &object->group_finished_event );
+ object->completed_event = NULL;
+ object->num_pending_callbacks = 0;
+ object->num_running_callbacks = 0;
+ object->num_associated_callbacks = 0;
+
+ if (environment)
+ {
+ if (environment->Version != 1 && environment->Version != 3)
+ FIXME( "unsupported environment version %lu\n",
environment->Version );
+
+ object->group = impl_from_TP_CLEANUP_GROUP( environment->CleanupGroup );
+ object->group_cancel_callback = environment->CleanupGroupCancelCallback;
+ object->finalization_callback = environment->FinalizationCallback;
+ object->may_run_long = environment->u.s.LongFunction != 0;
+ object->race_dll = environment->RaceDll;
+ if (environment->Version == 3)
+ {
+ TP_CALLBACK_ENVIRON_V3 *environment_v3 = (TP_CALLBACK_ENVIRON_V3
*)environment;
+
+ object->priority = environment_v3->CallbackPriority;
+ assert( object->priority < ARRAY_SIZE(pool->pools) );
+ }
+
+ if (environment->ActivationContext)
+ FIXME( "activation context not supported yet\n" );
+
+ if (environment->u.s.Persistent)
+ FIXME( "persistent threads not supported yet\n" );
+ }
+
+ if (object->race_dll)
+ LdrAddRefDll( 0, object->race_dll );
+
+ TRACE( "allocated object %p of type %u\n", object, object->type );
+
+ /* For simple callbacks we have to run tp_object_submit before adding this object
+ * to the cleanup group. As soon as the cleanup group members are released
->shutdown
+ * will be set, and tp_object_submit would fail with an assertion. */
+
+ if (is_simple_callback)
+ tp_object_submit( object, FALSE );
+
+ if (object->group)
+ {
+ struct threadpool_group *group = object->group;
+ InterlockedIncrement( &group->refcount );
+
+ RtlEnterCriticalSection( &group->cs );
+ list_add_tail( &group->members, &object->group_entry );
+ object->is_group_member = TRUE;
+ RtlLeaveCriticalSection( &group->cs );
+ }
+
+ if (is_simple_callback)
+ tp_object_release( object );
+}
+
+static void tp_object_prio_queue( struct threadpool_object *object )
+{
+ ++object->pool->num_busy_workers;
+ list_add_tail( &object->pool->pools[object->priority],
&object->pool_entry );
+}
+
+/***********************************************************************
+ * tp_object_submit (internal)
+ *
+ * Submits a threadpool object to the associated threadpool. This
+ * function has to be VOID because TpPostWork can never fail on Windows.
+ */
+static void tp_object_submit( struct threadpool_object *object, BOOL signaled )
+{
+ struct threadpool *pool = object->pool;
+ NTSTATUS status = STATUS_UNSUCCESSFUL;
+
+ assert( !object->shutdown );
+ assert( !pool->shutdown );
+
+ RtlEnterCriticalSection( &pool->cs );
+
+ /* Start new worker threads if required. */
+ if (pool->num_busy_workers >= pool->num_workers &&
+ pool->num_workers < pool->max_workers)
+ status = tp_new_worker_thread( pool );
+
+ /* Queue work item and increment refcount. */
+ InterlockedIncrement( &object->refcount );
+ if (!object->num_pending_callbacks++)
+ tp_object_prio_queue( object );
+
+ /* Count how often the object was signaled. */
+ if (object->type == TP_OBJECT_TYPE_WAIT && signaled)
+ object->u.wait.signaled++;
+
+ /* No new thread started - wake up one existing thread. */
+ if (status != STATUS_SUCCESS)
+ {
+ assert( pool->num_workers > 0 );
+ RtlWakeConditionVariable( &pool->update_event );
+ }
+
+ RtlLeaveCriticalSection( &pool->cs );
+}
+
+/***********************************************************************
+ * tp_object_cancel (internal)
+ *
+ * Cancels all currently pending callbacks for a specific object.
+ */
+static void tp_object_cancel( struct threadpool_object *object )
+{
+ struct threadpool *pool = object->pool;
+ LONG pending_callbacks = 0;
+
+ RtlEnterCriticalSection( &pool->cs );
+ if (object->num_pending_callbacks)
+ {
+ pending_callbacks = object->num_pending_callbacks;
+ object->num_pending_callbacks = 0;
+ list_remove( &object->pool_entry );
+
+ if (object->type == TP_OBJECT_TYPE_WAIT)
+ object->u.wait.signaled = 0;
+ }
+ if (object->type == TP_OBJECT_TYPE_IO)
+ {
+ object->u.io.skipped_count += object->u.io.pending_count;
+ object->u.io.pending_count = 0;
+ }
+ RtlLeaveCriticalSection( &pool->cs );
+
+ while (pending_callbacks--)
+ tp_object_release( object );
+}
+
+static BOOL object_is_finished( struct threadpool_object *object, BOOL group )
+{
+ if (object->num_pending_callbacks)
+ return FALSE;
+ if (object->type == TP_OBJECT_TYPE_IO && object->u.io.pending_count)
+ return FALSE;
+
+ if (group)
+ return !object->num_running_callbacks;
+ else
+ return !object->num_associated_callbacks;
+}
+
+/***********************************************************************
+ * tp_object_wait (internal)
+ *
+ * Waits until all pending and running callbacks of a specific object
+ * have been processed.
+ */
+static void tp_object_wait( struct threadpool_object *object, BOOL group_wait )
+{
+ struct threadpool *pool = object->pool;
+
+ RtlEnterCriticalSection( &pool->cs );
+ while (!object_is_finished( object, group_wait ))
+ {
+ if (group_wait)
+ RtlSleepConditionVariableCS( &object->group_finished_event,
&pool->cs, NULL );
+ else
+ RtlSleepConditionVariableCS( &object->finished_event,
&pool->cs, NULL );
+ }
+ RtlLeaveCriticalSection( &pool->cs );
+}
+
+static void tp_ioqueue_unlock( struct threadpool_object *io )
+{
+ assert( io->type == TP_OBJECT_TYPE_IO );
+
+ RtlEnterCriticalSection( &ioqueue.cs );
+
+ assert(ioqueue.objcount);
+
+ if (!io->shutdown && !--ioqueue.objcount)
+ NtSetIoCompletion( ioqueue.port, 0, 0, STATUS_SUCCESS, 0 );
+
+ RtlLeaveCriticalSection( &ioqueue.cs );
+}
+
+/***********************************************************************
+ * tp_object_prepare_shutdown (internal)
+ *
+ * Prepares a threadpool object for shutdown.
+ */
+static void tp_object_prepare_shutdown( struct threadpool_object *object )
+{
+ if (object->type == TP_OBJECT_TYPE_TIMER)
+ tp_timerqueue_unlock( object );
+ else if (object->type == TP_OBJECT_TYPE_WAIT)
+ tp_waitqueue_unlock( object );
+ else if (object->type == TP_OBJECT_TYPE_IO)
+ tp_ioqueue_unlock( object );
+}
+
+/***********************************************************************
+ * tp_object_release (internal)
+ *
+ * Releases a reference to a threadpool object.
+ */
+static BOOL tp_object_release( struct threadpool_object *object )
+{
+ if (InterlockedDecrement( &object->refcount ))
+ return FALSE;
+
+ TRACE( "destroying object %p of type %u\n", object, object->type );
+
+ assert( object->shutdown );
+ assert( !object->num_pending_callbacks );
+ assert( !object->num_running_callbacks );
+ assert( !object->num_associated_callbacks );
+
+ /* release reference to the group */
+ if (object->group)
+ {
+ struct threadpool_group *group = object->group;
+
+ RtlEnterCriticalSection( &group->cs );
+ if (object->is_group_member)
+ {
+ list_remove( &object->group_entry );
+ object->is_group_member = FALSE;
+ }
+ RtlLeaveCriticalSection( &group->cs );
+
+ tp_group_release( group );
+ }
+
+ tp_threadpool_unlock( object->pool );
+
+ if (object->race_dll)
+ LdrUnloadDll( object->race_dll );
+
+ if (object->completed_event && object->completed_event !=
INVALID_HANDLE_VALUE)
+ NtSetEvent( object->completed_event, NULL );
+
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return TRUE;
+}
+
+static struct list *threadpool_get_next_item( const struct threadpool *pool )
+{
+ struct list *ptr;
+ unsigned int i;
+
+ for (i = 0; i < ARRAY_SIZE(pool->pools); ++i)
+ {
+ if ((ptr = list_head( &pool->pools[i] )))
+ break;
+ }
+
+ return ptr;
+}
+
+/***********************************************************************
+ * tp_object_execute (internal)
+ *
+ * Executes a threadpool object callback, object->pool->cs has to be
+ * held.
+ */
+static void tp_object_execute( struct threadpool_object *object, BOOL wait_thread )
+{
+ TP_CALLBACK_INSTANCE *callback_instance;
+ struct threadpool_instance instance;
+ struct io_completion completion;
+ struct threadpool *pool = object->pool;
+ TP_WAIT_RESULT wait_result = 0;
+ NTSTATUS status;
+
+ object->num_pending_callbacks--;
+
+ /* For wait objects check if they were signaled or have timed out. */
+ if (object->type == TP_OBJECT_TYPE_WAIT)
+ {
+ wait_result = object->u.wait.signaled ? WAIT_OBJECT_0 : WAIT_TIMEOUT;
+ if (wait_result == WAIT_OBJECT_0) object->u.wait.signaled--;
+ }
+ else if (object->type == TP_OBJECT_TYPE_IO)
+ {
+ assert( object->u.io.completion_count );
+ completion = object->u.io.completions[--object->u.io.completion_count];
+ }
+
+ /* Leave critical section and do the actual callback. */
+ object->num_associated_callbacks++;
+ object->num_running_callbacks++;
+ RtlLeaveCriticalSection( &pool->cs );
+ if (wait_thread) RtlLeaveCriticalSection( &waitqueue.cs );
+
+ /* Initialize threadpool instance struct. */
+ callback_instance = (TP_CALLBACK_INSTANCE *)&instance;
+ instance.object = object;
+ instance.threadid = GetCurrentThreadId();
+ instance.associated = TRUE;
+ instance.may_run_long = object->may_run_long;
+ instance.cleanup.critical_section = NULL;
+ instance.cleanup.mutex = NULL;
+ instance.cleanup.semaphore = NULL;
+ instance.cleanup.semaphore_count = 0;
+ instance.cleanup.event = NULL;
+ instance.cleanup.library = NULL;
+
+ switch (object->type)
+ {
+ case TP_OBJECT_TYPE_SIMPLE:
+ {
+ TRACE( "executing simple callback %p(%p, %p)\n",
+ object->u.simple.callback, callback_instance, object->userdata
);
+ object->u.simple.callback( callback_instance, object->userdata );
+ TRACE( "callback %p returned\n", object->u.simple.callback );
+ break;
+ }
+
+ case TP_OBJECT_TYPE_WORK:
+ {
+ TRACE( "executing work callback %p(%p, %p, %p)\n",
+ object->u.work.callback, callback_instance, object->userdata,
object );
+ object->u.work.callback( callback_instance, object->userdata, (TP_WORK
*)object );
+ TRACE( "callback %p returned\n", object->u.work.callback );
+ break;
+ }
+
+ case TP_OBJECT_TYPE_TIMER:
+ {
+ TRACE( "executing timer callback %p(%p, %p, %p)\n",
+ object->u.timer.callback, callback_instance, object->userdata,
object );
+ object->u.timer.callback( callback_instance, object->userdata,
(TP_TIMER *)object );
+ TRACE( "callback %p returned\n", object->u.timer.callback );
+ break;
+ }
+
+ case TP_OBJECT_TYPE_WAIT:
+ {
+ TRACE( "executing wait callback %p(%p, %p, %p, %lu)\n",
+ object->u.wait.callback, callback_instance, object->userdata,
object, wait_result );
+ object->u.wait.callback( callback_instance, object->userdata, (TP_WAIT
*)object, wait_result );
+ TRACE( "callback %p returned\n", object->u.wait.callback );
+ break;
+ }
+
+ case TP_OBJECT_TYPE_IO:
+ {
+ TRACE( "executing I/O callback %p(%p, %p, %#Ix, %p, %p)\n",
+ object->u.io.callback, callback_instance, object->userdata,
+ completion.cvalue, &completion.iosb, (TP_IO *)object );
+ object->u.io.callback( callback_instance, object->userdata,
+ (void *)completion.cvalue, &completion.iosb, (TP_IO *)object );
+ TRACE( "callback %p returned\n", object->u.io.callback );
+ break;
+ }
+
+ default:
+ assert(0);
+ break;
+ }
+
+ /* Execute finalization callback. */
+ if (object->finalization_callback)
+ {
+ TRACE( "executing finalization callback %p(%p, %p)\n",
+ object->finalization_callback, callback_instance, object->userdata
);
+ object->finalization_callback( callback_instance, object->userdata );
+ TRACE( "callback %p returned\n", object->finalization_callback );
+ }
+
+ /* Execute cleanup tasks. */
+ if (instance.cleanup.critical_section)
+ {
+ RtlLeaveCriticalSection( instance.cleanup.critical_section );
+ }
+ if (instance.cleanup.mutex)
+ {
+ status = NtReleaseMutant( instance.cleanup.mutex, NULL );
+ if (status != STATUS_SUCCESS) goto skip_cleanup;
+ }
+ if (instance.cleanup.semaphore)
+ {
+ status = NtReleaseSemaphore( instance.cleanup.semaphore,
instance.cleanup.semaphore_count, NULL );
+ if (status != STATUS_SUCCESS) goto skip_cleanup;
+ }
+ if (instance.cleanup.event)
+ {
+ status = NtSetEvent( instance.cleanup.event, NULL );
+ if (status != STATUS_SUCCESS) goto skip_cleanup;
+ }
+ if (instance.cleanup.library)
+ {
+ LdrUnloadDll( instance.cleanup.library );
+ }
+
+skip_cleanup:
+ if (wait_thread) RtlEnterCriticalSection( &waitqueue.cs );
+ RtlEnterCriticalSection( &pool->cs );
+
+ /* Simple callbacks are automatically shutdown after execution. */
+ if (object->type == TP_OBJECT_TYPE_SIMPLE)
+ {
+ tp_object_prepare_shutdown( object );
+ object->shutdown = TRUE;
+ }
+
+ object->num_running_callbacks--;
+ if (object_is_finished( object, TRUE ))
+ RtlWakeAllConditionVariable( &object->group_finished_event );
+
+ if (instance.associated)
+ {
+ object->num_associated_callbacks--;
+ if (object_is_finished( object, FALSE ))
+ RtlWakeAllConditionVariable( &object->finished_event );
+ }
+}
+
+/***********************************************************************
+ * threadpool_worker_proc (internal)
+ */
+static void CALLBACK threadpool_worker_proc( void *param )
+{
+ struct threadpool *pool = param;
+ LARGE_INTEGER timeout;
+ struct list *ptr;
+
+ TRACE( "starting worker thread for pool %p\n", pool );
+ set_thread_name(L"wine_threadpool_worker");
+
+ RtlEnterCriticalSection( &pool->cs );
+ for (;;)
+ {
+ while ((ptr = threadpool_get_next_item( pool )))
+ {
+ struct threadpool_object *object = LIST_ENTRY( ptr, struct threadpool_object,
pool_entry );
+ assert( object->num_pending_callbacks > 0 );
+
+ /* If further pending callbacks are queued, move the work item to
+ * the end of the pool list. Otherwise remove it from the pool. */
+ list_remove( &object->pool_entry );
+ if (object->num_pending_callbacks > 1)
+ tp_object_prio_queue( object );
+
+ tp_object_execute( object, FALSE );
+
+ assert(pool->num_busy_workers);
+ pool->num_busy_workers--;
+
+ tp_object_release( object );
+ }
+
+ /* Shutdown worker thread if requested. */
+ if (pool->shutdown)
+ break;
+
+ /* Wait for new tasks or until the timeout expires. A thread only terminates
+ * when no new tasks are available, and the number of threads can be
+ * decreased without violating the min_workers limit. An exception is when
+ * min_workers == 0, then objcount is used to detect if the last thread
+ * can be terminated. */
+ timeout.QuadPart = (ULONGLONG)THREADPOOL_WORKER_TIMEOUT * -10000;
+ if (RtlSleepConditionVariableCS( &pool->update_event, &pool->cs,
&timeout ) == STATUS_TIMEOUT &&
+ !threadpool_get_next_item( pool ) && (pool->num_workers > max(
pool->min_workers, 1 ) ||
+ (!pool->min_workers && !pool->objcount)))
+ {
+ break;
+ }
+ }
+ pool->num_workers--;
+ RtlLeaveCriticalSection( &pool->cs );
+
+ TRACE( "terminating worker thread for pool %p\n", pool );
+ tp_threadpool_release( pool );
+ RtlExitUserThread( 0 );
+}
+
+/***********************************************************************
+ * TpAllocCleanupGroup (NTDLL.@)
+ */
+NTSTATUS WINAPI TpAllocCleanupGroup( TP_CLEANUP_GROUP **out )
+{
+ TRACE( "%p\n", out );
+
+ return tp_group_alloc( (struct threadpool_group **)out );
+}
+
+/***********************************************************************
+ * TpAllocIoCompletion (NTDLL.@)
+ */
+NTSTATUS WINAPI TpAllocIoCompletion( TP_IO **out, HANDLE file, PTP_IO_CALLBACK callback,
+ void *userdata, TP_CALLBACK_ENVIRON *environment )
+{
+ struct threadpool_object *object;
+ struct threadpool *pool;
+ NTSTATUS status;
+
+ TRACE( "%p %p %p %p %p\n", out, file, callback, userdata, environment );
+
+ if (!(object = RtlAllocateHeap( GetProcessHeap(), HEAP_ZERO_MEMORY, sizeof(*object)
)))
+ return STATUS_NO_MEMORY;
+
+ if ((status = tp_threadpool_lock( &pool, environment )))
+ {
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ object->type = TP_OBJECT_TYPE_IO;
+ object->u.io.callback = callback;
+ if (!(object->u.io.completions = RtlAllocateHeap( GetProcessHeap(), 0, 8 *
sizeof(*object->u.io.completions) )))
+ {
+ tp_threadpool_unlock( pool );
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ if ((status = tp_ioqueue_lock( object, file )))
+ {
+ tp_threadpool_unlock( pool );
+ RtlFreeHeap( GetProcessHeap(), 0, object->u.io.completions );
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ tp_object_initialize( object, pool, userdata, environment );
+
+ *out = (TP_IO *)object;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * TpAllocPool (NTDLL.@)
+ */
+NTSTATUS WINAPI TpAllocPool( TP_POOL **out, PVOID reserved )
+{
+ TRACE( "%p %p\n", out, reserved );
+
+ if (reserved)
+ FIXME( "reserved argument is nonzero (%p)\n", reserved );
+
+ return tp_threadpool_alloc( (struct threadpool **)out );
+}
+
+/***********************************************************************
+ * TpAllocTimer (NTDLL.@)
+ */
+NTSTATUS WINAPI TpAllocTimer( TP_TIMER **out, PTP_TIMER_CALLBACK callback, PVOID
userdata,
+ TP_CALLBACK_ENVIRON *environment )
+{
+ struct threadpool_object *object;
+ struct threadpool *pool;
+ NTSTATUS status;
+
+ TRACE( "%p %p %p %p\n", out, callback, userdata, environment );
+
+ object = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*object) );
+ if (!object)
+ return STATUS_NO_MEMORY;
+
+ status = tp_threadpool_lock( &pool, environment );
+ if (status)
+ {
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ object->type = TP_OBJECT_TYPE_TIMER;
+ object->u.timer.callback = callback;
+
+ status = tp_timerqueue_lock( object );
+ if (status)
+ {
+ tp_threadpool_unlock( pool );
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ tp_object_initialize( object, pool, userdata, environment );
+
+ *out = (TP_TIMER *)object;
+ return STATUS_SUCCESS;
+}
+
+static NTSTATUS tp_alloc_wait( TP_WAIT **out, PTP_WAIT_CALLBACK callback, PVOID
userdata,
+ TP_CALLBACK_ENVIRON *environment, DWORD flags )
+{
+ struct threadpool_object *object;
+ struct threadpool *pool;
+ NTSTATUS status;
+
+ object = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*object) );
+ if (!object)
+ return STATUS_NO_MEMORY;
+
+ status = tp_threadpool_lock( &pool, environment );
+ if (status)
+ {
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ object->type = TP_OBJECT_TYPE_WAIT;
+ object->u.wait.callback = callback;
+ object->u.wait.flags = flags;
+
+ status = tp_waitqueue_lock( object );
+ if (status)
+ {
+ tp_threadpool_unlock( pool );
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ tp_object_initialize( object, pool, userdata, environment );
+
+ *out = (TP_WAIT *)object;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * TpAllocWait (NTDLL.@)
+ */
+NTSTATUS WINAPI TpAllocWait( TP_WAIT **out, PTP_WAIT_CALLBACK callback, PVOID userdata,
+ TP_CALLBACK_ENVIRON *environment )
+{
+ TRACE( "%p %p %p %p\n", out, callback, userdata, environment );
+ return tp_alloc_wait( out, callback, userdata, environment, WT_EXECUTEONLYONCE );
+}
+
+/***********************************************************************
+ * TpAllocWork (NTDLL.@)
+ */
+NTSTATUS WINAPI TpAllocWork( TP_WORK **out, PTP_WORK_CALLBACK callback, PVOID userdata,
+ TP_CALLBACK_ENVIRON *environment )
+{
+ struct threadpool_object *object;
+ struct threadpool *pool;
+ NTSTATUS status;
+
+ TRACE( "%p %p %p %p\n", out, callback, userdata, environment );
+
+ object = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*object) );
+ if (!object)
+ return STATUS_NO_MEMORY;
+
+ status = tp_threadpool_lock( &pool, environment );
+ if (status)
+ {
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ object->type = TP_OBJECT_TYPE_WORK;
+ object->u.work.callback = callback;
+ tp_object_initialize( object, pool, userdata, environment );
+
+ *out = (TP_WORK *)object;
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * TpCancelAsyncIoOperation (NTDLL.@)
+ */
+void WINAPI TpCancelAsyncIoOperation( TP_IO *io )
+{
+ struct threadpool_object *this = impl_from_TP_IO( io );
+
+ TRACE( "%p\n", io );
+
+ RtlEnterCriticalSection( &this->pool->cs );
+
+ TRACE("pending_count %u.\n", this->u.io.pending_count);
+
+ this->u.io.pending_count--;
+ if (object_is_finished( this, TRUE ))
+ RtlWakeAllConditionVariable( &this->group_finished_event );
+ if (object_is_finished( this, FALSE ))
+ RtlWakeAllConditionVariable( &this->finished_event );
+
+ RtlLeaveCriticalSection( &this->pool->cs );
+}
+
+/***********************************************************************
+ * TpCallbackLeaveCriticalSectionOnCompletion (NTDLL.@)
+ */
+VOID WINAPI TpCallbackLeaveCriticalSectionOnCompletion( TP_CALLBACK_INSTANCE *instance,
CRITICAL_SECTION *crit )
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+
+ TRACE( "%p %p\n", instance, crit );
+
+ if (!this->cleanup.critical_section)
+ this->cleanup.critical_section = crit;
+}
+
+/***********************************************************************
+ * TpCallbackMayRunLong (NTDLL.@)
+ */
+NTSTATUS WINAPI TpCallbackMayRunLong( TP_CALLBACK_INSTANCE *instance )
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+ struct threadpool_object *object = this->object;
+ struct threadpool *pool;
+ NTSTATUS status = STATUS_SUCCESS;
+
+ TRACE( "%p\n", instance );
+
+ if (this->threadid != GetCurrentThreadId())
+ {
+ ERR("called from wrong thread, ignoring\n");
+ return STATUS_UNSUCCESSFUL; /* FIXME */
+ }
+
+ if (this->may_run_long)
+ return STATUS_SUCCESS;
+
+ pool = object->pool;
+ RtlEnterCriticalSection( &pool->cs );
+
+ /* Start new worker threads if required. */
+ if (pool->num_busy_workers >= pool->num_workers)
+ {
+ if (pool->num_workers < pool->max_workers)
+ {
+ status = tp_new_worker_thread( pool );
+ }
+ else
+ {
+ status = STATUS_TOO_MANY_THREADS;
+ }
+ }
+
+ RtlLeaveCriticalSection( &pool->cs );
+ this->may_run_long = TRUE;
+ return status;
+}
+
+/***********************************************************************
+ * TpCallbackReleaseMutexOnCompletion (NTDLL.@)
+ */
+VOID WINAPI TpCallbackReleaseMutexOnCompletion( TP_CALLBACK_INSTANCE *instance, HANDLE
mutex )
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+
+ TRACE( "%p %p\n", instance, mutex );
+
+ if (!this->cleanup.mutex)
+ this->cleanup.mutex = mutex;
+}
+
+/***********************************************************************
+ * TpCallbackReleaseSemaphoreOnCompletion (NTDLL.@)
+ */
+VOID WINAPI TpCallbackReleaseSemaphoreOnCompletion( TP_CALLBACK_INSTANCE *instance,
HANDLE semaphore, DWORD count )
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+
+ TRACE( "%p %p %lu\n", instance, semaphore, count );
+
+ if (!this->cleanup.semaphore)
+ {
+ this->cleanup.semaphore = semaphore;
+ this->cleanup.semaphore_count = count;
+ }
+}
+
+/***********************************************************************
+ * TpCallbackSetEventOnCompletion (NTDLL.@)
+ */
+VOID WINAPI TpCallbackSetEventOnCompletion( TP_CALLBACK_INSTANCE *instance, HANDLE event
)
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+
+ TRACE( "%p %p\n", instance, event );
+
+ if (!this->cleanup.event)
+ this->cleanup.event = event;
+}
+
+/***********************************************************************
+ * TpCallbackUnloadDllOnCompletion (NTDLL.@)
+ */
+VOID WINAPI TpCallbackUnloadDllOnCompletion( TP_CALLBACK_INSTANCE *instance, HMODULE
module )
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+
+ TRACE( "%p %p\n", instance, module );
+
+ if (!this->cleanup.library)
+ this->cleanup.library = module;
+}
+
+/***********************************************************************
+ * TpDisassociateCallback (NTDLL.@)
+ */
+VOID WINAPI TpDisassociateCallback( TP_CALLBACK_INSTANCE *instance )
+{
+ struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
+ struct threadpool_object *object = this->object;
+ struct threadpool *pool;
+
+ TRACE( "%p\n", instance );
+
+ if (this->threadid != GetCurrentThreadId())
+ {
+ ERR("called from wrong thread, ignoring\n");
+ return;
+ }
+
+ if (!this->associated)
+ return;
+
+ pool = object->pool;
+ RtlEnterCriticalSection( &pool->cs );
+
+ object->num_associated_callbacks--;
+ if (object_is_finished( object, FALSE ))
+ RtlWakeAllConditionVariable( &object->finished_event );
+
+ RtlLeaveCriticalSection( &pool->cs );
+ this->associated = FALSE;
+}
+
+/***********************************************************************
+ * TpIsTimerSet (NTDLL.@)
+ */
+BOOL WINAPI TpIsTimerSet( TP_TIMER *timer )
+{
+ struct threadpool_object *this = impl_from_TP_TIMER( timer );
+
+ TRACE( "%p\n", timer );
+
+ return this->u.timer.timer_set;
+}
+
+/***********************************************************************
+ * TpPostWork (NTDLL.@)
+ */
+VOID WINAPI TpPostWork( TP_WORK *work )
+{
+ struct threadpool_object *this = impl_from_TP_WORK( work );
+
+ TRACE( "%p\n", work );
+
+ tp_object_submit( this, FALSE );
+}
+
+/***********************************************************************
+ * TpReleaseCleanupGroup (NTDLL.@)
+ */
+VOID WINAPI TpReleaseCleanupGroup( TP_CLEANUP_GROUP *group )
+{
+ struct threadpool_group *this = impl_from_TP_CLEANUP_GROUP( group );
+
+ TRACE( "%p\n", group );
+
+ tp_group_shutdown( this );
+ tp_group_release( this );
+}
+
+/***********************************************************************
+ * TpReleaseCleanupGroupMembers (NTDLL.@)
+ */
+VOID WINAPI TpReleaseCleanupGroupMembers( TP_CLEANUP_GROUP *group, BOOL cancel_pending,
PVOID userdata )
+{
+ struct threadpool_group *this = impl_from_TP_CLEANUP_GROUP( group );
+ struct threadpool_object *object, *next;
+ struct list members;
+
+ TRACE( "%p %u %p\n", group, cancel_pending, userdata );
+
+ RtlEnterCriticalSection( &this->cs );
+
+ /* Unset group, increase references, and mark objects for shutdown */
+ LIST_FOR_EACH_ENTRY_SAFE( object, next, &this->members, struct
threadpool_object, group_entry )
+ {
+ assert( object->group == this );
+ assert( object->is_group_member );
+
+ if (InterlockedIncrement( &object->refcount ) == 1)
+ {
+ /* Object is basically already destroyed, but group reference
+ * was not deleted yet. We can safely ignore this object. */
+ InterlockedDecrement( &object->refcount );
+ list_remove( &object->group_entry );
+ object->is_group_member = FALSE;
+ continue;
+ }
+
+ object->is_group_member = FALSE;
+ tp_object_prepare_shutdown( object );
+ }
+
+ /* Move members to a new temporary list */
+ list_init( &members );
+ list_move_tail( &members, &this->members );
+
+ RtlLeaveCriticalSection( &this->cs );
+
+ /* Cancel pending callbacks if requested */
+ if (cancel_pending)
+ {
+ LIST_FOR_EACH_ENTRY( object, &members, struct threadpool_object, group_entry
)
+ {
+ tp_object_cancel( object );
+ }
+ }
+
+ /* Wait for remaining callbacks to finish */
+ LIST_FOR_EACH_ENTRY_SAFE( object, next, &members, struct threadpool_object,
group_entry )
+ {
+ tp_object_wait( object, TRUE );
+
+ if (!object->shutdown)
+ {
+ /* Execute group cancellation callback if defined, and if this was actually a
group cancel. */
+ if (cancel_pending && object->group_cancel_callback)
+ {
+ TRACE( "executing group cancel callback %p(%p, %p)\n",
+ object->group_cancel_callback, object->userdata, userdata
);
+ object->group_cancel_callback( object->userdata, userdata );
+ TRACE( "callback %p returned\n",
object->group_cancel_callback );
+ }
+
+ if (object->type != TP_OBJECT_TYPE_SIMPLE)
+ tp_object_release( object );
+ }
+
+ object->shutdown = TRUE;
+ tp_object_release( object );
+ }
+}
+
+/***********************************************************************
+ * TpReleaseIoCompletion (NTDLL.@)
+ */
+void WINAPI TpReleaseIoCompletion( TP_IO *io )
+{
+ struct threadpool_object *this = impl_from_TP_IO( io );
+ BOOL can_destroy;
+
+ TRACE( "%p\n", io );
+
+ RtlEnterCriticalSection( &this->pool->cs );
+ this->u.io.shutting_down = TRUE;
+ can_destroy = !this->u.io.pending_count && !this->u.io.skipped_count;
+ RtlLeaveCriticalSection( &this->pool->cs );
+
+ if (can_destroy)
+ {
+ tp_object_prepare_shutdown( this );
+ this->shutdown = TRUE;
+ tp_object_release( this );
+ }
+}
+
+/***********************************************************************
+ * TpReleasePool (NTDLL.@)
+ */
+VOID WINAPI TpReleasePool( TP_POOL *pool )
+{
+ struct threadpool *this = impl_from_TP_POOL( pool );
+
+ TRACE( "%p\n", pool );
+
+ tp_threadpool_shutdown( this );
+ tp_threadpool_release( this );
+}
+
+/***********************************************************************
+ * TpReleaseTimer (NTDLL.@)
+ */
+VOID WINAPI TpReleaseTimer( TP_TIMER *timer )
+{
+ struct threadpool_object *this = impl_from_TP_TIMER( timer );
+
+ TRACE( "%p\n", timer );
+
+ tp_object_prepare_shutdown( this );
+ this->shutdown = TRUE;
+ tp_object_release( this );
+}
+
+/***********************************************************************
+ * TpReleaseWait (NTDLL.@)
+ */
+VOID WINAPI TpReleaseWait( TP_WAIT *wait )
+{
+ struct threadpool_object *this = impl_from_TP_WAIT( wait );
+
+ TRACE( "%p\n", wait );
+
+ tp_object_prepare_shutdown( this );
+ this->shutdown = TRUE;
+ tp_object_release( this );
+}
+
+/***********************************************************************
+ * TpReleaseWork (NTDLL.@)
+ */
+VOID WINAPI TpReleaseWork( TP_WORK *work )
+{
+ struct threadpool_object *this = impl_from_TP_WORK( work );
+
+ TRACE( "%p\n", work );
+
+ tp_object_prepare_shutdown( this );
+ this->shutdown = TRUE;
+ tp_object_release( this );
+}
+
+/***********************************************************************
+ * TpSetPoolMaxThreads (NTDLL.@)
+ */
+VOID WINAPI TpSetPoolMaxThreads( TP_POOL *pool, DWORD maximum )
+{
+ struct threadpool *this = impl_from_TP_POOL( pool );
+
+ TRACE( "%p %lu\n", pool, maximum );
+
+ RtlEnterCriticalSection( &this->cs );
+ this->max_workers = max( maximum, 1 );
+ this->min_workers = min( this->min_workers, this->max_workers );
+ RtlLeaveCriticalSection( &this->cs );
+}
+
+/***********************************************************************
+ * TpSetPoolMinThreads (NTDLL.@)
+ */
+BOOL WINAPI TpSetPoolMinThreads( TP_POOL *pool, DWORD minimum )
+{
+ struct threadpool *this = impl_from_TP_POOL( pool );
+ NTSTATUS status = STATUS_SUCCESS;
+
+ TRACE( "%p %lu\n", pool, minimum );
+
+ RtlEnterCriticalSection( &this->cs );
+
+ while (this->num_workers < minimum)
+ {
+ status = tp_new_worker_thread( this );
+ if (status != STATUS_SUCCESS)
+ break;
+ }
+
+ if (status == STATUS_SUCCESS)
+ {
+ this->min_workers = minimum;
+ this->max_workers = max( this->min_workers, this->max_workers );
+ }
+
+ RtlLeaveCriticalSection( &this->cs );
+ return !status;
+}
+
+/***********************************************************************
+ * TpSetTimer (NTDLL.@)
+ */
+VOID WINAPI TpSetTimer( TP_TIMER *timer, LARGE_INTEGER *timeout, LONG period, LONG
window_length )
+{
+ struct threadpool_object *this = impl_from_TP_TIMER( timer );
+ struct threadpool_object *other_timer;
+ BOOL submit_timer = FALSE;
+ ULONGLONG timestamp;
+
+ TRACE( "%p %p %lu %lu\n", timer, timeout, period, window_length );
+
+ RtlEnterCriticalSection( &timerqueue.cs );
+
+ assert( this->u.timer.timer_initialized );
+ this->u.timer.timer_set = timeout != NULL;
+
+ /* Convert relative timeout to absolute timestamp and handle a timeout
+ * of zero, which means that the timer is submitted immediately. */
+ if (timeout)
+ {
+ timestamp = timeout->QuadPart;
+ if ((LONGLONG)timestamp < 0)
+ {
+ LARGE_INTEGER now;
+ NtQuerySystemTime( &now );
+ timestamp = now.QuadPart - timestamp;
+ }
+ else if (!timestamp)
+ {
+ if (!period)
+ timeout = NULL;
+ else
+ {
+ LARGE_INTEGER now;
+ NtQuerySystemTime( &now );
+ timestamp = now.QuadPart + (ULONGLONG)period * 10000;
+ }
+ submit_timer = TRUE;
+ }
+ }
+
+ /* First remove existing timeout. */
+ if (this->u.timer.timer_pending)
+ {
+ list_remove( &this->u.timer.timer_entry );
+ this->u.timer.timer_pending = FALSE;
+ }
+
+ /* If the timer was enabled, then add it back to the queue. */
+ if (timeout)
+ {
+ this->u.timer.timeout = timestamp;
+ this->u.timer.period = period;
+ this->u.timer.window_length = window_length;
+
+ LIST_FOR_EACH_ENTRY( other_timer, &timerqueue.pending_timers,
+ struct threadpool_object, u.timer.timer_entry )
+ {
+ assert( other_timer->type == TP_OBJECT_TYPE_TIMER );
+ if (this->u.timer.timeout < other_timer->u.timer.timeout)
+ break;
+ }
+ list_add_before( &other_timer->u.timer.timer_entry,
&this->u.timer.timer_entry );
+
+ /* Wake up the timer thread when the timeout has to be updated. */
+ if (list_head( &timerqueue.pending_timers ) ==
&this->u.timer.timer_entry )
+ RtlWakeAllConditionVariable( &timerqueue.update_event );
+
+ this->u.timer.timer_pending = TRUE;
+ }
+
+ RtlLeaveCriticalSection( &timerqueue.cs );
+
+ if (submit_timer)
+ tp_object_submit( this, FALSE );
+}
+
+/***********************************************************************
+ * TpSetWait (NTDLL.@)
+ */
+VOID WINAPI TpSetWait( TP_WAIT *wait, HANDLE handle, LARGE_INTEGER *timeout )
+{
+ struct threadpool_object *this = impl_from_TP_WAIT( wait );
+ ULONGLONG timestamp = MAXLONGLONG;
+
+ TRACE( "%p %p %p\n", wait, handle, timeout );
+
+ RtlEnterCriticalSection( &waitqueue.cs );
+
+ assert( this->u.wait.bucket );
+ this->u.wait.handle = handle;
+
+ if (handle || this->u.wait.wait_pending)
+ {
+ struct waitqueue_bucket *bucket = this->u.wait.bucket;
+ list_remove( &this->u.wait.wait_entry );
+
+ /* Convert relative timeout to absolute timestamp. */
+ if (handle && timeout)
+ {
+ timestamp = timeout->QuadPart;
+ if ((LONGLONG)timestamp < 0)
+ {
+ LARGE_INTEGER now;
+ NtQuerySystemTime( &now );
+ timestamp = now.QuadPart - timestamp;
+ }
+ }
+
+ /* Add wait object back into one of the queues. */
+ if (handle)
+ {
+ list_add_tail( &bucket->waiting, &this->u.wait.wait_entry );
+ this->u.wait.wait_pending = TRUE;
+ this->u.wait.timeout = timestamp;
+ }
+ else
+ {
+ list_add_tail( &bucket->reserved, &this->u.wait.wait_entry );
+ this->u.wait.wait_pending = FALSE;
+ }
+
+ /* Wake up the wait queue thread. */
+ NtSetEvent( bucket->update_event, NULL );
+ }
+
+ RtlLeaveCriticalSection( &waitqueue.cs );
+}
+
+/***********************************************************************
+ * TpSimpleTryPost (NTDLL.@)
+ */
+NTSTATUS WINAPI TpSimpleTryPost( PTP_SIMPLE_CALLBACK callback, PVOID userdata,
+ TP_CALLBACK_ENVIRON *environment )
+{
+ struct threadpool_object *object;
+ struct threadpool *pool;
+ NTSTATUS status;
+
+ TRACE( "%p %p %p\n", callback, userdata, environment );
+
+ object = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*object) );
+ if (!object)
+ return STATUS_NO_MEMORY;
+
+ status = tp_threadpool_lock( &pool, environment );
+ if (status)
+ {
+ RtlFreeHeap( GetProcessHeap(), 0, object );
+ return status;
+ }
+
+ object->type = TP_OBJECT_TYPE_SIMPLE;
+ object->u.simple.callback = callback;
+ tp_object_initialize( object, pool, userdata, environment );
+
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * TpStartAsyncIoOperation (NTDLL.@)
+ */
+void WINAPI TpStartAsyncIoOperation( TP_IO *io )
+{
+ struct threadpool_object *this = impl_from_TP_IO( io );
+
+ TRACE( "%p\n", io );
+
+ RtlEnterCriticalSection( &this->pool->cs );
+
+ this->u.io.pending_count++;
+
+ RtlLeaveCriticalSection( &this->pool->cs );
+}
+
+/***********************************************************************
+ * TpWaitForIoCompletion (NTDLL.@)
+ */
+void WINAPI TpWaitForIoCompletion( TP_IO *io, BOOL cancel_pending )
+{
+ struct threadpool_object *this = impl_from_TP_IO( io );
+
+ TRACE( "%p %d\n", io, cancel_pending );
+
+ if (cancel_pending)
+ tp_object_cancel( this );
+ tp_object_wait( this, FALSE );
+}
+
+/***********************************************************************
+ * TpWaitForTimer (NTDLL.@)
+ */
+VOID WINAPI TpWaitForTimer( TP_TIMER *timer, BOOL cancel_pending )
+{
+ struct threadpool_object *this = impl_from_TP_TIMER( timer );
+
+ TRACE( "%p %d\n", timer, cancel_pending );
+
+ if (cancel_pending)
+ tp_object_cancel( this );
+ tp_object_wait( this, FALSE );
+}
+
+/***********************************************************************
+ * TpWaitForWait (NTDLL.@)
+ */
+VOID WINAPI TpWaitForWait( TP_WAIT *wait, BOOL cancel_pending )
+{
+ struct threadpool_object *this = impl_from_TP_WAIT( wait );
+
+ TRACE( "%p %d\n", wait, cancel_pending );
+
+ if (cancel_pending)
+ tp_object_cancel( this );
+ tp_object_wait( this, FALSE );
+}
+
+/***********************************************************************
+ * TpWaitForWork (NTDLL.@)
+ */
+VOID WINAPI TpWaitForWork( TP_WORK *work, BOOL cancel_pending )
+{
+ struct threadpool_object *this = impl_from_TP_WORK( work );
+
+ TRACE( "%p %u\n", work, cancel_pending );
+
+ if (cancel_pending)
+ tp_object_cancel( this );
+ tp_object_wait( this, FALSE );
+}
+
+/***********************************************************************
+ * TpSetPoolStackInformation (NTDLL.@)
+ */
+NTSTATUS WINAPI TpSetPoolStackInformation( TP_POOL *pool, TP_POOL_STACK_INFORMATION
*stack_info )
+{
+ struct threadpool *this = impl_from_TP_POOL( pool );
+
+ TRACE( "%p %p\n", pool, stack_info );
+
+ if (!stack_info)
+ return STATUS_INVALID_PARAMETER;
+
+ RtlEnterCriticalSection( &this->cs );
+ this->stack_info = *stack_info;
+ RtlLeaveCriticalSection( &this->cs );
+
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * TpQueryPoolStackInformation (NTDLL.@)
+ */
+NTSTATUS WINAPI TpQueryPoolStackInformation( TP_POOL *pool, TP_POOL_STACK_INFORMATION
*stack_info )
+{
+ struct threadpool *this = impl_from_TP_POOL( pool );
+
+ TRACE( "%p %p\n", pool, stack_info );
+
+ if (!stack_info)
+ return STATUS_INVALID_PARAMETER;
+
+ RtlEnterCriticalSection( &this->cs );
+ *stack_info = this->stack_info;
+ RtlLeaveCriticalSection( &this->cs );
+
+ return STATUS_SUCCESS;
+}
+
+static void CALLBACK rtl_wait_callback( TP_CALLBACK_INSTANCE *instance, void *userdata,
TP_WAIT *wait, TP_WAIT_RESULT result )
+{
+ struct threadpool_object *object = impl_from_TP_WAIT(wait);
+ object->u.wait.rtl_callback( userdata, result != STATUS_WAIT_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( HANDLE *out, HANDLE handle, RTL_WAITORTIMERCALLBACKFUNC
callback,
+ void *context, ULONG milliseconds, ULONG flags )
+{
+ struct threadpool_object *object;
+ TP_CALLBACK_ENVIRON environment;
+ LARGE_INTEGER timeout;
+ NTSTATUS status;
+ TP_WAIT *wait;
+
+ TRACE( "out %p, handle %p, callback %p, context %p, milliseconds %lu, flags
%lx\n",
+ out, handle, callback, context, milliseconds, flags );
+
+ memset( &environment, 0, sizeof(environment) );
+ environment.Version = 1;
+ environment.u.s.LongFunction = (flags & WT_EXECUTELONGFUNCTION) != 0;
+ environment.u.s.Persistent = (flags & WT_EXECUTEINPERSISTENTTHREAD) != 0;
+
+ flags &= (WT_EXECUTEONLYONCE | WT_EXECUTEINWAITTHREAD | WT_EXECUTEINIOTHREAD);
+ if ((status = tp_alloc_wait( &wait, rtl_wait_callback, context, &environment,
flags )))
+ return status;
+
+ object = impl_from_TP_WAIT(wait);
+ object->u.wait.rtl_callback = callback;
+
+ RtlEnterCriticalSection( &waitqueue.cs );
+ TpSetWait( (TP_WAIT *)object, handle, get_nt_timeout( &timeout, milliseconds )
);
+
+ *out = object;
+ RtlLeaveCriticalSection( &waitqueue.cs );
+
+ return STATUS_SUCCESS;
+}
+
+/***********************************************************************
+ * 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 handle, HANDLE event )
+{
+ struct threadpool_object *object = handle;
+ NTSTATUS status;
+
+ TRACE( "handle %p, event %p\n", handle, event );
+
+ if (!object) return STATUS_INVALID_HANDLE;
+
+ TpSetWait( (TP_WAIT *)object, NULL, NULL );
+
+ if (event == INVALID_HANDLE_VALUE) TpWaitForWait( (TP_WAIT *)object, TRUE );
+ else
+ {
+ assert( object->completed_event == NULL );
+ object->completed_event = event;
+ }
+
+ RtlEnterCriticalSection( &object->pool->cs );
+ if (object->num_pending_callbacks + object->num_running_callbacks
+ + object->num_associated_callbacks) status = STATUS_PENDING;
+ else status = STATUS_SUCCESS;
+ RtlLeaveCriticalSection( &object->pool->cs );
+
+ TpReleaseWait( (TP_WAIT *)object );
+ 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);
+}