[PATCH 1/9] include: Add the rest of async callback flags.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 include/mfobjects.idl | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/include/mfobjects.idl b/include/mfobjects.idl
index 203acc770f..758fd82cb1 100644
--- a/include/mfobjects.idl
+++ b/include/mfobjects.idl
@@ -425,8 +425,11 @@ interface IMFRemoteAsyncCallback : IUnknown
     HRESULT Invoke([in] HRESULT hr, [in] IUnknown *pRemoteResult);
 }

-cpp_quote("#define MFASYNC_FAST_IO_PROCESSING_CALLBACK  0x0001")
-cpp_quote("#define MFASYNC_SIGNAL_CALLBACK              0x0002" )
+cpp_quote("#define MFASYNC_FAST_IO_PROCESSING_CALLBACK   0x00000001")
+cpp_quote("#define MFASYNC_SIGNAL_CALLBACK               0x00000002")
+cpp_quote("#define MFASYNC_BLOCKING_CALLBACK             0x00000004")
+cpp_quote("#define MFASYNC_REPLY_CALLBACK                0x00000008")
+cpp_quote("#define MFASYNC_LOCALIZE_REMOTE_CALLBACK      0x00000010")

cpp_quote("#define MFASYNC_CALLBACK_QUEUE_UNDEFINED      0x00000000")
 cpp_quote("#define MFASYNC_CALLBACK_QUEUE_STANDARD       0x00000001")

[PATCH 2/9] rtworkq: Use IUnknown wrapper for work item management.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 dlls/rtworkq/queue.c | 83 +++++++++++++++++++++++++++++++-------------
 1 file changed, 59 insertions(+), 24 deletions(-)

diff --git a/dlls/rtworkq/queue.c b/dlls/rtworkq/queue.c
index a2fb3824c3..866c2b532f 100644
--- a/dlls/rtworkq/queue.c
+++ b/dlls/rtworkq/queue.c
@@ -109,8 +109,9 @@ enum system_queue_index

struct work_item
 {
-    struct list entry;
+    IUnknown IUnknown_iface;
     LONG refcount;
+    struct list entry;
     IRtwqAsyncResult *result;
     struct queue *queue;
     RTWQWORKITEM_KEY key;
@@ -121,6 +122,11 @@ struct work_item
     } u;
 };

+static struct work_item *work_item_impl_from_IUnknown(IUnknown *iface)
+{
+    return CONTAINING_RECORD(iface, struct work_item, IUnknown_iface);
+}
+
 static const TP_CALLBACK_PRIORITY priorities[] =
 {
     TP_CALLBACK_PRIORITY_HIGH,
@@ -158,32 +164,59 @@ static void CALLBACK standard_queue_cleanup_callback(void *object_data, void *gr
 {
 }

-static struct work_item * alloc_work_item(struct queue *queue, IRtwqAsyncResult *result)
+static HRESULT WINAPI work_item_QueryInterface(IUnknown *iface, REFIID riid, void **obj)
 {
-    struct work_item *item;
+    if (IsEqualIID(riid, &IID_IUnknown))
+    {
+        *obj = iface;
+        IUnknown_AddRef(iface);
+        return S_OK;
+    }

-    item = heap_alloc_zero(sizeof(*item));
-    item->result = result;
-    IRtwqAsyncResult_AddRef(item->result);
-    item->refcount = 1;
-    item->queue = queue;
-    list_init(&item->entry);
+    *obj = NULL;
+    return E_NOINTERFACE;
+}

-    return item;
+static ULONG WINAPI work_item_AddRef(IUnknown *iface)
+{
+    struct work_item *item = work_item_impl_from_IUnknown(iface);
+    return InterlockedIncrement(&item->refcount);
 }

-static void release_work_item(struct work_item *item)
+static ULONG WINAPI work_item_Release(IUnknown *iface)
 {
-    if (InterlockedDecrement(&item->refcount) == 0)
+    struct work_item *item = work_item_impl_from_IUnknown(iface);
+    ULONG refcount = InterlockedDecrement(&item->refcount);
+
+    if (!refcount)
     {
          IRtwqAsyncResult_Release(item->result);
          heap_free(item);
     }
+
+    return refcount;
 }

-static struct work_item *grab_work_item(struct work_item *item)
+static const IUnknownVtbl work_item_vtbl =
 {
-    InterlockedIncrement(&item->refcount);
+    work_item_QueryInterface,
+    work_item_AddRef,
+    work_item_Release,
+};
+
+static struct work_item * alloc_work_item(struct queue *queue, IRtwqAsyncResult *result)
+{
+    struct work_item *item;
+
+    item = heap_alloc_zero(sizeof(*item));
+
+    item->IUnknown_iface.lpVtbl = &work_item_vtbl;
+    item->result = result;
+    IRtwqAsyncResult_AddRef(item->result);
+    item->refcount = 1;
+    item->queue = queue;
+    list_init(&item->entry);
+
     return item;
 }

@@ -294,7 +327,7 @@ static void shutdown_queue(struct queue *queue)
     LIST_FOR_EACH_ENTRY_SAFE(item, item2, &queue->pending_items, struct work_item, entry)
     {
         list_remove(&item->entry);
-        release_work_item(item);
+        IUnknown_Release(&item->IUnknown_iface);
     }
     LeaveCriticalSection(&queue->cs);

@@ -335,7 +368,7 @@ static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void

IRtwqAsyncCallback_Invoke(result->pCallback, item->result);

-    release_work_item(item);
+    IUnknown_Release(&item->IUnknown_iface);
 }

static HRESULT queue_submit_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
@@ -399,7 +432,7 @@ static void queue_release_pending_item(struct work_item *item)
     {
         list_remove(&item->entry);
         item->key = 0;
-        release_work_item(item);
+        IUnknown_Release(&item->IUnknown_iface);
     }
     LeaveCriticalSection(&item->queue->cs);
 }
@@ -413,7 +446,7 @@ static void CALLBACK waiting_item_callback(TP_CALLBACK_INSTANCE *instance, void

invoke_async_callback(item->result);

-    release_work_item(item);
+    IUnknown_Release(&item->IUnknown_iface);
 }

static void CALLBACK waiting_item_cancelable_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_WAIT *wait,
@@ -427,7 +460,7 @@ static void CALLBACK waiting_item_cancelable_callback(TP_CALLBACK_INSTANCE *inst

invoke_async_callback(item->result);

-    release_work_item(item);
+    IUnknown_Release(&item->IUnknown_iface);
 }

static void CALLBACK scheduled_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
@@ -438,7 +471,7 @@ static void CALLBACK scheduled_item_callback(TP_CALLBACK_INSTANCE *instance, voi

invoke_async_callback(item->result);

-    release_work_item(item);
+    IUnknown_Release(&item->IUnknown_iface);
 }

static void CALLBACK scheduled_item_cancelable_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
@@ -451,16 +484,18 @@ static void CALLBACK scheduled_item_cancelable_callback(TP_CALLBACK_INSTANCE *in

invoke_async_callback(item->result);

-    release_work_item(item);
+    IUnknown_Release(&item->IUnknown_iface);
 }

static void CALLBACK periodic_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 {
-    struct work_item *item = grab_work_item(context);
+    struct work_item *item = context;
+
+    IUnknown_AddRef(&item->IUnknown_iface);

invoke_async_callback(item->result);

-    release_work_item(item);
+    IUnknown_Release(&item->IUnknown_iface);
 }

static void queue_mark_item_pending(DWORD mask, struct work_item *item, RTWQWORKITEM_KEY *key)
@@ -470,7 +505,7 @@ static void queue_mark_item_pending(DWORD mask, struct work_item *item, RTWQWORK

EnterCriticalSection(&item->queue->cs);
     list_add_tail(&item->queue->pending_items, &item->entry);
-    grab_work_item(item);
+    IUnknown_AddRef(&item->IUnknown_iface);
     LeaveCriticalSection(&item->queue->cs);
 }

