Class: UringMachine

Inherits:
Object
  • Object
show all
Defined in:
lib/uringmachine.rb,
lib/uringmachine/actor.rb,
lib/uringmachine/version.rb,
lib/uringmachine/dns_resolver.rb,
lib/uringmachine/fiber_scheduler.rb,
ext/um/um_class.c

Overview

A UringMachine instance provides an interface for performing I/O operations and automatically switching between fibers. A single UringMachine instance should be used for each thread.

Defined Under Namespace

Modules: FiberExtensions, ThreadExtensions Classes: Actor, AsyncOp, BlockingOperationThreadPool, DNSResolver, Error, FiberScheduler, IO, Mutex, Queue, Terminate

Constant Summary collapse

TERMINATE_EXCEPTION = 
UM::Terminate.new
VERSION = 
'1.0.4'

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(*args) ⇒ Object 



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'ext/um/um_class.c', line 109

VALUE UM_initialize(int argc, VALUE *argv, VALUE self) {
  static ID kwargs_ids[3];
  struct um *machine = RTYPEDDATA_DATA(self);
  VALUE opts, kwargs[3] = {Qnil, Qnil, Qnil};

  if (!kwargs_ids[0]) {
    kwargs_ids[0] = rb_intern_const("size");
    kwargs_ids[1] = rb_intern_const("sqpoll");
    kwargs_ids[2] = rb_intern_const("sidecar");
  }
  rb_scan_args(argc, argv, "0:", &opts);
  if (!NIL_P(opts)) {
    rb_get_kwargs(opts, kwargs_ids, 0, 3, kwargs);
  }

  uint entries_i = TYPE(kwargs[0]) == T_FIXNUM ? NUM2UINT(kwargs[0]) : 0;
  uint sqpoll_timeout_msec = get_sqpoll_timeout_msec(kwargs[1]);
  um_setup(self, machine, entries_i, sqpoll_timeout_msec, RTEST(kwargs[2]));

  return self;
}

Class Method Details

.debug(str) ⇒ Object 



1327
1328
1329
1330
 
# File 'ext/um/um_class.c', line 1327

VALUE UM_debug(VALUE self, VALUE str) {
  fprintf(stderr, "%s\n", StringValueCStr(str));
  return Qnil;
}

.inotify_add_watch(fd, path, mask) ⇒ Object 



1362
1363
1364
1365
1366
1367
1368
1369
 
# File 'ext/um/um_class.c', line 1362

VALUE UM_inotify_add_watch(VALUE self, VALUE fd, VALUE path, VALUE mask) {
  int ret = inotify_add_watch(NUM2INT(fd), StringValueCStr(path), NUM2UINT(mask));
  if (ret == -1) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }
  return INT2NUM(ret);
}

.inotify_init ⇒ Object 



1341
1342
1343
1344
1345
1346
1347
1348
 
# File 'ext/um/um_class.c', line 1341

VALUE UM_inotify_init(VALUE self) {
  int fd = inotify_init();
  if (fd == -1) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }
  return INT2NUM(fd);
}

.kernel_version ⇒ Object 



1315
1316
1317
 
# File 'ext/um/um_class.c', line 1315

VALUE UM_kernel_version(VALUE self) {
  return INT2NUM(UM_KERNEL_VERSION);
}

.pidfd_open(pid) ⇒ Object 



1275
1276
1277
1278
1279
1280
1281
1282
1283
 
# File 'ext/um/um_class.c', line 1275

VALUE UM_pidfd_open(VALUE self, VALUE pid) {
  int fd = syscall(SYS_pidfd_open, NUM2INT(pid), 0);
  if (fd == -1) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }

  return INT2NUM(fd);
}

.pidfd_send_signal(fd, sig) ⇒ Object 



1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
 
# File 'ext/um/um_class.c', line 1296

VALUE UM_pidfd_send_signal(VALUE self, VALUE fd, VALUE sig) {
  int ret = syscall(
    SYS_pidfd_send_signal, NUM2INT(fd), NUM2INT(sig), NULL, 0
  );
  if (ret) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }

  return fd;
}

.pipe ⇒ Object 



1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
 
# File 'ext/um/um_class.c', line 1230

VALUE UM_pipe(VALUE self) {
  int fds[2];
  int ret = pipe(fds);
  if (ret) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }

  return rb_ary_new_from_args(2, INT2NUM(fds[0]), INT2NUM(fds[1]));
}

.pr_set_child_subreaper(set) ⇒ Object 



1422
1423
1424
1425
1426
1427
1428
1429
1430
 
# File 'ext/um/um_class.c', line 1422

VALUE UM_pr_set_child_subreaper(VALUE self, VALUE set) {
  int ret = prctl(PR_SET_CHILD_SUBREAPER, RTEST(set) ? 1 : 0);
  if (ret) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }

  return set;
}

.socketpair(domain, type, protocol) ⇒ Object 



1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
 
