Welcome! Please see the About page for a little more info on how this works.

0 votes
in core.async by

This issue came up here: https://groups.google.com/d/topic/clojure/w_p4f3gNo3s/discussion
Another (possibly) related previous discussion: https://groups.google.com/d/topic/clojure-dev/9Ai-ZuCezOY/discussion

The scenario:
1. a channel has a fixed buffer of 1 and an expanding transducer (mapcat identity)
2. a single put! is performed with a collection (C1)
3. a blocking take! is initiated in another thread
4. a second put! is performed with another collection

I expected any put!'s to fail until all elements of collection C1 are taken from the channel.

Non-blocking put!'s (via offer!) do fail until then.

Blocking put!'s do block -- but only until a single take! is executed on the channel. That's surprising. Where does the put!'s value go? Everything does continue to "work" as expected, but the buffer acts like it has grown to size 2 (2 input collections, 1 partially output via the transducer)

I believe this is resolved by checking whether the buffer is full during a take! after the call to (impl/remove! buffer) but before processing any puts. See the patch for exactly what I mean. I'm not sure whether this change breaks anything as I'm not too familiar with this codebase; though tests do pass.

2 Answers

0 votes
by

Comment made by: brianru

Replaced the patch with one including a test demonstrating the behavior with the change and without.

0 votes
by
Reference: https://clojure.atlassian.net/browse/ASYNC-210 (reported by brianru)
...