| blob | ef5ec7923f44c03c4eed957d0d557a22399117ae |
1 /**********************************************************************
3 scheduler.c
5 $Author$
7 Copyright (C) 2020 Samuel Grant Dawson Williams
9 **********************************************************************/
18 // For `ruby_thread_has_gvl_p`.
43 /*
44 * Document-class: Fiber::Scheduler
45 *
46 * This is not an existing class, but documentation of the interface that Scheduler
47 * object should comply to in order to be used as argument to Fiber.scheduler and handle non-blocking
48 * fibers. See also the "Non-blocking fibers" section in Fiber class docs for explanations
49 * of some concepts.
50 *
51 * Scheduler's behavior and usage are expected to be as follows:
52 *
53 * * When the execution in the non-blocking Fiber reaches some blocking operation (like
54 * sleep, wait for a process, or a non-ready I/O), it calls some of the scheduler's
55 * hook methods, listed below.
56 * * Scheduler somehow registers what the current fiber is waiting on, and yields control
57 * to other fibers with Fiber.yield (so the fiber would be suspended while expecting its
58 * wait to end, and other fibers in the same thread can perform)
59 * * At the end of the current thread execution, the scheduler's method #scheduler_close is called
60 * * The scheduler runs into a wait loop, checking all the blocked fibers (which it has
61 * registered on hook calls) and resuming them when the awaited resource is ready
62 * (e.g. I/O ready or sleep time elapsed).
63 *
64 * This way concurrent execution will be achieved transparently for every
65 * individual Fiber's code.
66 *
67 * Scheduler implementations are provided by gems, like
68 * Async[https://github.com/socketry/async].
69 *
70 * Hook methods are:
71 *
72 * * #io_wait, #io_read, #io_write, #io_pread, #io_pwrite, and #io_select, #io_close
73 * * #process_wait
74 * * #kernel_sleep
75 * * #timeout_after
76 * * #address_resolve
77 * * #block and #unblock
78 * * #blocking_operation_wait
79 * * (the list is expanded as Ruby developers make more methods having non-blocking calls)
80 *
81 * When not specified otherwise, the hook implementations are mandatory: if they are not
82 * implemented, the methods trying to call hook will fail. To provide backward compatibility,
83 * in the future hooks will be optional (if they are not implemented, due to the scheduler
84 * being created for the older Ruby version, the code which needs this hook will not fail,
85 * and will just behave in a blocking fashion).
86 *
87 * It is also strongly recommended that the scheduler implements the #fiber method, which is
88 * delegated to by Fiber.schedule.
89 *
90 * Sample _toy_ implementation of the scheduler can be found in Ruby's code, in
91 * <tt>test/fiber/scheduler.rb</tt>
92 *
93 */
94 void
96 {
138 rb_define_method(rb_cFiberScheduler, "blocking_operation_wait", rb_fiber_scheduler_blocking_operation_wait, -2);
139 #endif
140 }
142 VALUE
144 {
151 }
153 static void
155 {
158 }
162 }
166 }
170 }
171 }
173 static VALUE
175 {
177 }
179 static VALUE
181 {
186 }
188 VALUE
190 {
198 }
200 // We invoke Scheduler#close when setting it to something else, to ensure
201 // the previous scheduler runs to completion before changing the scheduler.
202 // That way, we do not need to consider interactions, e.g., of a Fiber from
203 // the previous scheduler with the new scheduler.
205 // rb_fiber_scheduler_close(thread->scheduler);
206 rb_ensure(fiber_scheduler_close, thread->scheduler, fiber_scheduler_close_ensure, (VALUE)thread);
207 }
212 }
214 static VALUE
216 {
221 }
224 }
225 }
227 VALUE
229 {
231 }
234 {
236 }
238 /*
239 *
240 * Document-method: Fiber::Scheduler#close
241 *
242 * Called when the current thread exits. The scheduler is expected to implement this
243 * method in order to allow all waiting fibers to finalize their execution.
244 *
245 * The suggested pattern is to implement the main event loop in the #close method.
246 *
247 */
248 VALUE
250 {
253 VALUE result;
255 // The reason for calling `scheduler_close` before calling `close` is for
256 // legacy schedulers which implement `close` and expect the user to call
257 // it. Subsequently, that method would call `Fiber.set_scheduler(nil)`
258 // which should call `scheduler_close`. If it were to call `close`, it
259 // would create an infinite loop.
268 }
270 VALUE
272 {
275 }
278 }
280 /*
281 * Document-method: Fiber::Scheduler#kernel_sleep
282 * call-seq: kernel_sleep(duration = nil)
283 *
284 * Invoked by Kernel#sleep and Mutex#sleep and is expected to provide
285 * an implementation of sleeping in a non-blocking way. Implementation might
286 * register the current fiber in some list of "which fiber wait until what
287 * moment", call Fiber.yield to pass control, and then in #close resume
288 * the fibers whose wait period has elapsed.
289 *
290 */
291 VALUE
293 {
295 }
297 VALUE
299 {
301 }
303 #if 0
304 /*
305 * Document-method: Fiber::Scheduler#timeout_after
306 * call-seq: timeout_after(duration, exception_class, *exception_arguments, &block) -> result of block
307 *
308 * Invoked by Timeout.timeout to execute the given +block+ within the given
309 * +duration+. It can also be invoked directly by the scheduler or user code.
310 *
311 * Attempt to limit the execution time of a given +block+ to the given
312 * +duration+ if possible. When a non-blocking operation causes the +block+'s
313 * execution time to exceed the specified +duration+, that non-blocking
314 * operation should be interrupted by raising the specified +exception_class+
315 * constructed with the given +exception_arguments+.
316 *
317 * General execution timeouts are often considered risky. This implementation
318 * will only interrupt non-blocking operations. This is by design because it's
319 * expected that non-blocking operations can fail for a variety of
320 * unpredictable reasons, so applications should already be robust in handling
321 * these conditions and by implication timeouts.
322 *
323 * However, as a result of this design, if the +block+ does not invoke any
324 * non-blocking operations, it will be impossible to interrupt it. If you
325 * desire to provide predictable points for timeouts, consider adding
326 * +sleep(0)+.
327 *
328 * If the block is executed successfully, its result will be returned.
329 *
330 * The exception will typically be raised using Fiber#raise.
331 */
332 VALUE
333 rb_fiber_scheduler_timeout_after(VALUE scheduler, VALUE timeout, VALUE exception, VALUE message)
334 {
335 VALUE arguments[] = {
337 };
340 }
342 VALUE
344 {
346 }
347 #endif
349 /*
350 * Document-method: Fiber::Scheduler#process_wait
351 * call-seq: process_wait(pid, flags)
352 *
353 * Invoked by Process::Status.wait in order to wait for a specified process.
354 * See that method description for arguments description.
355 *
356 * Suggested minimal implementation:
357 *
358 * Thread.new do
359 * Process::Status.wait(pid, flags)
360 * end.value
361 *
362 * This hook is optional: if it is not present in the current scheduler,
363 * Process::Status.wait will behave as a blocking method.
364 *
365 * Expected to return a Process::Status instance.
366 */
367 VALUE
369 {
370 VALUE arguments[] = {
372 };
375 }
377 /*
378 * Document-method: Fiber::Scheduler#block
379 * call-seq: block(blocker, timeout = nil)
380 *
381 * Invoked by methods like Thread.join, and by Mutex, to signify that current
382 * Fiber is blocked until further notice (e.g. #unblock) or until +timeout+ has
383 * elapsed.
384 *
385 * +blocker+ is what we are waiting on, informational only (for debugging and
386 * logging). There are no guarantee about its value.
387 *
388 * Expected to return boolean, specifying whether the blocking operation was
389 * successful or not.
390 */
391 VALUE
393 {
395 }
397 /*
398 * Document-method: Fiber::Scheduler#unblock
399 * call-seq: unblock(blocker, fiber)
400 *
401 * Invoked to wake up Fiber previously blocked with #block (for example, Mutex#lock
402 * calls #block and Mutex#unlock calls #unblock). The scheduler should use
403 * the +fiber+ parameter to understand which fiber is unblocked.
404 *
405 * +blocker+ is what was awaited for, but it is informational only (for debugging
406 * and logging), and it is not guaranteed to be the same value as the +blocker+ for
407 * #block.
408 *
409 */
410 VALUE
412 {
415 // `rb_fiber_scheduler_unblock` can be called from points where `errno` is expected to be preserved. Therefore, we should save and restore it. For example `io_binwrite` calls `rb_fiber_scheduler_unblock` and if `errno` is reset to 0 by user code, it will break the error handling in `io_write`.
416 // If we explicitly preserve `errno` in `io_binwrite` and other similar functions (e.g. by returning it), this code is no longer needed. I hope in the future we will be able to remove it.
424 }
426 /*
427 * Document-method: Fiber::Scheduler#io_wait
428 * call-seq: io_wait(io, events, timeout)
429 *
430 * Invoked by IO#wait, IO#wait_readable, IO#wait_writable to ask whether the
431 * specified descriptor is ready for specified events within
432 * the specified +timeout+.
433 *
434 * +events+ is a bit mask of <tt>IO::READABLE</tt>, <tt>IO::WRITABLE</tt>, and
435 * <tt>IO::PRIORITY</tt>.
436 *
437 * Suggested implementation should register which Fiber is waiting for which
438 * resources and immediately calling Fiber.yield to pass control to other
439 * fibers. Then, in the #close method, the scheduler might dispatch all the
440 * I/O resources to fibers waiting for it.
441 *
442 * Expected to return the subset of events that are ready immediately.
443 *
444 */
445 VALUE
447 {
449 }
451 VALUE
453 {
454 return rb_fiber_scheduler_io_wait(scheduler, io, RB_UINT2NUM(RUBY_IO_READABLE), rb_io_timeout(io));
455 }
457 VALUE
459 {
460 return rb_fiber_scheduler_io_wait(scheduler, io, RB_UINT2NUM(RUBY_IO_WRITABLE), rb_io_timeout(io));
461 }
463 /*
464 * Document-method: Fiber::Scheduler#io_select
465 * call-seq: io_select(readables, writables, exceptables, timeout)
466 *
467 * Invoked by IO.select to ask whether the specified descriptors are ready for
468 * specified events within the specified +timeout+.
469 *
470 * Expected to return the 3-tuple of Array of IOs that are ready.
471 *
472 */
473 VALUE rb_fiber_scheduler_io_select(VALUE scheduler, VALUE readables, VALUE writables, VALUE exceptables, VALUE timeout)
474 {
475 VALUE arguments[] = {
477 };
480 }
483 {
484 // I wondered about extracting argv, and checking if there is only a single
485 // IO instance, and instead calling `io_wait`. However, it would require a
486 // decent amount of work and it would be hard to preserve the exact
487 // semantics of IO.select.
490 }
492 /*
493 * Document-method: Fiber::Scheduler#io_read
494 * call-seq: io_read(io, buffer, length, offset) -> read length or -errno
495 *
496 * Invoked by IO#read or IO#Buffer.read to read +length+ bytes from +io+ into a
497 * specified +buffer+ (see IO::Buffer) at the given +offset+.
498 *
499 * The +length+ argument is the "minimum length to be read". If the IO buffer
500 * size is 8KiB, but the +length+ is +1024+ (1KiB), up to 8KiB might be read,
501 * but at least 1KiB will be. Generally, the only case where less data than
502 * +length+ will be read is if there is an error reading the data.
503 *
504 * Specifying a +length+ of 0 is valid and means try reading at least once and
505 * return any available data.
506 *
507 * Suggested implementation should try to read from +io+ in a non-blocking
508 * manner and call #io_wait if the +io+ is not ready (which will yield control
509 * to other fibers).
510 *
511 * See IO::Buffer for an interface available to return data.
512 *
513 * Expected to return number of bytes read, or, in case of an error,
514 * <tt>-errno</tt> (negated number corresponding to system's error code).
515 *
516 * The method should be considered _experimental_.
517 */
518 VALUE
519 rb_fiber_scheduler_io_read(VALUE scheduler, VALUE io, VALUE buffer, size_t length, size_t offset)
520 {
521 VALUE arguments[] = {
523 };
526 }
528 /*
529 * Document-method: Fiber::Scheduler#io_read
530 * call-seq: io_pread(io, buffer, from, length, offset) -> read length or -errno
531 *
532 * Invoked by IO#pread or IO::Buffer#pread to read +length+ bytes from +io+
533 * at offset +from+ into a specified +buffer+ (see IO::Buffer) at the given
534 * +offset+.
535 *
536 * This method is semantically the same as #io_read, but it allows to specify
537 * the offset to read from and is often better for asynchronous IO on the same
538 * file.
539 *
540 * The method should be considered _experimental_.
541 */
542 VALUE
543 rb_fiber_scheduler_io_pread(VALUE scheduler, VALUE io, rb_off_t from, VALUE buffer, size_t length, size_t offset)
544 {
545 VALUE arguments[] = {
547 };
550 }
552 /*
553 * Document-method: Scheduler#io_write
554 * call-seq: io_write(io, buffer, length, offset) -> written length or -errno
555 *
556 * Invoked by IO#write or IO::Buffer#write to write +length+ bytes to +io+ from
557 * from a specified +buffer+ (see IO::Buffer) at the given +offset+.
558 *
559 * The +length+ argument is the "minimum length to be written". If the IO
560 * buffer size is 8KiB, but the +length+ specified is 1024 (1KiB), at most 8KiB
561 * will be written, but at least 1KiB will be. Generally, the only case where
562 * less data than +length+ will be written is if there is an error writing the
563 * data.
564 *
565 * Specifying a +length+ of 0 is valid and means try writing at least once, as
566 * much data as possible.
567 *
568 * Suggested implementation should try to write to +io+ in a non-blocking
569 * manner and call #io_wait if the +io+ is not ready (which will yield control
570 * to other fibers).
571 *
572 * See IO::Buffer for an interface available to get data from buffer
573 * efficiently.
574 *
575 * Expected to return number of bytes written, or, in case of an error,
576 * <tt>-errno</tt> (negated number corresponding to system's error code).
577 *
578 * The method should be considered _experimental_.
579 */
580 VALUE
581 rb_fiber_scheduler_io_write(VALUE scheduler, VALUE io, VALUE buffer, size_t length, size_t offset)
582 {
583 VALUE arguments[] = {
585 };
588 }
590 /*
591 * Document-method: Fiber::Scheduler#io_pwrite
592 * call-seq: io_pwrite(io, buffer, from, length, offset) -> written length or -errno
593 *
594 * Invoked by IO#pwrite or IO::Buffer#pwrite to write +length+ bytes to +io+
595 * at offset +from+ into a specified +buffer+ (see IO::Buffer) at the given
596 * +offset+.
597 *
598 * This method is semantically the same as #io_write, but it allows to specify
599 * the offset to write to and is often better for asynchronous IO on the same
600 * file.
601 *
602 * The method should be considered _experimental_.
603 *
604 */
605 VALUE
606 rb_fiber_scheduler_io_pwrite(VALUE scheduler, VALUE io, rb_off_t from, VALUE buffer, size_t length, size_t offset)
607 {
608 VALUE arguments[] = {
610 };
613 }
615 VALUE
616 rb_fiber_scheduler_io_read_memory(VALUE scheduler, VALUE io, void *base, size_t size, size_t length)
617 {
625 }
627 VALUE
628 rb_fiber_scheduler_io_write_memory(VALUE scheduler, VALUE io, const void *base, size_t size, size_t length)
629 {
637 }
639 VALUE
640 rb_fiber_scheduler_io_pread_memory(VALUE scheduler, VALUE io, rb_off_t from, void *base, size_t size, size_t length)
641 {
649 }
651 VALUE
652 rb_fiber_scheduler_io_pwrite_memory(VALUE scheduler, VALUE io, rb_off_t from, const void *base, size_t size, size_t length)
653 {
661 }
663 VALUE
665 {
669 }
671 /*
672 * Document-method: Fiber::Scheduler#address_resolve
673 * call-seq: address_resolve(hostname) -> array_of_strings or nil
674 *
675 * Invoked by any method that performs a non-reverse DNS lookup. The most
676 * notable method is Addrinfo.getaddrinfo, but there are many other.
677 *
678 * The method is expected to return an array of strings corresponding to ip
679 * addresses the +hostname+ is resolved to, or +nil+ if it can not be resolved.
680 *
681 * Fairly exhaustive list of all possible call-sites:
682 *
683 * - Addrinfo.getaddrinfo
684 * - Addrinfo.tcp
685 * - Addrinfo.udp
686 * - Addrinfo.ip
687 * - Addrinfo.new
688 * - Addrinfo.marshal_load
689 * - SOCKSSocket.new
690 * - TCPServer.new
691 * - TCPSocket.new
692 * - IPSocket.getaddress
693 * - TCPSocket.gethostbyname
694 * - UDPSocket#connect
695 * - UDPSocket#bind
696 * - UDPSocket#send
697 * - Socket.getaddrinfo
698 * - Socket.gethostbyname
699 * - Socket.pack_sockaddr_in
700 * - Socket.sockaddr_in
701 * - Socket.unpack_sockaddr_in
702 */
703 VALUE
705 {
706 VALUE arguments[] = {
707 hostname
708 };
711 }
721 };
723 static VALUE
725 {
726 struct rb_blocking_operation_wait_arguments *arguments = (struct rb_blocking_operation_wait_arguments*)_arguments;
730 }
732 arguments->state->result = rb_nogvl(arguments->function, arguments->data, arguments->unblock_function, arguments->data2, arguments->flags);
735 // Make sure it's only invoked once.
739 }
741 /*
742 * Document-method: Fiber::Scheduler#blocking_operation_wait
743 * call-seq: blocking_operation_wait(work)
744 *
745 * Invoked by Ruby's core methods to run a blocking operation in a non-blocking way.
746 *
747 * Minimal suggested implementation is:
748 *
749 * def blocking_operation_wait(work)
750 * Thread.new(&work).join
751 * end
752 */
753 VALUE rb_fiber_scheduler_blocking_operation_wait(VALUE scheduler, void* (*function)(void *), void *data, rb_unblock_function_t *unblock_function, void *data2, int flags, struct rb_fiber_scheduler_blocking_operation_state *state)
754 {
762 };
767 }
769 /*
770 * Document-method: Fiber::Scheduler#fiber
771 * call-seq: fiber(&block)
772 *
773 * Implementation of the Fiber.schedule. The method is <em>expected</em> to immediately
774 * run the given block of code in a separate non-blocking fiber, and to return that Fiber.
775 *
776 * Minimal suggested implementation is:
777 *
778 * def fiber(&block)
779 * fiber = Fiber.new(blocking: false, &block)
780 * fiber.resume
781 * fiber
782 * end
783 */
784 VALUE
786 {
788 }