# File 'ext/um/um_class.c', line 1253

VALUE UM_socketpair(VALUE self, VALUE domain, VALUE type, VALUE protocol) {
  int fds[2];
  int ret = socketpair(NUM2INT(domain), NUM2INT(type), NUM2INT(protocol), fds);
  if (ret) {
    int e = errno;
    rb_syserr_fail(e, strerror(e));
  }

  return rb_ary_new_from_args(2, INT2NUM(fds[0]), INT2NUM(fds[1]));
}

Instance Method Details

#accept(server_fd) ⇒ Object 



621
622
623
624
 
# File 'ext/um/um_class.c', line 621

VALUE UM_accept(VALUE self, VALUE server_fd) {
  struct um *machine = um_get_machine(self);
  return um_accept(machine, NUM2INT(server_fd));
}

#accept_each(server_fd) ⇒ Object 



638
639
640
641
 
# File 'ext/um/um_class.c', line 638

VALUE UM_accept_each(VALUE self, VALUE server_fd) {
  struct um *machine = um_get_machine(self);
  return um_accept_each(machine, NUM2INT(server_fd));
}

#accept_into_queue(server_fd, queue) ⇒ Object 



656
657
658
659
 
# File 'ext/um/um_class.c', line 656

VALUE UM_accept_into_queue(VALUE self, VALUE server_fd, VALUE queue) {
  struct um *machine = um_get_machine(self);
  return um_accept_into_queue(machine, NUM2INT(server_fd), queue);
}

#await(*fibers) ⇒ Integer

Waits for the given fibers to terminate, without collecting their return values.

Parameters:

  • fibers (Fiber, Array<Fiber>) —

    fibers to wait for

Returns:

  • (Integer) —

    number of fibers



124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/uringmachine.rb', line 124

def await(*fibers)
  queue = Fiber.current.mailbox

  if fibers.size == 1
    first = fibers.first
    case first
    when Enumerable
      fibers = first
    when Fiber
      first = proc_spin(first) if first.is_a?(Proc)
      if !first.done?
        first.add_done_listener(queue)
        self.shift(queue)
      end
      return 1
    end
  end

  pending = nil
  fibers.each do |f|
    f = proc_spin(f) if f.is_a?(Proc)
    if !f.done?
      (pending ||= []) << f
      f.add_done_listener(queue)
    end
  end
  if pending
    while !pending.empty?
      f = self.shift(queue)
      pending.delete(f)
    end
  end
  fibers.count
end

#bind(fd, host, port) ⇒ Object 



873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
 
# File 'ext/um/um_class.c', line 873

VALUE UM_bind(VALUE self, VALUE fd, VALUE host, VALUE port) {
  struct sockaddr_in addr;
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = inet_addr(StringValueCStr(host));
  addr.sin_port = htons(NUM2INT(port));

#ifdef HAVE_IO_URING_PREP_BIND
  struct um *machine = um_get_machine(self);
  return um_bind(machine, NUM2INT(fd), (struct sockaddr *)&addr, sizeof(addr));
#else
  int res = bind(NUM2INT(fd), (struct sockaddr *)&addr, sizeof(addr));
  if (res)
    rb_syserr_fail(errno, strerror(errno));
  return INT2NUM(0);
#endif
}

#close(fd) ⇒ Object 



587
588
589
590
 
# File 'ext/um/um_class.c', line 587

VALUE UM_close(VALUE self, VALUE fd) {
  struct um *machine = um_get_machine(self);
  return um_close(machine, NUM2INT(fd));
}

#close_async(fd) ⇒ Object 



605
606
607
608
 
# File 'ext/um/um_class.c', line 605

VALUE UM_close_async(VALUE self, VALUE fd) {
  struct um *machine = um_get_machine(self);
  return um_close_async(machine, NUM2INT(fd));
}

#connect(fd, host, port) ⇒ Object 



762
763
764
765
766
767
768
769
770
771
772
 
# File 'ext/um/um_class.c', line 762

VALUE UM_connect(VALUE self, VALUE fd, VALUE host, VALUE port) {
  struct um *machine = um_get_machine(self);

  struct sockaddr_in addr;
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = inet_addr(StringValueCStr(host));
  addr.sin_port = htons(NUM2INT(port));

  return um_connect(machine, NUM2INT(fd), (struct sockaddr *)&addr, sizeof(addr));
}

#fiber_set ⇒ Object

Returns the set of running fibers.

return [Set]



17
18
19
 
# File 'lib/uringmachine.rb', line 17

def fiber_set
  @fiber_set ||= Set.new
end

#file_watch(root, mask) ⇒ void

This method returns an undefined value.

Watches for filesystem events using inotify in an infinite loop, yielding incoming events to the given block.

Parameters:

  • root (String) —

    directory to watch

  • mask (Integer) —

    event mask



175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/uringmachine.rb', line 175

