Class: Event::Selector::URing

Inherits:
Object
  • Object
show all
Defined in:
ext/event/selector/uring.c

Instance Method Summary collapse

Constructor Details

#initialize(loop) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'ext/event/selector/uring.c', line 93

VALUE Event_Selector_URing_initialize(VALUE self, VALUE loop) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  Event_Selector_initialize(&data->backend, loop);
  int result = io_uring_queue_init(URING_ENTRIES, &data->ring, 0);
  
  if (result < 0) {
    rb_syserr_fail(-result, "io_uring_queue_init");
  }
  
  rb_update_max_fd(data->ring.ring_fd);
  
  return self;
}

Instance Method Details

#close ⇒ Object 



109
110
111
112
113
114
115
116
 
# File 'ext/event/selector/uring.c', line 109

VALUE Event_Selector_URing_close(VALUE self) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  close_internal(data);
  
  return Qnil;
}

#io_close(io) ⇒ Object 



469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
 
# File 'ext/event/selector/uring.c', line 469

VALUE Event_Selector_URing_io_close(VALUE self, VALUE io) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  int descriptor = Event_Selector_io_descriptor(io);

  if (ASYNC_CLOSE) {
    struct io_uring_sqe *sqe = io_get_sqe(data);
    
    io_uring_prep_close(sqe, descriptor);
    io_uring_sqe_set_data(sqe, NULL);
    io_uring_submit_now(data);
  } else {
    close(descriptor);
  }

  // We don't wait for the result of close since it has no use in pratice:
  return Qtrue;
}

#io_read(fiber, io, buffer, _length) ⇒ Object 



382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
 
# File 'ext/event/selector/uring.c', line 382

VALUE Event_Selector_URing_io_read(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _length) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  int descriptor = Event_Selector_io_descriptor(io);
  
  void *base;
  size_t size;
  rb_io_buffer_get_mutable(buffer, &base, &size);
  
  size_t offset = 0;
  size_t length = NUM2SIZET(_length);
  
  while (length > 0) {
    size_t maximum_size = size - offset;
    int result = io_read(data, fiber, descriptor, (char*)base+offset, maximum_size);
    
    if (result == 0) {
      break;
    } else if (result > 0) {
      offset += result;
      if ((size_t)result >= length) break;
      length -= result;
    } else if (-result == EAGAIN || -result == EWOULDBLOCK) {
      Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(EVENT_READABLE));
    } else {
      rb_syserr_fail(-result, strerror(-result));
    }
  }
  
  return SIZET2NUM(offset);
}

#io_wait(fiber, io, events) ⇒ Object 



339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
 
# File 'ext/event/selector/uring.c', line 339

VALUE Event_Selector_URing_io_wait(VALUE self, VALUE fiber, VALUE io, VALUE events) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  int descriptor = Event_Selector_io_descriptor(io);
  struct io_uring_sqe *sqe = io_get_sqe(data);
  
  short flags = poll_flags_from_events(NUM2INT(events));
  
  if (DEBUG) fprintf(stderr, "Event_Selector_URing_io_wait:io_uring_prep_poll_add(descriptor=%d, flags=%d, fiber=%p)\n", descriptor, flags, (void*)fiber);
  
  io_uring_prep_poll_add(sqe, descriptor, flags);
  io_uring_sqe_set_data(sqe, (void*)fiber);
  
  // If we are going to wait, we assume that we are waiting for a while:
  io_uring_submit_pending(data);
  
  struct io_wait_arguments io_wait_arguments = {
    .data = data,
    .fiber = fiber,
    .flags = flags
  };
  
  return rb_rescue(io_wait_transfer, (VALUE)&io_wait_arguments, io_wait_rescue, (VALUE)&io_wait_arguments);
}

#io_write(fiber, io, buffer, _length) ⇒ Object 



431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
 
# File 'ext/event/selector/uring.c', line 431

VALUE Event_Selector_URing_io_write(VALUE self, VALUE fiber, VALUE io, VALUE buffer, VALUE _length) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  int descriptor = Event_Selector_io_descriptor(io);
  
  const void *base;
  size_t size;
  rb_io_buffer_get_immutable(buffer, &base, &size);
  
  size_t offset = 0;
  size_t length = NUM2SIZET(_length);
  
  if (length > size) {
    rb_raise(rb_eRuntimeError, "Length exceeds size of buffer!");
  }
  
  while (length > 0) {
    int result = io_write(data, fiber, descriptor, (char*)base+offset, length);
    
    if (result >= 0) {
      offset += result;
      if ((size_t)result >= length) break;
      length -= result;
    } else if (-result == EAGAIN || -result == EWOULDBLOCK) {
      Event_Selector_URing_io_wait(self, fiber, io, RB_INT2NUM(EVENT_WRITABLE));
    } else {
      rb_syserr_fail(-result, strerror(-result));
    }
  }
  
  return SIZET2NUM(offset);
}

