bencudafile.cu

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

/*
*System-independent library.
 *Written by Wenbin FANG ([e-mail])
 *
 *COMPILE MACROS:
 *1, __UNIX__ or __WIN32__
 *2, __TIMER__
 *3, __DEBUG__
 *4, __CUDA__
 *5, __MEM__
 *6, __ALLOC__
 */

/*********************************************************
	===Example1-iterate data chunks in a file===
	FStream_t *fs = D_OPEN_FILE("test.dat");

	size_t chunkSize = 100;
	char *gpubuf = D_MALLOC(chunkSize);
	size_t readSize = 0;
	while ((readSize = D_ITERATE_CHUNK(fs, gpubuf, chunkSize)))
	{
		test_kernel<<<4,4>>>(gpubuf);
	}
	D_FREE(gpubuf, chunkSize);

	D_CLOSE_FILE(fs);
**********************************************************
	===Example2-overwrite a file, then append chunks===
	FStream_t *fs = D_OPEN_FILE("test.dat");
	FStream_t *wfs = D_OPEN_FILE("wtest.dat");

	size_t chunkSize = 100;
	char *gpubuf = D_MALLOC(chunkSize);
	size_t readSize = 0;
	while ((readSize = D_ITERATE_CHUNK(fs, gpubuf, chunkSize)))
	{
		test_kernel<<<4,4>>>(gpubuf);
		D_WRITE_CHUNK(wfs, gpubuf, readSize);
	}
	D_FREE(gpubuf, chunkSize);

	D_CLOSE_FILE(fs);
	D_CLOSE_FILE(wfs);
***********************************************************
	===Example3-just append chunks to a file===
	FStream_t *fs = D_OPEN_FILE("test.dat");
	FStream_t *wfs = D_OPEN_FILE("wtest.dat");

	size_t chunkSize = 100;
	char *gpubuf = D_MALLOC(chunkSize);
	size_t readSize = 0;
	while ((readSize = D_ITERATE_CHUNK(fs, gpubuf, chunkSize)))
	{
		test_kernel<<<4,4>>>(gpubuf);
		D_APPEND_CHUNK(wfs, gpubuf, readSize);
	}
	D_FREE(gpubuf, chunkSize);

	D_CLOSE_FILE(fs);
	D_CLOSE_FILE(wfs);
 */

#include "BenLib.h"

FStream_t *D_OPEN_FILE(char *filename)
{
	BEN_ASSERT(filename != NULL);

	FStream_t *fs = (FStream_t*)BenMalloc(sizeof(FStream_t));
	fs->offset = 0;
#ifdef __UNIX__
#else //__UNIX__
	HANDLE hFile; 
	hFile = CreateFile(filename,               
                       GENERIC_WRITE|GENERIC_READ,          
                       FILE_SHARE_WRITE|FILE_SHARE_READ,      
                       NULL,               
                       OPEN_ALWAYS,   
                       FILE_ATTRIBUTE_NORMAL,
                       NULL);            
	BEN_ASSERT(hFile != INVALID_HANDLE_VALUE );

	fs->filename = filename;
	fs->hFile = hFile;
	size_t hi_size;
	fs->fileSize = GetFileSize(hFile, (LPDWORD)&hi_size);

#endif //__WIN32__
	return fs;
}

size_t D_ITERATE_CHUNK(FStream_t *fs, char *gpubuf, 
					 size_t chunkSize)
{
	BEN_ASSERT(fs != NULL);
	BEN_ASSERT(chunkSize != NULL);

	size_t readSize = 0;

	if (D_READ_CHUNK(fs, gpubuf, fs->offset, chunkSize, &readSize))
	{
		fs->offset += readSize;
		return readSize;
	}
	else 
		return 0;
}

void D_RESET_ITERATOR(FStream_t *fs)
{
	BEN_ASSERT(fs != NULL);

	fs->offset = 0;
}

char D_READ_CHUNK(FStream_t *fs, char *gpubuf, 
				  size_t offset, size_t size, size_t *read_size)
{
	BEN_ASSERT(gpubuf != NULL);
	BEN_ASSERT(fs != NULL);

	if (offset >= fs->fileSize) return 0;

	char *cpubuf = (char*)BenMalloc(size);
#ifndef __UNIX__
	HANDLE hFile = fs->hFile;

	OVERLAPPED o;
	BenMemset(&o, 0, sizeof(o));
	o.Offset = offset;
	BEN_ASSERT(ReadFile(hFile, cpubuf, size, (LPDWORD)read_size, &o));

	D_MEMCPY_H2D(gpubuf, cpubuf, *read_size);
#endif //__UNIX__
	BenFree((char**)&cpubuf, size);
	return 1;
}

char D_WRITE_CHUNK(FStream_t *fs, char *gpubuf, size_t size)
{
	char *cpubuf = (char*)BenMalloc(size);
	D_MEMCPY_D2H(cpubuf, gpubuf, size);

	size_t write_size = 0;
	BEN_ASSERT(WriteFile (fs->hFile, cpubuf, size, (LPDWORD)&write_size, NULL));
	BenFree((char**)&cpubuf, size);
	return 0;
}

char D_APPEND_CHUNK(FStream_t *fs, char *gpubuf, size_t size)
{
	BEN_ASSERT(fs != NULL);
	BEN_ASSERT(gpubuf != NULL);

	char *cpubuf = (char*)BenMalloc(size);
	D_MEMCPY_D2H(cpubuf, gpubuf, size);

#define FILE_WRITE_TO_END_OF_FILE       0xffffffff

	DWORD dwResult = 0; 
	OVERLAPPED o;
	BenMemset(&o, 0, sizeof(o));
	o.Offset = FILE_WRITE_TO_END_OF_FILE; 
	o.OffsetHigh = 0;
	BEN_ASSERT(WriteFile(fs->hFile, cpubuf, size, &dwResult, &o));
	
	BenFree((char**)&cpubuf, size);
	return 0;
}

void D_CLOSE_FILE(FStream_t *fs)
{
	BEN_ASSERT(fs != NULL);
#ifndef __UNIX__
	CloseHandle(fs->hFile);
#endif //__UNIX__
	BenFree((char**)&fs, sizeof(FStream_t));
}