def file_watch(root, mask)
  fd = UM.inotify_init
  wd_map = {}
  recursive_file_watch(fd, root, wd_map, mask)
  while true
    events = inotify_get_events(fd)
    events.each do |event|
      if event[:mask] | UM::IN_IGNORED == UM::IN_IGNORED
        wd_map.delete(event[:wd])
        next
      end
      transformed_event = transform_file_watch_event(event, wd_map)
      if event[:mask] == UM::IN_CREATE | UM::IN_ISDIR
        recursive_file_watch(fd, transformed_event[:fn], wd_map, mask)
      end
      yield transformed_event
    end
  end
ensure
  close_async(fd)
end

#fsync(fd) ⇒ Object 



571
572
573
574
 
# File 'ext/um/um_class.c', line 571

VALUE UM_fsync(VALUE self, VALUE fd) {
  struct um *machine = um_get_machine(self);
  return um_fsync(machine, NUM2INT(fd));
}

#getsockopt(fd, level, opt) ⇒ Object 



939
940
941
942
 
# File 'ext/um/um_class.c', line 939

VALUE UM_getsockopt(VALUE self, VALUE fd, VALUE level, VALUE opt) {
  struct um *machine = um_get_machine(self);
  return um_getsockopt(machine, NUM2INT(fd), NUM2INT(level), NUM2INT(opt));
}

#inotify_get_events(fd) ⇒ Object 



1403
1404
1405
1406
1407
1408
1409
1410
 
# File 'ext/um/um_class.c', line 1403

VALUE UM_inotify_get_events(VALUE self, VALUE fd) {
  struct um *machine = um_get_machine(self);

  char buf[4096] __attribute__ ((aligned(__alignof__(struct inotify_event))));

  size_t ret = um_read_raw(machine, NUM2INT(fd), buf, sizeof(buf));
  return inotify_get_events(buf, ret);
}

#io(target, mode = nil) ⇒ UringMachine::IO

call-seq:

machine.io(fd, mode = nil) -> conn
machine.io(fd, mode = nil) { |conn| }

Creates an UM::IO for the given target (fd or SSLSocket). The mode indicates the type of target and how it is read from:

  • :fd - read from the given fd using the buffer pool (default mode)
  • :socket - receive from the given socket fd using the buffer pool
  • :ssl - read from the given SSL socket

If a block is given, the block will be called with the IO instance as argument and the method will return the block's return value.

Parameters:

  • target (Integer, OpenSSL::SSL::SSLSocket) —

    fd or ssl socket

  • mode (Symbol, nil) (defaults to: nil) —

    IO mode

Returns:



214
215
216
217
218
219
220
221
 
# File 'lib/uringmachine.rb', line 214

def io(target, mode = nil)
  conn = UM::IO.new(self, target, mode)
  return conn if !block_given?

  res = yield(conn)
  conn.clear
  res
end

#join(*fibers) ⇒ Object

Waits for the given fibers to terminate, returning the return value for each given fiber. This method also accepts procs instead of fibers. When a proc is given, it is ran in a separate fiber which will be joined.

machine.join(
  -> { machine.sleep(0.01); :f1 },
  -> { machine.sleep(0.02); :f2 },
  -> { machine.sleep(0.03); :f3 }
) #=> [:f1, :f2, :f3]

values of the given fibers

Parameters:

  • *fibers (Array<Fiber, Proc>) —

    fibers @return [Array] return



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
 
# File 'lib/uringmachine.rb', line 77

def join(*fibers)
  queue = Fiber.current.mailbox

  if fibers.size == 1
    first = fibers.first
    case first
    when Enumerable
      fibers = first
    when Fiber
      first = proc_spin(first) if first.is_a?(Proc)
      if !first.done?
        first.add_done_listener(queue)
        self.shift(queue)
      end
      return first.result
    end
  end

  results = {}
  pending = nil
  fibers.each do |f|
    f = proc_spin(f) if f.is_a?(Proc)
    if f.done?
      results[f] = f.result
    else
      results[f] = nil
      (pending ||= []) << f
      f.add_done_listener(queue)
    end
  end

  if pending
    while !pending.empty?
      f = self.shift(queue)
      pending.delete(f)
      results[f] = f.result
    end
  end
  values = results.values
  fibers.size == 1 ? values.first : values
end

#listen(fd, backlog) ⇒ Object 



903
904
905
906
907
908
909
910
911
912
913
 
# File 'ext/um/um_class.c', line 903

VALUE UM_listen(VALUE self, VALUE fd, VALUE backlog) {
#ifdef HAVE_IO_URING_PREP_LISTEN
  struct um *machine = um_get_machine(self);
  return um_listen(machine, NUM2INT(fd), NUM2INT(backlog));
#else
  int res = listen(NUM2INT(fd), NUM2INT(backlog));
  if (res)
    rb_syserr_fail(errno, strerror(errno));
  return INT2NUM(0);
#endif
}

#mark(mark) ⇒ Object 



