threadpool.c File Reference

#include <assert.h>
#include <stdarg.h>
#include <limits.h>
#include "ntstatus.h"
#include "winternl.h"
#include "wine/debug.h"
#include "wine/list.h"
#include "ntdll_misc.h"

Include dependency graph for threadpool.c:

Go to the source code of this file.

Classes
struct	rtl_work_item
struct	queue_timer
struct	timer_queue
struct	threadpool
struct	io_completion
struct	threadpool_object
struct	threadpool_instance
struct	threadpool_group
struct	waitqueue_bucket

Macros
#define	WIN32_NO_STATUS
#define	EXPIRE_NEVER   (~(ULONGLONG)0)
#define	TIMER_QUEUE_MAGIC   0x516d6954 /* TimQ */
#define	THREADPOOL_WORKER_TIMEOUT   5000
#define	MAXIMUM_WAITQUEUE_OBJECTS   (MAXIMUM_WAIT_OBJECTS - 1)

Enumerations
enum	threadpool_objtype {   TP_OBJECT_TYPE_SIMPLE , TP_OBJECT_TYPE_WORK , TP_OBJECT_TYPE_TIMER , TP_OBJECT_TYPE_WAIT ,   TP_OBJECT_TYPE_IO }

Functions
	WINE_DEFAULT_DEBUG_CHANNEL (threadpool)
