Clojure STM 笔记-上篇
年前有时间关注了一下Clojure,兴趣点在Lisp和其对并发的解决方案.对于后者,老聂推荐了一篇比较不错的文章:"Software Transactional Memory" .这篇文章是个很好的切入点,一方面可以学习Clojure,一方面可以借这个机会温习一下"程序设计语言-实践之路"的第12章"并发",不拘泥于文章本身.文章比较长,笔记分而治之,分上中下三篇完成.
user=> (def long-calculation (future (apply + (range 1e8))))#'user/long-calculationuser=> (type long-calculation)clojure.core$future_call$reify__6110user=> @long-calculation4999999950000000user=> (deref long-calculation)4999999950000000
future宏的作用,在meta里面有很清晰的描述:
?
user=> (source future)(defmacro future "Takes a body of expressions and yields a future object that will invoke the body in another thread, and will cache the result and return it on all subsequent calls to deref/@. If the computation has not yet finished, calls to deref/@ will block, unless the variant of deref with timeout is used. See also - realized?." {:added "1.1"} [& body] `(future-call (^{:once true} fn* [] ~@body)))nil上面的@符号是deref运算符的简记符.
promise可以有超时特性的解析引用对象值,解析过程是阻塞的直到有值.promise只能一次赋值.但promise的特质是并不创建最终为变量赋值的code或者function.promise初始化一个空容器,后续通过deliver填充数据.
user=> (def a (promise))#'user/auser=> (def b (promise))#'user/buser=> (def c (promise))#'user/cuser=> (future (deliver c (+ @a @b)) (println "Delivery complete!"))#<core$future_call$reify__6110@74aa513b: :pending>user=> (deliver a 101)#<core$promise$reify__6153@23293541: 101>user=> (deliver b 102)Delivery complete!#<core$promise$reify__6153@582b0e7b: 102>user=>
看看它的实现:
user=> (source promise)(defn promise "Alpha - subject to change. Returns a promise object that can be read with deref/@, and set, once only, with deliver. Calls to deref/@ prior to delivery will block, unless the variant of deref with timeout is used. All subsequent derefs will return the same delivered value without blocking. See also - realized?." {:added "1.1" :static true} [] (let [d (java.util.concurrent.CountDownLatch. 1) v (atom d)] (reify clojure.lang.IDeref (deref [_] (.await d) @v) clojure.lang.IBlockingDeref (deref [_ timeout-ms timeout-val] (if (.await d timeout-ms java.util.concurrent.TimeUnit/MILLISECONDS) @v timeout-val)) clojure.lang.IPending (isRealized [this] (zero? (.getCount d))) clojure.lang.IFn (invoke [this x] (when (and (pos? (.getCount d)) (compare-and-set! v d x)) (.countDown d) this)))))C# Parallel FX "程序设计语言-实践之路"里面提到的C# Parallel FX的例子是比较老了,用现在的写法举两个例子:
Task.Factory.StartNew(() => { Console.WriteLine("Hello Parallel Program."); }); Parallel.For(0, 100, (index) => { index.Dump(); }); //Console.WriteLine("Hello Parallel Program.");被外包到其它线程的进行计算,Task背后的实现是进程池. //下面这个例子使用了Lazy Lazy<Task<string>> lazyTask = new Lazy<Task<string>>(() => { return new Task<string>(() => { Console.WriteLine("Task Body working......"); return "Task Result"; }); }); Console.WriteLine("Calling lazy variable"); Console.WriteLine("Result from task: {0}", lazyData.Value.Result);继续看图, 上图中的STM简单展开说一下:
如果允许对最新数据进行更新,对历史版本数据查询,称这是Partially Persistent.如果允许对任何版本进行更新和查询,称这是Full Persistence.纯函数式编程语言(Pure Functional)所有数据结构都是fully persistent. STM可以有很多实现方式,Clojure实现与众不同的是"data coordinated by transactions cannot be modified outside a transaction."(需要事务中协调的数据状态不可以在事务外进行修改).这是在语言层面做了限制而非要求开发者自觉遵守规则. 就到这里,下一篇将继续对"Software Transactional Memory"的研读,关注Clojure语言处理并发的基础原语. 新年快乐! [1] Race condition http://en.wikipedia.org/wiki/Race_condition[2] The Transactional Memory / Garbage Collection Analogy[3] Software transactional memory [4] Persistent data structures [5] Persistent Data Structure及其应用