使用並行計算大幅提高遞歸算法效率

前言：
不管什麼樣的並行計算方式，其終極目的都是爲了有效利用多機多核的計算能力，並能靈活知足各類需求。相對於傳統基於單機編寫的運行程序，若是使用該方式改寫爲多機並行程序，可以充分利用多機多核cpu的資源，使得運行效率獲得大幅度提高，那麼這是一個好的靠譜的並行計算方式，反之，又難使用又難直接看出並行計算優點，還要耗費大量學習成本，那就不是一個好的方式。

因爲並行計算在互聯網應用的業務場景都比較複雜，如海量數據商品搜索、廣告點擊算法、用戶行爲挖掘，關聯推薦模型等等，若是以真實場景舉例，初學者很容易被業務自己的複雜度繞暈了頭。所以，咱們須要一個通俗易懂的例子來直接看到並行計算的優點。

數字排列組合是個經典的算法問題，它很通俗易懂，適合不懂業務的人學習，咱們經過它來發現和運用並行計算的優點，能夠獲得一個很直觀的體會，並留下深入的印象。問題以下：

請寫一個程序，輸入M，而後打印出M個數字的全部排列組合（每一個數字爲1，2，3，4中的一個）。好比：M=3，輸出：
1，1，1
1，1，2
……
4，4，4
共64個
注意：這裏是使用計算機遍歷出全部排列組合，而不是求總數，若是隻求總數，能夠直接利用數學公式進行計算了。

1、單機解決方案：
一般，咱們在一臺電腦上寫這樣的排列組合算法，通常用遞歸或者迭代來作，咱們先分別看看這兩種方案。
1) 單機遞歸
能夠將n（1<=n<=4）看作深度，輸入的m看作廣度，獲得如下遞歸函數（完整代碼見附件CombTest.java） 

public void comb(String str){
  for(int i=1;i<n+1;i++){
    if(str.length()==m-1){
	System.out.println(str+i);
	total++;
    }else
	comb(str+i);
  }
}

可是當m數字很大時，會超出單臺機器的計算侷限致使緩慢，太大數字的排列組合在一臺計算機上幾乎很難運行出，不光是排列組合問題，其餘相似遍歷求解的遞歸或回溯等算法也都存在這個問題，如何突破單機計算性能的問題一直困擾着咱們。

2) 單機迭代
咱們觀察到，求的m個數字的排列組合，實際上均可以在m-1的結果基礎上獲得。
好比m=1，獲得排列爲1,2,3,4，記錄該結果爲r(1)
m=2, 能夠由(1,2,3,4)* r(1) = 11,12,13,14,21,22,…,43,44獲得, 記錄該結果爲r(2)
由此，r(m) =(1,2,3,4)*r(m-1)
若是咱們從1開始計算，每輪結果保存到一箇中間變量中，反覆迭代這個中間變量，直到算出m的結果爲止，這樣看上去也可行，彷彿還更簡單。
可是若是咱們估計一下這個中間變量的大小，估計會嚇一跳，由於當m=14的時候，結果已經上億了，一億個數字，每一個數字有14位長，而且爲了獲得m=15的結果，咱們須要將m=14的結果存儲在內存變量中用於迭代計算，不管以什麼格式存，幾乎都會遭遇到單臺機器的內存侷限，若是排列組合數字繼續增大下去，結果便會內存溢出了。

2、分佈式並行計算解決方案：
咱們看看如何利用多臺計算機來解決該問題，一樣以遞歸和迭代的方式進行分析。

1) 多機遞歸
作分佈式並行計算的核心是須要改變傳統的編程設計觀念，將算法從新設計按多機進行拆分和合並，有效利用多機並行計算優點去完成結果。
咱們觀察到，將一個n深度m廣度的遞歸結果記錄爲 r(n,m)，那麼它能夠由(1,2,…n)*r(n,m-1)獲得：
r(n,m)=1*r(n,m-1)+2*r(n,m-1)+…+n*r(n,m-1)
假設咱們有n臺計算機，每臺計算機的編號依次爲1到n，那麼每臺計算機實際上只要計算r(n,m-1)的結果就夠了，這裏實際上將遞歸降了一級， 而且讓多機並行計算。
若是咱們有更多的計算機，假設有n*n臺計算機，那麼：
r(n,m)=11*r(n,m-2)+12*r(n,m-2)+…+nn*r(n,m-2)
拆分到n*n臺計算機上就將遞歸降了兩級了
能夠推斷，只要咱們的機器足夠多，可以線性擴充下去，咱們的遞歸複雜度會逐漸降級，而且並行計算的能力會逐漸加強。

