thread_pthread.c

   1 /* -*-c-*- */
   2 /**********************************************************************
   3
   4   thread_pthread.c -
   5
   6   $Author$
   7
   8   Copyright (C) 2004-2007 Koichi Sasada
   9
  10 **********************************************************************/
  11
  12 #ifdef THREAD_SYSTEM_DEPENDENT_IMPLEMENTATION
  13
  14 #include "gc.h"
  15
  16 #ifdef HAVE_SYS_RESOURCE_H
  17 #include <sys/resource.h>
  18 #endif
  19
  20 static void native_mutex_lock(pthread_mutex_t *lock);
  21 static void native_mutex_unlock(pthread_mutex_t *lock);
  22 static int native_mutex_trylock(pthread_mutex_t *lock);
  23 static void native_mutex_initialize(pthread_mutex_t *lock);
  24 static void native_mutex_destroy(pthread_mutex_t *lock);
  25
  26 static void native_cond_signal(pthread_cond_t *cond);
  27 static void native_cond_broadcast(pthread_cond_t *cond);
  28 static void native_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
  29 static void native_cond_initialize(pthread_cond_t *cond);
  30 static void native_cond_destroy(pthread_cond_t *cond);
  31
  32 static void
  33 native_mutex_lock(pthread_mutex_t *lock)
  34 {
  35     int r;
  36     if ((r = pthread_mutex_lock(lock)) != 0) {
  37         rb_bug("pthread_mutex_lock: %d", r);
  38     }
  39 }
  40
  41 static void
  42 native_mutex_unlock(pthread_mutex_t *lock)
  43 {
  44     int r;
  45     if ((r = pthread_mutex_unlock(lock)) != 0) {
  46         rb_bug("native_mutex_unlock return non-zero: %d", r);
  47     }
  48 }
  49
  50 static inline int
  51 native_mutex_trylock(pthread_mutex_t *lock)
  52 {
  53     int r;
  54     if ((r = pthread_mutex_trylock(lock)) != 0) {
  55         if (r == EBUSY) {
  56             return EBUSY;
  57         }
  58         else {
  59             rb_bug("native_mutex_trylock return non-zero: %d", r);
  60         }
  61     }
  62     return 0;
  63 }
  64
  65 static void
  66 native_mutex_initialize(pthread_mutex_t *lock)
  67 {
  68     int r = pthread_mutex_init(lock, 0);
  69     if (r != 0) {
  70         rb_bug("native_mutex_initialize return non-zero: %d", r);
  71     }
  72 }
  73
  74 static void
  75 native_mutex_destroy(pthread_mutex_t *lock)
  76 {
  77     int r = pthread_mutex_destroy(lock);
  78     if (r != 0) {
  79         rb_bug("native_mutex_destroy return non-zero: %d", r);
  80     }
  81 }
  82
  83 static void
  84 native_cond_initialize(pthread_cond_t *cond)
  85 {
  86     int r = pthread_cond_init(cond, 0);
  87     if (r != 0) {
  88         rb_bug("native_cond_initialize return non-zero: %d", r);
  89     }
  90 }
  91
  92 static void
  93 native_cond_destroy(pthread_cond_t *cond)
  94 {
  95     int r = pthread_cond_destroy(cond);
  96     if (r != 0) {
  97         rb_bug("native_cond_destroy return non-zero: %d", r);
  98     }
  99 }
 100
 101 static void
 102 native_cond_signal(pthread_cond_t *cond)
 103 {
 104     pthread_cond_signal(cond);
 105 }
 106
 107 static void
 108 native_cond_broadcast(pthread_cond_t *cond)
 109 {
 110     pthread_cond_broadcast(cond);
 111 }
 112
 113 static void
 114 native_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex)
 115 {
 116     pthread_cond_wait(cond, mutex);
 117 }
 118
 119
 120 #define native_cleanup_push pthread_cleanup_push
 121 #define native_cleanup_pop  pthread_cleanup_pop
 122 #define native_thread_yield() sched_yield()
 123
 124 #ifndef __CYGWIN__
 125 static void add_signal_thread_list(rb_thread_t *th);
 126 #endif
 127 static void remove_signal_thread_list(rb_thread_t *th);
 128
 129 static rb_thread_lock_t signal_thread_list_lock;
 130
 131 static pthread_key_t ruby_native_thread_key;
 132
 133 static void
 134 null_func(int i)
 135 {
 136     /* null */
 137 }
 138
 139 static rb_thread_t *
 140 ruby_thread_from_native(void)
 141 {
 142     return pthread_getspecific(ruby_native_thread_key);
 143 }
 144
 145 static int
 146 ruby_thread_set_native(rb_thread_t *th)
 147 {
 148     return pthread_setspecific(ruby_native_thread_key, th) == 0;
 149 }
 150
 151 static void
 152 Init_native_thread(void)
 153 {
 154     rb_thread_t *th = GET_THREAD();
 155
 156     pthread_key_create(&ruby_native_thread_key, NULL);
 157     th->thread_id = pthread_self();
 158     native_cond_initialize(&th->native_thread_data.sleep_cond);
 159     ruby_thread_set_native(th);
 160     native_mutex_initialize(&signal_thread_list_lock);
 161     posix_signal(SIGVTALRM, null_func);
 162 }
 163
 164 static void
 165 native_thread_destroy(rb_thread_t *th)
 166 {
 167     pthread_mutex_destroy(&th->interrupt_lock);
 168     pthread_cond_destroy(&th->native_thread_data.sleep_cond);
 169 }
 170
 171 #define USE_THREAD_CACHE 0
 172
 173 static struct {
 174     rb_thread_id_t id;
 175     size_t stack_maxsize;
 176     VALUE *stack_start;
 177 #ifdef __ia64
 178     VALUE *register_stack_start;
 179 #endif
 180 } native_main_thread;
 181
 182 #ifdef STACK_END_ADDRESS
 183 extern void *STACK_END_ADDRESS;
 184 #endif
 185
 186 #undef ruby_init_stack
 187 void
 188 ruby_init_stack(VALUE *addr
 189 #ifdef __ia64
 190     , void *bsp
 191 #endif
 192     )
 193 {
 194     native_main_thread.id = pthread_self();
 195 #ifdef STACK_END_ADDRESS
 196     native_main_thread.stack_start = STACK_END_ADDRESS;
 197 #else
 198     if (!native_main_thread.stack_start ||
 199         STACK_UPPER(&addr,
 200                     native_main_thread.stack_start > addr,
 201                     native_main_thread.stack_start < addr)) {
 202         native_main_thread.stack_start = addr;
 203     }
 204 #endif
 205 #ifdef __ia64
 206     if (!native_main_thread.register_stack_start ||
 207         (VALUE*)bsp < native_main_thread.register_stack_start) {
 208         native_main_thread.register_stack_start = (VALUE*)bsp;
 209     }
 210 #endif
 211 #ifdef HAVE_GETRLIMIT
 212     {
 213         struct rlimit rlim;
 214
 215         if (getrlimit(RLIMIT_STACK, &rlim) == 0) {
 216             unsigned int space = rlim.rlim_cur/5;
 217
 218             if (space > 1024*1024) space = 1024*1024;
 219             native_main_thread.stack_maxsize = rlim.rlim_cur - space;
 220         }
 221     }
 222 #endif
 223 }
 224
 225 #define CHECK_ERR(expr) \
 226     {int err = (expr); if (err) {rb_bug("err: %d - %s", err, #expr);}}
 227
 228 static int
 229 native_thread_init_stack(rb_thread_t *th)
 230 {
 231     rb_thread_id_t curr = pthread_self();
 232
 233     if (pthread_equal(curr, native_main_thread.id)) {
 234         th->machine_stack_start = native_main_thread.stack_start;
 235         th->machine_stack_maxsize = native_main_thread.stack_maxsize;
 236     }
 237     else {
 238 #ifdef HAVE_PTHREAD_GETATTR_NP
 239         pthread_attr_t attr;
 240         void *start;
 241         CHECK_ERR(pthread_getattr_np(curr, &attr));
 242 # if defined HAVE_PTHREAD_ATTR_GETSTACK
 243         CHECK_ERR(pthread_attr_getstack(&attr, &start, &th->machine_stack_maxsize));
 244 # elif defined HAVE_PTHREAD_ATTR_GETSTACKSIZE && defined HAVE_PTHREAD_ATTR_GETSTACKADDR
 245         CHECK_ERR(pthread_attr_getstackaddr(&attr, &start));
 246         CHECK_ERR(pthread_attr_getstacksize(&attr, &th->machine_stack_maxsize));
 247 # endif
 248         th->machine_stack_start = start;
 249 #else
 250         rb_raise(rb_eNotImpError, "ruby engine can initialize only in the main thread");
 251 #endif
 252     }
 253 #ifdef __ia64
 254     th->machine_register_stack_start = native_main_thread.register_stack_start;
 255     th->machine_stack_maxsize /= 2;
 256     th->machine_register_stack_maxsize = th->machine_stack_maxsize;
 257 #endif
 258     return 0;
 259 }
 260
 261 static void *
 262 thread_start_func_1(void *th_ptr)
 263 {
 264 #if USE_THREAD_CACHE
 265   thread_start:
 266 #endif
 267     {
 268         rb_thread_t *th = th_ptr;
 269         VALUE stack_start;
 270
 271         /* run */
 272         thread_start_func_2(th, &stack_start, rb_ia64_bsp());
 273     }
 274 #if USE_THREAD_CACHE
 275     if (1) {
 276         /* cache thread */
 277         rb_thread_t *th;
 278         static rb_thread_t *register_cached_thread_and_wait(void);
 279         if ((th = register_cached_thread_and_wait()) != 0) {
 280             th_ptr = (void *)th;
 281             th->thread_id = pthread_self();
 282             goto thread_start;
 283         }
 284     }
 285 #endif
 286     return 0;
 287 }
 288
 289 void rb_thread_create_control_thread(void);
 290
 291 struct cached_thread_entry {
 292     volatile rb_thread_t **th_area;
 293     pthread_cond_t *cond;
 294     struct cached_thread_entry *next;
 295 };
 296
 297
 298 #if USE_THREAD_CACHE
 299 static pthread_mutex_t thread_cache_lock = PTHREAD_MUTEX_INITIALIZER;
 300 struct cached_thread_entry *cached_thread_root;
 301
 302 static rb_thread_t *
 303 register_cached_thread_and_wait(void)
 304 {
 305     pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
 306     volatile rb_thread_t *th_area = 0;
 307     struct cached_thread_entry *entry =
 308       (struct cached_thread_entry *)malloc(sizeof(struct cached_thread_entry));
 309
 310     struct timeval tv;
 311     struct timespec ts;
 312     gettimeofday(&tv, 0);
 313     ts.tv_sec = tv.tv_sec + 60;
 314     ts.tv_nsec = tv.tv_usec * 1000;
 315
 316     pthread_mutex_lock(&thread_cache_lock);
 317     {
 318         entry->th_area = &th_area;
 319         entry->cond = &cond;
 320         entry->next = cached_thread_root;
 321         cached_thread_root = entry;
 322
 323         pthread_cond_timedwait(&cond, &thread_cache_lock, &ts);
 324
 325         {
 326             struct cached_thread_entry *e = cached_thread_root;
 327             struct cached_thread_entry *prev = cached_thread_root;
 328
 329             while (e) {
 330                 if (e == entry) {
 331                     if (prev == cached_thread_root) {
 332                         cached_thread_root = e->next;
 333                     }
 334                     else {
 335                         prev->next = e->next;
 336                     }
 337                     break;
 338                 }
 339                 prev = e;
 340                 e = e->next;
 341             }
 342         }
 343
 344         free(entry); /* ok */
 345         pthread_cond_destroy(&cond);
 346     }
 347     pthread_mutex_unlock(&thread_cache_lock);
 348
 349     return (rb_thread_t *)th_area;
 350 }
 351 #endif
 352
 353 static int
 354 use_cached_thread(rb_thread_t *th)
 355 {
 356     int result = 0;
 357 #if USE_THREAD_CACHE
 358     struct cached_thread_entry *entry;
 359
 360     if (cached_thread_root) {
 361         pthread_mutex_lock(&thread_cache_lock);
 362         entry = cached_thread_root;
 363         {
 364             if (cached_thread_root) {
 365                 cached_thread_root = entry->next;
 366                 *entry->th_area = th;
 367                 result = 1;
 368             }
 369         }
 370         if (result) {
 371             pthread_cond_signal(entry->cond);
 372         }
 373         pthread_mutex_unlock(&thread_cache_lock);
 374     }
 375 #endif
 376     return result;
 377 }
 378
 379 static int
 380 native_thread_create(rb_thread_t *th)
 381 {
 382     int err = 0;
 383
 384     if (use_cached_thread(th)) {
 385         thread_debug("create (use cached thread): %p\n", th);
 386     }
 387     else {
 388         pthread_attr_t attr;
 389         size_t stack_size = 512 * 1024; /* 512KB */
 390         size_t space;
 391
 392 #ifdef PTHREAD_STACK_MIN
 393         if (stack_size < PTHREAD_STACK_MIN) {
 394             stack_size = PTHREAD_STACK_MIN * 2;
 395         }
 396 #endif
 397         space = stack_size/5;
 398         if (space > 1024*1024) space = 1024*1024;
 399         th->machine_stack_maxsize = stack_size - space;
 400 #ifdef __ia64
 401         th->machine_stack_maxsize /= 2;
 402         th->machine_register_stack_maxsize = th->machine_stack_maxsize;
 403 #endif
 404
 405         CHECK_ERR(pthread_attr_init(&attr));
 406
 407 #ifdef PTHREAD_STACK_MIN
 408         thread_debug("create - stack size: %lu\n", (unsigned long)stack_size);
 409         CHECK_ERR(pthread_attr_setstacksize(&attr, stack_size));
 410 #endif
 411
 412         CHECK_ERR(pthread_attr_setinheritsched(&attr, PTHREAD_INHERIT_SCHED));
 413         CHECK_ERR(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED));
 414
 415         err = pthread_create(&th->thread_id, &attr, thread_start_func_1, th);
 416         thread_debug("create: %p (%d)", th, err);
 417         CHECK_ERR(pthread_attr_destroy(&attr));
 418
 419         if (!err) {
 420             pthread_cond_init(&th->native_thread_data.sleep_cond, 0);
 421         }
 422         else {
 423             st_delete_wrap(th->vm->living_threads, th->self);
 424             th->status = THREAD_KILLED;
 425             rb_raise(rb_eThreadError, "can't create Thread (%d)", err);
 426         }
 427     }
 428     return err;
 429 }
 430
 431 static void
 432 native_thread_join(pthread_t th)
 433 {
 434     int err = pthread_join(th, 0);
 435     if (err) {
 436         rb_raise(rb_eThreadError, "native_thread_join() failed (%d)", err);
 437     }
 438 }
 439
 440 static void
 441 native_thread_apply_priority(rb_thread_t *th)
 442 {
 443 #if defined(_POSIX_PRIORITY_SCHEDULING) && (_POSIX_PRIORITY_SCHEDULING > 0)
 444     struct sched_param sp;
 445     int policy;
 446     int priority = 0 - th->priority;
 447     int max, min;
 448     pthread_getschedparam(th->thread_id, &policy, &sp);
 449     max = sched_get_priority_max(policy);
 450     min = sched_get_priority_min(policy);
 451
 452     if (min > priority) {
 453         priority = min;
 454     }
 455     else if (max < priority) {
 456         priority = max;
 457     }
 458
 459     sp.sched_priority = priority;
 460     pthread_setschedparam(th->thread_id, policy, &sp);
 461 #else
 462     /* not touched */
 463 #endif
 464 }
 465
 466 static void
 467 ubf_pthread_cond_signal(void *ptr)
 468 {
 469     rb_thread_t *th = (rb_thread_t *)ptr;
 470     thread_debug("ubf_pthread_cond_signal (%p)\n", th);
 471     pthread_cond_signal(&th->native_thread_data.sleep_cond);
 472 }
 473
 474 #ifndef __CYGWIN__
 475 static void
 476 ubf_select_each(rb_thread_t *th)
 477 {
 478     thread_debug("ubf_select_each (%p)\n", (void *)th->thread_id);
 479     if (th) {
 480         pthread_kill(th->thread_id, SIGVTALRM);
 481     }
 482 }
 483
 484 static void
 485 ubf_select(void *ptr)
 486 {
 487     rb_thread_t *th = (rb_thread_t *)ptr;
 488     add_signal_thread_list(th);
 489     ubf_select_each(th);
 490 }
 491 #else
 492 #define ubf_select 0
 493 #endif
 494
 495 static void
 496 native_sleep(rb_thread_t *th, struct timeval *tv, int deadlockable)
 497 {
 498     int prev_status = th->status;
 499     struct timespec ts;
 500     struct timeval tvn;
 501
 502     if (tv) {
 503         gettimeofday(&tvn, NULL);
 504         ts.tv_sec = tvn.tv_sec + tv->tv_sec;
 505         ts.tv_nsec = (tvn.tv_usec + tv->tv_usec) * 1000;
 506         if (ts.tv_nsec >= 1000000000){
 507             ts.tv_sec += 1;
 508             ts.tv_nsec -= 1000000000;
 509         }
 510     }
 511
 512     if (!tv && deadlockable) {
 513         th->status = THREAD_STOPPED_FOREVER;
 514         th->vm->sleeper++;
 515         rb_check_deadlock(th->vm);
 516     }
 517     else {
 518         th->status = THREAD_STOPPED;
 519     }
 520
 521     thread_debug("native_sleep %ld\n", tv ? tv->tv_sec : -1);
 522     GVL_UNLOCK_BEGIN();
 523     {
 524         pthread_mutex_lock(&th->interrupt_lock);
 525         th->unblock.func = ubf_pthread_cond_signal;
 526         th->unblock.arg = th;
 527
 528         if (RUBY_VM_INTERRUPTED(th)) {
 529             /* interrupted.  return immediate */
 530             thread_debug("native_sleep: interrupted before sleep\n");
 531         }
 532         else {
 533             if (tv == 0 || ts.tv_sec < tvn.tv_sec /* overflow */ ) {
 534                 int r;
 535                 thread_debug("native_sleep: pthread_cond_wait start\n");
 536                 r = pthread_cond_wait(&th->native_thread_data.sleep_cond,
 537                                   &th->interrupt_lock);
 538                 if (r) rb_bug("pthread_cond_wait: %d", r);
 539                 thread_debug("native_sleep: pthread_cond_wait end\n");
 540             }
 541             else {
 542                 int r;
 543                 thread_debug("native_sleep: pthread_cond_timedwait start (%ld, %ld)\n",
 544                              (unsigned long)ts.tv_sec, ts.tv_nsec);
 545                 r = pthread_cond_timedwait(&th->native_thread_data.sleep_cond,
 546                                            &th->interrupt_lock, &ts);
 547                 if (r && r != ETIMEDOUT) rb_bug("pthread_cond_timedwait: %d", r);
 548
 549                 thread_debug("native_sleep: pthread_cond_timedwait end (%d)\n", r);
 550             }
 551         }
 552         th->unblock.func = 0;
 553         th->unblock.arg = 0;
 554
 555         pthread_mutex_unlock(&th->interrupt_lock);
 556     }
 557     GVL_UNLOCK_END();
 558     th->status = prev_status;
 559     if (!tv && deadlockable) th->vm->sleeper--;
 560     RUBY_VM_CHECK_INTS();
 561
 562     thread_debug("native_sleep done\n");
 563 }
 564
 565 struct signal_thread_list {
 566     rb_thread_t *th;
 567     struct signal_thread_list *prev;
 568     struct signal_thread_list *next;
 569 };
 570
 571 #ifndef __CYGWIN__
 572 static struct signal_thread_list signal_thread_list_anchor = {
 573     0, 0, 0,
 574 };
 575 #endif
 576
 577 #define FGLOCK(lock, body) do { \
 578     native_mutex_lock(lock); \
 579     { \
 580         body; \
 581     } \
 582     native_mutex_unlock(lock); \
 583 } while (0)
 584
 585 #if 0 /* for debug */
 586 static void
 587 print_signal_list(char *str)
 588 {
 589     struct signal_thread_list *list =
 590       signal_thread_list_anchor.next;
 591     thread_debug("list (%s)> ", str);
 592     while(list){
 593         thread_debug("%p (%p), ", list->th, list->th->thread_id);
 594         list = list->next;
 595     }
 596     thread_debug("\n");
 597 }
 598 #endif
 599
 600 #ifndef __CYGWIN__
 601 static void
 602 add_signal_thread_list(rb_thread_t *th)
 603 {
 604     if (!th->native_thread_data.signal_thread_list) {
 605         FGLOCK(&signal_thread_list_lock, {
 606             struct signal_thread_list *list =
 607               malloc(sizeof(struct signal_thread_list));
 608
 609             if (list == 0) {
 610                 fprintf(stderr, "[FATAL] failed to allocate memory\n");
 611                 exit(1);
 612             }
 613
 614             list->th = th;
 615
 616             list->prev = &signal_thread_list_anchor;
 617             list->next = signal_thread_list_anchor.next;
 618             if (list->next) {
 619                 list->next->prev = list;
 620             }
 621             signal_thread_list_anchor.next = list;
 622             th->native_thread_data.signal_thread_list = list;
 623         });
 624     }
 625 }
 626 #endif
 627
 628 static void
 629 remove_signal_thread_list(rb_thread_t *th)
 630 {
 631     if (th->native_thread_data.signal_thread_list) {
 632         FGLOCK(&signal_thread_list_lock, {
 633             struct signal_thread_list *list =
 634               (struct signal_thread_list *)
 635                 th->native_thread_data.signal_thread_list;
 636
 637             list->prev->next = list->next;
 638             if (list->next) {
 639                 list->next->prev = list->prev;
 640             }
 641             th->native_thread_data.signal_thread_list = 0;
 642             list->th = 0;
 643             free(list); /* ok */
 644         });
 645     }
 646     else {
 647         /* */
 648     }
 649 }
 650
 651 static pthread_t timer_thread_id;
 652
 653 static void *
 654 thread_timer(void *dummy)
 655 {
 656     while (system_working) {
 657 #ifdef HAVE_NANOSLEEP
 658         struct timespec req, rem;
 659         req.tv_sec = 0;
 660         req.tv_nsec = 10 * 1000 * 1000; /* 10 ms */
 661         nanosleep(&req, &rem);
 662 #else
 663         struct timeval tv;
 664         tv.tv_sec = 0;
 665         tv.tv_usec = 10000;             /* 10 ms */
 666         select(0, NULL, NULL, NULL, &tv);
 667 #endif
 668 #ifndef __CYGWIN__
 669         if (signal_thread_list_anchor.next) {
 670             FGLOCK(&signal_thread_list_lock, {
 671                 struct signal_thread_list *list;
 672                 list = signal_thread_list_anchor.next;
 673                 while (list) {
 674                     ubf_select_each(list->th);
 675                     list = list->next;
 676                 }
 677             });
 678         }
 679 #endif
 680         timer_thread_function(dummy);
 681     }
 682     return NULL;
 683 }
 684
 685 static void
 686 rb_thread_create_timer_thread(void)
 687 {
 688     rb_enable_interrupt();
 689
 690     if (!timer_thread_id) {
 691         pthread_attr_t attr;
 692         int err;
 693
 694         pthread_attr_init(&attr);
 695 #ifdef PTHREAD_STACK_MIN
 696         pthread_attr_setstacksize(&attr,
 697                                   PTHREAD_STACK_MIN + (THREAD_DEBUG ? BUFSIZ : 0));
 698 #endif
 699         err = pthread_create(&timer_thread_id, &attr, thread_timer, GET_VM());
 700         if (err != 0) {
 701             rb_bug("rb_thread_create_timer_thread: return non-zero (%d)", err);
 702         }
 703     }
 704     rb_disable_interrupt(); /* only timer thread recieve signal */
 705 }
 706
 707 #endif /* THREAD_SYSTEM_DEPENDENT_IMPLEMENTATION */