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.0'

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size: 4096, sqpoll: false, sidecar: false) ⇒ void

Initializes a new UringMachine instance with the given options.

Parameters:

  • size (Integer) (defaults to: 4096) —

    SQ size (default: 4096)

  • sqpoll (bool, Number) (defaults to: false) —

    Set SQPOLL mode, SQPOLL timeout

  • sidecar (bool) (defaults to: false) —

    Set sidecar mode



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) ⇒ void

This method returns an undefined value.

Prints the given string to STDERR.

Parameters:

  • str (String) —

    debug message



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) ⇒ Integer

Adds a watch on the given inotify file descriptor.

Parameters:

  • fd (Integer) —

    inotify file descriptor

  • path (String) —

    file/directory path

  • mask (Integer) —

    inotify event mask

Returns:

  • (Integer) —

    watch descriptor



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 ⇒ Integer

Creates an inotify file descriptor.

Returns:

  • (Integer) —

    file descriptor



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 ⇒ Integer

Returns the kernel version.

Returns:

  • (Integer) —

    kernel version



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) ⇒ Integer

Creates a file descriptor representing the given process pid. The file descriptor can then be used with methods such as ‘#waitid` or `.pidfd_open`.

Parameters:

  • pid (Integer) —

    process pid

Returns:

  • (Integer) —

    file descriptor



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) ⇒ Integer

Sends a signal to a pidfd.

Parameters:

  • fd (Integer) —

    pidfd

  • sig (Integer) —

    signal

Returns:

  • (Integer) —

    pidfd



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 ⇒ Array

Creates a pipe, returning the file descriptors for the read and write ends.

Returns:

  • (Array)

Returns:

  • (Array<Integer>) —

    array containing the read and write file descriptors



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(vaule) ⇒ bool

Sets/unsets the “child subreaper” attribute of the calling process.

Parameters:

  • value (bool) —

    set/unset value

Returns:

  • (bool) —

    set/unset value



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) ⇒ Array

Creates a pair of connected sockets, returning the two file descriptors.

Returns:

  • (Array)

Parameters:

  • domain (Integer) —

    domain

  • type (Integer) —

    type

  • protocol (Integer) —

    protocol

Returns:

  • (Array<Integer>) —

    array containing the two file descriptors



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) ⇒ Integer

Accepts an incoming TCP connection.

Parameters:

  • server_fd (Integer) —

    listening socket file descriptor

Returns:

  • (Integer) —

    connection file descriptor



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) {|connection_fd| ... } ⇒ void

This method returns an undefined value.

Repeatedly accepts incoming TCP connections in a loop, yielding the connection file descriptors to the given block.

Yields:

  • (connection_fd)

Parameters:

  • server_fd (Integer) —

    listening socket file descriptor



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) ⇒ void

This method returns an undefined value.

Repeatedly accepts incoming TCP connections in a loop, pushing the connection file descriptors into the given queue.

Parameters:

  • server_fd (Integer) —

    listening socket file descriptor

  • queue (UM::Queue) —

    connection queue



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) ⇒ 0

Binds the given socket to the given host and port.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

  • host (String) —

    hostname or IP address

  • port (Integer) —

    port number

Returns:

  • (0) —

    success



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) ⇒ 0

Closes the given file descriptor.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

Returns:

  • (0) —

    success



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) ⇒ Hash

Closes the given file descriptor. This method submits the operation but does not wait for it to complete. This method may be used to improve performance in cases where the application des not care whether it succeeds or not.

Returns:

  • (Hash)

Parameters:

  • fd (Integer) —

    file descriptor

Returns:

  • (Integer) —

    file descriptor



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) ⇒ 0

Connects the given socket to the given host and port.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

  • host (String) —

    hostname or IP address

  • port (Integer) —

    port number

Returns:

  • (0) —

    success



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) ⇒ 0

Flushes all modified file data to the storage device.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    fd

Returns:

  • (Integer) —

    0 if successful



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) ⇒ Integer

Returns the value of a socket option.

Parameters:

  • fd (Integer) —

    file descriptor

  • level (Integer) —

    level

  • opt (Integer) —

    level

Returns:

  • (Integer) —

    option value



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) ⇒ Array<Hash>

Waits for and returns one or more events on the given inotify file descriptor. Each event is returned as a hash containing the watch descriptor, the file name, and the event mask.

Parameters:

  • fd (Integer) —

    inotify file descriptor

Returns:

  • (Array<Hash>) —

    array of one or more events



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<any>] 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) ⇒ 0

Starts listening for incoming connections on the given socket.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

  • backlog (String) —

    pending connection queue length

Returns:

  • (0) —

    success



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

:nodoc:



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 ⇒ Hash