142
143
144
145
146
 
# File 'ext/um/um_class.c', line 142

VALUE UM_mark_m(VALUE self, VALUE mark) {
  struct um *machine = um_get_machine(self);
  machine->mark = NUM2UINT(mark);
  return self;
}

#metrics ⇒ Object 



153
154
155
156
 
# File 'ext/um/um_class.c', line 153

VALUE UM_metrics(VALUE self) {
  struct um *machine = um_get_machine(self);
  return um_metrics(machine, &machine->metrics);
}

#open(pathname, flags) ⇒ Object 



1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
 
# File 'ext/um/um_class.c', line 1077

VALUE UM_open(VALUE self, VALUE pathname, VALUE flags) {
  struct um *machine = um_get_machine(self);
  // TODO: take optional perm (mode) arg
  VALUE fd = um_open(machine, pathname, NUM2INT(flags), 0666);
  if (rb_block_given_p()) {
    struct um_open_ctx ctx = { self, fd };
    return rb_ensure(rb_yield, fd, UM_open_complete, (VALUE)&ctx);
  }
  else
    return fd;
}

#pending_fibers ⇒ Object 



305
306
307
308
 
# File 'ext/um/um_class.c', line 305

VALUE UM_pending_fibers(VALUE self) {
  struct um *machine = um_get_machine(self);
  return machine->pending_fibers;
}

#periodically(interval) ⇒ Object 



357
358
359
360
 
# File 'ext/um/um_class.c', line 357

VALUE UM_periodically(VALUE self, VALUE interval) {
  struct um *machine = um_get_machine(self);
  return um_periodically(machine, NUM2DBL(interval));
}

#poll(fd, mask) ⇒ Object 



1102
1103
1104
1105
 
# File 'ext/um/um_class.c', line 1102

VALUE UM_poll(VALUE self, VALUE fd, VALUE mask) {
  struct um *machine = um_get_machine(self);
  return um_poll(machine, NUM2INT(fd), NUM2UINT(mask));
}

#pop(queue) ⇒ Object 



1010
1011
1012
1013
1014
 
# File 'ext/um/um_class.c', line 1010

VALUE UM_queue_pop(VALUE self, VALUE queue) {
  struct um *machine = um_get_machine(self);
  struct um_queue *que = Queue_data(queue);
  return um_queue_pop(machine, que);
}

#prep_timeout(interval) ⇒ Object 



1168
1169
1170
1171
 
# File 'ext/um/um_class.c', line 1168

VALUE UM_prep_timeout(VALUE self, VALUE interval) {
  struct um *machine = um_get_machine(self);
  return um_prep_timeout(machine, NUM2DBL(interval));
}

#profile_mode=(value) ⇒ Object 



181
182
183
184
185
186
187
188
189
 
# File 'ext/um/um_class.c', line 181

VALUE UM_profile_mode_set(VALUE self, VALUE value) {
  struct um *machine = um_get_machine(self);
  machine->profile_mode = RTEST(value);
  if (machine->profile_mode) {
    machine->metrics.time_total_wait = 0.0;
    machine->metrics.time_last_cpu = machine->metrics.time_first_cpu = um_get_time_cpu();
  }
  return value;
}

#profile_mode? ⇒ Object 



162
163
164
165
 
# File 'ext/um/um_class.c', line 162

VALUE UM_profile_mode_p(VALUE self) {
  struct um *machine = um_get_machine(self);
  return machine->profile_mode ? Qtrue : Qfalse;
}

#push(queue, value) ⇒ Object 



993
994
995
996
997
 
# File 'ext/um/um_class.c', line 993

VALUE UM_queue_push(VALUE self, VALUE queue, VALUE value) {
  struct um *machine = um_get_machine(self);
  struct um_queue *que = Queue_data(queue);
  return um_queue_push(machine, que, value);
}

#read(*args) ⇒ Object 



381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
 
# File 'ext/um/um_class.c', line 381

VALUE UM_read(int argc, VALUE *argv, VALUE self) {
  struct um *machine = um_get_machine(self);
  VALUE fd;
  VALUE buffer;
  VALUE maxlen;
  VALUE buffer_offset;
  VALUE file_offset;
  rb_scan_args(argc, argv, "32", &fd, &buffer, &maxlen, &buffer_offset, &file_offset);

  ssize_t maxlen_i = NIL_P(maxlen) ? -1 : NUM2INT(maxlen);
  ssize_t buffer_offset_i = NIL_P(buffer_offset) ? 0 : NUM2INT(buffer_offset);
  __u64 file_offset_i = NIL_P(file_offset) ? (__u64)-1 : NUM2UINT(file_offset);

  return um_read(machine, NUM2INT(fd), buffer, maxlen_i, buffer_offset_i, file_offset_i);
}

#read_each(fd) ⇒ Object 



408
409
410
411
 
# File 'ext/um/um_class.c', line 408

