pipeline   xtlang

(bind-func pipeline:[[void]*,[[void]*,!a]*,[[void]*,!a,!b]*,[[void]*,!b,!c]*,[[void]*,!c,!d]*,[[void]*,!d]*]*
  (lambda (a b c d e)
    (let ((depth 5) (i 0) (cnt 0)
    (f1 (pipeline_stage a depth))
    (f2 (pipeline_stage b (* (a.push) (filter_rate f1)) depth (filter_output f1)))
    (f3 (pipeline_stage c (* (b.push) (filter_rate f2)) depth (filter_output f2)))
    (f4 (pipeline_stage d (* (c.push) (filter_rate f3)) depth (filter_output f3)))
    (f5 (pipeline_stage f (* (e.push) (filter_rate f4)) (filter_output f4))))
      ;; do prework!
      ((filter_prework f1))
      ((filter_prework f2))
      ((filter_prework f3))
      ((filter_prework f4))
      ((filter_prework f5))
      ;; prime pipeline!
      (dotimes (i (- (* 5 (filter_rate f1)) 1)) ((filter_work f1)))
      (dotimes (i (- (* 4 (filter_rate f2)) 1)) ((filter_work f2)))
      (dotimes (i (- (* 3 (filter_rate f3)) 1)) ((filter_work f3)))
      (dotimes (i (- (* 2 (filter_rate f4)) 1)) ((filter_work f4)))
      (dotimes (i (- (* 1 (filter_rate f5)) 1)) ((filter_work f5)))
      ;; now ready for real work!
      (lambda ()
  (set! cnt (+ cnt 1))
        (sync
         (spawn (filter_lwork f1))
         (spawn (filter_lwork f2))
         (spawn (filter_lwork f3))
         (spawn (filter_lwork f4))
         (spawn (filter_lwork f5)))
        void))))