关键词搜索

源码搜索 ×
×

漫话Redis源码之八十

发布2022-02-13浏览451次

详情内容

socke相关的操作，需要搞清楚阻塞socket和非阻塞socket的用途哦：

  1. #include "anet.h"
  2.  
  3. static void anetSetError(char *err, const char *fmt, ...)
  4. {
  5.     va_list ap;
  6.  
  7.     if (!err) return;
  8.     va_start(ap, fmt);
  9.     vsnprintf(err, ANET_ERR_LEN, fmt, ap);
  10.     va_end(ap);
  11. }
  12.  
  13. int anetSetBlock(char *err, int fd, int non_block) {
  14.     int flags;
  15.  
  16.     /* Set the socket blocking (if non_block is zero) or non-blocking.
  17.      * Note that fcntl(2) for F_GETFL and F_SETFL can't be
  18.      * interrupted by a signal. */
  19.     if ((flags = fcntl(fd, F_GETFL)) == -1) {
  20.         anetSetError(err, "fcntl(F_GETFL): %s", strerror(errno));
  21.         return ANET_ERR;
  22.     }
  23.  
  24.     /* Check if this flag has been set or unset, if so, 
  25.      * then there is no need to call fcntl to set/unset it again. */
  26.     if (!!(flags & O_NONBLOCK) == !!non_block)
  27.         return ANET_OK;
  28.  
  29.     if (non_block)
  30.         flags |= O_NONBLOCK;
  31.     else
  32.         flags &= ~O_NONBLOCK;
  33.  
  34.     if (fcntl(fd, F_SETFL, flags) == -1) {
  35.         anetSetError(err, "fcntl(F_SETFL,O_NONBLOCK): %s", strerror(errno));
  36.         return ANET_ERR;
  37.     }
  38.     return ANET_OK;
  39. }
  40.  
  41. int anetNonBlock(char *err, int fd) {
  42.     return anetSetBlock(err,fd,1);
  43. }
  44.  
  45. int anetBlock(char *err, int fd) {
  46.     return anetSetBlock(err,fd,0);
  47. }
  48.  
  49. /* Enable the FD_CLOEXEC on the given fd to avoid fd leaks. 
  50.  * This function should be invoked for fd's on specific places 
  51.  * where fork + execve system calls are called. */
  52. int anetCloexec(int fd) {
  53.     int r;
  54.     int flags;
  55.  
  56.     do {
  57.         r = fcntl(fd, F_GETFD);
  58.     } while (r == -1 && errno == EINTR);
  59.  
  60.     if (r == -1 || (r & FD_CLOEXEC))
  61.         return r;
  62.  
  63.     flags = r | FD_CLOEXEC;
  64.  
  65.     do {
  66.         r = fcntl(fd, F_SETFD, flags);
  67.     } while (r == -1 && errno == EINTR);
  68.  
  69.     return r;
  70. }
  71.  
  72. /* Set TCP keep alive option to detect dead peers. The interval option
  73.  * is only used for Linux as we are using Linux-specific APIs to set
  74.  * the probe send time, interval, and count. */
  75. int anetKeepAlive(char *err, int fd, int interval)
  76. {
  77.     int val = 1;
  78.  
  79.     if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val)) == -1)
  80.     {
  81.         anetSetError(err, "setsockopt SO_KEEPALIVE: %s", strerror(errno));
  82.         return ANET_ERR;
  83.     }
  84.  
  85. #ifdef __linux__
  86.     /* Default settings are more or less garbage, with the keepalive time
  87.      * set to 7200 by default on Linux. Modify settings to make the feature
  88.      * actually useful. */
  89.  
  90.     /* Send first probe after interval. */
  91.     val = interval;
  92.     if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &val, sizeof(val)) < 0) {
  93.         anetSetError(err, "setsockopt TCP_KEEPIDLE: %s\n", strerror(errno));
  94.         return ANET_ERR;
  95.     }
  96.  
  97.     /* Send next probes after the specified interval. Note that we set the
  98.      * delay as interval / 3, as we send three probes before detecting
  99.      * an error (see the next setsockopt call). */
  100.     val = interval/3;
  101.     if (val == 0) val = 1;
  102.     if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &val, sizeof(val)) < 0) {
  103.         anetSetError(err, "setsockopt TCP_KEEPINTVL: %s\n", strerror(errno));
  104.         return ANET_ERR;
  105.     }
  106.  
  107.     /* Consider the socket in error state after three we send three ACK
  108.      * probes without getting a reply. */
  109.     val = 3;
  110.     if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &val, sizeof(val)) < 0) {
  111.         anetSetError(err, "setsockopt TCP_KEEPCNT: %s\n", strerror(errno));
  112.         return ANET_ERR;
  113.     }
  114. #else
  115.     ((void) interval); /* Avoid unused var warning for non Linux systems. */
  116. #endif
  117.  
  118.     return ANET_OK;
  119. }
  120.  
  121. static int anetSetTcpNoDelay(char *err, int fd, int val)
  122. {
  123.     if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof(val)) == -1)
  124.     {
  125.         anetSetError(err, "setsockopt TCP_NODELAY: %s", strerror(errno));
  126.         return ANET_ERR;
  127.     }
  128.     return ANET_OK;
  129. }
  130.  
  131. int anetEnableTcpNoDelay(char *err, int fd)
  132. {
  133.     return anetSetTcpNoDelay(err, fd, 1);
  134. }
  135.  
  136. int anetDisableTcpNoDelay(char *err, int fd)
  137. {
  138.     return anetSetTcpNoDelay(err, fd, 0);
  139. }
  140.  
  141. /* Set the socket send timeout (SO_SNDTIMEO socket option) to the specified
  142.  * number of milliseconds, or disable it if the 'ms' argument is zero. */
  143. int anetSendTimeout(char *err, int fd, long long ms) {
  144.     struct timeval tv;
  145.  
  146.     tv.tv_sec = ms/1000;
  147.     tv.tv_usec = (ms%1000)*1000;
  148.     if (setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)) == -1) {
  149.         anetSetError(err, "setsockopt SO_SNDTIMEO: %s", strerror(errno));
  150.         return ANET_ERR;
  151.     }
  152.     return ANET_OK;
  153. }
  154.  
  155. /* Set the socket receive timeout (SO_RCVTIMEO socket option) to the specified
  156.  * number of milliseconds, or disable it if the 'ms' argument is zero. */
  157. int anetRecvTimeout(char *err, int fd, long long ms) {
  158.     struct timeval tv;
  159.  
  160.     tv.tv_sec = ms/1000;
  161.     tv.tv_usec = (ms%1000)*1000;
  162.     if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) == -1) {
  163.         anetSetError(err, "setsockopt SO_RCVTIMEO: %s", strerror(errno));
  164.         return ANET_ERR;
  165.     }
  166.     return ANET_OK;
  167. }
  168.  
  169. /* Resolve the hostname "host" and set the string representation of the
  170.  * IP address into the buffer pointed by "ipbuf".
  171.  *
  172.  * If flags is set to ANET_IP_ONLY the function only resolves hostnames
  173.  * that are actually already IPv4 or IPv6 addresses. This turns the function
  174.  * into a validating / normalizing function. */
  175. int anetResolve(char *err, char *host, char *ipbuf, size_t ipbuf_len,
  176.                        int flags)
  177. {
  178.     struct addrinfo hints, *info;
  179.     int rv;
  180.  
  181.     memset(&hints,0,sizeof(hints));
  182.     if (flags & ANET_IP_ONLY) hints.ai_flags = AI_NUMERICHOST;
  183.     hints.ai_family = AF_UNSPEC;
  184.     hints.ai_socktype = SOCK_STREAM;  /* specify socktype to avoid dups */
  185.  
  186.     if ((rv = getaddrinfo(host, NULL, &hints, &info)) != 0) {
  187.         anetSetError(err, "%s", gai_strerror(rv));
  188.         return ANET_ERR;
  189.     }
  190.     if (info->ai_family == AF_INET) {
  191.         struct sockaddr_in *sa = (struct sockaddr_in *)info->ai_addr;
  192.         inet_ntop(AF_INET, &(sa->sin_addr), ipbuf, ipbuf_len);
  193.     } else {
  194.         struct sockaddr_in6 *sa = (struct sockaddr_in6 *)info->ai_addr;
  195.         inet_ntop(AF_INET6, &(sa->sin6_addr), ipbuf, ipbuf_len);
  196.     }
  197.  
  198.     freeaddrinfo(info);
  199.     return ANET_OK;
  200. }
  201.  
  202. static int anetSetReuseAddr(char *err, int fd) {
  203.     int yes = 1;
  204.     /* Make sure connection-intensive things like the redis benchmark
  205.      * will be able to close/open sockets a zillion of times */
  206.     if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) == -1) {
  207.         anetSetError(err, "setsockopt SO_REUSEADDR: %s", strerror(errno));
  208.         return ANET_ERR;
  209.     }
  210.     return ANET_OK;
  211. }
  212.  
  213. static int anetCreateSocket(char *err, int domain) {
  214.     int s;
  215.     if ((s = socket(domain, SOCK_STREAM, 0)) == -1) {
  216.         anetSetError(err, "creating socket: %s", strerror(errno));
  217.         return ANET_ERR;
  218.     }
  219.  
  220.     /* Make sure connection-intensive things like the redis benchmark
  221.      * will be able to close/open sockets a zillion of times */
  222.     if (anetSetReuseAddr(err,s) == ANET_ERR) {
  223.         close(s);
  224.         return ANET_ERR;
  225.     }
  226.     return s;
  227. }
  228.  
  229. #define ANET_CONNECT_NONE 0
  230. #define ANET_CONNECT_NONBLOCK 1
  231. #define ANET_CONNECT_BE_BINDING 2 /* Best effort binding. */
  232. static int anetTcpGenericConnect(char *err, const char *addr, int port,
  233.                                  const char *source_addr, int flags)
  234. {
  235.     int s = ANET_ERR, rv;
  236.     char portstr[6];  /* strlen("65535") + 1; */
  237.     struct addrinfo hints, *servinfo, *bservinfo, *p, *b;
  238.  
  239.     snprintf(portstr,sizeof(portstr),"%d",port);
  240.     memset(&hints,0,sizeof(hints));
  241.     hints.ai_family = AF_UNSPEC;
  242.     hints.ai_socktype = SOCK_STREAM;
  243.  
  244.     if ((rv = getaddrinfo(addr,portstr,&hints,&servinfo)) != 0) {
  245.         anetSetError(err, "%s", gai_strerror(rv));
  246.         return ANET_ERR;
  247.     }
  248.     for (p = servinfo; p != NULL; p = p->ai_next) {
  249.         /* Try to create the socket and to connect it.
  250.          * If we fail in the socket() call, or on connect(), we retry with
  251.          * the next entry in servinfo. */
  252.         if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1)
  253.             continue;
  254.         if (anetSetReuseAddr(err,s) == ANET_ERR) goto error;
  255.         if (flags & ANET_CONNECT_NONBLOCK && anetNonBlock(err,s) != ANET_OK)
  256.             goto error;
  257.         if (source_addr) {
  258.             int bound = 0;
  259.             /* Using getaddrinfo saves us from self-determining IPv4 vs IPv6 */
  260.             if ((rv = getaddrinfo(source_addr, NULL, &hints, &bservinfo)) != 0)
  261.             {
  262.                 anetSetError(err, "%s", gai_strerror(rv));
  263.                 goto error;
  264.             }
  265.             for (b = bservinfo; b != NULL; b = b->ai_next) {
  266.                 if (bind(s,b->ai_addr,b->ai_addrlen) != -1) {
  267.                     bound = 1;
  268.                     break;
  269.                 }
  270.             }
  271.             freeaddrinfo(bservinfo);
  272.             if (!bound) {
  273.                 anetSetError(err, "bind: %s", strerror(errno));
  274.                 goto error;
  275.             }
  276.         }
  277.         if (connect(s,p->ai_addr,p->ai_addrlen) == -1) {
  278.             /* If the socket is non-blocking, it is ok for connect() to
  279.              * return an EINPROGRESS error here. */
  280.             if (errno == EINPROGRESS && flags & ANET_CONNECT_NONBLOCK)
  281.                 goto end;
  282.             close(s);
  283.             s = ANET_ERR;
  284.             continue;
  285.         }
  286.  
  287.         /* If we ended an iteration of the for loop without errors, we
  288.          * have a connected socket. Let's return to the caller. */
  289.         goto end;
  290.     }
  291.     if (p == NULL)
  292.         anetSetError(err, "creating socket: %s", strerror(errno));
  293.  
  294. error:
  295.     if (s != ANET_ERR) {
  296.         close(s);
  297.         s = ANET_ERR;
  298.     }
  299.  
  300. end:
  301.     freeaddrinfo(servinfo);
  302.  
  303.     /* Handle best effort binding: if a binding address was used, but it is
  304.      * not possible to create a socket, try again without a binding address. */
  305.     if (s == ANET_ERR && source_addr && (flags & ANET_CONNECT_BE_BINDING)) {
  306.         return anetTcpGenericConnect(err,addr,port,NULL,flags);
  307.     } else {
  308.         return s;
  309.     }
  310. }
  311.  
  312. int anetTcpNonBlockConnect(char *err, const char *addr, int port)
  313. {
  314.     return anetTcpGenericConnect(err,addr,port,NULL,ANET_CONNECT_NONBLOCK);
  315. }
  316.  
  317. int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port,
  318.                                          const char *source_addr)
  319. {
  320.     return anetTcpGenericConnect(err,addr,port,source_addr,
  321.             ANET_CONNECT_NONBLOCK|ANET_CONNECT_BE_BINDING);
  322. }
  323.  
  324. int anetUnixGenericConnect(char *err, const char *path, int flags)
  325. {
  326.     int s;
  327.     struct sockaddr_un sa;
  328.  
  329.     if ((s = anetCreateSocket(err,AF_LOCAL)) == ANET_ERR)
  330.         return ANET_ERR;
  331.  
  332.     sa.sun_family = AF_LOCAL;
  333.     strncpy(sa.sun_path,path,sizeof(sa.sun_path)-1);
  334.     if (flags & ANET_CONNECT_NONBLOCK) {
  335.         if (anetNonBlock(err,s) != ANET_OK) {
  336.             close(s);
  337.             return ANET_ERR;
  338.         }
  339.     }
  340.     if (connect(s,(struct sockaddr*)&sa,sizeof(sa)) == -1) {
  341.         if (errno == EINPROGRESS &&
  342.             flags & ANET_CONNECT_NONBLOCK)
  343.             return s;
  344.  
  345.         anetSetError(err, "connect: %s", strerror(errno));
  346.         close(s);
  347.         return ANET_ERR;
  348.     }
  349.     return s;
  350. }
  351.  
  352. static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) {
  353.     if (bind(s,sa,len) == -1) {
  354.         anetSetError(err, "bind: %s", strerror(errno));
  355.         close(s);
  356.         return ANET_ERR;
  357.     }
  358.  
  359.     if (listen(s, backlog) == -1) {
  360.         anetSetError(err, "listen: %s", strerror(errno));
  361.         close(s);
  362.         return ANET_ERR;
  363.     }
  364.     return ANET_OK;
  365. }
  366.  
  367. static int anetV6Only(char *err, int s) {
  368.     int yes = 1;
  369.     if (setsockopt(s,IPPROTO_IPV6,IPV6_V6ONLY,&yes,sizeof(yes)) == -1) {
  370.         anetSetError(err, "setsockopt: %s", strerror(errno));
  371.         return ANET_ERR;
  372.     }
  373.     return ANET_OK;
  374. }
  375.  
  376. static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog)
  377. {
  378.     int s = -1, rv;
  379.     char _port[6];  /* strlen("65535") */
  380.     struct addrinfo hints, *servinfo, *p;
  381.  
  382.     snprintf(_port,6,"%d",port);
  383.     memset(&hints,0,sizeof(hints));
  384.     hints.ai_family = af;
  385.     hints.ai_socktype = SOCK_STREAM;
  386.     hints.ai_flags = AI_PASSIVE;    /* No effect if bindaddr != NULL */
  387.     if (bindaddr && !strcmp("*", bindaddr))
  388.         bindaddr = NULL;
  389.     if (af == AF_INET6 && bindaddr && !strcmp("::*", bindaddr))
  390.         bindaddr = NULL;
  391.  
  392.     if ((rv = getaddrinfo(bindaddr,_port,&hints,&servinfo)) != 0) {
  393.         anetSetError(err, "%s", gai_strerror(rv));
  394.         return ANET_ERR;
  395.     }
  396.     for (p = servinfo; p != NULL; p = p->ai_next) {
  397.         if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1)
  398.             continue;
  399.  
  400.         if (af == AF_INET6 && anetV6Only(err,s) == ANET_ERR) goto error;
  401.         if (anetSetReuseAddr(err,s) == ANET_ERR) goto error;
  402.         if (anetListen(err,s,p->ai_addr,p->ai_addrlen,backlog) == ANET_ERR) s = ANET_ERR;
  403.         goto end;
  404.     }
  405.     if (p == NULL) {
  406.         anetSetError(err, "unable to bind socket, errno: %d", errno);
  407.         goto error;
  408.     }
  409.  
  410. error:
  411.     if (s != -1) close(s);
  412.     s = ANET_ERR;
  413. end:
  414.     freeaddrinfo(servinfo);
  415.     return s;
  416. }
  417.  
  418. int anetTcpServer(char *err, int port, char *bindaddr, int backlog)
  419. {
  420.     return _anetTcpServer(err, port, bindaddr, AF_INET, backlog);
  421. }
  422.  
  423. int anetTcp6Server(char *err, int port, char *bindaddr, int backlog)
  424. {
  425.     return _anetTcpServer(err, port, bindaddr, AF_INET6, backlog);
  426. }
  427.  
  428. int anetUnixServer(char *err, char *path, mode_t perm, int backlog)
  429. {
  430.     int s;
  431.     struct sockaddr_un sa;
  432.  
  433.     if ((s = anetCreateSocket(err,AF_LOCAL)) == ANET_ERR)
  434.         return ANET_ERR;
  435.  
  436.     memset(&sa,0,sizeof(sa));
  437.     sa.sun_family = AF_LOCAL;
  438.     strncpy(sa.sun_path,path,sizeof(sa.sun_path)-1);
  439.     if (anetListen(err,s,(struct sockaddr*)&sa,sizeof(sa),backlog) == ANET_ERR)
  440.         return ANET_ERR;
  441.     if (perm)
  442.         chmod(sa.sun_path, perm);
  443.     return s;
  444. }
  445.  
  446. static int anetGenericAccept(char *err, int s, struct sockaddr *sa, socklen_t *len) {
  447.     int fd;
  448.     while(1) {
  449.         fd = accept(s,sa,len);
  450.         if (fd == -1) {
  451.             if (errno == EINTR)
  452.                 continue;
  453.             else {
  454.                 anetSetError(err, "accept: %s", strerror(errno));
  455.                 return ANET_ERR;
  456.             }
  457.         }
  458.         break;
  459.     }
  460.     return fd;
  461. }
  462.  
  463. int anetTcpAccept(char *err, int s, char *ip, size_t ip_len, int *port) {
  464.     int fd;
  465.     struct sockaddr_storage sa;
  466.     socklen_t salen = sizeof(sa);
  467.     if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == -1)
  468.         return ANET_ERR;
  469.  
  470.     if (sa.ss_family == AF_INET) {
  471.         struct sockaddr_in *s = (struct sockaddr_in *)&sa;
  472.         if (ip) inet_ntop(AF_INET,(void*)&(s->sin_addr),ip,ip_len);
  473.         if (port) *port = ntohs(s->sin_port);
  474.     } else {
  475.         struct sockaddr_in6 *s = (struct sockaddr_in6 *)&sa;
  476.         if (ip) inet_ntop(AF_INET6,(void*)&(s->sin6_addr),ip,ip_len);
  477.         if (port) *port = ntohs(s->sin6_port);
  478.     }
  479.     return fd;
  480. }
  481.  
  482. int anetUnixAccept(char *err, int s) {
  483.     int fd;
  484.     struct sockaddr_un sa;
  485.     socklen_t salen = sizeof(sa);
  486.     if ((fd = anetGenericAccept(err,s,(struct sockaddr*)&sa,&salen)) == -1)
  487.         return ANET_ERR;
  488.  
  489.     return fd;
  490. }
  491.  
  492. int anetFdToString(int fd, char *ip, size_t ip_len, int *port, int fd_to_str_type) {
  493.     struct sockaddr_storage sa;
  494.     socklen_t salen = sizeof(sa);
  495.  
  496.     if (fd_to_str_type == FD_TO_PEER_NAME) {
  497.         if (getpeername(fd, (struct sockaddr *)&sa, &salen) == -1) goto error;
  498.     } else {
  499.         if (getsockname(fd, (struct sockaddr *)&sa, &salen) == -1) goto error;
  500.     }
  501.     if (ip_len == 0) goto error;
  502.  
  503.     if (sa.ss_family == AF_INET) {
  504.         struct sockaddr_in *s = (struct sockaddr_in *)&sa;
  505.         if (ip) inet_ntop(AF_INET,(void*)&(s->sin_addr),ip,ip_len);
  506.         if (port) *port = ntohs(s->sin_port);
  507.     } else if (sa.ss_family == AF_INET6) {
  508.         struct sockaddr_in6 *s = (struct sockaddr_in6 *)&sa;
  509.         if (ip) inet_ntop(AF_INET6,(void*)&(s->sin6_addr),ip,ip_len);
  510.         if (port) *port = ntohs(s->sin6_port);
  511.     } else if (sa.ss_family == AF_UNIX) {
  512.         if (ip) snprintf(ip, ip_len, "/unixsocket");
  513.         if (port) *port = 0;
  514.     } else {
  515.         goto error;
  516.     }
  517.     return 0;
  518.  
  519. error:
  520.     if (ip) {
  521.         if (ip_len >= 2) {
  522.             ip[0] = '?';
  523.             ip[1] = '\0';
  524.         } else if (ip_len == 1) {
  525.             ip[0] = '\0';
  526.         }
  527.     }
  528.     if (port) *port = 0;
  529.     return -1;
  530. }
  531.  
  532. /* Format an IP,port pair into something easy to parse. If IP is IPv6
  533.  * (matches for ":"), the ip is surrounded by []. IP and port are just
  534.  * separated by colons. This the standard to display addresses within Redis. */
  535. int anetFormatAddr(char *buf, size_t buf_len, char *ip, int port) {
  536.     return snprintf(buf,buf_len, strchr(ip,':') ?
  537.            "[%s]:%d" : "%s:%d", ip, port);
  538. }
  539.  
  540. /* Like anetFormatAddr() but extract ip and port from the socket's peer/sockname. */
  541. int anetFormatFdAddr(int fd, char *buf, size_t buf_len, int fd_to_str_type) {
  542.     char ip[INET6_ADDRSTRLEN];
  543.     int port;
  544.  
  545.     anetFdToString(fd,ip,sizeof(ip),&port,fd_to_str_type);
  546.     return anetFormatAddr(buf, buf_len, ip, port);
  547. }

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载