parallel_radixsort.cpp

C/C++ 多任务下的数据结构与算法 （周伟明）华中科技大学出版社

		//收集操作，将数据重新放回ppData里
		UINT uIndex = 0;
		for ( k = 0; k < uRadix; k++ )
		{
			for ( i = 0; i < nProcessors; i++ )
			{
				UINT j;
				SORTTABLE *pTable = pBox[i].ppTable[k];
				if ( pTable != NULL )
				{
					UINT uCount = pTable->uCursorCount;
					for ( j = 0; j < uCount; j++ )
					{
						ppData[uIndex] = pTable->ppData[j];
						uIndex++;
					}
				}
			}
		}
		//释放表
		for ( i = 0; i < nProcessors; i++ )
		{
			for ( k = 0; k < uRadix; k ++ )
			{
				SortTable_Destroy(pBox[i].ppTable[k], NULL);
			}
		}
		//进行下一轮的分配与收集操作
	}	//for ( uKeyIndex = 0; uKeyIndex < uMaxKeyLen; uKeyIndex++ )
}



/**	填入描述

	@param	UINT uRadix - xxxx	
	@param	int nProcessors - xxxx	
	@param	UINT uDataBlockLen - xxxx	
	@return	ARRAYLIST_ARRAY * - xxxx	
*/
ARRAYLIST_ARRAY *BoxArray_Create(UINT uRadix, int nProcessors, UINT uDataBlockLen)
{
	int i;
	ARRAYLIST_ARRAY *pBoxArray;
	pBoxArray = (ARRAYLIST_ARRAY *)malloc(sizeof(ARRAYLIST_ARRAY) * nProcessors);
	if ( pBoxArray == NULL )
	{
		return NULL;
	}

	for ( i = 0; i < nProcessors; i++ )
	{
		pBoxArray[i].ppList = (ARRAYLIST **)malloc(sizeof(ARRAYLIST *) * uRadix);
		UINT k;
		for ( k = 0; k < uRadix; k++ )
		{
			pBoxArray[i].ppList[k] = ArrayList_Create(uDataBlockLen);
		}
	}
	pBoxArray->nProcessors = nProcessors;
	pBoxArray->uDataBlockLen = uDataBlockLen;
	pBoxArray->uRadix = uRadix;
	return pBoxArray;
}


/**	填入描述

	@param	ARRAYLIST_ARRAY *pBoxArray - xxxx	
	@return	void - xxxx	
*/
void BoxArray_Destroy(ARRAYLIST_ARRAY *pBoxArray)
{
	int i;
	if ( pBoxArray == NULL )
	{
		return ;
	}

	for ( i = 0; i < pBoxArray->nProcessors; i++ )
	{
		UINT k;
		for ( k = 0; k < pBoxArray->uRadix; k++ )
		{
			ArrayList_Destroy(pBoxArray[i].ppList[k], NULL);
		}
		free(pBoxArray[i].ppList);
	}
	free(pBoxArray);
}


