在Windows socket I/O模型 之 select（1）中。我们仅仅是在console中简单的模拟了select的处理方法。
还有非常多特性不能改动。比方仅仅能写，不能读。
没使用线程。也没有同步不同的读写线程。
先谈谈个人眼下对select的理解。
select就是监控一组套接字的变化情况。
比方一个fd_set的变量（暂且定义为fdRead）里面有5个套接字。当你传给select后，如果仅仅有2个被触发。
那么这个fdRead也就改变了。这是为什么select须要从全局fd_set拷贝一份给select的原因。
然后我们依据套接字的变化情况做对应的处理就OK了。可是大并发量还没有測试。
今晚我改进了。以下大家见代码。
。。。
。
。
// Select_Server.cpp : Defines the entry point for the console application.
// 服务端 #include "stdafx.h" #define STR_SERVER_IP "127.0.0.1"
#define INT_SERVER_PORT 8001
#define INT_DATABUFFER_SIZE 256 SOCKET g_soClient;
typedef std::list<SOCKET> LstSocket;
LstSocket g_lstSoClient; SOCKET g_soServer;
fd_set g_fdSocketSet; // for thread synchronize
CCriSec g_criSec; DWORD WINAPI ThreadRead(LPVOID lpvParam)
{
int iResult = 0;
sockaddr_in addrAccept;
int iAcceptLen = sizeof(addrAccept);
SOCKET soClient; FD_ZERO(&g_fdSocketSet);
FD_SET(g_soServer, &g_fdSocketSet);
fd_set fdRead, fdWrite;
while( TRUE ) {
// initialize
FD_ZERO(&fdRead);
FD_ZERO(&fdWrite);
fdRead = g_fdSocketSet;
fdWrite = g_fdSocketSet; int n1 = fdRead.fd_count;
int n2 = fdWrite.fd_count;
int n3 = g_fdSocketSet.fd_count;
int iResult = select(0, &fdRead, &fdWrite, NULL, NULL);
if( iResult == SOCKET_ERROR) {
break;
} if(FD_ISSET(g_soServer, &fdRead)) {
soClient = accept(g_soServer, (sockaddr*)&addrAccept, &iAcceptLen); CCriSecLock lock(g_criSec);
if(soClient == INVALID_SOCKET) {
continue;
} else {
printf("\n[%s:%d] has connected to server!\r\n", inet_ntoa(addrAccept.sin_addr),
ntohs(addrAccept.sin_port));
FD_SET(soClient, &g_fdSocketSet);
}
} else {
// check read
for(int i=0; i < (int)fdRead.fd_count; i++) {
if ( fdRead.fd_array[i] == g_soServer ) {
continue;
} if( FD_ISSET(fdRead.fd_array[i], &g_fdSocketSet) ) {
sockaddr_in name;
int namelen = sizeof(sockaddr_in);
getpeername(fdRead.fd_array[i], (sockaddr *)&name, &namelen); char buf[256] = {0};
int len = 256;
int ret = recv(fdRead.fd_array[i], buf, len, 0);
CCriSecLock lock(g_criSec);
if( ret == SOCKET_ERROR ) {
int nErr = GetLastError();
if( nErr == 10054 ) {
// Connection reset by peer.
FD_CLR(fdRead.fd_array[i], &g_fdSocketSet);
printf("\n[%s:%d] disconnect from server.\n", inet_ntoa(name.sin_addr), ntohs(name.sin_port) );
} else {
printf("\nfdread failed with %d\n", nErr);
}
} else {
printf("\nRecv from [%s:%d] : %s\n", inet_ntoa(name.sin_addr), ntohs(name.sin_port), buf);
}
}
} // check write
static bool b11 = false;
for(int i=0; i < (int)fdWrite.fd_count; i++) {
if( FD_ISSET(fdWrite.fd_array[i], &g_fdSocketSet) ) {
char buf[256] = "abcd";
int len = 256;
if( !b11 ) {
b11 = true;
//send(fdWrite.fd_array[i], buf, len ,0);
}
}
}
}
} return 0;
} DWORD WINAPI ThreadWrite(LPVOID lpvParam)
{
std::string str;
{
CCriSecLock lock(g_criSec);
std::cout << "Please input message to client: ";
}
while( getline(std::cin, str) ) {
if( str.compare("exit") == 0 ) {
{
CCriSecLock lock(g_criSec);
printf("close write thread\n");
}
break;
} for(int i = 1; i < (int)g_fdSocketSet.fd_count; i++) {
send(g_fdSocketSet.fd_array[i], str.data(), (int)str.size(), 0);
}
} return 0;
} int _tmain(int argc, _TCHAR* argv[])
{
WORD dwVersion = MAKEWORD(2, 2);
WSAData wsaData;
WSAStartup(WINSOCK_VERSION,&wsaData); g_soServer = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (INVALID_SOCKET == g_soServer) {
printf("Failed to create socket!\r\n");
WSACleanup();
return -1;
} sockaddr_in addrServer;
memset(&addrServer,0,sizeof(sockaddr_in));
addrServer.sin_family = AF_INET;
addrServer.sin_port = htons(INT_SERVER_PORT);
addrServer.sin_addr.S_un.S_addr = htonl(INADDR_ANY); int iResult; bool bReuseAddr = true;
iResult = setsockopt(g_soServer, SOL_SOCKET, SO_REUSEADDR, (char *)&bReuseAddr, sizeof(bReuseAddr));
if(SOCKET_ERROR == iResult) {
printf("Failed to set resueaddr socket!\r\n");
WSACleanup();
return -1;
} //设置非堵塞方式连接
unsigned long cmd = 1;
iResult = ioctlsocket(g_soServer, FIONBIO, &cmd); iResult = bind(g_soServer, (sockaddr *)&addrServer, sizeof(addrServer));
if (SOCKET_ERROR == iResult) {
printf("Failed to bind address!\r\n");
WSACleanup();
return -1;
} if (0 != listen(g_soServer, 5)) {
printf("Failed to listen client!\r\n");
WSACleanup();
return -1;
} printf("Start server...\r\n"); HANDLE hWorkRead = CreateThread(NULL, 0, ThreadRead, NULL, 0, NULL);
HANDLE hWorkWrite = CreateThread(NULL, 0, ThreadWrite, NULL, 0, NULL); ::WaitForSingleObject(hWorkRead, INFINITE);
::WaitForSingleObject(hWorkWrite, INFINITE); WSACleanup(); return 0;
}
以下是client代码：
// Select_Client.cpp : Defines the entry point for the console application.
// #include "stdafx.h" #define INT_SERVER_PORT 8001
#define STR_SERVER_IP "127.0.0.1"
#define INT_DATABUFFER_SIZE 256
#define STR_EXIT "exit"
#define STR_RECV "recv" //
SOCKET g_soClient;
fd_set g_fdSocketSet; // for thread synchronize
CCriSec g_criSec; DWORD WINAPI ThreadWorker(LPVOID lpvParam)
{
FD_ZERO(&g_fdSocketSet);
FD_SET(g_soClient, &g_fdSocketSet);
fd_set fdRead, fdWrite; while( TRUE ) {
// initialize
FD_ZERO(&fdRead);
FD_ZERO(&fdWrite);
fdRead = g_fdSocketSet;
fdWrite = g_fdSocketSet; int iResult = select(0, &fdRead, &fdWrite, NULL, NULL);
if( iResult == SOCKET_ERROR) {
break;
} else if( iResult == 0 ) {
printf("Time limit expired\n");
} else {
// check read
if (FD_ISSET(fdRead.fd_array[0], &g_fdSocketSet)) {
sockaddr_in name;
int namelen = sizeof(sockaddr_in);
getpeername(fdRead.fd_array[0], (sockaddr *)&name, &namelen); char buf[256] = {0};
int len = 256;
int ret = recv(fdRead.fd_array[0], buf, len, 0);
CCriSecLock lock(g_criSec);
if( ret == SOCKET_ERROR ) {
int nErr = GetLastError();
if( nErr == 10054 ) {
// Connection reset by peer.
FD_CLR(fdRead.fd_array[0], &g_fdSocketSet);
printf( "\n[%s:%d] is closed.\n", inet_ntoa(name.sin_addr), ntohs(name.sin_port) );
} else {
printf("fdread failed with %d\n", nErr);
}
} else {
CCriSecLock lock(g_criSec);
printf("\nRecv from [%s:%d] : %s\n", inet_ntoa(name.sin_addr), ntohs(name.sin_port), buf);
}
} // check write
if (FD_ISSET(fdWrite.fd_array[0], &g_fdSocketSet)) {
int a=2;
int b=a;
}
}
} return 0;
} void main(void)
{
WSAData wsaData;
WSAStartup(WINSOCK_VERSION,&wsaData); g_soClient = socket(AF_INET,SOCK_STREAM, IPPROTO_TCP);
if (INVALID_SOCKET == g_soClient) {
printf("Failed to create client!\r\n");
WSACleanup();
} sockaddr_in addrServer;
addrServer.sin_addr.S_un.S_addr = inet_addr(STR_SERVER_IP);
addrServer.sin_family = AF_INET;
addrServer.sin_port = htons(INT_SERVER_PORT); int iResult; //设置非堵塞方式连接
//unsigned long ul = 1;
//iResult = ioctlsocket(g_soClient, FIONBIO, (unsigned long*)&ul); iResult = connect(g_soClient, (sockaddr *)&addrServer, sizeof(sockaddr_in));
if (SOCKET_ERROR == iResult) {
printf("Failed to connect server!(Error: %d)\r\n", ::WSAGetLastError());
WSACleanup();
return;
} HANDLE hWorker = CreateThread(NULL, 0, ThreadWorker, NULL, 0, NULL); std::string str;
std::cout << "Please input message to server: ";
while( getline(std::cin, str) ) {
send(g_soClient, str.data(), str.size(), 0);
std::cout << "Please input message to client: ";
} closesocket(g_soClient);
WSACleanup();
}
头文件
// stdafx.h : include file for standard system include files,
// or project specific include files that are used frequently, but
// are changed infrequently
// #pragma once #ifndef _WIN32_WINNT // Allow use of features specific to Windows XP or later.
#define _WIN32_WINNT 0x0501 // Change this to the appropriate value to target other versions of Windows.
#endif #include <stdio.h>
#include <tchar.h>
#include <string>
#include <iostream>
#include <WINSOCK2.H> #pragma comment(lib,"ws2_32.lib") // TODO: reference additional headers your program requires here
#include "CriticalSection.h"
// CriticalSection.h
/*
* Copyright: JessMA Open Source (ldcsaa@gmail.com)
*
* Version : 2.3.2
* Author : Bruce Liang
* Website : http://www.jessma.org
* Project : https://github.com/ldcsaa
* Blog : http://www.cnblogs.com/ldcsaa
* Wiki : http://www.oschina.net/p/hp-socket
* QQ Group : 75375912
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/ /****************************************************************************
* *
* CriticalSection.h *
* *
* Create by : *
* Kingfisher 2003-10-15 *
* *
* Description: *
* 封装Win32临界量对象和相互排斥量内核对象 *
****************************************************************************/ #pragma once #include <windows.h> class CCriSec
{
public:
CCriSec() {::InitializeCriticalSection(&m_crisec);}
~CCriSec() {::DeleteCriticalSection(&m_crisec);} void Lock() {::EnterCriticalSection(&m_crisec);}
void Unlock() {::LeaveCriticalSection(&m_crisec);} private:
CCriSec(const CCriSec& cs);
CCriSec operator = (const CCriSec& cs); private:
CRITICAL_SECTION m_crisec;
}; class CCriSec2
{
public:
CCriSec2(BOOL bInitialize = TRUE)
{
if(bInitialize)
{
m_pcrisec = new CRITICAL_SECTION;
::InitializeCriticalSection(m_pcrisec);
}
else
m_pcrisec = NULL;
} ~CCriSec2() {Reset();} void Attach(CRITICAL_SECTION* pcrisec)
{
Reset();
m_pcrisec = pcrisec;
} CRITICAL_SECTION* Detach()
{
CRITICAL_SECTION* pcrisec = m_pcrisec;
m_pcrisec = NULL;
return pcrisec;
} void Lock() {::EnterCriticalSection(m_pcrisec);}
void Unlock() {::LeaveCriticalSection(m_pcrisec);} private:
CCriSec2(const CCriSec2& cs);
CCriSec2 operator = (const CCriSec2& cs); void Reset()
{
if(m_pcrisec)
{
::DeleteCriticalSection(m_pcrisec);
delete m_pcrisec;
m_pcrisec = NULL;
}
} private:
CRITICAL_SECTION* m_pcrisec;
}; template<class CLockObj> class CLocalLock
{
public:
CLocalLock(CLockObj& obj) : m_lock(obj) {m_lock.Lock();}
~CLocalLock() {m_lock.Unlock();}
private:
CLockObj& m_lock;
}; typedef CLocalLock<CCriSec> CCriSecLock;
typedef CLocalLock<CCriSec2> CCriSecLock2;