VALUE UM_read_each(VALUE self, VALUE fd) {
  struct um *machine = um_get_machine(self);
  return um_read_each(machine, NUM2INT(fd));
}

#recv(fd, buffer, maxlen, flags) ⇒ Object 



838
839
840
841
 
# File 'ext/um/um_class.c', line 838

VALUE UM_recv(VALUE self, VALUE fd, VALUE buffer, VALUE maxlen, VALUE flags) {
  struct um *machine = um_get_machine(self);
  return um_recv(machine, NUM2INT(fd), buffer, NUM2INT(maxlen), NUM2INT(flags));
}

#recv_each(fd, flags) ⇒ Object 



855
856
857
858
 
# File 'ext/um/um_class.c', line 855

VALUE UM_recv_each(VALUE self, VALUE fd, VALUE flags) {
  struct um *machine = um_get_machine(self);
  return um_recv_each(machine, NUM2INT(fd), NUM2INT(flags));
}

#recv_fd(sock_fd) ⇒ Object 



744
745
746
747
 
# File 'ext/um/um_class.c', line 744

VALUE UM_recv_fd(VALUE self, VALUE sock_fd) {
  struct um *machine = um_get_machine(self);
  return um_recv_fd(machine, NUM2INT(sock_fd));
}

#resolve(hostname, type = :A) ⇒ String

Resolves a hostname to an IP address by performing a DNS query.

Parameters:

  • hostname (String) —

    hostname

  • type (Symbol) (defaults to: :A) —

    record type

Returns:

  • (String) —

    IP address



164
165
166
167
 
# File 'lib/uringmachine.rb', line 164

def resolve(hostname, type = :A)
  @resolver ||= DNSResolver.new(self)
  @resolver.resolve(hostname, type)
end

#run(fiber, &block) ⇒ Fiber

Runs the given block in the given fiber. This method is used to run fibers indirectly.

Parameters:

  • fiber (Fiber) —

    fiber

  • block (Proc) —

    block to run

Returns:



58
59
60
61
62
63
 
# File 'lib/uringmachine.rb', line 58

def run(fiber, &block)
  run_block_in_fiber(block, fiber, nil)
  self.schedule(fiber, nil)
  fiber_set << fiber
  fiber
end

#schedule(fiber, value) ⇒ Object 



317
318
319
320
321
 
# File 'ext/um/um_class.c', line 317

VALUE UM_schedule(VALUE self, VALUE fiber, VALUE value) {
  struct um *machine = um_get_machine(self);
  um_schedule(machine, fiber, value);
  return self;
}

#select(read_fds, write_fds, except_fds) ⇒ Object 



1121
1122
1123
1124
 
# File 'ext/um/um_class.c', line 1121

VALUE UM_select(VALUE self, VALUE read_fds, VALUE write_fds, VALUE except_fds) {
  struct um *machine = um_get_machine(self);
  return um_select(machine, read_fds, write_fds, except_fds);
}

#send(fd, buffer, len, flags) ⇒ Object 



790
791
792
793
 
# File 'ext/um/um_class.c', line 790

VALUE UM_send(VALUE self, VALUE fd, VALUE buffer, VALUE len, VALUE flags) {
  struct um *machine = um_get_machine(self);
  return um_send(machine, NUM2INT(fd), buffer, NUM2INT(len), NUM2INT(flags));
}

#send_fd(sock_fd, fd) ⇒ Object 



728
729
730
731
 
# File 'ext/um/um_class.c', line 728

VALUE UM_send_fd(VALUE self, VALUE sock_fd, VALUE fd) {
  struct um *machine = um_get_machine(self);
  return um_send_fd(machine, NUM2INT(sock_fd), NUM2INT(fd));
}

#sendv(*args) ⇒ Object 



809
810
811
812
813
814
815
816
817
818
819
820
821
 
# File 'ext/um/um_class.c', line 809

VALUE UM_sendv(int argc, VALUE *argv, VALUE self) {
  struct um *machine = um_get_machine(self);
  if (argc < 1)
    rb_raise(rb_eArgError, "wrong number of arguments (given 0, expected 1+)");
  int fd = NUM2INT(argv[0]);
  if (argc < 2) return INT2NUM(0);

#ifdef HAVE_IO_URING_SEND_VECTORIZED
  return um_sendv(machine, fd, argc - 1, argv + 1);
#else
  return um_writev(machine, fd, argc - 1, argv + 1);
#endif
}

#setsockopt(fd, level, opt, value) ⇒ Object 



958
959
960
961
 
# File 'ext/um/um_class.c', line 958

VALUE UM_setsockopt(VALUE self, VALUE fd, VALUE level, VALUE opt, VALUE value) {
  struct um *machine = um_get_machine(self);
  return um_setsockopt(machine, NUM2INT(fd), NUM2INT(level), NUM2INT(opt), numeric_value(value));
}

#shift(queue) ⇒ Object 



