marsutils.cu

GPU实现的MapReduce framework,对于学习并行编程和cuda平台的编程方面有着极好的参考价值

	chunk->recCount = recCount;

	//get keySize & keyOffset & allocate key buffer
	size_t keyOffset = chunk->index[0].x;
	size_t keySize = chunk->index[recCount-1].x + 
		chunk->index[recCount-1].y - chunk->index[0].x;
	//chunk->keys = (char*)BenMalloc(keySize);
	chunk->keys = BenReadFile(file->keyFile, keyOffset, keySize);
	chunk->keySize = keySize;
	chunk->keyOffset = keyOffset;

	//get valSize & valOffset & allocate val buffer
	size_t valOffset = chunk->index[0].z;
	size_t valSize = chunk->index[recCount-1].z +
		chunk->index[recCount-1].w - chunk->index[0].z;
	//chunk->vals = (char*)BenMalloc(valSize);
	chunk->vals = BenReadFile(file->valFile, valOffset, valSize);
	chunk->valSize = valSize;
	chunk->valOffset = valOffset;

	return true;
}

bool WriteChunkToFile(ChunkInfo_t *chunk, FileName_t *file, 
					  size_t *totalRecCount, char mode)
{
	BEN_ASSERT(chunk != NULL);
	BEN_ASSERT(file != NULL);
	BEN_ASSERT(totalRecCount != NULL);
	
	//{!!! when map only appers
	if (mode & MAP_ONLY)
		for (int i = 0; i < chunk->recCount; i++)
		{
			chunk->index[i].x += chunk->keyOffset;
			chunk->index[i].z += chunk->valOffset;
		}
	//!!!}

	if (*totalRecCount > 0)
	{
		BenAppendFile(file->keyFile, chunk->keys, chunk->keySize);
		BenAppendFile(file->valFile, chunk->vals, chunk->valSize);
		BenAppendFile(file->indexFile, chunk->index, chunk->indexSize);
		if (!(mode & MAP_ONLY) || (mode & EXTERNAL_SORT))
			BenAppendFile(file->rangeFile, chunk->keyListRange, chunk->rangeSize);
	}
	else
	{
		BenWriteFile(file->keyFile, chunk->keys, chunk->keySize);
		BenWriteFile(file->valFile, chunk->vals, chunk->valSize);
		BenWriteFile(file->indexFile, chunk->index, chunk->indexSize);
		if (!(mode & MAP_ONLY) || (mode & EXTERNAL_SORT))
			BenWriteFile(file->rangeFile, chunk->keyListRange, chunk->rangeSize);	
	}
	*totalRecCount += chunk->recCount;
	chunk->keyOffset += chunk->keySize;
	chunk->valOffset += chunk->valSize;

	if (!(mode & EXTERNAL_SORT))
		FreeChunk(chunk);

	return true;
}
//------------------------------------------------
//return: 0 -- finish, 1 -- unfinish
//------------------------------------------------
char NextRec(RecIterator_t *it, Record_t *rec)
{
	BEN_ASSERT(rec != NULL);
	BEN_ASSERT(it->chunk != NULL);
		
	static size_t allKeyOffset = 0;
	static size_t allValOffset = 0;

	if (it->file == NULL)
	{
		rec->keyOffset = it->chunk->keyOffset;
		rec->valOffset = it->chunk->valOffset;
		allKeyOffset = rec->keyOffset;
		allValOffset = rec->valOffset;
	}

	if (it->cursor >= it->chunk->recCount)
	{
		if (it->file == NULL) 
			return 0;
		else 
		{
			if (it->fileCursor >= it->totalRecCount)
				return 0;
			BenFree((char**)&it->chunk->keys, it->chunk->keySize);
			BenFree((char**)&it->chunk->vals, it->chunk->valSize);
			BenFree((char**)&it->chunk->index, it->chunk->indexSize);	
			if (!ReadChunkFromFile(it->chunk, it->file, 
				it->totalRecCount, it->fileCursor, it->chunkSize))
				return 0;
			else
			{
				allKeyOffset = it->chunk->index[0].x;
				allValOffset = it->chunk->index[0].z;
			//	ReadChunkFromFile(it->chunk, it->file, 
			//	it->totalRecCount, it->fileCursor, it->chunkSize);
				it->fileCursor += it->chunk->recCount;
				it->cursor = 0;
			}
		}
	}

	int4 index = it->chunk->index[it->cursor];
	rec->key = it->chunk->keys + index.x - allKeyOffset;
	rec->val = it->chunk->vals + index.z - allValOffset;
	rec->keySize = index.y;
	rec->valSize = index.w;
	rec->keyOffset = it->chunk->keyOffset;
	rec->valOffset = it->chunk->valOffset;
	it->cursor++;

	return 1;
}
 
