zoukankan      html  css  js  c++  java

  • myfastchat

    https://github.com/wenwenqingqing/myfastchat

    chat_server.h

    点击(此处)折叠或打开

    1. #include <stdio.h>
    2. #include <string.h>
    3. #include <stdlib.h>
    4. #include <unistd.h>
    5. #include <sys/time.h>
    6. #include <sys/socket.h>
    7. #include <sys/types.h>
    8. #include <sys/select.h>
    9. #include <netinet/in.h>
    10. #include <arpa/inet.h>
    11. #include <sys/epoll.h>
    12. #include <fcntl.h>
    13. #include <netinet/tcp.h>
    14. #include <pthread.h>
    15. #include <errno.h>
    16. #include <assert.h>
    17. #define BUFSIZE 65535

    19. //#define PORT 9002
    20. #define SERVER_PORT 11111
    21. #define SERVER_HOST "127.0.0.1"

    23. struct {
    24.     int len;
    25.     char buf[65535];
    26. } TcpPacket;

    28. typedef struct worker {
    29.     void *(*process) (void *arg, char *buf);
    30.     void *arg;
    31.     char buf[255];
    32.     struct worker *next;
    33. } CThread_worker;

    35. typedef struct param {
    36.     int *epoll_fd;
    37.     int *accept_fd;
    38.     int *maxfd;
    39.     struct epoll_event *ev;
    40. } CThread_param;

    42. typedef struct {
    43.     pthread_mutex_t queue_lock;
    44.     pthread_cond_t queue_ready;
    45.     CThread_worker *queue_head;
    46.     int shutdown;
    47.     pthread_t *threadid;
    48.     int max_thread_num;
    49.     int cur_queue_size;
    50. } CThread_pool;

    52. void *myprocess(void *arg, char *buf);
    53. void pool_init(int);
    54. int pool_add_worker(void *(*process) (void *arg, char *buf), void *arg);
    55. void *thread_routine(void *arg);

    57. static CThread_pool *pool = NULL;

    59. int buildConnect();
    60. long int fd_A[5] = { 0, 0, 0, 0 };    //because fd_A maybe flow

    62. int acceptSelectClient(int);

    64. int acceptEpollClient(int);
    65. int set_noblocking(int);
    66. int readn(int, char *, size_t);
    67. int readrec(int, char *, size_t);
    68. //int pthread_test();
    69. //void* thread_fun(void*);
    70. //int reicveData();

    72. //int toQueue();

    74. //int dealData();

    76. //int sendData();


    chat_server.c

    点击(此处)折叠或打开

    1. #include "chat_server.h"
    2. int main()
    3. {
    4.     int fd;
    5.     pool_init(3);
    6.     fd = buildConnect();
    7.     //acceptSelectClient(fd);
    8.     acceptEpollClient(fd);
    9.     return 0;
    10. }

    12. int buildConnect()
    13. {
    14.     struct sockaddr_in server;
    15.     int listen_fd;
    16.     if ((listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
    17.         return -1;
    18.     }

    20.     server.sin_family = AF_INET;
    21.     server.sin_port = htons(SERVER_PORT);
    22.     server.sin_addr.s_addr = inet_addr(SERVER_HOST);

    24.     if (bind(listen_fd, (struct sockaddr *)&server, sizeof(server)) < 0) {
    25.         return -1;
    26.     }
    27.     listen(listen_fd, 5);
    28.     printf("server start listen.... ");
    29.     return listen_fd;

    31. }

    33. int set_nonblocking(int listen_fd)
    34. {
    35.     if (fcntl(listen_fd, F_SETFL, fcntl(listen_fd, F_GETFD, 0) | O_NONBLOCK) == -1) {
    36.         return -1;
    37.     }
    38.     return 0;
    39. }

    41. /*
    42. void * thread_fun(void *arg)
    43. {
    44.     pid_t pid;
    45.     pthread_t tid;
    46.     pid = getpid();
    47.     tid = pthread_self();
    48.     printf(" pid %u tid %u (0x%x) " ,(unsigned int)pid, (unsigned int )tid , (unsigned int ) tid);    
    49.     return ((void *)1);
    50. }
    51. int pthread_test()
    52. {
    53.     int err;
    54.     pthread_t tid;
    55.     err = pthread_create(&tid, NULL , thread_fun, NULL);
    56.     //err = pthread_join(tid , NULL);
    57.     return 0;

    59. }
    60. */
    61. int readn(int accept_fd, char *buf, size_t len)
    62. {
    63.     int cnt;
    64.     int rc;
    65.     cnt = len;
    66.     while (cnt > 0) {
    67.         rc = recv(accept_fd, buf, cnt, 0);
    68.         if (rc < 0) {
    69.             return -1;
    70.         }
    71.         if (rc == 0)
    72.             return len - cnt;
    73.         buf += rc;
    74.         cnt -= rc;
    75.     }
    76.     return len;
    77. }

    79. int readrec(int accept_fd, char *buf, size_t len)
    80. {
    81.     int reclen;
    82.     int rc;
    83.     rc = readn(accept_fd, (char *)&reclen, sizeof(reclen));
    84.     if (rc != sizeof(reclen))
    85.         return rc < 0 ? -1 : 0;
    86.     reclen = ntohl(reclen);

    88.     if (reclen > len) {
    89.         while (reclen > 0) {
    90.             rc = readn(accept_fd, buf, len);
    91.             if (rc != len)
    92.                 return rc < 0 ? -1 : 0;
    93.             reclen -= len;
    94.             if (reclen < len)
    95.                 len = reclen;
    96.         }
    97.     }
    98.     rc = readn(accept_fd, buf, reclen);
    99.     if (rc != reclen)
    100.         return rc < 0 ? -1 : 0;
    101.     return rc;
    102. }

    104. int acceptEpollClient(int listen_fd)
    105. {
    106.     int epoll_fd;
    107.     int maxfd;
    108.     int ret;
    109.     int accept_fd;
    110.     int i;
    111.     struct sockaddr_in client;
    112.     struct epoll_event ev;
    113.     struct epoll_event events[5];
    114.     //int len = sizeof(client);
    115.     socklen_t len = sizeof(client);
    116.     char buf[255];
    117.     if (set_nonblocking(listen_fd) < 0) {
    118.         printf("fcntl error ");
    119.         return -1;
    120.     }
    121.     epoll_fd = epoll_create(5);
    122.     if (epoll_fd == 0) {
    123.         printf("create error ");
    124.         return -1;
    125.     }
    126.     ev.events = EPOLLIN | EPOLLET;
    127.     ev.data.fd = listen_fd;
    128.     if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_fd, &ev) < 0) {
    129.         printf("epoll add error ");
    130.         return -1;
    131.     } else {
    132.         printf("listen id %d ", (int)listen_fd);
    133.     }
    134.     //set_keep_alive(listen_fd);
    135.     maxfd = 1;
    136.     while (1) {
    137.         ret = epoll_wait(epoll_fd, events, maxfd, -1);

    139.         if (ret == -1) {
    140.             printf("wait ");
    141.             return -1;
    142.         }
    143.         for (i = 0; i < ret; i++) {
    144.             if (events[i].data.fd == listen_fd) {
    145.                 accept_fd = accept(listen_fd, (struct sockaddr *)&client, &len);
    146.                 if (accept_fd < 0) {
    147.                     printf("accetp error ");
    148.                     continue;
    149.                 }
    150.                 set_nonblocking(accept_fd);
    151.                 ev.events = EPOLLIN | EPOLLET;
    152.                 ev.data.fd = accept_fd;
    153.                 if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, accept_fd, &ev) < 0) {
    154.                     printf("epoll add error ");
    155.                     return -1;
    156.                 } else {
    157.                     printf("new client %d ", (int)accept_fd);

    159.                     maxfd++;
    160.                 }
    161.             } else if (events[i].events & EPOLLIN) {
    162.                 CThread_param *CPThread_param = (CThread_param *) malloc(sizeof(CThread_param));
    163.                 CPThread_param->epoll_fd = &epoll_fd;
    164.                 CPThread_param->accept_fd = &(events[i].data.fd);
    165.                 CPThread_param->ev = &ev;
    166.                 CPThread_param->maxfd = &maxfd;

    168.                 pool_add_worker(myprocess, CPThread_param);
    169.                 /*
    170.                  while(1)
    171.                  {
    172.                  ret = readrec(events[i].data.fd, buf,sizeof(buf));
    173.                  // ret = recv(events[i].data.fd, buf, sizeof(buf),0);
    174.                  if(ret == -1)
    175.                  {
    176.                  if(errno == EAGAIN)
    177.                  {
    178.                  break;

    180.                  }

    182.                  break;
    183.                  }
    184.                  else if(ret == 0)
    185.                  {
    186.                  printf("client[%d] close ",i);
    187.                  epoll_ctl(epoll_fd, EPOLL_CTL_DEL, events[i].data.fd,&ev);
    188.                  maxfd--;
    189.                  break;

    191.                  }
    192.                  else
    193.                  {

    195.                  buf[ret] = '0';
    196.                  printf("client send %s ", buf);

    198.                  send(events[i].data.fd,buf, sizeof(buf) , 0);
    199.                  pthread_test();
    200.                  }
    201.                  } */
    202.             }

    204.         }
    205.     }
    206.     close(listen_fd);
    207.     return 0;

    209. }

    211. // select block way
    212. int acceptSelectClient(int listen_fd)
    213. {
    214.     struct sockaddr_in client;
    215.     int accept_fd;
    216.     int maxfd;
    217.     struct timeval tv;
    218.     fd_set select_fd;
    219.     int ret;
    220.     int i;
    221.     //int len = sizeof(client);
    222.     socklen_t len = sizeof(client);
    223.     int amount = 0;
    224.     char buf[256];

    226.     maxfd = listen_fd;

    228.     while (1) {
    229.         FD_ZERO(&select_fd);
    230.         FD_SET(listen_fd, &select_fd);
    231.         tv.tv_sec = 30;
    232.         tv.tv_usec = 0;
    233.         for (i = 0; i < 5; i++) {
    234.             if (fd_A[i] != 0) {
    235.                 FD_SET(fd_A[i], &select_fd);
    236.             }
    237.         }
    238.         ret = select(maxfd + 1, &select_fd, NULL, NULL, &tv);
    239.         if (ret < 0) {
    240.             return -1;
    241.         } else if (ret == 0) {
    242.             printf("timeout ");
    243.             continue;
    244.         }
    245.         for (i = 0; i < amount; i++) {
    246.             if (FD_ISSET(fd_A[i], &select_fd)) {
    247.                 ret = recv(fd_A[i], buf, sizeof(buf), 0);
    248.                 if (ret <= 0) {
    249.                     printf("client[%d] close ", i);
    250.                     close(fd_A[i]);
    251.                     FD_CLR(fd_A[i], &select_fd);
    252.                     fd_A[i] = 0;

    254.                 } else {
    255.                     buf[ret] = '0';
    256.                     printf("client send %s ", buf);
    257.                 }
    258.             }
    259.         }
    260.         if (FD_ISSET(listen_fd, &select_fd)) {
    261.             printf("new connect come on ");
    262.             accept_fd = accept(listen_fd, (struct sockaddr *)&client, &len);
    263.             if (accept_fd <= 0) {
    264.                 printf("accept error ");
    265.                 continue;
    266.             }

    268.             if (amount < 5) {
    269.                 fd_A[amount++] = accept_fd;
    270.                 if (accept_fd > maxfd)
    271.                     maxfd = accept_fd;
    272.             } else {
    273.                 printf("max connects arrive ");
    274.             }
    275.         }

    277.     }
    278.     for (i = 0; i < 5; i++) {
    279.         if (fd_A[i] != 0) {
    280.             close(fd_A[i]);
    281.         }
    282.     }
    283.     return 0;
    284. }

    286. void pool_init(int max_thread_num)
    287. {
    288.     pool = (CThread_pool *) malloc(sizeof(CThread_pool));
    289.     pthread_mutex_init(&(pool->queue_lock), NULL);
    290.     pthread_cond_init(&(pool->queue_ready), NULL);
    291.     pool->queue_head = NULL;
    292.     pool->max_thread_num = max_thread_num;
    293.     pool->cur_queue_size = 0;
    294.     pool->shutdown = 0;
    295.     pool->threadid = (pthread_t *) malloc(max_thread_num * sizeof(pthread_t));
    296.     int i = 0;
    297.     for (i = 0; i < max_thread_num; i++) {
    298.         pthread_create(&(pool->threadid[i]), NULL, thread_routine, NULL);
    299.     }
    300. }

    302. int pool_add_worker(void *(*process) (void *arg, char *buf), void *arg)
    303. {

    305.     CThread_worker *newworker = (CThread_worker *) malloc(sizeof(CThread_worker));
    306.     newworker->process = process;
    307.     newworker->arg = arg;
    308.     newworker->next = NULL;
    309.     pthread_mutex_lock(&(pool->queue_lock));

    311.     CThread_worker *member = pool->queue_head;
    312.     if (member != NULL) {
    313.         while (member->next != NULL)
    314.             member = member->next;
    315.         member->next = newworker;
    316.     } else {
    317.         pool->queue_head = newworker;
    318.     }
    319.     assert(pool->queue_head != NULL);
    320.     pool->cur_queue_size++;
    321.     printf("cur_____%d ", pool->cur_queue_size);
    322.     pthread_mutex_unlock(&(pool->queue_lock));

    324.     pthread_cond_signal(&(pool->queue_ready));
    325.     return 0;
    326. }

    328. int pool_destroy()
    329. {
    330.     if (pool->shutdown)
    331.         return -1;
    332.     pool->shutdown = 1;

    334.     pthread_cond_broadcast(&(pool->queue_ready));

    336.     int i;
    337.     for (i = 0; i < pool->max_thread_num; i++)
    338.         pthread_join(pool->threadid[i], NULL);
    339.     free(pool->threadid);

    341.     CThread_worker *head = NULL;
    342.     while (pool->queue_head != NULL) {
    343.         head = pool->queue_head;
    344.         pool->queue_head = pool->queue_head->next;
    345.         free(head);
    346.     }

    348.     pthread_mutex_destroy(&(pool->queue_lock));
    349.     pthread_cond_destroy(&(pool->queue_ready));

    351.     free(pool);

    353.     pool = NULL;
    354.     return 0;
    355. }

    357. void *thread_routine(void *arg)
    358. {
    359.     printf("starting thread 0x%x ", (unsigned int)pthread_self());
    360.     while (1) {
    361.         pthread_mutex_lock(&(pool->queue_lock));

    363.         while (pool->cur_queue_size == 0 && !pool->shutdown) {
    364.             printf("thread 0x%x is waiting ", (unsigned int)pthread_self());
    365.             pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
    366.         }

    368.         if (pool->shutdown) {

    370.             pthread_mutex_unlock(&(pool->queue_lock));
    371.             printf("thread 0x%x will exit ", (unsigned int)pthread_self());
    372.             pthread_exit(NULL);
    373.         }
    374.         printf("thread 0x%x is starting to work ", (unsigned int)pthread_self());

    376.         assert(pool->cur_queue_size != 0);
    377.         assert(pool->queue_head != NULL);

    379.         pool->cur_queue_size--;
    380.         CThread_worker *worker = pool->queue_head;
    381.         pool->queue_head = worker->next;
    382.         pthread_mutex_unlock(&(pool->queue_lock));
    383.         printf("start process... ");
    384.         (*(worker->process)) (worker->arg, worker->buf);
    385.         free(worker);
    386.         worker = NULL;
    387.     }

    389.     pthread_exit(NULL);
    390. }

    392. void *myprocess(void *arg, char *buf)
    393. {
    394.     int ret;
    395.     char buf1[255];
    396.     struct param *PP_param;
    397.     PP_param = (CThread_param *) arg;
    398.     printf("myprocess start... ");
    399.     while (1) {
    400.         //printf("fd_%d ",*(PP_param->accept_fd));
    401.         ret = readrec(*(PP_param->accept_fd), buf1, sizeof(buf1));
    402.         // ret = recv(events[i].data.fd, buf, sizeof(buf),0);
    403.         if (ret == -1) {
    404.             if (errno == EAGAIN) {
    405.                 break;

    407.             }

    409.             break;
    410.         } else if (ret == 0) {
    411.             epoll_ctl(*(PP_param->epoll_fd), EPOLL_CTL_DEL, *(PP_param->accept_fd), PP_param->ev);
    412.             *(PP_param->maxfd)--;
    413.             break;

    415.         } else {

    417.             buf1[ret] = '0';
    418.             printf("client send %s ", buf1);

    420.             printf("threadid is 0x%x ", (unsigned int)pthread_self());

    422.             send(*(PP_param->accept_fd), buf1, sizeof(buf1), 0);
    423.         }

    425.     }

    427.     return NULL;
    428. }


    chat_client.c

    点击(此处)折叠或打开

    1. #include <stdio.h>
    2. #include <stdlib.h>
    3. #include <string.h>
    4. #include <unistd.h>
    5. #include <netdb.h>
    6. #include <sys/socket.h>
    7. #include <sys/types.h>
    8. #include <netinet/in.h>
    9. #include <arpa/inet.h>
    10. #include <sys/time.h>

    12. #define BUFSIZE 8192
    13. #define SERVER_PORT 11111
    14. #define SERVER_HOST "127.0.0.1"

    16. struct {
    17.     int len;
    18.     char buf[65535];
    19. } TcpPacket;
    20. enum { CMD_NAME, DST_IP, DST_PORT };

    22. int main(int argc, char *argv[])
    23. {
    24.     struct sockaddr_in server;
    25.     unsigned long dst_ip;
    26.     int port;
    27.     int s;
    28.     int n;
    29.     char buf[BUFSIZE];
    30.     char buf1[BUFSIZE];
    31.     char cmd[BUFSIZE];
    32.     struct timeval tv;
    33.     fd_set readfd;

    35.     port = SERVER_PORT;
    36.     dst_ip = inet_addr(SERVER_HOST);
    37.     if ((s = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
    38.         perror("sokcet error");
    39.         exit(-1);
    40.     }
    41.     memset((char *)&server, 0, sizeof(server));
    42.     server.sin_family = AF_INET;
    43.     server.sin_addr.s_addr = dst_ip;
    44.     server.sin_port = htons(port);

    46.     if (connect(s, (struct sockaddr *)&server, sizeof(server)) < 0) {
    47.         perror("connect error");
    48.         exit(-1);
    49.     }
    50.     while (1) {
    51.         tv.tv_sec = 600;
    52.         tv.tv_usec = 0;
    53.         FD_ZERO(&readfd);
    54.         FD_SET(0, &readfd);
    55.         FD_SET(s, &readfd);
    56.         if ((select(s + 1, &readfd, NULL, NULL, &tv)) < 0) {
    57.             perror("timeout");
    58.             break;
    59.         }
    60.         if (FD_ISSET(0, &readfd)) {
    61.             if ((n = read(0, TcpPacket.buf, sizeof(TcpPacket.buf))) <= 0)
    62.                 break;
    63.             n = strlen(TcpPacket.buf);
    64.             // TcpPacket.buf[n-1] = '0';
    65.             TcpPacket.len = ntohl(n);
    66.             if (send(s, (char *)&TcpPacket, n + sizeof(TcpPacket.len), 0) <= 0)
    67.                 break;
    68.         }
    69.         if (FD_ISSET(s, &readfd)) {
    70.             if ((n = recv(s, buf1, BUFSIZE - 1, 0)) <= 0) {
    71.                 perror("recv error! ");
    72.                 exit(-1);
    73.             }
    74.             buf1[n] = '0';
    75.             printf("%s", buf1);
    76.             fflush(stdout);
    77.         }
    78.     }
    79.     close(s);
    80.     return 0;
    81. }

    g++ -lpthread chat_server.c -o chat_server

    g++ -lpthread chat_client.c -o chat_client




    <script>window._bd_share_config={"common":{"bdSnsKey":{},"bdText":"","bdMini":"2","bdMiniList":false,"bdPic":"","bdStyle":"0","bdSize":"16"},"share":{}};with(document)0[(getElementsByTagName('head')[0]||body).appendChild(createElement('script')).src='http://bdimg.share.baidu.com/static/api/js/share.js?v=89860593.js?cdnversion='+~(-new Date()/36e5)];</script>
    阅读(67) | 评论(0) | 转发(0) |
    0

    上一篇：epoll pthread—share the same epoll file descriptor among threads?

    下一篇：epoll pthread chat

    相关热门文章
    给主人留下些什么吧！~~
    评论热议
  • 相关阅读:
    apache和tomcat有什么不同，为什么要整合apache 和tomcat？
    servlet
    关于Spring配置文件xml文档的schema约束
    request对象和response对象
    多线程
    数据结构得到连续数据的手段java Enumeration
    程序员八荣八耻
    windows更改MySQL存储路径
    Tomcat源码学习（1）
    Tomcat源码学习（2）——启动过程分析
  • 原文地址：https://www.cnblogs.com/ztguang/p/12649336.html
Copyright © 2011-2022 走看看