cells.clj

;   Copyright (c) Rich Hickey. All rights reserved.
;   The use and distribution terms for this software are covered by the
;   Eclipse Public License 1.0 (http://opensource.org/licenses/eclipse-1.0.php)
;   which can be found in the file epl-v10.html at the root of this distribution.
;   By using this software in any fashion, you are agreeing to be bound by
;   the terms of this license.
;   You must not remove this notice, or any other, from this software.

(set! *warn-on-reflection* true)

(defprotocol Sentry
  (make-cell [sentry val] [sentry val transient]))

(defprotocol Cell
  (cell-sentry [cell])
  (cell-get-transient [cell])
  (cell-set-transient [cell t])
  (cell-render [cell]))

(defprotocol Editable
  (transient-of [value sentry]))

(defprotocol Transient
  (value-of [transient]))

(defonce #^java.lang.ThreadLocal tl-in-cells (ThreadLocal.))


(deftype ThreadCell [thread 
                     #^{:unsynchronized-mutable true} val
                     #^{:unsynchronized-mutable true} trans
                     __meta]
  Object
  (equals [this o] (identical? this o))
  (hashCode [this] (System/identityHashCode this))

  clojure.lang.IMeta
  (meta [_] __meta)

  clojure.lang.IDeref
  (deref [this] (cell-render this))

  Cell
  (cell-sentry [_] thread)
  (cell-get-transient [_] 
    (assert (identical? (Thread/currentThread) thread))
    (when (identical? ::none trans)
      (set! trans (transient-of val thread)))
    trans)
  (cell-set-transient [this t] (set! trans t) this)
  (cell-render [_] 
    (assert (identical? (Thread/currentThread) thread))
    (when-not (identical? trans ::none)
      (set! val (value-of trans))
      (set! trans ::none))
    val))

(deftype LockCell [#^java.util.concurrent.locks.ReentrantLock lock 
                   #^{:volatile-mutable true} val
                   #^{:unsynchronized-mutable true} trans
                   __meta]
  Object
  (equals [this o] (identical? this o))
  (hashCode [this] (System/identityHashCode this))

  Comparable
  (compareTo [this o] 
    (cond (identical? lock (:lock o)) 0
          (< (hash lock) (hash (:lock o))) -1
          (> (hash lock) (hash (:lock o))) 1
          :else (throw (IllegalStateException. (str "Duplicate lock hashes for distinct locks: " this " " o)))))

  clojure.lang.IMeta
  (meta [_] __meta)

  clojure.lang.IDeref
  (deref [this] 
    (if (.isHeldByCurrentThread lock) 
      (cell-render this)
      val))

  Cell
  (cell-sentry [_] lock)
  (cell-get-transient [_] 
    (assert (.isHeldByCurrentThread lock))
    (when (identical? ::none trans)
      (set! trans (transient-of val lock)))
    trans)
  (cell-set-transient [this t] (set! trans t) this)
  (cell-render [_] 
    (assert (.isHeldByCurrentThread lock))
    (when-not (identical? trans ::none)
      (set! val (value-of trans))
      (set! trans ::none))
    val))

(extend-protocol Sentry
  java.lang.Thread
  (make-cell 
   ([thread val] (ThreadCell. thread val ::none))
   ([thread val transient] (ThreadCell. thread val transient)))

  java.util.concurrent.locks.ReentrantLock
  (make-cell 
   ([lock val] (LockCell. lock val ::none))
   ([lock val transient] (LockCell. lock val transient))))

(defmacro pass [f cell & args]
  `(cell-set-transient ~cell (~f ~(with-meta `(cell-get-transient ~cell) (meta cell)) ~@args)))

(defmacro fetch [f cell & args]
  `(~f ~(with-meta `(cell-get-transient ~cell) (meta cell)) ~@args))

