xsortpub.cpp

归并排序的并行算法 intel 多线程优化大赛

/*/////////////////////////////////////////
// 版权所有(C)  2000-2008 邓辉           //
// Email：      [e-mail]  //
// 说明：       Intel线程优化大赛参赛作品//
/////////////////////////////////////////*/

#include "XSortPub.h"
#include "XFileMap.h"

#define XIOTHREAD	 16
#define XONENUMSIZE	 11

void Error (char *s)
{
	fprintf(stderr, "ERROR: %s\n",s);
	exit(1);
}

void Warning (char *s)
{
	fprintf(stderr, "WARNING: %s\n",s);
}

// 修改数组大小
void XSortArrayResize(XSortArray* pArray, uint32 nNewSize)
{
	if(pArray->nSize < nNewSize)
	{
		pArray->pData = (uint32*)scalable_realloc(pArray->pData, sizeof(pArray->pData[0]) * nNewSize);

		if(pArray->pData == NULL)
		{
			XSortArrayDestroy(&pArray);
			Error("XSortArrayResize Error!");
		}

		pArray->nSize = nNewSize;
	}
}

// 创建数组
XSortArray* XSortArrayCreate(uint32 nSize)
{
	XSortArray* pArray = (XSortArray*)scalable_malloc(sizeof(pArray[0]));

	if(pArray != NULL)
	{
		pArray->pData = (uint32*)scalable_malloc(sizeof(pArray->pData[0]) * nSize);

		if(pArray->pData == NULL)
		{
			XSortArrayDestroy(&pArray);
		}
		else
		{
			pArray->nData = 0;
			pArray->nSize = nSize;
		}	
	}

	return pArray;
}

// 释放数组
void XSortArrayDestroy(XSortArray** ppArray)
{
	if(ppArray != NULL && *ppArray != NULL)
	{
		if((*ppArray)->pData != NULL) scalable_free((*ppArray)->pData);

		scalable_free(*ppArray);

		*ppArray = NULL;
	}
}

// 将pSrc1, pSrc2合并到pSrc1中
void XSortArrayMerge(XSortArray* pSrc1, XSortArray* pSrc2)
{
	if(pSrc1 != NULL && pSrc2 != NULL)
	{
		XSortArrayResize(pSrc1, pSrc1->nData + pSrc2->nData);

		XMEMCOPY(pSrc2->pData, pSrc1->pData+pSrc1->nData, pSrc2->nData * sizeof(pSrc1->pData[0]));

		pSrc1->nData +=  pSrc2->nData;
	}
	else
	{
		Error("XSortArrayMerge Error!");
	}
}

// 构造随机数据
XSortArray*  XSortArrayRand(uint32 nSize)
{
	XSortArray* pArray = XSortArrayCreate(nSize);

	if(pArray != NULL)
	{
		for(uint32 i = 0; i < nSize; ++i)
		{
			pArray->pData[i] = (uint32)rand() * RAND_MAX + (uint32)rand();
		}

		pArray->nData = nSize;
	}

	return pArray;
}

// 构造随机数据
void XDataFileRand(uint32 nSize, const char* szFile)
{
	if(szFile != NULL)
	{
		FILE* fp = fopen(szFile, "wb");

		if(fp != NULL)
		{
			fprintf(fp, "%u\n", nSize);

			for(uint32 i = 0; i < nSize; ++i)
			{
				fprintf(fp, "%u\n", (uint32)rand() * RAND_MAX + (uint32)rand());
			}

			fclose(fp);
		}
	}
}

// 校验排序结果
void XSortArrayVerify(XSortArray* pArray)
{
	for(uint32 i = 1; i < pArray->nData; ++i)
	{
		if(pArray->pData[i-1] > pArray->pData[i])
		{
			printf("Error Sort Ret! Index[%d]\n", i);
			break;
		}
	}
}

