/*
*
*EPOLL ET 触发必须使用非阻塞，LT触发可以阻塞/非阻塞。
*read 函数 非阻塞读需 忙轮寻 soket关闭返回0，循环读完数据
*如果已经读完再读read返回 -1，errno=11(EAGIAN)则退出轮循
*
**/

#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#define WKEY 100
#define RKEY 101
//#define DEBUG

typedef unsigned int uint32;

pthread_mutex_t lock;
uint32 qcount=;

typedef struct{
int quenu;
sem_t sem;
}QUE_SEM;

QUE_SEM rque ,wque;

typedef struct{
int flag;
sem_t sem;
}MYLOCK;

typedef struct{
long type;
char buff[];
}REQUEST;

typedef struct{
long type;
int wrfd;
}RESPON;

struct sockaddr_in server;
MYLOCK mlock;
int g_epfd=-;

int rnum=;
int wnum=;

void printerror()
{
printf("fun %s,%d num",__func__,__LINE__);
printf("rnum =%d ,wnum =%d\n",rnum,wnum);
printf("%d--->%s\n",errno,strerror(errno));
msgctl(rque.quenu,IPC_RMID,);
msgctl(wque.quenu,IPC_RMID,);
exit(-);
}

void initREQUESTue(int rkey,int wkey){
int rq= msgget(rkey,IPC_CREAT|);
int wq= msgget(wkey,IPC_CREAT|);
sem_init(&rque.sem,,);
sem_init(&wque.sem,,);
rque.quenu=rq,wque.quenu=wq;

printf("REQUEST r is %d ,w is %d\n",rq,wq);
}

int ReadSocket(int fd){
REQUEST msg;
int rd=-;
memset(msg.buff,,sizeof (msg.buff));
msg.type=fd;
rd=read(fd,msg.buff,sizeof(msg.buff)-);
// printf("rd size is %d\n",rd);
if(rd==){
struct sockaddr_in client;
memset(&client,,sizeof(client));
int len=sizeof(client);
getpeername(fd,(struct sockaddr*)&client,&len);
// printf("client %s is closed fd is %d , bye!\n",inet_ntoa(client.sin_addr),fd);
close(fd);
return ;
}else if(rd>){
int res=-;
while(){
sem_wait(&rque.sem);
res =msgsnd(rque.quenu,&msg,sizeof(REQUEST)-sizeof(long),IPC_NOWAIT);
printf("read clietn and send to queue res=%d\n",res);
if(res==-){
printf("errno =%d msg=%s\n",errno,strerror(errno));
if(errno==ENOMEM){
sem_post(&rque.sem);
usleep();
continue;
}
sem_post(&rque.sem);
return ;
}else if(res==){
// printf("rnum=%d\n",rnum++);
sem_post(&rque.sem);
return ;
}
}

    }else if(rd==-){  
        close(fd);  
        return ;  
    }  

}

//int getMsgbyfd(int fd,REQUEST* msg){
// if(msgrcv(rque.quenu,msg,sizeof(REQUEST)-sizeof(long),fd,0)>0){
// return 1;
// }
// return 0;
//}

void* th_procce(void* p){
while(){
sem_wait(&wque.sem);
RESPON msg;
#ifdef DEBUG
printf("proc thread ready to feth msg from wque\n");
#endif
int res=msgrcv(wque.quenu,&msg,sizeof(RESPON)-sizeof(long),,IPC_NOWAIT);
usleep();
#ifdef DEBUG
printf("proc feth msg from wque type=%d\n",res);
#endif
sem_post(&wque.sem);
usleep();
if(res>){
sem_wait(&rque.sem);
REQUEST req;
int res2=msgrcv(rque.quenu,&req,sizeof(REQUEST)-sizeof(long),msg.wrfd,IPC_NOWAIT);
#ifdef DEBUG
printf("proc feth msg from rque and send to client res=%d\n",res2);
#endif
if(res2>){
sem_post(&rque.sem);
char buff[]={};
sprintf(buff,"%s -> %s","Server Snd To",req.buff);
write(msg.wrfd,buff,strlen(buff));
}else if(res2==-&& errno==ENOMSG){
sem_post(&rque.sem);
continue;
}
usleep();
#ifdef DEBUG
printf("yet send to client\n");
#endif
}else if(res==-&& errno==ENOMSG){
continue;
}
}
}

