core.async 通道 - 跟踪时发生的情况（示例）

Question

我在中国。 Paul Butcher 的 6 个7 Concurrency Models in 7 Weeks，专注于core.async。

我们有以下功能

(defn map-chan [f from]                                                         
  (let [to (chan)]
    (go-loop []
      (when-let [x (<! from)]    
        (>! to (f x))            
        (println "parking channel write.") 
        (recur))                         
      (close! to))
    (println "map-chan done.")
    to)) 

我自己添加了printlns，以探索确切的计算顺序，我想在这里询问。

我们可以这样运行

(def ch (to-chan (range 10)))             ; [1]
(def mapped (map-chan (partial * 2) ch))  ; [2]
(<!! (async/into [] mapped))              ; [3]

;; [1] Create & rtn a channel from els of seq, closing it when seq fin.
;; [2] map-chan returns immediately, with blocked go blocks inside of it.
;; [3] calling async/into finally triggers the parked channel writes, as seen below.

在repl中：

channels.core=> (def ch (to-chan (range 10)))
#'channels.core/ch
channels.core=> (def mapped (map-chan (partial * 2) ch))
map-chan done.
#'channels.core/mapped
channels.core=> (<!! (async/into [] mapped))
parking channel write.
parking channel write.
parking channel write.
parking channel write.
parking channel write.
parking channel write.
parking channel write.
parking channel write.
parking channel write.
parking channel write.
[0 2 4 6 8 10 12 14 16 18]
channels.core=> 

问题

我们在这里有一个（同步）（即无缓冲）通道，它的写入器和读取器都准备好了。为什么在调用async/into 之前我的上面的“停车通道写入”不会触发？ （触发它的不是<!! 读取的频道，而是async/into 本身 - 易于检查）。我不是在抱怨这个，只是想了解为什么痕迹是这样的。 频道实际上也有点懒惰吗？他还没有在书中提到这一点。

请注意，对这段代码的依赖是org.clojure/core.async "0.1.267.0-0d7780-alpha"，如果这有什么不同的话。

此外，他在书中使用了长度为 10 的缓冲通道。 然而，我也尝试使用无缓冲（同步）通道，结果似乎相同。

Answer 1

您的输出通道to 的大小为零，因此在请求相应的镜头之前无法进行写入。查看代码的修改版本：

(ns tst.demo.core
  (:use tupelo.core tupelo.test )
  (:require
    [clojure.core.async :as async]
    ))

(defn map-chan [f from]
  (let [to (async/chan)]
    (async/go
      (loop []
        (when-let [x (async/<! from)]
          (println "put - pre")
          (async/>! to (f x))
          (println "put - post")
          (recur)))
      (async/close! to))
    (println "map-chan returns output buffer")
    to))

(dotest
(println :1)
(spyx
  (def ch (async/to-chan (range 10)))) ; [1]

(Thread/sleep 2000) (println :2)
(spyx
  (def mapped (map-chan (partial * 2) ch))) ; [2]

(Thread/sleep 2000) (println :3)
(spyx
  (async/<!! (async/into [] mapped))) ; [3]
  )

结果：

-------------------------------
   Clojure 1.10.1    Java 13
-------------------------------

lein test tst.demo.core
:1
(def ch (async/to-chan (range 10))) => #'tst.demo.core/ch
:2
map-chan returns output buffer
(def mapped (map-chan (partial * 2) ch)) => #'tst.demo.core/mapped
put - pre
:3
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
(async/<!! (async/into [] mapped)) => [0 2 4 6 8 10 12 14 16 18]

因此，go 循环确实会立即开始运行，但第一个 put 操作会阻塞，直到步骤 [3] 中的 async/into 发生。

如果我们使用长度为 20 的缓冲输出通道，我们会看到 go 循环在步骤 [3] 发生之前运行：

...
(let [to (async/chan 20)]
   ...

结果：

:1
(def ch (async/to-chan (range 10))) => #'tst.demo.core/ch
:2
map-chan returns output buffer
(def mapped (map-chan (partial * 2) ch)) => #'tst.demo.core/mapped
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
put - pre
put - post
:3
(async/<!! (async/into [] mapped)) => [0 2 4 6 8 10 12 14 16 18]