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 

Parallel Java 1 - Task Parallelism

Async and Finish primitives

這是用high level眼光來看parallelism。

SUM = SUM1 + SUM2

我們說SUM1 和 SUM2 run asynchronizely ，也就是獨立運作，可能in parallel (在不同的CPU core) or not都有可能。

當SUM1 和 SUM2都完成computation，我們說finish，所以finish是定義所有async tasks的scope，而SUM需要等finish才能得到正確答案。

在Java中，async可以用fork()來實現。
finish可以用join()來實現。

例如以下：

或是可以把用parallelize的objects都放入invoke():

這樣會自動fork / join。

Modelling Parallelism

我們可以用graph來model parallelism，假設我們有以下的async/finish execution:

我們可以用以下的graph來model:

directed path意味著sequential dependency，所以沒有被path連接的node tasks，就是可以run in parallel，以上圖來說就是S2 S3。

Work and Span

要衡量computation graph的瓶頸，我們可以引入work / span衡量。
work: 所有node task的cost總和
span: 某條最長path的work

以上圖來說，work = 1 + 10 + 10 + 1 = 22
其中有兩條path的work 都是最大的，是12，所以span = 12

span可以說是瓶頸。
所以如果把 graph work / span稱為"ideal parallelism"，我們可以得出一個"多少parallelism"的metric，因為如果能夠parallelized tasks越多，則span越少，那parallelism越大。

ideal parallelism是這個computation graph的parallelism upper bound。

Execution time estimation on multi-core processors

假設有一個parallel computation graph如下：

如果在2-cord processor上面，依照OS的scheduler的演算法，有可能造成不同的分配給CPU的結果。定義Tn = 在n-core上需要執行的cost：

注意T1 = work，因為單一cpu壹定要完成所有的task。
T_inf  = span = 最大的path work，因為其他的tasks都可以跟最大path tasks run in parallel，這使得span成為瓶頸。

所以   T_inf <= Tp <= T1
我們定義 speedup = T1/Tp ，這很好理解。

speedup一定 <= p，因為speedup = p是最理想狀態，發生在work被平均分配給p個processor的graph，則speedup = T1/ (T1/p) = p。不過實際上通常會有一個longest path的出現。

另外speedup <= ideal parallelism，因為之前說過ideal parallelism是一個computation graph parallelism upper bound。

Amdahl's law

這定律很好證明，如果知道一個parallel program中sequential part佔了q %，則speedup <=  1/q。例如假設q = 50%，則直覺來想如果你的program平行化之後，只有50%能run in parallel，那當然最多只能獲得比現在執行時間少一半的效能，也就是 1/0.5 = 2。

這個定律在我們不知道computation graph時來預估speedup有用，證明如下：

1. span >= q * work，這容易理解，因為span就是最大的 path cost，當然 >= 那q percentage的不能被parallelize的部分的work (?? 這解釋有點疑問)

2. 又我們知道 speedup <= work/span這個upper bound，把1. 中的span帶入:
speedup <= 1/q

得證。

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。

Scala Parallel Programming課後心得

上完了！很短的課程。

課程內容我覺得稍微有點失望，主要是習題設計的問題。
習題多數時候比較是在implement application logic，但不見得是parallel construct的運用，所以其實我還是有一點一知半解。

比較好的習題方式應該是多個小的習題，但是每個習題都是給予一個sequential mini project，要我們怎麼利用lecture中交的來改成parallel program，然後比較performance，這樣就很容易懂和進步，因為畢竟之後真的要運用在工作中的話，一定就是找尋可以parallelize的sequential parts，去做performance tuning。

第四個禮拜教了某個parallel data structure的implementation，這有好有壞，不過可惜的是本課程老師口音有點難理解，字幕又是自動翻譯的，常常不正確，造成我很難做筆記，最後就放棄不做了，剛好作業其實跟parallelization的實作相關性不大，變成也不需要太理解lecture video中的內容，也是能過關。

不是很推薦本課程，不過如果想要拿到specialization，就只好拿囉，硬要給分數的話，會給75分。

Scala Parallel Programming筆記 12 - Conc Trees

why Conc Tree?

之前提到，我們採用intermediate data structure來解決"efficient" combining問題。但是另一個可能就是，是否存在本來就是efficient combine operation （concatenate/union)的data structure?

List & Tree

scala list定義如下：（又稱為conc list)

list天生就是sequential，但是tree可以parallel：

我們可以用以下的方法來implement tree parallel filter:

問題是這樣的implementation不能給出一個適合parallelization的balanced tree， 中間的tree不是optimal height因為Leaf case仍會return Empty，如果修改不讓Leaf case return Empty的話，會變成最右邊那樣，像list而非tree，非常不balanced:

Conclist  - balanced tree

為了達成balance，我們要另一個data structure Conc，trait 定義如下：

level :就是height

size: 就是這個tree有幾個elements

以下為Conc trait可能的concrete implementation:

conc tree有invariants必須要遵守：

1. <> (inner node) 不能有Empty child

2. <>的左右children的level差 <= 1，為了要達成balanced tree。以下這個tree就違反了invariant 2:

以下這個才符合variant 2:

所以invariant 2 保證conc tree的operations至少能在O(log n)完成。

