Class: Thread::SizedQueue

Inherits:
Object show all
Defined in:
thread_sync.c,
thread_sync.c

Overview

This class represents queues of specified size capacity. The push operation may be blocked if the capacity is full.

See Thread::Queue for an example of how a Thread::SizedQueue works.

Instance Method Summary collapse

Constructor Details

#new(max) ⇒ Object

Creates a fixed-length queue with a maximum size of max.



1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
 
# File 'thread_sync.c', line 1213

static VALUE
rb_szqueue_initialize(VALUE self, VALUE vmax)
{
    long max;
    struct rb_szqueue *sq = szqueue_ptr(self);

    max = NUM2LONG(vmax);
    if (max <= 0) {
        rb_raise(rb_eArgError, "queue size must be positive");
    }

    RB_OBJ_WRITE(self, szqueue_list(sq), ary_buf_new());
    ccan_list_head_init(szqueue_waitq(sq));
    ccan_list_head_init(szqueue_pushq(sq));
    sq->max = max;

    return self;
}

Instance Method Details

#clear ⇒ Object

Removes all objects from the queue.



1365
1366
1367
1368
1369
1370
1371
1372
1373
 
# File 'thread_sync.c', line 1365

static VALUE
rb_szqueue_clear(VALUE self)
{
    struct rb_szqueue *sq = szqueue_ptr(self);

    rb_ary_clear(check_array(self, sq->q.que));
    wakeup_all(szqueue_pushq(sq));
    return self;
}

#close ⇒ Object

Similar to Thread::Queue#close.

The difference is behavior with waiting enqueuing threads.

If there are waiting enqueuing threads, they are interrupted by raising ClosedQueueError(‘queue closed’).



1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
 
# File 'thread_sync.c', line 1244

static VALUE
rb_szqueue_close(VALUE self)
{
    if (!queue_closed_p(self)) {
        struct rb_szqueue *sq = szqueue_ptr(self);

        FL_SET(self, QUEUE_CLOSED);
        wakeup_all(szqueue_waitq(sq));
        wakeup_all(szqueue_pushq(sq));
    }
    return self;
}

#empty? ⇒ Boolean

Returns true if the queue is empty.

Returns:

  • (Boolean) 


1413
1414
1415
1416
1417
1418
1419
 
# File 'thread_sync.c', line 1413

static VALUE
rb_szqueue_empty_p(VALUE self)
{
    struct rb_szqueue *sq = szqueue_ptr(self);

    return RBOOL(queue_length(self, &sq->q) == 0);
}

#length ⇒ Object #size ⇒ Object Also known as: size

Returns the length of the queue.



1384
1385
1386
1387
1388
1389
1390
 
# File 'thread_sync.c', line 1384

static VALUE
rb_szqueue_length(VALUE self)
{
    struct rb_szqueue *sq = szqueue_ptr(self);

    return LONG2NUM(queue_length(self, &sq->q));
}

#max ⇒ Object

Returns the maximum size of the queue.



1263
1264
1265
1266
1267
 
# File 'thread_sync.c', line 1263

static VALUE
rb_szqueue_max_get(VALUE self)
{
    return LONG2NUM(szqueue_ptr(self)->max);
}

#max=(number) ⇒ Object

Sets the maximum size of the queue to the given number.



1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
 
# File 'thread_sync.c', line 1276

static VALUE
rb_szqueue_max_set(VALUE self, VALUE vmax)
{
    long max = NUM2LONG(vmax);
    long diff = 0;
    struct rb_szqueue *sq = szqueue_ptr(self);

    if (max <= 0) {
        rb_raise(rb_eArgError, "queue size must be positive");
    }
    if (max > sq->max) {
        diff = max - sq->max;
    }
    sq->max = max;
    sync_wakeup(szqueue_pushq(sq), diff);
    return vmax;
}

#num_waiting ⇒ Object

Returns the number of threads waiting on the queue.



1398
1399
1400
1401
1402
1403
1404
 
# File 'thread_sync.c', line 1398

static VALUE
rb_szqueue_num_waiting(VALUE self)
{
    struct rb_szqueue *sq = szqueue_ptr(self);

    return INT2NUM(sq->q.num_waiting + sq->num_waiting_push);
}