[PATCH v2 3/9] d3dx10: Add ID3DX10ThreadPump:AddWorkItem implementation.

[Piotr Caban]

From: Piotr Caban

Signed-off-by: Piotr Caban 
---
 dlls/d3dx10_43/d3dx10_43_main.c | 251 +++++++++++++++++++++++++++++++-
 1 file changed, 248 insertions(+), 3 deletions(-)

diff --git a/dlls/d3dx10_43/d3dx10_43_main.c b/dlls/d3dx10_43/d3dx10_43_main.c
index 357b0257496..d66d83d3e7f 100644
--- a/dlls/d3dx10_43/d3dx10_43_main.c
+++ b/dlls/d3dx10_43/d3dx10_43_main.c
@@ -29,10 +29,13 @@
 #include "winbase.h"
 #include "winuser.h"
 #include "objbase.h"
+#include "winternl.h"

#include "d3d10_1.h"
 #include "d3dx10.h"

+#include "wine/list.h"
+
 WINE_DEFAULT_DEBUG_CHANNEL(d3dx);

/***********************************************************************
@@ -177,10 +180,45 @@ HRESULT WINAPI D3DX10LoadTextureFromTexture(ID3D10Resource *src_texture, D3DX10_
     return E_NOTIMPL;
 }

+struct work_item
+{
+    struct list entry;
+
+    ID3DX10DataLoader *loader;
+    ID3DX10DataProcessor *processor;
+    HRESULT *result;
+    void **object;
+};
+
+static inline void work_item_free(struct work_item *work_item, BOOL cancel)
+{
+    ID3DX10DataLoader_Destroy(work_item->loader);
+    ID3DX10DataProcessor_Destroy(work_item->processor);
+    if (cancel && work_item->result)
+        *work_item->result = S_FALSE;
+    free(work_item);
+}
+
+#define THREAD_PUMP_EXITING UINT_MAX
 struct thread_pump
 {
     ID3DX10ThreadPump ID3DX10ThreadPump_iface;
     LONG refcount;
+
+    SRWLOCK io_lock;
+    UINT io_count;
+    struct list io_queue;
+
+    SRWLOCK proc_lock;
+    UINT proc_count;
+    struct list proc_queue;
+
+    SRWLOCK device_lock;
+    UINT device_count;
+    struct list device_queue;
+
+    int threads_no;
+    HANDLE threads[1];
 };

static inline struct thread_pump *impl_from_ID3DX10ThreadPump(ID3DX10ThreadPump *iface)
@@ -219,11 +257,48 @@ static ULONG WINAPI thread_pump_Release(ID3DX10ThreadPump *iface)
 {
     struct thread_pump *thread_pump = impl_from_ID3DX10ThreadPump(iface);
     ULONG refcount = InterlockedDecrement(&thread_pump->refcount);
+    struct work_item *iter, *next;
+    struct list list;
+    int i;

TRACE("%p decreasing refcount to %lu.\n", iface, refcount);

if (!refcount)
+    {
+        AcquireSRWLockExclusive(&thread_pump->io_lock);
+        thread_pump->io_count = THREAD_PUMP_EXITING;
+        ReleaseSRWLockExclusive(&thread_pump->io_lock);
+        RtlWakeAddressAll((void *)&thread_pump->io_count);
+
+        AcquireSRWLockExclusive(&thread_pump->proc_lock);
+        thread_pump->proc_count = THREAD_PUMP_EXITING;
+        ReleaseSRWLockExclusive(&thread_pump->proc_lock);
+        RtlWakeAddressAll((void *)&thread_pump->proc_count);
+
+        AcquireSRWLockExclusive(&thread_pump->device_lock);
+        thread_pump->device_count = THREAD_PUMP_EXITING;
+        ReleaseSRWLockExclusive(&thread_pump->device_lock);
+
+        for (i = 0; i < thread_pump->threads_no; i++)
+        {
+            if (!thread_pump->threads[i]) continue;
+
+            WaitForSingleObject(thread_pump->threads[i], INFINITE);
+            CloseHandle(thread_pump->threads[i]);
+        }
+
+        list_init(&list);
+        list_move_tail(&list, &thread_pump->io_queue);
+        list_move_tail(&list, &thread_pump->proc_queue);
+        list_move_tail(&list, &thread_pump->device_queue);
+        LIST_FOR_EACH_ENTRY_SAFE(iter, next, &list, struct work_item, entry)
+        {
+            list_remove(&iter->entry);
+            work_item_free(iter, TRUE);
+        }
+
         free(thread_pump);
+    }

return refcount;
 }
@@ -231,9 +306,30 @@ static ULONG WINAPI thread_pump_Release(ID3DX10ThreadPump *iface)
 static HRESULT WINAPI thread_pump_AddWorkItem(ID3DX10ThreadPump *iface, ID3DX10DataLoader *loader,
         ID3DX10DataProcessor *processor, HRESULT *result, void **object)
 {
-    FIXME("iface %p, loader %p, processor %p, result %p, object %p stub!\n",
+    struct thread_pump *thread_pump = impl_from_ID3DX10ThreadPump(iface);
+    struct work_item *work_item;
+
+    TRACE("iface %p, loader %p, processor %p, result %p, object %p.\n",
             iface, loader, processor, result, object);
-    return E_NOTIMPL;
+
+    work_item = malloc(sizeof(*work_item));
+    if (!work_item)
+        return E_OUTOFMEMORY;
+
+    work_item->loader = loader;
+    work_item->processor = processor;
+    work_item->result = result;
+    work_item->object = object;
+
+    if (object)
+        *object = NULL;
+
+    AcquireSRWLockExclusive(&thread_pump->io_lock);
+    thread_pump->io_count++;
+    list_add_tail(&thread_pump->io_queue, &work_item->entry);
+    ReleaseSRWLockExclusive(&thread_pump->io_lock);
+    RtlWakeAddressAll((void *)&thread_pump->io_count);
+    return S_OK;
 }

static UINT WINAPI thread_pump_GetWorkItemCount(ID3DX10ThreadPump *iface)
@@ -281,20 +377,169 @@ static const ID3DX10ThreadPumpVtbl thread_pump_vtbl =
     thread_pump_GetQueueStatus
 };

+static DWORD WINAPI io_thread(void *arg)
+{
+    struct thread_pump *thread_pump = arg;
+    struct work_item *work_item;
+    UINT zero = 0;
+    HRESULT hr;
+
+    TRACE("%p thread started.\n", thread_pump);
+
+    while (1)
+    {
+        RtlWaitOnAddress((void *)&thread_pump->io_count, &zero, sizeof(zero), NULL);
+        AcquireSRWLockExclusive(&thread_pump->io_lock);
+        if (!thread_pump->io_count)
+        {
+            ReleaseSRWLockExclusive(&thread_pump->io_lock);
+            continue;
+        }
+        if (thread_pump->io_count == THREAD_PUMP_EXITING)
+        {
+            ReleaseSRWLockExclusive(&thread_pump->io_lock);
+            return 0;
+        }
+
+        thread_pump->io_count--;
+        work_item = LIST_ENTRY(list_head(&thread_pump->io_queue), struct work_item, entry);
+        list_remove(&work_item->entry);
+        ReleaseSRWLockExclusive(&thread_pump->io_lock);
+
+        if (FAILED(hr = ID3DX10DataLoader_Load(work_item->loader)))
+        {
+            if (work_item->result)
+                *work_item->result = hr;
+            work_item_free(work_item, FALSE);
+            continue;
+        }
+
+        AcquireSRWLockExclusive(&thread_pump->proc_lock);
+        if (thread_pump->proc_count == THREAD_PUMP_EXITING)
+        {
+            ReleaseSRWLockExclusive(&thread_pump->proc_lock);
+            work_item_free(work_item, TRUE);
+            return 0;
+        }
+
+        list_add_tail(&thread_pump->proc_queue, &work_item->entry);
+        thread_pump->proc_count++;
+        ReleaseSRWLockExclusive(&thread_pump->proc_lock);
+        RtlWakeAddressAll((void *)&thread_pump->proc_count);
+    }
+    return 0;
+}
+
+static DWORD WINAPI proc_thread(void *arg)
+{
+    struct thread_pump *thread_pump = arg;
+    struct work_item *work_item;
+    UINT zero = 0;
+    SIZE_T size;
+    void *data;
+    HRESULT hr;
+
+    TRACE("%p thread started.\n", thread_pump);
+
+    while (1)
+    {
+        RtlWaitOnAddress((void *)&thread_pump->proc_count, &zero, sizeof(zero), NULL);
+        AcquireSRWLockExclusive(&thread_pump->proc_lock);
+        if (!thread_pump->proc_count)
+        {
+            ReleaseSRWLockExclusive(&thread_pump->proc_lock);
+            continue;
+        }
+        if (thread_pump->proc_count == THREAD_PUMP_EXITING)
+        {
+            ReleaseSRWLockExclusive(&thread_pump->proc_lock);
+            return 0;
+        }
+
+        thread_pump->proc_count--;
+        work_item = LIST_ENTRY(list_head(&thread_pump->proc_queue), struct work_item, entry);
+        list_remove(&work_item->entry);
+        ReleaseSRWLockExclusive(&thread_pump->proc_lock);
+
+        if (FAILED(hr = ID3DX10DataLoader_Decompress(work_item->loader, &data, &size)))
+        {
+            if (work_item->result)
+                *work_item->result = hr;
+            work_item_free(work_item, FALSE);
+            continue;
+        }
+
+        if (thread_pump->device_count == THREAD_PUMP_EXITING)
+        {
+            work_item_free(work_item, TRUE);
+            return 0;
+        }
+
+        if (FAILED(hr = ID3DX10DataProcessor_Process(work_item->processor, data, size)))
+        {
+            if (work_item->result)
+                *work_item->result = hr;
+            work_item_free(work_item, FALSE);
+            continue;
+        }
+
+        AcquireSRWLockExclusive(&thread_pump->device_lock);
+        if (thread_pump->device_count == THREAD_PUMP_EXITING)
+        {
+            ReleaseSRWLockExclusive(&thread_pump->device_lock);
+            work_item_free(work_item, TRUE);
+            return 0;
+        }
+
+        list_add_tail(&thread_pump->device_queue, &work_item->entry);
+        thread_pump->device_count++;
+        ReleaseSRWLockExclusive(&thread_pump->device_lock);
+    }
+    return 0;
+}
+
 HRESULT WINAPI D3DX10CreateThreadPump(UINT io_threads, UINT proc_threads, ID3DX10ThreadPump **pump)
 {
     struct thread_pump *object;
+    int i;

TRACE("io_threads %u, proc_threads %u, pump %p.\n", io_threads, proc_threads, pump);

if (io_threads >= 1024 || proc_threads >= 1024)
         return E_FAIL;

-    if (!(object = calloc(1, sizeof(*object))))
+    if (!io_threads)
+        io_threads = 1;
+    if (!proc_threads)
+    {
+        SYSTEM_INFO info;
+
+        GetSystemInfo(&info);
+        proc_threads = info.dwNumberOfProcessors;
+    }
+
+    if (!(object = calloc(1, FIELD_OFFSET(struct thread_pump, threads[io_threads + proc_threads]))))
         return E_OUTOFMEMORY;

object->ID3DX10ThreadPump_iface.lpVtbl = &thread_pump_vtbl;
     object->refcount = 1;
+    InitializeSRWLock(&object->io_lock);
+    list_init(&object->io_queue);
+    InitializeSRWLock(&object->proc_lock);
+    list_init(&object->proc_queue);
+    InitializeSRWLock(&object->device_lock);
+    list_init(&object->device_queue);
+    object->threads_no = io_threads + proc_threads;
+
+    for (i = 0; i < object->threads_no; i++)
+    {
+        object->threads[i] = CreateThread(NULL, 0, i < io_threads ? io_thread : proc_thread, object, 0, NULL);
+        if (!object->threads[i])
+        {
+            ID3DX10ThreadPump_Release(&object->ID3DX10ThreadPump_iface);
+            return E_FAIL;
+        }
+    }

*pump = &object->ID3DX10ThreadPump_iface;
     return S_OK;