/**	填入描述

	@param	ARRAYLIST_ARRAY *pBoxArray - xxxx	
	@param	UINT uProcessors - xxxx	
	@param	UINT uCount - xxxx	
	@return	RADIXBOX_POSITION * - xxxx	
*/
RADIXBOX_POSITION *GetPosArray(ARRAYLIST_ARRAY *pBoxArray, UINT uProcessors, UINT uCount)
{
	UINT		uTotalSize = 0;
	UINT		uSize;
	UINT		uPosIndex;
	ARRAYLIST	*pList;
//	ARRAYNODE	*pNode;
	UINT		uBoxArrayIndex = 0;
	UINT		uArrayListIndex = 0;

	RADIXBOX_POSITION	*pPos = 
		(RADIXBOX_POSITION *)malloc(sizeof(RADIXBOX_POSITION)*(uProcessors+1));

	pList = pBoxArray[uBoxArrayIndex].ppList[uArrayListIndex];
	if ( pList->pTail != NULL )
	{
		uSize = (pList->uNodeCount - 1) * pList->uDataLen + pList->pTail->uCurPos;
	}
	else
	{
		uSize = 0;
	}
	uTotalSize += uSize;

	pPos[0].uArrayListIndex = 0;
	pPos[0].uBoxArrayIndex = 0;
	pPos[0].uPos = 0;

	uPosIndex = 1;

	while ( uPosIndex < uProcessors )
	{
		if ( uTotalSize < uCount )
		{
			uBoxArrayIndex++;
			if ( uBoxArrayIndex >= uProcessors )
			{
				uBoxArrayIndex = 0;
				uArrayListIndex++;
			}
			pList = pBoxArray[uBoxArrayIndex].ppList[uArrayListIndex];
			if ( pList->pTail != NULL )
			{
				uSize = (pList->uNodeCount - 1) * pList->uDataLen + pList->pTail->uCurPos;
			}
			else
			{
				uSize = 0;
			}
			uTotalSize += uSize;
			continue;
		}
		else
		{
			pPos[uPosIndex].uArrayListIndex = uArrayListIndex;
			pPos[uPosIndex].uBoxArrayIndex = uBoxArrayIndex;
			pPos[uPosIndex].uPos = uCount + uSize - uTotalSize;
			uPosIndex++;

			uTotalSize -= uCount;
		}
	}

	pPos[uPosIndex].uArrayListIndex = pBoxArray->uRadix - 1;
	pPos[uPosIndex].uBoxArrayIndex = uProcessors - 1;
	pList = pBoxArray[uProcessors - 1].ppList[pBoxArray->uRadix-1];
	if ( pList->pTail != NULL )
	{
		pPos[uPosIndex].uPos = (pList->uNodeCount - 1) * pList->uDataLen 
			+ pList->pTail->uCurPos;
	}
	else
	{
		pPos[uPosIndex].uPos = 0;
	}
	return pPos;
}


