Lost input when using suspendable read-line

Lost input when using suspendable read-line

[Maxim Cournoyer]

Hello,

I'm pretty new to this, so hopefully I'm doing something wrong, but when
using a suspendable read-line with the following "minimal" program:

--8<---------------cut here---------------start------------->8---
(use-modules (ice-9 match)
             (ice-9 rdelim)
             (ice-9 suspendable-ports))

(install-suspendable-ports!)

(define (read-line* port cont)
        ;; Return, as a pair, the line and the terminated delimiter or end-of-file
        ;; object.  When a line cannot be read, return the '(suspended
        ;; . partial-continuation) pair, where partial-continuation can be
        ;; evaluated in the future when the port is ready to be read.
        (call-with-prompt 'continue
          (lambda ()
            (parameterize ((current-read-waiter
                            (lambda (_)
                              (abort-to-prompt 'continue))))
              (if cont
                  (cont)
                  (read-line port 'split))))
          (lambda (partial-continuation)
            (cons 'suspended partial-continuation))))

(define (main)

  ;; Create a pipe, and set its read side to non-blocking mode.
  (define child->parent-pipe (pipe))
  (let ((flags (fcntl (car child->parent-pipe) F_GETFL)))
    (fcntl (car child->parent-pipe) F_SETFL (logior O_NONBLOCK flags)))

  ;; Empty buffers to avoid duplicated output.
  (flush-all-ports)

  (match  (primitive-fork)
    (0                                  ;child

     ;; Connect the stdout and stderr outputs of the child process to the
     ;; pipe established in the parent.
     (close (car child->parent-pipe))   ;unused input pipe
     (dup2 (port->fdes (cdr child->parent-pipe)) 1)
     (dup2 1 2)

     (set-current-output-port (cdr child->parent-pipe))
     (set-current-error-port (cdr child->parent-pipe))

     (while #t
       (format #t "Line 1\n")
       (format #t "Line 2\nLine 3\n")
       (display "Done!\n")
       (force-output)

       (sleep 5)))

    (child-pid                          ;parent

     (close (cdr child->parent-pipe))   ;disconnect the write end of the pipe

     (define port (car child->parent-pipe))

     (let loop ((cont #f))
       (match (select (list (port->fdes port)) '() '())
         (((fdes ..1) () ())
          (let next-line ((line+delim (read-line* port cont)))
            (match line+delim
              (('suspended . partial-continuation)
               (loop partial-continuation))
              ((line . _)
               (format #t "~a~%" line)
               (next-line (read-line* port cont)))))))))))

(main)
--8<---------------cut here---------------end--------------->8---

The output looks like this (non-deterministic it seems):

--8<---------------cut here---------------start------------->8---
$ guile repro.scm
Line 1
Line 2
Line 3
Done!
Line 1
Line 2
Line 3
Done!
Line 1
Line 1
Line 1

[...]
--8<---------------cut here---------------end--------------->8---

And strace shows that more than a single line is buffered on each
read-line, but that it only returns the first line of such buffered
input:

--8<---------------cut here---------------start------------->8---
read(8, 0x7fcd06bfc020, 4096)           = -1 EAGAIN (Resource temporarily unavailable)
rt_sigprocmask(SIG_BLOCK, NULL, [], 8)  = 0
select(9, [3 8], [], [], NULL)          = 1 (in [8])
read(8, "Line 1\nLine 2\nLine 3\nDone!\n", 4096) = 27
write(1, "Line 1", 6Line 1)                   = 6
write(1, "\n", 1
)                       = 1
read(8, 0x7fcd06bfc020, 4096)           = -1 EAGAIN (Resource temporarily unavailable)
rt_sigprocmask(SIG_BLOCK, NULL, [], 8)  = 0
select(9, [3 8], [], [], NULL)          = 1 (in [8])
read(8, "Line 1\nLine 2\nLine 3\nDone!\n", 4096) = 27
write(1, "Line 1", 6Line 1)                   = 6
write(1, "\n", 1
)                       = 1

[...]
--8<---------------cut here---------------end--------------->8---


Bug?  PEBCAK?  I'm interested to know :-).

Thanks,

Maxim

Re: Lost input when using suspendable read-line

[Leo Prikler]

Hi,

Am Montag, den 23.08.2021, 00:04 -0400 schrieb Maxim Cournoyer:
> --8<---------------cut here---------------start------------->8---
> (use-modules (ice-9 match)
>              (ice-9 rdelim)
>              (ice-9 suspendable-ports))
> 
> (install-suspendable-ports!)
> 
> (define (read-line* port cont)
>         ;; Return, as a pair, the line and the terminated delimiter
> or end-of-file
>         ;; object.  When a line cannot be read, return the
> '(suspended
>         ;; . partial-continuation) pair, where partial-continuation
> can be
>         ;; evaluated in the future when the port is ready to be read.
>         (call-with-prompt 'continue
>           (lambda ()
>             (parameterize ((current-read-waiter
>                             (lambda (_)
>                               (abort-to-prompt 'continue))))
>               (if cont
>                   (cont)
>                   (read-line port 'split))))
>           (lambda (partial-continuation)
>             (cons 'suspended partial-continuation))))
> 
> (define (main)
>   [...]
>      (let loop ((cont #f))
>        (match (select (list (port->fdes port)) '() '())
>          (((fdes ..1) () ())
>           (let next-line ((line+delim (read-line* port cont)))
>             (match line+delim
>               (('suspended . partial-continuation)
>                (loop partial-continuation))
>               ((line . _)
>                (format #t "~a~%" line)
>                (next-line (read-line* port cont)))))))))))
> 
> (main)
> --8<---------------cut here---------------end--------------->8---
This main loop appears broken.  Look at the value of cont over time. 
On the first few lines, before suspending, you will read all the lines
just fine.  But afterwards it will be set to partial-continuation, even
if said continuation is no longer current.

This appears to be no issue if you need less than one continuation to
finish the line, but if you need two or more, the outer continuation
will cause you to redo the inner over and over again.

I tried to make this loop somewhat more sensible:

--8<---------------cut here---------------start------------->8---
     (define port (car child->parent-pipe))
     (define cont (make-parameter #f))
     (define do-read
       (lambda ()
         (match (select (list (port->fdes port)) '() '())
           (((fdes ..1) () ())
            (match (read-line* port (cont))
              (('suspended . partial-cont)
               partial-cont)
              ((line . _)
               (format #t "~a~%" line)
               #f))))))

     (let loop ((cont? (do-read)))
       (loop (parameterize ((cont cont?)) (do-read)))))))
--8<---------------cut here---------------end--------------->8---
Here, each continuation is used exactly once, and that is to finish the
current line.  With this, I typically get output of the style:

--8<---------------cut here---------------start------------->8---
Line 1
[wait for it...]
Line 2
Line 3
Done!
Line 1
[wait for it...]
--8<----------
-----cut here---------------end--------------->8---
So it's not perfect either, but it's somewhat better than what you have
currently.

I'd hazard a guess that there are simpler implementations that need not
make use of parameters, particularly because I kinda eliminated the
inner loop anyway.  A simpler fix, which retains your structure is to
use 
               (next-line (read-line* port #f)))))))))))
in the case where a line has already been read.  This also seems to
have the desired behaviour of waiting after the "Done!" line.

Regards,
Leo

Re: Lost input when using suspendable read-line

[Maxim Cournoyer]

Hello Leo!

Leo Prikler <leo.prikler@student.tugraz.at> writes:

> Hi,
>
> Am Montag, den 23.08.2021, 00:04 -0400 schrieb Maxim Cournoyer:
>> --8<---------------cut here---------------start------------->8---
>> (use-modules (ice-9 match)
>>              (ice-9 rdelim)
>>              (ice-9 suspendable-ports))
>> 
>> (install-suspendable-ports!)
>> 
>> (define (read-line* port cont)
>>         ;; Return, as a pair, the line and the terminated delimiter
>> or end-of-file
>>         ;; object.  When a line cannot be read, return the
>> '(suspended
>>         ;; . partial-continuation) pair, where partial-continuation
>> can be
>>         ;; evaluated in the future when the port is ready to be read.
>>         (call-with-prompt 'continue
>>           (lambda ()
>>             (parameterize ((current-read-waiter
>>                             (lambda (_)
>>                               (abort-to-prompt 'continue))))
>>               (if cont
>>                   (cont)
>>                   (read-line port 'split))))
>>           (lambda (partial-continuation)
>>             (cons 'suspended partial-continuation))))
>> 
>> (define (main)
>>   [...]
>>      (let loop ((cont #f))
>>        (match (select (list (port->fdes port)) '() '())
>>          (((fdes ..1) () ())
>>           (let next-line ((line+delim (read-line* port cont)))
>>             (match line+delim
>>               (('suspended . partial-continuation)
>>                (loop partial-continuation))
>>               ((line . _)
>>                (format #t "~a~%" line)
>>                (next-line (read-line* port cont)))))))))))
>> 
>> (main)
>> --8<---------------cut here---------------end--------------->8---
> This main loop appears broken.  Look at the value of cont over time. 
> On the first few lines, before suspending, you will read all the lines
> just fine.  But afterwards it will be set to partial-continuation, even
> if said continuation is no longer current.
>
> This appears to be no issue if you need less than one continuation to
> finish the line, but if you need two or more, the outer continuation
> will cause you to redo the inner over and over again.
>
> I tried to make this loop somewhat more sensible:
>
>      (define port (car child->parent-pipe))
>      (define cont (make-parameter #f))
>      (define do-read
>        (lambda ()
>          (match (select (list (port->fdes port)) '() '())
>            (((fdes ..1) () ())
>             (match (read-line* port (cont))
>               (('suspended . partial-cont)
>                partial-cont)
>               ((line . _)
>                (format #t "~a~%" line)
>                #f))))))
>
>      (let loop ((cont? (do-read)))
>        (loop (parameterize ((cont cont?)) (do-read)))))))
>
> Here, each continuation is used exactly once, and that is to finish the
> current line.  With this, I typically get output of the style:
>
> --8<---------------cut here---------------start------------->8---
> Line 1
> [wait for it...]
> Line 2
> Line 3
> Done!
> Line 1
> [wait for it...]
> --8<----------
> -----cut here---------------end--------------->8---
> So it's not perfect either, but it's somewhat better than what you have
> currently.
>
> I'd hazard a guess that there are simpler implementations that need not
> make use of parameters, particularly because I kinda eliminated the
> inner loop anyway.  A simpler fix, which retains your structure is to
> use 
>                (next-line (read-line* port #f)))))))))))
> in the case where a line has already been read.  This also seems to
> have the desired behaviour of waiting after the "Done!" line.

Thank you so much (to flatwhatson on #guile) :-).  I had indeed failed
to carry/refresh the partial continuity in the inner loop.  I had made a
similar mistake in the original problem which was capturing the
continuity at the wrong place in the closure, which was fixed like so:

--8<---------------cut here---------------start------------->8---
1 file changed, 7 insertions(+), 7 deletions(-)
src/mcron/base.scm | 14 +++++++-------

modified   src/mcron/base.scm
@@ -330,8 +330,7 @@ associated <job-data> instance."
     ;; could not be read.
     (let ((name (job-data:name data))
           (pid  (job-data:pid  data))
-          (port (job-data:port data))
-          (cont (job-data:continuation data)))
+          (port (job-data:port data)))
 
       (define (read-line*)
         ;; Return, as a pair, the line and the terminated delimiter or end-of-file
@@ -343,11 +342,12 @@ associated <job-data> instance."
             (parameterize ((current-read-waiter
                             (lambda (_)
                               (abort-to-prompt 'continue))))
-              (if cont
-                  (begin
-                    (set-job-data-continuation! data #f) ;reset continuation field
-                    (cont))
-                  (read-line port 'split))))
+              (let ((cont (job-data:continuation data)))
+                (if cont
+                    (begin
+                      (set-job-data-continuation! data #f) ;reset continuation
+                      (cont))
+                    (read-line port 'split)))))
           (lambda (partial-continuation)
             (cons 'suspended partial-continuation))))
 --8<---------------cut here---------------end--------------->8---

You've made my day.

Happy hacking!

Maxim