2010 英特尔® 线程挑战赛-出租车路径

2010 英特尔® 线程挑战赛-出租车路径

邓辉 denghui0815@hotmail.com
点击下载源码 问题描述

在笛卡尔网格中，曼哈顿距离是指从一个整数坐标点到另一个整数坐标点间必须经过的最小整数坐标网格点数，这样，两个坐标中只有一个的相邻网格点坐标可以相差 1。如果您在曼哈顿乘坐出租车旅行，那么可以将十字路口看作是网格上的点，街道看作是连接这些点的线。不能驱车穿过建筑物来缩短行驶距离；因此，两个十字路口之间的距离就是经过的街区数。但是，如果出租车是对称变维结构而且可以穿过建筑物，将会怎么样？

问题描述： 编写一个多线程程序来生成笛卡尔网格，对网格上原点 (0,0) 与某个给定点之间的路径进行统计和分类。路径由从一个整数网格点到另一个整数网格点的移动组成。网格点间有三种可能的移动方式：一个点向北移（与 x 轴平行），一个点向东移（与 y 轴平行）或向东北方向移（沿对角线方向朝目标移动）。路径的目的地坐标由命令行上的两个整数确定，另外此命令行上还会提供一个文件名，用于保存从原点 (0,0) 到目标点的路径。

路径将根据路径中使用的每种移动的次数进行分类，而不考虑移动的顺序。例如，如果目标点是 (5,4)，那么“2-3-1”路径将包括两次向北移动，三次向东北方向移动，和一次向东移动。此类别的两个示例是“NNneneneE”和“EneNneneN”。

输入描述：此程序要求输入两个非负整数和一个文件名，所有这些都从应用程序的命令行中输入。两个整数表示目标点的 x 坐标和 y坐标。给定的文件将用于保存从原点 (0,0) 到目标点时生成的路径。

输出描述：该应用程序将生成两个输出。第一个输出是每一种路径类别中的路径数以及所有类别的路径总数。程序报告的数字将打印到标准输出。

第二个预期输出是列出所有生成的路径，这些路径包含在一个给定的文本文件中，每条路径占一行。为使输出行的长度等于原位置与目标位置间的曼哈顿距离，向北/向东移动时打印一个字符（“N”）/（“E”），向东北方向移动时打印两个字符（“ne”）。人们阅读文件时，可以通过大小写来区分移动情况，例如先向北移动一次，再向东移动一次为“NE”，向东北方向移动一次为“ne”。这样不仅不必在路径的各次移动之间加空格，还可以减小文件大小。

输入命令行示例： taxi.exe 2 2 paths2x2.txt

输出（屏幕）示例：

Number of paths to ( 2, 2)

2-0-2 paths: 6

1-1-1 paths: 6

0-2-0 paths: 1

Total 13

输出（文件）示例，paths2x2.txt:

EENN

EneN

ENEN

ENne

ENNE

NEne

NENE

NneE

NNEE

nene

NEEN

neNE

neEN

计时：路径的生成和统计时间将作为此应用程序的唯一评分标准。将序列写入输出文件的时间不需要计算在内。为得到最准确的计时结果，所提交的代码需要包含计时代码并仅将算出的序列生成时间打印到标准输出，否则将使用外部秒表计算整个执行时间。
串行算法

到目标点的移动的分类数量由目标坐标(nDstX, nDstY)决定。

令nMax = max（nDstX, nDstY）

存在共计nMax + 1个分类：

nDstX-0-nDstY

(nDstX - 1)-1-(nDstY - 1)

(nDstX - 2)-2-(nDstY - 2)

… … …

… … …

… … ..

(nDstX - nMax)-nMax-(nDstY - nMax)

每一个分类的路径数量为其可重全排列的数量，所以分类E-ne-N的路径数量为

Path(E, ne, N) = (E + ne + N)! / E! / ne! / N!，由于要求生成路径并分类统计，所以这个公式只能用来验证程序的正确性。

假设当前点坐标（x， y），目标点坐标为（nDstX， nDstY），当x != nDstX 且 y != nDstY时可向三个方向移动，否则只能向一个方向移动，而路径所属分类由斜向移动次数决定。所以很容易可以编写递归的求解函数。

首先定义出租车和路径结构：

// 路径

typedef struct tagXPath

{

int nCnt; // 路径数量

uint8* pPath; // 路径内容

tagXPath* pNext; // 下一路径

}XPath;

// 出租车

typedef struct tagXTaxi

{

int nDstX,nDstY; // 目标点坐标

int nCutoff; // 串行求解条件

int nStep; // 路径最大长度

int nClass; // 路径分类数量

uint32* pCount; // 路径分类计数

uint8* pPath; // 移动路径

}XTaxi;