static struct threadpool *	impl_from_TP_POOL (TP_POOL *pool)
static struct threadpool_object *	impl_from_TP_WORK (TP_WORK *work)
static struct threadpool_object *	impl_from_TP_TIMER (TP_TIMER *timer)
static struct threadpool_object *	impl_from_TP_WAIT (TP_WAIT *wait)
static struct threadpool_object *	impl_from_TP_IO (TP_IO *io)
static struct threadpool_group *	impl_from_TP_CLEANUP_GROUP (TP_CLEANUP_GROUP *group)
static struct threadpool_instance *	impl_from_TP_CALLBACK_INSTANCE (TP_CALLBACK_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 BOOL	array_reserve (void **elements, unsigned int *capacity, unsigned int count, unsigned int size)
static void	set_thread_name (const WCHAR *name)
static void CALLBACK	process_rtl_work_item (TP_CALLBACK_INSTANCE *instance, void *userdata)
NTSTATUS WINAPI	RtlQueueWorkItem (PRTL_WORK_ITEM_ROUTINE function, PVOID context, ULONG flags)
static DWORD CALLBACK	iocp_poller (LPVOID Arg)
NTSTATUS WINAPI	RtlSetIoCompletionCallback (HANDLE FileHandle, PRTL_OVERLAPPED_COMPLETION_ROUTINE Function, ULONG Flags)
static PLARGE_INTEGER	get_nt_timeout (PLARGE_INTEGER pTime, ULONG timeout)
static void	queue_remove_timer (struct queue_timer *t)
static void	timer_cleanup_callback (struct queue_timer *t)
static DWORD WINAPI	timer_callback_wrapper (LPVOID p)
static ULONGLONG	queue_current_time (void)
static void	queue_add_timer (struct queue_timer *t, ULONGLONG time, BOOL set_event)
static void	queue_move_timer (struct queue_timer *t, ULONGLONG time, BOOL set_event)
static void	queue_timer_expire (struct timer_queue *q)
static ULONG	queue_get_timeout (struct timer_queue *q)
static void WINAPI	timer_queue_thread_proc (LPVOID p)
static void	queue_destroy_timer (struct queue_timer *t)
NTSTATUS WINAPI	RtlCreateTimerQueue (PHANDLE NewTimerQueue)
NTSTATUS WINAPI	RtlDeleteTimerQueueEx (HANDLE TimerQueue, HANDLE CompletionEvent)
static struct timer_queue *	get_timer_queue (HANDLE TimerQueue)
NTSTATUS WINAPI	RtlCreateTimer (HANDLE TimerQueue, HANDLE *NewTimer, RTL_WAITORTIMERCALLBACKFUNC Callback, PVOID Parameter, DWORD DueTime, DWORD Period, ULONG Flags)
NTSTATUS WINAPI	RtlUpdateTimer (HANDLE TimerQueue, HANDLE Timer, DWORD DueTime, DWORD Period)
NTSTATUS WINAPI	RtlDeleteTimer (HANDLE TimerQueue, HANDLE Timer, HANDLE CompletionEvent)
static void CALLBACK	timerqueue_thread_proc (void *param)
static NTSTATUS	tp_new_worker_thread (struct threadpool *pool)
static NTSTATUS	tp_timerqueue_lock (struct threadpool_object *timer)
static void	tp_timerqueue_unlock (struct threadpool_object *timer)
static void CALLBACK	waitqueue_thread_proc (void *param)
static NTSTATUS	tp_waitqueue_lock (struct threadpool_object *wait)
static void	tp_waitqueue_unlock (struct threadpool_object *wait)
static void CALLBACK	ioqueue_thread_proc (void *param)
static NTSTATUS	tp_ioqueue_lock (struct threadpool_object *io, HANDLE file)
static NTSTATUS	tp_threadpool_alloc (struct threadpool **out)
static void	tp_threadpool_shutdown (struct threadpool *pool)
static BOOL	tp_threadpool_release (struct threadpool *pool)
static NTSTATUS	tp_threadpool_lock (struct threadpool **out, TP_CALLBACK_ENVIRON *environment)
static void	tp_threadpool_unlock (struct threadpool *pool)
static NTSTATUS	tp_group_alloc (struct threadpool_group **out)
static void	tp_group_shutdown (struct threadpool_group *group)
static BOOL	tp_group_release (struct threadpool_group *group)
static void	tp_object_initialize (struct threadpool_object *object, struct threadpool *pool, PVOID userdata, TP_CALLBACK_ENVIRON *environment)
static void	tp_object_prio_queue (struct threadpool_object *object)
static void	tp_object_cancel (struct threadpool_object *object)
static BOOL	object_is_finished (struct threadpool_object *object, BOOL group)
static void	tp_object_wait (struct threadpool_object *object, BOOL group_wait)
static void	tp_ioqueue_unlock (struct threadpool_object *io)
static struct list *	threadpool_get_next_item (const struct threadpool *pool)
NTSTATUS WINAPI	TpAllocCleanupGroup (TP_CLEANUP_GROUP **out)
NTSTATUS WINAPI	TpAllocIoCompletion (TP_IO **out, HANDLE file, PTP_IO_CALLBACK callback, void *userdata, TP_CALLBACK_ENVIRON *environment)
NTSTATUS WINAPI	TpAllocPool (TP_POOL **out, PVOID reserved)
NTSTATUS WINAPI	TpAllocTimer (TP_TIMER **out, PTP_TIMER_CALLBACK callback, PVOID userdata, TP_CALLBACK_ENVIRON *environment)
static NTSTATUS	tp_alloc_wait (TP_WAIT **out, PTP_WAIT_CALLBACK callback, PVOID userdata, TP_CALLBACK_ENVIRON *environment, DWORD flags)
NTSTATUS WINAPI	TpAllocWait (TP_WAIT **out, PTP_WAIT_CALLBACK callback, PVOID userdata, TP_CALLBACK_ENVIRON *environment)
NTSTATUS WINAPI	TpAllocWork (TP_WORK **out, PTP_WORK_CALLBACK callback, PVOID userdata, TP_CALLBACK_ENVIRON *environment)
void WINAPI	TpCancelAsyncIoOperation (TP_IO *io)
VOID WINAPI	TpCallbackLeaveCriticalSectionOnCompletion (TP_CALLBACK_INSTANCE *instance, CRITICAL_SECTION *crit)
NTSTATUS WINAPI	TpCallbackMayRunLong (TP_CALLBACK_INSTANCE *instance)
VOID WINAPI	TpCallbackReleaseMutexOnCompletion (TP_CALLBACK_INSTANCE *instance, HANDLE mutex)
VOID WINAPI	TpCallbackReleaseSemaphoreOnCompletion (TP_CALLBACK_INSTANCE *instance, HANDLE semaphore, DWORD count)
VOID WINAPI	TpCallbackSetEventOnCompletion (TP_CALLBACK_INSTANCE *instance, HANDLE event)
VOID WINAPI	TpCallbackUnloadDllOnCompletion (TP_CALLBACK_INSTANCE *instance, HMODULE module)
VOID WINAPI	TpDisassociateCallback (TP_CALLBACK_INSTANCE *instance)
BOOL WINAPI	TpIsTimerSet (TP_TIMER *timer)
VOID WINAPI	TpPostWork (TP_WORK *work)
VOID WINAPI	TpReleaseCleanupGroup (TP_CLEANUP_GROUP *group)
VOID WINAPI	TpReleaseCleanupGroupMembers (TP_CLEANUP_GROUP *group, BOOL cancel_pending, PVOID userdata)
void WINAPI	TpReleaseIoCompletion (TP_IO *io)
VOID WINAPI	TpReleasePool (TP_POOL *pool)
VOID WINAPI	TpReleaseTimer (TP_TIMER *timer)
VOID WINAPI	TpReleaseWait (TP_WAIT *wait)
VOID WINAPI	TpReleaseWork (TP_WORK *work)
VOID WINAPI	TpSetPoolMaxThreads (TP_POOL *pool, DWORD maximum)
BOOL WINAPI	TpSetPoolMinThreads (TP_POOL *pool, DWORD minimum)
VOID WINAPI	TpSetTimer (TP_TIMER *timer, LARGE_INTEGER *timeout, LONG period, LONG window_length)
VOID WINAPI	TpSetWait (TP_WAIT *wait, HANDLE handle, LARGE_INTEGER *timeout)
NTSTATUS WINAPI	TpSimpleTryPost (PTP_SIMPLE_CALLBACK callback, PVOID userdata, TP_CALLBACK_ENVIRON *environment)
void WINAPI	TpStartAsyncIoOperation (TP_IO *io)
void WINAPI	TpWaitForIoCompletion (TP_IO *io, BOOL cancel_pending)
VOID WINAPI	TpWaitForTimer (TP_TIMER *timer, BOOL cancel_pending)
VOID WINAPI	TpWaitForWait (TP_WAIT *wait, BOOL cancel_pending)
VOID WINAPI	TpWaitForWork (TP_WORK *work, BOOL cancel_pending)
NTSTATUS WINAPI	TpSetPoolStackInformation (TP_POOL *pool, TP_POOL_STACK_INFORMATION *stack_info)
NTSTATUS WINAPI	TpQueryPoolStackInformation (TP_POOL *pool, TP_POOL_STACK_INFORMATION *stack_info)
static void CALLBACK	rtl_wait_callback (TP_CALLBACK_INSTANCE *instance, void *userdata, TP_WAIT *wait, TP_WAIT_RESULT result)
NTSTATUS WINAPI	RtlRegisterWait (HANDLE *out, HANDLE handle, RTL_WAITORTIMERCALLBACKFUNC callback, void *context, ULONG milliseconds, ULONG flags)
NTSTATUS WINAPI	RtlDeregisterWaitEx (HANDLE handle, HANDLE event)
NTSTATUS WINAPI	RtlDeregisterWait (HANDLE WaitHandle)

Variables
static RTL_CRITICAL_SECTION_DEBUG	critsect_compl_debug
struct {
HANDLE   compl_port
RTL_CRITICAL_SECTION   threadpool_compl_cs
}	old_threadpool
static RTL_CRITICAL_SECTION_DEBUG	timerqueue_debug
struct {
CRITICAL_SECTION   cs
LONG   objcount
BOOL   thread_running
struct list   pending_timers
RTL_CONDITION_VARIABLE   update_event
}	timerqueue
static RTL_CRITICAL_SECTION_DEBUG	waitqueue_debug
struct {
CRITICAL_SECTION   cs
LONG   num_buckets
struct list   buckets
}	waitqueue
static RTL_CRITICAL_SECTION_DEBUG	ioqueue_debug
struct {
CRITICAL_SECTION   cs
LONG   objcount
BOOL   thread_running
HANDLE   port
RTL_CONDITION_VARIABLE   update_event
}	ioqueue
static struct threadpool *	default_threadpool = NULL

Macro Definition Documentation

EXPIRE_NEVER

#define EXPIRE_NEVER   (~(ULONGLONG)0)

Definition at line 84 of file threadpool.c.

MAXIMUM_WAITQUEUE_OBJECTS

#define MAXIMUM_WAITQUEUE_OBJECTS   (MAXIMUM_WAIT_OBJECTS - 1)

Definition at line 145 of file threadpool.c.

THREADPOOL_WORKER_TIMEOUT

#define THREADPOOL_WORKER_TIMEOUT   5000

Definition at line 144 of file threadpool.c.

TIMER_QUEUE_MAGIC

#define TIMER_QUEUE_MAGIC   0x516d6954 /* TimQ */

Definition at line 85 of file threadpool.c.

WIN32_NO_STATUS

#define WIN32_NO_STATUS

Definition at line 63 of file threadpool.c.

Enumeration Type Documentation

threadpool_objtype

enum threadpool_objtype

Enumerator
TP_OBJECT_TYPE_SIMPLE
TP_OBJECT_TYPE_WORK
TP_OBJECT_TYPE_TIMER
TP_OBJECT_TYPE_WAIT
TP_OBJECT_TYPE_IO

Definition at line 166 of file threadpool.c.

{
    TP_OBJECT_TYPE_SIMPLE,
    TP_OBJECT_TYPE_WORK,
    TP_OBJECT_TYPE_TIMER,
    TP_OBJECT_TYPE_WAIT,
    TP_OBJECT_TYPE_IO,
};

Function Documentation

array_reserve()

static BOOL array_reserve ( void **  elements, unsigned int *  capacity, unsigned int  count, unsigned int  size  )

static

Definition at line 446 of file threadpool.c.

{
    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;
}

Referenced by ioqueue_thread_proc().

get_nt_timeout()

static PLARGE_INTEGER get_nt_timeout ( PLARGE_INTEGER  pTime, ULONG  timeout  )

inlinestatic

Definition at line 632 of file threadpool.c.

{
    if (timeout == INFINITE) return NULL;
    pTime->QuadPart = (ULONGLONG)timeout * -10000;
    return pTime;
}

Referenced by RtlRegisterWait(), and timer_queue_thread_proc().

get_timer_queue()

static struct timer_queue * get_timer_queue ( HANDLE  TimerQueue )

static

Definition at line 957 of file threadpool.c.

{
    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;
    }
}

