thread_win32.c

   1 /* -*-c-*- */
   2 /**********************************************************************
   3
   4   thread_win32.c -
   5
   6   $Author$
   7
   8   Copyright (C) 2004-2007 Koichi Sasada
   9
  10 **********************************************************************/
  11
  12 #ifdef THREAD_SYSTEM_DEPENDENT_IMPLEMENTATION
  13
  14 #include <process.h>
  15
  16 #define WIN32_WAIT_TIMEOUT 10   /* 10 ms */
  17 #undef Sleep
  18
  19 #define native_thread_yield() Sleep(0)
  20 #define remove_signal_thread_list(th)
  21
  22 static volatile DWORD ruby_native_thread_key = TLS_OUT_OF_INDEXES;
  23
  24 static int native_mutex_lock(rb_thread_lock_t *);
  25 static int native_mutex_unlock(rb_thread_lock_t *);
  26 static int native_mutex_trylock(rb_thread_lock_t *);
  27 static void native_mutex_initialize(rb_thread_lock_t *);
  28
  29 static void native_cond_signal(rb_thread_cond_t *cond);
  30 static void native_cond_broadcast(rb_thread_cond_t *cond);
  31 static void native_cond_wait(rb_thread_cond_t *cond, rb_thread_lock_t *mutex);
  32 static void native_cond_initialize(rb_thread_cond_t *cond);
  33 static void native_cond_destroy(rb_thread_cond_t *cond);
  34
  35 static rb_thread_t *
  36 ruby_thread_from_native(void)
  37 {
  38     return TlsGetValue(ruby_native_thread_key);
  39 }
  40
  41 static int
  42 ruby_thread_set_native(rb_thread_t *th)
  43 {
  44     return TlsSetValue(ruby_native_thread_key, th);
  45 }
  46
  47 static void
  48 Init_native_thread(void)
  49 {
  50     rb_thread_t *th = GET_THREAD();
  51
  52     ruby_native_thread_key = TlsAlloc();
  53     DuplicateHandle(GetCurrentProcess(),
  54                     GetCurrentThread(),
  55                     GetCurrentProcess(),
  56                     &th->thread_id, 0, FALSE, DUPLICATE_SAME_ACCESS);
  57
  58     th->native_thread_data.interrupt_event = CreateEvent(0, TRUE, FALSE, 0);
  59
  60     thread_debug("initial thread (th: %p, thid: %p, event: %p)\n",
  61                  th, GET_THREAD()->thread_id,
  62                  th->native_thread_data.interrupt_event);
  63 }
  64
  65 static void
  66 w32_error(void)
  67 {
  68     LPVOID lpMsgBuf;
  69     FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER |
  70                   FORMAT_MESSAGE_FROM_SYSTEM |
  71                   FORMAT_MESSAGE_IGNORE_INSERTS,
  72                   NULL,
  73                   GetLastError(),
  74                   MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
  75                   (LPTSTR) & lpMsgBuf, 0, NULL);
  76     rb_bug("%s", (char*)lpMsgBuf);
  77 }
  78
  79 static void
  80 w32_set_event(HANDLE handle)
  81 {
  82     if (SetEvent(handle) == 0) {
  83         w32_error();
  84     }
  85 }
  86
  87 static void
  88 w32_reset_event(HANDLE handle)
  89 {
  90     if (ResetEvent(handle) == 0) {
  91         w32_error();
  92     }
  93 }
  94
  95 static int
  96 w32_wait_events(HANDLE *events, int count, DWORD timeout, rb_thread_t *th)
  97 {
  98     HANDLE *targets = events;
  99     HANDLE intr;
 100     DWORD ret;
 101
 102     thread_debug("  w32_wait_events events:%p, count:%d, timeout:%ld, th:%p\n",
 103                  events, count, timeout, th);
 104     if (th && (intr = th->native_thread_data.interrupt_event)) {
 105         w32_reset_event(intr);
 106         if (RUBY_VM_INTERRUPTED(th)) {
 107             w32_set_event(intr);
 108         }
 109
 110         targets = ALLOCA_N(HANDLE, count + 1);
 111         memcpy(targets, events, sizeof(HANDLE) * count);
 112
 113         targets[count++] = intr;
 114         thread_debug("  * handle: %p (count: %d, intr)\n", intr, count);
 115     }
 116
 117     thread_debug("  WaitForMultipleObjects start (count: %d)\n", count);
 118     ret = WaitForMultipleObjects(count, targets, FALSE, timeout);
 119     thread_debug("  WaitForMultipleObjects end (ret: %lu)\n", ret);
 120
 121     if (ret == WAIT_OBJECT_0 + count - 1 && th) {
 122         errno = EINTR;
 123     }
 124     if (ret == -1 && THREAD_DEBUG) {
 125         int i;
 126         DWORD dmy;
 127         for (i = 0; i < count; i++) {
 128             thread_debug("  * error handle %d - %s\n", i,
 129                          GetHandleInformation(targets[i], &dmy) ? "OK" : "NG");
 130         }
 131     }
 132     return ret;
 133 }
 134
 135 static void ubf_handle(void *ptr);
 136 #define ubf_select ubf_handle
 137
 138 int
 139 rb_w32_wait_events_blocking(HANDLE *events, int num, DWORD timeout)
 140 {
 141     return w32_wait_events(events, num, timeout, GET_THREAD());
 142 }
 143
 144 int
 145 rb_w32_wait_events(HANDLE *events, int num, DWORD timeout)
 146 {
 147     int ret;
 148
 149     BLOCKING_REGION(ret = rb_w32_wait_events_blocking(events, num, timeout),
 150                     ubf_handle, GET_THREAD());
 151     return ret;
 152 }
 153
 154 static void
 155 w32_close_handle(HANDLE handle)
 156 {
 157     if (CloseHandle(handle) == 0) {
 158         w32_error();
 159     }
 160 }
 161
 162 static void
 163 w32_resume_thread(HANDLE handle)
 164 {
 165     if (ResumeThread(handle) == -1) {
 166         w32_error();
 167     }
 168 }
 169
 170 #ifdef _MSC_VER
 171 #define HAVE__BEGINTHREADEX 1
 172 #else
 173 #undef HAVE__BEGINTHREADEX
 174 #endif
 175
 176 #ifdef HAVE__BEGINTHREADEX
 177 #define start_thread (HANDLE)_beginthreadex
 178 typedef unsigned long (_stdcall *w32_thread_start_func)(void*);
 179 #else
 180 #define start_thread CreateThread
 181 typedef LPTHREAD_START_ROUTINE w32_thread_start_func;
 182 #endif
 183
 184 static HANDLE
 185 w32_create_thread(DWORD stack_size, w32_thread_start_func func, void *val)
 186 {
 187     return start_thread(0, stack_size, func, val, CREATE_SUSPENDED, 0);
 188 }
 189
 190 int
 191 rb_w32_sleep(unsigned long msec)
 192 {
 193     return w32_wait_events(0, 0, msec, GET_THREAD());
 194 }
 195
 196 int WINAPI
 197 rb_w32_Sleep(unsigned long msec)
 198 {
 199     int ret;
 200
 201     BLOCKING_REGION(ret = rb_w32_sleep(msec),
 202                     ubf_handle, GET_THREAD());
 203     return ret;
 204 }
 205
 206 static void
 207 native_sleep(rb_thread_t *th, struct timeval *tv)
 208 {
 209     DWORD msec;
 210
 211     if (tv) {
 212         msec = tv->tv_sec * 1000 + tv->tv_usec / 1000;
 213     }
 214     else {
 215         msec = INFINITE;
 216     }
 217
 218     GVL_UNLOCK_BEGIN();
 219     {
 220         DWORD ret;
 221
 222         native_mutex_lock(&th->interrupt_lock);
 223         th->unblock.func = ubf_handle;
 224         th->unblock.arg = th;
 225         native_mutex_unlock(&th->interrupt_lock);
 226
 227         if (RUBY_VM_INTERRUPTED(th)) {
 228             /* interrupted.  return immediate */
 229         }
 230         else {
 231             thread_debug("native_sleep start (%lu)\n", msec);
 232             ret = w32_wait_events(0, 0, msec, th);
 233             thread_debug("native_sleep done (%lu)\n", ret);
 234         }
 235
 236         native_mutex_lock(&th->interrupt_lock);
 237         th->unblock.func = 0;
 238         th->unblock.arg = 0;
 239         native_mutex_unlock(&th->interrupt_lock);
 240     }
 241     GVL_UNLOCK_END();
 242 }
 243
 244 static int
 245 native_mutex_lock(rb_thread_lock_t *lock)
 246 {
 247 #if USE_WIN32_MUTEX
 248     DWORD result;
 249     while (1) {
 250         thread_debug("native_mutex_lock: %p\n", *lock);
 251         result = w32_wait_events(&*lock, 1, INFINITE, 0);
 252         switch (result) {
 253           case WAIT_OBJECT_0:
 254             /* get mutex object */
 255             thread_debug("acquire mutex: %p\n", *lock);
 256             return 0;
 257           case WAIT_OBJECT_0 + 1:
 258             /* interrupt */
 259             errno = EINTR;
 260             thread_debug("acquire mutex interrupted: %p\n", *lock);
 261             return 0;
 262           case WAIT_TIMEOUT:
 263             thread_debug("timeout mutex: %p\n", *lock);
 264             break;
 265           case WAIT_ABANDONED:
 266             rb_bug("win32_mutex_lock: WAIT_ABANDONED");
 267             break;
 268           default:
 269             rb_bug("win32_mutex_lock: unknown result (%d)", result);
 270             break;
 271         }
 272     }
 273     return 0;
 274 #else
 275     EnterCriticalSection(lock);
 276     return 0;
 277 #endif
 278 }
 279
 280 static int
 281 native_mutex_unlock(rb_thread_lock_t *lock)
 282 {
 283 #if USE_WIN32_MUTEX
 284     thread_debug("release mutex: %p\n", *lock);
 285     return ReleaseMutex(*lock);
 286 #else
 287     LeaveCriticalSection(lock);
 288     return 0;
 289 #endif
 290 }
 291
 292 static int
 293 native_mutex_trylock(rb_thread_lock_t *lock)
 294 {
 295 #if USE_WIN32_MUTEX
 296     int result;
 297     thread_debug("native_mutex_trylock: %p\n", *lock);
 298     result = w32_wait_events(&*lock, 1, 1, 0);
 299     thread_debug("native_mutex_trylock result: %d\n", result);
 300     switch (result) {
 301       case WAIT_OBJECT_0:
 302         return 0;
 303       case WAIT_TIMEOUT:
 304         return EBUSY;
 305     }
 306     return EINVAL;
 307 #else
 308     return TryEnterCriticalSection(lock) == 0;
 309 #endif
 310 }
 311
 312 static void
 313 native_mutex_initialize(rb_thread_lock_t *lock)
 314 {
 315 #if USE_WIN32_MUTEX
 316     *lock = CreateMutex(NULL, FALSE, NULL);
 317     if (*lock == NULL) {
 318         w32_error();
 319     }
 320     /* thread_debug("initialize mutex: %p\n", *lock); */
 321 #else
 322     InitializeCriticalSection(lock);
 323 #endif
 324 }
 325
 326 static void
 327 native_mutex_destroy(rb_thread_lock_t *lock)
 328 {
 329 #if USE_WIN32_MUTEX
 330     w32_close_handle(lock);
 331 #else
 332     DeleteCriticalSection(lock);
 333 #endif
 334 }
 335
 336 struct cond_event_entry {
 337     struct cond_event_entry* next;
 338     HANDLE event;
 339 };
 340
 341 struct rb_thread_cond_struct {
 342     struct cond_event_entry *next;
 343     struct cond_event_entry *last;
 344 };
 345
 346 static void
 347 native_cond_signal(rb_thread_cond_t *cond)
 348 {
 349     /* cond is guarded by mutex */
 350     struct cond_event_entry *e = cond->next;
 351
 352     if (e) {
 353         cond->next = e->next;
 354         SetEvent(e->event);
 355     }
 356     else {
 357         rb_bug("native_cond_signal: no pending threads");
 358     }
 359 }
 360
 361 static void
 362 native_cond_broadcast(rb_thread_cond_t *cond)
 363 {
 364     /* cond is guarded by mutex */
 365     struct cond_event_entry *e = cond->next;
 366     cond->next = 0;
 367
 368     while (e) {
 369         SetEvent(e->event);
 370         e = e->next;
 371     }
 372 }
 373
 374 static void
 375 native_cond_wait(rb_thread_cond_t *cond, rb_thread_lock_t *mutex)
 376 {
 377     DWORD r;
 378     struct cond_event_entry entry;
 379
 380     entry.next = 0;
 381     entry.event = CreateEvent(0, FALSE, FALSE, 0);
 382
 383     /* cond is guarded by mutex */
 384     if (cond->next) {
 385         cond->last->next = &entry;
 386         cond->last = &entry;
 387     }
 388     else {
 389         cond->next = &entry;
 390         cond->last = &entry;
 391     }
 392
 393     native_mutex_unlock(mutex);
 394     {
 395         r = WaitForSingleObject(entry.event, INFINITE);
 396         if (r != WAIT_OBJECT_0) {
 397             rb_bug("native_cond_wait: WaitForSingleObject returns %lu", r);
 398         }
 399     }
 400     native_mutex_lock(mutex);
 401
 402     w32_close_handle(entry.event);
 403 }
 404
 405 static void
 406 native_cond_initialize(rb_thread_cond_t *cond)
 407 {
 408     cond->next = 0;
 409     cond->last = 0;
 410 }
 411
 412 static void
 413 native_cond_destroy(rb_thread_cond_t *cond)
 414 {
 415     /* */
 416 }
 417
 418 void
 419 ruby_init_stack(VALUE *addr)
 420 {
 421 }
 422
 423 #define CHECK_ERR(expr) \
 424     {if (!(expr)) {rb_bug("err: %lu - %s", GetLastError(), #expr);}}
 425
 426 static void
 427 native_thread_init_stack(rb_thread_t *th)
 428 {
 429     MEMORY_BASIC_INFORMATION mi;
 430     char *base, *end;
 431     DWORD size, space;
 432
 433     CHECK_ERR(VirtualQuery(&mi, &mi, sizeof(mi)));
 434     base = mi.AllocationBase;
 435     end = mi.BaseAddress;
 436     end += mi.RegionSize;
 437     size = end - base;
 438     space = size / 5;
 439     if (space > 1024*1024) space = 1024*1024;
 440     th->machine_stack_start = (VALUE *)end - 1;
 441     th->machine_stack_maxsize = size - space;
 442 }
 443
 444 static void
 445 native_thread_destroy(rb_thread_t *th)
 446 {
 447     HANDLE intr = th->native_thread_data.interrupt_event;
 448     native_mutex_destroy(&th->interrupt_lock);
 449     thread_debug("close handle - intr: %p, thid: %p\n", intr, th->thread_id);
 450     th->native_thread_data.interrupt_event = 0;
 451     w32_close_handle(intr);
 452 }
 453
 454 static unsigned long _stdcall
 455 thread_start_func_1(void *th_ptr)
 456 {
 457     rb_thread_t *th = th_ptr;
 458     VALUE stack_start;
 459     volatile HANDLE thread_id = th->thread_id;
 460
 461     native_thread_init_stack(th);
 462     th->native_thread_data.interrupt_event = CreateEvent(0, TRUE, FALSE, 0);
 463
 464     /* run */
 465     thread_debug("thread created (th: %p, thid: %p, event: %p)\n", th,
 466                  th->thread_id, th->native_thread_data.interrupt_event);
 467     thread_start_func_2(th, &stack_start, 0);
 468
 469     w32_close_handle(thread_id);
 470     thread_debug("thread deleted (th: %p)\n", th);
 471     return 0;
 472 }
 473
 474 static int
 475 native_thread_create(rb_thread_t *th)
 476 {
 477     size_t stack_size = 4 * 1024; /* 4KB */
 478     th->thread_id = w32_create_thread(stack_size, thread_start_func_1, th);
 479
 480     if ((th->thread_id) == 0) {
 481         st_delete_wrap(th->vm->living_threads, th->self);
 482         rb_raise(rb_eThreadError, "can't create Thread (%d)", errno);
 483     }
 484
 485     w32_resume_thread(th->thread_id);
 486
 487     if (THREAD_DEBUG) {
 488         Sleep(0);
 489         thread_debug("create: (th: %p, thid: %p, intr: %p), stack size: %d\n",
 490                      th, th->thread_id,
 491                      th->native_thread_data.interrupt_event, stack_size);
 492     }
 493     return 0;
 494 }
 495
 496 static void
 497 native_thread_join(HANDLE th)
 498 {
 499     w32_wait_events(&th, 1, 0, 0);
 500 }
 501
 502 #if USE_NATIVE_THREAD_PRIORITY
 503
 504 static void
 505 native_thread_apply_priority(rb_thread_t *th)
 506 {
 507     int priority = th->priority;
 508     if (th->priority > 0) {
 509         priority = THREAD_PRIORITY_ABOVE_NORMAL;
 510     }
 511     else if (th->priority < 0) {
 512         priority = THREAD_PRIORITY_BELOW_NORMAL;
 513     }
 514     else {
 515         priority = THREAD_PRIORITY_NORMAL;
 516     }
 517
 518     SetThreadPriority(th->thread_id, priority);
 519 }
 520
 521 #endif /* USE_NATIVE_THREAD_PRIORITY */
 522
 523 static void
 524 ubf_handle(void *ptr)
 525 {
 526     typedef BOOL (WINAPI *cancel_io_func_t)(HANDLE);
 527     static cancel_io_func_t cancel_func = NULL;
 528     rb_thread_t *th = (rb_thread_t *)ptr;
 529     thread_debug("ubf_handle: %p\n", th);
 530
 531     if (!cancel_func) {
 532         cancel_func = (cancel_io_func_t)GetProcAddress(GetModuleHandle("kernel32"), "CancelSynchronousIo");
 533         if (!cancel_func)
 534             cancel_func = (cancel_io_func_t)-1;
 535     }
 536     if (cancel_func != (cancel_io_func_t)-1)
 537         cancel_func((HANDLE)th->thread_id);
 538
 539     w32_set_event(th->native_thread_data.interrupt_event);
 540 }
 541
 542 static HANDLE timer_thread_id = 0;
 543
 544 static unsigned long _stdcall
 545 timer_thread_func(void *dummy)
 546 {
 547     thread_debug("timer_thread\n");
 548     while (system_working) {
 549         Sleep(WIN32_WAIT_TIMEOUT);
 550         timer_thread_function(dummy);
 551     }
 552     thread_debug("timer killed\n");
 553     return 0;
 554 }
 555
 556 void
 557 rb_thread_create_timer_thread(void)
 558 {
 559     if (timer_thread_id == 0) {
 560         timer_thread_id = w32_create_thread(1024 + (THREAD_DEBUG ? BUFSIZ : 0),
 561                                             timer_thread_func, GET_VM());
 562         w32_resume_thread(timer_thread_id);
 563     }
 564 }
 565
 566 #endif /* THREAD_SYSTEM_DEPENDENT_IMPLEMENTATION */