1045
1046
1047
1048
1049
 
# File 'ext/um/um_class.c', line 1045

VALUE UM_queue_shift(VALUE self, VALUE queue) {
  struct um *machine = um_get_machine(self);
  struct um_queue *que = Queue_data(queue);
  return um_queue_shift(machine, que);
}

#shutdown(fd, how) ⇒ Object 



692
693
694
695
 
# File 'ext/um/um_class.c', line 692

VALUE UM_shutdown(VALUE self, VALUE fd, VALUE how) {
  struct um *machine = um_get_machine(self);
  return um_shutdown(machine, NUM2INT(fd), NUM2INT(how));
}

#shutdown_async(fd, how) ⇒ Object 



711
712
713
714
 
# File 'ext/um/um_class.c', line 711

VALUE UM_shutdown_async(VALUE self, VALUE fd, VALUE how) {
  struct um *machine = um_get_machine(self);
  return um_shutdown_async(machine, NUM2INT(fd), NUM2INT(how));
}

#sidecar_mode? ⇒ Object 



206
207
208
209
 
# File 'ext/um/um_class.c', line 206

VALUE UM_sidecar_mode_p(VALUE self) {
  struct um *machine = um_get_machine(self);
  return machine->sidecar_mode ? Qtrue : Qfalse;
}

#sidecar_start ⇒ Object 



215
216
217
218
219
 
# File 'ext/um/um_class.c', line 215

VALUE UM_sidecar_start(VALUE self) {
  struct um *machine = um_get_machine(self);
  um_sidecar_setup(machine);
  return self;
}

#sidecar_stop ⇒ Object 



225
226
227
228
229
 
# File 'ext/um/um_class.c', line 225

VALUE UM_sidecar_stop(VALUE self) {
  struct um *machine = um_get_machine(self);
  um_sidecar_teardown(machine);
  return self;
}

#size ⇒ Object 



135
136
137
138
 
# File 'ext/um/um_class.c', line 135

VALUE UM_size(VALUE self) {
  struct um *machine = um_get_machine(self);
  return UINT2NUM(machine->size);
}

#sleep(duration) ⇒ Object 



345
346
347
348
 
# File 'ext/um/um_class.c', line 345

VALUE UM_sleep(VALUE self, VALUE duration) {
  struct um *machine = um_get_machine(self);
  return um_sleep(machine, NUM2DBL(duration));
}

#snooze ⇒ Object 



237
238
239
240
241
242
243
244
245
246
 
# File 'ext/um/um_class.c', line 237

VALUE UM_snooze(VALUE self) {
  struct um *machine = um_get_machine(self);
  um_schedule(machine, rb_fiber_current(), Qnil);

  // the current fiber is already scheduled, and the runqueue is GC-marked, so
  // we can safely call um_switch, which is faster than calling um_yield.
  VALUE ret = um_switch(machine);
  RAISE_IF_EXCEPTION(ret);
  return ret;
}

#socket(domain, type, protocol, flags) ⇒ Object 



675
676
677
678
 
# File 'ext/um/um_class.c', line 675

VALUE UM_socket(VALUE self, VALUE domain, VALUE type, VALUE protocol, VALUE flags) {
  struct um *machine = um_get_machine(self);
  return um_socket(machine, NUM2INT(domain), NUM2INT(type), NUM2INT(protocol), NUM2UINT(flags));
}

#spin(value = nil, klass = Fiber, &block) ⇒ Fiber

Creates a new fiber and schedules it to be ran.

Parameters:

  • value (any) (defaults to: nil) —

    Value to be passed to the given block

  • klass (Class) (defaults to: Fiber) —

    fiber class

  • block (Proc) —

    block to run in fiber

Returns:



31
32
33
34
35
36
37
 
# File 'lib/uringmachine.rb', line 31

def spin(value = nil, klass = Fiber, &block)
  fiber = klass.new(blocking: false) { |v| run_block_in_fiber(block, fiber, v) }
  self.schedule(fiber, value)

  fiber_set << fiber
  fiber
end

#spin_actor(mod, *a, **k) ⇒ Object 



4
5
6
7
8
9
10
11
 
# File 'lib/uringmachine/actor.rb', line 4

def spin_actor(mod, *a, **k)
  target = Object.new.extend(mod)
  mailbox = UM::Queue.new
  actor = spin(nil, Actor) { actor.run(self, target, mailbox) }
  target.setup(*a, **k)
  snooze
  actor
end

#spin_thread_actor(mod, *a, **k) ⇒ Object 



13
14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/uringmachine/actor.rb', line 13

def spin_thread_actor(mod, *a, **k)
  machine = UM.new
  target = Object.new.extend(mod)
  mailbox = UM::Queue.new
  actor = Actor.new
  Thread.new do
    actor.run(machine, target, mailbox)
  end
  target.setup(*a, **k)
  snooze
  actor
end

#splice(in_fd, out_fd, nbytes) ⇒ Object 