/**	填入描述

	@param	void **ppData - 待排序数据	
	@param	UINT uDataLen - 数据长度	
	@param	UINT uRadix - 基数	
	@param	UINT uMaxKeyLen - 关键词长度	
	@param	GETKEYFUNC GetKeyFunc - 关键词取位回调函数	
	@return	INT - CAPI_SUCCESS	
*/
INT Parallel_RadixSort_ArrayList(void **ppData, UINT uDataLen,
							  UINT uRadix, UINT uMaxKeyLen, GETKEYFUNC GetKeyFunc)
{
	UINT	i,j,k;
	ARRAYLIST_ARRAY	*pBoxArray;
	ARRAYLIST_ARRAY *pBoxArrayTemp;
	UINT nProcessors = omp_get_num_procs();
	UINT uCount = uDataLen / nProcessors;

	if ( uDataLen - uCount * nProcessors > 2 )
	{
		++uCount;
	}
	UINT	uDataBlockLen = uCount / uRadix + 128;
	pBoxArray = BoxArray_Create(uRadix, nProcessors, uDataBlockLen);
	
	UINT uKeyIndex = 0;

	// 进行第1轮分配操作
	for ( i = 0; i < nProcessors; i++ )
	{
		//计算各个线程操作数据的起始位置
		UINT begin = i * uCount;
		UINT end = (i+1) * uCount;
		if ( end > uDataLen )
		{
			end = uDataLen;
		}
		ARRAYLIST	*pList;
		ARRAYNODE	*pNode;

		for ( k = begin ; k < end; k++ )
		{
			UINT uPos = (*GetKeyFunc)(ppData[k], uKeyIndex);

			pList = pBoxArray[i].ppList[uPos];

			pNode = pList->pTail;

			if ( pNode->uCurPos < uDataBlockLen )
			{
				pNode->ppData[pNode->uCurPos] = ppData[k];
				pNode->uCurPos++;
			}
			else
			{
				pNode = ArrayNode_Create(uDataBlockLen);
				if ( pNode == NULL )
				{
					return CAPI_FAILED;
				}
				pList->pTail->pNext = pNode;
				pList->pTail = pNode;
				pList->uNodeCount++;

				pNode->ppData[0] = ppData[k];
				pNode->uCurPos = 1;
			}
		}

	}//第1轮分配操作结束

	for ( uKeyIndex = 1; uKeyIndex < uMaxKeyLen; uKeyIndex++ )
	{
		RADIXBOX_POSITION	*pPos = GetPosArray(pBoxArray, nProcessors, uCount);
		pBoxArrayTemp = BoxArray_Create(uRadix, nProcessors, uDataBlockLen);

		// 进行第uKeyIndex轮分配操作
		for ( i = 0; i < nProcessors; i++ )
		{
			ARRAYLIST	*pList;
			ARRAYNODE	*pNode;
			//计算各个线程操作数据的起始位置
			UINT	uArrayListIndexStart = pPos[i].uArrayListIndex;
			UINT	uArrayListIndexEnd = pPos[i+1].uArrayListIndex;
			
			UINT	uBoxArrayIndexStart = pPos[i].uBoxArrayIndex;
			UINT	uBoxArrayIndexEnd = pPos[i+1].uBoxArrayIndex;

			UINT	uStartPos = pPos[i].uPos;
			UINT	uEndPos = pPos[i+1].uPos;

			pList = pBoxArray[uBoxArrayIndexStart].ppList[uArrayListIndexStart];
			pNode = pList->pHead;

			UINT uNodeNo = uStartPos / pList->uDataLen;
			UINT uStart = uStartPos - uNodeNo * pList->uDataLen;
			for ( k = 0; k < uNodeNo; k++ )
			{
				pNode = pNode->pNext;
			}

			while ( uArrayListIndexStart < uArrayListIndexEnd )
			{
				while ( uBoxArrayIndexStart < nProcessors )
				{
					while ( pNode != NULL )
					{
						for ( k = uStart; k < pNode->uCurPos; k++ )
						{
							UINT uPos = (*GetKeyFunc)(pNode->ppData[k], uKeyIndex);
							pList = pBoxArrayTemp[i].ppList[uPos];
							ArrayList_Add(pList, pNode->ppData[k]);

							//Add(pNode->ppData[i]);
						}
						pNode = pNode->pNext;
						uStart = 0;
					}
					++uBoxArrayIndexStart;
					pList = pBoxArray[uBoxArrayIndexStart].ppList[uArrayListIndexStart];
					pNode = pList->pHead;
				}
				++uArrayListIndexStart;
			}
			uBoxArrayIndexStart = 0;

			while ( uBoxArrayIndexStart < uBoxArrayIndexEnd )
			{
				pList = pBoxArray[uBoxArrayIndexStart].ppList[uArrayListIndexEnd];
				pNode = pList->pHead;

				while ( pNode != NULL )
				{
					for ( k = uStart; k < pNode->uCurPos; k++ )
					{
						UINT uPos = (*GetKeyFunc)(pNode->ppData[k], uKeyIndex);
						pList = pBoxArrayTemp[i].ppList[uPos];
						ArrayList_Add(pList, pNode->ppData[k]);
						//Add(pNode->ppData[i]);
					}
					pNode = pNode->pNext;
					uStart = 0;
				}
				++uBoxArrayIndexStart;
			}

			pList = pBoxArray[uBoxArrayIndexEnd].ppList[uArrayListIndexEnd];
			pNode = pList->pHead;
			uNodeNo = uEndPos / pList->uDataLen;

			for ( j = 0; j < uNodeNo; j++ )
			{
				for ( k = 0; k < pNode->uCurPos; k++)
				{
					UINT uPos = (*GetKeyFunc)(pNode->ppData[k], uKeyIndex);
					pList = pBoxArrayTemp[i].ppList[uPos];
					ArrayList_Add(pList, pNode->ppData[k]);
					//Add(pNode->ppData[i]);
				}
				pNode = pNode->pNext; 
			}
			if ( pNode != NULL )
			{
				UINT uEnd = uEndPos - uNodeNo* pList->uDataLen;
				for ( i = 0; i < uEnd; i++ )
				{
					UINT uPos = (*GetKeyFunc)(pNode->ppData[k], uKeyIndex);
					pList = pBoxArrayTemp[i].ppList[uPos];
					ArrayList_Add(pList, pNode->ppData[k]);
					//Add(pNode->ppData[i]);			
				}
			}
		}//第uKeyIndex轮分配操作结束
		
		//释放pBoxArray
		BoxArray_Destroy(pBoxArray);
		pBoxArray = pBoxArrayTemp;
	}
    return CAPI_SUCCESS;
}



