一个c++buffer的设计和实现

#include <process.h> #include <fstream> #include <stdio.h> #include "NonLockingBuffer.h" #define PORT_NUMBER 9998 #define GROUP_ITEM_COUNT 128 #define THREAD_RUN_LOOP_TIMES 60 #define MAX_THREAD_COUNT_NUMBER 40 class CThreadTime { public: void BeginGetElapsedTime(); __int64 EndGetElapsedTime(); private: __int64 FileTimeToQuadWord(PFILETIME pft); private: FILETIME ftKernelTimeStart; FILETIME ftKernelTimeEnd; FILETIME ftUserTimeStart; FILETIME ftUserTimeEnd; FILETIME ftDummy; }; // Get the time elapsed since the thread start inline void CThreadTime::BeginGetElapsedTime() { GetThreadTimes(GetCurrentThread(), &ftDummy, &ftDummy, &ftKernelTimeStart, &ftUserTimeStart); } // Calculate the time elapsed inline __int64 CThreadTime::EndGetElapsedTime() { GetThreadTimes(GetCurrentThread(), &ftDummy, &ftDummy, &ftKernelTimeEnd, &ftUserTimeEnd); __int64 qwKernelTimeElapsed = FileTimeToQuadWord(&ftKernelTimeEnd) - FileTimeToQuadWord(&ftKernelTimeStart); __int64 qwUserTimeElapsed = FileTimeToQuadWord(&ftUserTimeEnd) - FileTimeToQuadWord(&ftUserTimeStart); // Get total time duration by adding the kernel and user times. // the default is 100ns, so we convert it to ms //return (qwKernelTimeElapsed + qwUserTimeElapsed) / 10000; return (qwKernelTimeElapsed + qwUserTimeElapsed) / 10; } inline __int64 CThreadTime::FileTimeToQuadWord(PFILETIME pft) { return (Int64ShllMod32(pft->dwHighDateTime, 32) | pft->dwLowDateTime); } class CStopwatch { public: CStopwatch() { QueryPerformanceFrequency(&m_liPerfFreq); Start(); } void Start() { QueryPerformanceCounter(&m_liPerfStart); } __int64 Now() const { //Returns # of milliseconds since //Start was called LARGE_INTEGER liPerfNow; QueryPerformanceCounter(&liPerfNow); return (((liPerfNow.QuadPart - m_liPerfStart.QuadPart) * 1000)/ m_liPerfFreq.QuadPart); } private: //Counts per second LARGE_INTEGER m_liPerfFreq; //Starting count LARGE_INTEGER m_liPerfStart; }; // 2 //CThreadTime ElapsedTime; //ElapsedTime.BeginGetElapsedTime(); // 4 //CStopwatch spWatch; // do ................ //2 //int iElapse = ElapsedTime.EndGetElapsedTime(); //4 //__int64 iElapse = spWatch.Now(); class WriteLogToFile { public: WriteLogToFile() : outfile("d:\\log\\log.txt", std::ofstream::binary | std::ofstream::app) { } virtual void Handle(Item* entry) { outfile.write (entry->GetItemPointer(), entry->GetDataLen()); outfile.flush(); entry->SetDataSign(false); //printf("entry size %d\n",entry->Size()); } ~WriteLogToFile() { outfile.close(); } private: std::ofstream outfile; }; class WriteLogToNet { public: WriteLogToNet() { if (InitSource()) { printf("Init source success....................\n"); } else { printf("Init source failed#########################\n"); } } virtual void Handle(Item* entry) { int ilen = send(nSocket, entry->GetItemPointer(), entry->GetDataLen(), 0); //printf("Sned len = %d, \n",ilen); entry->SetDataSign(false); } bool InitSource() { WSADATA wSaData; WORD dwSockVersion = MAKEWORD(2,2); if (0 != WSAStartup(dwSockVersion,&wSaData)) //协商版本号 { printf("Arrange Version Failure"); return false; } nSocket = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); //创建TCP socket if (INVALID_SOCKET == nSocket) { printf("invalid socket"); WSACleanup(); return false; } sockaddr_in sa; sa.sin_family = AF_INET; sa.sin_addr.s_addr = inet_addr("127.0.0.1"); sa.sin_port = htons(PORT_NUMBER); if ( 0 != connect( nSocket,( const SOCKADDR * )&sa, sizeof(sa) ) ) return false; return true; /* char buf[MAX_BUF_SIZE] = {0}; char tmp[MAX_BUF_SIZE] = {0}; strcpy(tmp,"this is Client!"); int nRecv = 0; nRecv = recv(nSocket, buf, MAX_BUF_SIZE, 0); if (nRecv > 0) { printf("%s/n",buf); send(nSocket, tmp, (int)strlen(tmp), 0); }*/ } ~WriteLogToNet() { closesocket(nSocket); WSACleanup(); } private: SOCKET nSocket; }; NonLockingBuffer g_nonlockingbuf; AtomicReallocationBuffer g_atoRealloctionBug; bool g_read_sign = true; int g_icount = 0; int MakeData(char *pch) { int iLen = 1024; memset(pch, 'w', iLen); return iLen; } unsigned __stdcall read_thread_fun(void*) { WriteLogToFile dataconsumer; printf("read thread func will do work \n"); while (g_read_sign) { std::list< std::map<unsigned int, Group*>* > ltGpPointers = g_nonlockingbuf.GetAllGroups(); std::list< std::map<unsigned int, Group*>* >::iterator it = ltGpPointers.begin(); for (; it!= ltGpPointers.end(); it++) { if (0 != (*it)) { std::map<unsigned int, Group*>::iterator iit = (*it)->begin(); for (; iit!=(*it)->end(); iit++) { //printf("Thr group id = %d \n", (*iit).second->GetGroupID()); Item *pit = iit->second->GetReadItem(); if (pit) { dataconsumer.Handle(pit); } else { //printf("item read is null....\n"); } } } } ::Sleep(2); } return 0; } unsigned __stdcall read_thread_realloc_fun(void*) { WriteLogToFile dataconsumer; printf("read thread func will do work \n"); while (g_read_sign) { unsigned int iReadCurID = 0; unsigned int iMaxUsingID = g_atoRealloctionBug.GetMaxUnsingGroupID(); std::map<unsigned int, Group*>* ltGpPointers = g_atoRealloctionBug.GetAllGroups(); if (NULL == ltGpPointers) { continue; } g_atoRealloctionBug.SetDataProcessSign(); std::map<unsigned int, Group*>::iterator iit = ltGpPointers->begin(); for (; iit!=ltGpPointers->end(); iit++) { //printf("Thr group id = %d \n", (*iit).second->GetGroupID()); Item *pit = iit->second->GetReadItem(); if (pit) { dataconsumer.Handle(pit); } else { //printf("item read is null....\n"); } iReadCurID++; if (iReadCurID > iMaxUsingID) { break; } } g_atoRealloctionBug.SetDataProcessSign(false); ::Sleep(2); } return 0; } unsigned __stdcall write_thread_fun(void*) { unsigned int iThreadid = ::GetCurrentThreadId(); printf("init thread %d finished will do work......\n", iThreadid); int im = 0; int ik = 0; for (int i=0; i<THREAD_RUN_LOOP_TIMES; i++) { im++; for (int j=0; j<GROUP_ITEM_COUNT; j++) { CThreadTime ElapsedTime; ElapsedTime.BeginGetElapsedTime(); Item *pIm = g_nonlockingbuf.GetWriteItem(iThreadid); if (pIm) { int iRetLen = 0; char *pch = pIm->GetItemPointer();