Returns a hash with different metrics about the functioning of the UringMachine instance.

Returns:

  • (Hash) —

    metrics hash



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) ⇒ Integer #open(pathname, flags) {|fd| ... } ⇒ Integer

Opens a file using the given pathname and flags. If a block is given, the file descriptor is passed to the block, and the file is automatically closed when the block returns.

Overloads:

  • #open(pathname, flags) {|fd| ... } ⇒ Integer

    Yields:

    • (fd)

Parameters:

  • pathname (String) —

    file path

  • flags (Integer) —

    flags mask

Returns:

  • (Integer) —

    fd



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 ⇒ Set

Returns a set containing all fibers in pending state (i.e. waiting for an operation to complete.)

Returns:

  • (Set) —

    set of pending fibers



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) ⇒ void

This method returns an undefined value.

Runs the given block at regular time intervals in an infinite loop.

Parameters:

  • interval (Number) —

    time interval (in seconds) between consecutive invocations



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) ⇒ Integer

Waits for readiness of the given file descriptor according to the given event mask.

Parameters:

  • fd (Integer) —

    file descriptor

  • mask (Integer) —

    events mask

Returns:

  • (Integer) —

    fd



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) ⇒ any

removes a value from the tail of the given queue.

Parameters:

Returns:

  • (any) —

    value



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) ⇒ UM::AsyncOp

Prepares an asynchronous timeout instance.

Parameters:

  • interval (Integer) —

    timeout interval in seconds

Returns:



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) ⇒ bool

Sets/resets profile mode.

Parameters:

  • value (bool) —

    profile mode on/off

Returns:

  • (bool) —

    profile mode on/off



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? ⇒ bool

Returns the profile mode state.

Returns:

  • (bool) —

    profile mode on/off



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) ⇒ UM::Queue

Pushes a value to the tail of the given queue.

Parameters:

  • queue (UM::Queue) —

    queue

  • value (any) —

    value

Returns:



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(fd, buffer, maxlen, buffer_offset = nil, file_offset = nil) ⇒ Integer

Reads up to ‘maxlen` bytes from the given `fd` into the given buffer. The optional `buffer_offset` parameter determines the position in the buffer into which the data will be read. A negative `buffer_offset` denotes a position relative to the end of the buffer, e.g. a value of `-1` means the data will be appended to the buffer.

Parameters:

  • fd (Integer) —

    file descriptor

  • buffer (String, IO::Buffer) —

    buffer

  • maxlen (Integer) —

    maximum number of bytes to read

  • buffer_offset (Integer) —

    optional buffer offset to read into

  • file_offset (Integer) —

    optional file offset to read from

Returns:

  • (Integer) —

    number of bytes read



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) {|data| ... } ⇒ void

This method returns an undefined value.

Reads repeatedly from the given fd. Read data is yielded in an infinite loop to the given block.

Yields:

  • (data)

Parameters:

  • fd (Integer) —

    file descriptor



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) ⇒ Integer

Receives data from the given socket.

Parameters:

  • fd (Integer) —

    file descriptor

  • buffer (String, IO::Buffer) —

    buffer

  • maxlen (Integer) —

    maximum number of bytes to receive

  • flags (Integer) —

    flags mask

Returns:

  • (Integer) —

    number of bytes received



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) {|data| ... } ⇒ void

This method returns an undefined value.

Repeatedlty receives data from the given socket in an infinite loop.

Yields:

  • (data)

Parameters:

  • fd (Integer) —

    file descriptor

  • flags (Integer) —

    flags mask



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) ⇒ Integer

Receives a file descriptor over the given socket.

Parameters:

  • sock_fd (Integer) —

    socket file descriptor

Returns:

  • (Integer) —

    rececived file descriptor



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) ⇒ UringMachine

Schedules the given fiber by adding it to the runqueue. The fiber will be resumed with the given value.

Parameters:

  • fiber (Fiber) —

    fiber to schedule

  • value (any) —

    resume value

Returns:



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) ⇒ Array

Waits for readyness of at least one fd for read, write or exception condition from the given fds. This method provides a similar interface to ‘IO#select`.

Returns:

  • (Array)

Parameters:

  • read_fds (Array<Integer>) —

    file descriptors for reading

  • write_fds (Array<Integer>) —

    file descriptors for writing

  • except_fds (Array<Integer>) —

    file descriptors for exception

Returns:

  • (Array<Array<Integer>>) —

    read-ready, write-ready, and exception-ready fds



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) ⇒ Integer