[PATCH 3/9] rtworkq: Store item priority and callback flags.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 dlls/rtworkq/queue.c | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/dlls/rtworkq/queue.c b/dlls/rtworkq/queue.c
index 866c2b532f..c6a0c68d54 100644
--- a/dlls/rtworkq/queue.c
+++ b/dlls/rtworkq/queue.c
@@ -115,6 +115,8 @@ struct work_item
     IRtwqAsyncResult *result;
     struct queue *queue;
     RTWQWORKITEM_KEY key;
+    LONG priority;
+    DWORD flags;
     union
     {
         TP_WAIT *wait_object;
@@ -204,8 +206,10 @@ static const IUnknownVtbl work_item_vtbl =
     work_item_Release,
 };

-static struct work_item * alloc_work_item(struct queue *queue, IRtwqAsyncResult *result)
+static struct work_item * alloc_work_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 {
+    RTWQASYNCRESULT *async_result = (RTWQASYNCRESULT *)result;
+    DWORD flags = 0, queue_id = 0;
     struct work_item *item;

item = heap_alloc_zero(sizeof(*item));
@@ -216,6 +220,10 @@ static struct work_item * alloc_work_item(struct queue *queue, IRtwqAsyncResult
     item->refcount = 1;
     item->queue = queue;
     list_init(&item->entry);
+    item->priority = priority;
+
+    if (SUCCEEDED(IRtwqAsyncCallback_GetParameters(async_result->pCallback, &flags, &queue_id)))
+        item->flags = flags;

return item;
 }
@@ -377,7 +385,7 @@ static HRESULT queue_submit_item(struct queue *queue, LONG priority, IRtwqAsyncR
     struct work_item *item;
     TP_WORK *work_object;

-    if (!(item = alloc_work_item(queue, result)))
+    if (!(item = alloc_work_item(queue, priority, result)))
         return E_OUTOFMEMORY;

if (priority == 0)
@@ -515,7 +523,7 @@ static HRESULT queue_submit_wait(struct queue *queue, HANDLE event, LONG priorit
     PTP_WAIT_CALLBACK callback;
     struct work_item *item;

-    if (!(item = alloc_work_item(queue, result)))
+    if (!(item = alloc_work_item(queue, priority, result)))
         return E_OUTOFMEMORY;

if (key)
@@ -543,7 +551,7 @@ static HRESULT queue_submit_timer(struct queue *queue, IRtwqAsyncResult *result,
     FILETIME filetime;
     LARGE_INTEGER t;

-    if (!(item = alloc_work_item(queue, result)))
+    if (!(item = alloc_work_item(queue, 0, result)))
         return E_OUTOFMEMORY;

if (key)

[PATCH 4/9] rtworkq: Add a structure for queue initialization parameters.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 dlls/rtworkq/queue.c | 32 ++++++++++++++++++++++++--------
 1 file changed, 24 insertions(+), 8 deletions(-)

diff --git a/dlls/rtworkq/queue.c b/dlls/rtworkq/queue.c
index c6a0c68d54..38d376342d 100644
--- a/dlls/rtworkq/queue.c
+++ b/dlls/rtworkq/queue.c
@@ -136,6 +136,11 @@ static const TP_CALLBACK_PRIORITY priorities[] =
     TP_CALLBACK_PRIORITY_LOW,
 };

+struct queue_desc
+{
+    RTWQ_WORKQUEUE_TYPE queue_type;
+};
+
 struct queue
 {
     TP_POOL *pool;
@@ -228,7 +233,7 @@ static struct work_item * alloc_work_item(struct queue *queue, LONG priority, IR
     return item;
 }

-static void init_work_queue(RTWQ_WORKQUEUE_TYPE queue_type, struct queue *queue)
+static void init_work_queue(const struct queue_desc *desc, struct queue *queue)
 {
     TP_CALLBACK_ENVIRON_V3 env;
     unsigned int max_thread, i;
@@ -250,12 +255,12 @@ static void init_work_queue(RTWQ_WORKQUEUE_TYPE queue_type, struct queue *queue)
     list_init(&queue->pending_items);
     InitializeCriticalSection(&queue->cs);

-    max_thread = (queue_type == RTWQ_STANDARD_WORKQUEUE || queue_type == RTWQ_WINDOW_WORKQUEUE) ? 1 : 4;
+    max_thread = (desc->queue_type == RTWQ_STANDARD_WORKQUEUE || desc->queue_type == RTWQ_WINDOW_WORKQUEUE) ? 1 : 4;

SetThreadpoolThreadMinimum(queue->pool, 1);
     SetThreadpoolThreadMaximum(queue->pool, max_thread);

-    if (queue_type == RTWQ_WINDOW_WORKQUEUE)
+    if (desc->queue_type == RTWQ_WINDOW_WORKQUEUE)
         FIXME("RTWQ_WINDOW_WORKQUEUE is not supported.\n");
 }

@@ -277,6 +282,8 @@ static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
     }
     else if (queue)
     {
+        struct queue_desc desc;
+
         EnterCriticalSection(&queues_section);
         switch (queue_id)
         {
@@ -288,7 +295,9 @@ static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
             default:
                 queue_type = RTWQ_STANDARD_WORKQUEUE;
         }
-        init_work_queue(queue_type, queue);
+
+        desc.queue_type = queue_type;
+        init_work_queue(&desc, queue);
         LeaveCriticalSection(&queues_section);
         *ret = queue;
         return S_OK;
@@ -604,7 +613,7 @@ static HRESULT queue_cancel_item(struct queue *queue, RTWQWORKITEM_KEY key)
     return hr;
 }

-static HRESULT alloc_user_queue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queue_id)
+static HRESULT alloc_user_queue(const struct queue_desc *desc, DWORD *queue_id)
 {
     struct queue_handle *entry;
     struct queue *queue;
@@ -618,7 +627,8 @@ static HRESULT alloc_user_queue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queue_id)
     queue = heap_alloc_zero(sizeof(*queue));
     if (!queue)
         return E_OUTOFMEMORY;
-    init_work_queue(queue_type, queue);
+
+    init_work_queue(desc, queue);

EnterCriticalSection(&queues_section);

@@ -839,6 +849,8 @@ HRESULT WINAPI RtwqUnlockPlatform(void)

static void init_system_queues(void)
 {
+    struct queue_desc desc;
+
     /* Always initialize standard queue, keep the rest lazy. */

EnterCriticalSection(&queues_section);
@@ -849,7 +861,8 @@ static void init_system_queues(void)
         return;
     }

-    init_work_queue(RTWQ_STANDARD_WORKQUEUE, &system_queues[SYS_QUEUE_STANDARD]);
+    desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
+    init_work_queue(&desc, &system_queues[SYS_QUEUE_STANDARD]);

LeaveCriticalSection(&queues_section);
 }
@@ -1096,9 +1109,12 @@ HRESULT WINAPI RtwqPutWorkItem(DWORD queue, LONG priority, IRtwqAsyncResult *res

HRESULT WINAPI RtwqAllocateWorkQueue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queue)
 {
+    struct queue_desc desc;
+
     TRACE("%d, %p.\n", queue_type, queue);

-    return alloc_user_queue(queue_type, queue);
+    desc.queue_type = queue_type;
+    return alloc_user_queue(&desc, queue);
 }

HRESULT WINAPI RtwqLockWorkQueue(DWORD queue)

[PATCH 5/9] rtworkq: Initialize queue with specific set of functionality.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 dlls/rtworkq/queue.c | 162 ++++++++++++++++++++++++++++---------------
 1 file changed, 108 insertions(+), 54 deletions(-)

diff --git a/dlls/rtworkq/queue.c b/dlls/rtworkq/queue.c
index 38d376342d..e5026b6298 100644
--- a/dlls/rtworkq/queue.c
+++ b/dlls/rtworkq/queue.c
@@ -16,6 +16,8 @@
  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  */

+#include <assert.h>
+
 #define COBJMACROS
 #define NONAMELESSUNION

@@ -136,13 +138,25 @@ static const TP_CALLBACK_PRIORITY priorities[] =
     TP_CALLBACK_PRIORITY_LOW,
 };

