diff options
| -rw-r--r-- | ChangeLog | 19 | ||||
| -rw-r--r-- | thread.c | 151 | ||||
| -rw-r--r-- | thread_pthread.ci | 60 | ||||
| -rw-r--r-- | thread_win32.ci | 13 | ||||
| -rw-r--r-- | yarvcore.h | 19 |
5 files changed, 181 insertions, 81 deletions
@@ -1,3 +1,22 @@ +Thu Feb 8 15:48:44 2007 Koichi Sasada <ko1@atdot.net> + + * yarvcore.h, thread.c: fix to use pthread on cygwin. + + * yarvcore.h, thread.c: move GVL_UNLOCK_BEGIN() and GVL_UNLOCK_END() + from yarvcore.h to thread.c. + + * thread.c: change GVL_UNLOCK_RANGE() arguments + (adding ubf as 2nd argument). + + * thread.c: fix to use polling in select on cygwin and mswin32. + + * thread.c, thread_pthread.ci, thread_win32.ci, yarvcore.h: + rename: + * rb_thread_t#interrupt_function -> unblock_function + * rb_interrupt_function_t -> rb_unblock_function + * some interrupt function name -> ubf_* + * yarv_* -> * + Thu Feb 8 16:08:02 2007 Koichi Sasada <ko1@atdot.net> * common.mk: fix to use RUNRUBY instead of BASERUBY if possible. @@ -76,21 +76,33 @@ st_delete_wrap(st_table * table, VALUE key) #define THREAD_SYSTEM_DEPENDENT_IMPLEMENTATION -static void native_thread_interrupt(rb_thread_t *th); -static void yarv_set_interrupt_function(rb_thread_t *th, rb_interrupt_function_t *func, int is_return); -static void yarv_clear_interrupt_function(rb_thread_t *th); +static void set_unblock_function(rb_thread_t *th, rb_unblock_function_t *func, int is_return); +static void clear_unblock_function(rb_thread_t *th); -#define GVL_UNLOCK_RANGE(exec) do { \ +NOINLINE(void rb_gc_set_stack_end(VALUE **stack_end_p)); +NOINLINE(void rb_gc_save_machine_context(rb_thread_t *)); + +#define GVL_UNLOCK_BEGIN() do { \ + rb_thread_t *_th_stored = GET_THREAD(); \ + rb_gc_save_machine_context(_th_stored); \ + native_mutex_unlock(&_th_stored->vm->global_interpreter_lock) + +#define GVL_UNLOCK_END() \ + native_mutex_lock(&_th_stored->vm->global_interpreter_lock); \ + rb_thread_set_current(_th_stored); \ +} while(0) + +#define GVL_UNLOCK_RANGE(exec, ubf) do { \ rb_thread_t *__th = GET_THREAD(); \ int __prev_status = __th->status; \ - yarv_set_interrupt_function(__th, native_thread_interrupt, 0); \ + set_unblock_function(__th, ubf, 0); \ __th->status = THREAD_STOPPED; \ GVL_UNLOCK_BEGIN(); {\ exec; \ } \ GVL_UNLOCK_END(); \ - yarv_remove_signal_thread_list(__th); \ - yarv_clear_interrupt_function(__th); \ + remove_signal_thread_list(__th); \ + clear_unblock_function(__th); \ if (__th->status == THREAD_STOPPED) { \ __th->status = __prev_status; \ } \ @@ -103,7 +115,7 @@ void thread_debug(const char *fmt, ...); #define thread_debug if(0)printf #endif -#if defined(_WIN32) || defined(__CYGWIN__) +#if defined(_WIN32) #include "thread_win32.ci" #define DEBUG_OUT() \ @@ -148,7 +160,7 @@ thread_debug(const char *fmt, ...) static void -yarv_set_interrupt_function(rb_thread_t *th, rb_interrupt_function_t *func, int is_return) +set_unblock_function(rb_thread_t *th, rb_unblock_function_t *func, int is_return) { check_ints: RUBY_VM_CHECK_INTS(); @@ -163,16 +175,16 @@ yarv_set_interrupt_function(rb_thread_t *th, rb_interrupt_function_t *func, int } } else { - th->interrupt_function = func; + th->unblock_function = func; } native_mutex_unlock(&th->interrupt_lock); } static void -yarv_clear_interrupt_function(rb_thread_t *th) +clear_unblock_function(rb_thread_t *th) { native_mutex_lock(&th->interrupt_lock); - th->interrupt_function = 0; + th->unblock_function = 0; native_mutex_unlock(&th->interrupt_lock); } @@ -182,8 +194,8 @@ rb_thread_interrupt(rb_thread_t *th) native_mutex_lock(&th->interrupt_lock); th->interrupt_flag = 1; - if (th->interrupt_function) { - (th->interrupt_function)(th); + if (th->unblock_function) { + (th->unblock_function)(th); } else { /* none */ @@ -586,14 +598,15 @@ rb_thread_s_critical(VALUE self) VALUE -rb_thread_run_parallel(VALUE(*func)(rb_thread_t *th, void *), void *data) +rb_thread_run_parallel(VALUE(*func)(rb_thread_t *th, void *), void *data, + rb_unblock_function_t *ubf) { VALUE val; rb_thread_t *th = GET_THREAD(); GVL_UNLOCK_RANGE({ val = func(th, data); - }); + }, ubf); return val; } @@ -691,7 +704,7 @@ rb_thread_ready(rb_thread_t *th) } static VALUE -yarv_thread_raise(int argc, VALUE *argv, rb_thread_t *th) +rb_thread_raise(int argc, VALUE *argv, rb_thread_t *th) { VALUE exc; @@ -718,7 +731,7 @@ rb_thread_signal_raise(void *thptr, const char *sig) } snprintf(buf, BUFSIZ, "SIG%s", sig); argv[0] = rb_exc_new3(rb_eSignal, rb_str_new2(buf)); - yarv_thread_raise(1, argv, th->vm->main_thread); + rb_thread_raise(1, argv, th->vm->main_thread); } void @@ -731,7 +744,7 @@ rb_thread_signal_exit(void *thptr) args[0] = INT2NUM(EXIT_SUCCESS); args[1] = rb_str_new2("exit"); argv[0] = rb_class_new_instance(2, args, rb_eSystemExit); - yarv_thread_raise(1, argv, th->vm->main_thread); + rb_thread_raise(1, argv, th->vm->main_thread); } int @@ -784,7 +797,7 @@ thread_raise_m(int argc, VALUE *argv, VALUE self) { rb_thread_t *th; GetThreadPtr(self, th); - yarv_thread_raise(argc, argv, th); + rb_thread_raise(argc, argv, th); return Qnil; } @@ -1607,27 +1620,86 @@ rb_fd_copy(rb_fdset_t *dst, const fd_set *src, int max) #endif -/* - * c: - */ +static long +cmp_tv(const struct timeval *a, const struct timeval *b) +{ + long d = (a->tv_sec - b->tv_sec); + return (d != 0) ? d : (a->tv_usec - b->tv_usec); +} + +static int +subst(struct timeval *rest, const struct timeval *wait) +{ + while (rest->tv_usec < wait->tv_usec) { + if (rest->tv_sec <= wait->tv_sec) { + return 0; + } + rest->tv_sec -= 1; + rest->tv_usec += 1000 * 1000; + } + rest->tv_sec -= wait->tv_sec; + rest->tv_usec -= wait->tv_usec; + return 1; +} + +static int +do_select(int n, fd_set *read, fd_set *write, fd_set *except, + struct timeval *timeout) +{ + int result, lerrno = 0; +#if defined(__CYGWIN__) || defined(_WIN32) + /* polling port */ + fd_set orig_read, orig_write, orig_except; + struct timeval wait_100ms, *wait; + + wait_100ms.tv_sec = 0; + wait_100ms.tv_usec = 100 * 1000; /* 100 ms */ + wait = (timeout == 0 || cmp_tv(&wait_100ms, timeout) > 0) ? &wait_100ms : timeout; + + do { + if (read) orig_read = *read; + if (write) orig_write = *write; + if (except) orig_except = *except; + + GVL_UNLOCK_RANGE({ + result = select(n, read, write, except, wait); + if (result < 0) lerrno = errno; + }, 0); + + if (result != 0) break; + if (read) *read = orig_read; + if (write) *write = orig_write; + if (except) *except = orig_except; + wait = &wait_100ms; + } while (timeout == 0 || subst(timeout, &wait_100ms)); +#else + GVL_UNLOCK_RANGE({ + result = select(n, read, write, except, timeout); + if (result < 0) lerrno = errno; + }, ubf_select); +#endif + errno = lerrno; + return result; +} + static void rb_thread_wait_fd_rw(int fd, char c) { - rb_fdset_t set; int result = 0; - rb_fd_init(&set); - FD_SET(fd, &set); - thread_debug("rb_thread_wait_fd_rw (%d, %c)\n", fd, c); while (result <= 0) { + rb_fdset_t set; + rb_fd_init(&set); + FD_SET(fd, &set); + switch(c) { case 'r': - GVL_UNLOCK_RANGE(result = select(fd + 1, rb_fd_ptr(&set), 0, 0, 0)); + result = do_select(fd + 1, rb_fd_ptr(&set), 0, 0, 0); break; case'w': - GVL_UNLOCK_RANGE(result = select(fd + 1, 0, rb_fd_ptr(&set), 0, 0)); + result = do_select(fd + 1, 0, rb_fd_ptr(&set), 0, 0); break; default: @@ -1656,7 +1728,7 @@ rb_thread_select(int max, fd_set * read, fd_set * write, fd_set * except, struct timeval *timeout) { struct timeval *tvp = timeout; - int lerrno, n; + int n; #ifndef linux double limit; struct timeval tv; @@ -1688,9 +1760,14 @@ rb_thread_select(int max, fd_set * read, fd_set * write, fd_set * except, #endif for (;;) { - GVL_UNLOCK_RANGE(n = select(max, read, write, except, tvp); - lerrno = errno; - ); +#ifndef linux + fd_set orig_read, orig_write, orig_except; + if (read) orig_read = *read; + if (write) orig_write = *write; + if (except) orig_except = *except; +#endif + + n = do_select(max, read, write, except, tvp); if (n < 0) { switch (errno) { @@ -1704,6 +1781,9 @@ rb_thread_select(int max, fd_set * read, fd_set * write, fd_set * except, double d = limit - timeofday(); tv = double2timeval(d); } + if (read) *read = orig_read; + if (write) *write = orig_write; + if (except) *except = orig_except; #endif continue; default: @@ -2395,3 +2475,8 @@ Init_Thread(void) rb_thread_create_timer_thread(); } +VALUE +is_ruby_native_thread() +{ + return Qtrue; +} diff --git a/thread_pthread.ci b/thread_pthread.ci index 2adde8a5b2..56abaa7d57 100644 --- a/thread_pthread.ci +++ b/thread_pthread.ci @@ -21,14 +21,15 @@ #define native_cleanup_pop pthread_cleanup_pop #define native_thread_yield() sched_yield() -static void yarv_add_signal_thread_list(rb_thread_t *th); -static void yarv_remove_signal_thread_list(rb_thread_t *th); +static void add_signal_thread_list(rb_thread_t *th); +static void remove_signal_thread_list(rb_thread_t *th); static rb_thread_lock_t signal_thread_list_lock; static void null_func() { + /* null */ } static void @@ -238,22 +239,31 @@ native_thread_apply_priority(rb_thread_t *th) } static void -interrupt_using_pthread_cond_signal(rb_thread_t *th) +ubf_pthread_cond_signal(rb_thread_t *th) { - thread_debug("interrupt_using_pthread_cond_signal (%p)\n", th); + thread_debug("ubf_pthread_cond_signal (%p)\n", th); pthread_cond_signal(&th->native_thread_data.sleep_cond); } +#ifndef __CYGWIN__ static void -native_thread_send_interrupt_signal(rb_thread_t *th) +ubf_select_each(rb_thread_t *th) { - thread_debug("native_thread_send_interrupt_signal (%p)\n", th->thread_id); + thread_debug("ubf_select_each (%p)\n", th->thread_id); if (th) { pthread_kill(th->thread_id, SIGVTALRM); } } static void +ubf_select(rb_thread_t *th) +{ + add_signal_thread_list(th); + ubf_select_each(th); +} +#endif + +static void native_sleep(rb_thread_t *th, struct timeval *tv) { int prev_status = th->status; @@ -283,7 +293,7 @@ native_sleep(rb_thread_t *th, struct timeval *tv) thread_debug("native_sleep: interrupted before sleep\n"); } else { - th->interrupt_function = interrupt_using_pthread_cond_signal; + th->unblock_function = ubf_pthread_cond_signal; if (tv == 0) { thread_debug("native_sleep: pthread_cond_wait start\n"); pthread_cond_wait(&th->native_thread_data.sleep_cond, @@ -298,7 +308,7 @@ native_sleep(rb_thread_t *th, struct timeval *tv) &th->interrupt_lock, &ts); thread_debug("native_sleep: pthread_cond_timedwait end (%d)\n", r); } - th->interrupt_function = 0; + th->unblock_function = 0; } pthread_mutex_unlock(&th->interrupt_lock); @@ -308,19 +318,13 @@ native_sleep(rb_thread_t *th, struct timeval *tv) thread_debug("native_sleep done\n"); } -static void -native_thread_interrupt(rb_thread_t *th) -{ - yarv_add_signal_thread_list(th); -} - -struct yarv_signal_thread_list { +struct signal_thread_list { rb_thread_t *th; - struct yarv_signal_thread_list *prev; - struct yarv_signal_thread_list *next; + struct signal_thread_list *prev; + struct signal_thread_list *next; }; -static struct yarv_signal_thread_list signal_thread_list_anchor = { +static struct signal_thread_list signal_thread_list_anchor = { 0, 0, 0, }; @@ -336,7 +340,7 @@ static struct yarv_signal_thread_list signal_thread_list_anchor = { static void print_signal_list(char *str) { - struct yarv_signal_thread_list *list = + struct signal_thread_list *list = signal_thread_list_anchor.next; thread_debug("list (%s)> ", str); while(list){ @@ -348,12 +352,12 @@ print_signal_list(char *str) #endif static void -yarv_add_signal_thread_list(rb_thread_t *th) +add_signal_thread_list(rb_thread_t *th) { if (!th->native_thread_data.signal_thread_list) { FGLOCK(&signal_thread_list_lock, { - struct yarv_signal_thread_list *list = - malloc(sizeof(struct yarv_signal_thread_list)); + struct signal_thread_list *list = + malloc(sizeof(struct signal_thread_list)); if (list == 0) { fprintf(stderr, "[FATAL] failed to allocate memory\n"); @@ -374,12 +378,12 @@ yarv_add_signal_thread_list(rb_thread_t *th) } static void -yarv_remove_signal_thread_list(rb_thread_t *th) +remove_signal_thread_list(rb_thread_t *th) { if (th->native_thread_data.signal_thread_list) { FGLOCK(&signal_thread_list_lock, { - struct yarv_signal_thread_list *list = - (struct yarv_signal_thread_list *) + struct signal_thread_list *list = + (struct signal_thread_list *) th->native_thread_data.signal_thread_list; list->prev->next = list->next; @@ -414,16 +418,18 @@ thread_timer(void *dummy) tv.tv_usec = 10000; /* 10 ms */ select(0, NULL, NULL, NULL, &tv); #endif +#ifndef __CYGWIN__ if (signal_thread_list_anchor.next) { FGLOCK(&signal_thread_list_lock, { - struct yarv_signal_thread_list *list; + struct signal_thread_list *list; list = signal_thread_list_anchor.next; while (list) { - native_thread_send_interrupt_signal(list->th); + ubf_select_each(list->th); list = list->next; } }); } +#endif timer_thread_function(); } return NULL; diff --git a/thread_win32.ci b/thread_win32.ci index e8d7facbaf..dd9a24aad1 100644 --- a/thread_win32.ci +++ b/thread_win32.ci @@ -18,7 +18,7 @@ #undef Sleep #define native_thread_yield() Sleep(0) -#define yarv_remove_signal_thread_list(th) +#define remove_signal_thread_list(th) static void Init_native_thread() @@ -93,6 +93,9 @@ w32_wait_event(HANDLE event, DWORD timeout, rb_thread_t *th) return ret; } +static void ubf_handle(rb_thread_t *th); +#define ubf_select ubf_handle + static void native_sleep(rb_thread_t *th, struct timeval *tv) { @@ -109,11 +112,11 @@ native_sleep(rb_thread_t *th, struct timeval *tv) DWORD ret; int status = th->status; th->status = THREAD_STOPPED; - th->interrupt_function = native_thread_interrupt; + th->unblock_function = ubf_handle; thread_debug("native_sleep start (%d)\n", (int)msec); ret = w32_wait_event(0, msec, th); thread_debug("native_sleep done (%d)\n", ret); - th->interrupt_function = 0; + th->unblock_function = 0; th->status = status; } GVL_UNLOCK_END(); @@ -283,9 +286,9 @@ native_thread_apply_priority(rb_thread_t *th) } static void -native_thread_interrupt(rb_thread_t *th) +ubf_handle(rb_thread_t *th) { - thread_debug("native_thread_interrupt: %p\n", th); + thread_debug("ubf_handle: %p\n", th); SetEvent(th->native_thread_data.interrupt_event); } diff --git a/yarvcore.h b/yarvcore.h index 8bfe5fe20f..4d98f42b90 100644 --- a/yarvcore.h +++ b/yarvcore.h @@ -24,7 +24,7 @@ #include "debug.h" #include "vm_opts.h" -#if defined(_WIN32) || defined(__CYGWIN__) +#if defined(_WIN32) #include "thread_win32.h" #elif defined(HAVE_PTHREAD_H) #include "thread_pthread.h" @@ -376,7 +376,7 @@ struct rb_vm_tag { struct rb_vm_tag *prev; }; -typedef void rb_interrupt_function_t(struct rb_thread_struct *); +typedef void rb_unblock_function_t(struct rb_thread_struct *); #define RUBY_VM_VALUE_CACHE_SIZE 0x1000 #define USE_VALUE_CACHE 1 @@ -423,7 +423,7 @@ typedef struct rb_thread_struct int exec_signal; int interrupt_flag; - rb_interrupt_function_t *interrupt_function; + rb_unblock_function_t *unblock_function; rb_thread_lock_t interrupt_lock; struct rb_vm_tag *tag; @@ -632,19 +632,6 @@ extern rb_vm_t *theYarvVM; #error "unsupported thread model" #endif -#define GVL_UNLOCK_BEGIN() do { \ - rb_thread_t *_th_stored = GET_THREAD(); \ - rb_gc_save_machine_context(_th_stored); \ - native_mutex_unlock(&_th_stored->vm->global_interpreter_lock) - -#define GVL_UNLOCK_END() \ - native_mutex_lock(&_th_stored->vm->global_interpreter_lock); \ - rb_thread_set_current(_th_stored); \ -} while(0) - -NOINLINE(void rb_gc_set_stack_end(VALUE **stack_end_p)); -NOINLINE(void rb_gc_save_machine_context(rb_thread_t *)); - void rb_thread_execute_interrupts(rb_thread_t *); #define RUBY_VM_CHECK_INTS_TH(th) do { \ |