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     PTP_SIMPLE_CALLBACK finalization_callback;
 135     union
 136     {
 137         TP_WAIT *wait_object;
 138         TP_TIMER *timer_object;
 139     } u;
 140 };
 141
 142 static struct work_item *work_item_impl_from_IUnknown(IUnknown *iface)
 143 {
 144     return CONTAINING_RECORD(iface, struct work_item, IUnknown_iface);
 145 }
 146
 147 static const TP_CALLBACK_PRIORITY priorities[] =
 148 {
 149     TP_CALLBACK_PRIORITY_HIGH,
 150     TP_CALLBACK_PRIORITY_NORMAL,
 151     TP_CALLBACK_PRIORITY_LOW,
 152 };
 153
 154 struct queue;
 155 struct queue_desc;
 156
 157 struct queue_ops
 158 {
 159     HRESULT (*init)(const struct queue_desc *desc, struct queue *queue);
 160     BOOL (*shutdown)(struct queue *queue);
 161     void (*submit)(struct queue *queue, struct work_item *item);
 162 };
 163
 164 struct queue_desc
 165 {
 166     RTWQ_WORKQUEUE_TYPE queue_type;
 167     const struct queue_ops *ops;
 168     DWORD target_queue;
 169 };
 170
 171 struct queue
 172 {
 173     IRtwqAsyncCallback IRtwqAsyncCallback_iface;
 174     const struct queue_ops *ops;
 175     TP_POOL *pool;
 176     TP_CALLBACK_ENVIRON_V3 envs[ARRAY_SIZE(priorities)];
 177     CRITICAL_SECTION cs;
 178     struct list pending_items;
 179     DWORD id;
 180     /* Data used for serial queues only. */
 181     PTP_SIMPLE_CALLBACK finalization_callback;
 182     DWORD target_queue;
 183 };
 184
 185 static void shutdown_queue(struct queue *queue);
 186
 187 static HRESULT lock_user_queue(DWORD queue)
 188 {
 189     HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
 190     struct queue_handle *entry;
 191
 192     if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
 193         return S_OK;
 194
 195     EnterCriticalSection(&queues_section);
 196     entry = get_queue_obj(queue);
 197     if (entry && entry->refcount)
 198     {
 199         entry->refcount++;
 200         hr = S_OK;
 201     }
 202     LeaveCriticalSection(&queues_section);
 203     return hr;
 204 }
 205
 206 static HRESULT unlock_user_queue(DWORD queue)
 207 {
 208     HRESULT hr = RTWQ_E_INVALID_WORKQUEUE;
 209     struct queue_handle *entry;
 210
 211     if (!(queue & RTWQ_CALLBACK_QUEUE_PRIVATE_MASK))
 212         return S_OK;
 213
 214     EnterCriticalSection(&queues_section);
 215     entry = get_queue_obj(queue);
 216     if (entry && entry->refcount)
 217     {
 218         if (--entry->refcount == 0)
 219         {
 220             if (shared_mt_queue == queue) shared_mt_queue = 0;
 221             shutdown_queue((struct queue *)entry->obj);
 222             free(entry->obj);
 223             entry->obj = next_free_user_queue;
 224             next_free_user_queue = entry;
 225         }
 226         hr = S_OK;
 227     }
 228     LeaveCriticalSection(&queues_section);
 229     return hr;
 230 }
 231
 232 static struct queue *queue_impl_from_IRtwqAsyncCallback(IRtwqAsyncCallback *iface)
 233 {
 234     return CONTAINING_RECORD(iface, struct queue, IRtwqAsyncCallback_iface);
 235 }
 236
 237 static HRESULT WINAPI queue_serial_callback_QueryInterface(IRtwqAsyncCallback *iface, REFIID riid, void **obj)
 238 {
 239     if (IsEqualIID(riid, &IID_IRtwqAsyncCallback) ||
 240             IsEqualIID(riid, &IID_IUnknown))
 241     {
 242         *obj = iface;
 243         IRtwqAsyncCallback_AddRef(iface);
 244         return S_OK;
 245     }
 246
 247     *obj = NULL;
 248     return E_NOINTERFACE;
 249 }
 250
 251 static ULONG WINAPI queue_serial_callback_AddRef(IRtwqAsyncCallback *iface)
 252 {
 253     return 2;
 254 }
 255
 256 static ULONG WINAPI queue_serial_callback_Release(IRtwqAsyncCallback *iface)
 257 {
 258     return 1;
 259 }
 260
 261 static HRESULT WINAPI queue_serial_callback_GetParameters(IRtwqAsyncCallback *iface, DWORD *flags, DWORD *queue_id)
 262 {
 263     struct queue *queue = queue_impl_from_IRtwqAsyncCallback(iface);
 264
 265     *flags = 0;
 266     *queue_id = queue->id;
 267
 268     return S_OK;
 269 }
 270
 271 static HRESULT WINAPI queue_serial_callback_Invoke(IRtwqAsyncCallback *iface, IRtwqAsyncResult *result)
 272 {
 273     /* Reply callback won't be called in a regular way, pending items and chained queues will make it
 274        unnecessary complicated to reach actual work queue that's able to execute this item. Instead
 275        serial queues are cleaned up right away on submit(). */
 276     return S_OK;
 277 }
 278
 279 static const IRtwqAsyncCallbackVtbl queue_serial_callback_vtbl =
 280 {
 281     queue_serial_callback_QueryInterface,
 282     queue_serial_callback_AddRef,
 283     queue_serial_callback_Release,
 284     queue_serial_callback_GetParameters,
 285     queue_serial_callback_Invoke,
 286 };
 287
 288 static struct queue system_queues[SYS_QUEUE_COUNT];
 289
 290 static struct queue *get_system_queue(DWORD queue_id)
 291 {
 292     switch (queue_id)
 293     {
 294         case RTWQ_CALLBACK_QUEUE_STANDARD:
 295         case RTWQ_CALLBACK_QUEUE_RT:
 296         case RTWQ_CALLBACK_QUEUE_IO:
 297         case RTWQ_CALLBACK_QUEUE_TIMER:
 298         case RTWQ_CALLBACK_QUEUE_MULTITHREADED:
 299         case RTWQ_CALLBACK_QUEUE_LONG_FUNCTION:
 300             return &system_queues[queue_id - 1];
 301         default:
 302             return NULL;
 303     }
 304 }
 305
 306 static HRESULT grab_queue(DWORD queue_id, struct queue **ret);
 307
 308 static void CALLBACK standard_queue_cleanup_callback(void *object_data, void *group_data)
 309 {
 310 }
 311
 312 static HRESULT pool_queue_init(const struct queue_desc *desc, struct queue *queue)
 313 {
 314     TP_CALLBACK_ENVIRON_V3 env;
 315     unsigned int max_thread, i;
 316
 317     queue->pool = CreateThreadpool(NULL);
 318
 319     memset(&env, 0, sizeof(env));
 320     env.Version = 3;
 321     env.Size = sizeof(env);
 322     env.Pool = queue->pool;
 323     env.CleanupGroup = CreateThreadpoolCleanupGroup();
 324     env.CleanupGroupCancelCallback = standard_queue_cleanup_callback;
 325     env.CallbackPriority = TP_CALLBACK_PRIORITY_NORMAL;
 326     for (i = 0; i < ARRAY_SIZE(queue->envs); ++i)
 327     {
 328         queue->envs[i] = env;
 329         queue->envs[i].CallbackPriority = priorities[i];
 330     }
 331     list_init(&queue->pending_items);
 332     InitializeCriticalSection(&queue->cs);
 333
 334     max_thread = (desc->queue_type == RTWQ_STANDARD_WORKQUEUE || desc->queue_type == RTWQ_WINDOW_WORKQUEUE) ? 1 : 4;
 335
 336     SetThreadpoolThreadMinimum(queue->pool, 1);
 337     SetThreadpoolThreadMaximum(queue->pool, max_thread);
 338
 339     if (desc->queue_type == RTWQ_WINDOW_WORKQUEUE)
 340         FIXME("RTWQ_WINDOW_WORKQUEUE is not supported.\n");
 341
 342     return S_OK;
 343 }
 344
 345 static BOOL pool_queue_shutdown(struct queue *queue)
 346 {
 347     if (!queue->pool)
 348         return FALSE;
 349
 350     CloseThreadpoolCleanupGroupMembers(queue->envs[0].CleanupGroup, TRUE, NULL);
 351     CloseThreadpool(queue->pool);
 352     queue->pool = NULL;
 353
 354     return TRUE;
 355 }
 356
 357 static void CALLBACK standard_queue_worker(TP_CALLBACK_INSTANCE *instance, void *context, TP_WORK *work)
 358 {
 359     struct work_item *item = context;
 360     RTWQASYNCRESULT *result = (RTWQASYNCRESULT *)item->result;
 361
 362     TRACE("result object %p.\n", result);
 363
 364     /* Submitting from serial queue in reply mode, use different result object acting as receipt token.
 365        It's submitted to user callback still, but when invoked, special serial queue callback will be used
 366        to ensure correct destination queue. */
 367
 368     IRtwqAsyncCallback_Invoke(result->pCallback, item->reply_result ? item->reply_result : item->result);
 369
 370     IUnknown_Release(&item->IUnknown_iface);
 371 }
 372
 373 static void pool_queue_submit(struct queue *queue, struct work_item *item)
 374 {
 375     TP_CALLBACK_PRIORITY callback_priority;
 376     TP_CALLBACK_ENVIRON_V3 env;
 377     TP_WORK *work_object;
 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     work_object = CreateThreadpoolWork(standard_queue_worker, item, (TP_CALLBACK_ENVIRON *)&env);
 393     SubmitThreadpoolWork(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->reply_result)
 555             IRtwqAsyncResult_Release(item->reply_result);
 556         IRtwqAsyncResult_Release(item->result);
 557         free(item);
 558     }
 559
 560     return refcount;
 561 }
 562
 563 static const IUnknownVtbl work_item_vtbl =
 564 {
 565     work_item_QueryInterface,
 566     work_item_AddRef,
 567     work_item_Release,
 568 };
 569
 570 static struct work_item * alloc_work_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 571 {
 572     RTWQASYNCRESULT *async_result = (RTWQASYNCRESULT *)result;
 573     DWORD flags = 0, queue_id = 0;
 574     struct work_item *item;
 575
 576     item = calloc(1, sizeof(*item));
 577
 578     item->IUnknown_iface.lpVtbl = &work_item_vtbl;
 579     item->result = result;
 580     IRtwqAsyncResult_AddRef(item->result);
 581     item->refcount = 1;
 582     item->queue = queue;
 583     list_init(&item->entry);
 584     item->priority = priority;
 585
 586     if (SUCCEEDED(IRtwqAsyncCallback_GetParameters(async_result->pCallback, &flags, &queue_id)))
 587         item->flags = flags;
 588
 589     return item;
 590 }
 591
 592 static void init_work_queue(const struct queue_desc *desc, struct queue *queue)
 593 {
 594     assert(desc->ops != NULL);
 595
 596     queue->ops = desc->ops;
 597     if (SUCCEEDED(queue->ops->init(desc, queue)))
 598     {
 599         list_init(&queue->pending_items);
 600         InitializeCriticalSection(&queue->cs);
 601     }
 602 }
 603
 604 static HRESULT grab_queue(DWORD queue_id, struct queue **ret)
 605 {
 606     struct queue *queue = get_system_queue(queue_id);
 607     RTWQ_WORKQUEUE_TYPE queue_type;
 608     struct queue_handle *entry;
 609
 610     *ret = NULL;
 611
 612     if (!system_queues[SYS_QUEUE_STANDARD].pool)
 613         return RTWQ_E_SHUTDOWN;
 614
 615     if (queue && queue->pool)
 616     {
 617         *ret = queue;
 618         return S_OK;
 619     }
 620     else if (queue)
 621     {
 622         struct queue_desc desc;
 623
 624         EnterCriticalSection(&queues_section);
 625         switch (queue_id)
 626         {
 627             case RTWQ_CALLBACK_QUEUE_IO:
 628             case RTWQ_CALLBACK_QUEUE_MULTITHREADED:
 629             case RTWQ_CALLBACK_QUEUE_LONG_FUNCTION:
 630                 queue_type = RTWQ_MULTITHREADED_WORKQUEUE;
 631                 break;
 632             default:
 633                 queue_type = RTWQ_STANDARD_WORKQUEUE;
 634         }
 635
 636         desc.queue_type = queue_type;
 637         desc.ops = &pool_queue_ops;
 638         desc.target_queue = 0;
 639         init_work_queue(&desc, queue);
 640         LeaveCriticalSection(&queues_section);
 641         *ret = queue;
 642         return S_OK;
 643     }
 644
 645     /* Handles user queues. */
 646     if ((entry = get_queue_obj(queue_id)))
 647         *ret = entry->obj;
 648
 649     return *ret ? S_OK : RTWQ_E_INVALID_WORKQUEUE;
 650 }
 651
 652 static void shutdown_queue(struct queue *queue)
 653 {
 654     struct work_item *item, *item2;
 655
 656     if (!queue->ops || !queue->ops->shutdown(queue))
 657         return;
 658
 659     EnterCriticalSection(&queue->cs);
 660     LIST_FOR_EACH_ENTRY_SAFE(item, item2, &queue->pending_items, struct work_item, entry)
 661     {
 662         list_remove(&item->entry);
 663         IUnknown_Release(&item->IUnknown_iface);
 664     }
 665     LeaveCriticalSection(&queue->cs);
 666
 667     DeleteCriticalSection(&queue->cs);
 668
 669     memset(queue, 0, sizeof(*queue));
 670 }
 671
 672 static HRESULT queue_submit_item(struct queue *queue, LONG priority, IRtwqAsyncResult *result)
 673 {
 674     struct work_item *item;
 675
 676     if (!(item = alloc_work_item(queue, priority, result)))
 677         return E_OUTOFMEMORY;
 678
 679     queue->ops->submit(queue, item);
 680
 681     return S_OK;
 682 }
 683
 684 static HRESULT queue_put_work_item(DWORD queue_id, LONG priority, IRtwqAsyncResult *result)
 685 {
 686     struct queue *queue;
 687     HRESULT hr;
 688
 689     if (FAILED(hr = grab_queue(queue_id, &queue)))
 690         return hr;
 691
 692     return queue_submit_item(queue, priority, result);
 693 }
 694
 695 static HRESULT invoke_async_callback(IRtwqAsyncResult *result)
 696 {
 697     RTWQASYNCRESULT *result_data = (RTWQASYNCRESULT *)result;
 698     DWORD queue = RTWQ_CALLBACK_QUEUE_STANDARD, flags;
 699     HRESULT hr;
 700
 701     if (FAILED(IRtwqAsyncCallback_GetParameters(result_data->pCallback, &flags, &queue)))
 702         queue = RTWQ_CALLBACK_QUEUE_STANDARD;
 703
 704     if (FAILED(lock_user_queue(queue)))
 705         queue = RTWQ_CALLBACK_QUEUE_STANDARD;
 706
 707     hr = queue_put_work_item(queue, 0, result);
 708
 709     unlock_user_queue(queue);
 710
 711     return hr;
 712 }
 713
 714 static void queue_release_pending_item(struct work_item *item)
 715 {
 716     EnterCriticalSection(&item->queue->cs);
 717     if (item->key)
 718     {
 719         list_remove(&item->entry);
 720         item->key = 0;
 721         IUnknown_Release(&item->IUnknown_iface);
 722     }
 723     LeaveCriticalSection(&item->queue->cs);
 724 }
 725
 726 static void CALLBACK waiting_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_WAIT *wait,
 727         TP_WAIT_RESULT wait_result)
 728 {
 729     struct work_item *item = context;
 730
 731     TRACE("result object %p.\n", item->result);
 732
 733     invoke_async_callback(item->result);
 734
 735     IUnknown_Release(&item->IUnknown_iface);
 736 }
 737
 738 static void CALLBACK waiting_item_cancelable_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_WAIT *wait,
 739         TP_WAIT_RESULT wait_result)
 740 {
 741     struct work_item *item = context;
 742
 743     TRACE("result object %p.\n", item->result);
 744
 745     queue_release_pending_item(item);
 746
 747     invoke_async_callback(item->result);
 748
 749     IUnknown_Release(&item->IUnknown_iface);
 750 }
 751
 752 static void CALLBACK scheduled_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 753 {
 754     struct work_item *item = context;
 755
 756     TRACE("result object %p.\n", item->result);
 757
 758     invoke_async_callback(item->result);
 759
 760     IUnknown_Release(&item->IUnknown_iface);
 761 }
 762
 763 static void CALLBACK scheduled_item_cancelable_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 764 {
 765     struct work_item *item = context;
 766
 767     TRACE("result object %p.\n", item->result);
 768
 769     queue_release_pending_item(item);
 770
 771     invoke_async_callback(item->result);
 772
 773     IUnknown_Release(&item->IUnknown_iface);
 774 }
 775
 776 static void CALLBACK periodic_item_callback(TP_CALLBACK_INSTANCE *instance, void *context, TP_TIMER *timer)
 777 {
 778     struct work_item *item = context;
 779
 780     IUnknown_AddRef(&item->IUnknown_iface);
 781
 782     invoke_async_callback(item->result);
 783
 784     IUnknown_Release(&item->IUnknown_iface);
 785 }
 786
 787 static void queue_mark_item_pending(DWORD mask, struct work_item *item, RTWQWORKITEM_KEY *key)
 788 {
 789     *key = generate_item_key(mask);
 790     item->key = *key;
 791
 792     EnterCriticalSection(&item->queue->cs);
 793     list_add_tail(&item->queue->pending_items, &item->entry);
 794     IUnknown_AddRef(&item->IUnknown_iface);
 795     LeaveCriticalSection(&item->queue->cs);
 796 }
 797
 798 static HRESULT queue_submit_wait(struct queue *queue, HANDLE event, LONG priority, IRtwqAsyncResult *result,
 799         RTWQWORKITEM_KEY *key)
 800 {
 801     PTP_WAIT_CALLBACK callback;
 802     struct work_item *item;
 803
 804     if (!(item = alloc_work_item(queue, priority, result)))
 805         return E_OUTOFMEMORY;
 806
 807     if (key)
 808     {
 809         queue_mark_item_pending(WAIT_ITEM_KEY_MASK, item, key);
 810         callback = waiting_item_cancelable_callback;
 811     }
 812     else
 813         callback = waiting_item_callback;
 814
 815     item->u.wait_object = CreateThreadpoolWait(callback, item,
 816             (TP_CALLBACK_ENVIRON *)&queue->envs[TP_CALLBACK_PRIORITY_NORMAL]);
 817     SetThreadpoolWait(item->u.wait_object, event, NULL);
 818
 819     TRACE("dispatched %p.\n", result);
 820
 821     return S_OK;
 822 }
 823
 824 static HRESULT queue_submit_timer(struct queue *queue, IRtwqAsyncResult *result, INT64 timeout, DWORD period,
 825         RTWQWORKITEM_KEY *key)
 826 {
 827     PTP_TIMER_CALLBACK callback;
 828     struct work_item *item;
 829     FILETIME filetime;
 830     LARGE_INTEGER t;
 831
 832     if (!(item = alloc_work_item(queue, 0, result)))
 833         return E_OUTOFMEMORY;
 834
 835     if (key)
 836     {
 837         queue_mark_item_pending(SCHEDULED_ITEM_KEY_MASK, item, key);
 838     }
 839
 840     if (period)
 841         callback = periodic_item_callback;
 842     else
 843         callback = key ? scheduled_item_cancelable_callback : scheduled_item_callback;
 844
 845     t.QuadPart = timeout * 1000 * 10;
 846     filetime.dwLowDateTime = t.u.LowPart;
 847     filetime.dwHighDateTime = t.u.HighPart;
 848
 849     item->u.timer_object = CreateThreadpoolTimer(callback, item,
 850             (TP_CALLBACK_ENVIRON *)&queue->envs[TP_CALLBACK_PRIORITY_NORMAL]);
 851     SetThreadpoolTimer(item->u.timer_object, &filetime, period, 0);
 852
 853     TRACE("dispatched %p.\n", result);
 854
 855     return S_OK;
 856 }
 857
 858 static HRESULT queue_cancel_item(struct queue *queue, RTWQWORKITEM_KEY key)
 859 {
 860     HRESULT hr = RTWQ_E_NOT_FOUND;
 861     struct work_item *item;
 862
 863     EnterCriticalSection(&queue->cs);
 864     LIST_FOR_EACH_ENTRY(item, &queue->pending_items, struct work_item, entry)
 865     {
 866         if (item->key == key)
 867         {
 868             key >>= 32;
 869             if ((key & WAIT_ITEM_KEY_MASK) == WAIT_ITEM_KEY_MASK)
 870             {
 871                 IRtwqAsyncResult_SetStatus(item->result, RTWQ_E_OPERATION_CANCELLED);
 872                 invoke_async_callback(item->result);
 873                 CloseThreadpoolWait(item->u.wait_object);
 874             }
 875             else if ((key & SCHEDULED_ITEM_KEY_MASK) == SCHEDULED_ITEM_KEY_MASK)
 876                 CloseThreadpoolTimer(item->u.timer_object);
 877             else
 878                 WARN("Unknown item key mask %#I64x.\n", key);
 879             queue_release_pending_item(item);
 880             hr = S_OK;
 881             break;
 882         }
 883     }
 884     LeaveCriticalSection(&queue->cs);
 885
 886     return hr;
 887 }
 888
 889 static HRESULT alloc_user_queue(const struct queue_desc *desc, DWORD *queue_id)
 890 {
 891     struct queue_handle *entry;
 892     struct queue *queue;
 893     unsigned int idx;
 894
 895     *queue_id = RTWQ_CALLBACK_QUEUE_UNDEFINED;
 896
 897     if (platform_lock <= 0)
 898         return RTWQ_E_SHUTDOWN;
 899
 900     if (!(queue = calloc(1, sizeof(*queue))))
 901         return E_OUTOFMEMORY;
 902
 903     init_work_queue(desc, queue);
 904
 905     EnterCriticalSection(&queues_section);
 906
 907     entry = next_free_user_queue;
 908     if (entry)
 909         next_free_user_queue = entry->obj;
 910     else if (next_unused_user_queue < user_queues + MAX_USER_QUEUE_HANDLES)
 911         entry = next_unused_user_queue++;
 912     else
 913     {
 914         LeaveCriticalSection(&queues_section);
 915         free(queue);
 916         WARN("Out of user queue handles.\n");
 917         return E_OUTOFMEMORY;
 918     }
 919
 920     entry->refcount = 1;
 921     entry->obj = queue;
 922     if (++queue_generation == 0xffff) queue_generation = 1;
 923     entry->generation = queue_generation;
 924     idx = entry - user_queues + FIRST_USER_QUEUE_HANDLE;
 925     *queue_id = (idx << 16) | entry->generation;
 926
 927     LeaveCriticalSection(&queues_section);
 928
 929     return S_OK;
 930 }
 931
 932 struct async_result
 933 {
 934     RTWQASYNCRESULT result;
 935     LONG refcount;
 936     IUnknown *object;
 937     IUnknown *state;
 938 };
 939
 940 static struct async_result *impl_from_IRtwqAsyncResult(IRtwqAsyncResult *iface)
 941 {
 942     return CONTAINING_RECORD(iface, struct async_result, result.AsyncResult);
 943 }
 944
 945 static HRESULT WINAPI async_result_QueryInterface(IRtwqAsyncResult *iface, REFIID riid, void **obj)
 946 {
 947     TRACE("%p, %s, %p.\n", iface, debugstr_guid(riid), obj);
 948
 949     if (IsEqualIID(riid, &IID_IRtwqAsyncResult) ||
 950             IsEqualIID(riid, &IID_IUnknown))
 951     {
 952         *obj = iface;
 953         IRtwqAsyncResult_AddRef(iface);
 954         return S_OK;
 955     }
 956
 957     *obj = NULL;
 958     WARN("Unsupported interface %s.\n", debugstr_guid(riid));
 959     return E_NOINTERFACE;
 960 }
 961
 962 static ULONG WINAPI async_result_AddRef(IRtwqAsyncResult *iface)
 963 {
 964     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
 965     ULONG refcount = InterlockedIncrement(&result->refcount);
 966
 967     TRACE("%p, %lu.\n", iface, refcount);
 968
 969     return refcount;
 970 }
 971
 972 static ULONG WINAPI async_result_Release(IRtwqAsyncResult *iface)
 973 {
 974     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
 975     ULONG refcount = InterlockedDecrement(&result->refcount);
 976
 977     TRACE("%p, %lu.\n", iface, refcount);
 978
 979     if (!refcount)
 980     {
 981         if (result->result.pCallback)
 982             IRtwqAsyncCallback_Release(result->result.pCallback);
 983         if (result->object)
 984             IUnknown_Release(result->object);
 985         if (result->state)
 986             IUnknown_Release(result->state);
 987         if (result->result.hEvent)
 988             CloseHandle(result->result.hEvent);
 989         free(result);
 990
 991         RtwqUnlockPlatform();
 992     }
 993
 994     return refcount;
 995 }
 996
 997 static HRESULT WINAPI async_result_GetState(IRtwqAsyncResult *iface, IUnknown **state)
 998 {
 999     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1000
1001     TRACE("%p, %p.\n", iface, state);
1002
1003     if (!result->state)
1004         return E_POINTER;
1005
1006     *state = result->state;
1007     IUnknown_AddRef(*state);
1008
1009     return S_OK;
1010 }
1011
1012 static HRESULT WINAPI async_result_GetStatus(IRtwqAsyncResult *iface)
1013 {
1014     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1015
1016     TRACE("%p.\n", iface);
1017
1018     return result->result.hrStatusResult;
1019 }
1020
1021 static HRESULT WINAPI async_result_SetStatus(IRtwqAsyncResult *iface, HRESULT status)
1022 {
1023     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1024
1025     TRACE("%p, %#lx.\n", iface, status);
1026
1027     result->result.hrStatusResult = status;
1028
1029     return S_OK;
1030 }
1031
1032 static HRESULT WINAPI async_result_GetObject(IRtwqAsyncResult *iface, IUnknown **object)
1033 {
1034     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1035
1036     TRACE("%p, %p.\n", iface, object);
1037
1038     if (!result->object)
1039         return E_POINTER;
1040
1041     *object = result->object;
1042     IUnknown_AddRef(*object);
1043
1044     return S_OK;
1045 }
1046
1047 static IUnknown * WINAPI async_result_GetStateNoAddRef(IRtwqAsyncResult *iface)
1048 {
1049     struct async_result *result = impl_from_IRtwqAsyncResult(iface);
1050
1051     TRACE("%p.\n", iface);
1052
1053     return result->state;
1054 }
1055
1056 static const IRtwqAsyncResultVtbl async_result_vtbl =
1057 {
1058     async_result_QueryInterface,
1059     async_result_AddRef,
1060     async_result_Release,
1061     async_result_GetState,
1062     async_result_GetStatus,
1063     async_result_SetStatus,
1064     async_result_GetObject,
1065     async_result_GetStateNoAddRef,
1066 };
1067
1068 static HRESULT create_async_result(IUnknown *object, IRtwqAsyncCallback *callback, IUnknown *state, IRtwqAsyncResult **out)
1069 {
1070     struct async_result *result;
1071
1072     if (!out)
1073         return E_INVALIDARG;
1074
1075     if (!(result = calloc(1, sizeof(*result))))
1076         return E_OUTOFMEMORY;
1077
1078     RtwqLockPlatform();
1079
1080     result->result.AsyncResult.lpVtbl = &async_result_vtbl;
1081     result->refcount = 1;
1082     result->object = object;
1083     if (result->object)
1084         IUnknown_AddRef(result->object);
1085     result->result.pCallback = callback;
1086     if (result->result.pCallback)
1087         IRtwqAsyncCallback_AddRef(result->result.pCallback);
1088     result->state = state;
1089     if (result->state)
1090         IUnknown_AddRef(result->state);
1091
1092     *out = &result->result.AsyncResult;
1093
1094     TRACE("Created async result object %p.\n", *out);
1095
1096     return S_OK;
1097 }
1098
1099 HRESULT WINAPI RtwqCreateAsyncResult(IUnknown *object, IRtwqAsyncCallback *callback, IUnknown *state,
1100         IRtwqAsyncResult **out)
1101 {
1102     TRACE("%p, %p, %p, %p.\n", object, callback, state, out);
1103
1104     return create_async_result(object, callback, state, out);
1105 }
1106
1107 HRESULT WINAPI RtwqLockPlatform(void)
1108 {
1109     InterlockedIncrement(&platform_lock);
1110
1111     return S_OK;
1112 }
1113
1114 HRESULT WINAPI RtwqUnlockPlatform(void)
1115 {
1116     InterlockedDecrement(&platform_lock);
1117
1118     return S_OK;
1119 }
1120
1121 static void init_system_queues(void)
1122 {
1123     struct queue_desc desc;
1124     HRESULT hr;
1125
1126     /* Always initialize standard queue, keep the rest lazy. */
1127
1128     EnterCriticalSection(&queues_section);
1129
1130     if (system_queues[SYS_QUEUE_STANDARD].pool)
1131     {
1132         LeaveCriticalSection(&queues_section);
1133         return;
1134     }
1135
1136     if (FAILED(hr = CoIncrementMTAUsage(&mta_cookie)))
1137         WARN("Failed to initialize MTA, hr %#lx.\n", hr);
1138
1139     desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
1140     desc.ops = &pool_queue_ops;
1141     desc.target_queue = 0;
1142     init_work_queue(&desc, &system_queues[SYS_QUEUE_STANDARD]);
1143
1144     LeaveCriticalSection(&queues_section);
1145 }
1146
1147 HRESULT WINAPI RtwqStartup(void)
1148 {
1149     if (InterlockedIncrement(&platform_lock) == 1)
1150     {
1151         init_system_queues();
1152     }
1153
1154     return S_OK;
1155 }
1156
1157 static void shutdown_system_queues(void)
1158 {
1159     unsigned int i;
1160     HRESULT hr;
1161
1162     EnterCriticalSection(&queues_section);
1163
1164     for (i = 0; i < ARRAY_SIZE(system_queues); ++i)
1165     {
1166         shutdown_queue(&system_queues[i]);
1167     }
1168
1169     if (FAILED(hr = CoDecrementMTAUsage(mta_cookie)))
1170         WARN("Failed to uninitialize MTA, hr %#lx.\n", hr);
1171
1172     LeaveCriticalSection(&queues_section);
1173 }
1174
1175 HRESULT WINAPI RtwqShutdown(void)
1176 {
1177     if (platform_lock <= 0)
1178         return S_OK;
1179
1180     if (InterlockedExchangeAdd(&platform_lock, -1) == 1)
1181     {
1182         shutdown_system_queues();
1183     }
1184
1185     return S_OK;
1186 }
1187
1188 HRESULT WINAPI RtwqPutWaitingWorkItem(HANDLE event, LONG priority, IRtwqAsyncResult *result, RTWQWORKITEM_KEY *key)
1189 {
1190     struct queue *queue;
1191     HRESULT hr;
1192
1193     TRACE("%p, %ld, %p, %p.\n", event, priority, result, key);
1194
1195     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1196         return hr;
1197
1198     hr = queue_submit_wait(queue, event, priority, result, key);
1199
1200     return hr;
1201 }
1202
1203 static HRESULT schedule_work_item(IRtwqAsyncResult *result, INT64 timeout, RTWQWORKITEM_KEY *key)
1204 {
1205     struct queue *queue;
1206     HRESULT hr;
1207
1208     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1209         return hr;
1210
1211     TRACE("%p, %s, %p.\n", result, wine_dbgstr_longlong(timeout), key);
1212
1213     return queue_submit_timer(queue, result, timeout, 0, key);
1214 }
1215
1216 HRESULT WINAPI RtwqScheduleWorkItem(IRtwqAsyncResult *result, INT64 timeout, RTWQWORKITEM_KEY *key)
1217 {
1218     TRACE("%p, %s, %p.\n", result, wine_dbgstr_longlong(timeout), key);
1219
1220     return schedule_work_item(result, timeout, key);
1221 }
1222
1223 struct periodic_callback
1224 {
1225     IRtwqAsyncCallback IRtwqAsyncCallback_iface;
1226     LONG refcount;
1227     RTWQPERIODICCALLBACK callback;
1228 };
1229
1230 static struct periodic_callback *impl_from_IRtwqAsyncCallback(IRtwqAsyncCallback *iface)
1231 {
1232     return CONTAINING_RECORD(iface, struct periodic_callback, IRtwqAsyncCallback_iface);
1233 }
1234
1235 static HRESULT WINAPI periodic_callback_QueryInterface(IRtwqAsyncCallback *iface, REFIID riid, void **obj)
1236 {
1237     if (IsEqualIID(riid, &IID_IRtwqAsyncCallback) ||
1238             IsEqualIID(riid, &IID_IUnknown))
1239     {
1240         *obj = iface;
1241         IRtwqAsyncCallback_AddRef(iface);
1242         return S_OK;
1243     }
1244
1245     *obj = NULL;
1246     return E_NOINTERFACE;
1247 }
1248
1249 static ULONG WINAPI periodic_callback_AddRef(IRtwqAsyncCallback *iface)
1250 {
1251     struct periodic_callback *callback = impl_from_IRtwqAsyncCallback(iface);
1252     ULONG refcount = InterlockedIncrement(&callback->refcount);
1253
1254     TRACE("%p, %lu.\n", iface, refcount);
1255
1256     return refcount;
1257 }
1258
1259 static ULONG WINAPI periodic_callback_Release(IRtwqAsyncCallback *iface)
1260 {
1261     struct periodic_callback *callback = impl_from_IRtwqAsyncCallback(iface);
1262     ULONG refcount = InterlockedDecrement(&callback->refcount);
1263
1264     TRACE("%p, %lu.\n", iface, refcount);
1265
1266     if (!refcount)
1267         free(callback);
1268
1269     return refcount;
1270 }
1271
1272 static HRESULT WINAPI periodic_callback_GetParameters(IRtwqAsyncCallback *iface, DWORD *flags, DWORD *queue)
1273 {
1274     return E_NOTIMPL;
1275 }
1276
1277 static HRESULT WINAPI periodic_callback_Invoke(IRtwqAsyncCallback *iface, IRtwqAsyncResult *result)
1278 {
1279     struct periodic_callback *callback = impl_from_IRtwqAsyncCallback(iface);
1280     IUnknown *context = NULL;
1281
1282     if (FAILED(IRtwqAsyncResult_GetObject(result, &context)))
1283         WARN("Expected object to be set for result object.\n");
1284
1285     callback->callback(context);
1286
1287     if (context)
1288         IUnknown_Release(context);
1289
1290     return S_OK;
1291 }
1292
1293 static const IRtwqAsyncCallbackVtbl periodic_callback_vtbl =
1294 {
1295     periodic_callback_QueryInterface,
1296     periodic_callback_AddRef,
1297     periodic_callback_Release,
1298     periodic_callback_GetParameters,
1299     periodic_callback_Invoke,
1300 };
1301
1302 static HRESULT create_periodic_callback_obj(RTWQPERIODICCALLBACK callback, IRtwqAsyncCallback **out)
1303 {
1304     struct periodic_callback *object;
1305
1306     if (!(object = calloc(1, sizeof(*object))))
1307         return E_OUTOFMEMORY;
1308
1309     object->IRtwqAsyncCallback_iface.lpVtbl = &periodic_callback_vtbl;
1310     object->refcount = 1;
1311     object->callback = callback;
1312
1313     *out = &object->IRtwqAsyncCallback_iface;
1314
1315     return S_OK;
1316 }
1317
1318 HRESULT WINAPI RtwqAddPeriodicCallback(RTWQPERIODICCALLBACK callback, IUnknown *context, DWORD *key)
1319 {
1320     IRtwqAsyncCallback *periodic_callback;
1321     RTWQWORKITEM_KEY workitem_key;
1322     IRtwqAsyncResult *result;
1323     struct queue *queue;
1324     HRESULT hr;
1325
1326     TRACE("%p, %p, %p.\n", callback, context, key);
1327
1328     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1329         return hr;
1330
1331     if (FAILED(hr = create_periodic_callback_obj(callback, &periodic_callback)))
1332         return hr;
1333
1334     hr = create_async_result(context, periodic_callback, NULL, &result);
1335     IRtwqAsyncCallback_Release(periodic_callback);
1336     if (FAILED(hr))
1337         return hr;
1338
1339     /* Same period MFGetTimerPeriodicity() returns. */
1340     hr = queue_submit_timer(queue, result, 0, 10, key ? &workitem_key : NULL);
1341
1342     IRtwqAsyncResult_Release(result);
1343
1344     if (key)
1345         *key = workitem_key;
1346
1347     return S_OK;
1348 }
1349
1350 HRESULT WINAPI RtwqRemovePeriodicCallback(DWORD key)
1351 {
1352     struct queue *queue;
1353     HRESULT hr;
1354
1355     TRACE("%#lx.\n", key);
1356
1357     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1358         return hr;
1359
1360     return queue_cancel_item(queue, get_item_key(SCHEDULED_ITEM_KEY_MASK, key));
1361 }
1362
1363 HRESULT WINAPI RtwqCancelWorkItem(RTWQWORKITEM_KEY key)
1364 {
1365     struct queue *queue;
1366     HRESULT hr;
1367
1368     TRACE("%s.\n", wine_dbgstr_longlong(key));
1369
1370     if (FAILED(hr = grab_queue(RTWQ_CALLBACK_QUEUE_TIMER, &queue)))
1371         return hr;
1372
1373     return queue_cancel_item(queue, key);
1374 }
1375
1376 HRESULT WINAPI RtwqInvokeCallback(IRtwqAsyncResult *result)
1377 {
1378     TRACE("%p.\n", result);
1379
1380     return invoke_async_callback(result);
1381 }
1382
1383 HRESULT WINAPI RtwqPutWorkItem(DWORD queue, LONG priority, IRtwqAsyncResult *result)
1384 {
1385     TRACE("%#lx, %ld, %p.\n", queue, priority, result);
1386
1387     return queue_put_work_item(queue, priority, result);
1388 }
1389
1390 HRESULT WINAPI RtwqAllocateWorkQueue(RTWQ_WORKQUEUE_TYPE queue_type, DWORD *queue)
1391 {
1392     struct queue_desc desc;
1393
1394     TRACE("%d, %p.\n", queue_type, queue);
1395
1396     desc.queue_type = queue_type;
1397     desc.ops = &pool_queue_ops;
1398     desc.target_queue = 0;
1399     return alloc_user_queue(&desc, queue);
1400 }
1401
1402 HRESULT WINAPI RtwqLockWorkQueue(DWORD queue)
1403 {
1404     TRACE("%#lx.\n", queue);
1405
1406     return lock_user_queue(queue);
1407 }
1408
1409 HRESULT WINAPI RtwqUnlockWorkQueue(DWORD queue)
1410 {
1411     TRACE("%#lx.\n", queue);
1412
1413     return unlock_user_queue(queue);
1414 }
1415
1416 HRESULT WINAPI RtwqSetLongRunning(DWORD queue_id, BOOL enable)
1417 {
1418     struct queue *queue;
1419     HRESULT hr;
1420     int i;
1421
1422     TRACE("%#lx, %d.\n", queue_id, enable);
1423
1424     lock_user_queue(queue_id);
1425
1426     if (SUCCEEDED(hr = grab_queue(queue_id, &queue)))
1427     {
1428         for (i = 0; i < ARRAY_SIZE(queue->envs); ++i)
1429             queue->envs[i].u.s.LongFunction = !!enable;
1430     }
1431
1432     unlock_user_queue(queue_id);
1433
1434     return hr;
1435 }
1436
1437 HRESULT WINAPI RtwqLockSharedWorkQueue(const WCHAR *usageclass, LONG priority, DWORD *taskid, DWORD *queue)
1438 {
1439     struct queue_desc desc;
1440     HRESULT hr;
1441
1442     TRACE("%s, %ld, %p, %p.\n", debugstr_w(usageclass), priority, taskid, queue);
1443
1444     if (!usageclass)
1445         return E_POINTER;
1446
1447     if (!*usageclass && taskid)
1448         return E_INVALIDARG;
1449
1450     if (*usageclass)
1451         FIXME("Class name is ignored.\n");
1452
1453     EnterCriticalSection(&queues_section);
1454
1455     if (shared_mt_queue)
1456         hr = lock_user_queue(shared_mt_queue);
1457     else
1458     {
1459         desc.queue_type = RTWQ_MULTITHREADED_WORKQUEUE;
1460         desc.ops = &pool_queue_ops;
1461         desc.target_queue = 0;
1462         hr = alloc_user_queue(&desc, &shared_mt_queue);
1463     }
1464
1465     *queue = shared_mt_queue;
1466
1467     LeaveCriticalSection(&queues_section);
1468
1469     return hr;
1470 }
1471
1472 HRESULT WINAPI RtwqSetDeadline(DWORD queue_id, LONGLONG deadline, HANDLE *request)
1473 {
1474     FIXME("%#lx, %s, %p.\n", queue_id, wine_dbgstr_longlong(deadline), request);
1475
1476     return E_NOTIMPL;
1477 }
1478
1479 HRESULT WINAPI RtwqSetDeadline2(DWORD queue_id, LONGLONG deadline, LONGLONG predeadline, HANDLE *request)
1480 {
1481     FIXME("%#lx, %s, %s, %p.\n", queue_id, wine_dbgstr_longlong(deadline), wine_dbgstr_longlong(predeadline), request);
1482
1483     return E_NOTIMPL;
1484 }
1485
1486 HRESULT WINAPI RtwqCancelDeadline(HANDLE request)
1487 {
1488     FIXME("%p.\n", request);
1489
1490     return E_NOTIMPL;
1491 }
1492
1493 HRESULT WINAPI RtwqAllocateSerialWorkQueue(DWORD target_queue, DWORD *queue)
1494 {
1495     struct queue_desc desc;
1496
1497     TRACE("%#lx, %p.\n", target_queue, queue);
1498
1499     desc.queue_type = RTWQ_STANDARD_WORKQUEUE;
1500     desc.ops = &serial_queue_ops;
1501     desc.target_queue = target_queue;
1502     return alloc_user_queue(&desc, queue);
1503 }
1504
1505 HRESULT WINAPI RtwqJoinWorkQueue(DWORD queue, HANDLE hFile, HANDLE *cookie)
1506 {
1507     FIXME("%#lx, %p, %p.\n", queue, hFile, cookie);
1508
1509     return E_NOTIMPL;
1510 }
1511
1512 HRESULT WINAPI RtwqUnjoinWorkQueue(DWORD queue, HANDLE cookie)
1513 {
1514     FIXME("%#lx, %p.\n", queue, cookie);
1515
1516     return E_NOTIMPL;
1517 }
1518
1519 HRESULT WINAPI RtwqGetWorkQueueMMCSSClass(DWORD queue, WCHAR *class, DWORD *length)
1520 {
1521     FIXME("%#lx, %p, %p.\n", queue, class, length);
1522
1523     return E_NOTIMPL;
1524 }
1525
1526 HRESULT WINAPI RtwqGetWorkQueueMMCSSTaskId(DWORD queue, DWORD *taskid)
1527 {
1528     FIXME("%#lx, %p.\n", queue, taskid);
1529
1530     return E_NOTIMPL;
1531 }
1532
1533 HRESULT WINAPI RtwqGetWorkQueueMMCSSPriority(DWORD queue, LONG *priority)
1534 {
1535     FIXME("%#lx, %p.\n", queue, priority);
1536
1537     return E_NOTIMPL;
1538 }
1539
1540 HRESULT WINAPI RtwqRegisterPlatformWithMMCSS(const WCHAR *class, DWORD *taskid, LONG priority)
1541 {
1542     FIXME("%s, %p, %ld.\n", debugstr_w(class), taskid, priority);
1543
1544     return E_NOTIMPL;
1545 }
1546
1547 HRESULT WINAPI RtwqUnregisterPlatformFromMMCSS(void)
1548 {
1549     FIXME("\n");
1550
1551     return E_NOTIMPL;
1552 }
1553
1554 HRESULT WINAPI RtwqBeginRegisterWorkQueueWithMMCSS(DWORD queue, const WCHAR *class, DWORD taskid, LONG priority,
1555         IRtwqAsyncCallback *callback, IUnknown *state)
1556 {
1557     FIXME("%#lx, %s, %lu, %ld, %p, %p.\n", queue, debugstr_w(class), taskid, priority, callback, state);
1558
1559     return E_NOTIMPL;
1560 }
1561
1562 HRESULT WINAPI RtwqEndRegisterWorkQueueWithMMCSS(IRtwqAsyncResult *result, DWORD *taskid)
1563 {
1564     FIXME("%p, %p.\n", result, taskid);
1565
1566     return E_NOTIMPL;
1567 }
1568
1569 HRESULT WINAPI RtwqBeginUnregisterWorkQueueWithMMCSS(DWORD queue, IRtwqAsyncCallback *callback, IUnknown *state)
1570 {
1571     FIXME("%#lx, %p, %p.\n", queue, callback, state);
1572
1573     return E_NOTIMPL;
1574 }
1575
1576 HRESULT WINAPI RtwqEndUnregisterWorkQueueWithMMCSS(IRtwqAsyncResult *result)
1577 {
1578     FIXME("%p.\n", result);
1579
1580     return E_NOTIMPL;
1581 }
1582
1583 HRESULT WINAPI RtwqRegisterPlatformEvents(IRtwqPlatformEvents *events)
1584 {
1585     FIXME("%p.\n", events);
1586
1587     return E_NOTIMPL;
1588 }
1589
1590 HRESULT WINAPI RtwqUnregisterPlatformEvents(IRtwqPlatformEvents *events)
1591 {
1592     FIXME("%p.\n", events);
1593
1594     return E_NOTIMPL;
1595 }