/**	并行基数排序

	@param	void **ppData - 待排序数据	
	@param	UINT uDataLen - 数据长度	
	@param	UINT uRadix - 基数	
	@param	UINT uMaxKeyLen - 最大关键词长度	
	@param	GETKEYFUNC GetKeyFunc - 关键词取位回调函数	
	@return	void - 无	
*/
void Parallel_RadixSort_Array2(void **ppData, UINT uDataLen,
							  UINT uRadix, UINT uMaxKeyLen, GETKEYFUNC GetKeyFunc)
{
	int i;
	int nProcessors = omp_get_num_procs();
	UINT uCount = uDataLen / nProcessors;

	if ( uDataLen - uCount * nProcessors > 2 )
	{
		uCount++;
	}

	RADIX_BOX2	*pBox = (RADIX_BOX2 *)malloc(sizeof(RADIX_BOX2) * nProcessors);

	void *pTemp = malloc(sizeof(DATA_ARRAY *) * uRadix * nProcessors);

	for ( i = 0; i < nProcessors; i++ )
	{
		pBox[i].pDataArray = (DATA_ARRAY *)((char *)pTemp + i * sizeof(DATA_ARRAY) * uRadix);
		pBox[i].uBoxLen = uRadix;
	}

	void ** ppTempData = (void **)malloc( sizeof(void *) * uDataLen );

	UINT uKeyIndex;

	for ( uKeyIndex = 0; uKeyIndex < uMaxKeyLen; uKeyIndex++ )
	{
		UINT k;
		//下面代码完成全局的一轮将数据分配到箱子里的操作
#pragma omp parallel for private(k)	
		for ( i = 0; i < nProcessors; i++ )
		{
			UINT	*pDataSum = (UINT *)malloc(uRadix * sizeof(UINT));

			// 初始化元素计数为0
			for ( k = 0; k < uRadix; k++ )
			{
				pDataSum[k] = 0;
			}

			//每个线程计算自己的数据中各个位的元素个数
			UINT begin = i * uCount;
			UINT end = (i+1) * uCount;
			if ( end > uDataLen )
			{
				end = uDataLen;
			}
			for ( k = begin ; k < end; k++ )
			{
				UINT uPos = (*GetKeyFunc)(ppData[k], uKeyIndex);
				pDataSum[uPos] += 1;
			}

			void **ppPrivateData = (void **)(&(ppTempData[begin]));
			DATA_ARRAY *pDataArray = pBox[i].pDataArray;

			pDataArray[0].ppData = ppPrivateData;

			int nDataSum = 0;
			//按计算出来的元素个数来分配箱子的空间大小
			for ( k = 1; k < uRadix; k++ )
			{
				//pBox[i].ppTable[k] = SortTable_Create(pDataSum[k]);
				nDataSum += pDataSum[k-1];
				pDataArray[k].ppData = (void **)( &(ppPrivateData[nDataSum]) ); 
				pDataArray[k].uCount = 0;
			}


			// 将数据分配到各个箱子里
			for ( k = begin ; k < end; k++ )
			{
				UINT uPos = (*GetKeyFunc)(ppData[k], uKeyIndex);
				pDataArray[uPos].ppData[pDataArray[uPos].uCount] = ppData[k];
				pDataArray[uPos].uCount += 1;

				//SortTable_Add(pBox[i].ppTable[uPos], ppData[k]);
			}
		}//for ( i = 0; i < nProcessors; i++ )

		//收集操作，将数据重新放回ppData里
		UINT uIndex = 0;
		for ( k = 0; k < uRadix; k++ )
		{
			for ( i = 0; i < nProcessors; i++ )
			{
				UINT j;
				DATA_ARRAY *pArray = &(pBox[i].pDataArray[k]);
				UINT uCount = pArray->uCount;
				for ( j = 0; j < uCount; j++ )
				{
					ppData[uIndex] = pArray->ppData[j];
					uIndex++;
				}
			}
		}
		//进行下一轮的分配与收集操作
	}	//for ( uKeyIndex = 0; uKeyIndex < uMaxKeyLen; uKeyIndex++ )
	free(ppTempData);
	free(pTemp);
	free(pBox);
}