int sndtowque(int fd){
RESPON msg;
msg.type=fd;
msg.wrfd=fd;
while () {
sem_wait(&wque.sem);
int res=msgsnd(wque.quenu,&msg,sizeof(RESPON)-sizeof(long),IPC_NOWAIT);
// #ifdef DEBUG
printf("write event com and sendto wque res=%d\n",res);
// #endif
if(res==-){
if(errno==ENOMEM){
sem_post(&wque.sem);
usleep();
continue;
}
sem_post(&wque.sem);
return ;
}else if(res==){
printf("wnum-----=%d\n",wnum++);
sem_post(&wque.sem);
return ;
}
}
}

void* th_hand(void* p){
sem_wait(&mlock.sem);
while(!mlock.flag){
sem_post(&mlock.sem);
sleep();
sem_wait(&mlock.sem);
}
sem_post(&mlock.sem);
struct epoll_event events[];
struct epoll_event event;
while(){
memset(&event,,sizeof(struct epoll_event));
memset(events,,sizeof(struct epoll_event)*);
int s = epoll_wait(g_epfd,events,,);
if(s==-){
printerror();
}else if(s==){
continue;
}else if(s>){
for(int i=;i<s;i++){
int fd=events[i].data.fd;
if(events[i].events & EPOLLIN){
if(ReadSocket(fd)){
events[i].events=EPOLLOUT|EPOLLET;
epoll_ctl(g_epfd,EPOLL_CTL_MOD,fd,&events[i]);
}else{
epoll_ctl(g_epfd,EPOLL_CTL_DEL,fd,);
continue;
}
}
if(events[i].events & EPOLLOUT){
if(sndtowque(fd)){
events[i].events=EPOLLIN|EPOLLET;
epoll_ctl(g_epfd,EPOLL_CTL_MOD,events[i].data.fd,&events[i]);
}else{
epoll_ctl(g_epfd,EPOLL_CTL_DEL,fd,);
continue;
}
}
}
}
}
}

int initSocket(int port){
memset(&server,,sizeof(server));
server.sin_family=AF_INET;
server.sin_addr.s_addr=htonl(INADDR_ANY);//INADDR_ANY==0
server.sin_port=htons(port);
int sockfd=socket(AF_INET,SOCK_STREAM,);
int flag=;
setsockopt(sockfd,SOL_SOCKET,SO_REUSEADDR,&flag,sizeof(flag));
if(sockfd==-){
printerror();
}
int res=bind(sockfd,(struct sockaddr*)&server,sizeof(struct sockaddr));
if(res==-){
printerror();
}
if(-==listen(sockfd,)){
printerror();
}

g\_epfd=epoll\_create();  
if(g\_epfd==-){  
    printerror();  
}

sem\_wait(&mlock.sem);  
    mlock.flag=;  
sem\_post(&mlock.sem);

printf("main before accept\\n");  
while(){  
    int fd;  
    if((fd=accept(sockfd,NULL,NULL))==-){  
        printerror();  
    }  
    printf("fd %d is connect\\n",fd);  
    struct epoll\_event event;  

// event.events=EPOLLIN;
event.events=EPOLLIN|EPOLLET;
event.data.fd=fd;

    fcntl(fd,F\_SETFL,fcntl(fd,F\_GETFL,)|O\_NONBLOCK);

    sem\_wait(&mlock.sem);  
    if(-==(epoll\_ctl(g\_epfd,EPOLL\_CTL\_ADD,fd,&event))){  
        printerror();  
    }  
    qcount++;  
    sem\_post(&mlock.sem);  
}  

}

void sig_hand(int signo){
if(signo==SIGINT){
msgctl(rque.quenu,IPC_RMID,);
msgctl(wque.quenu,IPC_RMID,);
printf("have %d client\n",qcount);
exit();
}
}

pthread_t pid;
pthread_t process;
int main(int argc,char** argv){
if(argc<){
puts("please input port\n");
exit(-);
}
initREQUESTue(RKEY,WKEY);
int port=atoi(argv[]);
signal(SIGINT,sig_hand);
memset(&mlock,,sizeof(mlock));
sem_init(&mlock.sem,,);
mlock.flag=;

pthread\_create(&pid,NULL,th\_hand,(void\*));  
pthread\_detach(pid);

pthread\_create(&process,NULL,th\_procce,(void\*));  
pthread\_detach(process);  
initSocket(port);

}