Referenced by RtlCreateTimer().

impl_from_TP_CALLBACK_INSTANCE()

static struct threadpool_instance * impl_from_TP_CALLBACK_INSTANCE ( TP_CALLBACK_INSTANCE *  instance )

inlinestatic

Definition at line 430 of file threadpool.c.

{
    return (struct threadpool_instance *)instance;
}

Referenced by TpCallbackLeaveCriticalSectionOnCompletion(), TpCallbackMayRunLong(), TpCallbackReleaseMutexOnCompletion(), TpCallbackReleaseSemaphoreOnCompletion(), TpCallbackSetEventOnCompletion(), TpCallbackUnloadDllOnCompletion(), and TpDisassociateCallback().

impl_from_TP_CLEANUP_GROUP()

static struct threadpool_group * impl_from_TP_CLEANUP_GROUP ( TP_CLEANUP_GROUP *  group )

inlinestatic

Definition at line 425 of file threadpool.c.

{
    return (struct threadpool_group *)group;
}

Referenced by tp_object_initialize(), TpReleaseCleanupGroup(), and TpReleaseCleanupGroupMembers().

impl_from_TP_IO()

static struct threadpool_object * impl_from_TP_IO ( TP_IO *  io )

inlinestatic

Definition at line 418 of file threadpool.c.

{
    struct threadpool_object *object = (struct threadpool_object *)io;
    assert( object->type == TP_OBJECT_TYPE_IO );
    return object;
}

Referenced by TpCancelAsyncIoOperation(), TpReleaseIoCompletion(), TpStartAsyncIoOperation(), and TpWaitForIoCompletion().

impl_from_TP_POOL()

static struct threadpool * impl_from_TP_POOL ( TP_POOL *  pool )

inlinestatic

Definition at line 392 of file threadpool.c.

{
    return (struct threadpool *)pool;
}

Referenced by TpQueryPoolStackInformation(), TpReleasePool(), TpSetPoolMaxThreads(), TpSetPoolMinThreads(), and TpSetPoolStackInformation().

impl_from_TP_TIMER()

static struct threadpool_object * impl_from_TP_TIMER ( TP_TIMER *  timer )

inlinestatic

Definition at line 404 of file threadpool.c.

{
    struct threadpool_object *object = (struct threadpool_object *)timer;
    assert( object->type == TP_OBJECT_TYPE_TIMER );
    return object;
}

Referenced by TpIsTimerSet(), TpReleaseTimer(), TpSetTimer(), and TpWaitForTimer().

impl_from_TP_WAIT()

static struct threadpool_object * impl_from_TP_WAIT ( TP_WAIT *  wait )

inlinestatic

Definition at line 411 of file threadpool.c.

{
    struct threadpool_object *object = (struct threadpool_object *)wait;
    assert( object->type == TP_OBJECT_TYPE_WAIT );
    return object;
}

Referenced by rtl_wait_callback(), RtlRegisterWait(), TpReleaseWait(), TpSetWait(), and TpWaitForWait().

impl_from_TP_WORK()

static struct threadpool_object * impl_from_TP_WORK ( TP_WORK *  work )

inlinestatic

Definition at line 397 of file threadpool.c.

{
    struct threadpool_object *object = (struct threadpool_object *)work;
    assert( object->type == TP_OBJECT_TYPE_WORK );
    return object;
}

Referenced by TpPostWork(), TpReleaseWork(), and TpWaitForWork().

iocp_poller()

static DWORD CALLBACK iocp_poller ( LPVOID  Arg )

static

Definition at line 544 of file threadpool.c.

{
    HANDLE cport = Arg;
 
    while( TRUE )
    {
        PRTL_OVERLAPPED_COMPLETION_ROUTINE callback;
        LPVOID overlapped;
        IO_STATUS_BLOCK iosb;
#ifdef __REACTOS__
        NTSTATUS res = NtRemoveIoCompletion( cport, (PVOID)&callback, (PVOID)&overlapped, &iosb, NULL );
#else
        NTSTATUS res = NtRemoveIoCompletion( cport, (PULONG_PTR)&callback, (PULONG_PTR)&overlapped, &iosb, NULL );
#endif
        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;
}

Referenced by RtlSetIoCompletionCallback().

ioqueue_thread_proc()

static void CALLBACK ioqueue_thread_proc ( void *  param )

static

Definition at line 1618 of file threadpool.c.

{
    struct io_completion *completion;
    struct threadpool_object *io;
    IO_STATUS_BLOCK iosb;
#ifdef __REACTOS__
    PVOID key, value;
#else
    ULONG_PTR key, value;
#endif
    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;
#ifdef __REACTOS__
                completion->cvalue = (ULONG_PTR)value;
#else
                completion->cvalue = value;
#endif
 
                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 );
 
#ifdef __REACTOS__
    return STATUS_SUCCESS;
#endif
}

Referenced by tp_ioqueue_lock().

object_is_finished()

static BOOL object_is_finished ( struct threadpool_object *  object, BOOL  group  )

static

Definition at line 2187 of file threadpool.c.

{
    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;
}

Referenced by tp_object_execute(), tp_object_wait(), TpCancelAsyncIoOperation(), and TpDisassociateCallback().

process_rtl_work_item()

