timers: Add cancel function that removes timer wheel entry.

Fixes #109.

Previously, an operation like:

  (choice-operation (sleep-operation 1234) (get-operation channel))

would accumulate timer wheel entries every time the ‘get’ operation wins
over the ‘sleep’ operation, potentially leading to unbounded memory
usage (each ‘sleep’ timer and its associated continuation would remain
on the wheel for 1234 seconds in this case).

This commit fixes it by removing the timer wheel entry as soon as the
timer operation is canceled.

* fibers/timers.scm (timer-operation)[wheel-entry]: New variable.
Set it in block function.  Use ‘make-timer-operation/internal’ and add
cancel function.
* fibers/scheduler.scm (scheduler-timers): Export.
* tests/cancel-timer.scm: New file.
* Makefile.am (TESTS): Add it.

Makefile.am

@@ -119,7 +119,8 @@ TESTS = \
 	tests/parameters.scm \
 	tests/preemption.scm \
 	tests/speedup.scm \
-	tests/timer-wheel.scm
+	tests/timer-wheel.scm \
+	tests/cancel-timer.scm
 
 # The following tests require SOCK_NONBLOCK and SOCK_CLOEXEC. For now we just
 # run them on a platform that supports epoll (probably Linux).

fibers/scheduler.scm

@@ -35,6 +35,7 @@
             (scheduler-kernel-thread/public . scheduler-kernel-thread)
             scheduler-remote-peers
             scheduler-work-pending?
+            scheduler-timers
             choose-parallel-scheduler
             run-scheduler
             destroy-scheduler

fibers/timers.scm

@@ -1,6 +1,6 @@
 ;; Fibers: cooperative, event-driven user-space threads.
 
-;;;; Copyright (C) 2016 Free Software Foundation, Inc.
+;;;; Copyright (C) 2016, 2024 Free Software Foundation, Inc.
 ;;;;
 ;;;; This library is free software; you can redistribute it and/or
 ;;;; modify it under the terms of the GNU Lesser General Public
@@ -19,6 +19,7 @@
 (define-module (fibers timers)
   #:use-module (fibers scheduler)
   #:use-module (fibers operations)
+  #:autoload   (fibers timer-wheel) (timer-wheel-remove!)
   #:use-module (ice-9 atomic)
   #:use-module (ice-9 match)
   #:use-module (ice-9 threads)
@@ -45,23 +46,37 @@
   "Make an operation that will succeed when the current time is
 greater than or equal to @var{expiry}, expressed in internal time
 units.  The operation will succeed with no values."
-  (make-base-operation #f
-                       (lambda ()
-                         (and (< expiry (get-internal-real-time))
-                              values))
-                       (lambda (flag sched resume)
-                         (define (timer)
-                           (match (atomic-box-compare-and-swap! flag 'W 'S)
-                             ('W (resume values))
-                             ('C (timer))
-                             ('S #f)))
-                         (if sched
-                             (schedule-task-at-time sched expiry timer)
-                             (schedule-task
-                              (timer-sched)
-                              (lambda ()
-                                (perform-operation (timer-operation expiry))
-                                (timer)))))))
+  (define wheel-entry
+    ;; If true, this is the currently active timer entry for this operation.
+    #f)
+
+  (make-base-operation/internal
+   #f                                             ;wrap
+   (lambda ()                                     ;try
+     (and (< expiry (get-internal-real-time))
+          values))
+   (lambda (flag sched resume)                    ;block
+     (define (timer)
+       (match (atomic-box-compare-and-swap! flag 'W 'S)
+         ('W (resume values))
+         ('C (timer))
+         ('S #f)))
+     (if sched
+         (set! wheel-entry
+               (schedule-task-at-time sched expiry timer))
+         (schedule-task
+          (timer-sched)
+          (lambda ()
+            (perform-operation (timer-operation expiry))
+            (timer)))))
+   (lambda (sched)                                ;cancel
+     ;; This operation is being canceled.
+     (when (and sched wheel-entry)
+       ;; Remove WHEEL-ENTRY from the timer wheel right away to avoid
+       ;; accumulating entries in the wheel.  See
+       ;; <https://github.com/wingo/fibers/issues/109>.
+       (timer-wheel-remove! (scheduler-timers sched) wheel-entry)
+       (set! wheel-entry #f)))))
 
 (define (sleep-operation seconds)
   "Make an operation that will succeed with no values when

tests/cancel-timer.scm

@@ -0,0 +1,68 @@
+;; Fibers: cooperative, event-driven user-space threads.
+
+;;;; Copyright (C) 2024 Ludovic Courtès <[email protected]>
+;;;;
+;;;; 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 program.  If not, see <http://www.gnu.org/licenses/>.
+;;;;
+
+(define-module (tests cancel-timer)
+  #:use-module (fibers)
+  #:use-module (fibers channels)
+  #:use-module (fibers operations)
+  #:use-module (fibers timers)
+  #:use-module (ice-9 format))
+
+(define (heap-size)
+  (assoc-ref (gc-stats) 'heap-size))
+
+(define iterations 200000)
+
+;;; Check the heap growth caused by repeated choice operations where one of
+;;; the base operations is a timer that always "loses" the choice.
+;;;
+;;; This situation used to cause timer continuations to accumulate, thereby
+;;; leading to unbounded heap growth.  The cancel function of
+;;; 'timer-operation' fixes that by immediately canceling timers that lost in
+;;; a choice operation.  See <https://github.com/wingo/fibers/issues/109>.
+
+(run-fibers
+ (lambda ()
+   (define channel
+     (make-channel))
+
+   (spawn-fiber
+    (lambda ()
+      (let loop ((i 0))
+        (when (< i iterations)
+          (put-message channel 'hello)
+          (loop (+ i 1))))))
+
+   (let ((initial-heap-size (heap-size)))
+     (let loop ((i 0))
+       (when (< i iterations)
+         (perform-operation
+          (choice-operation (sleep-operation 500)
+                            (get-operation channel)))
+         (loop (+ 1 i))))
+
+     (let ((final-heap-size (heap-size))
+           (MiB (lambda (size)
+                  (/ size (expt 2 20.)))))
+       (if (<= final-heap-size (* 2 initial-heap-size))
+           (format #t "final heap size: ~,2f MiB; initial heap size: ~,2f MiB~%"
+                   (MiB final-heap-size) (MiB initial-heap-size))
+           (begin
+             (format #t "heap grew too much: ~,2f MiB vs. ~,2f MiB~%"
+                     (MiB final-heap-size) (MiB initial-heap-size))
+             (primitive-exit 1)))))))