我們可以定義一個 <> method，用來當作concatenation (注意這根constructor 名字一樣）:

這method裡面先檢查了invariant 1，然後真正做balancing (滿足invariant 2)的是concat method，concat method就檢查invariant 2，如果滿足的話就new 一個inner node <>當成parent node：

否則的話，tree xs和tree ys必須要做combination。
先假設xs的height = d，ys的height = e <= d-2 。
所以要看xs的trees的leaning狀況（哪邊的tree較深）如何來決定要怎麼merge。


1. xs is left-leaning (左邊subtree較深）

注意右邊subtree最多高度d-1 (否則不會通過invariant 2)。則我們就把右邊的subtree與ys concat recursively，最後在形成一個parent node連接兩者：

2. xs is right-leaning，右邊的subtree比較長：

注意下圖把xs的右邊subtree的左右children subtree的可能深度配對都列出來了，分別是：
(d-2, d-2)
(d-3, d-2)
(d-2, d-3)

可能的combination結果如下圖：（沒有完全列出來，看code吧）：

重點是 這個combination有達到我們要的O(log(n) + log(m) )嗎？！

注意此演算法的base case是某個兩者高度差 < 某值，所以recursion深度取決於O(xs_height - ys_height) < O(logn)，可以說Conc tree是符合parallel combiners需求的data structure。

Conc tree to implement combiner!

Conc tree concatenation兩個tree可以有O(h1-h2)，但是我們還需要constant time append operation，才能實現"efficient" parallel combiner，所謂efficient就是 O(logm + logn)，n = combiner1 #elements, m = combiner2 #elements。

combiner trait的append就是 += method，假設我們用conc tree來當作combiner的instance variable xs:

我們知道至少要O(log n) time來做append，但是有沒有辦法constant time?

事實上是有ㄟ！
不過要loose Conc tree invariance 2: 仿造 <> node，定義一個Append node:

但是不限定append node的左右children的高度差！所以以下的subtree 是合法的：

所以append很快，可以是O(1)，因為只要把append node的children pointers指向兩者就好：

所以Append Node事實上也是intermediate data structure，因為不能parallel execution（因為tree unbalanced)，所以勢必也要轉換成balanced tree。

不過這個轉換本身至少需要O(n)，所以以上的implementation不適合達成"efficient" concatenation (至少O(logn))！因為在concatenation過程中，變成linked list，而linked list只能linear time traversal。

所以要達成O(log n) concatenation，另一個方法是 限制append node的數量，使得整體的concatenation effort能被O(log n) bounded。

這邊用到二進位的想法：二進位的數字n，但是所需要的bit數目卻是log(n)，所以數字線性成長但是bit數目log成長。

以下過程省略，課程字幕實在太爛了，老師口音又常常聽不清楚，放棄！

怎麼用conc tree implement Combiner

這個combiner叫做ConcBuffer，定義如下：

他有一個conc tree memeber，也有一個array，以及chunkSize來記錄第一個empty array entry index。這個index就用來append element，直到array滿了以後，就expand一個新的array，並且把此滿了的array  包成Chunk node放入 conc tree。

Chunk node定義如下，功能類似Conc tree中的SingleNode:

以下又省略了，實在聽得很痛苦，直接寫作業吧！
這門課的口音和字幕帶給我很大痛苦！

Scala Parallel Programming筆記 11 - Splitter and Combiners

Iterator 

首現我們知道Java有iterator:

Splitter (iterator parallel counterpart)

每一個parallel collection都依定會implement Splitter trait。
一但呼叫了split，就分成幾個disjoint splitter set:

原來的splitter instance會變成undefined state。
remaining是"estimate"目前splitter中有幾個element。

來看怎麼用splitter trait implement fold?

首先迴響 for .. yield會產生一個collection，所以children是一個collection for task[Splitter]，每個task會recursively呼叫fold，由於每次children都是remaining > threshold時 split出child Splitters而來，所以remaining會越來越少，直到抵達base case去做sequential foldLeft。最後一行還是要記得去呼叫foldLeft在這個children collection上，才能真正去merge。

Builder

result被呼叫的話，就會return collection，而此builder就undefined。

所以builder其實是把一個collection的“增加member"這件事抽象化的trait，

Combiner (parallel Builder)

combiner要efficient需要經過一番努力，所謂的efficient是要在O(logn + logm)時間內完成，當然落，要不然不等於linearly跑過一次，那哪叫什麼efficient combiner。簡單以四核心cpu為例：

一個reduction tree的leaves會分配給4 core cpus，假設剛好分成四份，每份1/4 N，但是在往root combine的過程中，其實總共累積了（粗略計算） 7/4 N > N的工作量，這比單一cpu sequentially map的時間還久！！！

combine對Map/Set來說，是union。
對sequence來說 (List/Vector/Array) combine是concatenation。

可惜的是，對常見的data  structure來說，combine的動作不可能達到efficient (O(logm + logn)。

兩階段parallel construction

但是實際上幾乎所有的scala collection都能轉換成parallel collection，為何？
因為採取了兩階段的collection construction，使用暫時的data structure來儲存state，例如Array 可能採用其他的intermediate data structure來表現，使得efficient combiner可以實現。

這個intermediate data structure有以下的性質：

第三點就是builder/combiner為什麼會有result method的原因，因為要捨棄這個intermediate data structure，convert到真正的data structure，需要再O(n/P) time內，n = size of the data structure, P = # processors。這個convert必須也要能parallellizable。

phase 1：每個processor先呼叫 += 來build intermediate data structures，然後每個ids被combine直到reduction tree root。

phase 2: parallelly build final data structure from ids。

總共約N/2 for 4 processors。

Array Combiners

intermediate data structure主要就是nested array:

先來看怎麼implement += :

這是O(1)。如果裏層array滿了，就new一個新的更大的。注意上圖中其他的第一層array element都是空的。

combine就很簡單了，因為第一層array是一堆Array pointer，所以只要指向另一個被combine的nested array就可：

這邊 ++= 倒是第一次看到的operator! 這是constant time operation 。

再來是conversion result:

後來的筆記都消失了....因為chrome給我當機 @@
火大ㄟ