這裏是進行拆分設計的分析是假設每臺計算機只跑1個實例，實際上每臺計算機能夠跑多個實例（如上圖），咱們下面的例子能夠看到，這種並行計算的方式相對傳統單機遞歸有大幅度的效率提高。

這裏使用fourinone框架設計分佈式並行計算，第一次使用能夠參考分佈式計算上手demo指南, 開發包下載地址：http://www.skycn.com/soft/68321.html

ParkServerDemo：負責工人註冊和分佈式協調
CombCtor：是一個包工頭實現，它負責接收用戶輸入的m，並將m保存到變量comb，和線上工人總數wknum一塊兒傳給各個工人，下達計算命令，並在計算完成後累加每一個工人的結果數量獲得一個結果總數。
CombWorker：是一個工人實現，它接收到工頭髮的comb和wknum參數用於遞歸條件，而且經過獲取本身在集羣的位置index，作爲遞歸初始條件用於降級，它找到一個排列組合會直接在本機輸出，可是計數保存到total，而後將本機的total發給包工頭統計整體數量。

運行步驟：
爲了方便演示,咱們在一臺計算機上運行:
一、啓動ParkServerDemo：它的IP端口已經在配置文件的PARK部分的SERVERS指定。
二、啓動4個CombWorker實例：傳入2個參數，依次是ip或者域名、端口（若是在同一臺機器能夠ip相同，可是端口不一樣），這裏啓動4個工人是因爲1<=n<=4，每一個工人實例恰好能夠經過集羣位置 index進行任務拆分。
三、運行CombCtor查看計算時間和結果

下面是在一臺普通4cpu雙核2.4Ghz內存4g開發機上和單機遞歸CombTest的測試對比

經過測試結果咱們能夠看到：
一、能夠推斷，因爲單機的性能限制，沒法完成m值很大的計算。
二、同是單機環境下，並行計算相對於傳統遞歸提高了將近1.6倍的效率，隨着m的值越大，節省的時間越多。
三、單機遞歸的CPU利用率不高，平均20-30%，在多核時代沒有充分利用機器資源，形成cpu閒置浪費，而並行計算則能打滿cpu，充分利用機器資源。
四、若是是多機分佈式並行計算，在4臺機器上，採用4*4的16個實例完成計算，效率還會成倍提高，並且機器數量越多，計算越快。
五、單機遞歸實現和運行簡單，使用c或者java寫個main函數完成便可，而分佈式並行程序，則須要利用並行框架，以包工頭+多個工人的全新並行計算思想去完成。

2) 多機迭代
咱們最後看看如何構思多機分佈式迭代方式實現。
思路一：
根據單機迭代的特色，咱們能夠將n臺計算機編號爲1到n
第一輪統計各工人發送編號給工頭，工頭合併獲得第一輪結果{1,2,3,…,n}
第二輪，工頭將第一輪結果發給各工人作爲計算輸入條件，各工人根據本身編號累加，返回結果給工頭合併，獲得第二輪結果：{11,12,13,1n,…,n1,n2,n3,nn}




這樣迭代下去，直到m輪結束，如上圖所示。
但很快就會發現，工頭合併每輪結果是個很大的瓶頸，很容易內存不夠致使計算崩潰。

思路二：
若是對思路一改進，各工人不發中間結果給工頭合併，而採起工人之間互相合並方式，將中間結果按編號分類，經過receive方式（工人互相合並及receive使用可參見sayhello demo），將屬於其餘工人編號的數據發給對方。這樣必定程度避免了工頭成爲瓶頸，可是通過實踐發現，隨着迭代變大，中間結果數據愈來愈大，工人合併耗用網絡也愈來愈大，若是中間結果保存在各工人內存中，隨着m變的更大，仍然存在內存溢出危險。

