CODEWITHSUNDEEP
clojurechannelcore.async
Petr Gladkikh
Petr Gladkikh

Reputation: 1932

Should clojure.core.async channel be drained to release parked puts

The problem: I have channel that consumer reads from and might stop reading when got enough data. When reader stops it closes channel with clojure.core.async/close!

The documentation says that at this moment all puts to channel after close is invoked should return false and do nothing. But the documentation also says that

Logically closing happens after all puts have been delivered. Therefore, any blocked or parked puts will remain blocked/parked until a taker releases them.

Does it mean that to release producers that were already blocked in parked puts at the moment of closing channel I should always also drain channel (read all remaining items) at consumer side? Following code shows that go block never finishes:

(require '[clojure.core.async :as a])
(let [c (a/chan)]
  (a/go
    (prn "Go")
    (prn "Put" (a/>! c 333)))
  (Thread/sleep 300) ;; Let go block to be scheduled
  (a/close! c))

If this is true, and I do not want to read all events then I should implement e.g. timeouts at producer side to detect that no more data is necessary?

Is there simpler way for consumer to tell "enough" to push back so producer stops also gracefully?

I found out that clojure.core.async/put! does not block and allows to avoid unnecessary blocking. Are there disadvantages of using it instead of clojure.core.aasync/>!?

Upvotes: 2

Views: 1029

Answers (2)

nottmey
nottmey

Reputation: 333

This behavior is intended, since they explicitly state it in the docs!

In your case, do (while (async/poll! c)) after closing channel c to release all blocked/parked (message sending) threads/go-blocks.

If you want to do anything with the content you can do:

(->> (repeatedly #(async/poll! c))
     (take-while identity))

Upvotes: 2

Arthur Ulfeldt
Arthur Ulfeldt

Reputation: 91554

closing chans frees all who are reading from them them, and leaves writers blocked

here is the reading case (where it works nicely):

user> (def a-chan (async/chan))
#'user/a-chan
user> (future (async/<!! a-chan)
              (println "continuting after take"))
#future[{:status :pending, :val nil} 0x5fb5a025]
user> (async/close! a-chan)
nil
user> continuting after take

And here is a test of the writing case where, as you say, draining it may be a good idea:

user> (def b-chan (async/chan))
#'user/b-chan
user> (future (try (async/>!! b-chan 4)
                   (println "continuting after put")
                   (catch Exception e
                     (println "got exception" e))
                   (finally
                     (println "finished in finally"))))
#future[{:status :pending, :val nil} 0x17be0f7b]
user> (async/close! b-chan)
nil

I don't find any evidence of the stuck writer unblocking here when the chan is closed

Upvotes: 2

Related Questions