
平台论坛博客文库

› 论坛 › 程序设计 › Linux环境编程 › epoll+事务处理线程池纠结的问题，服务端会挂掉！

[网络] epoll+事务处理线程池纠结的问题，服务端会挂掉！ [复制链接]

guojinshuai

白手起家

论坛徽章:: 0

电梯直达

1楼 [收藏(0)] [报告]

发表于 2015-04-07 10:32 |只看该作者 |倒序浏览

本帖最后由 guojinshuai 于 2015-04-07 10:34 编辑

服务端代码如下：

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <sys/select.h>
#include <pthread.h>
#include <iostream>
#include <sys/epoll.h>
#include <list>
#include <signal.h>
using namespace std;
pthread_mutex_t lock=PTHREAD_MUTEX_INITIALIZER;
enum falgrw{readflag = 0,writeflag};
typedef struct{
int fd;
int op;
}Task;
void *func( void *arg );
struct epoll_event ev,events[200];
int epfd=0;
list<Task> tasklist(0);
int main()
{
int guo=0;
pthread_t thread[3];
for(guo=0;guo<3;guo++)
{
pthread_create(&thread[guo],NULL,func,NULL);
}
int listenfd = 0;
int connfd = 0;
int serverPort= 8888;
int listenq = 5;
socklen_t socklen = 0;
int opt = 1;
int nready = 0;
int maxfd = 0;
int maxi = -1;
char buf[50]= {0};
struct sockaddr_in cliaddr, servaddr;
socklen = sizeof(struct sockaddr_in);
memset(&servaddr, 0, sizeof(servaddr));
memset(&cliaddr, 0, sizeof(cliaddr));
servaddr.sin_family = AF_INET;
servaddr.sin_addr.s_addr = htonl(INADDR_ANY);
servaddr.sin_port = htons(serverPort);
listenfd = socket(AF_INET, SOCK_STREAM, 0);
if (listenfd < 0)
{
perror("socket error");
return -1;
}
if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, (const char*)&opt, sizeof(opt)) < 0)
{
perror("setsockopt error");
}
if (bind(listenfd, (struct sockaddr *) &servaddr, socklen) < 0)
{
perror("bind error");
return -1;
}
if (listen(listenfd, listenq) < 0)
{
perror("listen error");
return -1;
}
//epoll LT
epfd=epoll_create(256);
ev.data.fd=listenfd;
//设置要处理的事件类型
ev.events=EPOLLIN;
//注册epoll事件
epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);
int nfds=0;
printf("echo server startup,listen on port:%d\n", serverPort);
while(1)
{
struct timeval tv;
tv.tv_sec = 1;
tv.tv_usec = 0;
nfds=epoll_wait(epfd,events,200,50);
if(nfds==0)
{
printf("0 00 0 0 0 0 0 0\n");
}
else if(nfds<0)
{
printf("<<<<<<<0 00 0 0 0 0 0 0\n");
}
for(int i=0;i<nfds;++i)
{
if(events[i].data.fd==listenfd)
{
connfd = accept(listenfd,(sockaddr *)&cliaddr, &socklen);
if(connfd<0){
perror("connfd<0");
exit(1);
}
//设置用于读操作的文件描述符
ev.data.fd=connfd;
//设置用于注测的读操作事件
ev.events=EPOLLIN|EPOLLONESHOT;
//注册ev
epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);
}
else if(events[i].events&EPOLLIN)
{
Task task={0,0};
task.fd=events[i].data.fd;
task.op= readflag;
pthread_mutex_lock(&lock);
tasklist.push_front(task);
pthread_mutex_unlock(&lock);
}
else if(events[i].events&EPOLLOUT)
{
//
}
}
}
close(listenfd);
return 0;
}
void *func( void *arg )
{
char buffer[1024]={0};
size_t size = 0;
pthread_detach(pthread_self());
Task task={0,0};
while(1)
{
pthread_mutex_lock(&lock);
size = tasklist.size();
pthread_mutex_unlock(&lock);
memset(buffer,0,1024);
memset(&task,0,sizeof(Task));
printf("size:%d\n",size);
if(size>0)
{
pthread_mutex_lock(&lock);
task = tasklist.front();
tasklist.pop_front();
pthread_mutex_unlock(&lock);
int xxx=read(task.fd,buffer,sizeof(buffer));
if(xxx==0)
{
printf("client %d close!\n",task.fd);
close(task.fd);
epoll_ctl(epfd,EPOLL_CTL_DEL,task.fd,&ev);
}
else if(xxx>0)
{
ev.data.fd=task.fd;
//设置用于注测的读操作事件
ev.events=EPOLLIN|EPOLLONESHOT;
//注册ev
epoll_ctl(epfd,EPOLL_CTL_MOD,task.fd,&ev);
}
else
{
printf("wrong!!\n");
close(task.fd);
epoll_ctl(epfd,EPOLL_CTL_DEL,task.fd,&ev);
}
}
else
{
sleep(1);
}
}
return 0;
}