static void CALLBACK process_rtl_work_item ( TP_CALLBACK_INSTANCE *  instance, void *  userdata  )

static

Definition at line 484 of file threadpool.c.

{
    struct rtl_work_item *item = userdata;
 
    TRACE("executing %p(%p)\n", item->function, item->context);
    item->function( item->context );
 
    RtlFreeHeap( GetProcessHeap(), 0, item );
}

Referenced by RtlQueueWorkItem().

queue_add_timer()

static void queue_add_timer ( struct queue_timer *  t, ULONGLONG  time, BOOL  set_event  )

static

Definition at line 689 of file threadpool.c.

{
    /* 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);
}

Referenced by queue_move_timer(), and RtlCreateTimer().

queue_current_time()

static ULONGLONG queue_current_time ( void  )

inlinestatic

Definition at line 682 of file threadpool.c.

{
    LARGE_INTEGER now, freq;
    NtQueryPerformanceCounter(&now, &freq);
    return now.QuadPart * 1000 / freq.QuadPart;
}

Referenced by queue_get_timeout(), queue_timer_expire(), RtlCreateTimer(), and RtlUpdateTimer().

queue_destroy_timer()

static void queue_destroy_timer ( struct queue_timer *  t )

static

Definition at line 840 of file threadpool.c.

{
    /* 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);
}

Referenced by RtlDeleteTimer(), and RtlDeleteTimerQueueEx().

queue_get_timeout()

static ULONG queue_get_timeout ( struct timer_queue *  q )

static

Definition at line 768 of file threadpool.c.

{
    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;
}

Referenced by timer_queue_thread_proc().

queue_move_timer()

static void queue_move_timer ( struct queue_timer *  t, ULONGLONG  time, BOOL  set_event  )

inlinestatic

Definition at line 715 of file threadpool.c.

{
    /* We MUST hold the queue cs while calling this function.  */
    list_remove(&t->entry);
    queue_add_timer(t, time, set_event);
}

Referenced by queue_destroy_timer(), queue_timer_expire(), and RtlUpdateTimer().

queue_remove_timer()

static void queue_remove_timer ( struct queue_timer *  t )

static

Definition at line 641 of file threadpool.c.