(def #^{:macro true} >> (deref (var pass)))
(def #^{:macro true} << (deref (var fetch)))

(defn cell 
  ([val] 
     (make-cell (Thread/currentThread) val))
  ([val transient] 
     (make-cell (Thread/currentThread) val transient)))

(defn locked-cell 
  ([val] 
     (make-cell (java.util.concurrent.locks.ReentrantLock.) val))
  ([val transient] 
     (make-cell (java.util.concurrent.locks.ReentrantLock.) val transient)))

(defn in-cells-fn [cells f]
  (assert (nil? (.get tl-in-cells)))
  (let [s (java.util.TreeSet. #^java.util.Collection cells)
        unlock-all #(doseq [cell %]
                      (let [lock #^java.util.concurrent.locks.ReentrantLock (:lock cell)]
                        (when (.isHeldByCurrentThread lock) (.unlock lock))))]
    (try
     (.set tl-in-cells true)
     (doseq [cell s]
       (assert (:lock cell))
       (.lock #^java.util.concurrent.locks.ReentrantLock (:lock cell)))
     (f)
     (finally
      (.set tl-in-cells nil)
      (unlock-all s)))))

(defmacro in-cells [cells & body]
  (case (count cells)
    0 `(do ~@body)
    1 `(let [lock#  #^java.util.concurrent.locks.ReentrantLock (:lock ~(first cells))] 
         (assert (identical? nil (.get tl-in-cells)))
         ;(assert lock#)
         (do
           (.set tl-in-cells true)
           (.lock lock#)
           (try
            ~@body
            (finally
             (.set tl-in-cells nil)
             (.unlock lock#)))))
    `(in-cells-fn [~@cells] (fn [] ~@body))))

(extend-type String
  Editable
  (transient-of [s _] (StringBuilder. s)))

(extend-type StringBuilder
  Transient
  (value-of [sb] (.toString sb)))

(extend-type clojure.lang.IEditableCollection
  Editable
  (transient-of [coll _] (.asTransient coll)))

(extend-type clojure.lang.ITransientCollection
  Transient
  (value-of [coll] (.persistent coll)))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; seqiters ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(defprotocol Iter
  (has-item [iter])
  (item [iter])
  (move! [iter]))

(definterface IterSeqImpl
  (mkseq []))

(deftype IterSeq [#^{:unsynchronized-mutable true} iter-cell 
                  #^{:unsynchronized-mutable true :tag clojure.lang.ISeq} seq-val]

  Iter
  (has-item [this] (.seq this))
  (item [this] (.first this))
  (move! [this] (.more this))

  user.IterSeqImpl
  (mkseq [this] 
    (locking this
      (when iter-cell
        (set! seq-val
          (when (<< has-item iter-cell)
            (clojure.lang.Cons. (<< item iter-cell) (IterSeq. (>> move! iter-cell) nil))))
        (set! iter-cell nil))
      seq-val))

  clojure.lang.Seqable
  (seq [this] (.mkseq this))

  clojure.lang.ISeq
  (first [this] (.mkseq this) (if (nil? seq-val) nil (.first seq-val)))
  (more [this] (.mkseq this) (if (nil? seq-val) () (.more seq-val)))

  user.Editable
  (transient-of [_ sentry] (transient-of (if iter-cell @iter-cell seq-val) sentry)))

(defn iter-seq [iter]
  (IterSeq. (cell nil iter) nil))

;todo - factor out the commonality below into a macro

(defn mapx
  "Returns a lazy sequence consisting of the result of applying f to the
  set of first items of each coll, followed by applying f to the set
  of second items in each coll, until any one of the colls is
  exhausted.  Any remaining items in other colls are ignored. Function
  f should accept number-of-colls arguments."
  ([f coll]
     (letfn [(iter [f seq-cell]
               (reify
                 Iter
                   (has-item [_] (<< has-item seq-cell))
                   (item [_] (f (<< item seq-cell)))
                   (move! [this] (>> move! seq-cell) this)
                 Transient 
                   (value-of [_] 
                     (reify Editable
                        (transient-of [_ sentry] 
                          (iter f (make-cell sentry (sequence @seq-cell))))))))]
       (iter-seq (iter f (cell (sequence coll)))))))

(defn filterx 
  "Returns a lazy sequence of the items in coll for which
  (pred item) returns true. pred must be free of side-effects."
  [pred coll]
  (letfn [(iter [pred seq-cell]
            (reify
              Iter
                (has-item [_] 
                  (loop [cc seq-cell] 
                    (when (and (<< has-item cc) (not (pred (<< item cc))))
                      (recur (>> move! cc))))
                  (<< has-item seq-cell))
                (item [_] (<< item seq-cell))
                (move! [this] (>> move! seq-cell) this)
              Transient 
                (value-of [_] 
                  (reify Editable
                    (transient-of [_ sentry] 
                      (iter pred (make-cell sentry (sequence @seq-cell))))))))]
    (iter-seq (iter pred (cell (sequence coll))))))

(defn takex
  "Returns a lazy sequence of the first n items in coll, or all items if
  there are fewer than n."
  [n coll]
  (letfn [(iter [ai n seq-cell]
            (reify
              Iter
                (has-item [_] (and (<< has-item seq-cell) (< @ai n)))
                (item [_] (<< item seq-cell))
                (move! [this] (swap! ai inc) (>> move! seq-cell) this)
              Transient 
                (value-of [_] 
                  (reify Editable
                    (transient-of [_ sentry] 
                      (iter (atom @ai) n (make-cell sentry (sequence @seq-cell))))))))]
    (iter-seq (iter (atom 0) n (cell (sequence coll))))))

(extend-type clojure.lang.LazySeq
  Editable
  (transient-of [ls _] ls)
  Iter
  (has-item [ls] (.seq ls))
  (item [ls] (.first ls))
  (move! [ls] (.more ls)))

(extend-type clojure.lang.ASeq
  Editable
  (transient-of [ls _] ls)
  Iter
  (has-item [ls] (.seq ls))
  (item [ls] (.first ls))
  (move! [ls] (.more ls)))

(extend-type clojure.lang.PersistentList$EmptyList
  Editable
  (transient-of [ls _] ls)
  Iter
  (has-item [ls] nil)
  (item [ls] nil)
  (move! [ls] ls))

(defn reducex [f init coll]
  (let [cc (cell (sequence coll))]
    (loop [ret init cc cc]
      (if (<< has-item cc)
        (recur (f ret (<< item cc))
               (>> move! cc))
        ret))))

(deftype Range [comp #^{:unsynchronized-mutable true} start end step]
  Iter
  (has-item [_] (comp start end))
  (item [_] start)
  (move! [this] (set! start (+ start step)) this)
  Transient 
  (value-of [_] 
    (reify Editable
      (transient-of [_ _] (Range. comp start end step)))))

(defn rangex [start end step]
  (let [comp (if (pos? step) < >)]
    (iter-seq (Range. comp start end step))))

(reducex + 0 (mapx inc (filterx even? (rangex 0 1000000 1))))