read issue on server-worker

server-worker-threads.rkt

#lang racket

#|

Save this file to "server-worker.rkt"
then run
$ time racket server-worker.rkt

Then toggle the commented expression in `read-msg` and again
$ time racket server-worker.rkt

Result:
First run takes ~20s, while second run takes ~10s (as it should).

|#

(struct worker (thd in out)
  #:property prop:evt
  (λ (self) (wrap-evt (worker-in self) (λ _ self)))
  #:transparent)

(define (make-worker)
  (define-values (in-for-in out-for-in) (make-pipe))
  (define-values (in-for-out out-for-out) (make-pipe))
  (worker
   (thread
    (λ ()
      (send-msg 'ready out-for-in)
      (let loop ()
        (define cmd (do/debug (read-msg in-for-out)))
        (sleep 2)
        (send-msg (list 'done cmd) out-for-in)
        (loop))))
   in-for-in
   out-for-out))


(define-syntax-rule (do/debug cmd)
  (let ()
    (define-values (res cpu real gc)
      (time-apply (λ () cmd) '()))
    (eprintf "~a cpu: ~a real: ~a gc: ~a\n" (~a 'cmd #:min-width 50) cpu real gc)
    (apply values res)))

(define (send-msg msg out)
  (writeln msg out)
  (flush-output out))

(define (read-msg in)
  ;; TOGGLE BETWEEN THESE TWO:
  (read in)
  #;(with-input-from-string (read-line in) read))

(module+ main
  (define workers (build-list 4 (λ _ (make-worker))))

  (for ([t 20])
    (printf "t: ~a\n" t)
    (define wk (begin #;do/debug (apply sync workers)))
    (begin #;do/debug (read-msg (worker-in wk)))
    (begin #;do/debug (send-msg (list 't t) (worker-out wk))))

  (for-each (λ (w) (kill-thread (worker-thd w))) workers))

server-worker.rkt

#lang racket

#|

Save this file to "server-worker.rkt"
then run
$ time racket server-worker.rkt

Then toggle the commented expression in `read-msg` and again
$ time racket server-worker.rkt

Result:
First run takes ~20s, while second run takes ~10s (as it should).

|#

(struct worker (cmd in out err pid handler)
  #:property prop:evt
  (λ (self)
    (wrap-evt (worker-in self) (λ _ self)))
  #:transparent)

(define (make-worker cmd)
  (define err (current-error-port))
  (define-values (in out pid _err handler)
    (apply values (apply process*/ports #f #f err cmd)))
  (worker cmd in out err pid handler))


(define-syntax-rule (do/debug cmd)
  (let ()
    (define-values (res cpu real gc)
      (time-apply (λ () cmd) '()))
    (eprintf "~a cpu: ~a real: ~a gc: ~a\n" (~a 'cmd #:min-width 50) cpu real gc)
    (apply values res)))

(define (send-msg msg [out (current-output-port)])
  (writeln msg out)
  (flush-output out))

(define (read-msg [in (current-input-port)])
  ;; TOGGLE BETWEEN THESE TWO:
  #;(read in)
  (with-input-from-string (read-line in) read))

;==============;
;=== Worker ===;
;==============;

(module+ worker

  (send-msg 'ready)
  
  (let loop ()
    (define cmd (do/debug (read-msg)))
    (unless (eq? cmd 'terminate)
      (sleep 2)
      (send-msg (list 'done cmd))
      (loop))))

;==============;
;=== Server ===;
;==============;

(module+ main

  (define cmd
    '("/usr/bin/env"
      "racket" "-l" "racket/init"
      "-e" "(require (submod \"server-worker.rkt\" worker))"))
  (define workers (build-list 4 (λ _ (make-worker cmd))))

  (for ([t 20])
    (printf "t: ~a\n" t)
    (define wk (begin #;do/debug (apply sync workers)))
    (begin #;do/debug (read-msg (worker-in wk)))
    (begin #;do/debug (send-msg (list 't t) (worker-out wk))))

  (for-each (λ (wk) ((worker-handler wk) 'kill)) workers)
  #;(for ([wk workers])
    (send-msg 'terminate (worker-out wk)))
  #;(sleep 2))

See here for an explanation: racket/racket#4184

Simplest solution: just encapsulate the data to read/write into a list, which is self-delimiting.