思路三：
繼續改進思路二，將中間結果變量不保存內存中，而每次寫入文件（詳見Fourinone2.0對分佈式文件的簡化操做），這樣能避免內存問題，可是增長了大量的文件io消耗。雖然能運行出結果，可是並不高效。

總結：
或許分佈式迭代在這裏並非最好的作法，上面的多機遞歸更合適。因爲迭代計算的特色，須要將中間結果進行保存，作爲下一輪計算的條件，若是爲了利用多機並行計算優點，又須要反覆合併產生中間結果，因此致使對內存、帶寬、文件io的耗用很大，處理不當容易形成性能低下。
咱們早已經進入多cpu多核時代，可是咱們的傳統程序設計和算法還停留在過去單機應用，所以合理利用並行計算的優點來改進傳統軟件設計思想，能爲咱們帶來更大效率的提高。

如下是分佈式並行遞歸的demo源碼：

// CombTest
import java.util.Date;
public class CombTest
{
	int m=0,n=0,total=0;
	CombTest(int n, int m){
		this.m=m;
		this.n=n;
	}
	public void comb(String str)
	{
		for(int i=1;i<n+1;i++){
			if(str.length()==m-1){
				//System.out.println(str+i);//打印出組合序列
				total++;
			}
			else
				comb(str+i);
		}
	}
	
	public static void main(String[] args)
	{
		CombTest ct = new CombTest(Integer.parseInt(args[0]), Integer.parseInt(args[1]));
		long begin = (new Date()).getTime();
		ct.comb("");
		System.out.println("total:"+ct.total);
		long end = (new Date()).getTime();
		System.out.println("time:"+(end-begin)/1000+"s");
	}
}

// ParkServerDemo
import com.fourinone.BeanContext;
public class ParkServerDemo{
	public static void main(String[] args){
		BeanContext.startPark();
	}
}

// CombCtor
import com.fourinone.Contractor;
import com.fourinone.WareHouse;
import com.fourinone.WorkerLocal;
import java.util.Date;
public class CombCtor extends Contractor
{
	public WareHouse giveTask(WareHouse wh)
	{
		WorkerLocal[] wks = getWaitingWorkers("CombWorker");
		System.out.println("wks.length:"+wks.length+";"+wh);
		wh.setObj("wknum",wks.length);
		WareHouse[] hmarr = doTaskBatch(wks, wh);//批量執行任務,全部工人完成才返回
		int total=0;
		for(WareHouse hm:hmarr)
			total+=(Integer)hm.getObj("total");
		System.out.println("total:"+total);
		return wh;
	}
	
	public static void main(String[] args)
	{
		CombCtor a = new CombCtor();
		WareHouse wh = new WareHouse("comb", Integer.parseInt(args[0]));
		long begin = (new Date()).getTime();
		a.doProject(wh);
		long end = (new Date()).getTime();
		System.out.println("time:"+(end-begin)/1000+"s");
		a.exit();
	}
}

//CombWorker
import com.fourinone.MigrantWorker;
import com.fourinone.WareHouse;
public class CombWorker extends MigrantWorker
{
	private int m=0,n=0,total=0,index=-1;

	public WareHouse doTask(WareHouse wh)
	{
		total=0;
		n = (Integer)wh.getObj("wknum");
		m = (Integer)wh.getObj("comb");
		index = getSelfIndex()+1;
		System.out.println("index:"+index);
		comb(index+"");
		System.out.println("total:"+total);
		return new WareHouse("total",total);
	}
	
	public void comb(String str)
	{
		for(int i=1;i<n+1;i++){
			if(str.length()==m-1){
				//System.out.println(str+i);//打印出組合序列
				total++;
			}
			else
				comb(str+i);
		}
	}
	
	public static void main(String[] args)
	{
		CombWorker mw = new CombWorker();
		mw.waitWorking(args[0],Integer.parseInt(args[1]),"CombWorker");
	}
}