仅仅统计分类数量的函数：

// 串行查找路径仅统计路径数量

void XTaxiPathCount_Serial(XTaxi* pTaxi, int x, int y, int nClass)

{

if(x == pTaxi->nDstX || y == pTaxi->nDstY)

{ // 更新路径数量

++pTaxi->pCount[nClass];

}

else

{ // 向东移动

XTaxiPathCount_Serial(pTaxi,x + 1, y, nClass);

// 向北移动

XTaxiPathCount_Serial(pTaxi, x, y + 1, nClass);

// 向东北移动

XTaxiPathCount_Serial(pTaxi, x + 1, y + 1, nClass + 1);

}

}

Debug版本统计移动到目标点（11，11）的路径数量花费的时间为1.121672秒。由于路径的数量非常多，在保存路径时为了降低内存消耗，必须对路径进行压缩。每次移动仅有3种选择，所以用2个bit位来表示一次移动。路径使用链表存放，每次分配的长度可保存XMAX_PATH_COUNT个路径。

#define XMA
159f8
X_PATH_COUNT 256

// 掩码数组

uint8 g_xMask1[4] = {0x40, 0x10, 0x04, 0x01};

uint8 g_xMask2[4] = {0x80, 0x20, 0x08, 0x02};

uint8 g_xMask3[4] = {0xC0, 0x30, 0x0C, 0x03};

// 设置bit位

#define XSETBIT_1(pBit, nPos) (pBit[(nPos) >> 2] |= g_xMask1[(nPos) & 0x03])

#define XSETBIT_2(pBit, nPos) (pBit[(nPos) >> 2] |= g_xMask2[(nPos) & 0x03])

#define XSETBIT_3(pBit, nPos) (pBit[(nPos) >> 2] |= g_xMask3[(nPos) & 0x03])

// 串行查找路径

void XTaxiFindPath_Serial(XTaxi* pTaxi, int x, int y, int nLen, XPath* pPath)

{

if(x == pTaxi->nDstX || y == pTaxi->nDstY)

{ // 加入路径

XPathPush(pTaxi, x, y, nLen, pPath);

}

else

{ // 备份路径 向东移动

uint8 cBack = pTaxi->pPath[nLen >> 2];

XSETBIT_1(pTaxi->pPath, nLen);

XTaxiFindPath_Serial(pTaxi,x + 1, y, nLen + 1, pPath);

// 恢复备份 向北移动

pTaxi->pPath[nLen >> 2] = cBack;

XSETBIT_2(pTaxi->pPath, nLen);

XTaxiFindPath_Serial(pTaxi, x, y + 1, nLen + 1, pPath);

// 恢复备份 向东北移动

pTaxi->pPath[nLen >> 2] = cBack;

XSETBIT_3(pTaxi->pPath, nLen);

XTaxiFindPath_Serial(pTaxi, x + 1, y + 1, nLen + 1, pPath);

// 恢复备份

pTaxi->pPath[nLen >> 2] = cBack;

}

}

在XPathPush函数中进行路径的保存和分类计数，参数（x, y, nLen）表示的路径进行了x + y – nLen次向东北方向的移动，所以只需++pTaxi->pCount[x + y - nLen]即可完成分类计数。

// 添加路径到链表

__inline void XPathPush(XTaxi* pTaxi, int x, int y, int nLen, XPath* pPath)

{

// 路径分类计数

++pTaxi->pCount[x + y - nLen];

// 复制路径内容

if(pPath->nCnt == XMAX_PATH_COUNT) XPathExt(pPath, pTaxi->nStep);

uint8* pBit = pPath->pPath + pPath->nCnt * pTaxi->nStep;

int nPos, nEnd = (nLen + 3) >> 2;

for(nPos = 0; nPos < nEnd; ++nPos) pBit[nPos] = pTaxi->pPath[nPos];

if(x != pTaxi->nDstX)

{ // 水平方向移动

nEnd = pTaxi->nDstX - x + nLen;

nPos = nLen - 1;

while(++nPos < nEnd) XSETBIT_1(pBit, nPos);

}

else if(y != pTaxi->nDstY)

{ // 垂直方向移动

nEnd = pTaxi->nDstY - y + nLen;

nPos = nLen - 1;

while(++nPos < nEnd) XSETBIT_2(pBit, nPos);

}

++pPath->nCnt;

}

使用Amplifier检测Hotspots结果如下：

进入XPushPath内部：

 


复制路径部分占用了334ms，将其优化为4字节复制可以提高效率。

// 添加路径到链表

__inline void XPathPush(XTaxi* pTaxi, int x, int y, int nLen, XPath* pPath)

