1 // 
  2 // a simple agi server using epoll in linux
  3 // 
  4 // 2010-12-20
  5 // by nsy
  6 //
  7 #include <sys/socket.h>
  8 #include <sys/epoll.h>
  9 #include <netinet/in.h>
 10 #include <arpa/inet.h>
 11 #include <fcntl.h>
 12 #include <unistd.h>
 13 #include <stdio.h>
 14 #include <stdlib.h>
 15 #include <errno.h>
 16 #include <string.h>
 17 #include "CallSvr.h"
 18 #include <pthread.h>
 19 #include "epoll.h"
 20 
 21 //test
 22 #include "msg.h"
 23 
 24 // set event
 25 void EventSet(struct myevent_s *ev, int fd,int status)
 26 {
 27   ev->fd = fd;
 28   ev->status = status;
 29   ev->last_active = time(NULL);
 30   fprintf(stderr,"function=%s,line=%d,fd=%d\n",__func__,__LINE__,fd);
 31 }
 32 // add/mod an event to epoll
 33 void EventAdd(int epollFd, int events,struct myevent_s *ev)
 34 {
 35   struct epoll_event epv = {0, {0}};
 36   int op;
 37   epv.data.ptr = ev;
 38   epv.events = events;
 39   if(ev->status == 1){
 40     op = EPOLL_CTL_MOD;
 41     fprintf(stderr,"mod:function=%s,line=%d,fd=%d,status=%d\n",__func__,__LINE__,ev->fd,ev->status);
 42   }
 43   else{
 44     op = EPOLL_CTL_ADD;
 45     ev->status = 1;
 46     fprintf(stderr,"add:function=%s,line=%d,fd=%d,status=%d\n",__func__,__LINE__,ev->fd,ev->status);
 47   }
 48   if(epoll_ctl(epollFd, op, ev->fd, &epv) < 0)
 49     {
 50       fprintf(stderr,"failed:function=%s,line=%d,fd=%d:errno=%d\n",__func__,__LINE__,ev->fd,errno);
 51     }
 52   else
 53     {
 54       fprintf(stderr,"success:function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
 55     }
 56 }
 57 // delete an event from epoll
 58 void EventDel(int epollFd,struct myevent_s *ev)
 59 {
 60   struct epoll_event epv = {0, {0}};
 61   if(ev->status != 1) return;
 62   epv.data.ptr = ev;
 63   ev->status = 0;
 64   epoll_ctl(epollFd, EPOLL_CTL_DEL, ev->fd, &epv);
 65   fprintf(stderr,"function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
 66 }
 67 
 68 // receive data
 69 void RecvData(struct myevent_s *ev)
 70 {
 71   msg_header header;
 72   int recvbytes;
 73   if ((recvbytes=recv(ev->fd, &header, sizeof(msg_header), 0)) ==-1) 
 74     {
 75       fprintf(stderr,"RecvHeaderErr:function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
 76       goto errret;
 77     }
 78   if(recvbytes == sizeof(msg_header))
 79     {
 80       switch(header.msg_type)
 81     {
 82     case msg_lost:
 83       {
 84         rq_lost rq;
 85         if ((recvbytes=recv(ev->fd, ((char*)(&rq))+sizeof(msg_header), sizeof(rq_lost)-sizeof(msg_header), 0)) ==-1)
 86           {
 87         fprintf(stderr,"RecvAfter:function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
 88         goto errret;
 89           }
 90         else if(recvbytes > 0)
 91           {                
 92         printf("recv sucess%d,hope to recv %d\n",recvbytes,sizeof(rq_lost)-sizeof(msg_header));
 93         printf("cardno is '%s'\n",rq.cardno);
 94         printf("password is '%s'\n",rq.password);
 95     
 96         //init ev recv
 97         memcpy(&ev->header,&header,sizeof(msg_header));
 98         memcpy(&ev->recv_buff,&rq,sizeof(rq_lost));
 99         ev->recv_len = recvbytes;
100         fprintf(stderr,"addtologicqueue:function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
101         //add to logic queue
102         sem_wait(&bin_sem_logic_data_produce);
103         struct QUEUE_LOGIC_DATA_ITEM *item;
104         item = malloc(sizeof(struct QUEUE_LOGIC_DATA_ITEM));
105         item->ev = ev;
106         pthread_mutex_lock(&queue_logic_data_mutex);
107         TAILQ_INSERT_TAIL(&queue_logic_data_header,item,logic_data_entries);
108         pthread_mutex_unlock(&queue_logic_data_mutex);
109         sem_post(&bin_sem_logic_data_consume);
110         fprintf(stderr,"addtologicqueue--end:function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
111           }else
112           {
113         fprintf(stderr,"RecvAfter:function=%s,line=%d,fd=%d:errno=%d\n",__func__,__LINE__,ev->fd,errno);
114         goto errret;
115           }
116         break;
117       }
118     }
119       return;//switch end function end
120     }else
121     {
122       fprintf(stderr,"function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
123       goto errret;
124     }
125  errret:
126   EventDel(g_epollFd, ev);
127   close(ev->fd);
128 }
129 // send data
130 void SendData(struct myevent_s *ev)
131 {
132   fprintf(stderr,"JustIn:function=%s,line=%d,fd=%d\n",__func__,__LINE__,ev->fd);
133   int len;
134   // send data
135   len = send(ev->fd, ev->send_buff, ev->send_len, 0);
136   ev->send_len = 0;
137   fprintf(stderr,"sendlen=%d:function=%s,line=%d,fd=%d\n",len,__func__,__LINE__,ev->fd);
138   if(len < 0)   
139     {
140       close(ev->fd);
141       fprintf(stderr,"err=%d:function=%s,line=%d,fd=%d\n",errno,__func__,__LINE__,ev->fd);
142     }else
143     {
144       //let system known can recv
145       EventAdd(g_epollFd, EPOLLIN|EPOLLET,ev);
146     }
147 }
148 
149 void *accept_thread_work(void *arg)
150 {
151   while(1)
152     {
153       int* plistenFd = (int*)arg;
154       fprintf(stderr,"function=%s,line=%d,listenfd=%d\n",__func__,__LINE__,*plistenFd);
155       struct sockaddr_in sin;
156       socklen_t len = sizeof(struct sockaddr_in);
157       int nfd, i;
158       // accept
159       if((nfd = accept(*plistenFd, (struct sockaddr*)&sin, &len)) == -1)
160     {
161       if(errno != EAGAIN && errno != EINTR)
162         {
163           fprintf(stderr,"%s: bad accept", __func__);
164         }
165       continue;
166     }
167       do
168     {
169       for(i = 0; i < MAX_EVENTS; i++)
170         {
171           if(g_Events[i].status == 0)
172         {
173           fprintf(stderr,"function=%s,line=%d,listenfd=%d,currentindex=%d\n",__func__,__LINE__,*plistenFd,i);
174           break;
175         }
176         }
177       if(i == MAX_EVENTS)
178         {
179           fprintf(stderr,"max events:function=%s,line=%d,listenFd=%d\n",__func__,__LINE__,*plistenFd);
180           break;
181         }
182       // set nonblocking
183       fprintf(stderr,"set nonblocking:function=%s,line=%d,listenfd=%d\n",__func__,__LINE__,*plistenFd);
184       if(fcntl(nfd, F_SETFL, O_NONBLOCK) < 0) break;
185       // add a read event for receive data
186       EventSet(&g_Events[i], nfd,0);
187       EventAdd(g_epollFd, EPOLLIN|EPOLLET, &g_Events[i]);
188       fprintf(stderr,"new conn[%s:%d][time:%d]\n", inet_ntoa(sin.sin_addr), ntohs(sin.sin_port),(int) g_Events[i].last_active);
189     }while(0);
190     }
191   return NULL;
192 }
193 
194 void *epoll_wait_thread_work(void *arg)
195 {
196   fprintf(stderr,"justin:function=%s,line=%d\n",__func__,__LINE__);
197   // event loop
198   struct epoll_event events[MAX_EVENTS];
199 
200   int checkPos = 0;
201   while(1){
202     // a simple timeout check here, every time 100, better to use a mini-heap, and add timer event
203     long now = time(NULL);
204     int i;
205     for(i = 0; i < 100; i++, checkPos++) // doesn't check listen fd
206       {
207     if(checkPos == MAX_EVENTS) checkPos = 0; // recycle
208     if(g_Events[checkPos].status != 1) continue;
209     long duration = now - g_Events[checkPos].last_active;
210     if(duration >= 60) // 60s timeout
211       {
212         close(g_Events[checkPos].fd);
213         fprintf(stderr,"[fd=%d] timeout[%d--%d].\n",(int) g_Events[checkPos].fd,(int) g_Events[checkPos].last_active, (int)now);
214         EventDel(g_epollFd, &g_Events[checkPos]);
215       }
216       }
217     // wait for events to happen
218     int fds = epoll_wait(g_epollFd, events, MAX_EVENTS, 1000);
219     if(fds < 0){
220       fprintf(stderr,"epoll_wait error, exit\n");
221       break;
222     }
223     for(i = 0; i < fds; i++){
224       struct myevent_s *ev = (struct myevent_s*)events[i].data.ptr;
225       if(events[i].events&EPOLLIN) // read event
226     {
227       sem_wait(&bin_sem_recv_fd_produce);
228       fprintf(stderr,"readEvent:function=%s,line=%d:fd=%d\n",__func__,__LINE__,ev->fd);
229       //ev->call_back(ev->fd, events[i].events, ev->arg);
230       struct QUEUE_RECV_FD_ITEM *item;
231       item = malloc(sizeof(struct QUEUE_RECV_FD_ITEM));
232       item->ev = ev;
233       pthread_mutex_lock(&queue_recv_fd_mutex);
234       TAILQ_INSERT_TAIL(&queue_recv_fd_header,item,recv_fd_entries);
235       pthread_mutex_unlock(&queue_recv_fd_mutex);
236       sem_post(&bin_sem_recv_fd_consume);
237     }else if(events[i].events&EPOLLOUT) // write event
238     {
239       sem_post(&bin_sem_send_fd_consume);
240       fprintf(stderr,"post send fd consume:function=%s,line=%d:fd=%d\n",__func__,__LINE__,ev->fd);
241     }
242     }
243   }
244   return NULL;
245 }
246 
247 void *recv_data_thread_work(void *arg)
248 {
249   while(1)
250     {
251       sem_wait(&bin_sem_recv_fd_consume);
252       fprintf(stderr,"justin:function=%s,line=%d\n",__func__,__LINE__);
253       int index = (int)arg;
254       fprintf(stderr,"recv thread id is %d\n",index);
255       pthread_mutex_lock(&queue_recv_fd_mutex);
256       struct QUEUE_RECV_FD_ITEM *item;
257       item = TAILQ_FIRST(&queue_recv_fd_header);
258       TAILQ_REMOVE(&queue_recv_fd_header,item,recv_fd_entries); 
259       pthread_mutex_unlock(&queue_recv_fd_mutex);
260       RecvData(item->ev);
261     }
262   return NULL;
263 }
264 
265 void *send_data_thread_work(void *arg)
266 {
267   while(1)
268     {  
269       sem_wait(&bin_sem_send_fd_consume);
270       fprintf(stderr,"justin:function=%s,line=%d\n",__func__,__LINE__);
271       pthread_mutex_lock(&queue_send_fd_mutex);
272       struct QUEUE_SEND_FD_ITEM *item;
273       item = TAILQ_FIRST(&queue_send_fd_header);
274       TAILQ_REMOVE(&queue_send_fd_header,item,send_fd_entries); 
275       pthread_mutex_unlock(&queue_send_fd_mutex);
276       SendData(item->ev);
277     }
278   return NULL;
279 }
280 
281 void *logic_data_thread_work(void *arg)
282 {
283   while(1)
284     {
285       //remove logic queue
286       sem_wait(&bin_sem_logic_data_consume);
287       //for test
288       int index = (int)arg;
289       fprintf(stderr,"logic thread id is %d\n",index);
290 
291       pthread_mutex_lock(&queue_logic_data_mutex);
292       struct QUEUE_LOGIC_DATA_ITEM *item;
293       item = TAILQ_FIRST(&queue_logic_data_header);
294       TAILQ_REMOVE(&queue_logic_data_header,item,logic_data_entries); 
295       pthread_mutex_unlock(&queue_logic_data_mutex);
296       //logic header
297       switch(item->ev->header.msg_type)
298     {
299     case msg_lost:
300       {
301         rq_lost* rq = (rq_lost*)item->ev->recv_buff;
302         
303         rs_lost rs;
304         rs.header.msg_type = msg_lost;
305         rs.header.size = sizeof(rs_lost);
306         rs.header.length = 0;
307 
308         if(strcmp(rq->cardno,"12345")==0)
309           {
310         rs.is_ok = 1;                        
311           }
312         else
313           {
314         rs.is_ok = 0;
315           }
316         memcpy(&item->ev->header,&rs.header,sizeof(msg_header));
317         item->ev->send_len = sizeof(rs);
318         memcpy(item->ev->send_buff,&rs,sizeof(rs));
319         break;
320       }
321     }
322    
323       //add to send fd queue
324       sem_wait(&bin_sem_send_fd_produce);
325       fprintf(stderr,"after wait send fd produce\n");
326       struct QUEUE_SEND_FD_ITEM *sendItem;
327       sendItem = malloc(sizeof(struct QUEUE_SEND_FD_ITEM));
328       sendItem->ev = item->ev;
329       pthread_mutex_lock(&queue_send_fd_mutex);
330       TAILQ_INSERT_TAIL(&queue_send_fd_header,sendItem,send_fd_entries);
331       pthread_mutex_unlock(&queue_send_fd_mutex);
332       //let system known can send
333       EventAdd(g_epollFd, EPOLLOUT|EPOLLET, item->ev);
334     }
335   return NULL;
336 }
337 
338 int main(int argc, char **argv)
339 {
340   int res;
341   //recv fd queue
342   TAILQ_INIT(&queue_recv_fd_header);
343   res = sem_init(&bin_sem_recv_fd_consume,0,0);
344   if(res)
345     {
346       fprintf(stderr,"sem init consume failed\n");
347       exit(EXIT_FAILURE);
348     }
349 
350   res = sem_init(&bin_sem_recv_fd_produce,0,MAX_EVENTS);
351   if(res)
352     {
353       fprintf(stderr,"sem init produce failed\n");
354       exit(EXIT_FAILURE);
355     }
356 
357   res = pthread_mutex_init(&queue_recv_fd_mutex,NULL);
358   if(res!=0)
359     {
360       perror("create mutex for queue recv failed\n");
361       exit(EXIT_FAILURE);
362     }
363   //logic data queue
364   TAILQ_INIT(&queue_logic_data_header);
365   res = sem_init(&bin_sem_logic_data_consume,0,0);
366   if(res)
367     {
368       fprintf(stderr,"sem init logic data consume failed\n");
369       exit(EXIT_FAILURE);
370     }
371 
372   res = sem_init(&bin_sem_logic_data_produce,0,MAX_EVENTS);
373   if(res)
374     {
375       fprintf(stderr,"sem init logic data produce failed\n");
376       exit(EXIT_FAILURE);
377     }
378 
379   res = pthread_mutex_init(&queue_logic_data_mutex,NULL);
380   if(res!=0)
381     {
382       perror("create mutex for queue logic data failed\n");
383       exit(EXIT_FAILURE);
384     }
385 
386   //send fd queue
387   TAILQ_INIT(&queue_send_fd_header);
388   res = sem_init(&bin_sem_send_fd_consume,0,0);
389   if(res)
390     {
391       fprintf(stderr,"sem init send fd consume failed\n");
392       exit(EXIT_FAILURE);
393     }
394 
395   res = sem_init(&bin_sem_send_fd_produce,0,MAX_EVENTS);
396   if(res)
397     {
398       fprintf(stderr,"sem init send fd produce failed\n");
399       exit(EXIT_FAILURE);
400     }
401 
402   res = pthread_mutex_init(&queue_send_fd_mutex,NULL);
403   if(res!=0)
404     {
405       perror("create mutex for queue send fd failed\n");
406       exit(EXIT_FAILURE);
407     }
408 
409   short port = 3342; // default port
410   if(argc == 2){
411     port = atoi(argv[1]);
412   }
413   // create epoll
414   g_epollFd = epoll_create(MAX_EVENTS);
415   if(g_epollFd <= 0) 
416     {
417       fprintf(stderr,"create epoll failed:fd=%d:function=%s,line=%d\n", g_epollFd,__func__,__LINE__);
418       exit(EXIT_FAILURE);
419     }
420   // create & bind listen socket
421   int listenFd = socket(AF_INET, SOCK_STREAM, 0);
422   // bind & listen
423   struct sockaddr_in sin;
424   bzero(&sin, sizeof(sin));
425   sin.sin_family = AF_INET;
426   sin.sin_addr.s_addr = INADDR_ANY;
427   sin.sin_port = htons(port);
428   bind(listenFd, (const struct sockaddr*)&sin, sizeof(sin));
429   listen(listenFd, 5);
430   fprintf(stderr,"server running:port[%d]\n", port);
431   //create accept thread
432 
433   void *thread_result;
434   pthread_t accept_t;  
435   res = pthread_create(&accept_t,NULL,accept_thread_work,(void *)&listenFd);
436   if(res != 0)
437     {
438       perror("accept create failed\n");
439       exit(EXIT_FAILURE);
440     }
441 
442   //create epoll wait thread
443   pthread_t epoll_wait_t;
444   res = pthread_create(&epoll_wait_t,NULL,epoll_wait_thread_work,NULL);
445   if(res != 0)
446     {
447       perror("create epoll wait thread failed\n");
448       exit(EXIT_FAILURE);
449     }
450   //create two recv data thread
451   pthread_t recv_data_t;
452   res = pthread_create(&recv_data_t,NULL,recv_data_thread_work,(void*)1);
453   if(res!=0)
454     {
455       perror("create recv data thread failed\n");
456       exit(EXIT_FAILURE);
457     }
458 
459   pthread_t recv_data_t_1;
460   res = pthread_create(&recv_data_t_1,NULL,recv_data_thread_work,(void*)2);
461   if(res!=0)
462     {
463       perror("create recv data thread failed\n");
464       exit(EXIT_FAILURE);
465     }
466   //create two send data thread
467   pthread_t send_data_t;
468   res = pthread_create(&send_data_t,NULL,send_data_thread_work,(void*)1);
469   if(res!=0)
470     {
471       perror("create send data thread failed\n");
472       exit(EXIT_FAILURE);
473     }
474 
475   pthread_t send_data_t_1;
476   res = pthread_create(&send_data_t_1,NULL,send_data_thread_work,(void*)2);
477   if(res!=0)
478     {
479       perror("create send data thread failed\n");
480       exit(EXIT_FAILURE);
481     }
482 
483   //create two logic work thread
484   pthread_t logic_work_t;
485   res = pthread_create(&logic_work_t,NULL,logic_data_thread_work,(void*)1);
486   if(res!=0)
487     {
488       perror("create logic work thread failed\n");
489       exit(EXIT_FAILURE);
490     }
491 
492  pthread_t logic_work_t_1;
493   res = pthread_create(&logic_work_t_1,NULL,logic_data_thread_work,(void*)2);
494   if(res!=0)
495     {
496       perror("create logic work thread failed\n");
497       exit(EXIT_FAILURE);
498     }
499 
500   //wait child thread
501   res = pthread_join(accept_t,&thread_result);
502   if(res!=0)
503     {
504       perror("accept thread join failed\n");
505       exit(EXIT_FAILURE);
506     }
507 
508   //wait child thread
509   res = pthread_join(epoll_wait_t,&thread_result);
510   if(res!=0)
511     {
512       perror("epoll wait thread join failed\n");
513       exit(EXIT_FAILURE);
514     }
515 
516   //wait child thread
517   res = pthread_join(recv_data_t,&thread_result);
518   if(res!=0)
519     {
520       perror("recv data thread join failed\n");
521       exit(EXIT_FAILURE);      
522     }
523   //wait child thread
524   res = pthread_join(recv_data_t_1,&thread_result);
525   if(res!=0)
526     {
527       perror("recv data thread join failed\n");
528       exit(EXIT_FAILURE);      
529     }
530 
531   //wait child thread
532   res = pthread_join(send_data_t,&thread_result);
533   if(res!=0)
534     {
535       perror("send data thread join failed\n");
536       exit(EXIT_FAILURE);      
537     }
538   //wait child thread
539   res = pthread_join(send_data_t_1,&thread_result);
540   if(res!=0)
541     {
542       perror("send data thread join failed\n");
543       exit(EXIT_FAILURE);      
544     }
545   //wait child thread
546   res = pthread_join(logic_work_t,&thread_result);
547   if(res!=0)
548     {
549       perror("logic work thread join failed\n");
550       exit(EXIT_FAILURE);      
551     }
552   //wait child thread
553   res = pthread_join(logic_work_t_1,&thread_result);
554   if(res!=0)
555     {
556       perror("logic work thread join failed\n");
557       exit(EXIT_FAILURE);      
558     }
559   // free resource
560   close(g_epollFd);
561   sem_destroy(&bin_sem_recv_fd_consume);
562   sem_destroy(&bin_sem_recv_fd_produce);
563   pthread_mutex_destroy(&queue_recv_fd_mutex);
564 
565   sem_destroy(&bin_sem_logic_data_consume);
566   sem_destroy(&bin_sem_logic_data_produce);
567   pthread_mutex_destroy(&queue_logic_data_mutex);
568 
569   sem_destroy(&bin_sem_send_fd_consume);
570   sem_destroy(&bin_sem_send_fd_produce);
571   pthread_mutex_destroy(&queue_send_fd_mutex);
572   return 0;
573