From: keiju@... (Keiju ISHITSUKA) Date: 2011-01-24T23:08:19+09:00 Subject: [ruby-dev:43092] pthread_cond を用いたConditionVariable --Multipart_Mon_Jan_24_23:08:05_2011-1 Content-Type: text/plain; charset=ISO-2022-JP けいじゅ＠いしつかです. POSIX thread の condition variable を直接用いて ConditionVariable を作っ てみました. とはいえ, まだまだ完全ではないのですが... ConditionVariableを用いたときのThreadのスイッチのコストは ConditionVariableの実行コスト分足されるので, 現行のConditonVariableで はあまりパフォーマンスが良くないと思い作成しました. パフォーマンスの測定はなかなか難しいのですが, 2から4倍ぐらいにはなって いるようです. 実装についてですが, Thread等の仕組みが十分に分かっていないので, 私だけ では完全なものにするのは難しいです. 良く知っていらっしゃる方々に完全な ものにして, Ruby に取り込んでいただければと思っています. まだ, 不十分だと思われるところ, 1. deadlockのチェックができていない. thread->vm->sleeper の辺りが怪し いと思ったのですが, うまくいきませんでした. 2. ConditionVariable.wait(mutex, timeout)でtimeoutの指定されているとき, timeoutで終了しない. 私の環境ではオリジナルのConditionVariableでも timeoutしないのですが... 3. その他色々ありそう... __ ---------------------------------------------------->> 石塚 圭樹 <<--- ---------------------------------->> e-mail: keiju@ishitsuka.com <<--- --Multipart_Mon_Jan_24_23:08:05_2011-1 Content-Type: application/octet-stream; type=patch Content-Disposition: attachment; filename="thread.diff" Content-Transfer-Encoding: quoted-printable diff --git a/thread.c b/thread.c index 69fc299..eca206d 100644 --- a/thread.c +++ b/thread.c @@ -351,7 +351,7 @@ rb_thread_terminate_all(void) =20 /* unlock all locking mutexes */ if (th->keeping_mutexes) { - rb_mutex_unlock_all(th->keeping_mutexes, GET_THREAD()); + rb_mutex_unlock_all(th->keeping_mutexes, GET_THREAD()); } =20 thread_debug("rb_thread_terminate_all (main thread: %p)\n", (void *)th= ); @@ -3494,6 +3494,207 @@ rb_barrier_destroy(VALUE self) return rb_mutex_unlock(mutex); } =20 +VALUE rb_cConditionVariable; + +typedef struct rb_cv_struct +{ + rb_thread_lock_t lock; + rb_thread_cond_t cond; +} cv_t; + +#define GetCVPtr(obj, tobj) \ + TypedData_Get_Struct((obj), cv_t, &cv_data_type, (tobj)) + +#define cv_mark NULL + +static void +cv_free(void *ptr) +{ + if (ptr) { + cv_t *cv =3D ptr; + native_mutex_destroy(&cv->lock); + native_cond_destroy(&cv->cond); + } + ruby_xfree(ptr); +} + +static size_t +cv_memsize(const void *ptr) +{ + return ptr ? sizeof(cv_t) : 0; +} + +static const rb_data_type_t cv_data_type =3D { + "cv", + {cv_mark, cv_free, cv_memsize,}, +}; + +static VALUE +cv_alloc(VALUE klass) +{ + VALUE volatile obj; + cv_t *cv; + + obj =3D TypedData_Make_Struct(klass, cv_t, &cv_data_type, cv); + native_mutex_initialize(&cv->lock); + native_cond_initialize(&cv->cond); + return obj; +} + +/* + * call-seq: + * ConditionVariable.new -> condition_variable + * + * Creates a new ConditionVariable + */ +static VALUE +cv_initialize(VALUE self) +{ + return self; +} + +VALUE +cv_new(void) +{ + return cv_alloc(rb_cConditionVariable); +} + +struct cv_wait_arg { + VALUE mutex; + cv_t *cv; + struct timeval *tv; + enum rb_thread_status prev_status; + struct rb_unblock_callback oldubf; +}; + +static VALUE +rb_cv_wait_forever(struct cv_wait_arg *arg) +{ + rb_thread_t *th =3D GET_THREAD(); + + th->transition_for_lock =3D 1; + BLOCKING_REGION_CORE({ + native_mutex_lock(&arg->cv->lock); + th->transition_for_lock =3D 0; + native_cond_wait(&arg->cv->cond, &arg->cv->lock); + th->transition_for_lock =3D 1; + native_mutex_unlock(&arg->cv->lock); + }); + th->transition_for_lock =3D 0; +} + +static VALUE +rb_cv_wait_for(struct cv_wait_arg *arg) +{ + struct timespec ts; + struct timeval tvn; + rb_thread_t *th =3D GET_THREAD(); + int r; + + gettimeofday(&tvn, NULL); + ts.tv_sec =3D tvn.tv_sec + arg->tv->tv_sec; + ts.tv_nsec =3D (tvn.tv_usec + arg->tv->tv_usec) * 1000; + if (ts.tv_nsec >=3D PER_NANO){ + ts.tv_sec +=3D 1; + ts.tv_nsec -=3D PER_NANO; + } + + th->transition_for_lock =3D 1; + BLOCKING_REGION_CORE({ + native_mutex_lock(&arg->cv->lock); + th->transition_for_lock =3D 0; + native_cond_timedwait(&arg->cv->cond, &arg->cv->lock, &ts); + th->transition_for_lock =3D 1; + native_mutex_unlock(&arg->cv->lock); + }); + th->transition_for_lock =3D 0; +} + +static VALUE +rb_cv_wait_ensure(struct cv_wait_arg *arg) +{ + rb_thread_t *th =3D GET_THREAD(); + =20 + if (th->status =3D=3D THREAD_STOPPED_FOREVER) { + th->status =3D arg->prev_status; + } + reset_unblock_function(th, &arg->oldubf); + rb_mutex_lock(arg->mutex); +} + +static void +rb_cv_wait_interrupt(void *ptr) +{ + cv_t *cv =3D (cv_t *)ptr; + native_mutex_lock(&cv->lock); + native_cond_broadcast(&cv->cond); + native_mutex_unlock(&cv->lock); +} + +VALUE +rb_cv_wait(int argc, VALUE *argv, VALUE self) +{ + VALUE mutex; + VALUE timeout; + cv_t *cv; + struct timeval t; + struct cv_wait_arg arg; + =20 + rb_thread_t *th =3D GET_THREAD(); + =20 + GetCVPtr(self, cv); + =20 + rb_scan_args(argc, argv, "11", &mutex, &timeout); + + if (!NIL_P(timeout)) { + t =3D rb_time_interval(timeout); + } + + arg.mutex =3D mutex; + arg.cv =3D cv; + arg.tv =3D &t; + =20 + rb_mutex_unlock(mutex); + + arg.prev_status =3D th->status; + + set_unblock_function(th, rb_cv_wait_interrupt, cv, &arg.oldubf); + th->status =3D THREAD_STOPPED_FOREVER; + + if (NIL_P(timeout)) { + rb_ensure(rb_cv_wait_forever, (VALUE)&arg, rb_cv_wait_ensure, (VALUE)&= arg); + } + else { + rb_ensure(rb_cv_wait_for, (VALUE)&arg, rb_cv_wait_ensure, (VALUE)&arg); + } + return self; +} + +VALUE +rb_cv_signal(VALUE self) +{ + cv_t *cv; + GetCVPtr(self, cv); + =20 + native_mutex_lock(&cv->lock); + native_cond_signal(&cv->cond); + native_mutex_unlock(&cv->lock); + return self; +} + +VALUE +rb_cv_broadcast(VALUE self) +{ + cv_t *cv; + GetCVPtr(self, cv); + =20 + native_mutex_lock(&cv->lock); + native_cond_broadcast(&cv->cond); + native_mutex_unlock(&cv->lock); + return self; +} + + /* variables for recursive traversals */ static ID recursive_key; =20 @@ -4382,6 +4583,13 @@ Init_Thread(void) rb_define_method(rb_cMutex, "unlock", rb_mutex_unlock, 0); rb_define_method(rb_cMutex, "sleep", mutex_sleep, -1); =20 + rb_cConditionVariable =3D rb_define_class("XConditionVariable", rb_cOb= ject); + rb_define_alloc_func(rb_cConditionVariable, cv_alloc); + rb_define_method(rb_cConditionVariable, "initialize", cv_initialize, 0= ); + rb_define_method(rb_cConditionVariable, "wait", rb_cv_wait, -1); + rb_define_method(rb_cConditionVariable, "signal", rb_cv_signal, 0); + rb_define_method(rb_cConditionVariable, "broadcast", rb_cv_broadcast, = 0); + recursive_key =3D rb_intern("__recursive_key__"); rb_eThreadError =3D rb_define_class("ThreadError", rb_eStandardError); =20 --Multipart_Mon_Jan_24_23:08:05_2011-1--