Update to Java commit 87e9a57 (2013.08.19): added limits to pending put! and take! queues ... AND fix adding to buffer pending lists

Author: dmiller

src/clojure/clojure/core/async/impl/channels.clj

@@ -16,6 +16,11 @@
 
 (set! *warn-on-reflection* true)
 
+(defmacro assert-unlock [lock test msg]
+  `(when-not ~test
+     (.unlock ~lock)
+     (throw (new Exception (str "Assert failed: " ~msg "\n" (pr-str '~test))))))       ;;;  AssertionError
+
 (defn box [val]
   (reify clojure.lang.IDeref
          (deref [_] val)))
@@ -118,7 +123,12 @@
                      nil))))
            (do
              (when (impl/active? handler)
-               (.AddFirst puts [handler val]))                        ;;; .add
+               (assert-unlock mutex
+                               (< (.Count puts) impl/MAX-QUEUE-SIZE)                         ;;; .size
+                               (str "No more than " impl/MAX-QUEUE-SIZE
+                                    " pending puts are allowed on a single channel."
+                                    " Consider using a windowed buffer."))			 
+               (.AddLast puts [handler val]))                                                ;;; .add
              (.unlock mutex)
              nil))))))
 
@@ -216,7 +226,11 @@
                  (box nil)
                  nil))
              (do
-               (.AddFirst takes handler)      ;;; .add
+                (assert-unlock mutex
+                               (< (.Count takes) impl/MAX-QUEUE-SIZE)                        ;;; .size
+                               (str "No more than " impl/MAX-QUEUE-SIZE
+                                    " pending takes are allowed on a single channel."))
+			   (.AddLast takes handler)                                                      ;;; .add
                (.unlock mutex)
                nil)))))))
 

src/clojure/clojure/core/async/impl/protocols.clj

@@ -11,6 +11,9 @@
 (ns ^{:skip-wiki true}
   clojure.core.async.impl.protocols)
 
+
+(def ^:const ^int MAX-QUEUE-SIZE 1024)
+
 (defprotocol ReadPort
   (take! [port fn1-handler] "derefable val if taken, nil if take was enqueued"))
 

test/clojure/clojure/core/async_test.clj

@@ -109,6 +109,22 @@
       "When on-caller? requested, but no reader can consume the value,
       put!'s callback executes on a different thread."))
 
+(deftest limit-async-take!-put!
+  (testing "async put! limit"
+    (let [c (chan)]
+      (dotimes [x 1024]
+        (put! c x))
+      (is (thrown? Exception                                             ;;;AssertionError
+                   (put! c 42)))
+      (is (= (<!! c) 0)))) ;; make sure the channel unlocks
+  (testing "async take! limit"
+    (let [c (chan)]
+      (dotimes [x 1024]
+        (take! c (fn [x])))
+      (is (thrown? Exception                                             ;;;AssertionError
+                   (take! c (fn [x]))))
+      (is (nil? (>!! c 42)))))) ;; make sure the channel unlocks
+ 
 (deftest puts-fulfill-when-buffer-available
   (is (= :proceeded
          (let [c (chan 1)