讓R語言走向多線程并行計算課件

1、SupR：讓 R 語言走向多線程并行計算1概要2R 并行計算概覽并行模式：進程 vs 線程SupR：當 R 遇上大數(shù)據(jù)概要3R 并行計算概覽并行模式：進程 vs 線程SupR：當 R 遇上大數(shù)據(jù)R 中可實現(xiàn)的并行方式4C/C+/FORTRAN 并行多進程并行Hadoop/Spark 并行GPU 并行C/C+/FORTRAN 并行5由 C/C+/FORTRAN 代碼進行并行，R 直接調用優(yōu)勢：使用簡單（作為用戶），性能良好缺點：編程困難（作為開發(fā)者），涉及 R 對象時并非線程安全例子：OpenBLAS（Pthread/OpenMP）xgboost 包（OpenMP）recosystem 包（C+

2、 11 線程庫）RcppParallel 包（Intel TBB）多進程并行6創(chuàng)建多個 R 進程同時進行計算，使用 Socket 或 MPI 等軟件 庫進行進程通信優(yōu)勢：R 中使用范圍最廣的方式，可直接編寫 R 代碼并行缺點：通信成本較高，內存使用量大例子：parallel 包，mclapply() 和 parLapply() 函數(shù)snow，Rmpi 和 pbdMPI 等使用 MPI 接口的軟件包Hadoop/Spark 并行從 R 調用 Hadoop 和 Spark 等并行計算平臺優(yōu)勢：基于成熟的并行計算平臺，能夠處理大 規(guī)模的數(shù)據(jù)缺點：與 R 之間通信成本非常高，內存占用大例子：RHIPE

3、 包（Hadoop）SparkR 包（Spark）7GPU 并行使用 GPU 進行并行計算優(yōu)勢：處理器數(shù)目多，并行效果顯著缺點：硬件要求高，實現(xiàn)的算法較少例子：gputools 包（CUDA）gpuR 包（OpenCL）8概要9R 并行計算概覽并行模式：進程 vs 線程SupR：當 R 遇上大數(shù)據(jù)兩種并行模式多進程并行多線程并行10進程 vs 線程一個進程可包含多個線程線程比進程更為輕量級同一進程中的線程可共享內存資源線程間的通信更為便捷，開銷較小11R 中的并行模式12R 中大部分已有的并行方式都是基于多進程并行多線程并行一般依賴于 C/C+/FORTRAN/Java相較于多進程并行，多線程

4、并行具有諸多優(yōu)勢然而，在 R 中進行多線程并行非常困難解釋器、內存分配和垃圾回收都不是線程安全的多個線程同時對 R 對象進行操作時，可能會破壞 R 的運行機 制和內部結構要實現(xiàn)多線程的 R，就必須更改 R 的底層源代碼概要13R 并行計算概覽并行模式：進程 vs 線程SupR：當 R 遇上大數(shù)據(jù)當 R 遇上大數(shù)據(jù)14基于種種原因，R 對于真正的大數(shù)據(jù)支持尚不理想其受限的并行機制是其中重要的一點理想的大數(shù)據(jù)分析平臺單機上進行多線程并行集群上進行多機分布式計算良好的交互性完善的社區(qū)支持SupR 誕生15SupR 是一款同時支持多線程和分布式計算的修改版 R由普渡大學統(tǒng)計系劉傳海教授開發(fā)力圖實現(xiàn)基于

5、 R 的大數(shù)據(jù)分析平臺主要特性保持 R 的語法和內部數(shù)據(jù)結構不變提供多線程并行計算支持提供類似 Spark 的分布式集群運算分布式文件系統(tǒng)支持SupR 多線程并行計算16SupR 在源代碼級別對 R 進行了修改在可能引起線程沖突的部分加入互斥鎖，保證程序正常運行提供完整的線程創(chuàng)建、查詢、打斷、同步和取消機制主要函數(shù)17new.thread()：創(chuàng)建線程start.thread()：開啟線程sync.eval()：同步線程執(zhí)行wait()：令線程睡眠，等待信號notify()：喚醒線程例子簡單并行18將需要并行處理的兩個表達式放到兩個線程中執(zhí)行set.seed(123) n = 10A = ma

6、trix(rnorm(n2), n) B = matrix(rnorm(n2), n) th1 = new.thread(C1 - A %*% B, 1:(n/2)th2 = new.thread(C2 - A %*% B, (n/2+1):n)start.thread(th1); start.thread(th2) ls()# 1 ABC1 C2 nth1 th2例子線程同步所有具有相同 x 對象的 sync.eval() 語句，任何時間只 能有一個被執(zhí)行x = any objectthreads = vector(list, 5) for(i in 1:5) threadsi = new.

7、thread(sync.eval(x, cat(current.thread(), : , as.character(Sys.time(),n, sep = ) thread.sleep(2)for(i in 1:5) start.thread(threadsi) # thread_1: 2016-05-25 09:21:24# thread_3: 2016-05-25 09:21:27# thread_5: 2016-05-25 09:21:29# thread_4: 2016-05-25 09:21:31# thread_2: 2016-05-25 09:21:3319例子線程通信（1）w

8、ait() 使線程進入睡眠，直到被 notify() 喚醒主線程創(chuàng)建主線程打印歡迎信息喚醒工人，等待工人工作結束打印結束信息sync.m = master barrier sync.w = worker barrier threads = vector(list, 3) count = 0master = new.thread(sync.eval(sync.m, cat(Thread M: Program startsn) # Notify workers starting to worksync.eval(sync.w, notify(sync.w, all = TRUE) # Wait f

9、or workers to finishwait(sync.m)cat(Thread M: Program endsn)20例子線程通信（2）工人線程創(chuàng)建工人線程，等待主線程“宣布開工”把 count 加 1 并打印信息，任一時刻只有一個線程在工作工作完成后（count 到 10）通知主線程，線程結束work = function() sync.eval(sync.w, wait(sync.w) while(TRUE) sync.eval(sync.w, if(count 10) count - count + 1cat(current.thread(), is working, count

10、=, count, n) else break)sync.eval(sync.m, notify(sync.m)for(i in 1:3) new.thread(work(), start = TRUE)21例子線程通信（3）輸出結果start.thread(master)# Thread M: Program starts# thread_3 is working, count = 1 # thread_4 is working, count = 2 # thread_2 is working, count = 3 # thread_3 is working, count = 4 # thread_4 is working, count = 5 # thread_2 is working, count = 6 # thread_3 is working, count = 7 # thread_