Merge pull request #243 from pixie-lang/csp

CSP anyone?

Author: halgari

pixie/buffers.pxi

@@ -0,0 +1,146 @@
+(ns pixie.buffers)
+
+(defn acopy [src src-start dest dest-start len]
+  (loop [cnt 0]
+    (when (< cnt len)
+      (aset dest
+            (+ dest-start cnt)
+            (aget src (+ src-start cnt)))
+      (recur (inc cnt)))))
+
+
+(defprotocol IMutableBuffer
+  (remove! [this])
+  (add! [this])
+  (full? [this]))
+
+(defprotocol IResizableMutableBuffer
+  (add-unbounded! [this val])
+  (resize! [this new-size]))
+
+(deftype RingBuffer [head tail length arr]
+  IMutableBuffer
+  (remove! [this]
+    (when-not (zero? length)
+      (let [x (aget arr tail)]
+        (aset arr tail nil)
+        (set-field! this :tail (int (rem (inc tail) (alength arr))))
+        (set-field! this :length (dec length))
+        x)))
+  (add! [this x]
+    (assert (< length (alength arr)))
+    (aset arr head x)
+    (set-field! this :head (int (rem (inc head) (alength arr))))
+    (set-field! this :length (inc length))
+    nil)
+
+  (full? [this]
+    (= length (alength arr)))
+
+
+  IResizableMutableBuffer
+  (resize! [this new-size]
+    (let [new-arr (make-array new-size)]
+      (cond
+       (< tail head)
+       (do (acopy arr tail new-arr 0 length)
+           (set-field! this :tail 0)
+           (set-field! this :head length)
+           (set-field! this :arr new-arr))
+
+       (> tail head)
+       (do (acopy arr tail new-arr 0 (- (alength arr) tail))
+           (acopy arr 0 new-arr (- (alength arr) tail) head)
+           (set-field! this :tail 0)
+           (set-field! this :head length)
+           (set-field! this :arr new-arr))
+
+
+       (full? this)
+       (do (acopy arr tail new-arr 0 length)
+           (set-field! this :tail 0)
+           (set-field! this :head length)
+           (set-field! this :arr new-arr))
+
+
+       :else
+       (do (set-field! this :tail 0)
+           (set-field! this :head 0)
+           (set-field! this :arr new-arr)))))
+
+  (add-unbounded! [this val]
+    (when (full? this)
+      (resize! this (* 2 length)))
+    (add! this val))
+
+  ICounted
+  (-count [this]
+    length))
+
+
+(defn ring-buffer [size]
+  (assert (> size 0) "Can't create a ring buffer of size <= 0")
+  (->RingBuffer 0 0 0 (make-array size)))
+
+
+(defn fixed-buffer [size]
+  (ring-buffer size))
+
+
+(deftype DroppingBuffer [buf]
+  IMutableBuffer
+  (full? [this]
+    false)
+  (remove! [this]
+    (remove! buf))
+  (add! [this val]
+    (when-not (full? buf)
+      (add! buf val)))
+
+  ICounted
+  (-count [this]
+    (count buf)))
+
+(defn dropping-buffer [size]
+  (->DroppingBuffer (ring-buffer size)))
+
+
+(deftype SlidingBuffer [buf]
+  IMutableBuffer
+  (full? [this]
+    false)
+  (remove! [this]
+    (remove! buf))
+  (add! [this val]
+    (when (full? buf)
+      (remove! buf))
+    (add! buf val))
+
+  ICounted
+  (-count [this]
+    (count buf)))
+
+(defn sliding-buffer [size]
+  (->SlidingBuffer (ring-buffer size)))
+
+(defn empty-buffer? [buf]
+  (= (count buf) 0))
+
+
+(deftype NullBuffer []
+  IMutableBuffer
+  (full? [this]
+    true)
+  ICounted
+  (-count [this] 0))
+
+(def null-buffer (->NullBuffer))
+
+(extend -reduce IMutableBuffer
+        (fn [buf f acc]
+          (loop [acc acc]
+            (if (reduced? acc)
+              @acc
+              (if (pos? (count buf))
+                (recur (f acc (remove! buf)))
+                acc)))))