536
537
538
539
 
# File 'ext/um/um_class.c', line 536

VALUE UM_splice(VALUE self, VALUE in_fd, VALUE out_fd, VALUE nbytes) {
  struct um *machine = um_get_machine(self);
  return um_splice(machine, NUM2INT(in_fd), NUM2INT(out_fd), NUM2UINT(nbytes));
}

#sqpoll_mode? ⇒ Object 



171
172
173
174
 
# File 'ext/um/um_class.c', line 171

VALUE UM_sqpoll_mode_p(VALUE self) {
  struct um *machine = um_get_machine(self);
  return machine->sqpoll_mode ? Qtrue : Qfalse;
}

#ssl_read(ssl, buf, maxlen) ⇒ Object 



1198
1199
1200
1201
1202
 
# File 'ext/um/um_class.c', line 1198

VALUE UM_ssl_read(VALUE self, VALUE ssl, VALUE buf, VALUE maxlen) {
  struct um *machine = um_get_machine(self);
  int ret = um_ssl_read(machine, ssl, buf, NUM2INT(maxlen));
  return INT2NUM(ret);
}

#ssl_set_bio(ssl) ⇒ Object 



1181
1182
1183
1184
1185
 
# File 'ext/um/um_class.c', line 1181

VALUE UM_ssl_set_bio(VALUE self, VALUE ssl) {
  struct um *machine = um_get_machine(self);
  um_ssl_set_bio(machine, ssl);
  return self;
}

#ssl_write(ssl, buf, len) ⇒ Object 



1215
1216
1217
1218
1219
 
# File 'ext/um/um_class.c', line 1215

VALUE UM_ssl_write(VALUE self, VALUE ssl, VALUE buf, VALUE len) {
  struct um *machine = um_get_machine(self);
  int ret = um_ssl_write(machine, ssl, buf, NUM2INT(len));
  return INT2NUM(ret);
}

#statx(dirfd, path, flags, mask) ⇒ Object 



517
518
519
520
 
# File 'ext/um/um_class.c', line 517

VALUE UM_statx(VALUE self, VALUE dirfd, VALUE path, VALUE flags, VALUE mask) {
  struct um *machine = um_get_machine(self);
  return um_statx(machine, NUM2INT(dirfd), path, NUM2INT(flags), NUM2UINT(mask));
}

#submit ⇒ Object 



294
295
296
297
298
 
# File 'ext/um/um_class.c', line 294

VALUE UM_submit(VALUE self) {
  struct um *machine = um_get_machine(self);
  uint ret = um_submit(machine);
  return UINT2NUM(ret);
}

#switch ⇒ Object 



271
272
273
274
275
276
277
 
# File 'ext/um/um_class.c', line 271

VALUE UM_switch(VALUE self) {
  struct um *machine = um_get_machine(self);

  VALUE ret = um_switch(machine);
  RAISE_IF_EXCEPTION(ret);
  return ret;
}

#synchronize(mutex) ⇒ Object 



975
976
977
978
979
 
# File 'ext/um/um_class.c', line 975

VALUE UM_mutex_synchronize(VALUE self, VALUE mutex) {
  struct um *machine = um_get_machine(self);
  struct um_mutex *mutex_data = Mutex_data(mutex);
  return um_mutex_synchronize(machine, mutex_data);
}

#tcp_connect(host, port) ⇒ Integer

Creates and connects a TCP socket to the given host and port.

Parameters:

  • host (String) —

    host IP address

  • port (Integer) —

    TCP port

Returns:

  • (Integer) —

    socket fd



241
242
243
244
245
 
# File 'lib/uringmachine.rb', line 241

def tcp_connect(host, port)
  fd = socket(UM::AF_INET, UM::SOCK_STREAM, 0, 0)
  connect(fd, host, port)
  fd
end

#tcp_listen(host, port) ⇒ Integer

Creates, binds and sets up a TCP socket for listening on the given host and port.

Parameters:

  • host (String) —

    host IP address

  • port (Integer) —

    TCP port

Returns:

  • (Integer) —

    socket fd



229
230
231
232
233
234
 
# File 'lib/uringmachine.rb', line 229

def tcp_listen(host, port)
  fd = socket(UM::AF_INET, UM::SOCK_STREAM, 0, 0)
  bind(fd, host, port)
  listen(fd, UM::SOMAXCONN)
  fd
end

#tee(in_fd, out_fd, nbytes) ⇒ Object 



555
556
557
558
 
# File 'ext/um/um_class.c', line 555

VALUE UM_tee(VALUE self, VALUE in_fd, VALUE out_fd, VALUE nbytes) {
  struct um *machine = um_get_machine(self);
  return um_tee(machine, NUM2INT(in_fd), NUM2INT(out_fd), NUM2UINT(nbytes));
}

#terminate(*fibers) ⇒ void

This method returns an undefined value.

