Scheme-GNUnet requires the new operations 'wait-until-port-readable-operation' and 'wait-until-port-readable-operation' for communicating with services. This patch has been previously submitted at , on Sep 16, 2021. As of Feb 3, 2022, upstream has not responded yet. diff --git a/Makefile.am b/Makefile.am index e2db57e..0134255 100644 --- a/Makefile.am +++ b/Makefile.am @@ -33,6 +33,7 @@ SOURCES = \ fibers/deque.scm \ fibers/epoll.scm \ fibers/interrupts.scm \ + fibers/io-wakeup.scm \ fibers/nameset.scm \ fibers/operations.scm \ fibers/posix-clocks.scm \ @@ -67,6 +68,7 @@ TESTS = \ tests/conditions.scm \ tests/channels.scm \ tests/foreign.scm \ + tests/io-wakeup.scm \ tests/parameters.scm \ tests/preemption.scm \ tests/speedup.scm diff --git a/fibers.texi b/fibers.texi index 52f7177..0990c8f 100644 --- a/fibers.texi +++ b/fibers.texi @@ -12,6 +12,7 @@ This manual is for Fibers (version @value{VERSION}, updated @value{UPDATED}) Copyright 2016-2022 Andy Wingo +Copyright 2021 Maxime Devos @quotation @c For more information, see COPYING.docs in the fibers @@ -453,6 +454,7 @@ of operations for channels and timers, and an internals interface. * Channels:: Share memory by communicating. * Timers:: Operations on time. * Conditions:: Waiting for simple state changes. +* Port Readiness:: Waiting until a port is ready for I/O. * REPL Commands:: Experimenting with Fibers at the console. * Schedulers and Tasks:: Fibers are built from lower-level primitives. @end menu @@ -722,6 +724,28 @@ signalled. Equivalent to @code{(perform-operation (wait-operation cvar))}. @end defun +@node Port Readiness +@section Port Readiness + +These two operations can be used on file ports to wait until +they are readable or writable. Spurious wake-ups are possible. +This is complementary to Guile's suspendable ports. + +@example +(use-modules (fibers io-wakeup)) +@end example + +@defun wait-until-port-readable-operation port +Make an operation that will succeed with no values when the input +port @var{port} becomes readable. For passive sockets, this operation +succeeds when a connection becomes available. +@end defun + +@defun wait-until-port-writable-operation +Make an operation that will succeed with no values when the output +port @var{port} becomes writable. +@end defun + @node REPL Commands @section REPL Commands diff --git a/fibers/io-wakeup.scm b/fibers/io-wakeup.scm new file mode 100644 index 0000000..5df03f1 --- /dev/null +++ b/fibers/io-wakeup.scm @@ -0,0 +1,93 @@ +;; Fibers: cooperative, event-driven user-space threads. + +;;;; Copyright (C) 2016,2021 Free Software Foundation, Inc. +;;;; Copyright (C) 2021 Maxime Devos +;;;; +;;;; This library is free software; you can redistribute it and/or +;;;; modify it under the terms of the GNU Lesser General Public +;;;; License as published by the Free Software Foundation; either +;;;; version 3 of the License, or (at your option) any later version. +;;;; +;;;; This library is distributed in the hope that it will be useful, +;;;; but WITHOUT ANY WARRANTY; without even the implied warranty of +;;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +;;;; Lesser General Public License for more details. +;;;; +;;;; You should have received a copy of the GNU Lesser General Public +;;;; License along with this library; if not, write to the Free Software +;;;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +;;;; + +(define-module (fibers io-wakeup) + #:use-module (fibers scheduler) + #:use-module (fibers operations) + #:use-module (ice-9 atomic) + #:use-module (ice-9 match) + #:use-module (ice-9 threads) + #:use-module (ice-9 ports internal) + #:export (wait-until-port-readable-operation + wait-until-port-writable-operation)) + +(define *poll-sched* (make-atomic-box #f)) + +(define (poll-sched) + (or (atomic-box-ref *poll-sched*) + (let ((sched (make-scheduler))) + (cond + ((atomic-box-compare-and-swap! *poll-sched* #f sched)) + (else + ;; FIXME: Would be nice to clean up this thread at some point. + (call-with-new-thread + (lambda () + (define (finished?) #f) + (run-scheduler sched finished?))) + sched))))) + +;; These procedure are subject to spurious wakeups. + +(define (readable? port) + "Test if PORT is writable." + (match (select (vector port) #() #() 0) + ((#() #() #()) #f) + ((#(_) #() #()) #t))) + +(define (writable? port) + "Test if PORT is writable." + (match (select #() (vector port) #() 0) + ((#() #() #()) #f) + ((#() #(_) #()) #t))) + +(define (make-wait-operation ready? schedule-when-ready port port-ready-fd this-procedure) + (make-base-operation #f + (lambda _ + (and (ready? port) values)) + (lambda (flag sched resume) + (define (commit) + (match (atomic-box-compare-and-swap! flag 'W 'S) + ('W (resume values)) + ('C (commit)) + ('S #f))) + (if sched + (schedule-when-ready + sched (port-ready-fd port) commit) + (schedule-task + (poll-sched) + (lambda () + (perform-operation (this-procedure port)) + (commit))))))) + +(define (wait-until-port-readable-operation port) + "Make an operation that will succeed when PORT is readable." + (unless (input-port? port) + (error "refusing to wait forever for input on non-input port")) + (make-wait-operation readable? schedule-task-when-fd-readable port + port-read-wait-fd + wait-until-port-readable-operation)) + +(define (wait-until-port-writable-operation port) + "Make an operation that will succeed when PORT is writable." + (unless (output-port? port) + (error "refusing to wait forever for output on non-output port")) + (make-wait-operation writable? schedule-task-when-fd-writable port + port-write-wait-fd + wait-until-port-writable-operation)) diff --git a/tests/io-wakeup.scm b/tests/io-wakeup.scm new file mode 100644 index 0000000..c14fa81 --- /dev/null +++ b/tests/io-wakeup.scm @@ -0,0 +1,167 @@ +;; Fibers: cooperative, event-driven user-space threads. + +;;;; Copyright (C) 2016 Free Software Foundation, Inc. +;;;; Copyright (C) 2021 Maxime Devos +;;;; +;;;; This library is free software; you can redistribute it and/or +;;;; modify it under the terms of the GNU Lesser General Public +;;;; License as published by the Free Software Foundation; either +;;;; version 3 of the License, or (at your option) any later version. +;;;; +;;;; This library is distributed in the hope that it will be useful, +;;;; but WITHOUT ANY WARRANTY; without even the implied warranty of +;;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +;;;; Lesser General Public License for more details. +;;;; +;;;; You should have received a copy of the GNU Lesser General Public +;;;; License along with this library; if not, write to the Free Software +;;;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +;;;; + +(define-module (tests io-wakeup) + #:use-module (rnrs bytevectors) + #:use-module (ice-9 control) + #:use-module (ice-9 suspendable-ports) + #:use-module (ice-9 binary-ports) + #:use-module (fibers) + #:use-module (fibers io-wakeup) + #:use-module (fibers operations) + #:use-module (fibers timers)) + +(define failed? #f) + +(define-syntax-rule (assert-equal expected actual) + (let ((x expected)) + (format #t "assert ~s equal to ~s: " 'actual x) + (force-output) + (let ((y actual)) + (cond + ((equal? x y) (format #t "ok\n")) + (else + (format #t "no (got ~s)\n" y) + (set! failed? #t)))))) + +(define-syntax-rule (assert-run-fibers-terminates exp) + (begin + (format #t "assert run-fibers on ~s terminates: " 'exp) + (force-output) + (let ((start (get-internal-real-time))) + (call-with-values (lambda () (run-fibers (lambda () exp))) + (lambda vals + (format #t "ok (~a s)\n" (/ (- (get-internal-real-time) start) + 1.0 internal-time-units-per-second)) + (apply values vals)))))) + +(define-syntax-rule (assert-run-fibers-returns (expected ...) exp) + (begin + (call-with-values (lambda () (assert-run-fibers-terminates exp)) + (lambda run-fiber-return-vals + (assert-equal '(expected ...) run-fiber-return-vals))))) + + +;; Note that theoretically, on very slow systems, SECONDS might need +;; to be increased. However, readable/timeout? and writable/timeout? +;; call this 5 times in a loop anyways, so the effective timeout is +;; a fourth of a second, which should be plenty in practice. +(define* (with-timeout op #:key (seconds 0.05) (wrap values)) + (choice-operation op + (wrap-operation (sleep-operation seconds) wrap))) + +(define* (readable/timeout? port #:key (allowed-spurious 5)) + "Does waiting for readability time-out? +Allow @var{allowed-spurious} spurious wakeups." + (or (perform-operation + (with-timeout + (wrap-operation (wait-until-port-readable-operation port) + (lambda () #f)) + #:wrap (lambda () #t))) + (and (> allowed-spurious 0) + (readable/timeout? port #:allowed-spurious + (- allowed-spurious 1))))) + +(define* (writable/timeout? port #:key (allowed-spurious 5)) + "Does waiting for writability time-out? +Allow @var{allowed-spurious} spurious wakeups." + (or (perform-operation + (with-timeout + (wrap-operation (wait-until-port-writable-operation port) + (lambda () #f)) + #:wrap (lambda () #t))) + (and (> allowed-spurious 0) + (writable/timeout? port #:allowed-spurious + (- allowed-spurious 1))))) + +;; Tests: +;; * wait-until-port-readable-operaton / wait-until-port-writable-operation +;; blocks if the port isn't ready for input / output. +;; +;; This is tested with a pipe (read & write) +;; and a listening socket (read, or accept in this case). +;; +;; Due to the possibility of spurious wakeups, +;; a limited few spurious wakeups are tolerated. +;; +;; * these operations succeed if the port is ready for input / output. +;; +;; These are again tested with a pipe and a listening socket +;; +;; Blocking is detected with a small time-out. + +(define (make-listening-socket) + (let ((server (socket PF_INET SOCK_DGRAM 0))) + (bind server AF_INET INADDR_LOOPBACK 0) + server)) + +(let ((s (make-listening-socket))) + (assert-run-fibers-returns (#t) + (readable/timeout? s)) + (assert-equal #t (readable/timeout? s)) + (close s)) + +(define (set-nonblocking! sock) + (let ((flags (fcntl sock F_GETFL))) + (fcntl sock F_SETFL (logior O_NONBLOCK flags)))) + +(define-syntax-rule (with-pipes (A B) exp exp* ...) + (let* ((pipes (pipe)) + (A (car pipes)) + (B (cdr pipes))) + exp exp* ... + (close A) + (close B))) + +(with-pipes (A B) + (setvbuf A 'none) + (setvbuf B 'none) + (assert-run-fibers-returns (#t) + (readable/timeout? A)) + (assert-equal #t (readable/timeout? A)) + + ;; The buffer is empty, so writability is expected. + (assert-run-fibers-returns (#f) + (writable/timeout? B)) + (assert-equal #f (writable/timeout? B)) + + ;; Fill the buffer + (set-nonblocking! B) + (let ((bv (make-bytevector 1024))) + (let/ec k + (parameterize ((current-write-waiter k)) + (let loop () + (put-bytevector B bv) + (loop))))) + + ;; As the buffer is full, writable/timeout? should return + ;; #t. + (assert-run-fibers-returns (#t) + (writable/timeout? B)) + ;; There's plenty to read now, so readable/timeout? should + ;; return #f. + (assert-run-fibers-returns (#f) + (readable/timeout? A))) + +(exit (if failed? 1 0)) + +;; Local Variables: +;; eval: (put 'with-pipes 'scheme-indent-function 1) +;; End: