Concurrent Java 5 - Concurrent Data Strucutures

Optimistic Concurrency

這是一個implementation strategy，主是要給impelement concurrent library的人使用的，例如Java的AtomicInteger。

optimistic concurrency是假設multithreads在讀寫shared variables的時候，終究會發生atomic operation，也就是某一thread不受其他thread干擾，彷彿single thread一樣做了該做的事，當此情況發生(當然是opimistic)，即便在多執行續狀況中，答案就是正確的。

不過要這樣搞，就得要在implementation中加入retry 的機制，並且還是要有atomic construct才能達到，例如以下是AtomicInteger可能的使用optimistic concurrency strategy的implementation:

上圖中可以看到，GET_AND_ADD這個implementation有個while (true) loop，loop內就是optimistic trial，不過還\是得依賴COMPARE_AND_SET這個atomic operation，這個operation檢查cur是否還是原來的值，沒被其他thread汙染，如果是的話就set成新值，並且return。

這個implementation保證沒有deadlock，因為沒有任何lock。
也保證沒有livelock，雖然在此沒有證明。

Concurrent Queue

這個也應該會用optimistic concurrency 來比較efficiently implement:

注意tail應該是用AtomicReference instance，所以才會有COMPARE_AND_SET method。
BJ4。

Linearizability

這是在講concurrent program中，任一個thread的執行順序至少要符合其內的先後順序，例如先執行x再執行y，如果發現結果出現最後才執行x的話，就一定是有錯誤的。

某些operation可以linearizable去reason正確性，但有一些operation沒辦法，例如deleteAll()，因為沒有明確定義delete的順序。

Concurrent Minimum Spanning Tree

這邊嘗試把sequential的MST edge contraction演算法給parallelize:

這邊有幾個sychronization要注意:
1. 兩個vertices要被merge的話，應該都要acquire這兩個vertices的lock，避免不同thread同時對其中一個 (不可能是兩個，因為REMOVE是concurrent data structuree的operation，保證是thread-safe)做edge contraction。

2. data structure要選用concurrent data structure，例如ConcurrentLinkedQueue，這樣就能確保REMOVE / INSERT這樣的 operation是thread-safe。

Concurrent Java 4 - Actors

Even higher abstraction

isolation需要小心，因為同一個memory只要有個地方沒有用到isolation保護，整個邏輯可能就破功。

一個方法就是atomic variable，之前有提過。
另一個方法就是Actors pattern，他相當於是某個variable的代理人，透過message passing的機制來確保synchronization不會破功。

Actors

actors像是一個代理人，只對messages做出反應，他有三個部分：mailbox / methods / states，actors要保護的對象就是他的local states:

actors 可以動態建立啟動其他actors，形成一個pipeline:

上圖是一個actor pipeline來實作列印出所有prime numbers。

注意actors model裡面，所有參與者都必須是actors，才能完全在high level abstraction層面運作，否則就一定要面對lock / critical sections等。

Actors for UNBOUNDED Consumer-Buffer-Producer

一個可能的design如下：

producer actor 傳送INSERT message給buffer actor，如果有東西要放入。
buffer actor傳送 REMOVE message給 consumer actor (好奇怪），通知有東西可以拿。
consumer actor傳送 READY message給 buffer actor，通知consumed。

Actors for BOUNDED Consumer-Buffer-Producer

如果buffer是bounded，那塞入product的主動權可能落在buffer上面比較好，所以producer actor會被動的給予buffer actor product，如果收到REQUEST message:

不過可惜JAVA本身沒有實作Actors model 

Concurrent Java 3 - Higher level abstraction

Critical Sections (Isolation)

我們定義被(low level construct, e.g. lock)保護的某段code稱為critical section，critical section被視為一個atomic unit (mutual exclusion)，所以任何thread一旦開始執行critical section，其他的thread只會看到離開critical section的結果 (可能因為要acquire lock被block住)，而不會看到中間的state變化。

可以抽象的說，我們把此段code獨立出來了，稱為isolated。

Object-based Isolation (Monitor)

critical sections保護的是一段code，但是有時候其中牽涉到的物件其實並沒有shared memory，例如以下linked list deletion:

thread T1 delete B只牽涉到node A和node C，而thread T3 delete E只牽涉到 node D和node F，沒有道理不能同時執行delete(B)和delete(E)

所以我們需要一個object-based isolation，也就是跟關聯物件群有關的孤立，基本規則是:如果兩個isolation關聯的物件群是空集合的話，這兩個isolation就可以run in parallel，否則只能run mutual exclusively。

跟monitor的關係是: monitor是一個class，其instance method可能都isloated on 某個object set M。所以兩個monitors A和B，如果A和B的methods的isolation set都無交集的話，則使用這兩個monitor保護的code可以run in parallel。


一個範例是找出某個undirected graph的spanning tree (可能有多個可能)_:

可以用DFS來走片整個graph:

每個neighbor c可以生出一個thread來執行recursion COMPUTE(c)，但是在MAKEPARENT的method就要注意，有可能兩個vertices要搶當同一個neighbor c的parent，這時候就可以用object isolation on c來保證mutual exclusiveness。 當然同時也不會阻擋跟c無關的parallelism被犧牲，如果把整個MAKEPARENT用critical section保護的話就會犧牲了一些parallelism。

Java Atomic Variable: Object isolation semantics

Java atomic variable實現了部分object isolation semantics，對某個物件提供了atomic operation，並且針對需要atomic operation的patterns可以考慮使用atomic variable，因為這在硬體上有efficient implementation:

1. get and add pattern:

舉例來說，以下integer update就是一個 get-and-add pattern，可以使用atomic integer:

2. compare-and-set
如果要比對某個object reference是否相等才去做事，也可以用atomic reference。

Concurrent Java 2 - Unstructured lock vs Structured lock

Unstructured Lock

Java提供另一種比較彈性的lock機制: Lock interface。

Lock object跟每個object的intrinsic lock沒什不同，也是有wait/notify，但有個tryLock的method，會return一個boolean，這讓嘗試要acquire lock的thread不會被block住，可以去做其他的事，之後再來try。

但是麻煩的就是獲得lock的人一定要想辦法在對的地方呼叫unlock，即便是exception發生 (永遠在finally clause做unlock)，由於不會block structure自動acquire/release lock，所以稱為unstructured lock。

Lock l = ...;
     l.lock();
     try {
         // access the resource protected by this lock
     } finally {
         l.unlock();
     }

ReentrantLock

這是一個implement Lock interface的class。

ReadWriteLock

這比較特別，他能提供concurrent read lock，也就是多個thread 能夠同時acquire這個read lock的時候，可以concurrent執行被這個read lock "保護"的code，但是只有一個thread能acquire write lock，而且一旦write lock被acquire，則acuire read lock的threads全部被block，直到write lock被unlock。


Performance Comparison: Structured vs Unstructured
我們來比較在四核心上，structured vs unstructured對linked list的 read/write performance比較:

CoarseList是使用ReentrantLock ，RWCoarseList是使用ReentrantReadWriteLock ，而SyncList是使用intrinsic lock，以SyncList當baseline











(1) CoarseList vs. SyncList (Large Random)

=========================================================

1.0839688041594453x improvement in add throughput

0.9205057152753724x improvement in contains throughput

0.8414179104477613x improvement in remove throughput

=========================================================

可以看到overall ReentrantLock的overhead比intrinsic lock大一些。

(2) RWCoarseList vs. SyncList (Large Random)

=========================================================

0.7537446750034354x improvement in add throughput

3.2617096018735365x improvement in contains throughput

0.9670546105640107x improvement in remove throughput

=========================================================

ReentrantReadWriteLock 可以讓read operation的次數多很多，達到3倍以上。

(3) RWCoarseList vs. SyncList (Small Random)

=========================================================

1.1236133122028527x improvement in add throughput

3.2494758909853254x improvement in contains throughput

0.9303955933900853x improvement in remove throughput

=========================================================

(4) CoarseList vs. SyncList (Large Repeating)

=========================================================

0.7279840555104452x improvement in add throughput

0.7392840160315196x improvement in contains throughput

1.5410764872521248x improvement in remove throughput

=========================================================

(5) RWCoarseList vs. SyncList (Large Repeating)

=========================================================

0.7709759000297531x improvement in add throughput

3.6224951519069166x improvement in contains throughput

0.8615888615888616x improvement in remove throughput

=========================================================

(6) RWCoarseList vs. SyncList (Small Repeating)

=========================================================

0.6785995899700362x improvement in add throughput

3.8316566063044935x improvement in contains throughput

1.2968299711815563x improvement in remove throughput

=========================================================

Concurrent Java 1 - Structured / Intrinsic Lock / Monitors

Structured Lock (intrinsic lock / monitors)

這就是synchronized keyword，特徵就是thread mutual exclusion是implicit達成，沒有任何一個explicit lock出現。

每個Java object都有一個intrinsic lock，一個thread要對此object獲得mutual exclusive right的話，就要acquire這個object的intrinsic lock:

thread X: acquire A's lock ----- (own the lock) ------> release A's lock
thread Y: --------------block when try to acquire A's lock-------acquire A's lock ..........

一個thead 如何acquire object A's intrinsic lock?

1. 呼叫A的synchronized instance method會自動獲得A's lock，在method returns/throws exceptions 會自動release lock。如果是synchronized class method，則獲得此class的關聯Class object的instrinsic lock，保護static fields。

2. 使用 synchronized(A)，這可以放在比較fine-grained小區域保護範圍。

Reentrant Synchronization

Java lock可以被同一個thread acquire多次，這保護這個thread不會自己把自己block住了，所以不用害怕會發生這樣的事。不過既然稱為structured lock，則acquire lock的順序會配合上相反順序的release。

Wait / Notify

假設以下的程式在等待一個flag joy被設為true:

public void guardedJoy() {
    // Simple loop guard. Wastes
    // processor time. Don't do this!
    while(!joy) {}
    System.out.println("Joy has been achieved!");
}

某個thread要是執行這個method，就會在while loop裡面一直等待，但是OS還是會分配CPU resource來執行這個while loop，等於浪費CPU資源。如果把此method加上synchronized keyword，此時可以呼叫此object instance 的wait()來suspend這個thread:

public synchronized void guardedJoy() {
    // This guard only loops once for each special event, which may not
    // be the event we're waiting for.
    while(!joy) {
        try {
            wait(); //the thread releases the lock and is suspended
        } catch (InterruptedException e) {}
    }
    System.out.println("Joy and efficiency have been achieved!");
}

此thread被suspended，只有當某個Interrupt發生才會resume，例如其他的thread呼叫notifyAll，但是這不一定是我們在等待的interrupt (e.g. 不一定是joy被改變的事件)，所以一定要在檢查condition的loop中呼叫wait()，才能在interrupt發生的時候檢查condition來決定是否繼續wait。

為什麼要把此method加上synchronized? 其實是因為要呼叫此object的wait() method必須要acquire 此object's intrinsic lock。

一旦呼叫wait()，則thread會release此object lock然後suspended。

某個method會讓另一個thread acquire lock並且產生interrupt來notify所有suspended threads:

public synchronized notifyJoy() {
    joy = true;
    notifyAll();
}

被interrupt甦醒的thread 會重新獲得這個object's lock (因為另一個thread在method returns就release lock)，繼續在while loop中檢查是否要繼續wait。

Java Concurrency 6 - Don't Share!

預防勝於治療

不要share memory是最簡單規避multithreading coordination的方法！
怎麼做？

confined to a single thread

例如限定某個物件只能在某個thread上執行，進而實現了"singlethreaded"的要求。
不過這必須在program structure設計上著手，Java只能提供輔助性的class來達成thread confinement。不過這類的設計（稱為ad-hoc thread confinement）是很脆弱的，因為沒有語言層面的支援，純粹靠設計者的架構，通常用在UI Subsystem，讓UI access變成single threaded。

Stack confinement (within-thread, thread-local confinement)

local variable所在的記憶體scope會是同一個stack，所以宣告local variable就只有其包含的thread能access。這個有用到語言runtime結構的支持，所以比ad-hoc confinement好一些，可靠一些。

在Java中，沒有任何方式可以share local primitive type variable across threads。

但是對local object reference來說，是有可能被publish出去給其他thread看到的。另外雖然在withi-thread confinement的限制中，使用non-thread safety不會造成問題，但是這邊又變成依賴programmer的記性和架構，這是不利的地方，除非能夠被明確的documented。

ThreadLocal

較為正式的做法是使用Java ThreadLocal class。

簡單說來，就是某個object經過ThreadLocal包裹之後，每個thread都會有一個object copy，透過accessors來update或取得自己thread內的object，彼此互不干擾。

這通常是class中某個private static member，希望是可以在程式內是globally accessible，但是又希望每個thread有自己的值，例如每個thread有自己的global  temporary buffer，random ID等等。

例如以下範例，每個thread會拿到自己的一份connection instance:

這個ThreadLocal object會存在Thread object裡面，當thread結束後，自然也會被garbage collected。

不過由於ThreadLocal通常用在global object，所以還是要遵守盡量不要使用global object的原則。

Immutable Object （Java final keyword)

stateless object出生就是天然thread-safe，不廢話！
要做一個stateless class，其所有data members必須要是final，且沒有任何escape機會。

對了注意object reference immutable不等於object 真的immutable。“部分states immutable"的object還是比所有states都mutable的object來得少問題，盡量宣告自己的class field final。

Java Concurrency 5 - Publish objects

Publish ＆ Escape

我們把某個internal state / object 提供介面讓外面的程式存取的話，就稱為publish。

如果此object在我們不允許的時候被publish了，稱為escape!

最爛的publish

就是static object
@@～～～ 其實蠻多時候迫不得已要用啊？！

爛的原因在於：
(1)由於是static，thread-safety是個issue

(2)object本身雖然刻意被publish，但是不想被publish的相關物件可能也被publish了。例如此object所包含的data有可能通過public interface被存取，這同樣也有了thread safety的issue。

其次爛的publish: publish internal mutable variables

應該是要給setter，而不是直接給object reference，這樣很危險：

小心constructor ＥＳＣＡＰＥ

通常是在constructor，我們會init某些data member，給予某個anonymous listener當作參數丟入，不過這邊就是一個escape ! 因為首先 這個inner class instance是有一個hidden的"this" reference到外面的這個object instance，所以洩露了！

另一個危險是，故意去用這個escape instance的人也會陷入危險，因為ThisEscape object還沒有完成construction就洩露了object reference，很多state不一定會符合spec規定。