复制代码

客户端代码如下：

#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <stdlib.h>
#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <fcntl.h>
#include <sys/select.h>
#include <pthread.h>
#include <iostream>
#include <list>
using namespace std;
int main()
{
int cfd;
int recbytes;
int sin_size;
char buffer[1024]={0};
struct sockaddr_in s_add,c_add;
unsigned short portnum=8888;
printf("Hello,welcome to client !\r\n");
cfd = socket(AF_INET, SOCK_STREAM, 0);
if(-1 == cfd)
{
printf("socket fail ! \r\n");
return -1;
}
printf("socket ok !\r\n");
bzero(&s_add,sizeof(struct sockaddr_in));
s_add.sin_family=AF_INET;
s_add.sin_addr.s_addr= inet_addr("127.0.0.1");
s_add.sin_port=htons(portnum);
if(-1 == connect(cfd,(struct sockaddr *)(&s_add), sizeof(struct sockaddr)))
{
printf("connect fail !\r\n");
return -1;
}
printf("connect ok !\r\n");
int j=0;
struct timeval tv;
while(1)
{
memset(buffer,0,sizeof(buffer));
sprintf(buffer,"%d%d",j,j);
if(-1 == (recbytes = write(cfd,buffer,strlen(buffer)+1)))
{
printf("write data fail !\r\n");
return -1;
}
printf("write ok.buffer=[%s]\n",buffer);
sleep(1);
j++;
//test select
tv.tv_sec = 0;
tv.tv_usec = 1000;
int ret = select(0, NULL, NULL, NULL, &tv);
if (-1 == ret)
{
printf("wweeeeeee\n");
}
}
close(cfd);
return 0;
}

复制代码

编译完客户端后，用一个脚本，nohup ./xxx & ,模拟700个客户端同时执行，运行几分钟后服务端就挂了！！！
也看了core文件，输出如下：
(gdb) bt
#0  0x00002ac36656af45 in raise () from /lib64/libc.so.6
#1  0x00002ac36656c340 in abort () from /lib64/libc.so.6
#2  0x00002ac3665a178b in __libc_message () from /lib64/libc.so.6
#3  0x00002ac3665a677e in malloc_printerr () from /lib64/libc.so.6
#4  0x00002ac3665a7ecc in free () from /lib64/libc.so.6
#5  0x0000000000401ca7 in __gnu_cxx::new_allocator<std::_List_node<Task> >::deallocate(std::_List_node<Task>*, unsigned long) ()
#6  0x0000000000401ccf in std::_List_base<Task, std::allocator<Task> >::_M_put_node(std::_List_node<Task>*) ()
#7  0x0000000000401e7b in std::list<Task, std::allocator<Task> >::_M_erase(std::_List_iterator<Task>) ()
#8  0x0000000000401ea3 in std::list<Task, std::allocator<Task> >::pop_front() ()
#9  0x0000000000401284 in func(void*) ()
#10 0x00002ac365fc7193 in start_thread () from /lib64/libpthread.so.0
#11 0x00002ac3665fc0dd in clone () from /lib64/libc.so.6

我看了半天代码，也修改测试，没看明白线程函数有什么问题。。。请大侠们帮忙看看是啥原因。
PS：看到网上有个方法，如果 MALLOC_CHECK_=0 ./xxx 这样运行的话，服务端就不会挂了.

服务端, 客户端