学步园

#define BUFSIZE 1024 //max size of incoming data buffer 
#define MAXADDRSTR 16 

#define DEFAULT_GROUP_ADDRESS "239.254.1.2" 
#define DEFAULT_PORT 7125  

LONG nCount = 0; 
HANDLE g_hCompletionPort; 

DWORD WINAPI WorkerThread( LPVOID WorkContext ); 

BOOL HandleIncomingData( UCHAR* pBuf); 
BOOL CreateNetConnections( VOID ); 
BOOL CreateWorkers( UINT ); 
void InitWinsock2(); 
void UnInitWinsock2(); 

HANDLE g_hReadEvent; 
SOCKET g_hSocket; 
UCHAR achInBuf [BUFSIZE]; 
char achMCAddr[MAXADDRSTR] = DEFAULT_GROUP_ADDRESS; 
u_short nPort = DEFAULT_PORT; 

OVERLAPPED Overlapped; 
 //----------------------------------------------------------------------------- 
void InitWinsock2() 
{  
             WSADATA data;  
             WORD version;  
             int ret = 0;   

             version = (MAKEWORD(2, 2));  
             ret = WSAStartup(version, &data);  
             if (ret != 0)  
             {   
                          ret = WSAGetLastError();  
                          if (ret == WSANOTINITIALISED)  
                          {   
                                       printf("not initialised");  
                          } 
            } 
} 
//----------------------------------------------------------------------------- 
void UnInitWinsock2() 
{  
           WSACleanup(); 
} 

//----------------------------------------------------------------------------- 
BOOL CreateNetConnections (void) 
{  
             DWORD nbytes;  
             BOOL b;  
             BOOL fFlag = TRUE;  
             int nRet=0;  
             SOCKADDR_IN stLclAddr;   
             struct ip_mreq stMreq; // Multicast interface structure   

             // Get a datagram socket   
             g_hSocket = socket(AF_INET, SOCK_DGRAM,0);  
             if (g_hSocket == INVALID_SOCKET)   
             {  
                          printf ("socket() failed, Err: %d ", WSAGetLastError());  
                          return FALSE;   
             }   

             nRet = setsockopt(g_hSocket,SOL_SOCKET,SO_REUSEADDR, (char *)&fFlag, sizeof(fFlag));   
             if (nRet == SOCKET_ERROR)   
             {  
                          printf ("setsockopt() SO_REUSEADDR failed, Err: %d ",WSAGetLastError());  
             }  

             // Name the socket (assign the local port number to receive on)   
             stLclAddr.sin_family = AF_INET;  
             stLclAddr.sin_addr.s_addr = htonl(INADDR_ANY);  
             stLclAddr.sin_port = htons(nPort);  
             nRet = bind(g_hSocket,(struct sockaddr*) &stLclAddr,sizeof(stLclAddr));  
             if (nRet == SOCKET_ERROR)   
             {  
                          printf ("bind() port: %d failed, Err: %d ", nPort,WSAGetLastError());  
             }  

             // Join the multicast group so we can receive from it   
             stMreq.imr_multiaddr.s_addr = inet_addr(achMCAddr);  
             stMreq.imr_interface.s_addr = INADDR_ANY;  
             nRet = setsockopt(g_hSocket,IPPROTO_IP,IP_ADD_MEMBERSHIP,(char *)&stMreq,sizeof             (stMreq));  
             if (nRet == SOCKET_ERROR)   
             {  
             printf("setsockopt() IP_ADD_MEMBERSHIP address %s failed, Err: %d ",achMCAddr,                     WSAGetLastError());  
             }   

             //  
             // note the 10 says how many concurrent cpu bound threads to allow thru  
             // this should be tunable based on the requests. CPU bound requests will  
             // really really honor this.  
             //  

             g_hCompletionPort = CreateIoCompletionPort (INVALID_HANDLE_VALUE,NULL,0,3);  
             if (!g_hCompletionPort)  
             {  
                          fprintf (stdout, "g_hCompletionPort Create Failed ");  
                          return FALSE;  
             }  

             //Associate this socket to this I/O completion port  
             CreateIoCompletionPort ((HANDLE)g_hSocket,g_hCompletionPort,(DWORD)g_hSocket,3);   

             //   
             // Start off an asynchronous read on the socket.   
             //   

             Overlapped.hEvent = g_hReadEvent;   
             Overlapped.Internal = 0;   
             Overlapped.InternalHigh = 0;   
             Overlapped.Offset = 0;   
             Overlapped.OffsetHigh = 0;   

             b = ReadFile ((HANDLE)g_hSocket,&achInBuf,sizeof (achInBuf),&nbytes,&Overlapped);   

             if (!b && GetLastError () != ERROR_IO_PENDING)   
             {   
                          fprintf (stdout, "ReadFile Failed ");   
                          return FALSE;   
             }   

             return TRUE; 
} 