#process_wait(fiber, pid, flags) ⇒ Object 



255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
 
# File 'ext/event/selector/uring.c', line 255

VALUE Event_Selector_URing_process_wait(VALUE self, VALUE fiber, VALUE pid, VALUE flags) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  struct process_wait_arguments process_wait_arguments = {
    .data = data,
    .pid = NUM2PIDT(pid),
    .flags = NUM2INT(flags),
  };
  
  process_wait_arguments.descriptor = pidfd_open(process_wait_arguments.pid, 0);
  rb_update_max_fd(process_wait_arguments.descriptor);
  
  struct io_uring_sqe *sqe = io_get_sqe(data);

  if (DEBUG) fprintf(stderr, "Event_Selector_URing_process_wait:io_uring_prep_poll_add(%p)\n", (void*)fiber);
  io_uring_prep_poll_add(sqe, process_wait_arguments.descriptor, POLLIN|POLLHUP|POLLERR);
  io_uring_sqe_set_data(sqe, (void*)fiber);
  io_uring_submit_pending(data);

  return rb_ensure(process_wait_transfer, (VALUE)&process_wait_arguments, process_wait_ensure, (VALUE)&process_wait_arguments);
}

#push(fiber) ⇒ Object 



148
149
150
151
152
153
154
155
156
 
# File 'ext/event/selector/uring.c', line 148

VALUE Event_Selector_URing_push(VALUE self, VALUE fiber)
{
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  Event_Selector_queue_push(&data->backend, fiber);
  
  return Qnil;
}

#raise(*args) ⇒ Object 



158
159
160
161
162
163
164
 
# File 'ext/event/selector/uring.c', line 158

VALUE Event_Selector_URing_raise(int argc, VALUE *argv, VALUE self)
{
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  return Event_Selector_wait_and_raise(&data->backend, argc, argv);
}

#ready? ⇒ Boolean

Returns:

  • (Boolean) 


166
167
168
169
170
171
 
# File 'ext/event/selector/uring.c', line 166

VALUE Event_Selector_URing_ready_p(VALUE self) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  return data->backend.ready ? Qtrue : Qfalse;
}

#resume(*args) ⇒ Object 



128
129
130
131
132
133
134
135
136
 
# File 'ext/event/selector/uring.c', line 128

VALUE Event_Selector_URing_resume(int argc, VALUE *argv, VALUE self)
{
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  Event_Selector_resume(&data->backend, argc, argv);
  
  return Qnil;
}

#select(duration) ⇒ Object 



589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
 
# File 'ext/event/selector/uring.c', line 589

VALUE Event_Selector_URing_select(VALUE self, VALUE duration) {
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  int ready = Event_Selector_queue_flush(&data->backend);
  
  int result = select_process_completions(&data->ring);

  // If the ready list was empty and we didn't process any completions:
  if (!ready && result == 0) {
    // We might need to wait for events:
    struct select_arguments arguments = {
      .data = data,
      .timeout = NULL,
    };
    
    arguments.timeout = make_timeout(duration, &arguments.storage);
    
    if (!data->backend.ready && !timeout_nonblocking(arguments.timeout)) {
      // This is a blocking operation, we wait for events:
      result = select_internal_without_gvl(&arguments);
    } else {
      // The timeout specified required "nonblocking" behaviour so we just flush the SQ if required:
      io_uring_submit_flush(data);
    }

    // After waiting/flushing the SQ, check if there are any completions:
    result = select_process_completions(&data->ring);
  }
  
  return RB_INT2NUM(result);
}

#transfer ⇒ Object 



118
119
120
121
122
123
124
125
126
 
# File 'ext/event/selector/uring.c', line 118

VALUE Event_Selector_URing_transfer(VALUE self)
{
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  Event_Selector_fiber_transfer(data->backend.loop, 0, NULL);
  
  return Qnil;
}

#yield ⇒ Object 



138
139
140
141
142
143
144
145
146
 
# File 'ext/event/selector/uring.c', line 138

VALUE Event_Selector_URing_yield(VALUE self)
{
  struct Event_Selector_URing *data = NULL;
  TypedData_Get_Struct(self, struct Event_Selector_URing, &Event_Selector_URing_Type, data);
  
  Event_Selector_yield(&data->backend);
  
  return Qnil;
}