{

// 路径分类计数

++pTaxi->pCount[x + y - nLen];

// 复制路径内容

if(pPath->nCnt == XMAX_PATH_COUNT) XPathExt(pPath, pTaxi->nStep);

uint8* pBit = pPath->pPath + pPath->nCnt * pTaxi->nStep;

int nPos = 0, nEnd = (nLen + 3) >> 2,nEnd4 = nEnd >> 2 << 2;

for(; nPos < nEnd4; nPos += 4) *(uint32*)(pBit + nPos) = *(uint32*)(pTaxi->pPath + nPos);

for(; nPos < nEnd; ++nPos) pBit[nPos] = pTaxi->pPath[nPos];

// 未到达目标点

if(x != pTaxi->nDstX)

{ // 水平方向移动

nEnd = pTaxi->nDstX - x + nLen;

nPos = nLen - 1;

while(++nPos < nEnd) XSETBIT_1(pBit, nPos);

}

else if(y != pTaxi->nDstY)

{ // 垂直方向移动

nEnd = pTaxi->nDstY - y + nLen;

nPos = nLen - 1;

while(++nPos < nEnd) XSETBIT_2(pBit, nPos);

}

++pPath->nCnt;

}

使用Amplifier检测Hotspots结果如下：

 


XPushPath的执行时间由813ms下降到687ms。
并行算法

Cilk提供的cilk_spawn，可以用于递归算法的并行优化。由于对pTaxi和pPath的访问存在数据竞争问题，通常采用锁机制来解决数据竞争，但常常会导致程序性能下降。更好的解决办法是设计非阻塞算法：1，复制pTaxi；2, 分别产生pPath；3, 合并pPath。

// 并行查找路径

void XTaxiFindPath_Parallel(XTaxi* pTaxi, int x, int y, int nLen, XPath* pPath)

{

if(x == pTaxi->nDstX || y == pTaxi->nDstY)

{ // 加入路径

XPathPush(pTaxi, x, y, nLen, pPath);

}

else

{

uint8 cBack = pTaxi->pPath[nLen >> 2];

// 创建临时Taxi和Path

XTaxi* pTaxi1 = XTaxiCopy(pTaxi);

XTaxi* pTaxi2 = XTaxiCopy(pTaxi);

XPath* pPath1 = XPathCreate(pTaxi->nStep, XMAX_PATH_COUNT);

XPath* pPath2 = XPathCreate(pTaxi->nStep, XMAX_PATH_COUNT);

// 设置路径标记

XSETBIT_1(pTaxi->pPath, nLen);

XSETBIT_2(pTaxi1->pPath, nLen);

XSETBIT_3(pTaxi2->pPath, nLen);

XTaxiFindPath_Parallel(pTaxi, x + 1, y, nLen + 1, pPath);

cilk_spawn XTaxiFindPath_Parallel(pTaxi1, x, y + 1, nLen + 1, pPath1);

cilk_spawn XTaxiFindPath_Parallel(pTaxi2, x + 1, y + 1, nLen + 1, pPath2);

cilk_sync;

// 将路径数量累加到pTaxi->pCount中

for(int i = 0; i < pTaxi->nClass; ++i)

{

pTaxi->pCount[i] += pTaxi1->pCount[i] + pTaxi2->pCount[i];

}

XTaxiFree(pTaxi1);

XTaxiFree(pTaxi2);

XPathUnite(pPath1, pPath2);

XPathUnite(pPath, pPath1);

pTaxi->pPath[nLen >> 2] = cBack;

}

}

为降低内存浪费将XMAX_PATH_COUNT定义为2，使用Debug版本统计移动到目标点（9，9）的路径数量花费的时间为1.298604秒，效率严重下降。

使用Amplifier检测Concurrency结果如下：


 

检测结果显示由于反复的拷贝、释放（Taxi和Path）程序效率严重下降。当路径长度等于某个阈值时调用串行算法，可以加大并行的粒度，减少拷贝、释放、cilk线程开销。

为减少内存分配次数将XMAX_PATH_COUNT定义为256。

测试Debug版本统计移动到目标点（10，10）的路径数量。串行版本为1.293096秒，并行版本为0.678521秒，加速比为1.90。

使用Amplifier检测Concurrency结果如下：

 



性能测试

Release[b]版本双核笔记本测试结果（测试时关闭了文件输出）：

[/b]

目标点坐标 算法版本	（9， 9）	（10， 10）	（9， 12）	（11， 11）
串行版本 XTaxiPathCount_Serial	0.081140秒	0.419633秒	0.796909秒	2.306538秒
并行版本 XTaxiPathCount_Parallel	0.041478秒	0.210360秒	0.385175秒	1.169847秒