dlls/rtworkq/queue.c

   1 /*
   2  * Copyright 2019-2020 Nikolay Sivov for CodeWeavers
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Lesser General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2.1 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Lesser General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Lesser General Public
  15  * License along with this library; if not, write to the Free Software
  16  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
  17  */
  18
  19 #include <assert.h>
  20
  21 #define COBJMACROS
  22 #define NONAMELESSUNION
  23
  24 #include "initguid.h"
  25 #include "rtworkq.h"
  26 #include "wine/debug.h"
  27 #include "wine/list.h"
  28
  29 WINE_DEFAULT_DEBUG_CHANNEL(mfplat);
  30
  31 #define FIRST_USER_QUEUE_HANDLE 5
  32 #define MAX_USER_QUEUE_HANDLES 124
  33
  34 #define WAIT_ITEM_KEY_MASK      (0x82000000)
  35 #define SCHEDULED_ITEM_KEY_MASK (0x80000000)
  36
  37 static LONG next_item_key;
  38
  39 static RTWQWORKITEM_KEY get_item_key(DWORD mask, DWORD key)
  40 {
  41     return ((RTWQWORKITEM_KEY)mask << 32) | key;
  42 }
  43
  44 static RTWQWORKITEM_KEY generate_item_key(DWORD mask)
  45 {
  46     return get_item_key(mask, InterlockedIncrement(&next_item_key));
  47 }
  48
  49 struct queue_handle
  50 {
  51     void *obj;
  52     LONG refcount;
  53     WORD generation;
  54 };
  55
  56 static struct queue_handle user_queues[MAX_USER_QUEUE_HANDLES];
  57 static struct queue_handle *next_free_user_queue;
  58 static struct queue_handle *next_unused_user_queue = user_queues;
  59 static WORD queue_generation;
  60 static DWORD shared_mt_queue;
  61
  62 static CRITICAL_SECTION queues_section;
  63 static CRITICAL_SECTION_DEBUG queues_critsect_debug =
  64 {
  65     0, 0, &queues_section,
  66     { &queues_critsect_debug.ProcessLocksList, &queues_critsect_debug.ProcessLocksList },
  67       0, 0, { (DWORD_PTR)(__FILE__ ": queues_section") }
  68 };
  69 static CRITICAL_SECTION queues_section = { &queues_critsect_debug, -1, 0, 0, 0, 0 };
  70
  71 static LONG platform_lock;
  72 static CO_MTA_USAGE_COOKIE mta_cookie;
  73
  74 static struct queue_handle *get_queue_obj(DWORD handle)
  75 {
  76     unsigned int idx = HIWORD(handle) - FIRST_USER_QUEUE_HANDLE;
  77
  78     if (idx < MAX_USER_QUEUE_HANDLES && user_queues[idx].refcount)
  79     {
  80         if (LOWORD(handle) == user_queues[idx].generation)
  81             return &user_queues[idx];
  82     }
  83
  84     return NULL;
  85 }
  86
  87 /* Should be kept in sync with corresponding MFASYNC_CALLBACK_ constants. */
  88 enum rtwq_callback_queue_id
  89 {
  90     RTWQ_CALLBACK_QUEUE_UNDEFINED     = 0x00000000,
  91     RTWQ_CALLBACK_QUEUE_STANDARD      = 0x00000001,
  92     RTWQ_CALLBACK_QUEUE_RT            = 0x00000002,
  93     RTWQ_CALLBACK_QUEUE_IO            = 0x00000003,
  94     RTWQ_CALLBACK_QUEUE_TIMER         = 0x00000004,
  95     RTWQ_CALLBACK_QUEUE_MULTITHREADED = 0x00000005,
  96     RTWQ_CALLBACK_QUEUE_LONG_FUNCTION = 0x00000007,
  97     RTWQ_CALLBACK_QUEUE_PRIVATE_MASK  = 0xffff0000,
  98     RTWQ_CALLBACK_QUEUE_ALL           = 0xffffffff,
  99 };
 100
 101 /* Should be kept in sync with corresponding MFASYNC_ constants. */
 102 enum rtwq_callback_flags
 103 {
 104     RTWQ_FAST_IO_PROCESSING_CALLBACK = 0x00000001,
 105     RTWQ_SIGNAL_CALLBACK             = 0x00000002,
 106     RTWQ_BLOCKING_CALLBACK           = 0x00000004,
 107     RTWQ_REPLY_CALLBACK              = 0x00000008,
 108     RTWQ_LOCALIZE_REMOTE_CALLBACK    = 0x00000010,
 109 };
 110
 111 enum system_queue_index
 112 {
 113     SYS_QUEUE_STANDARD = 0,
 114     SYS_QUEUE_RT,
 115     SYS_QUEUE_IO,
 116     SYS_QUEUE_TIMER,
 117     SYS_QUEUE_MULTITHREADED,
 118     SYS_QUEUE_DO_NOT_USE,
 119     SYS_QUEUE_LONG_FUNCTION,
 120     SYS_QUEUE_COUNT,
 121 };
 122
 123 struct work_item
 124 {
 125     IUnknown IUnknown_iface;
 126     LONG refcount;
 127     struct list entry;
 128     IRtwqAsyncResult *result;
 129     IRtwqAsyncResult *reply_result;
 130     struct queue *queue;
 131     RTWQWORKITEM_KEY key;
 132     LONG priority;
 133     DWORD flags;
 134     TP_WORK *work_object;
 135     PTP_SIMPLE_CALLBACK finalization_callback;
 136     union
 137     {
 138         TP_WAIT *wait_object;
 139         TP_TIMER *timer_object;
 140     } u;
 141 };
 142
 143 static struct work_item *work_item_impl_from_IUnknown(IUnknown *iface)
 144 {
 145     return CONTAINING_RECORD(iface, struct work_item, IUnknown_iface);
 146 }
 147
 148 static const TP_CALLBACK_PRIORITY priorities[] =
 149 {
 150     TP_CALLBACK_PRIORITY_HIGH,
 151     TP_CALLBACK_PRIORITY_NORMAL,
 152     TP_CALLBACK_PRIORITY_LOW,
 153 };
 154
 155 struct queue;
 156 struct queue_desc;
 157
 158 struct queue_ops
 159 {
 160     HRESULT (*init)(const struct queue_desc *desc, struct queue *queue);
 161     BOOL (*shutdown)(struct queue *queue);
 162     void (*submit)(struct queue *queue, struct work_item *item);
 163 };
 164
 165 struct queue_desc
 166 {
 167     RTWQ_WORKQUEUE_TYPE queue_type;
 168     const struct queue_ops *ops;
 169     DWORD target_queue;
 170 };
 171
 172 struct queue
 173 {
 174     IRtwqAsyncCallback IRtwqAsyncCallback_iface;
 175     const struct queue_ops *ops;
 176     TP_POOL *pool;
 177     TP_CALLBACK_ENVIRON_V3 envs[ARRAY_SIZE(priorities)];
 178     CRITICAL_SECTION cs;
 179     struct list pending_items;
 180     DWORD id;
 181     /* Data used for serial queues only. */
 182     PTP_SIMPLE_CALLBACK finalization_callback;
 183     DWORD target_queue;
 184 };
 185
 186 static void shutdown_queue(struct queue *queue);
 187
 188 static HRESULT lock_user_queue(DWORD queue)
 189 {
 190     HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
 191     struct queue_handle *entry;
 192
 193     if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
 194         return S_OK;
 195
 196     EnterCriticalSection(&queues_section);
 197     entry = get_queue_obj(queue);
 198     if (entry && entry->refcount)
 199     {
 200         entry->refcount++;
 201         hr = S_OK;
 202     }
 203     LeaveCriticalSection(&queues_section);
 204     return hr;
 205 }
 206
 207 static HRESULT unlock_user_queue(DWORD queue)
 208 {
 209     HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
 210     struct queue_handle *entry;
 211
 212     if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
 213         return S_OK;
 214
 215     EnterCriticalSection(&queues_section);
 216     entry = get_queue_obj(queue);
 217     if (entry && entry->refcount)
 218     {
 219         if (--entry->refcount == 0)
 220         {
 221             if (shared_mt_queue == queue) shared_mt_queue = 0;
 222             shutdown_queue((struct queue *)entry->obj);
 223             free(entry->obj);
 224             entry->obj = next_free_user_queue;
 225             next_free_user_queue = entry;
 226         }
 227         hr = S_OK;
 228     }
 229     LeaveCriticalSection(&queues_section);
 230     return hr;
 231 }
 232
 233 static struct queue *queue_impl_from_IRtwqAsyncCallback(IRtwqAsyncCallback *iface)
 234 {
 235     return CONTAINING_RECORD(iface, struct queue, IRtwqAsyncCallback_iface);
 236 }
 237
 238 static HRESULT WINAPI queue_serial_callback_QueryInterface(IRtwqAsyncCallback *iface, REFIID riid, void **obj)
 239 {
 240     if (IsEqualIID(riid, &IID_IRtwqAsyncCallback) ||
 241             IsEqualIID(riid, &IID_IUnknown))
 242     {
 243         *obj = iface;
 244         IRtwqAsyncCallback_AddRef(iface);
 245         return S_OK;
 246     }
 247
 248     *obj = NULL;
 249     return E_NOINTERFACE;
 250 }
 251
 252 static ULONG WINAPI queue_serial_callback_AddRef(IRtwqAsyncCallback *iface)
 253 {
 254     return 2;
 255 }
 256
 257 static ULONG WINAPI queue_serial_callback_Release(IRtwqAsyncCallback *iface)
 258 {
 259     return 1;
 260 }
 261
 262 static HRESULT WINAPI queue_serial_callback_GetParameters(IRtwqAsyncCallback *iface, DWORD *flags, DWORD *queue_id)
 263 {
 264     struct queue *queue = queue_impl_from_IRtwqAsyncCallback(iface);
 265
 266     *flags = 0;
 267     *queue_id = queue->id;
 268
 269     return S_OK;
 270 }
 271
 272 static HRESULT WINAPI queue_serial_callback_Invoke(IRtwqAsyncCallback *iface, IRtwqAsyncResult *result)
 273 {
 274     /* Reply callback won't be called in a regular way, pending items and chained queues will make it
 275        unnecessary complicated to reach actual work queue that's able to execute this item. Instead
 276        serial queues are cleaned up right away on submit(). */
 277     return S_OK;
 278 }
 279
 280 static const IRtwqAsyncCallbackVtbl queue_serial_callback_vtbl =
 281 {
 282     queue_serial_callback_QueryInterface,
 283     queue_serial_callback_AddRef,
 284     queue_serial_callback_Release,
 285     queue_serial_callback_GetParameters,
 286     queue_serial_callback_Invoke,
 287 };
 288
 289 static struct queue system_queues[SYS_QUEUE_COUNT];
 290
 291 static struct queue *get_system_queue(DWORD queue_id)
 292 {
 293     switch (queue_id)
 294     {
 295         case RTWQ_CALLBACK_QUEUE_STANDARD:
 296         case RTWQ_CALLBACK_QUEUE_RT:
 297         case RTWQ_CALLBACK_QUEUE_IO:
 298         case RTWQ_CALLBACK_QUEUE_TIMER:
 299         case RTWQ_CALLBACK_QUEUE_MULTITHREADED:
 300         case RTWQ_CALLBACK_QUEUE_LONG_FUNCTION:
 301             return &system_queues[queue_id - 1];
 302         default:
 303             return NULL;
 304     }
 305 }
 306
 307 static HRESULT grab_queue(DWORD queue_id, struct queue **ret);
 308
 309 static void CALLBACK standard_queue_cleanup_callback(void *object_data, void *group_data)
 310 {
 311 }
 312
 313 static HRESULT pool_queue_init(const struct queue_desc *desc, struct queue *queue)
 314 {
 315     TP_CALLBACK_ENVIRON_V3 env;
 316     unsigned int max_thread, i;
 317
 318     queue->pool = CreateThreadpool(NULL);
 319
 320     memset(&env, 0, sizeof(env));
 321     env.Version = 3;
 322     env.Size = sizeof(env);
 323     env.Pool = queue->pool;
 324     env.CleanupGroup = CreateThreadpoolCleanupGroup();
 325     env.CleanupGroupCancelCallback = standard_queue_cleanup_callback;
 326     env.CallbackPriority = TP_CALLBACK_PRIORITY_NORMAL;
 327     for (i = 0; i < ARRAY_SIZE(queue->envs); ++i)
 328     {
 329         queue->envs[i] = env;
 330         queue->envs[i].CallbackPriority = priorities[i];
 331     }
 332     list_init(&queue->pending_items);
 333     InitializeCriticalSection(&queue->cs);
 334
 335     max_thread = (desc->queue_type == RTWQ_STANDARD_WORKQUEUE || desc->queue_type == RTWQ_WINDOW_WORKQUEUE) ? 1 : 4;
 336
 337     SetThreadpoolThreadMinimum(queue->pool, 1);
 338     SetThreadpoolThreadMaximum(queue->pool, max_thread);
 339
 340     if (desc->queue_type == RTWQ_WINDOW_WORKQUEUE)
 341         FIXME("RTWQ_WINDOW_WORKQUEUE is not supported.\n");
 342
 343     return S_OK;
 344 }
 345
 346 static BOOL pool_queue_shutdown(struct queue *queue)
 347 {
 348     if (!queue->pool)
 349         return FALSE;
 350
 351     CloseThreadpoolCleanupGroupMembers(queue->envs[0].CleanupGroup, TRUE, NULL);
 352     CloseThreadpool(queue->pool);
 353     queue->pool = NULL;
 354
 355     return TRUE;
 356 }
 357
 358 static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void *context, TP_WORK *work)
 359 {
 360     struct work_item *item = context;
 361     RTWQASYNCRESULT *result = (RTWQASYNCRESULT *)item->result;
 362
 363     TRACE("result object %p.\n", result);
 364
 365     /* Submitting from serial queue in reply mode, use different result object acting as receipt token.
 366        It's submitted to user callback still, but when invoked, special serial queue callback will be used
 367        to ensure correct destination queue. */
 368
 369     IRtwqAsyncCallback_Invoke(result->pCallback, item->reply_result ? item->reply_result : item->result);
 370
 371     IUnknown_Release(&item->IUnknown_iface);
 372 }
 373
 374 static void pool_queue_submit(struct queue *queue, struct work_item *item)
 375 {
 376     TP_CALLBACK_PRIORITY callback_priority;
 377     TP_CALLBACK_ENVIRON_V3 env;
 378
 379     if (item->priority == 0)
 380         callback_priority = TP_CALLBACK_PRIORITY_NORMAL;
 381     else if (item->priority < 0)
 382         callback_priority = TP_CALLBACK_PRIORITY_LOW;
 383     else
 384         callback_priority = TP_CALLBACK_PRIORITY_HIGH;
 385
 386     env = queue->envs[callback_priority];
 387     env.FinalizationCallback = item->finalization_callback;
 388     /* Worker pool callback will release one reference. Grab one more to keep object alive when
 389        we need finalization callback. */
 390     if (item->finalization_callback)
 391         IUnknown_AddRef(&item->IUnknown_iface);
 392     item->work_object = CreateThreadpoolWork(standard_queue_worker, item, (TP_CALLBACK_ENVIRON *)&env);
 393     SubmitThreadpoolWork(item->work_object);
 394
 395     TRACE("dispatched %p.\n", item->result);
 396 }
 397
 398 static const struct queue_ops pool_queue_ops =
 399 {
 400     pool_queue_init,
 401     pool_queue_shutdown,
 402     pool_queue_submit,
 403 };
 404
 405 static struct work_item * serial_queue_get_next(struct queue *queue, struct work_item *item)
 406 {
 407     struct work_item *next_item = NULL;
 408
 409     list_remove(&item->entry);
 410     if (!list_empty(&item->queue->pending_items))
 411         next_item = LIST_ENTRY(list_head(&item->queue->pending_items), struct work_item, entry);
 412
 413     return next_item;
 414 }
 415
 416 static void CALLBACK serial_queue_finalization_callback(PTP_CALLBACK_INSTANCE instance, void *user_data)
 417 {
 418     struct work_item *item = (struct work_item *)user_data, *next_item;
 419     struct queue *target_queue, *queue = item->queue;
 420     HRESULT hr;
 421
 422     EnterCriticalSection(&queue->cs);
 423
 424     if ((next_item = serial_queue_get_next(queue, item)))
 425     {
 426         if (SUCCEEDED(hr = grab_queue(queue->target_queue, &target_queue)))
 427             target_queue->ops->submit(target_queue, next_item);
 428         else
 429             WARN("Failed to grab queue for id %#lx, hr %#lx.\n", queue->target_queue, hr);
 430     }
 431
 432     LeaveCriticalSection(&queue->cs);
 433
 434     IUnknown_Release(&item->IUnknown_iface);
 435 }
 436
 437 static HRESULT serial_queue_init(const struct queue_desc *desc, struct queue *queue)
 438 {
 439     queue->IRtwqAsyncCallback_iface.lpVtbl = &queue_serial_callback_vtbl;
 440     queue->target_queue = desc->target_queue;
 441     lock_user_queue(queue->target_queue);
 442     queue->finalization_callback = serial_queue_finalization_callback;
 443
 444     return S_OK;
 445 }
 446
 447 static BOOL serial_queue_shutdown(struct queue *queue)
 448 {
 449     unlock_user_queue(queue->target_queue);
 450
 451     return TRUE;
 452 }
 453
 454 static struct work_item * serial_queue_is_ack_token(struct queue *queue, struct work_item *item)
 455 {
 456     RTWQASYNCRESULT *async_result = (RTWQASYNCRESULT *)item->result;
 457     struct work_item *head;
 458
 459     if (list_empty(&queue->pending_items))
 460         return NULL;
 461
 462     head = LIST_ENTRY(list_head(&queue->pending_items), struct work_item, entry);
 463     if (head->reply_result == item->result && async_result->pCallback == &queue->IRtwqAsyncCallback_iface)
 464         return head;
 465
 466     return NULL;
 467 }
 468
 469 static void serial_queue_submit(struct queue *queue, struct work_item *item)
 470 {
 471     struct work_item *head, *next_item = NULL;
 472     struct queue *target_queue;
 473     HRESULT hr;
 474
 475     /* In reply mode queue will advance when 'reply_result' is invoked, in regular mode it will advance automatically,
 476        via finalization callback. */
 477
 478     if (item->flags & RTWQ_REPLY_CALLBACK)
 479     {
 480         if (FAILED(hr = RtwqCreateAsyncResult(NULL, &queue->IRtwqAsyncCallback_iface, NULL, &item->reply_result)))
 481             WARN("Failed to create reply object, hr %#lx.\n", hr);
 482     }
 483     else
 484         item->finalization_callback = queue->finalization_callback;
 485
 486     /* Serial queues could be chained together, detach from current queue before transitioning item to this one.
 487        Items are not detached when submitted to pool queues, because pool queues won't forward them further. */
 488     EnterCriticalSection(&item->queue->cs);
 489     list_remove(&item->entry);
 490     LeaveCriticalSection(&item->queue->cs);
 491
 492     EnterCriticalSection(&queue->cs);
 493
 494     item->queue = queue;
 495
 496     if ((head = serial_queue_is_ack_token(queue, item)))
 497     {
 498         /* Ack receipt token - pop waiting item, advance. */
 499         next_item = serial_queue_get_next(queue, head);
 500         IUnknown_Release(&head->IUnknown_iface);
 501     }
 502     else
 503     {
 504         if (list_empty(&queue->pending_items))
 505             next_item = item;
 506         list_add_tail(&queue->pending_items, &item->entry);
 507         IUnknown_AddRef(&item->IUnknown_iface);
 508     }
 509
 510     if (next_item)
 511     {
 512         if (SUCCEEDED(hr = grab_queue(queue->target_queue, &target_queue)))
 513             target_queue->ops->submit(target_queue, next_item);
 514         else
 515             WARN("Failed to grab queue for id %#lx, hr %#lx.\n", queue->target_queue, hr);
 516     }
 517
 518     LeaveCriticalSection(&queue->cs);
 519 }
 520
 521 static const struct queue_ops serial_queue_ops =
 522 {
 523     serial_queue_init,
 524     serial_queue_shutdown,
 525     serial_queue_submit,
 526 };
 527
 528 static HRESULT WINAPI work_item_QueryInterface(IUnknown *iface, REFIID riid, void **obj)
 529 {
 530     if (IsEqualIID(riid, &IID_IUnknown))
 531     {
 532         *obj = iface;
 533         IUnknown_AddRef(iface);
 534         return S_OK;
 535     }
 536
 537     *obj = NULL;
 538     return E_NOINTERFACE;
 539 }
 540
 541 static ULONG WINAPI work_item_AddRef(IUnknown *iface)
 542 {
 543     struct work_item *item = work_item_impl_from_IUnknown(iface);
 544     return InterlockedIncrement(&item->refcount);
 545 }
 546
 547 static ULONG WINAPI work_item_Release(IUnknown *iface)
 548 {
 549     struct work_item *item = work_item_impl_from_IUnknown(iface);
 550     ULONG refcount = InterlockedDecrement(&item->refcount);
 551
 552     if (!refcount)
 553     {
 554         if (item->work_object)
 555             CloseThreadpoolWork(item->work_object);
 556         if (item->reply_result)
 557             IRtwqAsyncResult_Release(item->reply_result);
 558         IRtwqAsyncResult_Release(item->result);
 559         free(item);
 560     }
 561
 562     return refcount;
 563 }
 564
 565 static const IUnknownVtbl work_item_vtbl =
 566 {
 567     work_item_QueryInterface,
 568     work_item_AddRef,
 569     work_item_Release,
 570 };
 571
 572 static struct work_item * alloc_work_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 573 {
 574     RTWQASYNCRESULT *async_result = (RTWQASYNCRESULT *)result;
 575     DWORD flags = 0, queue_id = 0;
 576     struct work_item *item;
 577
 578     item = calloc(1, sizeof(*item));
 579
 580     item->IUnknown_iface.lpVtbl = &work_item_vtbl;
 581     item->result = result;
 582     IRtwqAsyncResult_AddRef(item->result);
 583     item->refcount = 1;
 584     item->queue = queue;
 585     list_init(&item->entry);
 586     item->priority = priority;
 587
 588     if (SUCCEEDED(IRtwqAsyncCallback_GetParameters(async_result->pCallback, &flags, &queue_id)))
 589         item->flags = flags;
 590
 591     return item;
 592 }
 593
 594 static void init_work_queue(const struct queue_desc *desc, struct queue *queue)
 595 {
 596     assert(desc->ops != NULL);
 597
 598     queue->ops = desc->ops;
 599     if (SUCCEEDED(queue->ops->init(desc, queue)))
 600     {
 601         list_init(&queue->pending_items);
 602         InitializeCriticalSection(&queue->cs);
 603     }
 604 }
 605
 606 static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
 607 {
 608     struct queue *queue = get_system_queue(queue_id);
 609     RTWQ_WORKQUEUE_TYPE queue_type;
 610     struct queue_handle *entry;
 611
 612     *ret = NULL;
 613
 614     if (!system_queues[SYS_QUEUE_STANDARD].pool)
 615         return RTWQ_E_SHUTDOWN;
 616
 617     if (queue && queue->pool)
 618     {
 619         *ret = queue;
 620         return S_OK;
 621     }
 622     else if (queue)
 623     {
 624         struct queue_desc desc;
 625
 626         EnterCriticalSection(&queues_section);
 627         switch (queue_id)
 628         {
 629             case RTWQ_CALLBACK_QUEUE_IO:
 630             case RTWQ_CALLBACK_QUEUE_MULTITHREADED:
 631             case RTWQ_CALLBACK_QUEUE_LONG_FUNCTION:
 632                 queue_type = RTWQ_MULTITHREADED_WORKQUEUE;
 633                 break;
 634             default:
 635                 queue_type = RTWQ_STANDARD_WORKQUEUE;
 636         }
 637
 638         desc.queue_type = queue_type;
 639         desc.ops = &pool_queue_ops;
 640         desc.target_queue = 0;
 641         init_work_queue(&desc, queue);
 642         LeaveCriticalSection(&queues_section);
 643         *ret = queue;
 644         return S_OK;
 645     }
 646
 647     /* Handles user queues. */
 648     if ((entry = get_queue_obj(queue_id)))
 649         *ret = entry->obj;
 650
 651     return *ret ? S_OK : RTWQ_E_INVALID_WORKQUEUE;
 652 }
 653
 654 static void shutdown_queue(struct queue *queue)
 655 {
 656     struct work_item *item, *item2;
 657
 658     if (!queue->ops || !queue->ops->shutdown(queue))
 659         return;
 660
 661     EnterCriticalSection(&queue->cs);
 662     LIST_FOR_EACH_ENTRY_SAFE(item, item2, &queue->pending_items, struct work_item, entry)
 663     {
 664         list_remove(&item->entry);
 665         IUnknown_Release(&item->IUnknown_iface);
 666     }
 667     LeaveCriticalSection(&queue->cs);
 668
 669     DeleteCriticalSection(&queue->cs);
 670
 671     memset(queue, 0, sizeof(*queue));
 672 }
 673
 674 static HRESULT queue_submit_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 675 {
 676     struct work_item *item;
 677
 678     if (!(item = alloc_work_item(queue, priority, result)))
 679         return E_OUTOFMEMORY;
 680
 681     queue->ops->submit(queue, item);
 682
 683     return S_OK;
 684 }
 685
 686 static HRESULT queue_put_work_item(DWORD queue_id, LONG priority, IRtwqAsyncResult *result)
 687 {
 688     struct queue *queue;
 689     HRESULT hr;
 690
 691     if (FAILED(hr = grab_queue(queue_id, &queue)))
 692         return hr;
 693
 694     return queue_submit_item(queue, priority, result);
 695 }
 696
 697 static HRESULT invoke_async_callback(IRtwqAsyncResult *result)
 698 {
 699     RTWQASYNCRESULT *result_data = (RTWQASYNCRESULT *)result;
 700     DWORD queue = RTWQ_CALLBACK_QUEUE_STANDARD, flags;
 701     HRESULT hr;
 702
 703     if (FAILED(IRtwqAsyncCallback_GetParameters(result_data->pCallback, &flags, &queue)))
 704         queue = RTWQ_CALLBACK_QUEUE_STANDARD;
 705
 706     if (FAILED(lock_user_queue(queue)))
 707         queue = RTWQ_CALLBACK_QUEUE_STANDARD;
 708
 709     hr = queue_put_work_item(queue, 0, result);
 710
 711     unlock_user_queue(queue);
 712
 713     return hr;
 714 }
 715
 716 static void queue_release_pending_item(struct work_item *item)
 717 {
 718     EnterCriticalSection(&item->queue->cs);
 719     if (item->key)
 720     {
 721         list_remove(&item->entry);
 722         item->key = 0;
 723         IUnknown_Release(&item->IUnknown_iface);
 724     }
 725     LeaveCriticalSection(&item->queue->cs);
 726 }
 727
 728 static void CALLBACK waiting_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_WAIT *wait,
 729         TP_WAIT_RESULT wait_result)
 730 {
 731     struct work_item *item = context;
 732
 733     TRACE("result object %p.\n", item->result);
 734
 735     invoke_async_callback(item->result);
 736
 737     IUnknown_Release(&item->IUnknown_iface);
 738 }
 739
 740 static void CALLBACK waiting_item_cancelable_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_WAIT *wait,
 741         TP_WAIT_RESULT wait_result)
 742 {
 743     struct work_item *item = context;
 744
 745     TRACE("result object %p.\n", item->result);
 746
 747     queue_release_pending_item(item);
 748
 749     invoke_async_callback(item->result);
 750
 751     IUnknown_Release(&item->IUnknown_iface);
 752 }
 753
 754 static void CALLBACK scheduled_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 755 {
 756     struct work_item *item = context;
 757
 758     TRACE("result object %p.\n", item->result);
 759
 760     invoke_async_callback(item->result);
 761
 762     IUnknown_Release(&item->IUnknown_iface);
 763 }
 764
 765 static void CALLBACK scheduled_item_cancelable_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 766 {
 767     struct work_item *item = context;
 768
 769     TRACE("result object %p.\n", item->result);
 770
 771     queue_release_pending_item(item);
 772
 773     invoke_async_callback(item->result);
 774
 775     IUnknown_Release(&item->IUnknown_iface);
 776 }
 777
 778 static void CALLBACK periodic_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 779 {
 780     struct work_item *item = context;
 781
 782     IUnknown_AddRef(&item->IUnknown_iface);
 783
 784     invoke_async_callback(item->result);
 785
 786     IUnknown_Release(&item->IUnknown_iface);
 787 }
 788
 789 static void queue_mark_item_pending(DWORD mask, struct work_item *item, RTWQWORKITEM_KEY *key)
 790 {
 791     *key = generate_item_key(mask);
 792     item->key = *key;
 793
 794     EnterCriticalSection(&item->queue->cs);
 795     list_add_tail(&item->queue->pending_items, &item->entry);
 796     IUnknown_AddRef(&item->IUnknown_iface);
 797     LeaveCriticalSection(&item->queue->cs);
 798 }
 799
 800 static HRESULT queue_submit_wait(struct queue *queue, HANDLE event, LONG priority, IRtwqAsyncResult *result,
 801         RTWQWORKITEM_KEY *key)
 802 {
 803     PTP_WAIT_CALLBACK callback;
 804     struct work_item *item;
 805
 806     if (!(item = alloc_work_item(queue, priority, result)))
 807         return E_OUTOFMEMORY;
 808
 809     if (key)
 810     {
 811         queue_mark_item_pending(WAIT_ITEM_KEY_MASK, item, key);
 812         callback = waiting_item_cancelable_callback;
 813     }
 814     else
 815         callback = waiting_item_callback;
 816
 817     item->u.wait_object = CreateThreadpoolWait(callback, item,
 818             (TP_CALLBACK_ENVIRON *)&queue->envs[TP_CALLBACK_PRIORITY_NORMAL]);
 819     SetThreadpoolWait(item->u.wait_object, event, NULL);
 820
 821     TRACE("dispatched %p.\n", result);
 822
 823     return S_OK;
 824 }
 825
 826 static HRESULT queue_submit_timer(struct queue *queue, IRtwqAsyncResult *result, INT64 timeout, DWORD period,
 827         RTWQWORKITEM_KEY *key)
 828 {
 829     PTP_TIMER_CALLBACK callback;
 830     struct work_item *item;
 831     FILETIME filetime;
 832     LARGE_INTEGER t;
 833
 834     if (!(item = alloc_work_item(queue, 0, result)))
 835         return E_OUTOFMEMORY;
 836
 837     if (key)
 838     {
 839         queue_mark_item_pending(SCHEDULED_ITEM_KEY_MASK, item, key);
 840     }
 841
 842     if (period)
 843         callback = periodic_item_callback;
 844     else
 845         callback = key ? scheduled_item_cancelable_callback : scheduled_item_callback;
 846
 847     t.QuadPart = timeout * 1000 * 10;
 848     filetime.dwLowDateTime = t.u.LowPart;
 849     filetime.dwHighDateTime = t.u.HighPart;
 850
 851     item->u.timer_object = CreateThreadpoolTimer(callback, item,
 852             (TP_CALLBACK_ENVIRON *)&queue->envs[TP_CALLBACK_PRIORITY_NORMAL]);
 853     SetThreadpoolTimer(item->u.timer_object, &filetime, period, 0);
 854
 855     TRACE("dispatched %p.\n", result);
 856
 857     return S_OK;
 858 }
 859
 860 static HRESULT queue_cancel_item(struct queue *queue, RTWQWORKITEM_KEY key)
 861 {
 862     HRESULT hr = RTWQ_E_NOT_FOUND;
 863     struct work_item *item;
 864
 865     EnterCriticalSection(&queue->cs);
 866     LIST_FOR_EACH_ENTRY(item, &queue->pending_items, struct work_item, entry)
 867     {
 868         if (item->key == key)
 869         {
 870             key >>= 32;
 871             if ((key & WAIT_ITEM_KEY_MASK) == WAIT_ITEM_KEY_MASK)
 872             {
 873                 IRtwqAsyncResult_SetStatus(item->result, RTWQ_E_OPERATION_CANCELLED);
 874                 invoke_async_callback(item->result);
 875                 CloseThreadpoolWait(item->u.wait_object);
 876             }
 877             else if ((key & SCHEDULED_ITEM_KEY_MASK) == SCHEDULED_ITEM_KEY_MASK)
 878                 CloseThreadpoolTimer(item->u.timer_object);
 879             else
 880                 WARN("Unknown item key mask %#I64x.\n", key);
 881             queue_release_pending_item(item);
 882             hr = S_OK;
 883             break;
 884         }
 885     }
 886     LeaveCriticalSection(&queue->cs);
 887
 888     return hr;
 889 }
 890
 891 static HRESULT alloc_user_queue(const struct queue_desc *desc, DWORD *queue_id)
 892 {
 893     struct queue_handle *entry;
 894     struct queue *queue;
 895     unsigned int idx;
 896
 897     *queue_id = RTWQ_CALLBACK_QUEUE_UNDEFINED;
 898
 899     if (platform_lock <= 0)
 900         return RTWQ_E_SHUTDOWN;
 901
 902     if (!(queue = calloc(1, sizeof(*queue))))
 903         return E_OUTOFMEMORY;
 904
 905     init_work_queue(desc, queue);
 906
 907     EnterCriticalSection(&queues_section);
 908
 909     entry = next_free_user_queue;
 910     if (entry)
 911         next_free_user_queue = entry->obj;
 912     else if (next_unused_user_queue < user_queues + MAX_USER_QUEUE_HANDLES)
 913         entry = next_unused_user_queue++;
 914     else
 915     {
 916         LeaveCriticalSection(&queues_section);
 917         free(queue);
 918         WARN("Out of user queue handles.\n");
 919         return E_OUTOFMEMORY;
 920     }
 921
 922     entry->refcount = 1;
 923     entry->obj = queue;
 924     if (++queue_generation == 0xffff) queue_generation = 1;
 925     entry->generation = queue_generation;
 926     idx = entry - user_queues + FIRST_USER_QUEUE_HANDLE;
 927     *queue_id = (idx << 16) | entry->generation;
 928
 929     LeaveCriticalSection(&queues_section);
 930
 931     return S_OK;
 932 }
 933
 934 struct async_result
 935 {
 936     RTWQASYNCRESULT result;
 937     LONG refcount;
 938     IUnknown *object;
 939     IUnknown *state;
 940 };
 941
 942 static struct async_result *impl_from_IRtwqAsyncResult(IRtwqAsyncResult *iface)
 943 {
 944     return CONTAINING_RECORD(iface, struct async_result, result.AsyncResult);
 945 }
 946
 947 static HRESULT WINAPI async_result_QueryInterface(IRtwqAsyncResult *iface, REFIID riid, void **obj)
 948 {
 949     TRACE("%p, %s, %p.\n", iface, debugstr_guid(riid), obj);
 950
 951     if (IsEqualIID(riid, &IID_IRtwqAsyncResult) ||
 952             IsEqualIID(riid, &IID_IUnknown))
 953     {
 954         *obj = iface;
 955         IRtwqAsyncResult_AddRef(iface);
 956         return S_OK;
 957     }
 958
 959     *obj = NULL;
 960     WARN("Unsupported interface %s.\n", debugstr_guid(riid));
 961     return E_NOINTERFACE;
 962 }
 963
 964 static ULONG WINAPI async_result_AddRef(IRtwqAsyncResult *iface)
 965 {
 966     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
 967     ULONG refcount = InterlockedIncrement(&result->refcount);
 968
 969     TRACE("%p, %lu.\n", iface, refcount);
 970
 971     return refcount;
 972 }
 973
 974 static ULONG WINAPI async_result_Release(IRtwqAsyncResult *iface)
 975 {
 976     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
 977     ULONG refcount = InterlockedDecrement(&result->refcount);
 978
 979     TRACE("%p, %lu.\n", iface, refcount);
 980
 981     if (!refcount)
 982     {
 983         if (result->result.pCallback)
 984             IRtwqAsyncCallback_Release(result->result.pCallback);
 985         if (result->object)
 986             IUnknown_Release(result->object);
 987         if (result->state)
 988             IUnknown_Release(result->state);
 989         if (result->result.hEvent)
 990             CloseHandle(result->result.hEvent);
 991         free(result);
 992
 993         RtwqUnlockPlatform();
 994     }
 995
 996     return refcount;
 997 }
 998
 999 static HRESULT WINAPI async_result_GetState(IRtwqAsyncResult *iface, IUnknown **state)