//------------------------------------------------
//return: 0 -- finish, 1 -- unfinish
//------------------------------------------------
char NextGroup(GroupIterator_t *it, Group_t *rec)
{
	BEN_ASSERT(rec != NULL);
	BEN_ASSERT(it->chunk != NULL);
		
	if (it->cursor >= it->chunk->diffKeyCount)
		return 0;

	//read from main memory
	if (it->file == NULL)
	{
		int2 keyListIndex = it->chunk->keyListRange[it->cursor];
		int4 index = it->chunk->index[keyListIndex.x];
		rec->index = &(it->chunk->index[keyListIndex.x]);
		rec->key = it->chunk->keys + index.x;
		rec->vals = it->chunk->vals;
		rec->keySize = index.y;
		rec->valCount = keyListIndex.y - keyListIndex.x;
		it->cursor++;
	}
	else
	{
	}

	return 1;
}

char NextRecInGroup(Group_t *group, Record_t *rec)
{
	BEN_ASSERT(rec != NULL);
	BEN_ASSERT(group != NULL);

	if (group->cursor >= group->valCount)
	{
		group->cursor = 0;
		return 0;	
	}

	int4 *index = group->index + group->cursor;
		
	rec->key = group->key;
	rec->val = group->vals + index->z;
	rec->keySize = index->y;
	rec->valSize = index->w;
	rec->keyOffset = index->x;
	rec->valOffset = index->z;
	group->cursor++;
	return 1;
}

void FinishIterator(RecIterator_t *it)
{
	BenFree((char**)&it, sizeof(RecIterator_t));
}

void FinishGroupIterator(GroupIterator_t *it)
{
	BenFree((char**)&it, sizeof(GroupIterator_t));
}
//========================================================
//For Mars debug
//========================================================
//--------------------------------------------------------
//Generate test data
//--------------------------------------------------------
#define STRING_LEN	6
void GenRecords(Spec_t *spec, char keyType, char valType, int maxRec)
{
	char *alpha = "abcdefghijklmnopqrstuvwxyz";
	srand(time(0));
	
	for (int i = 0; i < maxRec; i++)
	{
		char *key = NULL;
		char *val = NULL;
		size_t keySize = 0;
		size_t valSize = 0;

		switch (keyType)
		{
		case INT:
			key = (char*)BenMalloc(sizeof(int));
			*(int*)key = rand()%maxRec;
			keySize = sizeof(int);
			break;
		case STRING:
			keySize = rand()%STRING_LEN+1;
			key = (char*)BenMalloc(keySize+1);
			for (int j = 0; j < keySize; j++)
				key[j] = alpha[rand()%26];
			key[keySize] = '\0';
			keySize++;
			break;
		case FLOAT:
			keySize = sizeof(float);
			key = (char*)BenMalloc(sizeof(float));
			*(float*)key = (float)(rand()%maxRec);
			break;
		}

		switch (valType)
		{
		case INT:
			val = (char*)BenMalloc(sizeof(int));
			*(int*)val = rand()%maxRec;
			valSize = sizeof(int);
			break;
		case STRING:
			valSize = rand()%STRING_LEN+1;
			val = (char*)BenMalloc(valSize+1);
			for (int j = 0; j < valSize; j++)
				val[j] = alpha[rand()%26];
			val[valSize] = '\0';
			valSize++;
			break;
		case FLOAT:
			valSize = sizeof(float);
			val = (char*)BenMalloc(sizeof(float));
			*(float*)val = (float)(rand()%maxRec);
			break;
		}

		AddMapInputRecord(spec, key, val, keySize, valSize);
		BenFree((char**)&key, keySize);
		BenFree((char**)&val, valSize);
	}
}

