socket服务端select模式+管道+多线程，select可以替换为epoll
天下发表于 2014-03-24 02:35:00
#include "stdafx.h"
using namespace std;
#define PORT 5006 /* Port that will be opened */
#define BACKLOG 5 /* Number of allowed connections */
#define MAXDATASIZE 4096
typedef map<unsigned int,sockaddr_in> socket_maptable_t;
bool process_client(int sockfd,sockaddr_in addr,char* recvbuf, int len);
socket_maptable_t socket_maptable;
int fds[2];
void* dispatch_thread(void *arg)
{
char buf[MAXDATASIZE];
arg = NULL;
pr_debug("fds[0]:%d,fds[1]:%d",fds[0],fds[1]);
while (1)
{
int readed = read(fds[0],buf,MAXDATASIZE);
pr_debug("readed:%d\r\n%s",readed,buf);
}
}
int main()
{
int sock_fd;
ssize_t recv_size;
fd_set read_set;
fd_set all_set;
int listen_fd;
int connect_fd;
int ready_size;
socklen_t addr_size;
sockaddr_in srv_addr;
sockaddr_in addr;
unsigned int max_fd = 0;
int errcode;
char recv_buf[MAXDATASIZE];
if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
pr_debug("Creating socket failed.");
exit(1);
}
int opt = SO_REUSEADDR;
setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
bzero(&srv_addr,sizeof(srv_addr));
srv_addr.sin_family=AF_INET;
srv_addr.sin_port=htons(PORT);
srv_addr.sin_addr.s_addr = htonl (INADDR_ANY);
if (bind(listen_fd, (struct sockaddr *)&srv_addr, sizeof(struct sockaddr)) == -1)
{
pr_debug("Bind error.");
exit(1);
}
if(listen(listen_fd,BACKLOG) == -1)
{
pr_debug("listen() error\n");
exit(1);
}
//创建管道
errcode = pipe(fds);
if (errcode < 0)
{
pr_debug("pipe() error");
return -1;
}
pr_debug("pipe() succeed,fds[0]:%d,fds[1]:%d",fds[0],fds[1]);
//创建接收连接线程
pthread_t dispatch_tid;
if (pthread_create(&dispatch_tid, NULL, dispatch_thread, (void*)NULL))
{
pr_debug("pthread_create() dispatch_thread error");
return -1;
}
pr_debug("pthread_create() dispatch_thread succeed");
max_fd = listen_fd;
addr_size=sizeof(sockaddr_in);
FD_ZERO(&all_set);
FD_SET(listen_fd, &all_set);
socket_maptable_t::iterator iter;
while(1)
{
read_set = all_set;
ready_size = select(max_fd+1, &read_set, NULL, NULL, NULL);
if ( FD_ISSET(listen_fd,&read_set) ) {
connect_fd = accept(listen_fd,(struct sockaddr *)&addr,&addr_size);
if (connect_fd ==-1 ) {
pr_debug("accept() error\n");
continue;
}
socket_maptable[connect_fd] = addr;
max_fd = connect_fd;
FD_SET(connect_fd, &all_set);
pr_debug("(%u) %s:%d connected,socket:%u",socket_maptable.size(),inet_ntoa(addr.sin_addr),ntohs(addr.sin_port),connect_fd);
}
for ( iter = socket_maptable.begin(); iter!=socket_maptable.end(); iter++)
{
sock_fd = iter->first;
addr = iter->second;
if ( FD_ISSET(sock_fd,&read_set) )
{
recv_size = recv(sock_fd, recv_buf, MAXDATASIZE,0);
bool bRet = (recv_size > 0);
bRet = bRet && process_client(sock_fd,addr,recv_buf,recv_size);
if ( !bRet)
{
close(sock_fd);
FD_CLR(sock_fd, &all_set);
socket_maptable.erase(sock_fd);
pr_debug("%s:%d disconnected\n",inet_ntoa(addr.sin_addr),ntohs(addr.sin_port) );
}
}
}
}
close(listen_fd);
}
bool process_client(int sockfd,sockaddr_in addr,char* recvbuf, int len)
{
int writed = write(fds[1],recvbuf,len);
pr_debug("datalen:%d,writed:%d",len,writed);
return (writed==len);
return true;
}
using namespace std;
#define PORT 5006 /* Port that will be opened */
#define BACKLOG 5 /* Number of allowed connections */
#define MAXDATASIZE 4096
typedef map<unsigned int,sockaddr_in> socket_maptable_t;
bool process_client(int sockfd,sockaddr_in addr,char* recvbuf, int len);
socket_maptable_t socket_maptable;
int fds[2];
void* dispatch_thread(void *arg)
{
char buf[MAXDATASIZE];
arg = NULL;
pr_debug("fds[0]:%d,fds[1]:%d",fds[0],fds[1]);
while (1)
{
int readed = read(fds[0],buf,MAXDATASIZE);
pr_debug("readed:%d\r\n%s",readed,buf);
}
}
int main()
{
int sock_fd;
ssize_t recv_size;
fd_set read_set;
fd_set all_set;
int listen_fd;
int connect_fd;
int ready_size;
socklen_t addr_size;
sockaddr_in srv_addr;
sockaddr_in addr;
unsigned int max_fd = 0;
int errcode;
char recv_buf[MAXDATASIZE];
if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
pr_debug("Creating socket failed.");
exit(1);
}
int opt = SO_REUSEADDR;
setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
bzero(&srv_addr,sizeof(srv_addr));
srv_addr.sin_family=AF_INET;
srv_addr.sin_port=htons(PORT);
srv_addr.sin_addr.s_addr = htonl (INADDR_ANY);
if (bind(listen_fd, (struct sockaddr *)&srv_addr, sizeof(struct sockaddr)) == -1)
{
pr_debug("Bind error.");
exit(1);
}
if(listen(listen_fd,BACKLOG) == -1)
{
pr_debug("listen() error\n");
exit(1);
}
//创建管道
errcode = pipe(fds);
if (errcode < 0)
{
pr_debug("pipe() error");
return -1;
}
pr_debug("pipe() succeed,fds[0]:%d,fds[1]:%d",fds[0],fds[1]);
//创建接收连接线程
pthread_t dispatch_tid;
if (pthread_create(&dispatch_tid, NULL, dispatch_thread, (void*)NULL))
{
pr_debug("pthread_create() dispatch_thread error");
return -1;
}
pr_debug("pthread_create() dispatch_thread succeed");
max_fd = listen_fd;
addr_size=sizeof(sockaddr_in);
FD_ZERO(&all_set);
FD_SET(listen_fd, &all_set);
socket_maptable_t::iterator iter;
while(1)
{
read_set = all_set;
ready_size = select(max_fd+1, &read_set, NULL, NULL, NULL);
if ( FD_ISSET(listen_fd,&read_set) ) {
connect_fd = accept(listen_fd,(struct sockaddr *)&addr,&addr_size);
if (connect_fd ==-1 ) {
pr_debug("accept() error\n");
continue;
}
socket_maptable[connect_fd] = addr;
max_fd = connect_fd;
FD_SET(connect_fd, &all_set);
pr_debug("(%u) %s:%d connected,socket:%u",socket_maptable.size(),inet_ntoa(addr.sin_addr),ntohs(addr.sin_port),connect_fd);
}
for ( iter = socket_maptable.begin(); iter!=socket_maptable.end(); iter++)
{
sock_fd = iter->first;
addr = iter->second;
if ( FD_ISSET(sock_fd,&read_set) )
{
recv_size = recv(sock_fd, recv_buf, MAXDATASIZE,0);
bool bRet = (recv_size > 0);
bRet = bRet && process_client(sock_fd,addr,recv_buf,recv_size);
if ( !bRet)
{
close(sock_fd);
FD_CLR(sock_fd, &all_set);
socket_maptable.erase(sock_fd);
pr_debug("%s:%d disconnected\n",inet_ntoa(addr.sin_addr),ntohs(addr.sin_port) );
}
}
}
}
close(listen_fd);
}
bool process_client(int sockfd,sockaddr_in addr,char* recvbuf, int len)
{
int writed = write(fds[1],recvbuf,len);
pr_debug("datalen:%d,writed:%d",len,writed);
return (writed==len);
return true;
}