使用并行计算大幅提高递归算法效率

前言：
不管什么样的并行计算方式，其终极目的都是为了有效利用多机多核的计算能力，并能灵活知足各类需求。相对于传统基于单机编写的运行程序，若是使用该方式改写为多机并行程序，可以充分利用多机多核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");
	}
}