/**	分区的并行基数排序函数

	@param	void **ppData - 待排序数据	
	@param	UINT uDataLen - 数据长度	
	@param	UINT uRadix - 基数	
	@param	UINT uMaxKeyLen - 关键词长度	
	@param	GETKEYFUNC GetKeyFunc - 关键词取位回调函数	
	@return	void - 无	
*/
void Parallel_Partitioned_RadixSort(void **ppData, UINT uDataLen, UINT uRadix, 
									UINT uMaxKeyLen, GETKEYFUNC GetKeyFunc)
{

	//各个线程计算自己的各个箱子大小
	int     i, nProcessors;
    UINT    k;

	nProcessors = omp_get_num_procs();
	UINT uCount = uDataLen / nProcessors;

	if ( uDataLen - uCount * nProcessors > 2 )
	{
		uCount++;
	}

	UINT	**ppDataSum = (UINT **)malloc(sizeof(UINT *) * nProcessors);
#pragma omp parallel for private(k)	
	for ( i = 0; i < nProcessors; i++ )
	{
		UINT	*pDataSum;
		pDataSum = (UINT *)malloc(uRadix * sizeof(UINT));
		ppDataSum[i] = pDataSum;

		// 初始化元素计数为0
		for ( k = 0; k < uRadix; k++ )
		{
			pDataSum[k] = 0;
		}

		//每个线程计算自己的数据中各个位的元素个数
		UINT begin = i * uCount;
		UINT end = (i+1) * uCount;
		if ( end > uDataLen )
		{
			end = uDataLen;
		}
		for ( k = begin ; k < end; k++ )
		{
			UINT uPos = (*GetKeyFunc)(ppData[k], 0);
			pDataSum[uPos] += 1;
		}
	}

	//计算全局的各个箱子的大小
	UINT	*pGlobalDataSum = (UINT *)malloc(uRadix * sizeof(UINT));
	for ( k = 0; k < uRadix; k++ )
	{
		pGlobalDataSum[k] = 0;
	}

	for ( i = 1; i < nProcessors; i++ )
	{
		UINT *pTempDataSum = ppDataSum[i];
		for ( k = 0; k < uRadix; k++ )
		{
			pGlobalDataSum[k] += pTempDataSum[k];
		}
	}


	//分配一个大的数组，数组长度等于输入数据的长度
	void **ppTempData = (void **)malloc(uDataLen * sizeof(void *));

	//将数组划分成一系列的箱子，保证箱子按照收集操作的顺序摆放
	RADIX_BOX2 *pBox = (RADIX_BOX2 *)malloc(sizeof(RADIX_BOX2)* nProcessors);

	void *pTemp = malloc(sizeof(DATA_ARRAY *) * uRadix * nProcessors);

	for ( i = 0; i < nProcessors; i++ )
	{
		pBox[i].pDataArray = (DATA_ARRAY *)((char *)pTemp + i * sizeof(DATA_ARRAY) * uRadix);
		pBox[i].uBoxLen = uRadix;
	}


	void **pp = ppTempData;

	for ( i = 0; i < (int)uRadix; i++ )
	{
		for ( k = 0; k < (UINT)nProcessors; k++ )
		{
			pBox[k].pDataArray[i].ppData = pp;//pBox[i].pDataArray[k-1].ppData + ppDataSum[i][k-1]; 
			pp += ppDataSum[k][i];
			pBox[k].pDataArray[i].uCount = 0;
		}
	}

	//各个线程将数据放入到对应的箱子里，这样可以得到前一个全局箱子的所有数据
    //一定小于后一个全局箱子的所有数据
	UINT begin = 0;
	UINT end = 0;
	for ( i = 0; i < nProcessors; i++ )
	{
		begin = i * uCount;
		end = (i+1) * uCount;
		if ( end > uDataLen )
		{
			end = uDataLen;
		}
		DATA_ARRAY	*pDataArray = pBox[i].pDataArray;
		for ( k = begin ; k < end; k++ )
		{
			UINT uPos = (*GetKeyFunc)(ppData[k], 0);
			pDataArray[uPos].ppData[pDataArray[uPos].uCount] = ppData[k];
			pDataArray[uPos].uCount += 1;

			//SortTable_Add(pBox[i].ppTable[uPos], ppData[k]);
		}
	}


	//对uRadix个全局箱子内的数据进行排序，每个线程内使用串行排序算法即可
	begin = 0;
	end = 0;

	for ( i = 0; i < (int)uRadix; i++ )
	{
		end = begin + pGlobalDataSum[i];  
//		Serial_RadixSort(pTempData, 
	}


	//对所有的全局箱子排好序后，自然得到了一个有序的数据序列
}