另外，別從constructor啟動另一個timer或是thread! 理由同上。建議把啟動thread的動作放到constructor之外，就不會有上述的問題。

Java Concurrency 4 - 小心compiler optimization造成的stale data!

奇怪的問題

以下的程式碼，我在工作中也是這樣寫，邏輯上似乎沒有問題才對，沒想到讀了這段文章之後，才知道我可能寫錯了！！！

public class NoVisibility {
    private static boolean ready;
    private static int number;
    private static class ReaderThread extends Thread {
        public void run() {
            while (!ready)
                Thread.yield();
            System.out.println(number);
        }
    }
    public static void main(String[] args) {
        new ReaderThread().start();
        number = 42;
        ready = true;
    } 
}

錯的原因在於ReaderThread不見得看得到ready和number這兩個變數的改變！為什麼會這樣？

ReaderThread可能發生情況包括：(1) 無窮迴圈 (2) 印出number為0 (3)按照我們的期望執行印出42

(1)(2)發生的原因難以理解，因為這是我們programmer看不到的邏輯：compiler optimization造成的execution reordering，因為對compiler來說某些指令如果沒有dependency的話，先後順序可以不按照程式行碼的順序來編排，這是為了多核心CPU所做的最佳化。

所以在main method中，創造新的ReaderThread, 寫入number = 42 和 寫入 ready = true這三行順序可能被compiler重新安排順序，因為兩者沒有關聯性，所以我們依照這個順序性成立的假設來實作ReaderThread的run method是錯誤的假設！

不過不理解為什麼會無窮迴圈？因為ready最終會被設為true，好奇怪？按照書中說法，當讀寫shared variable時候，其他thread不保證會看到ready = true被寫入這件事，書中提到可能是因為CPU cache造成ReaderThread讀到outdated ready value，這倒是說得通，不過這樣的optimization也未免太嚇人，multithreading的時候要注意optimization造成的side effect。


我個人在桌機執行上面的程式，儘管開了5條ReaderThread也是印出42，所以可能跟硬體和JVM compiler的設定有關。

不求甚解的我，conclude書中的話好了：反正有被multithreads shared的variables，都請用synchronization保護！

Stale data

