java_clj_haskell.txt

>SimpleExampleMVar 100000 1000 6
Starting...
2100000000
Computation time: 11.781 sec
Done.

>SimpleExampleSTM 100000 1000 6
Starting...
2100000000
Computation time: 53.797 sec
Done.

>java -cp classes SimpleLocking
Sum node values: 2100000000
Time: 15.703 sec

java -cp classes;%CLOJURE_JAR% simple_example lock 1000 6 100000
"Elapsed time: 27.545 secs"
Locking: 2100000000

java -cp classes;%CLOJURE_JAR% simple_example lock-native 1000 6 100000
"Elapsed time: 80.913 secs"
Native locking: 2100000000

java -cp classes;%CLOJURE_JAR% simple_example atom 1000 6 100000
"Elapsed time: 95.143 secs"
Atom: 2100000000

java -cp classes;%CLOJURE_JAR% simple_example stm 1000 6 100000
"Elapsed time: 990.255 secs"
STM: 2100000000

simple_example.clj

(ns simple-example (:gen-class))
(set! *warn-on-reflection* true)
;; tomado de: http://clojure.org/concurrent_programming  
;; usando vectores en lugar de secuencias
(import '(java.util.concurrent Executors Future)
	SimpleLocking$Node)  

(defn test-concur [iter refs nthreads niters]
  (let [pool  (Executors/newFixedThreadPool nthreads)  
        tasks (map (fn [t]  
                      (fn []  
                        (dotimes [n niters]  
                          (iter refs t))))  
                   (range nthreads))]  
    (doseq [^Future future (.invokeAll pool tasks)]  
      (.get future))  
    (.shutdown pool)))