// 统计输出分割参数,返回线程数量,以及文件大小
int XGetOutSplitParam(uint32 nGran, XSortArray* pArray, uint32* pBeg, uint32* pMapOff, uint32& nFileSize)
{
	// 统计不同字数的数据数量
	uint32 nTotal[XONENUMSIZE] = {0};
	uint32 nCount[XONENUMSIZE] = {0};

	// 回车占用pArray->nSize个字符
	nFileSize = pArray->nSize;
	for(int i = 1; i < XONENUMSIZE; ++i)
	{
		// 使用2分查找取得小于10, 100, 1000, 10000 .......的数的个数
		nTotal[i] = i == XONENUMSIZE - 1 ? pArray->nData : XSearch<uint32>(pow(10,i), pArray->pData, pArray->nData, FALSE);
		nCount[i] = nTotal[i] - nTotal[i-1];
		nFileSize += i * nCount[i];
	}

	// 文件足够大使用多线程
	const int nThread = nFileSize > nGran ? (XIOTHREAD < nFileSize / nGran ? XIOTHREAD : nFileSize / nGran) : 1;
	// 计算平均每个线程写入的块大小
	const int nBlockSize = nFileSize / nThread;

	// 当前线程ID
	int nThreadIndex = 0;
	// 当前文件偏移
	int nFilePos  = 0;
	// 当前数的字符数
	int nNumChars = 1;
	// 当前线程要写入的数据的总的大小
	int nCurSize = 0;

	// 定位每个线程要写入文件的数组区间和相对应的文件偏移
	while(nFilePos < nFileSize)
	{
		if(nFilePos + nCount[nNumChars] * (nNumChars + 1) >  (nThreadIndex + 1) * nBlockSize)
		{
			++nThreadIndex;
			// 计算填满nBlockSize需要的数的个数
			const int nTmpCnt = (nThreadIndex * nBlockSize - nFilePos) / (nNumChars + 1);
			// 增加文件偏移
			nFilePos += nTmpCnt * (nNumChars + 1);
			// 减少未非配的数的数量
			nCount[nNumChars] -= nTmpCnt;
			// 计算出当前线程处理数据的结束位置
			pBeg[nThreadIndex] = pBeg[nThreadIndex - 1] + nCurSize + nTmpCnt;
			// 记录当前线程的文件偏移结束位置
			pMapOff[nThreadIndex] = nFilePos;
			nCurSize = 0;
		}
		else
		{
			// 增加当前线程要写入的数据的总的大小
			nCurSize += nCount[nNumChars];
			// 增加文件偏移
			nFilePos += nCount[nNumChars] * (nNumChars + 1);
			// 输出为nNumChars的数字已全部分配完毕
			nCount[nNumChars++] = 0;
		}
	}

	// 最后一个线程的数据为数组的结尾和文件尾
	pBeg[nThread] = pArray->nData;
	pMapOff[nThread] = nFileSize;

	return nThread;
}