Sends data on the given socket. This method is not guaranteed to send all of the data in the buffer, unless ‘UM::MSG_WAITALL` is specified in the flags mask.

Parameters:

  • fd (Integer) —

    file descriptor

  • buffer (String, IO::Buffer) —

    buffer

  • len (Integer) —

    number of bytes to send

  • flags (Integer) —

    flags mask

Returns:

  • (Integer) —

    number of bytes sent



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) ⇒ Integer

Sends the given file descriptor over the given socket.

Parameters:

  • sock_fd (Integer) —

    socket file descriptor

  • fd (Integer) —

    file descriptor to send

Returns:

  • (Integer) —

    file descriptor to send



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(fd, *buffers) ⇒ Integer #sendv(fd, *buffers) ⇒ Object

Sends data on the given socket from the given buffers. This method is only available on Linux kernel >= 6.17. This method is guaranteed to send

Overloads:

  • #sendv(fd, *buffers) ⇒ Integer

    Returns number of bytes sent.

    Parameters:

    • fd (Integer) —

      file descriptor

    • *buffers (Array<String, IO::Buffer>) —

      buffers

    Returns:

    • (Integer) —

      number of bytes sent



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) ⇒ 0

Sets the value of a socket option.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

  • level (Integer) —

    level

  • opt (Integer) —

    level

  • value (Integer) —

    option value

Returns:

  • (0) —

    success



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));
}

#pop(queue) ⇒ any

removes a value from the head of the given queue.

Parameters:

Returns:

  • (any) —

    value



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) ⇒ 0

Shuts down a socket for sending and/or receiving.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

  • how (Integer) —

    how the socket should be shutdown

Returns:

  • (0) —

    success



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) ⇒ 0

Shuts down a socket for sending and/or receiving. This method may be used to improve performance in situations where the application does not care about whether the operation succeeds or not.

Returns:

  • (0)

Parameters:

  • fd (Integer) —

    file descriptor

  • how (Integer) —

    how the socket should be shutdown

Returns:

  • (0) —

    success



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? ⇒ bool

Returns the sidecar mode state.

Returns:

  • (bool) —

    sidecar mode on/off



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 ⇒ UringMachine

Starts sidecar mode.

Returns:



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 ⇒ UringMachine

Stops sidecar mode.

Returns:



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 ⇒ Integer

Returns the SQ (submission queue) size.

Returns:

  • (Integer) —

    SQ size



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) ⇒ void

This method returns an undefined value.

Puts the current fiber to sleep for the given time duration (in seconds), yielding control to the next fiber in the runqueue.

Parameters:

  • duration (Number) —

    sleep duration in seconds



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 ⇒ UringMachine

Adds the current fiber to the end of the runqueue and yields control to the next fiber in the runqueue. This method is usually used to yield control while performing CPU-intensive work in order not starve other fibers.

Returns:



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) ⇒ Integer

Creates a socket.

Parameters:

  • domain (Integer) —

    socket domain

  • type (Integer) —

    socket type

  • protocol (Integer) —

    socket protocol

  • flags (Integer) —

    flags

Returns:

  • (Integer) —

    file descriptor



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) ⇒ Integer

Splices bytes from in_fd to out_fd. At least one of the given fds must be a pipe.

Parameters:

  • in_fd (Integer) —

    fd to splice from

  • out_fd (Integer) —

    fd to splice to

  • nbytes (Integer) —

    number of bytes to splice

Returns:

  • (Integer) —

    number of bytes spliced



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? ⇒ bool

Returns the SQPOLL mode state.

Returns:

  • (bool) —

    SQPOLL mode on/off



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) ⇒ Integer

Reads from the given SSL socket. This method should be used after first having called ‘#ssl_set_bio`.

Parameters:

  • ssl (OpenSSL::SSL::SSLSocket) —

    ssl socket

  • buf (String, IO::Buffer) —

    buffer

  • maxlen (Integer) —

    maximum number of bytes to read

Returns:

  • (Integer) —

    number of bytes read



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) ⇒ UringMachine

Sets up the given ssl socket to use the machine for sending and receiving.

Parameters:

  • ssl (OpenSSL::SSL::SSLSocket) —

    ssl socket

Returns:



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) ⇒ Integer

Writes to the given SSL socket. This method should be used after first having called ‘#ssl_set_bio`.

Parameters:

  • ssl (OpenSSL::SSL::SSLSocket) —

    ssl socket

  • buf (String, IO::Buffer) —

    buffer

  • len (Integer) —

    number of bytes to write

Returns:

  • (Integer) —

    number of bytes written



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(fd, nil, flags, mask) ⇒ Hash #statx(dirfd, path, flags, mask) ⇒ Hash #statx(UM: :AT_FDCWD, path, flags, mask) ⇒ Hash

Returns information about a file.