上面例子說到ReaderThread可能看到非預期的變數值，因為compiler optimization造成的，這種過期（有時候混合正確的值）的資料稱為stale data，非常難以用邏輯理解（因為我們的邏輯需要建設在某些假設上，但這些假設在compiler optimization之後就不成立）。

重點在“讀取”shared variables也一定要synchronized保護，否則stale data是可能發生的。只有這樣能保證不同thread讀到最新的值。

一個可能發生的怪現象是JVM允許implementation把64 bit non-volatile long/double 用兩個32 bit read/write指令來完成指令來完成 原本應該是atomic operation的fetch and store operation，所以也有可能發生upper 或 lower 32 bits是stale value，相當恐怖，這一般應該查不出原因，如果不知道這麼眉角的話。

Java提供了volatile keyword，告訴compiler這個variable是shared，不要做reordering，可以解決這個問題，但是並沒有synchronization的保護。

Java concurrency 3 - intrinsic lock continued

Lock的持有與釋放

當一個lock被某個thread要求持有時，JVM會增值這個lock的計數器，當這個thread釋放這個lock的時候，計數器會減值。


JVM對lock的持有與釋放是以thread為請求單位，而非呼叫者（例如unix)，這個設計考量是因為OOP裡面通常subclass要呼叫super class的method來做initialization，如果是以呼叫者來當作請求單位的話，就會deadlock。

public class Widget {
    public synchronized void doSomething() {
        
    }
}
public class LoggingWidget extends Widget {
    public synchronized void doSomething() {
        System.out.println(toString() + ": calling doSomething");
        super.doSomething();
    }
}

一個class要保護哪些variables?

只要有被多執行緒存取（不止write，也包括read) 到的可能，都應該要被lock機制保護。我們很容易在啟動timer的程式中忘了要做synchronization保護，事實上這就是另一個thread嘗試要存取shared data。

另外如果一個應該是atomic operation的block，其中相關的shared variables都要受到保護，這個我們已經在上一篇文章中有範例說明。注意的是一個operation即便所有方法都是lock保護，這整個operation也不是thread safe，要將整個operation block保護才能說是thread-safe：

if (!vector.contains(element))
        vector.add(element);

vector的所有methods都是synchronized methods，但是在執行contains和add之間仍有可能有多執行緒造成的不同順序結果，不是thread-safe。

不必要的synchronization

我們之前說的白做工的SynchronizedFactorizer由於把最關鍵的service method宣告為synchronized，使得所有thread都要等待上一個thread完整執行整個block後才能進入，這會造成嚴重的效能問題，特別是這是一個web service!!

我們可以縮小lock保護的範圍：

@ThreadSafepublic class CachedFactorizer implements Servlet {
    @GuardedBy("this") private BigInteger lastNumber;
    @GuardedBy("this") private BigInteger[] lastFactors;
    @GuardedBy("this") private long hits;
    @GuardedBy("this") private long cacheHits;
    public synchronized long getHits() { return hits; }
    public synchronized double getCacheHitRatio() {
        return (double) cacheHits / (double) hits;
    }
    public void service(ServletRequest req, ServletResponse resp) {
        BigInteger i = extractFromRequest(req);
        BigInteger[] factors = null;
        synchronized (this) {
            ++hits;
            if (i.equals(lastNumber)) {
                ++cacheHits;
                factors = lastFactors.clone();
            }
        }
        if (factors == null) {
            factors = factor(i);
            synchronized (this)  {
                lastNumber = i;
                lastFactors = factors.clone();
            }
        }
        encodeIntoResponse(resp, factors);
    }
}

不過synchronization有overhead，所以不要分拆過細，只要包覆著atomic operation即可。

耗時間的block不要synchronized!

Java concurrency 2 - Locks

thread-safe variable不見得構成一個thread-safe class

如果我們要用更多的state在我們的class的話，有比較好的方法嗎? 例如以下:

@NotThreadSafepublic class UnsafeCachingFactorizer implements Servlet {
    private final AtomicReference<BigInteger> lastNumber           = new AtomicReference<BigInteger>();

    private final AtomicReference<BigInteger[]> lastFactors            = new AtomicReference<BigInteger[]>();

    public void service(ServletRequest req, ServletResponse resp) {
        BigInteger i = extractFromRequest(req);
        if (i.equals(lastNumber.get()))
            encodeIntoResponse(resp, lastFactors.get() );
        else {
            BigInteger[] factors = factor(i);
            lastNumber.set(i);
            lastFactors.set(factors);
            encodeIntoResponse(resp, factors);
        }
    }
}

