diff --git a/mrbgems/mruby-task/README.md b/mrbgems/mruby-task/README.md index c26179574..092da97a7 100644 --- a/mrbgems/mruby-task/README.md +++ b/mrbgems/mruby-task/README.md @@ -13,8 +13,9 @@ The primary purpose of `mruby-task` is to enable mruby applications to: - Schedule tasks based on priority (0-255, where 0 is highest priority). - Provide cooperative yielding with `Task.pass`. - Support preemptive scheduling via timer-based interrupts. -- Synchronize tasks using `sleep` and `join` operations. +- Synchronize tasks using `sleep`, `join` and `Task::Queue`. - Suspend and resume tasks programmatically. +- Coordinate producers and consumers via `Task::Queue` without polling. ## Architecture @@ -186,6 +187,116 @@ Task.run puts "Worker finished" ``` +### Task::Error + +`Task::Error` is the base error class for queue-related errors. It inherits +from `StandardError`. + +```ruby +begin + q = Task::Queue.new + q.close + q.push(1) # raises Task::Error: queue closed +rescue Task::Error => e + puts e.message # => "queue closed" +end +``` + +Errors raised by `Task::Queue`: + +| Situation | Error class | Message | +| ---------------------------------- | ------------- | ---------------- | +| `push` on a closed queue | `Task::Error` | `"queue closed"` | +| `pop(true)` on an empty open queue | `Task::Error` | `"queue empty"` | + +Internal consistency errors (programming errors, not queue logic) still use +`RuntimeError`: + +| Situation | Error class | +| ------------------------------------------------------- | -------------- | +| Blocking `pop` called from root context | `RuntimeError` | +| Blocking `pop` called from inside a C function boundary | `RuntimeError` | + +### Task::Queue + +`Task::Queue` is a thread-safe FIFO queue for inter-task communication, analogous to CRuby's `Queue`. A task that calls `pop` on an empty queue is automatically moved to the WAITING state and rescheduled when another task pushes an item. +No polling or explicit sleep is required. + +#### Creating a Queue + +```ruby +q = Task::Queue.new +``` + +#### Pushing Items + +```ruby +q.push(item) # add to the back of the queue; raises Task::Error if closed +q << item # alias for push +q.enq(item) # alias for push +``` + +#### Popping Items + +```ruby +item = q.pop # block until an item is available +item = q.pop(true) # non-blocking: raises Task::Error if empty +item = q.deq # alias for pop +item = q.shift # alias for pop +``` + +Behavior when the queue is **closed**: + +- `pop` on a non-empty closed queue returns the remaining items normally. +- `pop` on an empty closed queue returns `nil` immediately (no blocking). + +#### Inspecting the Queue + +```ruby +q.size # => Integer: number of items currently in the queue +q.length # alias for size +q.empty? # => true if the queue has no items +q.num_waiting # => Integer: number of tasks currently blocked in pop +``` + +#### Clearing and Closing + +```ruby +q.clear # remove all items; returns self +q.close # close the queue; returns self +q.closed? # => true if the queue has been closed +``` + +After `close`: + +- `push` raises `Task::Error`. +- Tasks blocked in `pop` are woken and receive `nil` for an empty queue. +- `pop` on remaining items still returns them normally; returns `nil` when empty. + +#### Producer/Consumer Example + +```ruby +q = Task::Queue.new + +Task.new(name: "producer") do + 10.times do |i| + q.push(i) + sleep 0.1 + end + q.close +end + +Task.new(name: "consumer") do + loop do + item = q.pop # blocks until an item arrives or the queue closes + break if item.nil? + puts "got #{item}" + end +end + +Task.run +``` + ### Kernel Methods (Sleep) The task scheduler provides task-aware sleep methods that cooperatively yield @@ -657,6 +768,49 @@ sleep 1 task.terminate # Stop permanently ``` +### Producer/Consumer with Task::Queue + +Using `Task::Queue` eliminates polling. The consumer blocks on `pop` and wakes +automatically when the producer pushes an item. + +```ruby +q = Task::Queue.new +results = [] + +Task.new(name: "producer") do + ["a", "b", "c"].each do |v| + q.push(v) + sleep 0.1 + end + q.close +end + +Task.new(name: "consumer") do + loop do + item = q.pop # blocks here until an item is available or the queue closes + break if item.nil? + results << item + end +end + +Task.run +puts results.inspect # => ["a", "b", "c"] +``` + +Multiple producers and consumers work naturally: + +```ruby +q = Task::Queue.new + +3.times { |i| Task.new { q.push(i) } } + +received = [] +3.times { Task.new { received << q.pop } } + +Task.run +puts received.sort.inspect # => [0, 1, 2] +``` + ## Limitations and Compatibility ### Relationship with Fiber @@ -701,6 +855,10 @@ The gem includes tests that verify: - Join synchronization - Suspend and resume - Task.pass cooperative yielding +- Task::Queue push/pop FIFO order +- Task::Queue blocking pop and wakeup on push +- Task::Queue close semantics +- Task::Queue num_waiting count Run tests with: @@ -717,7 +875,7 @@ Each task can be in one of five states: - `DORMANT (0x00)`: Not started or finished - `READY (0x02)`: Ready to run - `RUNNING (0x03)`: Currently executing -- `WAITING (0x04)`: Waiting (sleep, join, mutex) +- `WAITING (0x04)`: Waiting (sleep, join, queue) - `SUSPENDED (0x08)`: Manually suspended ### Wait Reasons @@ -728,6 +886,7 @@ When a task is in WAITING state, the reason indicates why: - `SLEEP (0x01)`: Sleeping for time - `MUTEX (0x02)`: Waiting for mutex (reserved, not yet implemented) - `JOIN (0x04)`: Waiting for another task +- `QUEUE (0x08)`: Waiting for an item to be pushed to a `Task::Queue` ### Scheduler Algorithm @@ -756,6 +915,7 @@ The scheduler includes a lock counter (`mrb->task.scheduler_lock`) that prevents Planned features not yet implemented: - **Mutex support**: Thread-safe synchronization primitives +- **Task::SizedQueue**: Bounded queue with backpressure (push blocks when full) - **Task.raise**: Throw exceptions to other tasks - **Task#value**: Retrieve task return value (like Thread#value) - **Per-task timeslice configuration** diff --git a/mrbgems/mruby-task/examples/queue.rb b/mrbgems/mruby-task/examples/queue.rb new file mode 100644 index 000000000..0308de8bf --- /dev/null +++ b/mrbgems/mruby-task/examples/queue.rb @@ -0,0 +1,54 @@ +# Task::Queue Example +# Demonstrates producer/consumer coordination using Task::Queue. +# No polling required - consumers block until items are available. + +puts "=== Task::Queue Producer/Consumer Demo ===" +puts + +TOTAL_ITEMS = 10 +q = Task::Queue.new +produced = 0 +consumed = 0 + +producer = Task.new(name: "producer") do + TOTAL_ITEMS.times do |i| + item = "item-#{i}" + q.push(item) + produced += 1 + puts "Producer: pushed #{item}" + sleep 0.1 + end + q.close + puts "Producer: closed queue after #{produced} items" +end + +consumer1 = Task.new(name: "consumer-1") do + loop do + item = q.pop # blocks until an item is available or queue closes + break if item.nil? + consumed += 1 + puts "Consumer-1: got #{item}" + sleep 0.15 + end + puts "Consumer-1: done" +end + +consumer2 = Task.new(name: "consumer-2") do + loop do + item = q.pop + break if item.nil? + consumed += 1 + puts "Consumer-2: got #{item}" + sleep 0.2 + end + puts "Consumer-2: done" +end + +Task.run + +puts +puts "=== Summary ===" +puts "Produced: #{produced}" +puts "Consumed: #{consumed}" +puts "Queue size: #{q.size}" +puts "Queue closed: #{q.closed?}" diff --git a/mrbgems/mruby-task/include/task.h b/mrbgems/mruby-task/include/task.h index 25154ba1b..121e32c02 100644 --- a/mrbgems/mruby-task/include/task.h +++ b/mrbgems/mruby-task/include/task.h @@ -28,8 +28,11 @@ enum { MRB_TASK_REASON_SLEEP = 0x01, /* Sleeping for time */ MRB_TASK_REASON_MUTEX = 0x02, /* Waiting for mutex (reserved) */ MRB_TASK_REASON_JOIN = 0x04, /* Waiting for another task */ + MRB_TASK_REASON_QUEUE = 0x08, /* Waiting for queue item */ }; +struct mrb_task_queue; + /* * Task structure - represents a single task in the scheduler * @@ -53,6 +56,7 @@ typedef struct mrb_task { uint32_t wakeup_tick; /* Tick count to wake up (REASON_SLEEP) */ const struct mrb_task *join; /* Task being waited on (REASON_JOIN) */ void *mutex; /* Mutex pointer (REASON_MUTEX, reserved) */ + struct mrb_task_queue *queue; /* Queue being waited on (REASON_QUEUE) */ } wait; mrb_value self; /* Ruby Task object reference */ @@ -140,4 +144,38 @@ MRB_API void mrb_task_init_context(mrb_state *mrb, mrb_value task, struct RProc MRB_API void mrb_task_reset_context(mrb_state *mrb, mrb_value task); MRB_API void mrb_task_proc_set(mrb_state *mrb, mrb_value task, struct RProc *proc); +/* + * Internal helpers - used by task.c and task_queue.c + */ +#include +#include "task_hal.h" + +/* Scheduler state accessors (require a local mrb variable in scope) */ +#define q_dormant_ (mrb->task.queues[MRB_TASK_QUEUE_DORMANT]) +#define q_ready_ (mrb->task.queues[MRB_TASK_QUEUE_READY]) +#define q_waiting_ (mrb->task.queues[MRB_TASK_QUEUE_WAITING]) +#define q_suspended_ (mrb->task.queues[MRB_TASK_QUEUE_SUSPENDED]) +#define tick_ (mrb->task.tick) +#define wakeup_tick_ (mrb->task.wakeup_tick) +#define switching_ (mrb->task.switching) + +/* Recover the mrb_task that owns the current mruby context */ +#define MRB2TASK(mrb) ((mrb_task *)((uint8_t *)(mrb)->c - offsetof(mrb_task, c))) + +/* Raise if the scheduler is locked (synchronous execution in progress) */ +static inline void +task_check_scheduler_lock(mrb_state *mrb) +{ + if (mrb->task.scheduler_lock > 0) { + mrb_raise(mrb, E_RUNTIME_ERROR, "Cannot use asynchronous Task API during synchronous execution"); + } +} + +/* Priority-queue insert/delete - defined in task.c */ +void q_insert_task(mrb_state *mrb, mrb_task *t); +void q_delete_task(mrb_state *mrb, mrb_task *t); + +/* Task::Queue class registration - defined in task_queue.c */ +void mrb_init_task_queue(mrb_state *mrb, struct RClass *task_class); + #endif /* MRUBY_TASK_H */ diff --git a/mrbgems/mruby-task/mrblib/queue.rb b/mrbgems/mruby-task/mrblib/queue.rb new file mode 100644 index 000000000..792dab885 --- /dev/null +++ b/mrbgems/mruby-task/mrblib/queue.rb @@ -0,0 +1,30 @@ +class Task + class Queue + # WAIT_RETRY is defined in C (task_queue.c gem init) + + def push(obj) + __push(obj) + self + end + alias enq push + alias << push + + # Blocks until an item is available (default), or raises if non_block is true. + # Returns nil if the queue is closed and empty. + # + # The loop is not a busy-wait. When __pop_try finds the queue empty it moves + # the current task to WAITING and sets switching_=TRUE before returning + # WAIT_RETRY. The VM detects switching_ at the next opcode boundary and + # exits mrb_vm_exec, handing control back to the scheduler. This task does + # not run again until a push (or close) moves it back to READY. The loop + # body therefore executes at most once per wakeup event. + def pop(non_block = false) + loop do + v = __pop_try(non_block) + return v unless v.equal?(WAIT_RETRY) + end + end + alias deq pop + alias shift pop + end +end diff --git a/mrbgems/mruby-task/src/task.c b/mrbgems/mruby-task/src/task.c index a8a627409..c49e25a7f 100644 --- a/mrbgems/mruby-task/src/task.c +++ b/mrbgems/mruby-task/src/task.c @@ -19,21 +19,6 @@ #include #include #include "task.h" -#include "task_hal.h" - -/* - * Queue helper macros - */ -#define q_dormant_ (mrb->task.queues[MRB_TASK_QUEUE_DORMANT]) -#define q_ready_ (mrb->task.queues[MRB_TASK_QUEUE_READY]) -#define q_waiting_ (mrb->task.queues[MRB_TASK_QUEUE_WAITING]) -#define q_suspended_ (mrb->task.queues[MRB_TASK_QUEUE_SUSPENDED]) -#define tick_ (mrb->task.tick) -#define wakeup_tick_ (mrb->task.wakeup_tick) -#define switching_ (mrb->task.switching) - -/* Get task from current context using pointer arithmetic */ -#define MRB2TASK(mrb) ((mrb_task *)((uint8_t *)mrb->c - offsetof(mrb_task, c))) /* Get task pointer from self with validation */ #define TASK_GET_PTR_OR_RAISE(var, self) \ @@ -50,15 +35,6 @@ /* Maximum value for scheduler_lock (uint8_t max) */ #define MRB_TASK_SCHEDULER_LOCK_MAX 255 -/* Check scheduler lock and raise error if locked */ -static inline void -task_check_scheduler_lock(mrb_state *mrb) -{ - if (mrb->task.scheduler_lock > 0) { - mrb_raise(mrb, E_RUNTIME_ERROR, "Cannot use asynchronous Task API during synchronous execution"); - } -} - /* * Task data type for GC */ @@ -171,7 +147,7 @@ q_get_queue(mrb_state *mrb, mrb_task *t) } /* Insert task into queue based on priority (higher priority = lower number = earlier in queue) */ -static void +void q_insert_task(mrb_state *mrb, mrb_task *t) { mrb_task **q = q_get_queue(mrb, t); @@ -195,7 +171,7 @@ q_insert_task(mrb_state *mrb, mrb_task *t) } /* Delete task from its current queue */ -static void +void q_delete_task(mrb_state *mrb, mrb_task *t) { mrb_task **q = q_get_queue(mrb, t); @@ -1528,6 +1504,12 @@ mrb_mruby_task_gem_init(mrb_state *mrb) task_class = mrb_define_class_id(mrb, MRB_SYM(Task), mrb->object_class); MRB_SET_INSTANCE_TT(task_class, MRB_TT_DATA); + /* Task::Error - base error class for task synchronization errors */ + mrb_define_class_under_id(mrb, task_class, MRB_SYM(Error), mrb->eStandardError_class); + + /* Task::Queue */ + mrb_init_task_queue(mrb, task_class); + /* Class methods */ mrb_define_class_method_id(mrb, task_class, MRB_SYM(new), mrb_task_s_new, MRB_ARGS_KEY(2,0)|MRB_ARGS_BLOCK()); mrb_define_class_method_id(mrb, task_class, MRB_SYM(current), mrb_task_s_current, MRB_ARGS_NONE()); diff --git a/mrbgems/mruby-task/src/task_queue.c b/mrbgems/mruby-task/src/task_queue.c new file mode 100644 index 000000000..bf46c0ded --- /dev/null +++ b/mrbgems/mruby-task/src/task_queue.c @@ -0,0 +1,253 @@ +/* +** task_queue.c - Task::Queue implementation +*/ + +#include +#include +#include +#include +#include +#include +#include "task.h" + +typedef struct mrb_task_queue { + uint8_t closed; +} mrb_task_queue; + +static void +mrb_task_queue_free(mrb_state *mrb, void *ptr) +{ + mrb_free(mrb, ptr); +} + +static struct RClass * +queue_task_error_class(mrb_state *mrb) +{ + return mrb_class_get_under(mrb, mrb_class_get(mrb, "Task"), "Error"); +} + +static const struct mrb_data_type mrb_task_queue_type = { + "Task::Queue", mrb_task_queue_free, +}; + +static mrb_value wait_retry_; + +/* Wake the highest-priority task waiting on this queue */ +static void +queue_wake_one_waiter(mrb_state *mrb, mrb_task_queue *q) +{ + mrb_task_disable_irq(); + mrb_task *curr = q_waiting_; + while (curr) { + mrb_task *next = curr->next; + if (curr->reason == MRB_TASK_REASON_QUEUE && curr->wait.queue == q) { + q_delete_task(mrb, curr); + curr->status = MRB_TASK_STATUS_READY; + curr->reason = MRB_TASK_REASON_NONE; + curr->wait.queue = NULL; + q_insert_task(mrb, curr); + switching_ = TRUE; + break; + } + curr = next; + } + mrb_task_enable_irq(); +} + +/* Wake all tasks waiting on this queue (used by close) */ +static void +queue_wake_all_waiters(mrb_state *mrb, mrb_task_queue *q) +{ + mrb_bool woke_any = FALSE; + mrb_task_disable_irq(); + mrb_task *curr = q_waiting_; + while (curr) { + mrb_task *next = curr->next; + if (curr->reason == MRB_TASK_REASON_QUEUE && curr->wait.queue == q) { + q_delete_task(mrb, curr); + curr->status = MRB_TASK_STATUS_READY; + curr->reason = MRB_TASK_REASON_NONE; + curr->wait.queue = NULL; + q_insert_task(mrb, curr); + woke_any = TRUE; + } + curr = next; + } + if (woke_any) { + switching_ = TRUE; + } + mrb_task_enable_irq(); +} + +static mrb_value +queue_initialize(mrb_state *mrb, mrb_value self) +{ + mrb_task_queue *q = (mrb_task_queue*)mrb_malloc(mrb, sizeof(mrb_task_queue)); + q->closed = 0; + mrb_data_init(self, q, &mrb_task_queue_type); + mrb_iv_set(mrb, self, mrb_intern_lit(mrb, "@items"), mrb_ary_new(mrb)); + return self; +} + +static mrb_value +queue_push(mrb_state *mrb, mrb_value self) +{ + mrb_value obj; + mrb_get_args(mrb, "o", &obj); + + mrb_task_queue *q = (mrb_task_queue*)mrb_data_get_ptr(mrb, self, &mrb_task_queue_type); + if (!q) mrb_raise(mrb, E_ARGUMENT_ERROR, "invalid queue"); + if (q->closed) mrb_raise(mrb, queue_task_error_class(mrb), "queue closed"); + + mrb_value items = mrb_iv_get(mrb, self, mrb_intern_lit(mrb, "@items")); + mrb_ary_push(mrb, items, obj); + queue_wake_one_waiter(mrb, q); + return self; +} + +/* + * __pop_try: try to pop one item. Returns: + * - the item if available + * - nil if closed and empty + * - raises Task::Error if non_block and empty + * - Task::Queue::WAIT_RETRY sentinel if the current task was put to WAITING + * + * Ruby-level pop loops on WAIT_RETRY. + */ +static mrb_value +queue_pop_try(mrb_state *mrb, mrb_value self) +{ + mrb_bool non_block = FALSE; + mrb_get_args(mrb, "|b", &non_block); + + mrb_task_queue *q = (mrb_task_queue*)mrb_data_get_ptr(mrb, self, &mrb_task_queue_type); + if (!q) mrb_raise(mrb, E_ARGUMENT_ERROR, "invalid queue"); + + mrb_value items = mrb_iv_get(mrb, self, mrb_intern_lit(mrb, "@items")); + + /* Item available - return it */ + if (RARRAY_LEN(items) > 0) { + return mrb_ary_shift(mrb, items); + } + + /* Closed and empty */ + if (q->closed) { + return mrb_nil_value(); + } + + /* Non-blocking and empty */ + if (non_block) { + mrb_raise(mrb, queue_task_error_class(mrb), "queue empty"); + } + + /* Blocking pop only works inside a task */ + if (mrb->c == mrb->root_c) { + mrb_raise(mrb, E_RUNTIME_ERROR, "blocking pop can only be called from within a task"); + } + + /* Guard against yielding from inside a C function boundary */ + mrb_callinfo *ci; + for (ci = mrb->c->ci; ci >= mrb->c->cibase; ci--) { + if (ci->cci > 0) { + mrb_raise(mrb, E_RUNTIME_ERROR, "blocking pop cannot be called from within a C function boundary"); + } + } + + /* Move current task to WAITING */ + mrb_task *current = MRB2TASK(mrb); + mrb_task_disable_irq(); + q_delete_task(mrb, current); + current->status = MRB_TASK_STATUS_WAITING; + current->reason = MRB_TASK_REASON_QUEUE; + current->wait.queue = q; + q_insert_task(mrb, current); + mrb_task_enable_irq(); + switching_ = TRUE; + + /* Return sentinel; the Ruby pop loop will retry after wakeup */ + return wait_retry_; +} + +static mrb_value +queue_size(mrb_state *mrb, mrb_value self) +{ + mrb_value items = mrb_iv_get(mrb, self, mrb_intern_lit(mrb, "@items")); + return mrb_int_value(mrb, RARRAY_LEN(items)); +} + +static mrb_value +queue_empty_p(mrb_state *mrb, mrb_value self) +{ + mrb_value items = mrb_iv_get(mrb, self, mrb_intern_lit(mrb, "@items")); + return mrb_bool_value(RARRAY_LEN(items) == 0); +} + +static mrb_value +queue_clear(mrb_state *mrb, mrb_value self) +{ + mrb_value items = mrb_iv_get(mrb, self, mrb_intern_lit(mrb, "@items")); + mrb_ary_clear(mrb, items); + return self; +} + +static mrb_value +queue_close(mrb_state *mrb, mrb_value self) +{ + mrb_task_queue *q = (mrb_task_queue*)mrb_data_get_ptr(mrb, self, &mrb_task_queue_type); + if (!q) mrb_raise(mrb, E_ARGUMENT_ERROR, "invalid queue"); + if (!q->closed) { + q->closed = 1; + queue_wake_all_waiters(mrb, q); + } + return self; +} + +static mrb_value +queue_closed_p(mrb_state *mrb, mrb_value self) +{ + mrb_task_queue *q = (mrb_task_queue*)mrb_data_get_ptr(mrb, self, &mrb_task_queue_type); + if (!q) mrb_raise(mrb, E_ARGUMENT_ERROR, "invalid queue"); + return mrb_bool_value(q->closed); +} + +static mrb_value +queue_num_waiting(mrb_state *mrb, mrb_value self) +{ + mrb_task_queue *q = (mrb_task_queue*)mrb_data_get_ptr(mrb, self, &mrb_task_queue_type); + if (!q) mrb_raise(mrb, E_ARGUMENT_ERROR, "invalid queue"); + uint32_t count = 0; + mrb_task_disable_irq(); + mrb_task *curr = q_waiting_; + while (curr) { + if (curr->reason == MRB_TASK_REASON_QUEUE && curr->wait.queue == q) { + count++; + } + curr = curr->next; + } + mrb_task_enable_irq(); + return mrb_int_value(mrb, (mrb_int)count); +} + +void +mrb_init_task_queue(mrb_state *mrb, struct RClass *task_class) +{ + struct RClass *queue_class; + + queue_class = mrb_define_class_under_id(mrb, task_class, MRB_SYM(Queue), mrb->object_class); + MRB_SET_INSTANCE_TT(queue_class, MRB_TT_DATA); + + /* Allocate and store WAIT_RETRY sentinel (rooted by the class constant table) */ + wait_retry_ = mrb_obj_new(mrb, mrb->object_class, 0, NULL); + mrb_define_const_id(mrb, queue_class, MRB_SYM(WAIT_RETRY), wait_retry_); + + mrb_define_method_id(mrb, queue_class, MRB_SYM(initialize), queue_initialize, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM(__push), queue_push, MRB_ARGS_REQ(1)); + mrb_define_method_id(mrb, queue_class, MRB_SYM(__pop_try), queue_pop_try, MRB_ARGS_OPT(1)); + mrb_define_method_id(mrb, queue_class, MRB_SYM(size), queue_size, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM(length), queue_size, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM_Q(empty), queue_empty_p, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM(clear), queue_clear, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM(close), queue_close, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM_Q(closed), queue_closed_p, MRB_ARGS_NONE()); + mrb_define_method_id(mrb, queue_class, MRB_SYM(num_waiting), queue_num_waiting, MRB_ARGS_NONE()); +} diff --git a/mrbgems/mruby-task/test/queue.rb b/mrbgems/mruby-task/test/queue.rb new file mode 100644 index 000000000..33c5f99e2 --- /dev/null +++ b/mrbgems/mruby-task/test/queue.rb @@ -0,0 +1,163 @@ +# Task::Queue tests + +assert("Task::Queue.new creates a queue") do + q = Task::Queue.new + assert_kind_of Task::Queue, q +end + +assert("Task::Queue push and non-blocking pop return item in FIFO order") do + q = Task::Queue.new + q.push(1) + q.push(2) + q.push(3) + assert_equal 1, q.pop(true) + assert_equal 2, q.pop(true) + assert_equal 3, q.pop(true) +end + +assert("Task::Queue << alias works") do + q = Task::Queue.new + q << :a + q << :b + assert_equal :a, q.pop(true) + assert_equal :b, q.pop(true) +end + +assert("Task::Queue enq/deq aliases work") do + q = Task::Queue.new + q.enq(10) + assert_equal 10, q.deq(true) +end + +assert("Task::Queue shift alias works") do + q = Task::Queue.new + q.push(:x) + assert_equal :x, q.shift(true) +end + +assert("Task::Queue size and length") do + q = Task::Queue.new + assert_equal 0, q.size + assert_equal 0, q.length + q.push(1) + assert_equal 1, q.size + q.push(2) + assert_equal 2, q.length + q.pop(true) + assert_equal 1, q.size +end + +assert("Task::Queue empty?") do + q = Task::Queue.new + assert_true q.empty? + q.push(1) + assert_false q.empty? + q.pop(true) + assert_true q.empty? +end + +assert("Task::Queue clear") do + q = Task::Queue.new + q.push(1) + q.push(2) + q.clear + assert_true q.empty? + assert_equal 0, q.size +end + +assert("Task::Queue pop(true) raises Task::Error when empty") do + q = Task::Queue.new + assert_raise(Task::Error) { q.pop(true) } +end + +assert("Task::Queue close and closed?") do + q = Task::Queue.new + assert_false q.closed? + q.close + assert_true q.closed? +end + +assert("Task::Queue push raises Task::Error after close") do + q = Task::Queue.new + q.close + assert_raise(Task::Error) { q.push(1) } +end + +assert("Task::Queue pop(true) returns nil when closed and empty") do + q = Task::Queue.new + q.close + assert_equal nil, q.pop(true) +end + +assert("Task::Queue pops remaining items after close, then nil") do + q = Task::Queue.new + q.push(1) + q.push(2) + q.close + assert_equal 1, q.pop(true) + assert_equal 2, q.pop(true) + assert_equal nil, q.pop(true) +end + +assert("Task::Queue double close is no-op") do + q = Task::Queue.new + q.close + assert_nothing_raised { q.close } + assert_true q.closed? +end + +assert("Task::Queue num_waiting is 0 with no blocked tasks") do + q = Task::Queue.new + assert_equal 0, q.num_waiting +end + +assert("Task::Queue blocking pop wakes on push") do + q = Task::Queue.new + results = [] + + Task.new { results << q.pop } + Task.new { q.push(99) } + Task.run + + assert_equal [99], results +end + +assert("Task::Queue multiple producers and consumers") do + q = Task::Queue.new + received = [] + + Task.new { q.push(1) } + Task.new { q.push(2) } + Task.new { q.push(3) } + Task.new { received << q.pop } + Task.new { received << q.pop } + Task.new { received << q.pop } + Task.run + + assert_equal [1, 2, 3], received.sort +end + +assert("Task::Queue blocking pop returns nil when queue is closed") do + q = Task::Queue.new + results = [] + + Task.new { results << q.pop } + Task.new { q.close } + Task.run + + assert_equal [nil], results +end + +assert("Task::Queue num_waiting reflects blocked task count") do + q = Task::Queue.new + counts = [] + + Task.new { q.pop } + Task.new do + counts << q.num_waiting # consumer should be waiting + q.push(:done) + end + Task.run + + assert_equal [1], counts +end