From: Nobuyoshi Nakada Date: 2010-07-18T09:05:25+09:00 Subject: [ruby-core:31334] Re: [rubysoc] Queue C-extension patch to come Hi, At Sun, 18 Jul 2010 06:00:56 +0900, Ricardo Panaggio wrote in [ruby-core:31333]: > My implementation is almost 4 times faster than the existing one. > Maybe it's enough, maybe it's not. Let me know if it isn't, so that I > can make any extra tuning, if needed. It was 3 and a half faster on my machine. user system total real Queue: 100| 0.990000 0.030000 1.020000 ( 1.036857) Queue: 1000| 9.910000 0.320000 10.230000 ( 10.677039) Queue:10000| 99.010000 3.130000 102.140000 (105.404587) user system total real Thread::Queue: 100| 0.280000 0.020000 0.300000 ( 0.315329) Thread::Queue: 1000| 2.870000 0.140000 3.010000 ( 3.050338) Thread::Queue:10000| 28.540000 1.450000 29.990000 ( 30.711049) Thread::Queue is C-implementation. Some changes: * moved to an extension library, with renaming as Thread::Queue. * split Thread::Queue#initialize from Thread::Queue.allocate. * untrust queue->que and queue->waiting. otherwise can't modify them when $SAFE == 4. * hide member objects from ObjectSpace, and count those size in queue_memsize. * most importantly, must not access other objects in dfree(), which may be already freed. GC cann't guarantee the order of dfree() calls. diff --git c/ext/thread/queue/extconf.rb i/ext/thread/queue/extconf.rb new file mode 100644 index 0000000..ac7e570 --- /dev/null +++ i/ext/thread/queue/extconf.rb @@ -0,0 +1 @@ +create_makefile('thread/queue') diff --git c/ext/thread/queue/queue.c i/ext/thread/queue/queue.c new file mode 100644 index 0000000..a3de199 --- /dev/null +++ i/ext/thread/queue/queue.c @@ -0,0 +1,315 @@ +#include + +/* + * Document-class: Queue + * + * This class provides a way to synchronize communication between threads. + * + * Example: + * + * require 'thread' + * queue = Queue.new + * + * producer = Thread.new do + * 5.times do |i| + * sleep rand(i) # simulate expense + * queue << i + * puts "#{i} produced" + * end + * end + * + * consumer = Thread.new do + * 5.times do |i| + * value = queue.pop + * sleep rand(i/2) # simulate expense + * puts "consumed #{value}" + * end + * end + * + */ + +typedef struct _Queue { + VALUE mutex; + VALUE que; + VALUE waiting; +} Queue; + +static void +queue_mark(void *ptr) +{ + Queue *queue = ptr; + rb_gc_mark(queue->mutex); + rb_gc_mark(queue->que); + rb_gc_mark(queue->waiting); +} + +#if 0 +static void +queue_free(void *ptr) +{ + if (ptr) { + Queue *queue = ptr; + long i, size = RARRAY_LEN(queue->waiting); + for (i = 0; i < size; i++) { + rb_thread_kill(rb_ary_entry(queue->waiting, i)); + } + ruby_xfree(ptr); + } +} +#else +#define queue_free RUBY_TYPED_DEFAULT_FREE +#endif + +RUBY_EXTERN size_t rb_objspace_data_type_memsize(VALUE); +RUBY_EXTERN size_t rb_ary_memsize(VALUE); + +static size_t +queue_memsize(const void *ptr) +{ + size_t size = 0; + if (ptr) { + const Queue *queue = ptr; + size = sizeof(Queue); + size += rb_objspace_data_type_memsize(queue->mutex); + size += rb_ary_memsize(queue->que); + size += rb_ary_memsize(queue->waiting); + } + return size; +} + +static const rb_data_type_t queue_data_type = { + "queue", + {queue_mark, queue_free, queue_memsize,}, +}; + +#define GetQueuePtr(obj, tobj) \ + TypedData_Get_Struct(obj, Queue, &queue_data_type, tobj) + +static Queue * +get_queue_ptr(VALUE self) +{ + Queue *queue; + GetQueuePtr(self, queue); + if (!queue->mutex || !queue->que || !queue->waiting) { + rb_raise(rb_eArgError, "uninitialized Queue"); + } + return queue; +} + +static VALUE +rb_ary_buf_new(void) +{ + VALUE ary = rb_ary_tmp_new(1); + OBJ_UNTRUST(ary); + return ary; +} + +static VALUE +queue_alloc(VALUE klass) +{ + Queue *queue; + return TypedData_Make_Struct(klass, Queue, &queue_data_type, queue); +} + +/* + * Document-method: new + * call-seq: new + * + * Creates a new queue. + * + */ + +static VALUE +rb_queue_initialize(VALUE self) +{ + Queue *queue; + GetQueuePtr(self, queue); + + queue->mutex = rb_mutex_new(); + RBASIC(queue->mutex)->klass = 0; + queue->que = rb_ary_buf_new(); + queue->waiting = rb_ary_buf_new(); + + return self; +} + +/* + * Document-method: push + * call-seq: push(obj) + * + * Pushes +obj+ to the queue. + * + */ + +static VALUE +rb_queue_push(VALUE self, VALUE obj) +{ + VALUE thread; + Queue *queue = get_queue_ptr(self); + + rb_mutex_lock(queue->mutex); + + rb_ary_push(queue->que, obj); + + do { + thread = rb_ary_shift(queue->waiting); + } while (!NIL_P(thread) && RTEST(rb_thread_wakeup_alive(thread))); + + rb_mutex_unlock(queue->mutex); + + return self; +} + +/* + * Document-method: pop + * call_seq: pop(non_block=false) + * + * Retrieves data from the queue. If the queue is empty, the calling thread is + * suspended until data is pushed onto the queue. If +non_block+ is true, the + * thread isn't suspended, and an exception is raised. + * + */ + +static VALUE +rb_queue_pop(int argc, VALUE *argv, VALUE self) +{ + int should_block; + VALUE poped, current_thread; + Queue *queue = get_queue_ptr(self); + + switch (argc) { + case 0: + should_block = 1; + break; + case 1: + should_block = !RTEST(argv[0]); + break; + default: + rb_raise(rb_eArgError, "wrong number of arguments (%d for 1)", argc); + } + + rb_mutex_lock(queue->mutex); + + while (!RARRAY_LEN(queue->que)) { + if (!should_block) { + rb_mutex_unlock(queue->mutex); + rb_raise(rb_eThreadError, "queue empty"); + } + current_thread = rb_thread_current(); + rb_ary_push(queue->waiting, current_thread); + + rb_mutex_sleep(queue->mutex, Qnil); + } + + poped = rb_ary_shift(queue->que); + + rb_mutex_unlock(queue->mutex); + + return poped; +} + +/* + * Document-method: empty? + * call-seq: empty? + * + * Returns +true+ if the queue is empty. + * + */ + +static VALUE +rb_queue_empty_p(VALUE self) +{ + VALUE result; + Queue *queue = get_queue_ptr(self); + + rb_mutex_lock(queue->mutex); + result = RARRAY_LEN(queue->que) == 0 ? Qtrue : Qfalse; + rb_mutex_unlock(queue->mutex); + + return result; +} + +/* + * Document-method: clear + * call-seq: clear + * + * Removes all objects from the queue. + * + */ + +static VALUE +rb_queue_clear(VALUE self) +{ + Queue *queue = get_queue_ptr(self); + + rb_mutex_lock(queue->mutex); + rb_ary_clear(queue->que); + rb_mutex_unlock(queue->mutex); + + return self; +} + +/* + * Document-method: length + * call-seq: length + * + * Returns the length of the queue. + * + */ + +static VALUE +rb_queue_length(VALUE self) +{ + long len; + VALUE result; + Queue *queue = get_queue_ptr(self); + + rb_mutex_lock(queue->mutex); + len = RARRAY_LEN(queue->que); + result = ULONG2NUM(len); + rb_mutex_unlock(queue->mutex); + + return result; +} + +/* + * Document-method: num_waiting + * call-seq: num_waiting + * + * Returns the number of threads waiting on the queue. + * + */ + +static VALUE +rb_queue_num_waiting(VALUE self) +{ + long len; + VALUE result; + Queue *queue = get_queue_ptr(self); + + rb_mutex_lock(queue->mutex); + len = RARRAY_LEN(queue->waiting); + result = ULONG2NUM(len); + rb_mutex_unlock(queue->mutex); + + return result; +} + +void +Init_queue(void) +{ + VALUE rb_cQueue = rb_define_class_under(rb_cThread, "Queue", rb_cObject); + rb_define_alloc_func(rb_cQueue, queue_alloc); + rb_define_method(rb_cQueue, "initialize", rb_queue_initialize, 0); + rb_define_method(rb_cQueue, "push", rb_queue_push, 1); + rb_define_method(rb_cQueue, "pop", rb_queue_pop, -1); + rb_define_method(rb_cQueue, "empty?", rb_queue_empty_p, 0); + rb_define_method(rb_cQueue, "clear", rb_queue_clear, 0); + rb_define_method(rb_cQueue, "length", rb_queue_length, 0); + rb_define_method(rb_cQueue, "num_waiting", rb_queue_num_waiting, 0); + rb_alias(rb_cQueue, rb_intern("enq"), rb_intern("push")); + rb_alias(rb_cQueue, rb_intern("<<"), rb_intern("push")); + rb_alias(rb_cQueue, rb_intern("deq"), rb_intern("pop")); + rb_alias(rb_cQueue, rb_intern("shift"), rb_intern("pop")); + rb_alias(rb_cQueue, rb_intern("size"), rb_intern("length")); +} -- Nobu Nakada