//----------------------------------------------------------------------------- 
BOOL CreateWorkers (UINT dwNumberOfWorkers) 
{  
        DWORD ThreadId;  
        HANDLE ThreadHandle;  
        DWORD i;  

        for (i = 0; i < dwNumberOfWorkers; i++)  
        {  
                ThreadHandle = CreateThread (NULL,0,WorkerThread,NULL,0,&ThreadId);  
                if (!ThreadHandle)  
                {  
                        fprintf (stdout, "Create Worker Thread Failed ");  
                        return FALSE;  
                }  

                CloseHandle (ThreadHandle);  
        }  

        return TRUE; 
} 

//----------------------------------------------------------------------------- 
DWORD WINAPI WorkerThread (LPVOID WorkContext) 
{  

        DWORD nSocket;  
        BOOL b;  
        OVERLAPPED ovl;  
        LPOVERLAPPED lpo=&ovl;  
        DWORD nBytesRead=0;  
        DWORD nBytesToBeRead;  
        UCHAR ReadBuffer[BUFSIZE];  
        LPVOID lpMsgBuf;  

        memset(&ReadBuffer,0,BUFSIZE);  

        for (;;)  
        {  

                b = GetQueuedCompletionStatus (g_hCompletionPort,&nBytesToBeRead,&nSocket,&lpo,INFINITE);  

                if (b || lpo)  
                {  
                        if (b)  
                        {  
                                //  
                                // Determine how long a response was desired by the client.  
                                //  
                                OVERLAPPED ol;  
                                ol.hEvent = g_hReadEvent;  
                                ol.Offset = 0;  
                                ol.OffsetHigh = 0;  

                                b = ReadFile ((HANDLE)nSocket,&ReadBuffer,nBytesToBeRead,&nBytesRead,&ol);  
                                if (!b )   
                                {  
                                               DWORD dwErrCode = GetLastError();  
                                               if( dwErrCode != ERROR_IO_PENDING )  
                                               {  
                                                    // something has gone wrong here...  
                                                    printf("Something has gone wrong:Error code - %d ",dwErrCode );  

                                                    FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER |   
                                                    FORMAT_MESSAGE_FROM_SYSTEM |   
                                                                                                FORMAT_MESSAGE_IGNORE_INSERTS,  
                                                                                                NULL,  
                                                                                                dwErrCode ,  
                                                     MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT), // Default language  
                                                                                                (LPTSTR) &lpMsgBuf,  
                                                                                                0,  
                                                                                                NULL   
                                                                                                           );  

                                                    OutputDebugString((LPCTSTR)lpMsgBuf);  

                                                    // Free the buffer.  
                                                    LocalFree( lpMsgBuf );  
                                }  
                                else  
                                if( dwErrCode == ERROR_IO_PENDING )  
                                { 

                                      // I had to do this for my UDP sample 

                                      //Never did for my TCP servers  
                                      WaitForSingleObject(ol.hEvent,INFINITE);   

                                      HandleIncomingData(ReadBuffer);  
                                 } 
                        }  
                        else  
                        {  
                                                HandleIncomingData(ReadBuffer);  
                        }  

                        continue; 
            } 
           else  
           {  
           fprintf (stdout, "WorkThread Wait Failed ");  
           //exit (1);  
           } 
        } 
       return 1; 
      } 
} 

//----------------------------------------------------------------------------- 
BOOL HandleIncomingData( UCHAR* pBuf) 
{  
           InterlockedIncrement(&nCount);  
           SYSTEMTIME *lpstSysTime;  
           lpstSysTime = (SYSTEMTIME *)(pBuf);  

           printf("[%d]UTC Time %02d:%02d:%02d:%03d on %02d-%02d-%d  ",nCount,  
           lpstSysTime->wHour, lpstSysTime->wMinute,   
           lpstSysTime->wSecond, lpstSysTime->wMilliseconds,  
           lpstSysTime->wMonth, lpstSysTime->wDay, lpstSysTime->wYear);  

           memset(&pBuf,0,BUFSIZE); //just making sure that i am not showing stale data  

           return TRUE; 
} 

//----------------------------------------------------------------------------- 
main () 
{  
           //You can modify your program to take some arguments for port number  
           //and multicast group address here  
           printf(" *************************************** ");  
           printf("Group IP address: %s ",achMCAddr);  
           printf("Port number : %d ",nPort);  
           printf(" *************************************** ");  

           //Initialize winsock 2   
           InitWinsock2();  

           //We want to keep the main thread running  
           HANDLE hWait2Exit = CreateEvent(NULL,FALSE,TRUE,"MCLIENT");  
           ResetEvent(hWait2Exit );  

           //This OVERLAPPED event  
           g_hReadEvent = CreateEvent(NULL,TRUE,TRUE,NULL);   

           //  
           // try to get timing more accurate... Avoid context  
           // switch that could occur when threads are released  
           //  

           SetThreadPriority (GetCurrentThread (), THREAD_PRIORITY_TIME_CRITICAL);  

           if (!CreateNetConnections ())  
           {  
                      printf("Error condition @ CreateNetConnections , exiting ");  
                      return 1;  
           }  

           if (!CreateWorkers (5))  
           {  
                      printf("Error condition @CreateWorkers, exiting ");  
                      return 1;  
           }  

           WaitForSingleObject(hWait2Exit,INFINITE);  

           UnInitWinsock2();  

           return 1; 
}