// 加载测试数据
XSortArray* XDataFileRead(const char* szFile)
{
#ifdef XCODE_WIN
	// 获取系统信息
	SYSTEM_INFO SysInfo;
	GetSystemInfo(&SysInfo);
	const uint32 nGran = SysInfo.dwAllocationGranularity;
#else
	const uint32 nGran = 1 << 16;
#endif

	// 测试数据
	XSortArray* pArray = NULL;
	// 每个线程加载的数据
	XSortArray* ppArray[XIOTHREAD] = {0};

	uint32  nFileSize = 0;

	XFILEMAPHANDLE hFileMap = XFileMapOpen(szFile, XFILEMAP_READ, nFileSize);

	if(hFileMap == NULL) Error("XFileMapOpen Error!");


	// 读取测试数据的大小
	const char *pInput = (const char*)XFileMapView(hFileMap, XFILEMAP_READ, 0, 16);

	if(pInput == XMAPMEMINVALID) Error("XFileMapView Error!");

	uint32 nSize = 0;

	const char * pTmp = pInput;

	while(*pTmp < '0' || *pTmp > '9') ++pTmp;

	XREADNUM(pTmp, pTmp+16, nSize);

	pArray = XSortArrayCreate(nSize);

	if(pArray == NULL) Error("No Memory!");

	XFileMapUnView((void*)pInput, 16);

	// 每次加载的文件块大小
	const int nBlockSize = nFileSize < 2 * nGran ? nFileSize : 2 * nGran;

	int nThread = (nFileSize + nBlockSize - 1) / nBlockSize;

	int i,j,nTime = (nThread + XIOTHREAD - 1) / XIOTHREAD ;

	if(nThread  > XIOTHREAD) nThread = XIOTHREAD;

	for(i = 0; i < nThread; ++i) ppArray[i] = XSortArrayCreate(nBlockSize * 2 / XONENUMSIZE);

	// 分次加载数据
	for(i = 0; i < nTime; ++i)
	{
		const int nMapBeg = i * nBlockSize * nThread;
#pragma omp parallel for
		for(j = 0; j < nThread; ++j)
		{
			XSortArray* pCur = ppArray[j];

			pCur->nData = 0;

			if(nMapBeg + nBlockSize * j < nFileSize)
			{
				// 进行文件块映射
				const int nMapOff  = nMapBeg + nBlockSize * j;
				const int nMapSize = (nBlockSize + XONENUMSIZE < nFileSize - nMapOff) ? nBlockSize + XONENUMSIZE : nFileSize - nMapOff;
				const char *pBeg = (const char*)XFileMapView(hFileMap, XFILEMAP_READ, nMapOff, nMapSize);
				if(pBeg == XMAPMEMINVALID) Error("XFileMapView Error!");

				// 定位当前块的数据尾
				const char *pEnd = pBeg + (nMapSize < nBlockSize ? nMapSize : nBlockSize);
				const char *pTmp = pBeg + nMapSize;

				while(pEnd < pTmp && *pEnd != '\n') ++pEnd;

				// 定位当前块的数据头
				pTmp = pBeg;
				while(pTmp < pEnd && *pTmp != '\n') ++pTmp;
				while(pTmp < pEnd && (*pTmp < '0' || *pTmp > '9')) ++pTmp;

				// 开始读取数据
				uint32  nRead = 0;

				while(pTmp < pEnd) 
				{
					XREADNUM(pTmp, pEnd, pCur->pData[nRead])
					// 如果数据多于Array大小 扩展空间
					if(++nRead == pCur->nSize) XSortArrayResize(pCur, nRead * 5 / 4);
				}

				pCur->nData = nRead;

				// 关闭映射
				XFileMapUnView((void*)pBeg, nMapSize);
			}
		}

		// 将每个线程读取的数据合并到pArray中
		for(j = 0; j < nThread; ++j) 
		{
			XSortArrayMerge(pArray, ppArray[j]);
		}

		if(pArray->nData > nSize) break;
	}
	
	// 销毁临时空间
	for(i = 0; i < nThread; ++i) XSortArrayDestroy(&ppArray[i]);

	XFileMapClose(&hFileMap);

	if(pArray->nData > nSize) pArray->nData = nSize;
	else if(pArray->nData < nSize) Error("Read File Error: Do not have sufficient numbers!");

	return pArray;
}

// 加载测试数据
void XDataFileSave(const char* szFile, XSortArray* pArray)
{
#ifdef XCODE_WIN
	// 获取系统信息
	SYSTEM_INFO SysInfo;
	GetSystemInfo(&SysInfo);
	const uint32 nGran = SysInfo.dwAllocationGranularity;
#else
	const uint32 nGran = 1 << 16;
#endif

	// 每个线程的写入的数组下标
	uint32 pBegPos[XIOTHREAD+1] = {0};
	// 每个线程的写入的文件偏移
	uint32 pMapOff[XIOTHREAD+1] = {0};

	uint32 nFileSize = 0;

	// 获取线程输出的相关参数
	const int nThread = XGetOutSplitParam(nGran, pArray, pBegPos, pMapOff, nFileSize);

	XFILEMAPHANDLE hFileMap = XFileMapOpen(szFile, XFILEMAP_WRITE, nFileSize);

	if(hFileMap == NULL) Error("XFileMapOpen Error!");

	// 并行写文件
#pragma omp parallel for
	for(int i = 0; i < nThread; ++i)
	{
		if(pBegPos[i] < pBegPos[i + 1])
		{
			// 进行文件块映射
			const int nMapOff = pMapOff[i] / nGran * nGran;

			const int nMapSize = (pMapOff[i + 1] - pMapOff[i] + nGran < nFileSize - pMapOff[i] + pMapOff[i] % nGran) ? 
								  pMapOff[i + 1] - pMapOff[i] + nGran : nFileSize - pMapOff[i] + pMapOff[i] % nGran;

			char* pOut = (char*)XFileMapView(hFileMap, XFILEMAP_WRITE, nMapOff, nMapSize);

			if(pOut == XMAPMEMINVALID) Error("XFileMapView Error!");

			// 开始写入数据
			char* pBeg = pOut + pMapOff[i] % nGran;

			for(int j = pBegPos[i]; j < pBegPos[i + 1]; ++j) XSAVENUM(pBeg, pArray->pData[j]);

			// 关闭映射
			XFileMapUnView((void*)pOut, nMapSize);
		}
	}

	XFileMapClose(&hFileMap);
}