(defn test-stm [nitems nthreads niters]  
  (let [refs  (vec (map ref (repeat nitems 0)))
	iter  #(dosync (doseq [r %]  (alter r + 1 %2)))]
    (test-concur iter  refs nthreads niters)
    (map deref refs))) 
 
(defn test-atom [nitems nthreads niters]  
  (let [refs  (vec (map atom (repeat nitems 0)))
	iter  #(doseq [r %]  (swap! r + 1 %2))]  
    (test-concur iter refs nthreads niters)
    (map deref refs)))

;; SimpleLocking$Node es la clase Node de mi version de java 
;; Hasta lo que se yo este codigo en clojure hace lo mismo que el main 
(defn test-locking [nitems nthreads niters] 
  (let [refs (->> (repeatedly #(SimpleLocking$Node.))
		  (take nitems) vec) 
	iter #(doseq [^SimpleLocking$Node n %]
		  (.sum n (+ 1 %2)))]
    (test-concur iter refs nthreads niters)
    (map (fn [^SimpleLocking$Node n] (.read n)) refs)))

(definterface INode
  (read [])
  (add [v]))

(deftype Node [^{:unsynchronized-mutable true} value]
  INode
  (read [_] value)
  (add [this v] (set! value (+ value v))))

(defn test-locking-native [nitems nthreads niters] 
  (let [refs (->> (repeatedly #(Node. 0))
		  (take nitems) vec) 
	iter #(doseq [^Node n %]
		  (locking n (.add n (+ 1 %2))))]
    (test-concur iter refs nthreads niters)
    (map (fn [^Node n] (.read n)) refs)))

(defn -main [& args]
  (read-line)
  (let [[type nitems nthreads niters] (map read-string args)
	t #(apply + (time (% nitems nthreads niters)))]
    (case type
	  lock (println "Locking:" (t test-locking)) 
	  atom (println "Atom:" (t test-atom))
	  stm (println "STM:" (t test-stm))
	  lock-native (println "Native locking:" (t test-locking-native)))))

SimpleExampleMVar.hs

{-# LANGUAGE BangPatterns #-} 
module SimpleExampleMvar where
import Control.Concurrent 
import Control.Monad 
import System.Environment(getArgs) 
import Data.Int 
import Text.Printf
import Control.Exception
import System.CPUTime
-- modified version of @__josejuan__ 
-- http://www.solveet.com/exercises/STM-vs-Bloqueos/119/solution-832 
time :: IO t -> IO t
time a = do
    start <- getCPUTime
    v <- a
    end   <- getCPUTime
    let diff = (fromIntegral (end - start)) / (10^12)
    printf "Computation time: %0.3f sec\n" (diff :: Double)
    return v
     
addMany :: MVar Int64 -> [MVar Int64] -> Int64 -> Int64 -> IO () 
addMany !c !a !n !i = forM_ [1..i] (\_ -> forM_ a (shW n)) >> shW 1 c 
 
shR = readMVar 
shW !k !r = 
  do x' <- takeMVar r 
     putMVar r $! (x' + k) -- <<<< ¡¡¡¡ éste es el culpable de todo !!!! 
 
main = do
    putStrLn "Starting..."
    time $ main'
    putStrLn "Done." 
 
main' = do 
  (niters': nvars': nthreads': _) <- getArgs 
  let (niters, nvars, nthreads) = (read niters', read nvars', read nthreads') 
  c <- newMVar 0 
  a <- mapM (\_ -> newMVar 0) [1..nvars] 
  mapM_ (\k -> forkIO $ addMany c a k niters) [1..nthreads] 
  waitTo c nthreads 
  z <- mapM shR a 
  putStrLn $ show $ sum z 
 
-- ¿no hay forma de hacer un "join" entre procesos? 
waitTo !c !nth = do 
  threadDelay (100*1000) 
  w' <- shR c 
  if w' == nth 
    then return () 
    else waitTo c nth 

SimpleExampleSTM.hs

{-# LANGUAGE BangPatterns #-} 
module SimpleExampleSTM where
import Control.Concurrent 
import Control.Monad 
import System.Environment(getArgs) 
import Data.Int 
import Control.Concurrent.STM
import Text.Printf
import Control.Exception
import System.CPUTime 
-- modified version of @__josejuan__ 
-- http://www.solveet.com/exercises/STM-vs-Bloqueos/119/solution-830
time :: IO t -> IO t
time a = do
    start <- getCPUTime
    v <- a
    end   <- getCPUTime
    let diff = (fromIntegral (end - start)) / (10^12)
    printf "Computation time: %0.3f sec\n" (diff :: Double)
    return v
 
addMany :: TVar Int64 -> [TVar Int64] -> Int64 -> Int64 -> IO () 
addMany !c !a !n !i = forM_ [1..i] (\_ -> forM_ a (shW n)) >> shW 1 c 
 
shR = atomically . readTVar 
shW !k !r = atomically $ 
  do x' <- readTVar r 
     writeTVar r $! (x' + k) -- <<<< ¡¡¡¡ éste es el culpable de todo !!!! 
 
main = do
    putStrLn "Starting..."
    time $ main'
    putStrLn "Done."  
 
main' = do 
  (x: y: z: _) <- getArgs 
  let (niters, nvars, nthreads) = (read x, read y, read z) 
  c <- atomically $ newTVar 0 
  a <- mapM (\_ -> atomically $ newTVar 0) [1..nvars] 
  mapM_ (\k -> forkIO $ addMany c a k niters) [1..nthreads] 
  waitTo c nthreads 
  z <- mapM shR a 
  putStrLn $ show $ sum z 
 
-- ¿no hay forma de hacer un "join" entre procesos? 
waitTo !c !nth = do 
  threadDelay (100*1000) 
  w' <- shR c 
  if w' == nth 
    then return () 
    else waitTo c nth 

SimpleLocking.java

import java.util.concurrent.*;
import java.util.*;
import static java.util.Collections.*;

public class SimpleLocking {
    public static void main (String[] args) {
	long begin=System.currentTimeMillis();
	int nitems=100;
	int nthreads=10;
	final int niters=1000;
	final Node[] g=new Node[nitems];
	for(int i=0;i<nitems;i++) g[i]=new Node();
	ExecutorService pool=Executors.newFixedThreadPool(nthreads);
	Callable<Object>[] tasks=new Callable[nthreads];
	for (int i=0;i<nthreads;i++) {
	    final int ii=i;
	    tasks[i]=new Callable<Object>() {
		public Object call() {
		    int z=niters;
		    while (z-- > 0)
			for(Node n : g) n.sum(ii+1);
		    return null;
		}};
	}
	try {
	    Collection<Future<Object>> futures=pool.invokeAll(Arrays.asList(tasks));
	    for (Future f : futures) f.get();
	}catch(Exception e){
	    e.printStackTrace();
	}finally { pool.shutdown();}
	long sum=0; for (Node n : g) sum+=n.read();
	long end=System.currentTimeMillis();
	System.out.println("Sum node values: "+sum);
	System.out.println("Time: "+(end-begin));
    }
    public static class Node {
	//private Object m=new Object();
	private int n=0;
	public synchronized void sum(int x) {n=n+x;}
	public int read() {return n;}
    }

}

threads.txt

CPU Profile with visualVM (self-time|invocations)
"Hot Spots - Method","Self time [%]","Self time","Invocations"

* java using manual locking:
"SimpleLocking$1.call()","53.661407","4851.595 ms","10"
"java.util.concurrent.locks.LockSupport.park(Object)","40.622433","3672.725 ms","86"
"java.util.concurrent.FutureTask$Sync.innerRun()","1.5622613","141.246 ms","10"
"java.lang.Thread.join(long)","0.90515286","81.836 ms","2"
"SimpleLocking$Node.sum(int)","0.65801543","59.492 ms","1000000"

* clojure using manual locking
"java.util.concurrent.locks.LockSupport.park(Object)","24.18844","19382.324 ms","10"
"java.net.SocketInputStream.read(byte[], int, int)","21.631039","17333.064 ms","198"
"simple_example$test_locking$iter__5712.invoke(Object, Object)","14.780218","11843.466 ms","10000"
"clojure.lang.Numbers.add(Object, Object)","7.621496","6107.145 ms","1000000"
"java.lang.reflect.Method.invoke(Object, Object[])","6.386695","5117.692 ms","4670"
"clojure.lang.Numbers$IntegerOps.add(Number, Number)","4.955416","3970.801 ms","1000000"
"java.lang.Integer.longValue()","3.5020137","2806.182 ms","2000000"
"clojure.lang.Numbers.ops(Object)","3.0372021","2433.726 ms","2000000"
"clojure.lang.ArrayChunk.nth(int)","1.9125521","1532.538 ms","1000000"
"clojure.lang.Numbers.lt(int, int)","1.0501778","841.513 ms","1060010"
"clojure.lang.Numbers$IntegerOps.combine(clojure.lang.Numbers.Ops)","0.94665796","758.562 ms","1000000"
"clojure.lang.Numbers.unchecked_inc(int)","0.893446","715.923 ms","1010000"

* clojure using atoms:
"java.net.SocketInputStream.read(byte[], int, int)","31.079458","115642.085 ms","923"
"java.util.concurrent.locks.LockSupport.park(Object)","15.172445","56454.433 ms","49"
"java.lang.reflect.Method.invoke(Object, Object[])","9.630574","35833.949 ms","25037"
"clojure.lang.Numbers.add(Object, Object)","3.386504","12600.682 ms","2007778"
"clojure.lang.RestFn.invoke(Object, Object, Object)","2.605458","9694.525 ms","1003889"
"clojure.core$r.invoke(Object, Object, Object)","2.5635095","9538.441 ms","1003889"
"clojure.lang.Numbers$IntegerOps.add(Number, Number)","2.4460144","9101.259 ms","2007778"
"clojure.lang.Var.get()","2.1666043","8061.615 ms","8203328"
"simple_example$test_atom$iter__5699.invoke(Object, Object)","1.5325866","5702.529 ms","10000"
"java.lang.Number.<init>()","1.2142249","4517.952 ms","2015281"
"java.lang.Integer.<init>(int)","1.2033328","4477.424 ms","2006978"
"clojure.lang.Numbers$IntegerOps.opsWith(clojure.lang.Numbers.IntegerOps)","1.1291922","4201.558 ms","2007778"
"clojure.lang.RT.seq(Object)","1.0424906","3878.954 ms","2067775"
"clojure.core$seq.invoke(Object)","1.0080996","3750.99 ms","2067775"
"java.lang.Integer.valueOf(int)","0.990004","3683.659 ms","2008757"
"clojure.lang.Numbers.ops(Object)","0.95642483","3558.716 ms","4015556"
"clojure.lang.Atom.swap(clojure.lang.IFn, Object, Object)","0.8817882","3281.004 ms","999998"

* clojure using stm
"java.util.concurrent.locks.LockSupport.parkNanos(Object, long)","53.995064","1352278.162 ms","52379"
"java.util.concurrent.locks.LockSupport.park(Object)","25.66637","642800.797 ms","14136"
"sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run()","10.56046","264481.191 ms","4"
"java.util.concurrent.locks.LockSupport.unpark(Thread)","1.2520777","31357.632 ms","84043"
"simple_example$test_stm$iter__5686.invoke(Object, Object)","0.23980345","6005.752 ms","10000"
"clojure.lang.Numbers.add(Object, Object)","0.19171205","4801.328 ms","2852124"
"clojure.core$r.invoke(Object, Object, Object)","0.17566432","4399.421 ms","1426062"
"clojure.lang.ASeq.<init>()","0.15851691","3969.973 ms","5897305"
"clojure.lang.Var.get()","0.15223883","3812.742 ms","11978843"
"clojure.lang.RestFn.applyTo(clojure.lang.ISeq)","0.1460502","3657.751 ms","1426062"
"java.util.HashMap.getEntry(Object)","0.14354251","3594.947 ms","4263732"
"simple_example$test_stm$iter__5686$fn__5687.invoke()","0.14162919","3547.029 ms","80685"
"clojure.lang.LockingTransaction.doSet(clojure.lang.Ref, Object)","0.12876955","3224.966 ms","1426062"
"clojure.lang.ArraySeq.<init>(Object, int)","0.12827718","3212.635 ms","4283487"
"clojure.lang.RestFn.invoke(Object, Object, Object, Object)","0.1268663","3177.3 ms","1431363"
"java.util.HashMap.put(Object, Object)","0.12443997","3116.534 ms","2780173"
"clojure.lang.Numbers$IntegerOps.add(Number, Number)","0.12424572","3111.669 ms","2852124"

* clojure using stm