+struct queue;
+struct queue_desc;
+
+struct queue_ops
+{
+    HRESULT (*init)(const struct queue_desc *desc, struct queue *queue);
+    BOOL (*shutdown)(struct queue *queue);
+    void (*submit)(struct queue *queue, struct work_item *item);
+};
+
 struct queue_desc
 {
     RTWQ_WORKQUEUE_TYPE queue_type;
+    const struct queue_ops *ops;
 };

struct queue
 {
+    const struct queue_ops *ops;
     TP_POOL *pool;
     TP_CALLBACK_ENVIRON_V3 envs[ARRAY_SIZE(priorities)];
     CRITICAL_SECTION cs;
@@ -171,6 +185,88 @@ static void CALLBACK standard_queue_cleanup_callback(void *object_data, void *gr
 {
 }

+static HRESULT pool_queue_init(const struct queue_desc *desc, struct queue *queue)
+{
+    TP_CALLBACK_ENVIRON_V3 env;
+    unsigned int max_thread, i;
+
+    queue->pool = CreateThreadpool(NULL);
+
+    memset(&env, 0, sizeof(env));
+    env.Version = 3;
+    env.Size = sizeof(env);
+    env.Pool = queue->pool;
+    env.CleanupGroup = CreateThreadpoolCleanupGroup();
+    env.CleanupGroupCancelCallback = standard_queue_cleanup_callback;
+    env.CallbackPriority = TP_CALLBACK_PRIORITY_NORMAL;
+    for (i = 0; i < ARRAY_SIZE(queue->envs); ++i)
+    {
+        queue->envs[i] = env;
+        queue->envs[i].CallbackPriority = priorities[i];
+    }
+    list_init(&queue->pending_items);
+    InitializeCriticalSection(&queue->cs);
+
+    max_thread = (desc->queue_type == RTWQ_STANDARD_WORKQUEUE || desc->queue_type == RTWQ_WINDOW_WORKQUEUE) ? 1 : 4;
+
+    SetThreadpoolThreadMinimum(queue->pool, 1);
+    SetThreadpoolThreadMaximum(queue->pool, max_thread);
+
+    if (desc->queue_type == RTWQ_WINDOW_WORKQUEUE)
+        FIXME("RTWQ_WINDOW_WORKQUEUE is not supported.\n");
+
+    return S_OK;
+}
+
+static BOOL pool_queue_shutdown(struct queue *queue)
+{
+    if (!queue->pool)
+        return FALSE;
+
+    CloseThreadpoolCleanupGroupMembers(queue->envs[0].CleanupGroup, TRUE, NULL);
+    CloseThreadpool(queue->pool);
+    queue->pool = NULL;
+
+    return TRUE;
+}
+
+static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void *context, TP_WORK *work)
+{
+    struct work_item *item = context;
+    RTWQASYNCRESULT *result = (RTWQASYNCRESULT *)item->result;
+
+    TRACE("result object %p.\n", result);
+
+    IRtwqAsyncCallback_Invoke(result->pCallback, item->result);
+
+    IUnknown_Release(&item->IUnknown_iface);
+}
+
+static void pool_queue_submit(struct queue *queue, struct work_item *item)
+{
+    TP_CALLBACK_PRIORITY callback_priority;
+    TP_WORK *work_object;
+
+    if (item->priority == 0)
+        callback_priority = TP_CALLBACK_PRIORITY_NORMAL;
+    else if (item->priority < 0)
+        callback_priority = TP_CALLBACK_PRIORITY_LOW;
+    else
+        callback_priority = TP_CALLBACK_PRIORITY_HIGH;
+    work_object = CreateThreadpoolWork(standard_queue_worker, item,
+            (TP_CALLBACK_ENVIRON *)&queue->envs[callback_priority]);
+    SubmitThreadpoolWork(work_object);
+
+    TRACE("dispatched %p.\n", item->result);
+}
+
+static const struct queue_ops pool_queue_ops =
+{
+    pool_queue_init,
+    pool_queue_shutdown,
+    pool_queue_submit,
+};
+
 static HRESULT WINAPI work_item_QueryInterface(IUnknown *iface, REFIID riid, void **obj)
 {
     if (IsEqualIID(riid, &IID_IUnknown))
@@ -235,33 +331,14 @@ static struct work_item * alloc_work_item(struct queue *queue, LONG priority, IR

static void init_work_queue(const struct queue_desc *desc, struct queue *queue)
 {
-    TP_CALLBACK_ENVIRON_V3 env;
-    unsigned int max_thread, i;
-
-    queue->pool = CreateThreadpool(NULL);
+    assert(desc->ops != NULL);

-    memset(&env, 0, sizeof(env));
-    env.Version = 3;
-    env.Size = sizeof(env);
-    env.Pool = queue->pool;
-    env.CleanupGroup = CreateThreadpoolCleanupGroup();
-    env.CleanupGroupCancelCallback = standard_queue_cleanup_callback;
-    env.CallbackPriority = TP_CALLBACK_PRIORITY_NORMAL;
-    for (i = 0; i < ARRAY_SIZE(queue->envs); ++i)
+    queue->ops = desc->ops;
+    if (SUCCEEDED(queue->ops->init(desc, queue)))
     {
-        queue->envs[i] = env;
-        queue->envs[i].CallbackPriority = priorities[i];
+        list_init(&queue->pending_items);
+        InitializeCriticalSection(&queue->cs);
     }
-    list_init(&queue->pending_items);
-    InitializeCriticalSection(&queue->cs);
-
-    max_thread = (desc->queue_type == RTWQ_STANDARD_WORKQUEUE || desc->queue_type == RTWQ_WINDOW_WORKQUEUE) ? 1 : 4;
-
-    SetThreadpoolThreadMinimum(queue->pool, 1);
-    SetThreadpoolThreadMaximum(queue->pool, max_thread);
-
-    if (desc->queue_type == RTWQ_WINDOW_WORKQUEUE)
-        FIXME("RTWQ_WINDOW_WORKQUEUE is not supported.\n");
 }

static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
@@ -297,6 +374,7 @@ static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
         }

desc.queue_type = queue_type;
+        desc.ops = &pool_queue_ops;
         init_work_queue(&desc, queue);
         LeaveCriticalSection(&queues_section);
         *ret = queue;
@@ -333,13 +411,9 @@ static void shutdown_queue(struct queue *queue)
 {
     struct work_item *item, *item2;

-    if (!queue->pool)
+    if (!queue->ops || !queue->ops->shutdown(queue))
         return;

-    CloseThreadpoolCleanupGroupMembers(queue->envs[0].CleanupGroup, TRUE, NULL);
-    CloseThreadpool(queue->pool);
-    queue->pool = NULL;
-
     EnterCriticalSection(&queue->cs);
     LIST_FOR_EACH_ENTRY_SAFE(item, item2, &queue->pending_items, struct work_item, entry)
     {
@@ -349,6 +423,8 @@ static void shutdown_queue(struct queue *queue)
     LeaveCriticalSection(&queue->cs);

DeleteCriticalSection(&queue->cs);
+
+    memset(queue, 0, sizeof(*queue));
 }

static HRESULT unlock_user_queue(DWORD queue)
@@ -376,38 +452,14 @@ static HRESULT unlock_user_queue(DWORD queue)
     return hr;
 }

-static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void *context, TP_WORK *work)
-{
-    struct work_item *item = context;
-    RTWQASYNCRESULT *result = (RTWQASYNCRESULT *)item->result;
-
-    TRACE("result object %p.\n", result);
-
-    IRtwqAsyncCallback_Invoke(result->pCallback, item->result);
-
-    IUnknown_Release(&item->IUnknown_iface);
-}
-
 static HRESULT queue_submit_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 {
-    TP_CALLBACK_PRIORITY callback_priority;
     struct work_item *item;
-    TP_WORK *work_object;

if (!(item = alloc_work_item(queue, priority, result)))
         return E_OUTOFMEMORY;

-    if (priority == 0)
-        callback_priority = TP_CALLBACK_PRIORITY_NORMAL;
-    else if (priority < 0)
-        callback_priority = TP_CALLBACK_PRIORITY_LOW;
-    else
-        callback_priority = TP_CALLBACK_PRIORITY_HIGH;
-    work_object = CreateThreadpoolWork(standard_queue_worker, item,
-            (TP_CALLBACK_ENVIRON *)&queue->envs[callback_priority]);
-    SubmitThreadpoolWork(work_object);
-
-    TRACE("dispatched %p.\n", result);
+    queue->ops->submit(queue, item);

return S_OK;
 }
@@ -862,6 +914,7 @@ static void init_system_queues(void)
     }

desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
+    desc.ops = &pool_queue_ops;
     init_work_queue(&desc, &system_queues[SYS_QUEUE_STANDARD]);

LeaveCriticalSection(&queues_section);
@@ -1114,6 +1167,7 @@ HRESULT WINAPI RtwqAllocateWorkQueue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queu
     TRACE("%d, %p.\n", queue_type, queue);

desc.queue_type = queue_type;
+    desc.ops = &pool_queue_ops;
     return alloc_user_queue(&desc, queue);
 }

[PATCH 6/9] rtworkq: Add support for serial work queues.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 dlls/rtworkq/queue.c      | 327 ++++++++++++++++++++++++++++++++------
 dlls/rtworkq/rtworkq.spec |   2 +-
 include/rtworkq.idl       |   1 +
 3 files changed, 280 insertions(+), 50 deletions(-)

diff --git a/dlls/rtworkq/queue.c b/dlls/rtworkq/queue.c
index e5026b6298..65007602c7 100644
--- a/dlls/rtworkq/queue.c
+++ b/dlls/rtworkq/queue.c
@@ -97,6 +97,16 @@ enum rtwq_callback_queue_id
     RTWQ_CALLBACK_QUEUE_ALL           = 0xffffffff,
 };

+/* Should be kept in sync with corresponding MFASYNC_ constants. */
+enum rtwq_callback_flags
+{
+    RTWQ_FAST_IO_PROCESSING_CALLBACK = 0x00000001,
+    RTWQ_SIGNAL_CALLBACK             = 0x00000002,
+    RTWQ_BLOCKING_CALLBACK           = 0x00000004,
+    RTWQ_REPLY_CALLBACK              = 0x00000008,
+    RTWQ_LOCALIZE_REMOTE_CALLBACK    = 0x00000010,
+};
+
 enum system_queue_index
 {
     SYS_QUEUE_STANDARD = 0,
@@ -115,10 +125,12 @@ struct work_item
     LONG refcount;
     struct list entry;
     IRtwqAsyncResult *result;
+    IRtwqAsyncResult *reply_result;
     struct queue *queue;
     RTWQWORKITEM_KEY key;
     LONG priority;
     DWORD flags;
+    PTP_SIMPLE_CALLBACK finalization_callback;
     union
     {
         TP_WAIT *wait_object;
@@ -152,15 +164,123 @@ struct queue_desc
 {
     RTWQ_WORKQUEUE_TYPE queue_type;
     const struct queue_ops *ops;
+    DWORD target_queue;
 };

struct queue
 {
+    IRtwqAsyncCallback IRtwqAsyncCallback_iface;
     const struct queue_ops *ops;
     TP_POOL *pool;
     TP_CALLBACK_ENVIRON_V3 envs[ARRAY_SIZE(priorities)];
     CRITICAL_SECTION cs;
     struct list pending_items;
+    DWORD id;
+    /* Data used for serial queues only. */
+    PTP_SIMPLE_CALLBACK finalization_callback;
+    DWORD target_queue;
+};
+
+static void shutdown_queue(struct queue *queue);
+
+static HRESULT lock_user_queue(DWORD queue)
+{
+    HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
+    struct queue_handle *entry;
+
+    if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
+        return S_OK;
+
+    EnterCriticalSection(&queues_section);
+    entry = get_queue_obj(queue);
+    if (entry && entry->refcount)
+    {
+        entry->refcount++;
+        hr = S_OK;
+    }
+    LeaveCriticalSection(&queues_section);
+    return hr;
+}
+
+static HRESULT unlock_user_queue(DWORD queue)
+{
+    HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
+    struct queue_handle *entry;
+
+    if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
+        return S_OK;
+
+    EnterCriticalSection(&queues_section);
+    entry = get_queue_obj(queue);
+    if (entry && entry->refcount)
+    {
+        if (--entry->refcount == 0)
+        {
+            shutdown_queue((struct queue *)entry->obj);
+            heap_free(entry->obj);
+            entry->obj = next_free_user_queue;
+            next_free_user_queue = entry;
+        }
+        hr = S_OK;
+    }
+    LeaveCriticalSection(&queues_section);
+    return hr;
+}
+
+static struct queue *queue_impl_from_IRtwqAsyncCallback(IRtwqAsyncCallback *iface)
+{
+    return CONTAINING_RECORD(iface, struct queue, IRtwqAsyncCallback_iface);
+}
+
+static HRESULT WINAPI queue_serial_callback_QueryInterface(IRtwqAsyncCallback *iface, REFIID riid, void **obj)
+{
+    if (IsEqualIID(riid, &IID_IRtwqAsyncCallback) ||
+            IsEqualIID(riid, &IID_IUnknown))
+    {
+        *obj = iface;
+        IRtwqAsyncCallback_AddRef(iface);
+        return S_OK;
+    }
+
+    *obj = NULL;
+    return E_NOINTERFACE;
+}
+
+static ULONG WINAPI queue_serial_callback_AddRef(IRtwqAsyncCallback *iface)
+{
+    return 2;
+}
+
+static ULONG WINAPI queue_serial_callback_Release(IRtwqAsyncCallback *iface)
+{
+    return 1;
+}
+
+static HRESULT WINAPI queue_serial_callback_GetParameters(IRtwqAsyncCallback *iface, DWORD *flags, DWORD *queue_id)
+{
+    struct queue *queue = queue_impl_from_IRtwqAsyncCallback(iface);
+
+    *flags = 0;
+    *queue_id = queue->id;
+
+    return S_OK;
+}
+
+static HRESULT WINAPI queue_serial_callback_Invoke(IRtwqAsyncCallback *iface, IRtwqAsyncResult *result)
+{
+    /* Reply callback won't be called in a regular way, pending items and chained queues will make it
+       unnecessary complicated to reach actual work queue that's able to execute this item. Instead
+       serial queues are cleaned up right away on submit(). */
+    return S_OK;
+}
+
+static const IRtwqAsyncCallbackVtbl queue_serial_callback_vtbl =
+{
+    queue_serial_callback_QueryInterface,
+    queue_serial_callback_AddRef,
+    queue_serial_callback_Release,
+    queue_serial_callback_GetParameters,
+    queue_serial_callback_Invoke,
 };

static struct queue system_queues[SYS_QUEUE_COUNT];
@@ -181,6 +301,8 @@ static struct queue *get_system_queue(DWORD queue_id)
     }
 }

