Signed-off-by: Nikolay Sivov nsivov@codeweavers.com --- dlls/mfplat/queue.c | 150 ++++++++++++++++++++++++++++++++++++- dlls/mfplat/tests/mfplat.c | 29 +++++++ 2 files changed, 176 insertions(+), 3 deletions(-)
diff --git a/dlls/mfplat/queue.c b/dlls/mfplat/queue.c index 3b599840cb..8c7cf195e5 100644 --- a/dlls/mfplat/queue.c +++ b/dlls/mfplat/queue.c @@ -33,9 +33,16 @@ WINE_DEFAULT_DEBUG_CHANNEL(mfplat); #define FIRST_USER_QUEUE_HANDLE 5 #define MAX_USER_QUEUE_HANDLES 124
+struct work_item +{ + LONG refcount; + IMFAsyncResult *result; +}; + struct queue { TP_POOL *pool; + TP_CALLBACK_ENVIRON env; };
struct queue_handle @@ -86,9 +93,104 @@ enum system_queue_index
static struct queue system_queues[SYS_QUEUE_COUNT];
+static struct queue *get_system_queue(DWORD queue_id) +{ + switch (queue_id) + { + case MFASYNC_CALLBACK_QUEUE_STANDARD: + case MFASYNC_CALLBACK_QUEUE_RT: + case MFASYNC_CALLBACK_QUEUE_IO: + case MFASYNC_CALLBACK_QUEUE_TIMER: + case MFASYNC_CALLBACK_QUEUE_MULTITHREADED: + case MFASYNC_CALLBACK_QUEUE_LONG_FUNCTION: + return &system_queues[queue_id - 1]; + default: + return NULL; + } +} + +static void CALLBACK standard_queue_cleanup_callback(void *object_data, void *group_data) +{ +} + +static struct work_item * alloc_work_item(struct queue *queue, IMFAsyncResult *result) +{ + struct work_item *item; + + item = heap_alloc_zero(sizeof(*item)); + item->result = result; + IMFAsyncResult_AddRef(item->result); + item->refcount = 1; + + return item; +} + +static void release_work_item(struct work_item *item) +{ + if (InterlockedDecrement(&item->refcount) == 0) + { + IMFAsyncResult_Release(item->result); + heap_free(item); + } +} + static void init_work_queue(MFASYNC_WORKQUEUE_TYPE queue_type, struct queue *queue) { + unsigned int max_thread; + queue->pool = CreateThreadpool(NULL); + queue->env.Version = 1; + queue->env.Pool = queue->pool; + queue->env.CleanupGroup = CreateThreadpoolCleanupGroup(); + queue->env.CleanupGroupCancelCallback = standard_queue_cleanup_callback; + + max_thread = (queue_type == MF_STANDARD_WORKQUEUE || queue_type == MF_WINDOW_WORKQUEUE) ? 1 : 4; + + SetThreadpoolThreadMinimum(queue->pool, 1); + SetThreadpoolThreadMaximum(queue->pool, max_thread); + + if (queue_type == MF_WINDOW_WORKQUEUE) + FIXME("MF_WINDOW_WORKQUEUE is not supported.\n"); +} + +static HRESULT grab_queue(DWORD queue_id, struct queue **ret) +{ + struct queue *queue = get_system_queue(queue_id); + MFASYNC_WORKQUEUE_TYPE queue_type; + struct queue_handle *entry; + + if (!system_queues[SYS_QUEUE_STANDARD].pool) + return MF_E_SHUTDOWN; + + if (queue && queue->pool) + { + *ret = queue; + return S_OK; + } + else if (queue) + { + EnterCriticalSection(&queues_section); + switch (queue_id) + { + case MFASYNC_CALLBACK_QUEUE_IO: + case MFASYNC_CALLBACK_QUEUE_MULTITHREADED: + case MFASYNC_CALLBACK_QUEUE_LONG_FUNCTION: + queue_type = MF_MULTITHREADED_WORKQUEUE; + break; + default: + queue_type = MF_STANDARD_WORKQUEUE; + } + init_work_queue(queue_type, queue); + LeaveCriticalSection(&queues_section); + *ret = queue; + return S_OK; + } + + /* Handles user queues. */ + if ((entry = get_queue_obj(queue_id))) + *ret = entry->obj; + + return *ret ? S_OK : MF_E_INVALID_WORKQUEUE; }
void init_system_queues(void) @@ -113,6 +215,7 @@ static void shutdown_queue(struct queue *queue) if (!queue->pool) return;
+ CloseThreadpoolCleanupGroupMembers(queue->env.CleanupGroup, TRUE, NULL); CloseThreadpool(queue->pool); queue->pool = NULL; } @@ -131,6 +234,47 @@ void shutdown_system_queues(void) LeaveCriticalSection(&queues_section); }
+static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void *context, TP_WORK *work) +{ + struct work_item *item = context; + MFASYNCRESULT *result = (MFASYNCRESULT *)item->result; + + TRACE("result object %p.\n", result); + + IMFAsyncCallback_Invoke(result->pCallback, item->result); + + release_work_item(item); +} + +static HRESULT queue_submit_item(struct queue *queue, IMFAsyncResult *result) +{ + struct work_item *item; + TP_WORK *work_object; + + if (!(item = alloc_work_item(queue, result))) + return E_OUTOFMEMORY; + + work_object = CreateThreadpoolWork(standard_queue_worker, item, &queue->env); + SubmitThreadpoolWork(work_object); + + TRACE("dispatched %p.\n", result); + + return S_OK; +} + +static HRESULT queue_put_work_item(DWORD queue_id, IMFAsyncResult *result) +{ + struct queue *queue; + HRESULT hr; + + if (FAILED(hr = grab_queue(queue_id, &queue))) + return hr; + + hr = queue_submit_item(queue, result); + + return hr; +} + static HRESULT alloc_user_queue(MFASYNC_WORKQUEUE_TYPE queue_type, DWORD *queue_id) { struct queue_handle *entry; @@ -440,7 +584,7 @@ HRESULT WINAPI MFPutWorkItem(DWORD queue, IMFAsyncCallback *callback, IUnknown * if (FAILED(hr = MFCreateAsyncResult(NULL, callback, state, &result))) return hr;
- hr = MFPutWorkItemEx(queue, result); + hr = queue_put_work_item(queue, result);
IMFAsyncResult_Release(result);
@@ -452,9 +596,9 @@ HRESULT WINAPI MFPutWorkItem(DWORD queue, IMFAsyncCallback *callback, IUnknown * */ HRESULT WINAPI MFPutWorkItemEx(DWORD queue, IMFAsyncResult *result) { - FIXME("%#x, %p\n", queue, result); + TRACE("%#x, %p\n", queue, result);
- return E_NOTIMPL; + return queue_put_work_item(queue, result); }
/*********************************************************************** diff --git a/dlls/mfplat/tests/mfplat.c b/dlls/mfplat/tests/mfplat.c index cc8a0f47b8..55142a79cf 100644 --- a/dlls/mfplat/tests/mfplat.c +++ b/dlls/mfplat/tests/mfplat.c @@ -1092,6 +1092,34 @@ todo_wine ok(hr == S_OK, "Failed to shutdown, hr %#x.\n", hr); }
+static void test_submit_work_item(void) +{ + static const DWORD queue_ids[] = + { + MFASYNC_CALLBACK_QUEUE_STANDARD, + MFASYNC_CALLBACK_QUEUE_RT, + MFASYNC_CALLBACK_QUEUE_IO, + MFASYNC_CALLBACK_QUEUE_TIMER, + MFASYNC_CALLBACK_QUEUE_MULTITHREADED, + MFASYNC_CALLBACK_QUEUE_LONG_FUNCTION, + }; + IMFAsyncCallback callback = { &testcallbackvtbl }; + unsigned int i; + HRESULT hr; + + hr = MFStartup(MF_VERSION, MFSTARTUP_FULL); + ok(hr == S_OK, "Failed to start up, hr %#x.\n", hr); + + for (i = 0; i < ARRAY_SIZE(queue_ids); ++i) + { + hr = MFPutWorkItem(queue_ids[i], &callback, NULL); + ok(hr == S_OK, "Failed to submit work item, hr %#x.\n", hr); + } + + hr = MFShutdown(); + ok(hr == S_OK, "Failed to shut down, hr %#x.\n", hr); +} + START_TEST(mfplat) { CoInitialize(NULL); @@ -1114,6 +1142,7 @@ START_TEST(mfplat) test_MFCreateCollection(); test_MFHeapAlloc(); test_scheduled_items(); + test_submit_work_item();
CoUninitialize(); }