1000 {
1001     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1002
1003     TRACE("%p, %p.\n", iface, state);
1004
1005     if (!result->state)
1006         return E_POINTER;
1007
1008     *state = result->state;
1009     IUnknown_AddRef(*state);
1010
1011     return S_OK;
1012 }
1013
1014 static HRESULT WINAPI async_result_GetStatus(IRtwqAsyncResult *iface)
1015 {
1016     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1017
1018     TRACE("%p.\n", iface);
1019
1020     return result->result.hrStatusResult;
1021 }
1022
1023 static HRESULT WINAPI async_result_SetStatus(IRtwqAsyncResult *iface, HRESULT status)
1024 {
1025     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1026
1027     TRACE("%p, %#lx.\n", iface, status);
1028
1029     result->result.hrStatusResult = status;
1030
1031     return S_OK;
1032 }
1033
1034 static HRESULT WINAPI async_result_GetObject(IRtwqAsyncResult *iface, IUnknown **object)
1035 {
1036     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1037
1038     TRACE("%p, %p.\n", iface, object);
1039
1040     if (!result->object)
1041         return E_POINTER;
1042
1043     *object = result->object;
1044     IUnknown_AddRef(*object);
1045
1046     return S_OK;
1047 }
1048
1049 static IUnknown * WINAPI async_result_GetStateNoAddRef(IRtwqAsyncResult *iface)
1050 {
1051     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1052
1053     TRACE("%p.\n", iface);
1054
1055     return result->state;
1056 }
1057
1058 static const IRtwqAsyncResultVtbl async_result_vtbl =
1059 {
1060     async_result_QueryInterface,
1061     async_result_AddRef,
1062     async_result_Release,
1063     async_result_GetState,
1064     async_result_GetStatus,
1065     async_result_SetStatus,
1066     async_result_GetObject,
1067     async_result_GetStateNoAddRef,
1068 };
1069
1070 static HRESULT create_async_result(IUnknown *object, IRtwqAsyncCallback *callback, IUnknown *state, IRtwqAsyncResult **out)
1071 {
1072     struct async_result *result;
1073
1074     if (!out)
1075         return E_INVALIDARG;
1076
1077     if (!(result = calloc(1, sizeof(*result))))
1078         return E_OUTOFMEMORY;
1079
1080     RtwqLockPlatform();
1081
1082     result->result.AsyncResult.lpVtbl = &async_result_vtbl;
1083     result->refcount = 1;
1084     result->object = object;
1085     if (result->object)
1086         IUnknown_AddRef(result->object);
1087     result->result.pCallback = callback;
1088     if (result->result.pCallback)
1089         IRtwqAsyncCallback_AddRef(result->result.pCallback);
1090     result->state = state;
1091     if (result->state)
1092         IUnknown_AddRef(result->state);
1093
1094     *out = &result->result.AsyncResult;
1095
1096     TRACE("Created async result object %p.\n", *out);
1097
1098     return S_OK;
1099 }
1100
1101 HRESULT WINAPI RtwqCreateAsyncResult(IUnknown *object, IRtwqAsyncCallback *callback, IUnknown *state,
1102         IRtwqAsyncResult **out)
1103 {
1104     TRACE("%p, %p, %p, %p.\n", object, callback, state, out);
1105
1106     return create_async_result(object, callback, state, out);
1107 }
1108
1109 HRESULT WINAPI RtwqLockPlatform(void)
1110 {
1111     InterlockedIncrement(&platform_lock);
1112
1113     return S_OK;
1114 }
1115
1116 HRESULT WINAPI RtwqUnlockPlatform(void)
1117 {
1118     InterlockedDecrement(&platform_lock);
1119
1120     return S_OK;
1121 }
1122
1123 static void init_system_queues(void)
1124 {
1125     struct queue_desc desc;
1126     HRESULT hr;
1127
1128     /* Always initialize standard queue, keep the rest lazy. */
1129
1130     EnterCriticalSection(&queues_section);
1131
1132     if (system_queues[SYS_QUEUE_STANDARD].pool)
1133     {
1134         LeaveCriticalSection(&queues_section);
1135         return;
1136     }
1137
1138     if (FAILED(hr = CoIncrementMTAUsage(&mta_cookie)))
1139         WARN("Failed to initialize MTA, hr %#lx.\n", hr);
1140
1141     desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
1142     desc.ops = &pool_queue_ops;
1143     desc.target_queue = 0;
1144     init_work_queue(&desc, &system_queues[SYS_QUEUE_STANDARD]);
1145
1146     LeaveCriticalSection(&queues_section);
1147 }
1148
1149 HRESULT WINAPI RtwqStartup(void)
1150 {
1151     if (InterlockedIncrement(&platform_lock) == 1)
1152     {
1153         init_system_queues();
1154     }
1155
1156     return S_OK;
1157 }
1158
1159 static void shutdown_system_queues(void)
1160 {
1161     unsigned int i;
1162     HRESULT hr;
1163
1164     EnterCriticalSection(&queues_section);
1165
1166     for (i = 0; i < ARRAY_SIZE(system_queues); ++i)
1167     {
1168         shutdown_queue(&system_queues[i]);
1169     }
1170
1171     if (FAILED(hr = CoDecrementMTAUsage(mta_cookie)))
1172         WARN("Failed to uninitialize MTA, hr %#lx.\n", hr);
1173
1174     LeaveCriticalSection(&queues_section);
1175 }
1176
1177 HRESULT WINAPI RtwqShutdown(void)
1178 {
1179     if (platform_lock <= 0)
1180         return S_OK;
1181
1182     if (InterlockedExchangeAdd(&platform_lock, -1) == 1)
1183     {
1184         shutdown_system_queues();
1185     }
1186
1187     return S_OK;
1188 }
1189
1190 HRESULT WINAPI RtwqPutWaitingWorkItem(HANDLE event, LONG priority, IRtwqAsyncResult *result, RTWQWORKITEM_KEY *key)
1191 {
1192     struct queue *queue;
1193     HRESULT hr;
1194
1195     TRACE("%p, %ld, %p, %p.\n", event, priority, result, key);
1196
1197     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1198         return hr;
1199
1200     hr = queue_submit_wait(queue, event, priority, result, key);
1201
1202     return hr;
1203 }
1204
1205 static HRESULT schedule_work_item(IRtwqAsyncResult *result, INT64 timeout, RTWQWORKITEM_KEY *key)
1206 {
1207     struct queue *queue;
1208     HRESULT hr;
1209
1210     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1211         return hr;
1212
1213     TRACE("%p, %s, %p.\n", result, wine_dbgstr_longlong(timeout), key);
1214
1215     return queue_submit_timer(queue, result, timeout, 0, key);
1216 }
1217
1218 HRESULT WINAPI RtwqScheduleWorkItem(IRtwqAsyncResult *result, INT64 timeout, RTWQWORKITEM_KEY *key)
1219 {
1220     TRACE("%p, %s, %p.\n", result, wine_dbgstr_longlong(timeout), key);
1221
1222     return schedule_work_item(result, timeout, key);
1223 }
1224
1225 struct periodic_callback
1226 {
1227     IRtwqAsyncCallback IRtwqAsyncCallback_iface;
1228     LONG refcount;
1229     RTWQPERIODICCALLBACK callback;
1230 };
1231
1232 static struct periodic_callback *impl_from_IRtwqAsyncCallback(IRtwqAsyncCallback *iface)
1233 {
1234     return CONTAINING_RECORD(iface, struct periodic_callback, IRtwqAsyncCallback_iface);
1235 }
1236
1237 static HRESULT WINAPI periodic_callback_QueryInterface(IRtwqAsyncCallback *iface, REFIID riid, void **obj)
1238 {
1239     if (IsEqualIID(riid, &IID_IRtwqAsyncCallback) ||
1240             IsEqualIID(riid, &IID_IUnknown))
1241     {
1242         *obj = iface;
1243         IRtwqAsyncCallback_AddRef(iface);
1244         return S_OK;
1245     }
1246
1247     *obj = NULL;
1248     return E_NOINTERFACE;
1249 }
1250
1251 static ULONG WINAPI periodic_callback_AddRef(IRtwqAsyncCallback *iface)
1252 {
1253     struct periodic_callback *callback = impl_from_IRtwqAsyncCallback(iface);
1254     ULONG refcount = InterlockedIncrement(&callback->refcount);
1255
1256     TRACE("%p, %lu.\n", iface, refcount);
1257
1258     return refcount;
1259 }
1260
1261 static ULONG WINAPI periodic_callback_Release(IRtwqAsyncCallback *iface)
1262 {
1263     struct periodic_callback *callback = impl_from_IRtwqAsyncCallback(iface);
1264     ULONG refcount = InterlockedDecrement(&callback->refcount);
1265
1266     TRACE("%p, %lu.\n", iface, refcount);
1267
1268     if (!refcount)
1269         free(callback);
1270
1271     return refcount;
1272 }
1273
1274 static HRESULT WINAPI periodic_callback_GetParameters(IRtwqAsyncCallback *iface, DWORD *flags, DWORD *queue)
1275 {
1276     return E_NOTIMPL;
1277 }
1278
1279 static HRESULT WINAPI periodic_callback_Invoke(IRtwqAsyncCallback *iface, IRtwqAsyncResult *result)
1280 {
1281     struct periodic_callback *callback = impl_from_IRtwqAsyncCallback(iface);
1282     IUnknown *context = NULL;
1283
1284     if (FAILED(IRtwqAsyncResult_GetObject(result, &context)))
1285         WARN("Expected object to be set for result object.\n");
1286
1287     callback->callback(context);
1288
1289     if (context)
1290         IUnknown_Release(context);
1291
1292     return S_OK;
1293 }
1294
1295 static const IRtwqAsyncCallbackVtbl periodic_callback_vtbl =
1296 {
1297     periodic_callback_QueryInterface,
1298     periodic_callback_AddRef,
1299     periodic_callback_Release,
1300     periodic_callback_GetParameters,
1301     periodic_callback_Invoke,
1302 };
1303
1304 static HRESULT create_periodic_callback_obj(RTWQPERIODICCALLBACK callback, IRtwqAsyncCallback **out)
1305 {
1306     struct periodic_callback *object;
1307
1308     if (!(object = calloc(1, sizeof(*object))))
1309         return E_OUTOFMEMORY;
1310
1311     object->IRtwqAsyncCallback_iface.lpVtbl = &periodic_callback_vtbl;
1312     object->refcount = 1;
1313     object->callback = callback;
1314
1315     *out = &object->IRtwqAsyncCallback_iface;
1316
1317     return S_OK;
1318 }
1319
1320 HRESULT WINAPI RtwqAddPeriodicCallback(RTWQPERIODICCALLBACK callback, IUnknown *context, DWORD *key)
1321 {
1322     IRtwqAsyncCallback *periodic_callback;
1323     RTWQWORKITEM_KEY workitem_key;
1324     IRtwqAsyncResult *result;
1325     struct queue *queue;
1326     HRESULT hr;
1327
1328     TRACE("%p, %p, %p.\n", callback, context, key);
1329
1330     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1331         return hr;
1332
1333     if (FAILED(hr = create_periodic_callback_obj(callback, &periodic_callback)))
1334         return hr;
1335
1336     hr = create_async_result(context, periodic_callback, NULL, &result);
1337     IRtwqAsyncCallback_Release(periodic_callback);
1338     if (FAILED(hr))
1339         return hr;
1340
1341     /* Same period MFGetTimerPeriodicity() returns. */
1342     hr = queue_submit_timer(queue, result, 0, 10, key ? &workitem_key : NULL);
1343
1344     IRtwqAsyncResult_Release(result);
1345
1346     if (key)
1347         *key = workitem_key;
1348
1349     return S_OK;
1350 }
1351
1352 HRESULT WINAPI RtwqRemovePeriodicCallback(DWORD key)
1353 {
1354     struct queue *queue;
1355     HRESULT hr;
1356
1357     TRACE("%#lx.\n", key);
1358
1359     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1360         return hr;
1361
1362     return queue_cancel_item(queue, get_item_key(SCHEDULED_ITEM_KEY_MASK, key));
1363 }
1364
1365 HRESULT WINAPI RtwqCancelWorkItem(RTWQWORKITEM_KEY key)
1366 {
1367     struct queue *queue;
1368     HRESULT hr;
1369
1370     TRACE("%s.\n", wine_dbgstr_longlong(key));
1371
1372     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1373         return hr;
1374
1375     return queue_cancel_item(queue, key);
1376 }
1377
1378 HRESULT WINAPI RtwqInvokeCallback(IRtwqAsyncResult *result)
1379 {
1380     TRACE("%p.\n", result);
1381
1382     return invoke_async_callback(result);
1383 }
1384
1385 HRESULT WINAPI RtwqPutWorkItem(DWORD queue, LONG priority, IRtwqAsyncResult *result)
1386 {
1387     TRACE("%#lx, %ld, %p.\n", queue, priority, result);
1388
1389     return queue_put_work_item(queue, priority, result);
1390 }
1391
1392 HRESULT WINAPI RtwqAllocateWorkQueue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queue)
1393 {
1394     struct queue_desc desc;
1395
1396     TRACE("%d, %p.\n", queue_type, queue);
1397
1398     desc.queue_type = queue_type;
1399     desc.ops = &pool_queue_ops;
1400     desc.target_queue = 0;
1401     return alloc_user_queue(&desc, queue);
1402 }
1403
1404 HRESULT WINAPI RtwqLockWorkQueue(DWORD queue)
1405 {
1406     TRACE("%#lx.\n", queue);
1407
1408     return lock_user_queue(queue);
1409 }
1410
1411 HRESULT WINAPI RtwqUnlockWorkQueue(DWORD queue)
1412 {
1413     TRACE("%#lx.\n", queue);
1414
1415     return unlock_user_queue(queue);
1416 }
1417
1418 HRESULT WINAPI RtwqSetLongRunning(DWORD queue_id, BOOL enable)
1419 {
1420     struct queue *queue;
1421     HRESULT hr;
1422     int i;
1423
1424     TRACE("%#lx, %d.\n", queue_id, enable);
1425
1426     lock_user_queue(queue_id);
1427
1428     if (SUCCEEDED(hr = grab_queue(queue_id, &queue)))
1429     {
1430         for (i = 0; i < ARRAY_SIZE(queue->envs); ++i)
1431             queue->envs[i].u.s.LongFunction = !!enable;
1432     }
1433
1434     unlock_user_queue(queue_id);
1435
1436     return hr;
1437 }
1438
1439 HRESULT WINAPI RtwqLockSharedWorkQueue(const WCHAR *usageclass, LONG priority, DWORD *taskid, DWORD *queue)
1440 {
1441     struct queue_desc desc;
1442     HRESULT hr;
1443
1444     TRACE("%s, %ld, %p, %p.\n", debugstr_w(usageclass), priority, taskid, queue);
1445
1446     if (!usageclass)
1447         return E_POINTER;
1448
1449     if (!*usageclass && taskid)
1450         return E_INVALIDARG;
1451
1452     if (*usageclass)
1453         FIXME("Class name is ignored.\n");
1454
1455     EnterCriticalSection(&queues_section);
1456
1457     if (shared_mt_queue)
1458         hr = lock_user_queue(shared_mt_queue);
1459     else
1460     {
1461         desc.queue_type = RTWQ_MULTITHREADED_WORKQUEUE;
1462         desc.ops = &pool_queue_ops;
1463         desc.target_queue = 0;
1464         hr = alloc_user_queue(&desc, &shared_mt_queue);
1465     }
1466
1467     *queue = shared_mt_queue;
1468
1469     LeaveCriticalSection(&queues_section);
1470
1471     return hr;
1472 }
1473
1474 HRESULT WINAPI RtwqSetDeadline(DWORD queue_id, LONGLONG deadline, HANDLE *request)
1475 {
1476     FIXME("%#lx, %s, %p.\n", queue_id, wine_dbgstr_longlong(deadline), request);
1477
1478     return E_NOTIMPL;
1479 }
1480
1481 HRESULT WINAPI RtwqSetDeadline2(DWORD queue_id, LONGLONG deadline, LONGLONG predeadline, HANDLE *request)
1482 {
1483     FIXME("%#lx, %s, %s, %p.\n", queue_id, wine_dbgstr_longlong(deadline), wine_dbgstr_longlong(predeadline), request);
1484
1485     return E_NOTIMPL;
1486 }
1487
1488 HRESULT WINAPI RtwqCancelDeadline(HANDLE request)
1489 {
1490     FIXME("%p.\n", request);
1491
1492     return E_NOTIMPL;
1493 }
1494
1495 HRESULT WINAPI RtwqAllocateSerialWorkQueue(DWORD target_queue, DWORD *queue)
1496 {
1497     struct queue_desc desc;
1498
1499     TRACE("%#lx, %p.\n", target_queue, queue);
1500
1501     desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
1502     desc.ops = &serial_queue_ops;
1503     desc.target_queue = target_queue;
1504     return alloc_user_queue(&desc, queue);
1505 }
1506
1507 HRESULT WINAPI RtwqJoinWorkQueue(DWORD queue, HANDLE hFile, HANDLE *cookie)
1508 {
1509     FIXME("%#lx, %p, %p.\n", queue, hFile, cookie);
1510
1511     return E_NOTIMPL;
1512 }
1513
1514 HRESULT WINAPI RtwqUnjoinWorkQueue(DWORD queue, HANDLE cookie)
1515 {
1516     FIXME("%#lx, %p.\n", queue, cookie);
1517
1518     return E_NOTIMPL;
1519 }
1520
1521 HRESULT WINAPI RtwqGetWorkQueueMMCSSClass(DWORD queue, WCHAR *class, DWORD *length)
1522 {
1523     FIXME("%#lx, %p, %p.\n", queue, class, length);
1524
1525     return E_NOTIMPL;
1526 }
1527
1528 HRESULT WINAPI RtwqGetWorkQueueMMCSSTaskId(DWORD queue, DWORD *taskid)
1529 {
1530     FIXME("%#lx, %p.\n", queue, taskid);
1531
1532     return E_NOTIMPL;
1533 }
1534
1535 HRESULT WINAPI RtwqGetWorkQueueMMCSSPriority(DWORD queue, LONG *priority)
1536 {
1537     FIXME("%#lx, %p.\n", queue, priority);
1538
1539     return E_NOTIMPL;
1540 }
1541
1542 HRESULT WINAPI RtwqRegisterPlatformWithMMCSS(const WCHAR *class, DWORD *taskid, LONG priority)
1543 {
1544     FIXME("%s, %p, %ld.\n", debugstr_w(class), taskid, priority);
1545
1546     return E_NOTIMPL;
1547 }
1548
1549 HRESULT WINAPI RtwqUnregisterPlatformFromMMCSS(void)
1550 {
1551     FIXME("\n");
1552
1553     return E_NOTIMPL;
1554 }
1555
1556 HRESULT WINAPI RtwqBeginRegisterWorkQueueWithMMCSS(DWORD queue, const WCHAR *class, DWORD taskid, LONG priority,
1557         IRtwqAsyncCallback *callback, IUnknown *state)
1558 {
1559     FIXME("%#lx, %s, %lu, %ld, %p, %p.\n", queue, debugstr_w(class), taskid, priority, callback, state);
1560
1561     return E_NOTIMPL;
1562 }
1563
1564 HRESULT WINAPI RtwqEndRegisterWorkQueueWithMMCSS(IRtwqAsyncResult *result, DWORD *taskid)
1565 {
1566     FIXME("%p, %p.\n", result, taskid);
1567
1568     return E_NOTIMPL;
1569 }
1570
1571 HRESULT WINAPI RtwqBeginUnregisterWorkQueueWithMMCSS(DWORD queue, IRtwqAsyncCallback *callback, IUnknown *state)
1572 {
1573     FIXME("%#lx, %p, %p.\n", queue, callback, state);
1574
1575     return E_NOTIMPL;
1576 }
1577
1578 HRESULT WINAPI RtwqEndUnregisterWorkQueueWithMMCSS(IRtwqAsyncResult *result)
1579 {
1580     FIXME("%p.\n", result);
1581
1582     return E_NOTIMPL;
1583 }
1584
1585 HRESULT WINAPI RtwqRegisterPlatformEvents(IRtwqPlatformEvents *events)
1586 {
1587     FIXME("%p.\n", events);
1588
1589     return E_NOTIMPL;
1590 }
1591
1592 HRESULT WINAPI RtwqUnregisterPlatformEvents(IRtwqPlatformEvents *events)
1593 {
1594     FIXME("%p.\n", events);
1595
1596     return E_NOTIMPL;
1597 }