pthread_workqueue-0.8.2/src/windows/manager.c

   1 /*-
   2  * Copyright (c) 2011, Mark Heily <mark@heily.com>
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions
   7  * are met:
   8  * 1. Redistributions of source code must retain the above copyright
   9  *    notice unmodified, this list of conditions, and the following
  10  *    disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  *
  15  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  16  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  17  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  18  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  19  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  20  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  21  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  22  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  23  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  24  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  25  *
  26  */
  27
  28 #include "platform.h"
  29 #include "../private.h"
  30 #include "pthread_workqueue.h"
  31
  32 #ifdef PROVIDE_LEGACY_XP_SUPPORT
  33
  34 static LIST_HEAD(, _pthread_workqueue) wqlist[WORKQ_NUM_PRIOQUEUE];
  35 static pthread_rwlock_t wqlist_mtx;
  36
  37 int
  38 manager_init(void)
  39 {
  40         pthread_rwlock_init(&wqlist_mtx, NULL);
  41     return (0);
  42 }
  43
  44 void
  45 manager_workqueue_create(struct _pthread_workqueue *workq)
  46 {
  47     pthread_rwlock_wrlock(&wqlist_mtx);
  48     LIST_INSERT_HEAD(&wqlist[workq->queueprio], workq, wqlist_entry);
  49         pthread_rwlock_unlock(&wqlist_mtx);
  50
  51         pthread_spin_init(&workq->mtx, PTHREAD_PROCESS_PRIVATE);
  52 }
  53
  54 /* The caller must hold the wqlist_mtx. */
  55 static struct work *
  56 wqlist_scan(void)
  57 {
  58     pthread_workqueue_t workq;
  59     struct work *witem = NULL;
  60     int i;
  61
  62     pthread_rwlock_rdlock(&wqlist_mtx);
  63     for (i = 0; i < WORKQ_NUM_PRIOQUEUE; i++) {
  64         LIST_FOREACH(workq, &wqlist[i], wqlist_entry) {
  65                         pthread_spin_lock(&workq->mtx);
  66
  67             if (STAILQ_EMPTY(&workq->item_listhead)) {
  68                                 pthread_spin_unlock(&workq->mtx);
  69                                 continue;
  70                         }
  71
  72             witem = STAILQ_FIRST(&workq->item_listhead);
  73             if (witem != NULL)
  74                 STAILQ_REMOVE_HEAD(&workq->item_listhead, item_entry);
  75
  76                         pthread_spin_unlock(&workq->mtx);
  77             goto out;
  78         }
  79     }
  80
  81 out:
  82     pthread_rwlock_unlock(&wqlist_mtx);
  83     return (witem);
  84 }
  85
  86 DWORD WINAPI
  87 worker_main(LPVOID arg)
  88 {
  89     struct work *witem;
  90
  91     witem = wqlist_scan();
  92     if (witem == NULL)
  93             return (0);
  94
  95     witem->func(witem->func_arg);
  96     free(witem);
  97     return (0);
  98 }
  99
 100 void
 101 manager_workqueue_additem(struct _pthread_workqueue *workq, struct work *witem)
 102 {
 103     pthread_spin_lock(&workq->mtx);
 104     STAILQ_INSERT_TAIL(&workq->item_listhead, witem, item_entry);
 105     pthread_spin_unlock(&workq->mtx);
 106     if (!QueueUserWorkItem(worker_main, NULL, WT_EXECUTELONGFUNCTION))
 107             abort();
 108 }
 109
 110 #else
 111
 112 int
 113 manager_init(void)
 114 {
 115     return (0);
 116 }
 117
 118 void
 119 manager_workqueue_create(struct _pthread_workqueue *workq)
 120 {
 121         PTP_POOL pool;
 122         PTP_CALLBACK_ENVIRON callback;
 123         SYSTEM_INFO sysinfo;
 124
 125         pool = CreateThreadpool(NULL);
 126         if(pool == NULL){
 127                 dbg_lasterror("CreateThreadpool()");
 128                 return;
 129         }
 130
 131         InitializeThreadpoolEnvironment(&workq->win_callback_env);
 132         callback = &workq->win_callback_env;
 133         SetThreadpoolCallbackPool(callback, pool);
 134
 135         switch(workq->queueprio){
 136         case WORKQ_HIGH_PRIOQUEUE:
 137                 // weird but this seems the only valid solution !?
 138                 SetThreadpoolCallbackPriority(callback, TP_CALLBACK_PRIORITY_LOW);
 139                 break;
 140         case WORKQ_LOW_PRIOQUEUE:
 141                 // see above
 142                 SetThreadpoolCallbackPriority(callback, TP_CALLBACK_PRIORITY_HIGH);
 143                 break;
 144         default:
 145                 SetThreadpoolCallbackPriority(callback, TP_CALLBACK_PRIORITY_NORMAL);
 146                 break;
 147         }
 148
 149         // we need a proper way to implement overcommitting on windows
 150         if(workq->overcommit){
 151                 GetSystemInfo(&sysinfo);
 152                 SetThreadpoolThreadMaximum(pool, sysinfo.dwNumberOfProcessors * 2);
 153         }
 154
 155         workq->win_thread_pool = pool;
 156 }
 157
 158 VOID CALLBACK
 159 worker_main( PTP_CALLBACK_INSTANCE instance, PVOID Parameter, PTP_WORK work )
 160 {
 161         struct work* witem = (struct work*)Parameter;
 162
 163     assert(witem);
 164         witem->func(witem->func_arg);
 165     free(witem);
 166         CloseThreadpoolWork(work);
 167 }
 168
 169 void
 170 manager_workqueue_additem(struct _pthread_workqueue *workq, struct work *witem)
 171 {
 172         PTP_WORK work = CreateThreadpoolWork(worker_main, witem, &workq->win_callback_env);
 173         if(work == NULL) {
 174                 dbg_lasterror("CreateThreadpoolWork()");
 175                 return;
 176         }
 177         SubmitThreadpoolWork(work);
 178 }
 179
 180 // TODO: We need to cleanly close the environment and threadpools!
 181
 182 #endif
 183
 184 unsigned long
 185 manager_peek(const char *key)
 186 {
 187     unsigned long rv;
 188
 189     if (strcmp(key, "combined_idle") == 0) {
 190         dbg_puts("TODO");
 191         abort();
 192     } else {
 193         dbg_printf("invalid key: ", key);
 194         abort();
 195     }
 196
 197     return rv;
 198 }