static VALUE
rb_szqueue_initialize(VALUE self, VALUE vmax)
{
long max;
max = NUM2LONG(vmax);
if (max <= 0) {
rb_raise(rb_eArgError, "queue size must be positive");
}
RSTRUCT_SET(self, QUEUE_QUE, ary_buf_new());
RSTRUCT_SET(self, QUEUE_WAITERS, ary_buf_new());
RSTRUCT_SET(self, SZQUEUE_WAITERS, ary_buf_new());
RSTRUCT_SET(self, SZQUEUE_MAX, vmax);
return self;
}
Creates a fixed-length queue with a maximum size of max.
static VALUE
rb_szqueue_clear(VALUE self)
{
rb_ary_clear(GET_QUEUE_QUE(self));
wakeup_all_threads(GET_SZQUEUE_WAITERS(self));
return self;
}
Removes all objects from the queue.
static VALUE
rb_szqueue_close(VALUE self)
{
return queue_do_close(self, TRUE);
}
Similar to Queue#close.
The difference is behavior with waiting enqueuing threads.
If there are waiting enqueuing threads, they are interrupted by raising ClosedQueueError(‘queue closed’).
static VALUE
rb_szqueue_max_get(VALUE self)
{
return GET_SZQUEUE_MAX(self);
}
Returns the maximum size of the queue.
static VALUE
rb_szqueue_max_set(VALUE self, VALUE vmax)
{
long max = NUM2LONG(vmax), diff = 0;
VALUE t;
if (max <= 0) {
rb_raise(rb_eArgError, "queue size must be positive");
}
if ((unsigned long)max > GET_SZQUEUE_ULONGMAX(self)) {
diff = max - GET_SZQUEUE_ULONGMAX(self);
}
RSTRUCT_SET(self, SZQUEUE_MAX, vmax);
while (diff-- > 0 && !NIL_P(t = rb_ary_shift(GET_SZQUEUE_WAITERS(self)))) {
rb_thread_wakeup_alive(t);
}
return vmax;
}
Sets the maximum size of the queue to the given number.
static VALUE
rb_szqueue_num_waiting(VALUE self)
{
long len = queue_num_waiting(self) + szqueue_num_waiting_producer(self);
return ULONG2NUM(len);
}
Returns the number of threads waiting on the queue.
static VALUE
rb_szqueue_pop(int argc, VALUE *argv, VALUE self)
{
int should_block = queue_pop_should_block(argc, argv);
return szqueue_do_pop(self, should_block);
}
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_szqueue_push(int argc, VALUE *argv, VALUE self)
{
struct waiting_delete args;
int should_block = szqueue_push_should_block(argc, argv);
args.waiting = GET_SZQUEUE_WAITERS(self);
args.th = rb_thread_current();
while (queue_length(self) >= GET_SZQUEUE_ULONGMAX(self)) {
if (!should_block) {
rb_raise(rb_eThreadError, "queue full");
}
else if (queue_closed_p(self)) {
goto closed;
}
else {
rb_ary_push(args.waiting, args.th);
rb_ensure((VALUE (*)())rb_thread_sleep_deadly, (VALUE)0, queue_delete_from_waiting, (VALUE)&args);
}
}
if (queue_closed_p(self)) {
closed:
raise_closed_queue_error(self);
}
return queue_do_push(self, argv[0]);
}
Pushes object to the queue.
If there is no space left in the queue, waits until space becomes available, unless non_block is true. If non_block is true, the thread isn’t suspended, and an exception is raised.