+static HRESULT grab_queue(DWORD queue_id, struct queue **ret);
+
 static void CALLBACK standard_queue_cleanup_callback(void *object_data, void *group_data)
 {
 }
@@ -237,7 +359,11 @@ static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void

TRACE("result object %p.\n", result);

-    IRtwqAsyncCallback_Invoke(result->pCallback, item->result);
+    /* Submitting from serial queue in reply mode, use different result object acting as receipt token.
+       It's submitted to user callback still, but when invoked, special serial queue callback will be used
+       to ensure correct destination queue. */
+
+    IRtwqAsyncCallback_Invoke(result->pCallback, item->reply_result ? item->reply_result : item->result);

IUnknown_Release(&item->IUnknown_iface);
 }
@@ -245,6 +371,7 @@ static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void
 static void pool_queue_submit(struct queue *queue, struct work_item *item)
 {
     TP_CALLBACK_PRIORITY callback_priority;
+    TP_CALLBACK_ENVIRON_V3 env;
     TP_WORK *work_object;

if (item->priority == 0)
@@ -253,8 +380,14 @@ static void pool_queue_submit(struct queue *queue, struct work_item *item)
         callback_priority = TP_CALLBACK_PRIORITY_LOW;
     else
         callback_priority = TP_CALLBACK_PRIORITY_HIGH;
-    work_object = CreateThreadpoolWork(standard_queue_worker, item,
-            (TP_CALLBACK_ENVIRON *)&queue->envs[callback_priority]);
+
+    env = queue->envs[callback_priority];
+    env.FinalizationCallback = item->finalization_callback;
+    /* Worker pool callback will release one reference. Grab one more to keep object alive when
+       we need finalization callback. */
+    if (item->finalization_callback)
+        IUnknown_AddRef(&item->IUnknown_iface);
+    work_object = CreateThreadpoolWork(standard_queue_worker, item, (TP_CALLBACK_ENVIRON *)&env);
     SubmitThreadpoolWork(work_object);

TRACE("dispatched %p.\n", item->result);
@@ -267,6 +400,129 @@ static const struct queue_ops pool_queue_ops =
     pool_queue_submit,
 };

+static struct work_item * serial_queue_get_next(struct queue *queue, struct work_item *item)
+{
+    struct work_item *next_item = NULL;
+
+    list_remove(&item->entry);
+    if (!list_empty(&item->queue->pending_items))
+        next_item = LIST_ENTRY(list_head(&item->queue->pending_items), struct work_item, entry);
+
+    return next_item;
+}
+
+static void CALLBACK serial_queue_finalization_callback(PTP_CALLBACK_INSTANCE instance, void *user_data)
+{
+    struct work_item *item = (struct work_item *)user_data, *next_item;
+    struct queue *target_queue, *queue = item->queue;
+    HRESULT hr;
+
+    EnterCriticalSection(&queue->cs);
+
+    if ((next_item = serial_queue_get_next(queue, item)))
+    {
+        if (SUCCEEDED(hr = grab_queue(queue->target_queue, &target_queue)))
+            target_queue->ops->submit(target_queue, next_item);
+        else
+            WARN("Failed to grab queue for id %#x, hr %#x.\n", queue->target_queue, hr);
+    }
+
+    LeaveCriticalSection(&queue->cs);
+
+    IUnknown_Release(&item->IUnknown_iface);
+}
+
+static HRESULT serial_queue_init(const struct queue_desc *desc, struct queue *queue)
+{
+    queue->IRtwqAsyncCallback_iface.lpVtbl = &queue_serial_callback_vtbl;
+    queue->target_queue = desc->target_queue;
+    lock_user_queue(queue->target_queue);
+    queue->finalization_callback = serial_queue_finalization_callback;
+
+    return S_OK;
+}
+
+static BOOL serial_queue_shutdown(struct queue *queue)
+{
+    unlock_user_queue(queue->target_queue);
+
+    return TRUE;
+}
+
+static struct work_item * serial_queue_is_ack_token(struct queue *queue, struct work_item *item)
+{
+    RTWQASYNCRESULT *async_result = (RTWQASYNCRESULT *)item->result;
+    struct work_item *head;
+
+    if (list_empty(&queue->pending_items))
+        return NULL;
+
+    head = LIST_ENTRY(list_head(&queue->pending_items), struct work_item, entry);
+    if (head->reply_result == item->result && async_result->pCallback == &queue->IRtwqAsyncCallback_iface)
+        return head;
+
+    return NULL;
+}
+
+static void serial_queue_submit(struct queue *queue, struct work_item *item)
+{
+    struct work_item *head, *next_item = NULL;
+    struct queue *target_queue;
+    HRESULT hr;
+
+    /* In reply mode queue will advance when 'reply_result' is invoked, in regular mode it will advance automatically,
+       via finalization callback. */
+
+    if (item->flags & RTWQ_REPLY_CALLBACK)
+    {
+        if (FAILED(hr = RtwqCreateAsyncResult(NULL, &queue->IRtwqAsyncCallback_iface, NULL, &item->reply_result)))
+            WARN("Failed to create reply object, hr %#x.\n", hr);
+    }
+    else
+        item->finalization_callback = queue->finalization_callback;
+
+    /* Serial queues could be chained together, detach from current queue before transitioning item to this one.
+       Items are not detached when submitted to pool queues, because pool queues won't forward them further. */
+    EnterCriticalSection(&item->queue->cs);
+    list_remove(&item->entry);
+    LeaveCriticalSection(&item->queue->cs);
+
+    EnterCriticalSection(&queue->cs);
+
+    item->queue = queue;
+
+    if ((head = serial_queue_is_ack_token(queue, item)))
+    {
+        /* Ack receipt token - pop waiting item, advance. */
+        next_item = serial_queue_get_next(queue, head);
+        IUnknown_Release(&head->IUnknown_iface);
+    }
+    else
+    {
+        if (list_empty(&queue->pending_items))
+            next_item = item;
+        list_add_tail(&queue->pending_items, &item->entry);
+        IUnknown_AddRef(&item->IUnknown_iface);
+    }
+
+    if (next_item)
+    {
+        if (SUCCEEDED(hr = grab_queue(queue->target_queue, &target_queue)))
+            target_queue->ops->submit(target_queue, next_item);
+        else
+            WARN("Failed to grab queue for id %#x, hr %#x.\n", queue->target_queue, hr);
+    }
+
+    LeaveCriticalSection(&queue->cs);
+}
+
+static const struct queue_ops serial_queue_ops =
+{
+    serial_queue_init,
+    serial_queue_shutdown,
+    serial_queue_submit,
+};
+
 static HRESULT WINAPI work_item_QueryInterface(IUnknown *iface, REFIID riid, void **obj)
 {
     if (IsEqualIID(riid, &IID_IUnknown))
@@ -293,8 +549,10 @@ static ULONG WINAPI work_item_Release(IUnknown *iface)

if (!refcount)
     {
-         IRtwqAsyncResult_Release(item->result);
-         heap_free(item);
+        if (item->reply_result)
+            IRtwqAsyncResult_Release(item->reply_result);
+        IRtwqAsyncResult_Release(item->result);
+        heap_free(item);
     }

return refcount;
@@ -375,6 +633,7 @@ static HRESULT grab_queue(DWORD queue_id, struct queue **ret)

desc.queue_type = queue_type;
         desc.ops = &pool_queue_ops;
+        desc.target_queue = 0;
         init_work_queue(&desc, queue);
         LeaveCriticalSection(&queues_section);
         *ret = queue;
@@ -388,25 +647,6 @@ static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
     return *ret ? S_OK : RTWQ_E_INVALID_WORKQUEUE;
 }

