lib/ruby-1.9.1-rc1/thread_win32.c

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