Overloads:

  • #statx(fd, nil, flags, mask) ⇒ Hash

    Returns:

    • (Hash)
  • #statx(dirfd, path, flags, mask) ⇒ Hash

    Returns:

    • (Hash)
  • #statx(UM: :AT_FDCWD, path, flags, mask) ⇒ Hash

    Returns:

    • (Hash)

Parameters:

  • dirfd (Integer) —

    file or directory descriptor

  • path (String, nil) —

    file path

  • flags (Integer) —

    flags

  • mask (Integer) —

    mask of information to return

Returns:

  • (hash) —

    hash containing file information



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 ⇒ Integer

Submits any pending I/O operations that are not yet submitted.

Returns:

  • (Integer) —

    number of I/O operations submitted



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 ⇒ any

Yields control to the next fiber in the runqueue. The current fiber will not be resumed unless it is scheduled again by some other fiber. The call to ‘#yield` will return the value with which the current fiber will be eventually scheduled. If resumed with an exception, that exception will be raised when the fiber is resumed.

Returns:

  • (any) —

    resume value



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) { ... } ⇒ any

Synchronizes access to the given mutex. The mutex is locked, the given block is executed and finally the mutex is unlocked.

Yields:

Parameters:

Returns:

  • (any) —

    block return value



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) ⇒ Integer

Duplicates bytes from in_fd to out_fd. At least one of the given fds must be a pipe.

Parameters:

  • in_fd (Integer) —

    fd to copy from

  • out_fd (Integer) —

    fd to copy to

  • nbytes (Integer) —

    number of bytes to duplicate

Returns:

  • (Integer) —

    number of bytes duplicated



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) ⇒ bool

Sets/resets test mode.

Parameters:

  • value (bool) —

    test mode on/off

Returns:

  • (bool) —

    test mode on/off



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) ⇒ any

Runs the given block, interrupting its execution if its runtime exceeds the given timeout interval (in seconds), raising the specified exception.

Parameters:

  • interval (Number) —

    timeout interval in seconds

  • exception (any) —

    timeout exception class or instance

Returns:

  • (any) —

    block’s return value



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) ⇒ UM::Queue

Pushes a value to the head of the given queue.

Parameters:

  • queue (UM::Queue) —

    queue

  • value (any) —

    value

Returns:



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) ⇒ Array

Waits for a process to change state. The process to wait for can be specified as a pid or as a pidfd, according to the given ‘idtype` and `id`.

Returns:

  • (Array)

Parameters:

  • idtype (Integer) —

    id type

  • id (Integer) —

    id

  • options (Integer) —

    options

Returns:

  • (Array<Integer>) —

    pid status



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

:nodoc:

This method depends on the availability of ‘rb_process_status_new`. See: github.com/ruby/ruby/pull/15213



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 ⇒ void

This method returns an undefined value.

Wakes up the machine in order to start processing its runqueue again. This method is normally called from another thread in order to resume processing of the runqueue when a machine is waiting for I/O completions.



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(fd, buffer, len = nil, file_offset = nil) ⇒ Integer

Writes up to ‘len` bytes from the given buffer to the given `fd`. If `len`, is not given, the entire buffer length is used.

Parameters:

  • fd (Integer) —

    file descriptor

  • buffer (String, IO::Buffer) —

    buffer

  • len (Integer) —

    maximum number of bytes to write

  • file_offset (Integer) —

    optional file offset to write to

Returns:

  • (Integer) —

    number of bytes written



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(fd, buffer, len = nil, file_offset = nil) ⇒ String, IO::Buffer

Writes up to ‘len` bytes from the given buffer to the given `fd`. If `len`, is not given, the entire buffer length is used. This method submits the operation but does not wait for it to complete. This method may be used to improve performance in situations where the application does not care about whether the I/O operation succeeds or not.

Parameters:

  • fd (Integer) —

    file descriptor

  • buffer (String, IO::Buffer) —

    buffer

  • len (Integer) —

    maximum number of bytes to write

  • file_offset (Integer) —

    optional file offset to write to

Returns:

  • (String, IO::Buffer) —

    buffer



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(fd, *buffers, file_offset = nil) ⇒ Integer

Writes from the given buffers into the given fd. This method does not guarantee that all data will be written. The application code should check the return value which indicates the number of bytes written and potentially repeat the operation after adjusting the buffers accordingly. See also ‘#sendv`.

Parameters:

  • fd (Integer) —

    file descriptor

  • *buffers (Array<String, IO::Buffer>) —

    data buffers

  • file_offset (Integer) —

    optional file offset to write to

Returns:

  • (Integer) —

    number of bytes written



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 ⇒ any

Yields control to the next fiber in the runqueue. The current fiber will not be resumed unless it is scheduled again by some other fiber. The call to ‘#yield` will return the value with which the current fiber will be eventually scheduled.

Returns:

  • (any) —

    resume value



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;
}