pixie/channels.pxi

@@ -0,0 +1,183 @@
+(ns pixie.channels
+  (require pixie.stacklets :as st)
+  (require pixie.buffers :as b))
+
+(defprotocol ICancelable
+  (-canceled? [this] "Determines if a request (such as a callback) that can be canceled")
+  (-commit! [this]))
+
+(defprotocol IReadPort
+  (-take! [this cfn] "Take a value from this port passing it to a cancellable function"))
+
+(defprotocol IWritePort
+  (-put! [this itm cfn] "Write a value to this port passing true if the write succeeds and the
+                         callback isn't canceled"))
+
+(defprotocol ICloseable
+  (-close! [this] "Closes the channel, future writes will be rejected, future reads will
+                   drain the channel before returning nil."))
+
+(deftype OpCell [val cfn]
+  IIndexed
+  (-nth [this idx]
+    (cond
+     (= idx 0) val
+     (= idx 1) cfn
+     :else (throw "Index out of range")))
+  (-nth-not-found [this idx not-found]
+    (cond
+     (= idx 0) val
+     (= idx 1) cfn
+     :else not-found))
+  ICounted
+  (-count [this]
+    2)
+  ICancelable
+  (-canceled? [this]
+    (canceled? cfn)))
+
+(defn canceled? [this]
+  (-canceled? this))
+
+
+(defn -move-puts-to-buffer [puts buffer]
+  (loop []
+    (if (or (b/full? buffer)
+            (b/empty-buffer? puts))
+      nil
+      (let [[val cfn] (b/remove! puts)]
+        (if (cancelled? cfn)
+          (recur)
+          (do (st/-run-later (partial cfn true))
+              (b/add! buffer val)
+              (recur)))))))
+
+(defn -get-non-canceled! [buffer]
+  (loop []
+    (if (b/empty-buffer? buffer)
+      nil
+      (let [v (b/remove! buffer)]
+        (if (canceled? v)
+          (recur)
+          v)))))
+
+
+(deftype MultiReaderWriterChannel [puts takes buffer closed? ops-since-last-clean]
+  IReadPort
+  (-take! [this cfn]
+    (if (canceled? cfn)
+      false
+      (if (and closed?
+               (b/empty-buffer? buffer)
+               (b/empty-buffer? puts))
+        (do (-commit! cfn)
+            (st/-run-later (partial cfn nil))
+            false)
+        (if (not (b/empty-buffer? buffer))
+          (do (-commit! cfn)
+              (st/-run-later (partial cfn (b/remove! buffer)))
+              (-move-puts-to-buffer puts buffer))
+
+          (if-let [[v pcfn] (-get-non-canceled! puts)]
+            (do (-commit! pcfn)
+                (-commit! cfn)
+                (st/-run-later (partial pcfn true))
+                (st/-run-later (partial cfn v))
+                true)
+            (do (set-field! this :ops-since-last-clean (inc ops-since-last-clean))
+                (b/add-unbounded! takes cfn)
+                true))))))
+  IWritePort
+  (-put! [this val cfn]
+    (if (or (canceled? cfn))
+      false
+      (if closed?
+        (do (-commit! cfn)
+            (st/-run-later (partial cfn false))
+            false)
+        (if-let [tfn (-get-non-canceled! takes)]
+          (do (-commit! cfn)
+              (-commit! tfn)
+              (st/-run-later (partial tfn val))
+              (st/-run-later (partial cfn true))
+              true)
+          (if (not (b/full? buffer))
+            (do (b/add! buffer val)
+                (-commit! cfn)
+                (st/-run-later (partial cfn true))
+                true)
+            (do (b/add-unbounded! puts (->OpCell val cfn))
+                (set-field! this :ops-since-last-clean (inc ops-since-last-clean))
+                true))))))
+  ICloseable
+  (-close! [this]
+    (set-field! this :closed? true)
+    (when (not (b/empty-buffer? takes))
+      (loop []
+        (when-let [tfn (-get-non-canceled! takes)]
+          (-commit! tfn)
+          (st/-run-later (partial tfn nil))
+          (recur))))))
+
+(defn chan
+  "Creates a CSP channel with the given buffer. If an integer is provided as the argument
+   creates a channel with a fixed buffer of that size. "
+  ([]
+   (chan 0))
+  ([size-or-buffer]
+    (if (= 0 size-or-buffer)
+      (->MultiReaderWriterChannel (b/ring-buffer 8)
+                                  (b/ring-buffer 8)
+                                  b/null-buffer
+                                  false
+                                  0)
+      (if (integer? size-or-buffer)
+        (->MultiReaderWriterChannel (b/ring-buffer 8)
+                                    (b/ring-buffer 8)
+                                    (b/fixed-buffer size-or-buffer)
+                                    false
+                                    0)
+        (->MultiReaderWriterChannel (b/ring-buffer 8)
+                                    (b/ring-buffer 8)
+                                    size-or-buffer
+                                    false
+                                    0)))))
+
+(deftype AltHandler [atm f]
+  ICancelable
+  (-canceled? [this]
+    @atm)
+  (-commit! [this]
+    (reset! atm true))
+  IFn
+  (-invoke [this & args]
+    (apply f args)))
+
+(defn alt-handlers [fns]
+  (mapv (partial ->AltHandler (atom false)) fns))
+
+(extend -canceled? IFn
+        (fn [this] false))
+
+(extend -commit! IFn
+        (fn [this] nil))
+
+(defn alts! [ops k options]
+  (let [handler-atom (atom false)]
+    (reduce
+     (fn [_ op]
+       (if (vector? op)
+         (let [[c val] op
+               f (fn [v]
+                   (st/-run-later (partial k [c v])))]
+           (-put! c val (->AltHandler handler-atom f)))
+         (let [c op
+               f (fn [v]
+                   (st/-run-later (partial k [c v])))]
+           (-take! c (->AltHandler handler-atom f)))))
+     nil
+     ops)
+    (when (and (contains? options :default)
+               (not @handler-atom))
+      (reset! handler-atom true)
+      (st/-run-later (partial k [:default (:default options)])))))

pixie/csp.pxi

@@ -0,0 +1,62 @@
+(ns pixie.csp
+  (require pixie.stacklets :as st)
+  (require pixie.buffers :as b)
+  (require pixie.channels :as chans))
+
+(def chan chans/chan)
+
+(defn close!
+  "Closes the channel, future writes will be rejected, future reads will
+   drain the channel before returning nil."
+  [c]
+  (chans/-close! c))
+
+(def -null-callback (fn [_] nil))
+
+(defn put!
+  "Puts the value into the channel, calling the optional callback when the operation has
+   completed."
+  ([c v]
+   (chans/-put! c v -null-callback))
+  ([c v f]
+   (chans/-put! c v f)))
+
+(defn take!
+  "Takes a value from a channel, calling the provided callback when completed"
+  ([c f]
+   (chans/-take! c f)))
+
+(defn >! [c v]
+  (st/call-cc (fn [k]
+                (chans/-put! c v (partial st/run-and-process k)))))
+
+(defn <! [c]
+  (st/call-cc (fn [k]
+                (chans/-take! c (partial st/run-and-process k)))))
+
+
+(defmacro go [& body]
+  `(let [ret-chan# (chans/chan 1)]
+     (st/spawn (put! ret-chan# (do ~@body))
+               (close! ret-chan#))
+     ret-chan#))
+
+
+(extend -reduce chans/IReadPort
+        (fn [c f init]
+          (loop [acc init]
+            (if (reduced? acc)
+              @acc
+              (let [v (<! c)]
+                (if (nil? v)
+                  acc
+                  (recur (f acc v))))))))
+
+
+(defn alts!
+  ([ops]
+   (st/call-cc (fn [k]
+                 (chans/alts! ops (partial st/run-and-process k) nil))))
+  ([ops & opts]
+   (st/call-cc (fn [k]
+                 (chans/alts! ops (partial st/run-and-process k) (apply hashmap opts))))))