//--------------------------------------------------------
//Print key/value pairs
//type: STRING/CHAR/INT/FLOAT
//--------------------------------------------------------
void PrintRecords(ChunkInfo_t *chunk, FileName_t *file, size_t totalRec, Spec_t *spec,
				 char keyType, char valType, int maxRec)
{
	if (file != NULL)
		BenMemset(chunk, 0, sizeof(ChunkInfo_t));

	RecIterator_t *it = InitIterator(chunk, file, totalRec, spec);

	BenLog("********total record %d*********\n", it->totalRecCount);
	Record_t rec;
	int count = maxRec;
	if (maxRec > it->totalRecCount)
		count = it->totalRecCount;

	while (NextRec(it, &rec))
	{
		if (count <= 0) break;
		switch (keyType)
		{
		case INT:
			BenLog("key:%d\t", *(int*)rec.key);
			break;
		case STRING:
			BenLog("key:%s\t", (char*)rec.key);
			break;
		case FLOAT:
			BenLog("key:%f\t", *(float*)rec.key);
			break;
		}

		switch (valType)
		{
		case INT:
			BenLog("val:%d\t", *(int*)rec.val);
			break;
		case STRING:
			BenLog("val:%s\t", (char*)rec.val);
			break;
		case FLOAT:
			BenLog("val:%f\t", *(float*)rec.val);
			break;
		}

		BenLog("keySize:%d\tvalSize:%d\tkeyOffset:%d\tvalOffset:%d\n",
			rec.keySize, rec.valSize, rec.keyOffset, rec.valOffset);
		count --;
	}

	FinishIterator(it);
	BenLog("************%d records are printed********\n", maxRec);
}

void PrintGroup(ChunkInfo_t *chunk,  FileName_t *file, size_t totalGroupNum, Spec_t *spec,
				 char keyType, char valType, int maxGroup, int maxRec)
{
	if (file != NULL)
		BenMemset(chunk, 0, sizeof(ChunkInfo_t));

	GroupIterator_t *it = InitGroupIterator(chunk, file, totalGroupNum, spec);

	BenLog("********totally %d groups, %d records*********\n", 
		it->chunk->diffKeyCount, it->chunk->recCount);

	Group_t group;
	BenMemset(&group, 0, sizeof(Group_t));
	Record_t rec;
	BenMemset(&rec, 0, sizeof(Record_t));

	int count = maxGroup;

	while (NextGroup(it, &group))
	{
		if (count <= 0) break;
		switch (keyType)
		{
		case INT:
			BenLog("-------key:%d, valCount:%d------\n", 
				*(int*)group.key, group.valCount);
			break;
		case STRING:
			BenLog("-------key:%s, valCount:%d------\n", 
				(char*)group.key, group.valCount);
			break;
		case FLOAT:
			BenLog("-------key:%f, valCount:%d------\n", 
				*(float*)group.key, group.valCount);
			break;
		}

		int recCount = maxRec;
		while (NextRecInGroup(&group, &rec))
		{
			if (recCount <= 0) break;
			switch (valType)
			{
			case INT:
				BenLog("val:%d\t", *(int*)rec.val);
				break;
			case STRING:
				BenLog("val:%s\t", (char*)rec.val);
				break;
			case FLOAT:
				BenLog("val:%f\t", *(float*)rec.val);
				break;
			}
			recCount--;
		}
		BenLog("\n");
		count --;
	}

	FinishGroupIterator(it);
	BenLog("************up to %d groups, each up to %d recs are printed********\n", 
		maxGroup, maxRec);
}