Terminates the given fibers by scheduling them with a UM::Terminate exception. This method does not wait for the fibers to be done.

Parameters:

  • *fibers (Array<Fiber>) —

    fibers to terminate



46
47
48
49
50
 
# File 'lib/uringmachine.rb', line 46

def terminate(*fibers)
  fibers = fibers.first if fibers.size == 1 && fibers.first.is_a?(Enumerable)

  fibers.each { schedule(it, TERMINATE_EXCEPTION) unless it.done? }
end

#test_mode=(value) ⇒ Object 



196
197
198
199
200
 
# File 'ext/um/um_class.c', line 196

VALUE UM_test_mode_set(VALUE self, VALUE value) {
  struct um *machine = um_get_machine(self);
  machine->test_mode = RTEST(value);
  return value;
}

#timeout(interval, exception) ⇒ Object 



332
333
334
335
 
# File 'ext/um/um_class.c', line 332

VALUE UM_timeout(VALUE self, VALUE interval, VALUE exception) {
  struct um *machine = um_get_machine(self);
  return um_timeout(machine, interval, exception);
}

#unshift(queue, value) ⇒ Object 



1028
1029
1030
1031
1032
 
# File 'ext/um/um_class.c', line 1028

VALUE UM_queue_unshift(VALUE self, VALUE queue, VALUE value) {
  struct um *machine = um_get_machine(self);
  struct um_queue *que = Queue_data(queue);
  return um_queue_unshift(machine, que, value);
}

#waitid(idtype, id, options) ⇒ Object 



1140
1141
1142
1143
 
# File 'ext/um/um_class.c', line 1140

VALUE UM_waitid(VALUE self, VALUE idtype, VALUE id, VALUE options) {
  struct um *machine = um_get_machine(self);
  return um_waitid(machine, NUM2INT(idtype), NUM2INT(id), NUM2INT(options));
}

#waitid_status(idtype, id, options) ⇒ Object 



1153
1154
1155
1156
 
# File 'ext/um/um_class.c', line 1153

VALUE UM_waitid_status(VALUE self, VALUE idtype, VALUE id, VALUE options) {
  struct um *machine = um_get_machine(self);
  return um_waitid_status(machine, NUM2INT(idtype), NUM2INT(id), NUM2INT(options));
}

#wakeup ⇒ Object 



285
286
287
288
 
# File 'ext/um/um_class.c', line 285

VALUE UM_wakeup(VALUE self) {
  struct um *machine = um_get_machine(self);
  return um_wakeup(machine);
}

#write(*args) ⇒ Object 



428
429
430
431
432
433
434
435
436
437
438
439
440
 
# File 'ext/um/um_class.c', line 428

VALUE UM_write(int argc, VALUE *argv, VALUE self) {
  struct um *machine = um_get_machine(self);
  VALUE fd;
  VALUE buffer;
  VALUE len;
  VALUE file_offset;
  rb_scan_args(argc, argv, "22", &fd, &buffer, &len, &file_offset);

  size_t len_i = NIL_P(len) ? (size_t)-1 : NUM2UINT(len);
  __u64 file_offset_i = NIL_P(file_offset) ? (__u64)-1 : NUM2UINT(file_offset);

  return um_write(machine, NUM2INT(fd), buffer, len_i, file_offset_i);
}

#write_async(*args) ⇒ Object 



487
488
489
490
491
492
493
494
495
496
497
498
499
 
# File 'ext/um/um_class.c', line 487

VALUE UM_write_async(int argc, VALUE *argv, VALUE self) {
  struct um *machine = um_get_machine(self);
  VALUE fd;
  VALUE buffer;
  VALUE len;
  VALUE file_offset;
  rb_scan_args(argc, argv, "22", &fd, &buffer, &len, &file_offset);

  size_t len_i = NIL_P(len) ? (size_t)-1 : NUM2UINT(len);
  __u64 file_offset_i = NIL_P(file_offset) ? (__u64)-1 : NUM2UINT(file_offset);

  return um_write_async(machine, NUM2INT(fd), buffer, len_i, file_offset_i);
}

#writev(*args) ⇒ Object 



459
460
461
462
463
464
465
466
467
 
# File 'ext/um/um_class.c', line 459

VALUE UM_writev(int argc, VALUE *argv, VALUE self) {
  struct um *machine = um_get_machine(self);
  if (argc < 1)
    rb_raise(rb_eArgError, "wrong number of arguments (given 0, expected 1+)");
  int fd = NUM2INT(argv[0]);
  if (argc < 2) return INT2NUM(0);

  return um_writev(machine, fd, argc - 1, argv + 1);
}

#yield ⇒ Object 



255
256
257
258
259
260
261
 
# File 'ext/um/um_class.c', line 255

VALUE UM_yield(VALUE self) {
  struct um *machine = um_get_machine(self);

  VALUE ret = um_yield(machine);
  RAISE_IF_EXCEPTION(ret);
  return ret;
}