我們想要也採用thread-safe object type來當作我們的data member type，不過不幸的是，

以上class並非thread-safe! 上面的問題在於，雖然兩個變數都用了AtomicReference來想要達成

thread-safe，但是lastNumer事實上是要等於lastFactors的乘積(後者為前者的因數)，所以在service 

method中，兩者在update的時候，並非一個完整的atomic operation，所以在multithreading的時候，

會造成race condition。

我們需要一個機制來將某些程式碼，組合成一個atomic operation。

Java內建的Lock (intrinsic lock / monitor / mutex)

Java提供了一個lock機制，利用synchronized這個關鍵字，可以保衛一段被{ } 框住的程式碼，確保

裡面的所有指令構成一個atomic operation。

所謂的lock，其實就是某個object reference，例如一個宣告synchroized的method，則保衛這個method

裡面所有指令的lock就是這個method所屬的instance:

synchronized void test() {
    //以下所有的程式碼都視為一個atomic operation    //lock為此class instance}

public static synchronized void staticTest() {
    //以下所有的程式碼都視為一個atomic operation    //lock為此class object}

public void test2() {
    synchronized (this) {
      //lock為this
}
}

這個lock只能一次被一個thread佔有，所以其他的thread要執行被lock保衛的block的時候，只能等待。

lock在一個thread開始進入block時獲得，在離開或丟出exception後釋放。

我們可以改寫 UnsafeCachingFactorizer 裡面的service method，使用intrinsic lock來保護這個method:\

@ThreadSafepublic class SynchronizedFactorizer implements Servlet {
    @GuardedBy("this") private BigInteger lastNumber;
    @GuardedBy("this") private BigInteger[] lastFactors;
    public synchronized void service(ServletRequest req,
                                     ServletResponse resp) {
        BigInteger i = extractFromRequest(req);
        if (i.equals(lastNumber))
            encodeIntoResponse(resp, lastFactors);
        else {
            BigInteger[] factors = factor(i);
            lastNumber = i;
            lastFactors = factors;
            encodeIntoResponse(resp, factors);
        }
    }
}

不過這其實就讓這個class變成non-multithreading class! 因為所有的thread都要排隊等待結果，等於

single thread的效能，白寫了!

Java concurrency 1 - Thread safe是什麼意思？

Thread-safe class

可以定義成：如果一個class instance在多個threads存取他的過程中，所有可能的執行緒的存取順序都能產出spec定義的行為，而且存取他的threads並不用做任何synchronization之類的工程，那這個class就可以說是thread safe。

Stateless object永遠是thread-safe的

下面這個class只有一個method，而且沒有任何data member，換句話說，不論是class instance或是method裡面都沒有state的存在，這讓access這個class instance的thread彷彿在access不同的instance，所以是一個thread safe class:

Non Atomic operation is not thread-safe

如果把上面的class加入一個state來計算request數目，我們可能會這樣做：

++count這行程式實際上不是atomic operation！所謂atomic operation是指單一thread會完整執行完畢產生預期效果後，此operation才會換其他thread執行。

++count實際上是三個CPU指令：(1)讀取couont (2)計算新的count值 = count+1 (3)寫入新的count值，這是所謂的read-modify-write operation。

兩個thread如果執行這個method，可能會產生以下的非預期狀況：

Race condition

一個多執行緒程式，如果必須要依賴時間點的巧合才能產生預期結果的話，那代表可能有race condition。一種常見的race condition pattern是check-then-act的設計：透過觀察某一個事件的發生，去決定要進行的動作。問題出在多執行緒的程式執行中，這個觀察到的事件，可能在要進行動作的時候就不成立了，主要是因為觀察和動作並非atomic。

一個常見的check-then-act設計是singleton pattern :

singleton pattern通常利用lazy initialization的技巧來達成單一instance的目的，但是由於getInstance()並沒有impelemnt成atomic operation，所以會產生race condition。

利用現有Thread-safe object來達成thread-safety

上面的UnsafeCountingFactorizer class，我們可以利用java提供的AtomicLong來取代long，進而保證thread-safety。

這是一個快速的解決方案，能如此達成的話，盡量先採取這個approach。