-static HRESULT lock_user_queue(DWORD queue)
-{
-    HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
-    struct queue_handle *entry;
-
-    if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
-        return S_OK;
-
-    EnterCriticalSection(&queues_section);
-    entry = get_queue_obj(queue);
-    if (entry && entry->refcount)
-    {
-        entry->refcount++;
-        hr = S_OK;
-    }
-    LeaveCriticalSection(&queues_section);
-    return hr;
-}
-
 static void shutdown_queue(struct queue *queue)
 {
     struct work_item *item, *item2;
@@ -427,31 +667,6 @@ static void shutdown_queue(struct queue *queue)
     memset(queue, 0, sizeof(*queue));
 }

-static HRESULT unlock_user_queue(DWORD queue)
-{
-    HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
-    struct queue_handle *entry;
-
-    if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
-        return S_OK;
-
-    EnterCriticalSection(&queues_section);
-    entry = get_queue_obj(queue);
-    if (entry && entry->refcount)
-    {
-        if (--entry->refcount == 0)
-        {
-            shutdown_queue((struct queue *)entry->obj);
-            heap_free(entry->obj);
-            entry->obj = next_free_user_queue;
-            next_free_user_queue = entry;
-        }
-        hr = S_OK;
-    }
-    LeaveCriticalSection(&queues_section);
-    return hr;
-}
-
 static HRESULT queue_submit_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 {
     struct work_item *item;
@@ -915,6 +1130,7 @@ static void init_system_queues(void)

desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
     desc.ops = &pool_queue_ops;
+    desc.target_queue = 0;
     init_work_queue(&desc, &system_queues[SYS_QUEUE_STANDARD]);

LeaveCriticalSection(&queues_section);
@@ -1168,6 +1384,7 @@ HRESULT WINAPI RtwqAllocateWorkQueue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queu

desc.queue_type = queue_type;
     desc.ops = &pool_queue_ops;
+    desc.target_queue = 0;
     return alloc_user_queue(&desc, queue);
 }

@@ -1233,3 +1450,15 @@ HRESULT WINAPI RtwqCancelDeadline(HANDLE request)

return E_NOTIMPL;
 }
+
+HRESULT WINAPI RtwqAllocateSerialWorkQueue(DWORD target_queue, DWORD *queue)
+{
+    struct queue_desc desc;
+
+    TRACE("%#x, %p.\n", target_queue, queue);
+
+    desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
+    desc.ops = &serial_queue_ops;
+    desc.target_queue = target_queue;
+    return alloc_user_queue(&desc, queue);
+}
diff --git a/dlls/rtworkq/rtworkq.spec b/dlls/rtworkq/rtworkq.spec
index 1a8909e098..adc05d679d 100644
--- a/dlls/rtworkq/rtworkq.spec
+++ b/dlls/rtworkq/rtworkq.spec
@@ -1,5 +1,5 @@
 @ stdcall RtwqAddPeriodicCallback(ptr ptr ptr)
-@ stub RtwqAllocateSerialWorkQueue
+@ stdcall RtwqAllocateSerialWorkQueue(long ptr)
 @ stdcall RtwqAllocateWorkQueue(long ptr)
 @ stub RtwqBeginRegisterWorkQueueWithMMCSS
 @ stub RtwqBeginUnregisterWorkQueueWithMMCSS
diff --git a/include/rtworkq.idl b/include/rtworkq.idl
index 325bebfde5..1a452c3edc 100644
--- a/include/rtworkq.idl
+++ b/include/rtworkq.idl
@@ -78,6 +78,7 @@ cpp_quote("} RTWQASYNCRESULT;")
 cpp_quote("typedef void (WINAPI *RTWQPERIODICCALLBACK)(IUnknown *context);")

cpp_quote("HRESULT WINAPI RtwqAddPeriodicCallback(RTWQPERIODICCALLBACK callback, IUnknown *context, DWORD *key);")
+cpp_quote("HRESULT WINAPI RtwqAllocateSerialWorkQueue(DWORD target_queue, DWORD *queue);")
 cpp_quote("HRESULT WINAPI RtwqAllocateWorkQueue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queue);")
 cpp_quote("HRESULT WINAPI RtwqCancelDeadline(HANDLE request);")
 cpp_quote("HRESULT WINAPI RtwqCancelWorkItem(RTWQWORKITEM_KEY key);")

[PATCH 7/9] mfplat: Add MFAllocateSerialWorkQueue() export.

[Nikolay Sivov]

Signed-off-by: Nikolay Sivov 
---
 dlls/mfplat/mfplat.spec    | 1 +
 dlls/mfplat/tests/mfplat.c | 2 +-
 include/mfapi.h            | 1 +
 3 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/dlls/mfplat/mfplat.spec b/dlls/mfplat/mfplat.spec
index ba835b694a..85e28792c6 100644
--- a/dlls/mfplat/mfplat.spec
+++ b/dlls/mfplat/mfplat.spec
@@ -16,6 +16,7 @@
 @ stub GetD3DFormatFromMFSubtype
 @ stub LFGetGlobalPool
 @ stdcall MFAddPeriodicCallback(ptr ptr ptr) rtworkq.RtwqAddPeriodicCallback
+@ stdcall MFAllocateSerialWorkQueue(long ptr) rtworkq.RtwqAllocateSerialWorkQueue
 @ stdcall MFAllocateWorkQueue(ptr)
 @ stdcall MFAllocateWorkQueueEx(long ptr) rtworkq.RtwqAllocateWorkQueue
 @ stub MFAppendCollection
diff --git a/dlls/mfplat/tests/mfplat.c b/dlls/mfplat/tests/mfplat.c
index d0e05650c2..56aa23c68c 100644
--- a/dlls/mfplat/tests/mfplat.c
+++ b/dlls/mfplat/tests/mfplat.c
@@ -2251,7 +2251,7 @@ static void test_serial_queue(void)

if (!pMFAllocateSerialWorkQueue)
     {
-        skip("Serial queues are not supported.\n");
+        win_skip("Serial queues are not supported.\n");
         return;
     }

diff --git a/include/mfapi.h b/include/mfapi.h
index 7c9e713ac2..910b45e68f 100644
--- a/include/mfapi.h
+++ b/include/mfapi.h
@@ -389,6 +389,7 @@ typedef enum
 #define MFSESSIONCAP_DOES_NOT_USE_NETWORK  0x00000040