{
    /* 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);
}

Referenced by queue_destroy_timer(), and timer_cleanup_callback().

queue_timer_expire()

static void queue_timer_expire ( struct timer_queue *  q )

static

Definition at line 723 of file threadpool.c.

{
    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);
        }
    }
}

Referenced by timer_queue_thread_proc().

rtl_wait_callback()

static void CALLBACK rtl_wait_callback ( TP_CALLBACK_INSTANCE *  instance, void *  userdata, TP_WAIT *  wait, TP_WAIT_RESULT  result  )

static

Definition at line 3383 of file threadpool.c.

{
    struct threadpool_object *object = impl_from_TP_WAIT(wait);
    object->u.wait.rtl_callback( userdata, result != STATUS_WAIT_0 );
}

Referenced by RtlRegisterWait().

RtlCreateTimer()

NTSTATUS WINAPI RtlCreateTimer	(	HANDLE	TimerQueue,
		HANDLE *	NewTimer,
		RTL_WAITORTIMERCALLBACKFUNC	Callback,
		PVOID	Parameter,
		DWORD	DueTime,
		DWORD	Period,
		ULONG	Flags
	)

Definition at line 1006 of file threadpool.c.

{
    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;
}

RtlCreateTimerQueue()

NTSTATUS WINAPI RtlCreateTimerQueue

(

PHANDLE

NewTimerQueue

)

Definition at line 867 of file threadpool.c.

{
    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;
}

Referenced by BasepCreateDefaultTimerQueue(), CreateTimerQueue(), and get_timer_queue().

RtlDeleteTimer()

NTSTATUS WINAPI RtlDeleteTimer	(	HANDLE	TimerQueue,
		HANDLE	Timer,
		HANDLE	CompletionEvent
	)

Definition at line 1101 of file threadpool.c.

{
    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;
}

RtlDeleteTimerQueueEx()

NTSTATUS WINAPI RtlDeleteTimerQueueEx	(	HANDLE	TimerQueue,
		HANDLE	CompletionEvent
	)

Definition at line 913 of file threadpool.c.

{
    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;
}

Referenced by get_timer_queue().

RtlDeregisterWait()

NTSTATUS WINAPI RtlDeregisterWait

(

HANDLE

WaitHandle

)

Definition at line 3501 of file threadpool.c.

{
    return RtlDeregisterWaitEx(WaitHandle, NULL);
}

RtlDeregisterWaitEx()

NTSTATUS WINAPI RtlDeregisterWaitEx	(	HANDLE	handle,
		HANDLE	event
	)

Definition at line 3460 of file threadpool.c.

{
    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;
}

RtlQueueWorkItem()

NTSTATUS WINAPI RtlQueueWorkItem	(	PRTL_WORK_ITEM_ROUTINE	function,
		PVOID	context,
		ULONG	flags
	)

Definition at line 516 of file threadpool.c.

{
    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;
}

RtlRegisterWait()

NTSTATUS WINAPI RtlRegisterWait	(	HANDLE *	out,
		HANDLE	handle,
		RTL_WAITORTIMERCALLBACKFUNC	callback,
		void *	context,
		ULONG	milliseconds,
		ULONG	flags
	)

Definition at line 3414 of file threadpool.c.

{
    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;
}

RtlSetIoCompletionCallback()

NTSTATUS WINAPI RtlSetIoCompletionCallback	(	HANDLE	FileHandle,
		PRTL_OVERLAPPED_COMPLETION_ROUTINE	Function,
		ULONG	Flags
	)

Definition at line 594 of file threadpool.c.

{
    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 );
}

RtlUpdateTimer()

NTSTATUS WINAPI RtlUpdateTimer	(	HANDLE	TimerQueue,
		HANDLE	Timer,
		DWORD	DueTime,
		DWORD	Period
	)

Definition at line 1065 of file threadpool.c.

{
    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;
}

set_thread_name()

static void set_thread_name ( const WCHAR *  name )

static

Definition at line 473 of file threadpool.c.

{
#ifndef __REACTOS__ // This is impossible on non vista+
    THREAD_NAME_INFORMATION info;
 
    RtlInitUnicodeString(&info.ThreadName, name);
    NtSetInformationThread(GetCurrentThread(), ThreadNameInformation, &info, sizeof(info));
#endif
}

Referenced by ioqueue_thread_proc(), threadpool_worker_proc(), timer_queue_thread_proc(), timerqueue_thread_proc(), and waitqueue_thread_proc().

threadpool_get_next_item()

static struct list * threadpool_get_next_item ( const struct threadpool *  pool )

static

Definition at line 2295 of file threadpool.c.

{
    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;
}

Referenced by threadpool_worker_proc().

threadpool_worker_proc()

static void CALLBACK threadpool_worker_proc ( void *  param )

static

Definition at line 2474 of file threadpool.c.

{
    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 );
#ifdef __REACTOS__
    return STATUS_SUCCESS;
#endif
}

Referenced by tp_new_worker_thread().

timer_callback_wrapper()

static DWORD WINAPI timer_callback_wrapper ( LPVOID  p )

static

Definition at line 674 of file threadpool.c.

{
    struct queue_timer *t = p;
    t->callback(t->param, TRUE);
    timer_cleanup_callback(t);
    return 0;
}

Referenced by queue_timer_expire().

timer_cleanup_callback()

static void timer_cleanup_callback ( struct queue_timer *  t )

static

Definition at line 660 of file threadpool.c.

{
    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);
}

Referenced by queue_timer_expire(), and timer_callback_wrapper().

timer_queue_thread_proc()

static void WINAPI timer_queue_thread_proc ( LPVOID  p )

static

Definition at line 793 of file threadpool.c.

{
    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 );
#ifdef __REACTOS__
    return STATUS_SUCCESS;
#endif
}

Referenced by RtlCreateTimerQueue().

timerqueue_thread_proc()

static void CALLBACK timerqueue_thread_proc ( void *  param )

static

Definition at line 1147 of file threadpool.c.

{
    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 );
#ifdef __REACTOS__
    return STATUS_SUCCESS;
#endif
}

Referenced by tp_timerqueue_lock().

tp_alloc_wait()

static NTSTATUS tp_alloc_wait ( TP_WAIT **  out, PTP_WAIT_CALLBACK  callback, PVOID  userdata, TP_CALLBACK_ENVIRON *  environment, DWORD  flags  )

static

Definition at line 2641 of file threadpool.c.

{
    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;
}

Referenced by RtlRegisterWait(), and TpAllocWait().

tp_group_alloc()

static NTSTATUS tp_group_alloc ( struct threadpool_group **  out )

static

Definition at line 1963 of file threadpool.c.

{
    struct threadpool_group *group;
 
    group = RtlAllocateHeap( GetProcessHeap(), 0, sizeof(*group) );
    if (!group)
        return STATUS_NO_MEMORY;
 
    group->refcount     = 1;
    group->shutdown     = FALSE;
 
#ifdef __REACTOS__
    RtlInitializeCriticalSection( &group->cs );
#else
    RtlInitializeCriticalSectionEx( &group->cs, 0, RTL_CRITICAL_SECTION_FLAG_FORCE_DEBUG_INFO );
 
    group->cs.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ": threadpool_group.cs");
#endif
 
    list_init( &group->members );
 
    TRACE( "allocated group %p\n", group );
 
    *out = group;
    return STATUS_SUCCESS;
}

Referenced by TpAllocCleanupGroup().

tp_group_release()

static BOOL tp_group_release ( struct threadpool_group *  group )

static

Definition at line 2005 of file threadpool.c.

{
    if (InterlockedDecrement( &group->refcount ))
        return FALSE;
 
    TRACE( "destroying group %p\n", group );
 
    assert( group->shutdown );
    assert( list_empty( &group->members ) );
 
#ifndef __REACTOS__
    group->cs.DebugInfo->Spare[0] = 0;
#endif
    RtlDeleteCriticalSection( &group->cs );
 
    RtlFreeHeap( GetProcessHeap(), 0, group );
    return TRUE;
}

Referenced by tp_object_release(), and TpReleaseCleanupGroup().

tp_group_shutdown()

static void tp_group_shutdown ( struct threadpool_group *  group )

static

Definition at line 1995 of file threadpool.c.

{
    group->shutdown = TRUE;
}

Referenced by TpReleaseCleanupGroup().

tp_ioqueue_lock()

static NTSTATUS tp_ioqueue_lock ( struct threadpool_object *  io, HANDLE  file  )

static

Definition at line 1727 of file threadpool.c.

{
    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;
 
#ifdef __REACTOS__
        info.Port = ioqueue.port;
        info.Key = io;
#else
        info.CompletionPort = ioqueue.port;
        info.CompletionKey = (ULONG_PTR)io;
#endif
 
        status = NtSetInformationFile( file, &iosb, &info, sizeof(info), FileCompletionInformation );
    }
 
    if (status == STATUS_SUCCESS)
    {
        if (!ioqueue.objcount++)
            RtlWakeConditionVariable( &ioqueue.update_event );
    }
 
    RtlLeaveCriticalSection( &ioqueue.cs );
    return status;
}

Referenced by TpAllocIoCompletion().

tp_ioqueue_unlock()

static void tp_ioqueue_unlock ( struct threadpool_object *  io )

static

Definition at line 2221 of file threadpool.c.

{
    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 );
}

Referenced by tp_object_prepare_shutdown().

tp_new_worker_thread()

static NTSTATUS tp_new_worker_thread ( struct threadpool *  pool )

static

Definition at line 1245 of file threadpool.c.

{
    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;
}

Referenced by tp_object_submit(), tp_threadpool_lock(), TpCallbackMayRunLong(), and TpSetPoolMinThreads().

tp_object_cancel()

static void tp_object_cancel ( struct threadpool_object *  object )

static

Definition at line 2161 of file threadpool.c.

{
    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 );
}

Referenced by TpReleaseCleanupGroupMembers(), TpWaitForIoCompletion(), TpWaitForTimer(), TpWaitForWait(), and TpWaitForWork().

tp_object_execute()

static void tp_object_execute ( struct threadpool_object *  object, BOOL  wait_thread  )

static

Definition at line 2315 of file threadpool.c.

{
    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 );
    }
}

Referenced by threadpool_worker_proc(), and waitqueue_thread_proc().

tp_object_initialize()

static void tp_object_initialize ( struct threadpool_object *  object, struct threadpool *  pool, PVOID  userdata, TP_CALLBACK_ENVIRON *  environment  )

static

Definition at line 2029 of file threadpool.c.

{
    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;
#ifndef __REACTOS__ //Windows 7 stuff
        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) );
        }
#endif
        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 );
}

Referenced by tp_alloc_wait(), TpAllocIoCompletion(), TpAllocTimer(), TpAllocWork(), and TpSimpleTryPost().

tp_object_prepare_shutdown()

static void tp_object_prepare_shutdown ( struct threadpool_object *  object )

static

Definition at line 2240 of file threadpool.c.

{
    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 );
}

Referenced by tp_object_execute(), TpReleaseCleanupGroupMembers(), TpReleaseIoCompletion(), TpReleaseTimer(), TpReleaseWait(), and TpReleaseWork().

tp_object_prio_queue()

static void tp_object_prio_queue ( struct threadpool_object *  object )

static

Definition at line 2110 of file threadpool.c.

{
    ++object->pool->num_busy_workers;
    list_add_tail( &object->pool->pools[object->priority], &object->pool_entry );
}

Referenced by threadpool_worker_proc(), and tp_object_submit().

tp_object_release()

static BOOL tp_object_release ( struct threadpool_object *  object )

static

Definition at line 2255 of file threadpool.c.

{
    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;
}

Referenced by ioqueue_thread_proc(), threadpool_worker_proc(), tp_object_cancel(), tp_object_initialize(), TpReleaseCleanupGroupMembers(), TpReleaseIoCompletion(), TpReleaseTimer(), TpReleaseWait(), TpReleaseWork(), and waitqueue_thread_proc().

tp_object_submit()

static void tp_object_submit ( struct threadpool_object *  object, BOOL  signaled  )

static

Definition at line 2122 of file threadpool.c.

{
    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 );
}

Referenced by ioqueue_thread_proc(), timerqueue_thread_proc(), tp_object_initialize(), TpPostWork(), TpSetTimer(), and waitqueue_thread_proc().

tp_object_wait()

static void tp_object_wait ( struct threadpool_object *  object, BOOL  group_wait  )

static

Definition at line 2206 of file threadpool.c.

{
    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 );
}

Referenced by TpReleaseCleanupGroupMembers(), TpWaitForIoCompletion(), TpWaitForTimer(), TpWaitForWait(), and TpWaitForWork().

tp_threadpool_alloc()

static NTSTATUS tp_threadpool_alloc ( struct threadpool **  out )

static

Definition at line 1785 of file threadpool.c.

{
#ifdef __REACTOS__
    IMAGE_NT_HEADERS *nt = RtlImageNtHeader( NtCurrentTeb()->ProcessEnvironmentBlock->ImageBaseAddress );
#else
    IMAGE_NT_HEADERS *nt = RtlImageNtHeader( NtCurrentTeb()->Peb->ImageBaseAddress );
#endif
    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;
 
#ifdef __REACTOS__
    RtlInitializeCriticalSection( &pool->cs );
#else
    RtlInitializeCriticalSectionEx( &pool->cs, 0, RTL_CRITICAL_SECTION_FLAG_FORCE_DEBUG_INFO );
 
    pool->cs.DebugInfo->Spare[0] = (DWORD_PTR)(__FILE__ ": threadpool.cs");
#endif
 
    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;
}

Referenced by tp_threadpool_lock(), and TpAllocPool().

tp_threadpool_lock()

static NTSTATUS tp_threadpool_lock ( struct threadpool **  out, TP_CALLBACK_ENVIRON *  environment  )

static

Definition at line 1877 of file threadpool.c.

{
    struct threadpool *pool = NULL;
    NTSTATUS status = STATUS_SUCCESS;
 
    if (environment)
    {
#ifndef __REACTOS__ //Windows 7 stuff 
        /* 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;
            }
        }
#endif
        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;
}

Referenced by tp_alloc_wait(), TpAllocIoCompletion(), TpAllocTimer(), TpAllocWork(), and TpSimpleTryPost().

tp_threadpool_release()

static BOOL tp_threadpool_release ( struct threadpool *  pool )

static

Definition at line 1848 of file threadpool.c.

{
    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] ) );
#ifndef __REACTOS__
    pool->cs.DebugInfo->Spare[0] = 0;
#endif
    RtlDeleteCriticalSection( &pool->cs );
 
    RtlFreeHeap( GetProcessHeap(), 0, pool );
    return TRUE;
}

Referenced by threadpool_worker_proc(), tp_threadpool_lock(), tp_threadpool_unlock(), and TpReleasePool().

tp_threadpool_shutdown()

static void tp_threadpool_shutdown ( struct threadpool *  pool )

static

Definition at line 1835 of file threadpool.c.

{
    assert( pool != default_threadpool );
 
    pool->shutdown = TRUE;
    RtlWakeAllConditionVariable( &pool->update_event );
}

Referenced by tp_threadpool_lock(), and TpReleasePool().

tp_threadpool_unlock()

static void tp_threadpool_unlock ( struct threadpool *  pool )

static

Definition at line 1950 of file threadpool.c.

{
    RtlEnterCriticalSection( &pool->cs );
    pool->objcount--;
    RtlLeaveCriticalSection( &pool->cs );
    tp_threadpool_release( pool );
}

Referenced by tp_alloc_wait(), tp_object_release(), TpAllocIoCompletion(), and TpAllocTimer().

tp_timerqueue_lock()

static NTSTATUS tp_timerqueue_lock ( struct threadpool_object *  timer )

static

Definition at line 1268 of file threadpool.c.

{
    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;
}

Referenced by TpAllocTimer().

tp_timerqueue_unlock()

static void tp_timerqueue_unlock ( struct threadpool_object *  timer )

static

Definition at line 1310 of file threadpool.c.

{
    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 );
}

Referenced by tp_object_prepare_shutdown().

tp_waitqueue_lock()

static NTSTATUS tp_waitqueue_lock ( struct threadpool_object *  wait )

static

Definition at line 1518 of file threadpool.c.

{
    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,
                                  (PTHREAD_START_ROUTINE)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;
}

Referenced by tp_alloc_wait().

tp_waitqueue_unlock()

static void tp_waitqueue_unlock ( struct threadpool_object *  wait )

static

Definition at line 1596 of file threadpool.c.

{
    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 );
}

Referenced by tp_object_prepare_shutdown().

TpAllocCleanupGroup()

NTSTATUS WINAPI TpAllocCleanupGroup

(

TP_CLEANUP_GROUP **

out

)

Definition at line 2537 of file threadpool.c.

{
    TRACE( "%p\n", out );
 
    return tp_group_alloc( (struct threadpool_group **)out );
}

Referenced by init_threadpool().

TpAllocIoCompletion()

NTSTATUS WINAPI TpAllocIoCompletion	(	TP_IO **	out,
		HANDLE	file,
		PTP_IO_CALLBACK	callback,
		void *	userdata,
		TP_CALLBACK_ENVIRON *	environment
	)

Definition at line 2547 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpAllocPool()

NTSTATUS WINAPI TpAllocPool	(	TP_POOL **	out,
		PVOID	reserved
	)

Definition at line 2591 of file threadpool.c.

{
    TRACE( "%p %p\n", out, reserved );
 
    if (reserved)
        FIXME( "reserved argument is nonzero (%p)\n", reserved );
 
    return tp_threadpool_alloc( (struct threadpool **)out );
}

Referenced by init_threadpool().

TpAllocTimer()

NTSTATUS WINAPI TpAllocTimer	(	TP_TIMER **	out,
		PTP_TIMER_CALLBACK	callback,
		PVOID	userdata,
		TP_CALLBACK_ENVIRON *	environment
	)

Definition at line 2604 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpAllocWait()

NTSTATUS WINAPI TpAllocWait	(	TP_WAIT **	out,
		PTP_WAIT_CALLBACK	callback,
		PVOID	userdata,
		TP_CALLBACK_ENVIRON *	environment
	)

Definition at line 2680 of file threadpool.c.

{
    TRACE( "%p %p %p %p\n", out, callback, userdata, environment );
    return tp_alloc_wait( out, callback, userdata, environment, WT_EXECUTEONLYONCE );
}

Referenced by init_threadpool().

TpAllocWork()

NTSTATUS WINAPI TpAllocWork	(	TP_WORK **	out,
		PTP_WORK_CALLBACK	callback,
		PVOID	userdata,
		TP_CALLBACK_ENVIRON *	environment
	)

Definition at line 2690 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpCallbackLeaveCriticalSectionOnCompletion()

VOID WINAPI TpCallbackLeaveCriticalSectionOnCompletion	(	TP_CALLBACK_INSTANCE *	instance,
		CRITICAL_SECTION *	crit
	)

Definition at line 2743 of file threadpool.c.

{
    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()

NTSTATUS WINAPI TpCallbackMayRunLong

(

TP_CALLBACK_INSTANCE *

instance

)

Definition at line 2756 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpCallbackReleaseMutexOnCompletion()

VOID WINAPI TpCallbackReleaseMutexOnCompletion	(	TP_CALLBACK_INSTANCE *	instance,
		HANDLE	mutex
	)

Definition at line 2798 of file threadpool.c.

{
    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()

VOID WINAPI TpCallbackReleaseSemaphoreOnCompletion	(	TP_CALLBACK_INSTANCE *	instance,
		HANDLE	semaphore,
		DWORD	count
	)

Definition at line 2811 of file threadpool.c.

{
    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;
    }
}

Referenced by init_threadpool().

TpCallbackSetEventOnCompletion()

VOID WINAPI TpCallbackSetEventOnCompletion	(	TP_CALLBACK_INSTANCE *	instance,
		HANDLE	event
	)

Definition at line 2827 of file threadpool.c.

{
    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()

VOID WINAPI TpCallbackUnloadDllOnCompletion	(	TP_CALLBACK_INSTANCE *	instance,
		HMODULE	module
	)

Definition at line 2840 of file threadpool.c.

{
    struct threadpool_instance *this = impl_from_TP_CALLBACK_INSTANCE( instance );
 
    TRACE( "%p %p\n", instance, module );
 
    if (!this->cleanup.library)
        this->cleanup.library = module;
}

TpCancelAsyncIoOperation()

void WINAPI TpCancelAsyncIoOperation

(

TP_IO *

io

)

Definition at line 2721 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpDisassociateCallback()

VOID WINAPI TpDisassociateCallback

(

TP_CALLBACK_INSTANCE *

instance

)

Definition at line 2853 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpIsTimerSet()

BOOL WINAPI TpIsTimerSet

(

TP_TIMER *

timer

)

Definition at line 2884 of file threadpool.c.

{
    struct threadpool_object *this = impl_from_TP_TIMER( timer );
 
    TRACE( "%p\n", timer );
 
    return this->u.timer.timer_set;
}

Referenced by init_threadpool().

TpPostWork()

VOID WINAPI TpPostWork

(

TP_WORK *

work

)

Definition at line 2896 of file threadpool.c.

{
    struct threadpool_object *this = impl_from_TP_WORK( work );
 
    TRACE( "%p\n", work );
 
    tp_object_submit( this, FALSE );
}

Referenced by init_threadpool().

TpQueryPoolStackInformation()

NTSTATUS WINAPI TpQueryPoolStackInformation	(	TP_POOL *	pool,
		TP_POOL_STACK_INFORMATION *	stack_info
	)

Definition at line 3367 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpReleaseCleanupGroup()

VOID WINAPI TpReleaseCleanupGroup

(

TP_CLEANUP_GROUP *

group

)

Definition at line 2908 of file threadpool.c.

{
    struct threadpool_group *this = impl_from_TP_CLEANUP_GROUP( group );
 
    TRACE( "%p\n", group );
 
    tp_group_shutdown( this );
    tp_group_release( this );
}

Referenced by init_threadpool().

TpReleaseCleanupGroupMembers()

VOID WINAPI TpReleaseCleanupGroupMembers	(	TP_CLEANUP_GROUP *	group,
		BOOL	cancel_pending,
		PVOID	userdata
	)

Definition at line 2921 of file threadpool.c.

{
    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 );
    }
}

Referenced by init_threadpool().

TpReleaseIoCompletion()

void WINAPI TpReleaseIoCompletion

(

TP_IO *

io

)

Definition at line 2994 of file threadpool.c.

{
    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 );
    }
}

Referenced by init_threadpool().

TpReleasePool()

VOID WINAPI TpReleasePool

(

TP_POOL *

pool

)

Definition at line 3017 of file threadpool.c.

{
    struct threadpool *this = impl_from_TP_POOL( pool );
 
    TRACE( "%p\n", pool );
 
    tp_threadpool_shutdown( this );
    tp_threadpool_release( this );
}

Referenced by init_threadpool().

TpReleaseTimer()

VOID WINAPI TpReleaseTimer

(

TP_TIMER *

timer

)

Definition at line 3030 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpReleaseWait()

VOID WINAPI TpReleaseWait

(

TP_WAIT *

wait

)

Definition at line 3044 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool(), and RtlDeregisterWaitEx().

TpReleaseWork()

VOID WINAPI TpReleaseWork

(

TP_WORK *

work

)

Definition at line 3058 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpSetPoolMaxThreads()

VOID WINAPI TpSetPoolMaxThreads	(	TP_POOL *	pool,
		DWORD	maximum
	)

Definition at line 3072 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpSetPoolMinThreads()

BOOL WINAPI TpSetPoolMinThreads	(	TP_POOL *	pool,
		DWORD	minimum
	)

Definition at line 3087 of file threadpool.c.

{
    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;
}

TpSetPoolStackInformation()

NTSTATUS WINAPI TpSetPoolStackInformation	(	TP_POOL *	pool,
		TP_POOL_STACK_INFORMATION *	stack_info
	)

Definition at line 3348 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool().

TpSetTimer()

VOID WINAPI TpSetTimer	(	TP_TIMER *	timer,
		LARGE_INTEGER *	timeout,
		LONG	period,
		LONG	window_length
	)

Definition at line 3116 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpSetWait()

VOID WINAPI TpSetWait	(	TP_WAIT *	wait,
		HANDLE	handle,
		LARGE_INTEGER *	timeout
	)

Definition at line 3194 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool(), RtlDeregisterWaitEx(), and RtlRegisterWait().

TpSimpleTryPost()

NTSTATUS WINAPI TpSimpleTryPost	(	PTP_SIMPLE_CALLBACK	callback,
		PVOID	userdata,
		TP_CALLBACK_ENVIRON *	environment
	)

Definition at line 3246 of file threadpool.c.

{
    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;
}

Referenced by init_threadpool(), and RtlQueueWorkItem().

TpStartAsyncIoOperation()

void WINAPI TpStartAsyncIoOperation

(

TP_IO *

io

)

Definition at line 3276 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpWaitForIoCompletion()

void WINAPI TpWaitForIoCompletion	(	TP_IO *	io,
		BOOL	cancel_pending
	)

Definition at line 3292 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpWaitForTimer()

VOID WINAPI TpWaitForTimer	(	TP_TIMER *	timer,
		BOOL	cancel_pending
	)

Definition at line 3306 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

TpWaitForWait()

VOID WINAPI TpWaitForWait	(	TP_WAIT *	wait,
		BOOL	cancel_pending
	)

Definition at line 3320 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool(), and RtlDeregisterWaitEx().

TpWaitForWork()

VOID WINAPI TpWaitForWork	(	TP_WORK *	work,
		BOOL	cancel_pending
	)

Definition at line 3334 of file threadpool.c.

{
    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 );
}

Referenced by init_threadpool().

waitqueue_thread_proc()

static void CALLBACK waitqueue_thread_proc ( void *  param )

static

Definition at line 1342 of file threadpool.c.

{
    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 );
}

Referenced by tp_waitqueue_lock().

WINE_DEFAULT_DEBUG_CHANNEL()

WINE_DEFAULT_DEBUG_CHANNEL

(

threadpool

)

Variable Documentation

buckets

struct list buckets

Definition at line 329 of file threadpool.c.

Referenced by add_hash(), delete_hash_entry(), hash_lookup(), and new_hash().

compl_port

HANDLE compl_port

Definition at line 93 of file threadpool.c.

critsect_compl_debug

static RTL_CRITICAL_SECTION_DEBUG critsect_compl_debug

static

Initial value:

=
{
    0, 0, &old_threadpool.threadpool_compl_cs,
    { &critsect_compl_debug.ProcessLocksList, &critsect_compl_debug.ProcessLocksList },
      0, 0, { (DWORD_PTR)(__FILE__ ": threadpool_compl_cs") }
}

Definition at line 88 of file threadpool.c.

cs

CRITICAL_SECTION cs

Definition at line 290 of file threadpool.c.

Referenced by TpQueryPoolStackInformation(), TpReleaseCleanupGroupMembers(), TpSetPoolMaxThreads(), TpSetPoolMinThreads(), and TpSetPoolStackInformation().

default_threadpool

struct threadpool* default_threadpool = NULL

static

Definition at line 444 of file threadpool.c.

Referenced by tp_threadpool_lock(), and tp_threadpool_shutdown().

struct { ... } ioqueue

Initial value:

=
{
 
 
 
    .cs = { &ioqueue_debug, -1, 0, 0, 0, 0 },
 
}

Referenced by ioqueue_thread_proc(), tp_ioqueue_lock(), and tp_ioqueue_unlock().

ioqueue_debug

static RTL_CRITICAL_SECTION_DEBUG ioqueue_debug

static

Initial value:

=
{
    0, 0, &ioqueue.cs,
    { &ioqueue_debug.ProcessLocksList, &ioqueue_debug.ProcessLocksList },
      0, 0, { (DWORD_PTR)(__FILE__ ": ioqueue.cs") }
}

Definition at line 363 of file threadpool.c.

num_buckets

LONG num_buckets

Definition at line 328 of file threadpool.c.

Referenced by hash_table_hash(), and hash_table_init().

objcount

LONG objcount

Definition at line 291 of file threadpool.c.

struct { ... } old_threadpool

Initial value:

=
{
    NULL,                                       
 
 
 
    { &critsect_compl_debug, -1, 0, 0, 0, 0 },  
 
}

Referenced by RtlSetIoCompletionCallback().

pending_timers

struct list pending_timers

Definition at line 293 of file threadpool.c.

port

HANDLE port

Definition at line 371 of file threadpool.c.

thread_running

BOOL thread_running

Definition at line 292 of file threadpool.c.

threadpool_compl_cs

RTL_CRITICAL_SECTION threadpool_compl_cs

Definition at line 94 of file threadpool.c.

struct { ... } timerqueue

Initial value:

=
{
 
 
 
    { &timerqueue_debug, -1, 0, 0, 0, 0 },      
 
    0,                                          
    FALSE,                                      
    LIST_INIT( timerqueue.pending_timers ),     
 
 
 
    RTL_CONDITION_VARIABLE_INIT                 
 
}

Referenced by timerqueue_thread_proc(), tp_timerqueue_lock(), tp_timerqueue_unlock(), and TpSetTimer().

timerqueue_debug

static RTL_CRITICAL_SECTION_DEBUG timerqueue_debug

static

Initial value:

=
{
    0, 0, &timerqueue.cs,
    { &timerqueue_debug.ProcessLocksList, &timerqueue_debug.ProcessLocksList },
      0, 0, { (DWORD_PTR)(__FILE__ ": timerqueue.cs") }
}

Definition at line 285 of file threadpool.c.

update_event

RTL_CONDITION_VARIABLE update_event

Definition at line 294 of file threadpool.c.

struct { ... } waitqueue

Initial value:

=
{
 
 
 
    { &waitqueue_debug, -1, 0, 0, 0, 0 },       
 
    0,                                          
              
}

Referenced by RtlRegisterWait(), tp_object_execute(), tp_waitqueue_lock(), tp_waitqueue_unlock(), TpSetWait(), and waitqueue_thread_proc().

waitqueue_debug

static RTL_CRITICAL_SECTION_DEBUG waitqueue_debug

static

Initial value:

=
{
    0, 0, &waitqueue.cs,
    { &waitqueue_debug.ProcessLocksList, &waitqueue_debug.ProcessLocksList },
      0, 0, { (DWORD_PTR)(__FILE__ ": waitqueue.cs") }
}

Definition at line 322 of file threadpool.c.