@cryptotaxi247 / netdata-1 / commits / 3efadbbc0

multithreaded web server using poll()

Costa Tsaousis committed May 29, 2016 at 17:50 UTC 3efadbbc0b90c89db732e89f51ae42b80b39bc12
2 files changed +71 -60
src/web_client.c
+57 -42
@@ -16,6 +16,7 @@
16 #include <pwd.h>
17 #include <grp.h>
18 #include <ctype.h>
19 +#include <poll.h>
20
21 #include "common.h"
22 #include "log.h"
@@ -90,7 +91,8 @@ struct web_client *web_client_create(int listener)
91 strcpy(w->client_ip, &w->client_ip[7]);
92 debug(D_WEB_CLIENT_ACCESS, "%llu: New IPv4 web client from %s port %s on socket %d.", w->id, w->client_ip, w->client_port, w->ifd);
93 }
93 - debug(D_WEB_CLIENT_ACCESS, "%llu: New IPv6 web client from %s port %s on socket %d.", w->id, w->client_ip, w->client_port, w->ifd);
94 + else
95 + debug(D_WEB_CLIENT_ACCESS, "%llu: New IPv6 web client from %s port %s on socket %d.", w->id, w->client_ip, w->client_port, w->ifd);
96 break;
97
98 default:
@@ -2103,9 +2105,8 @@ void *web_client_main(void *ptr)
2105 error("Cannot set pthread cancel state to ENABLE.");
2106
2107 struct web_client *w = ptr;
2106 - int retval;
2107 - fd_set ifds, ofds, efds;
2108 - int fdmax = 0;
2108 + struct pollfd fds[2], *ifd, *ofd;
2109 + int retval, fdmax = 0, timeout;
2110
2111 log_access("%llu: %s port %s connected on thread task id %d", w->id, w->client_ip, w->client_port, gettid());
2112
@@ -2119,62 +2120,73 @@ void *web_client_main(void *ptr)
2120 break;
2121 }
2122
2122 - FD_ZERO (&ifds);
2123 - FD_ZERO (&ofds);
2124 - FD_ZERO (&efds);
2125 -
2126 - if(w->ifd < 0 || w->ifd >= FD_SETSIZE || w->ofd < 0 || w->ofd >= FD_SETSIZE) {
2127 - error("%llu: invalid file descriptor, ifd = %d, ofd = %d (required 0 <= fd < FD_SETSIZE (%d)", w->id, w->ifd, w->ofd, FD_SETSIZE);
2123 + if(unlikely(w->ifd < 0 || w->ofd < 0)) {
2124 + error("%llu: invalid file descriptor, ifd = %d, ofd = %d (required 0 <= fd", w->id, w->ifd, w->ofd);
2125 break;
2126 }
2127
2131 - FD_SET(w->ifd, &efds);
2128 + if(w->ifd == w->ofd) {
2129 + fds[0].fd = w->ifd;
2130 + fds[0].events = 0;
2131 + fds[0].revents = 0;
2132
2133 - if(w->ifd != w->ofd)
2134 - FD_SET(w->ofd, &efds);
2133 + if(w->wait_receive) fds[0].events += POLLIN;
2134 + if(w->wait_send) fds[0].events += POLLOUT;
2135
2136 - if (w->wait_receive) {
2137 - FD_SET(w->ifd, &ifds);
2138 - if(w->ifd > fdmax) fdmax = w->ifd;
2139 - }
2136 + fds[1].fd = -1;
2137 + fds[1].events = 0;
2138 + fds[1].revents = 0;
2139
2141 - if (w->wait_send) {
2142 - FD_SET(w->ofd, &ofds);
2143 - if(w->ofd > fdmax) fdmax = w->ofd;
2144 - }
2140 + ifd = ofd = &fds[0];
2141
2146 - debug(D_WEB_CLIENT, "%llu: Waiting socket async I/O for %s %s", w->id, w->wait_receive?"INPUT":"", w->wait_send?"OUTPUT":"");
2147 - struct timeval tv = { .tv_sec = web_client_timeout, .tv_usec = 0 };
2148 - retval = select(fdmax+1, &ifds, &ofds, &efds, &tv);
2149 -
2150 - if(retval == -1) {
2151 - error("%llu: LISTENER: select() failed (input fd = %d, output fd = %d). Closing client.", w->id, w->ifd, w->ofd);
2152 - break;
2142 + fdmax = 1;
2143 }
2154 - else if(!retval) {
2155 - // timeout
2156 - debug(D_WEB_CLIENT_ACCESS, "%llu: LISTENER: timeout.", w->id);
2157 - break;
2144 + else {
2145 + fds[0].fd = w->ifd;
2146 + fds[0].events = 0;
2147 + fds[0].revents = 0;
2148 + if(w->wait_receive) fds[0].events += POLLIN;
2149 + ifd = &fds[0];
2150 +
2151 + fds[1].fd = w->ofd;
2152 + fds[1].events = 0;
2153 + fds[1].revents = 0;
2154 + if(w->wait_send) fds[1].events += POLLOUT;
2155 + ofd = &fds[1];
2156 +
2157 + fdmax = 2;
2158 }
2159
2160 - if(FD_ISSET(w->ifd, &efds)) {
2161 - debug(D_WEB_CLIENT_ACCESS, "%llu: Received error on input socket.", w->id);
2160 + debug(D_WEB_CLIENT, "%llu: Waiting socket async I/O for %s %s", w->id, w->wait_receive?"INPUT":"", w->wait_send?"OUTPUT":"");
2161 + errno = 0;
2162 + timeout = web_client_timeout * 1000;
2163 + retval = poll(fds, fdmax, timeout);
2164 +
2165 + if(unlikely(retval == -1)) {
2166 + if(errno == EAGAIN || errno == EINTR) {
2167 + debug(D_WEB_CLIENT, "%llu: EAGAIN received.", w->id);
2168 + continue;
2169 + }
2170 +
2171 + error("%llu: LISTENER: poll() failed (input fd = %d, output fd = %d). Closing client.", w->id, w->ifd, w->ofd);
2172 break;
2173 }
2164 -
2165 - if(FD_ISSET(w->ofd, &efds)) {
2166 - debug(D_WEB_CLIENT_ACCESS, "%llu: Received error on output socket.", w->id);
2174 + else if(unlikely(!retval)) {
2175 + debug(D_WEB_CLIENT, "%llu: Timeout while waiting socket async I/O for %s %s", w->id, w->wait_receive?"INPUT":"", w->wait_send?"OUTPUT":"");
2176 break;
2177 }
2178
2170 - if(w->wait_send && FD_ISSET(w->ofd, &ofds)) {
2179 + int used = 0;
2180 + if(w->wait_send && ofd->revents & POLLOUT) {
2181 + used++;
2182 if(web_client_send(w) < 0) {
2183 debug(D_WEB_CLIENT, "%llu: Cannot send data to client. Closing client.", w->id);
2184 break;
2185 }
2186 }
2187
2177 - if(w->wait_receive && FD_ISSET(w->ifd, &ifds)) {
2188 + if(w->wait_receive && (ifd->revents & POLLIN || ifd->revents & POLLPRI)) {
2189 + used++;
2190 if(web_client_receive(w) < 0) {
2191 debug(D_WEB_CLIENT, "%llu: Cannot receive data from client. Closing client.", w->id);
2192 break;
@@ -2185,13 +2197,16 @@ void *web_client_main(void *ptr)
2197 web_client_process(w);
2198 }
2199 }
2200 +
2201 + if(unlikely(!used)) {
2202 + debug(D_WEB_CLIENT_ACCESS, "%llu: Received error on socket.", w->id);
2203 + break;
2204 + }
2205 }
2206
2207 log_access("%llu: %s port %s disconnected from thread task id %d", w->id, w->client_ip, w->client_port, gettid());
2208 debug(D_WEB_CLIENT, "%llu: done...", w->id);
2209
2193 - web_client_reset(w);
2194 -
2210 // close the sockets/files now
2211 // to free file descriptors
2212 if(w->ifd == w->ofd) {
@@ -2199,7 +2214,7 @@ void *web_client_main(void *ptr)
2214 }
2215 else {
2216 if(w->ifd != -1) close(w->ifd);
2202 - if(w->ofd != -1) close(w->ifd);
2217 + if(w->ofd != -1) close(w->ofd);
2218 }
2219 w->ifd = -1;
2220 w->ofd = -1;
src/web_server.c
+14 -18
@@ -13,6 +13,7 @@
13 #include <fcntl.h>
14 #include <netinet/tcp.h>
15 #include <malloc.h>
16 +#include <poll.h>
17
18 #include "common.h"
19 #include "log.h"
@@ -205,40 +206,32 @@ int create_listen_socket(void) {
206 #define CLEANUP_EVERY_EVENTS 1000
207
208 void *socket_listen_main_multi_threaded(void *ptr) {
208 - if(ptr) { ; }
209 + (void)ptr;
210
211 web_server_mode = WEB_SERVER_MODE_MULTI_THREADED;
212 info("Multi-threaded WEB SERVER thread created with task id %d", gettid());
213
214 struct web_client *w;
214 - struct timeval tv;
215 int retval, failures = 0, counter = 0;
216
217 - if(ptr) { ; }
218 -
217 if(pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
218 error("Cannot set pthread cancel type to DEFERRED.");
219
220 if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
221 error("Cannot set pthread cancel state to ENABLE.");
222
225 - if(listen_fd < 0 || listen_fd >= FD_SETSIZE) fatal("LISTENER: Listen socket %d is not ready, or invalid.", listen_fd);
226 -
227 - fd_set ifds;
228 - FD_ZERO (&ifds);
223 + if(listen_fd < 0)
224 + fatal("LISTENER: Listen socket %d is not ready, or invalid.", listen_fd);
225
226 for(;;) {
231 - tv.tv_sec = 0;
232 - tv.tv_usec = 100000;
233 -
234 - if(likely(listen_fd >= 0))
235 - FD_SET(listen_fd, &ifds);
227 + struct pollfd fd = { .fd = listen_fd, .events = POLLIN, .revents = 0 };
228 + int timeout = 10 * 1000;
229
230 // debug(D_WEB_CLIENT, "LISTENER: Waiting...");
238 - retval = select(listen_fd + 1, &ifds, NULL, NULL, &tv);
231 + retval = poll(&fd, 1, timeout);
232
233 if(unlikely(retval == -1)) {
241 - error("LISTENER: select() failed.");
234 + error("LISTENER: poll() failed.");
235 failures++;
236
237 if(failures > 10) {
@@ -257,7 +250,7 @@ void *socket_listen_main_multi_threaded(void *ptr) {
250 }
251 else if(likely(retval)) {
252 // check for new incoming connections
260 - if(likely(FD_ISSET(listen_fd, &ifds))) {
253 + if(fd.revents & POLLIN || fd.revents & POLLPRI) {
254 w = web_client_create(listen_fd);
255 if(unlikely(!w)) {
256 // no need for error log - web_client_create already logged the error
@@ -273,8 +266,11 @@ void *socket_listen_main_multi_threaded(void *ptr) {
266 w->obsolete = 1;
267 }
268 }
276 - else debug(D_WEB_CLIENT, "LISTENER: select() didn't do anything.");
277 -
269 + else {
270 + failures++;
271 + debug(D_WEB_CLIENT, "LISTENER: select() didn't do anything.");
272 + continue;
273 + }
274 }
275 else {
276 debug(D_WEB_CLIENT, "LISTENER: select() timeout.");