HRESULT WINAPI MFAddPeriodicCallback(MFPERIODICCALLBACK callback, IUnknown *context, DWORD *key);
+HRESULT WINAPI MFAllocateSerialWorkQueue(DWORD target_queue, DWORD *queue);
 HRESULT WINAPI MFAllocateWorkQueue(DWORD *queue);
 HRESULT WINAPI MFAllocateWorkQueueEx(MFASYNC_WORKQUEUE_TYPE queue_type, DWORD *queue);
 HRESULT WINAPI MFBeginCreateFile(MF_FILE_ACCESSMODE access_mode, MF_FILE_OPENMODE open_mode, MF_FILE_FLAGS flags,

[PATCH 7/9] mfplat: Add MFAllocateSerialWorkQueue() export.

[Marvin]

Hi,

While running your changed tests, I think I found new failures.
Being a bot and all I'm not very good at pattern recognition, so I might be
wrong, but could you please double-check?

Full results can be found at:
https://testbot.winehq.org/JobDetails.pl?Key=64907

Your paranoid android.

=== wxppro (32 bit report) ===

=== w8 (32 bit report) ===

=== w8adm (32 bit report) ===

=== w864 (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1507 (32 bit report) ===

=== w1064v1809 (32 bit report) ===

mfplat:
mfplat.c:2409: Test failed: Unexpected return value 0x102.
mfplat.c:1760: Test failed: Failed to get event, hr 0xc00d3e85.
mfplat.c:1763: Test failed: Failed to get event, hr 0xc00d3e85.
mfplat.c:1766: Test failed: Failed to finalize GetEvent, hr 0xc00d3e85.
mfplat.c:1769: Test failed: Unexpected result, hr 0xc00d3e85.
18c4:mfplat: unhandled exception c0000005 at 00404EFB

=== w1064v1809_2scr (32 bit report) ===

=== w1064v1809_ar (32 bit report) ===

=== w1064v1809_he (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1809_ja (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1809_zh_CN (32 bit report) ===

mfplat:
mfplat.c:2409: Test failed: Unexpected return value 0x102.

Report validation errors:
mfplat:mfplat has unaccounted for failure messages

=== w864 (64 bit report) ===

=== w1064v1507 (64 bit report) ===

=== w1064v1809 (64 bit report) ===

=== debian10 (32 bit report) ===

=== debian10 (32 bit French report) ===

=== debian10 (32 bit Japanese:Japan report) ===

=== debian10 (32 bit Chinese:China report) ===

=== debian10 (32 bit WoW report) ===

=== debian10 (64 bit WoW report) ===

[PATCH 8/9] mfplat: Implement IMFByteStream::GetLength() for file streams.

[Nikolay Sivov]

From: Derek Lesho

Signed-off-by: Derek Lesho 
Signed-off-by: Nikolay Sivov 
---
 dlls/mfplat/main.c         | 27 +++++++++++++++++++--------
 dlls/mfplat/tests/mfplat.c | 10 ++++++++++
 2 files changed, 29 insertions(+), 8 deletions(-)

diff --git a/dlls/mfplat/main.c b/dlls/mfplat/main.c
index e96a972909..1ebd443131 100644
--- a/dlls/mfplat/main.c
+++ b/dlls/mfplat/main.c
@@ -3278,13 +3278,6 @@ static HRESULT WINAPI bytestream_GetCapabilities(IMFByteStream *iface, DWORD *ca
     return S_OK;
 }

-static HRESULT WINAPI mfbytestream_GetLength(IMFByteStream *iface, QWORD *length)
-{
-    FIXME("%p, %p.\n", iface, length);
-
-    return E_NOTIMPL;
-}
-
 static HRESULT WINAPI mfbytestream_SetLength(IMFByteStream *iface, QWORD length)
 {
     mfbytestream *This = impl_from_IMFByteStream(iface);
@@ -3312,6 +3305,24 @@ static HRESULT WINAPI mfbytestream_SetCurrentPosition(IMFByteStream *iface, QWOR
     return E_NOTIMPL;
 }

+static HRESULT WINAPI bytestream_file_GetLength(IMFByteStream *iface, QWORD *length)
+{
+    struct bytestream *stream = impl_from_IMFByteStream(iface);
+    LARGE_INTEGER li;
+
+    TRACE("%p, %p.\n", iface, length);
+
+    if (!length)
+        return E_INVALIDARG;
+
+    if (GetFileSizeEx(stream->hfile, &li))
+        *length = li.QuadPart;
+    else
+        return HRESULT_FROM_WIN32(GetLastError());
+
+    return S_OK;
+}
+
 static HRESULT WINAPI bytestream_file_IsEndOfStream(IMFByteStream *iface, BOOL *ret)
 {
     struct bytestream *stream = impl_from_IMFByteStream(iface);
@@ -3440,7 +3451,7 @@ static const IMFByteStreamVtbl bytestream_file_vtbl =
     bytestream_AddRef,
     bytestream_Release,
     bytestream_GetCapabilities,
-    mfbytestream_GetLength,
+    bytestream_file_GetLength,
     mfbytestream_SetLength,
     mfbytestream_GetCurrentPosition,
     mfbytestream_SetCurrentPosition,
diff --git a/dlls/mfplat/tests/mfplat.c b/dlls/mfplat/tests/mfplat.c
index 56aa23c68c..7ec0b01982 100644
--- a/dlls/mfplat/tests/mfplat.c
+++ b/dlls/mfplat/tests/mfplat.c
@@ -1342,6 +1342,7 @@ static void test_file_stream(void)
     IMFByteStream *bytestream, *bytestream2;
     IMFAttributes *attributes = NULL;
     MF_ATTRIBUTE_TYPE item_type;
+    QWORD bytestream_length;
     WCHAR pathW[MAX_PATH];
     DWORD caps, count;
     WCHAR *filename;
@@ -1405,6 +1406,15 @@ static void test_file_stream(void)

IMFAttributes_Release(attributes);

+    /* Length. */
+    hr = IMFByteStream_GetLength(bytestream, NULL);
+    ok(hr == E_INVALIDARG, "Unexpected hr %#x.\n", hr);
+
+    bytestream_length = 0;
+    hr = IMFByteStream_GetLength(bytestream, &bytestream_length);
+    ok(hr == S_OK, "Failed to get bytestream length, hr %#x.\n", hr);
+    ok(bytestream_length > 0, "Unexpected bytestream length %s.\n", wine_dbgstr_longlong(bytestream_length));
+
     hr = MFCreateFile(MF_ACCESSMODE_READ, MF_OPENMODE_FAIL_IF_NOT_EXIST,
                       MF_FILEFLAGS_NONE, filename, &bytestream2);
     ok(hr == S_OK, "got 0x%08x\n", hr);

[PATCH 8/9] mfplat: Implement IMFByteStream::GetLength() for file streams.

[Marvin]

Hi,

While running your changed tests, I think I found new failures.
Being a bot and all I'm not very good at pattern recognition, so I might be
wrong, but could you please double-check?

Full results can be found at:
https://testbot.winehq.org/JobDetails.pl?Key=64908

Your paranoid android.

=== wxppro (32 bit report) ===

=== w8 (32 bit report) ===

=== w8adm (32 bit report) ===

=== w864 (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1507 (32 bit report) ===

=== w1064v1809 (32 bit report) ===

=== w1064v1809_2scr (32 bit report) ===

=== w1064v1809_ar (32 bit report) ===

=== w1064v1809_he (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1809_ja (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1809_zh_CN (32 bit report) ===

=== w864 (64 bit report) ===

=== w1064v1507 (64 bit report) ===

=== w1064v1809 (64 bit report) ===

mfplat:
mfplat.c:2419: Test failed: Unexpected return value 0x102.
mfplat.c:1770: Test failed: Failed to get event, hr 0xc00d3e85.
mfplat.c:1773: Test failed: Failed to get event, hr 0xc00d3e85.
mfplat.c:1776: Test failed: Failed to finalize GetEvent, hr 0xc00d3e85.
mfplat.c:1779: Test failed: Unexpected result, hr 0xc00d3e85.
190c:mfplat: unhandled exception c0000005 at 00000000004045C3

=== debian10 (32 bit report) ===

=== debian10 (32 bit French report) ===

=== debian10 (32 bit Japanese:Japan report) ===

mfplat:
mfplat.c:2342: Test failed: Unexpected counter value 0.
mfplat.c:2419: Test failed: Unexpected return value 0x102.
mfplat.c:2901: Test failed: Unexpected refcount 1.
Unhandled exception: page fault on execute access to 0x00000000 in 32-bit code (0x00000000).

Report validation errors:
mfplat:mfplat crashed (c0000005)

=== debian10 (32 bit Chinese:China report) ===

=== debian10 (32 bit WoW report) ===

=== debian10 (64 bit WoW report) ===

[PATCH 9/9] mfplat: Implement IMFByteStream::SetCurrentPosition() for file streams.

[Nikolay Sivov]

From: Derek Lesho

Signed-off-by: Derek Lesho 
Signed-off-by: Nikolay Sivov 
---
 dlls/mfplat/main.c         | 19 ++++++++++++-------
 dlls/mfplat/tests/mfplat.c | 20 +++++++++++++++++++-
 2 files changed, 31 insertions(+), 8 deletions(-)

diff --git a/dlls/mfplat/main.c b/dlls/mfplat/main.c
index 1ebd443131..264c10e5cf 100644
--- a/dlls/mfplat/main.c
+++ b/dlls/mfplat/main.c
@@ -3287,22 +3287,27 @@ static HRESULT WINAPI mfbytestream_SetLength(IMFByteStream *iface, QWORD length)
     return E_NOTIMPL;
 }

-static HRESULT WINAPI mfbytestream_GetCurrentPosition(IMFByteStream *iface, QWORD *position)
+static HRESULT WINAPI mfbytestream_SetCurrentPosition(IMFByteStream *iface, QWORD position)
 {
     mfbytestream *This = impl_from_IMFByteStream(iface);

-    FIXME("%p, %p\n", This, position);
+    FIXME("%p, %s\n", This, wine_dbgstr_longlong(position));

return E_NOTIMPL;
 }

-static HRESULT WINAPI mfbytestream_SetCurrentPosition(IMFByteStream *iface, QWORD position)
+static HRESULT WINAPI bytestream_file_GetCurrentPosition(IMFByteStream *iface, QWORD *position)
 {
-    mfbytestream *This = impl_from_IMFByteStream(iface);
+    struct bytestream *stream = impl_from_IMFByteStream(iface);

-    FIXME("%p, %s\n", This, wine_dbgstr_longlong(position));
+    TRACE("%p, %p.\n", iface, position);

-    return E_NOTIMPL;
+    if (!position)
+        return E_INVALIDARG;
+
+    *position = stream->position;
+
+    return S_OK;
 }

static HRESULT WINAPI bytestream_file_GetLength(IMFByteStream *iface, QWORD *length)
@@ -3453,7 +3458,7 @@ static const IMFByteStreamVtbl bytestream_file_vtbl =
     bytestream_GetCapabilities,
     bytestream_file_GetLength,
     mfbytestream_SetLength,
-    mfbytestream_GetCurrentPosition,
+    bytestream_file_GetCurrentPosition,
     mfbytestream_SetCurrentPosition,
     bytestream_file_IsEndOfStream,
     bytestream_file_Read,
diff --git a/dlls/mfplat/tests/mfplat.c b/dlls/mfplat/tests/mfplat.c
index 7ec0b01982..67123d0b69 100644
--- a/dlls/mfplat/tests/mfplat.c
+++ b/dlls/mfplat/tests/mfplat.c
@@ -1340,15 +1340,16 @@ static void test_MFCreateMFByteStreamOnStream(void)
 static void test_file_stream(void)
 {
     IMFByteStream *bytestream, *bytestream2;
+    QWORD bytestream_length, position;
     IMFAttributes *attributes = NULL;
     MF_ATTRIBUTE_TYPE item_type;
-    QWORD bytestream_length;
     WCHAR pathW[MAX_PATH];
     DWORD caps, count;
     WCHAR *filename;
     IUnknown *unk;
     HRESULT hr;
     WCHAR *str;
+    BOOL eos;

static const WCHAR newfilename[] = {'n','e','w','.','m','p','4',0};

@@ -1415,6 +1416,23 @@ static void test_file_stream(void)
     ok(hr == S_OK, "Failed to get bytestream length, hr %#x.\n", hr);
     ok(bytestream_length > 0, "Unexpected bytestream length %s.\n", wine_dbgstr_longlong(bytestream_length));

+    hr = IMFByteStream_SetCurrentPosition(bytestream, bytestream_length);
+    ok(hr == S_OK, "Failed to set bytestream position, hr %#x.\n", hr);
+
+    hr = IMFByteStream_IsEndOfStream(bytestream, &eos);
+    ok(hr == S_OK, "Failed query end of stream, hr %#x.\n", hr);
+    ok(eos == TRUE, "Unexpected IsEndOfStream result, %u.\n", eos);
+
+    hr = IMFByteStream_SetCurrentPosition(bytestream, 2 * bytestream_length);
+    ok(hr == S_OK, "Failed to set bytestream position, hr %#x.\n", hr);
+
+    hr = IMFByteStream_GetCurrentPosition(bytestream, NULL);
+    ok(hr == E_INVALIDARG, "Unexpected hr %#x.\n", hr);
+
+    hr = IMFByteStream_GetCurrentPosition(bytestream, &position);
+    ok(hr == S_OK, "Unexpected hr %#x.\n", hr);
+    ok(position == 2 * bytestream_length, "Unexpected position.\n");
+
     hr = MFCreateFile(MF_ACCESSMODE_READ, MF_OPENMODE_FAIL_IF_NOT_EXIST,
                       MF_FILEFLAGS_NONE, filename, &bytestream2);
     ok(hr == S_OK, "got 0x%08x\n", hr);

[PATCH 9/9] mfplat: Implement IMFByteStream::SetCurrentPosition() for file streams.

[Marvin]

Hi,

While running your changed tests, I think I found new failures.
Being a bot and all I'm not very good at pattern recognition, so I might be
wrong, but could you please double-check?

Full results can be found at:
https://testbot.winehq.org/JobDetails.pl?Key=64909

Your paranoid android.

=== wxppro (32 bit report) ===

=== w8 (32 bit report) ===

=== w8adm (32 bit report) ===

=== w864 (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1507 (32 bit report) ===

=== w1064v1809 (32 bit report) ===

=== w1064v1809_2scr (32 bit report) ===

=== w1064v1809_ar (32 bit report) ===

=== w1064v1809_he (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1809_ja (32 bit report) ===

Report validation errors:
mfplat:mfplat has no test summary line (early exit of the main process?)

=== w1064v1809_zh_CN (32 bit report) ===

=== w864 (64 bit report) ===

=== w1064v1507 (64 bit report) ===

=== w1064v1809 (64 bit report) ===

=== debian10 (32 bit report) ===

mfplat:
mfplat.c:1420: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1424: Test failed: Unexpected IsEndOfStream result, 0.
mfplat.c:1427: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1434: Test failed: Unexpected position.

Report validation errors:
mfplat:mfplat has unaccounted for failure messages

=== debian10 (32 bit French report) ===

mfplat:
mfplat.c:1420: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1424: Test failed: Unexpected IsEndOfStream result, 0.
mfplat.c:1427: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1434: Test failed: Unexpected position.

Report validation errors:
mfplat:mfplat has unaccounted for failure messages

=== debian10 (32 bit Japanese:Japan report) ===

mfplat:
mfplat.c:1420: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1424: Test failed: Unexpected IsEndOfStream result, 0.
mfplat.c:1427: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1434: Test failed: Unexpected position.
mfplat.c:2919: Test failed: Unexpected refcount 1.
Unhandled exception: page fault on read access to 0x00000009 in 32-bit code (0x658820b7).

Report validation errors:
mfplat:mfplat crashed (c0000005)

=== debian10 (32 bit Chinese:China report) ===

mfplat:
mfplat.c:1420: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1424: Test failed: Unexpected IsEndOfStream result, 0.
mfplat.c:1427: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1434: Test failed: Unexpected position.

Report validation errors:
mfplat:mfplat has unaccounted for failure messages

=== debian10 (32 bit WoW report) ===

mfplat:
mfplat.c:1420: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1424: Test failed: Unexpected IsEndOfStream result, 0.
mfplat.c:1427: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1434: Test failed: Unexpected position.

Report validation errors:
mfplat:mfplat has unaccounted for failure messages

=== debian10 (64 bit WoW report) ===

mfplat:
mfplat.c:1420: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1424: Test failed: Unexpected IsEndOfStream result, 0.
mfplat.c:1427: Test failed: Failed to set bytestream position, hr 0x80004001.
mfplat.c:1434: Test failed: Unexpected position.

Report validation errors:
mfplat:mfplat has unaccounted for failure messages