added single threaded web server
Costa Tsaousis committed
May 28, 2016 at 15:57 UTC
4f63fa6fe9ed7e508f0f74bc014200ae97aba1ca
5 files changed
+454
-163
src/main.c
+29
-36
@@ -70,26 +70,38 @@ struct netdata_static_thread {
70
71
void (*init_routine) (void);
72
void *(*start_routine) (void *);
73
-};
74
-
75
-struct netdata_static_thread static_threads[] = {
76
- {"tc", "plugins", "tc", 1, NULL, NULL, tc_main},
77
- {"idlejitter", "plugins", "idlejitter", 1, NULL, NULL, cpuidlejitter_main},
78
- {"proc", "plugins", "proc", 1, NULL, NULL, proc_main},
79
- {"cgroups", "plugins", "cgroups", 1, NULL, NULL, cgroups_main},
80
-
73
+} static_threads[] = {
74
#ifdef INTERNAL_PLUGIN_NFACCT
82
- // nfacct requires root access
75
+// nfacct requires root access
76
// so, we build it as an external plugin with setuid to root
84
- {"nfacct", "plugins", "nfacct", 1, NULL, NULL, nfacct_main},
77
+ {"nfacct", "plugins", "nfacct", 1, NULL, NULL, nfacct_main},
78
#endif
79
87
- {"plugins.d", NULL, NULL, 1, NULL, NULL, pluginsd_main},
88
- {"check", "plugins", "checks", 0, NULL, NULL, checks_main},
89
- {"web", NULL, NULL, 1, NULL, NULL, socket_listen_main},
90
- {NULL, NULL, NULL, 0, NULL, NULL, NULL}
80
+ {"tc", "plugins", "tc", 1, NULL, NULL, tc_main},
81
+ {"idlejitter", "plugins", "idlejitter", 1, NULL, NULL, cpuidlejitter_main},
82
+ {"proc", "plugins", "proc", 1, NULL, NULL, proc_main},
83
+ {"cgroups", "plugins", "cgroups", 1, NULL, NULL, cgroups_main},
84
+ {"plugins.d", NULL, NULL, 1, NULL, NULL, pluginsd_main},
85
+ {"check", "plugins", "checks", 0, NULL, NULL, checks_main},
86
+ {"web", NULL, NULL, 1, NULL, NULL, socket_listen_main_multi_threaded},
87
+ {"web-single-threaded", NULL, NULL, 0, NULL, NULL, socket_listen_main_single_threaded},
88
+ {NULL, NULL, NULL, 0, NULL, NULL, NULL}
89
};
90
91
+void web_server_threading_selection(void) {
92
+ int threaded = config_get_boolean("global", "multi threaded web server", 1);
93
+
94
+ int i;
95
+ for(i = 0; static_threads[i].name ; i++) {
96
+ if(static_threads[i].start_routine == socket_listen_main_multi_threaded)
97
+ static_threads[i].enabled = threaded?1:0;
98
+
99
+ if(static_threads[i].start_routine == socket_listen_main_single_threaded)
100
+ static_threads[i].enabled = threaded?0:1;
101
+ }
102
+}
103
+
104
+
105
int killpid(pid_t pid, int sig)
106
{
107
int ret = -1;
@@ -591,28 +603,7 @@ int main(int argc, char **argv)
603
604
// --------------------------------------------------------------------
605
594
- listen_backlog = (int) config_get_number("global", "http port listen backlog", LISTEN_BACKLOG);
595
-
596
- listen_port = (int) config_get_number("global", "port", LISTEN_PORT);
597
- if(listen_port < 1 || listen_port > 65535) {
598
- info("Invalid listen port %d given. Defaulting to %d.", listen_port, LISTEN_PORT);
599
- listen_port = LISTEN_PORT;
600
- }
601
- else debug(D_OPTIONS, "Listen port set to %d.", listen_port);
602
-
603
- int ip = 0;
604
- char *ipv = config_get("global", "ip version", "any");
605
- if(!strcmp(ipv, "any") || !strcmp(ipv, "both") || !strcmp(ipv, "all")) ip = 0;
606
- else if(!strcmp(ipv, "ipv4") || !strcmp(ipv, "IPV4") || !strcmp(ipv, "IPv4") || !strcmp(ipv, "4")) ip = 4;
607
- else if(!strcmp(ipv, "ipv6") || !strcmp(ipv, "IPV6") || !strcmp(ipv, "IPv6") || !strcmp(ipv, "6")) ip = 6;
608
- else info("Cannot understand ip version '%s'. Assuming 'any'.", ipv);
609
-
610
- if(ip == 0 || ip == 6) listen_fd = create_listen_socket6(config_get("global", "bind socket to IP", "*"), listen_port, listen_backlog);
611
- if(listen_fd < 0) {
612
- listen_fd = create_listen_socket4(config_get("global", "bind socket to IP", "*"), listen_port, listen_backlog);
613
- if(listen_fd >= 0 && ip != 4) info("Managed to open an IPv4 socket on port %d.", listen_port);
614
- }
615
-
606
+ listen_fd = create_listen_socket();
607
if(listen_fd < 0) fatal("Cannot listen socket.");
608
}
609
@@ -656,6 +647,8 @@ int main(int argc, char **argv)
647
// ------------------------------------------------------------------------
648
// spawn the threads
649
650
+ web_server_threading_selection();
651
+
652
for (i = 0; static_threads[i].name != NULL ; i++) {
653
struct netdata_static_thread *st = &static_threads[i];
654
src/web_client.c
+145
-84
@@ -159,16 +159,16 @@ void web_client_reset(struct web_client *w)
159
w->stats_received_bytes = 0;
160
w->stats_sent_bytes = 0;
161
162
- long sent = (w->mode == WEB_CLIENT_MODE_FILECOPY)?w->response.rlen:w->response.data->len;
162
+ size_t sent = (w->mode == WEB_CLIENT_MODE_FILECOPY)?w->response.rlen:w->response.data->len;
163
164
#ifdef NETDATA_WITH_ZLIB
165
- if(likely(w->response.zoutput)) sent = (long)w->response.zstream.total_out;
165
+ if(likely(w->response.zoutput)) sent = (size_t)w->response.zstream.total_out;
166
#endif
167
168
- long size = (w->mode == WEB_CLIENT_MODE_FILECOPY)?w->response.rlen:w->response.data->len;
168
+ size_t size = (w->mode == WEB_CLIENT_MODE_FILECOPY)?w->response.rlen:w->response.data->len;
169
170
if(likely(w->last_url[0]))
171
- log_access("%llu: (sent/all = %ld/%ld bytes %0.0f%%, prep/sent/total = %0.2f/%0.2f/%0.2f ms) %s: %d '%s'",
171
+ log_access("%llu: (sent/all = %zu/%zu bytes %0.0f%%, prep/sent/total = %0.2f/%0.2f/%0.2f ms) %s: %d '%s'",
172
w->id,
173
sent, size, -((size>0)?((float)(size-sent)/(float)size * 100.0):0.0),
174
(float)usecdiff(&w->tv_ready, &w->tv_in) / 1000.0,
@@ -180,9 +180,11 @@ void web_client_reset(struct web_client *w)
180
);
181
182
if(unlikely(w->mode == WEB_CLIENT_MODE_FILECOPY)) {
183
- debug(D_WEB_CLIENT, "%llu: Closing filecopy input file.", w->id);
184
- close(w->ifd);
185
- w->ifd = w->ofd;
183
+ if(w->ifd != w->ofd) {
184
+ debug(D_WEB_CLIENT, "%llu: Closing filecopy input file descriptor %d.", w->id, w->ifd);
185
+ close(w->ifd);
186
+ w->ifd = w->ofd;
187
+ }
188
}
189
190
w->last_url[0] = '\0';
@@ -192,6 +194,7 @@ void web_client_reset(struct web_client *w)
194
w->origin[1] = '\0';
195
196
w->mode = WEB_CLIENT_MODE_NORMAL;
197
+
198
w->keepalive = 0;
199
w->decoded_url[0] = '\0';
200
@@ -226,19 +229,17 @@ void web_client_reset(struct web_client *w)
229
struct web_client *web_client_free(struct web_client *w)
230
{
231
struct web_client *n = w->next;
232
+ if(w == web_clients) web_clients = n;
233
234
debug(D_WEB_CLIENT_ACCESS, "%llu: Closing web client from %s port %s.", w->id, w->client_ip, w->client_port);
235
236
if(w->prev) w->prev->next = w->next;
237
if(w->next) w->next->prev = w->prev;
234
-
235
- if(w == web_clients) web_clients = w->next;
236
-
238
if(w->response.header_output) buffer_free(w->response.header_output);
239
if(w->response.header) buffer_free(w->response.header);
240
if(w->response.data) buffer_free(w->response.data);
240
- close(w->ifd);
241
- if(w->ofd != w->ifd) close(w->ofd);
241
+ if(w->ifd != -1) close(w->ifd);
242
+ if(w->ofd != -1 && w->ofd != w->ifd) close(w->ofd);
243
free(w);
244
245
global_statistics.connected_clients--;
@@ -425,7 +426,7 @@ int mysendfile(struct web_client *w, char *filename)
426
427
#ifdef NETDATA_WITH_ZLIB
428
void web_client_enable_deflate(struct web_client *w, int gzip) {
428
- if(w->response.zinitialized == 1) {
429
+ if(w->response.zinitialized) {
430
error("%llu: Compression has already be initialized for this client.", w->id);
431
return;
432
}
@@ -1715,11 +1716,11 @@ void web_client_process(struct web_client *w) {
1716
1717
buffer_strcat(w->response.header_output, "\r\n");
1718
1718
- // disable TCP_NODELAY, to buffer the header
1719
+/* // disable TCP_NODELAY, to buffer the header
1720
int flag = 0;
1721
if(setsockopt(w->ofd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)) != 0)
1722
error("%llu: failed to disable TCP_NODELAY on socket.", w->id);
1722
-
1723
+*/
1724
// sent the HTTP header
1725
debug(D_WEB_DATA, "%llu: Sending response HTTP header of size %d: '%s'"
1726
, w->id
@@ -1728,18 +1729,23 @@ void web_client_process(struct web_client *w) {
1729
);
1730
1731
bytes = send(w->ofd, buffer_tostring(w->response.header_output), buffer_strlen(w->response.header_output), 0);
1731
- if(bytes != (ssize_t) buffer_strlen(w->response.header_output))
1732
- error("%llu: HTTP Header failed to be sent (I sent %d bytes but the system sent %d bytes)."
1733
- , w->id
1734
- , buffer_strlen(w->response.header_output)
1735
- , bytes);
1732
+ if(bytes != (ssize_t) buffer_strlen(w->response.header_output)) {
1733
+ if(bytes > 0)
1734
+ w->stats_sent_bytes += bytes;
1735
+
1736
+ error("%llu: HTTP Header failed to be sent (I sent %d bytes but the system sent %d bytes). Closing web client.", w->id,
1737
+ buffer_strlen(w->response.header_output), bytes);
1738
+
1739
+ WEB_CLIENT_IS_DEAD(w);
1740
+ return;
1741
+ }
1742
else
1743
w->stats_sent_bytes += bytes;
1744
1739
- // enable TCP_NODELAY, to send all data immediately at the next send()
1745
+/* // enable TCP_NODELAY, to send all data immediately at the next send()
1746
flag = 1;
1747
if(setsockopt(w->ofd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int)) != 0) error("%llu: failed to enable TCP_NODELAY on socket.", w->id);
1742
-
1748
+*/
1749
// enable sending immediately if we have data
1750
if(w->response.data->len) w->wait_send = 1;
1751
else w->wait_send = 0;
@@ -1780,50 +1786,78 @@ void web_client_process(struct web_client *w) {
1786
}
1787
}
1788
1783
-long web_client_send_chunk_header(struct web_client *w, long len)
1789
+ssize_t web_client_send_chunk_header(struct web_client *w, size_t len)
1790
{
1791
debug(D_DEFLATE, "%llu: OPEN CHUNK of %d bytes (hex: %x).", w->id, len, len);
1792
char buf[1024];
1793
sprintf(buf, "%lX\r\n", len);
1788
- ssize_t bytes = send(w->ofd, buf, strlen(buf), MSG_DONTWAIT);
1794
+
1795
+ ssize_t bytes = send(w->ofd, buf, strlen(buf), 0);
1796
+ if(bytes > 0) {
1797
+ debug(D_DEFLATE, "%llu: Sent chunk header %d bytes.", w->id, bytes);
1798
+ w->stats_sent_bytes += bytes;
1799
+ }
1800
1790
- if(bytes > 0) debug(D_DEFLATE, "%llu: Sent chunk header %d bytes.", w->id, bytes);
1791
- else if(bytes == 0) debug(D_DEFLATE, "%llu: Did not send chunk header to the client.", w->id);
1792
- else debug(D_DEFLATE, "%llu: Failed to send chunk header to client.", w->id);
1801
+ else if(bytes == 0) {
1802
+ error("%llu: Did not send chunk header to the client.", w->id);
1803
+ WEB_CLIENT_IS_DEAD(w);
1804
+ }
1805
+ else {
1806
+ error("%llu: Failed to send chunk header to client.", w->id);
1807
+ WEB_CLIENT_IS_DEAD(w);
1808
+ }
1809
1810
return bytes;
1811
}
1812
1797
-long web_client_send_chunk_close(struct web_client *w)
1813
+ssize_t web_client_send_chunk_close(struct web_client *w)
1814
{
1815
//debug(D_DEFLATE, "%llu: CLOSE CHUNK.", w->id);
1816
1801
- ssize_t bytes = send(w->ofd, "\r\n", 2, MSG_DONTWAIT);
1817
+ ssize_t bytes = send(w->ofd, "\r\n", 2, 0);
1818
+ if(bytes > 0) {
1819
+ debug(D_DEFLATE, "%llu: Sent chunk suffix %d bytes.", w->id, bytes);
1820
+ w->stats_sent_bytes += bytes;
1821
+ }
1822
1803
- if(bytes > 0) debug(D_DEFLATE, "%llu: Sent chunk suffix %d bytes.", w->id, bytes);
1804
- else if(bytes == 0) debug(D_DEFLATE, "%llu: Did not send chunk suffix to the client.", w->id);
1805
- else debug(D_DEFLATE, "%llu: Failed to send chunk suffix to client.", w->id);
1823
+ else if(bytes == 0) {
1824
+ error("%llu: Did not send chunk suffix to the client.", w->id);
1825
+ WEB_CLIENT_IS_DEAD(w);
1826
+ }
1827
+ else {
1828
+ error("%llu: Failed to send chunk suffix to client.", w->id);
1829
+ WEB_CLIENT_IS_DEAD(w);
1830
+ }
1831
1832
return bytes;
1833
}
1834
1810
-long web_client_send_chunk_finalize(struct web_client *w)
1835
+ssize_t web_client_send_chunk_finalize(struct web_client *w)
1836
{
1837
//debug(D_DEFLATE, "%llu: FINALIZE CHUNK.", w->id);
1838
1814
- ssize_t bytes = send(w->ofd, "\r\n0\r\n\r\n", 7, MSG_DONTWAIT);
1839
+ ssize_t bytes = send(w->ofd, "\r\n0\r\n\r\n", 7, 0);
1840
+ if(bytes > 0) {
1841
+ debug(D_DEFLATE, "%llu: Sent chunk suffix %d bytes.", w->id, bytes);
1842
+ w->stats_sent_bytes += bytes;
1843
+ }
1844
1816
- if(bytes > 0) debug(D_DEFLATE, "%llu: Sent chunk suffix %d bytes.", w->id, bytes);
1817
- else if(bytes == 0) debug(D_DEFLATE, "%llu: Did not send chunk suffix to the client.", w->id);
1818
- else debug(D_DEFLATE, "%llu: Failed to send chunk suffix to client.", w->id);
1845
+ else if(bytes == 0) {
1846
+ error("%llu: Did not send chunk suffix to the client.", w->id);
1847
+ WEB_CLIENT_IS_DEAD(w);
1848
+ }
1849
+ else {
1850
+ error("%llu: Failed to send chunk suffix to client.", w->id);
1851
+ WEB_CLIENT_IS_DEAD(w);
1852
+ }
1853
1854
return bytes;
1855
}
1856
1857
#ifdef NETDATA_WITH_ZLIB
1824
-long web_client_send_deflate(struct web_client *w)
1858
+ssize_t web_client_send_deflate(struct web_client *w)
1859
{
1826
- long len = 0, t = 0;
1860
+ ssize_t len = 0, t = 0;
1861
1862
// when using compression,
1863
// w->response.sent is the amount of bytes passed through compression
@@ -1836,21 +1870,19 @@ long web_client_send_deflate(struct web_client *w)
1870
debug(D_WEB_CLIENT, "%llu: Out of output data.", w->id);
1871
1872
// finalize the chunk
1839
- if(w->response.sent != 0)
1840
- t += web_client_send_chunk_finalize(w);
1841
-
1842
- // there can be two cases for this
1843
- // A. we have done everything
1844
- // B. we temporarily have nothing to send, waiting for the buffer to be filled by ifd
1873
+ if(w->response.sent != 0) {
1874
+ t = web_client_send_chunk_finalize(w);
1875
+ if(t < 0) return t;
1876
+ }
1877
1846
- if(w->mode == WEB_CLIENT_MODE_FILECOPY && w->wait_receive && w->ifd != w->ofd && w->response.rlen && w->response.rlen > w->response.data->len) {
1878
+ if(w->mode == WEB_CLIENT_MODE_FILECOPY && w->wait_receive && w->response.rlen && w->response.rlen > w->response.data->len) {
1879
// we have to wait, more data will come
1880
debug(D_WEB_CLIENT, "%llu: Waiting for more data to become available.", w->id);
1881
w->wait_send = 0;
1850
- return(0);
1882
+ return t;
1883
}
1884
1853
- if(w->keepalive == 0) {
1885
+ if(unlikely(w->keepalive == 0)) {
1886
debug(D_WEB_CLIENT, "%llu: Closing (keep-alive is not enabled). %ld bytes sent.", w->id, w->response.sent);
1887
errno = 0;
1888
return(-1);
@@ -1858,24 +1890,23 @@ long web_client_send_deflate(struct web_client *w)
1890
1891
// reset the client
1892
web_client_reset(w);
1861
- debug(D_WEB_CLIENT, "%llu: Done sending all data on socket. Waiting for next request on the same socket.", w->id);
1862
- return(0);
1893
+ debug(D_WEB_CLIENT, "%llu: Done sending all data on socket.", w->id);
1894
+ return t;
1895
}
1896
1897
if(w->response.zhave == w->response.zsent) {
1898
// compress more input data
1899
1900
// close the previous open chunk
1869
- if(w->response.sent != 0) t += web_client_send_chunk_close(w);
1901
+ if(w->response.sent != 0) {
1902
+ t = web_client_send_chunk_close(w);
1903
+ if(t < 0) return t;
1904
+ }
1905
1906
debug(D_DEFLATE, "%llu: Compressing %d new bytes starting from %d (and %d left behind).", w->id, (w->response.data->len - w->response.sent), w->response.sent, w->response.zstream.avail_in);
1907
1908
// give the compressor all the data not passed through the compressor yet
1909
if(w->response.data->len > w->response.sent) {
1875
-#ifdef NETDATA_INTERNAL_CHECKS
1876
- if((long)w->response.sent - (long)w->response.zstream.avail_in < 0)
1877
- error("internal error: avail_in is corrupted.");
1878
-#endif
1910
w->response.zstream.next_in = (Bytef *)&w->response.data->buffer[w->response.sent - w->response.zstream.avail_in];
1911
w->response.zstream.avail_in += (uInt) (w->response.data->len - w->response.sent);
1912
}
@@ -1911,31 +1942,39 @@ long web_client_send_deflate(struct web_client *w)
1942
debug(D_DEFLATE, "%llu: Compression produced %d bytes.", w->id, w->response.zhave);
1943
1944
// open a new chunk
1914
- t += web_client_send_chunk_header(w, w->response.zhave);
1945
+ ssize_t t2 = web_client_send_chunk_header(w, w->response.zhave);
1946
+ if(t2 < 0) return t2;
1947
+ t += t2;
1948
}
1949
1950
debug(D_WEB_CLIENT, "%llu: Sending %d bytes of data (+%d of chunk header).", w->id, w->response.zhave - w->response.zsent, t);
1951
1952
len = send(w->ofd, &w->response.zbuffer[w->response.zsent], (size_t) (w->response.zhave - w->response.zsent), MSG_DONTWAIT);
1953
if(len > 0) {
1954
+ w->stats_sent_bytes += len;
1955
w->response.zsent += len;
1922
- if(t > 0) len += t;
1956
+ len += t;
1957
debug(D_WEB_CLIENT, "%llu: Sent %d bytes.", w->id, len);
1958
}
1925
- else if(len == 0) debug(D_WEB_CLIENT, "%llu: Did not send any bytes to the client (zhave = %ld, zsent = %ld, need to send = %ld).", w->id, w->response.zhave, w->response.zsent, w->response.zhave - w->response.zsent);
1926
- else debug(D_WEB_CLIENT, "%llu: Failed to send data to client. Reason: %s", w->id, strerror(errno));
1959
+ else if(len == 0) {
1960
+ error("%llu: Did not send any bytes to the client (zhave = %ld, zsent = %ld, need to send = %ld).", w->id, w->response.zhave, w->response.zsent, w->response.zhave - w->response.zsent);
1961
+ WEB_CLIENT_IS_DEAD(w);
1962
+ }
1963
+ else {
1964
+ error("%llu: Failed to send data to client.", w->id);
1965
+ WEB_CLIENT_IS_DEAD(w);
1966
+ }
1967
1968
return(len);
1969
}
1970
#endif // NETDATA_WITH_ZLIB
1971
1932
-long web_client_send(struct web_client *w)
1933
-{
1972
+ssize_t web_client_send(struct web_client *w) {
1973
#ifdef NETDATA_WITH_ZLIB
1974
if(likely(w->response.zoutput)) return web_client_send_deflate(w);
1975
#endif // NETDATA_WITH_ZLIB
1976
1938
- long bytes;
1977
+ ssize_t bytes;
1978
1979
if(unlikely(w->response.data->len - w->response.sent == 0)) {
1980
// there is nothing to send
@@ -1946,7 +1985,7 @@ long web_client_send(struct web_client *w)
1985
// A. we have done everything
1986
// B. we temporarily have nothing to send, waiting for the buffer to be filled by ifd
1987
1949
- if(w->mode == WEB_CLIENT_MODE_FILECOPY && w->wait_receive && w->ifd != w->ofd && w->response.rlen && w->response.rlen > w->response.data->len) {
1988
+ if(w->mode == WEB_CLIENT_MODE_FILECOPY && w->wait_receive && w->response.rlen && w->response.rlen > w->response.data->len) {
1989
// we have to wait, more data will come
1990
debug(D_WEB_CLIENT, "%llu: Waiting for more data to become available.", w->id);
1991
w->wait_send = 0;
@@ -1966,22 +2005,29 @@ long web_client_send(struct web_client *w)
2005
2006
bytes = send(w->ofd, &w->response.data->buffer[w->response.sent], w->response.data->len - w->response.sent, MSG_DONTWAIT);
2007
if(likely(bytes > 0)) {
2008
+ w->stats_sent_bytes += bytes;
2009
w->response.sent += bytes;
2010
debug(D_WEB_CLIENT, "%llu: Sent %d bytes.", w->id, bytes);
2011
}
1972
- else if(likely(bytes == 0)) debug(D_WEB_CLIENT, "%llu: Did not send any bytes to the client.", w->id);
1973
- else debug(D_WEB_CLIENT, "%llu: Failed to send data to client.", w->id);
2012
+ else if(likely(bytes == 0)) {
2013
+ error("%llu: Did not send any bytes to the client.", w->id);
2014
+ WEB_CLIENT_IS_DEAD(w);
2015
+ }
2016
+ else {
2017
+ error("%llu: Failed to send data to client.", w->id);
2018
+ WEB_CLIENT_IS_DEAD(w);
2019
+ }
2020
2021
return(bytes);
2022
}
2023
1978
-long web_client_receive(struct web_client *w)
2024
+ssize_t web_client_receive(struct web_client *w)
2025
{
2026
// do we have any space for more data?
2027
buffer_need_bytes(w->response.data, WEB_REQUEST_LENGTH);
2028
1983
- long left = w->response.data->size - w->response.data->len;
1984
- long bytes;
2029
+ ssize_t left = w->response.data->size - w->response.data->len;
2030
+ ssize_t bytes;
2031
2032
if(unlikely(w->mode == WEB_CLIENT_MODE_FILECOPY))
2033
bytes = read(w->ifd, &w->response.data->buffer[w->response.data->len], (size_t) (left - 1));
@@ -1989,6 +2035,9 @@ long web_client_receive(struct web_client *w)
2035
bytes = recv(w->ifd, &w->response.data->buffer[w->response.data->len], (size_t) (left - 1), MSG_DONTWAIT);
2036
2037
if(likely(bytes > 0)) {
2038
+ if(w->mode != WEB_CLIENT_MODE_FILECOPY)
2039
+ w->stats_received_bytes += bytes;
2040
+
2041
size_t old = w->response.data->len;
2042
w->response.data->len += bytes;
2043
w->response.data->buffer[w->response.data->len] = '\0';
@@ -1998,7 +2047,9 @@ long web_client_receive(struct web_client *w)
2047
2048
if(w->mode == WEB_CLIENT_MODE_FILECOPY) {
2049
w->wait_send = 1;
2001
- if(w->response.rlen && w->response.data->len >= w->response.rlen) w->wait_receive = 0;
2050
+
2051
+ if(w->response.rlen && w->response.data->len >= w->response.rlen)
2052
+ w->wait_receive = 0;
2053
}
2054
}
2055
else if(likely(bytes == 0)) {
@@ -2012,13 +2063,20 @@ long web_client_receive(struct web_client *w)
2063
// we are copying data from ifd to ofd
2064
// let it finish copying...
2065
w->wait_receive = 0;
2015
- debug(D_WEB_CLIENT, "%llu: Disabling input.", w->id);
2066
+
2067
+ debug(D_WEB_CLIENT, "%llu: Read the whole file.", w->id);
2068
+ if(w->ifd != w->ofd) close(w->ifd);
2069
+ w->ifd = w->ofd;
2070
}
2071
else {
2018
- bytes = -1;
2019
- errno = 0;
2072
+ error("%llu: failed to receive data.", w->id);
2073
+ WEB_CLIENT_IS_DEAD(w);
2074
}
2075
}
2076
+ else {
2077
+ error("%llu: receive data failed.", w->id);
2078
+ WEB_CLIENT_IS_DEAD(w);
2079
+ }
2080
2081
return(bytes);
2082
}
@@ -2040,7 +2098,6 @@ void *web_client_main(void *ptr)
2098
if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
2099
error("Cannot set pthread cancel state to ENABLE.");
2100
2043
- struct timeval tv;
2101
struct web_client *w = ptr;
2102
int retval;
2103
fd_set ifds, ofds, efds;
@@ -2049,6 +2106,15 @@ void *web_client_main(void *ptr)
2106
log_access("%llu: %s port %s connected on thread task id %d", w->id, w->client_ip, w->client_port, gettid());
2107
2108
for(;;) {
2109
+ if(unlikely(w->dead)) {
2110
+ error("%llu: client is dead.");
2111
+ break;
2112
+ }
2113
+ else if(unlikely(!w->wait_receive && !w->wait_send)) {
2114
+ error("%llu: client is not set for neither receiving nor sending data.");
2115
+ break;
2116
+ }
2117
+
2118
FD_ZERO (&ifds);
2119
FD_ZERO (&ofds);
2120
FD_ZERO (&efds);
@@ -2068,15 +2134,13 @@ void *web_client_main(void *ptr)
2134
if(w->ofd > fdmax) fdmax = w->ofd;
2135
}
2136
2071
- tv.tv_sec = web_client_timeout;
2072
- tv.tv_usec = 0;
2073
-
2137
debug(D_WEB_CLIENT, "%llu: Waiting socket async I/O for %s %s", w->id, w->wait_receive?"INPUT":"", w->wait_send?"OUTPUT":"");
2138
+ struct timeval tv = { .tv_sec = web_client_timeout, .tv_usec = 0 };
2139
retval = select(fdmax+1, &ifds, &ofds, &efds, &tv);
2140
2141
if(retval == -1) {
2078
- debug(D_WEB_CLIENT_ACCESS, "%llu: LISTENER: select() failed.", w->id);
2079
- continue;
2142
+ error("%llu: LISTENER: select() failed (input fd = %d, output fd = %d). Closing client.", w->id, w->ifd, w->ofd);
2143
+ break;
2144
}
2145
else if(!retval) {
2146
// timeout
@@ -2095,25 +2159,21 @@ void *web_client_main(void *ptr)
2159
}
2160
2161
if(w->wait_send && FD_ISSET(w->ofd, &ofds)) {
2098
- long bytes;
2162
+ ssize_t bytes;
2163
if((bytes = web_client_send(w)) < 0) {
2164
debug(D_WEB_CLIENT, "%llu: Cannot send data to client. Closing client.", w->id);
2165
errno = 0;
2166
break;
2167
}
2104
- else
2105
- w->stats_sent_bytes += bytes;
2168
}
2169
2170
if(w->wait_receive && FD_ISSET(w->ifd, &ifds)) {
2109
- long bytes;
2171
+ ssize_t bytes;
2172
if((bytes = web_client_receive(w)) < 0) {
2173
debug(D_WEB_CLIENT, "%llu: Cannot receive data from client. Closing client.", w->id);
2174
errno = 0;
2175
break;
2176
}
2115
- else
2116
- w->stats_received_bytes += bytes;
2177
2178
if(w->mode == WEB_CLIENT_MODE_NORMAL) {
2179
debug(D_WEB_CLIENT, "%llu: Attempting to process received data (%ld bytes).", w->id, bytes);
@@ -2129,6 +2189,7 @@ void *web_client_main(void *ptr)
2189
2190
web_client_reset(w);
2191
w->obsolete = 1;
2192
+ pthread_exit(NULL);
2193
2194
return NULL;
2195
}
src/web_client.h
+30
-19
@@ -37,16 +37,16 @@ struct response {
37
38
int code; // the HTTP response code
39
40
- size_t rlen; // if non-zero, the excepted size of ifd (input)
40
+ size_t rlen; // if non-zero, the excepted size of ifd (input of firecopy)
41
size_t sent; // current data length sent to output
42
43
int zoutput; // if set to 1, web_client_send() will send compressed data
44
#ifdef NETDATA_WITH_ZLIB
45
z_stream zstream; // zlib stream for sending compressed output to client
46
Bytef zbuffer[ZLIB_CHUNK]; // temporary buffer for storing compressed output
47
- long zsent; // the compressed bytes we have sent to the client
48
- long zhave; // the compressed bytes that we have to send
49
- int zinitialized;
47
+ size_t zsent; // the compressed bytes we have sent to the client
48
+ size_t zhave; // the compressed bytes that we have received from zlib
49
+ int zinitialized:1;
50
#endif
51
52
};
@@ -54,6 +54,23 @@ struct response {
54
struct web_client {
55
unsigned long long id;
56
57
+ uint8_t obsolete:1; // if set to 1, the listener will remove this client
58
+ // after setting this to 1, you should not touch
59
+ // this web_client
60
+
61
+ uint8_t dead:1; // if set to 1, this client is dead
62
+
63
+ uint8_t keepalive:1; // if set to 1, the web client will be re-used
64
+ uint8_t enable_gzip:1; // if set to 1, the response will be compressed
65
+
66
+ uint8_t mode:3; // the operational mode of the client
67
+
68
+ uint8_t wait_receive:1; // 1 = we are waiting more input data
69
+ uint8_t wait_send:1; // 1 = we have data to send to the client
70
+
71
+ int ifd;
72
+ int ofd;
73
+
74
char client_ip[NI_MAXHOST+1];
75
char client_port[NI_MAXSERV+1];
76
@@ -66,30 +83,20 @@ struct web_client {
83
char cookie2[COOKIE_MAX+1];
84
char origin[ORIGIN_MAX+1];
85
69
- int mode;
70
- int keepalive;
71
- int enable_gzip;
72
-
86
struct sockaddr_storage clientaddr;
74
-
75
- pthread_t thread; // the thread servicing this client
76
- int obsolete; // if set to 1, the listener will remove this client
77
-
78
- int ifd;
79
- int ofd;
80
-
87
struct response response;
88
83
- int wait_receive;
84
- int wait_send;
89
+ size_t stats_received_bytes;
90
+ size_t stats_sent_bytes;
91
86
- unsigned long stats_received_bytes;
87
- unsigned long stats_sent_bytes;
92
+ pthread_t thread; // the thread servicing this client
93
94
struct web_client *prev;
95
struct web_client *next;
96
};
97
98
+#define WEB_CLIENT_IS_DEAD(w) (w)->dead=1
99
+
100
extern struct web_client *web_clients;
101
102
extern uid_t web_files_uid(void);
@@ -97,6 +104,10 @@ extern uid_t web_files_gid(void);
104
105
extern struct web_client *web_client_create(int listener);
106
extern struct web_client *web_client_free(struct web_client *w);
107
+extern ssize_t web_client_send(struct web_client *w);
108
+extern ssize_t web_client_receive(struct web_client *w);
109
+extern void web_client_process(struct web_client *w);
110
+extern void web_client_reset(struct web_client *w);
111
112
extern void *web_client_main(void *ptr);
113
src/web_server.c
+247
-23
@@ -86,7 +86,7 @@ int create_listen_socket4(const char *ip, int port, int listen_backlog)
86
87
if(is_ip_anything(ip)) {
88
name.sin_addr.s_addr = htonl(INADDR_ANY);
89
- info("Listening on any IPs (IPv4).");
89
+ // info("Listening on any IPs (IPv4).");
90
}
91
else {
92
int ret = inet_pton(AF_INET, ip, (void *)&name.sin_addr.s_addr);
@@ -95,7 +95,7 @@ int create_listen_socket4(const char *ip, int port, int listen_backlog)
95
close(sock);
96
return -1;
97
}
98
- info("Listening on IP '%s' (IPv4).", ip);
98
+ // info("Listening on IP '%s' (IPv4).", ip);
99
}
100
101
if(bind (sock, (struct sockaddr *) &name, sizeof (name)) < 0) {
@@ -137,7 +137,7 @@ int create_listen_socket6(const char *ip, int port, int listen_backlog)
137
138
if(is_ip_anything(ip)) {
139
name.sin6_addr = in6addr_any;
140
- info("Listening on all IPs (IPv6 and IPv4)");
140
+ // info("Listening on all IPs (IPv6 and IPv4)");
141
}
142
else {
143
int ret = inet_pton(AF_INET6, ip, (void *)&name.sin6_addr.s6_addr);
@@ -146,7 +146,7 @@ int create_listen_socket6(const char *ip, int port, int listen_backlog)
146
close(sock);
147
return -1;
148
}
149
- info("Listening on IP '%s' (IPv6)", ip);
149
+ // info("Listening on IP '%s' (IPv6)", ip);
150
}
151
152
name.sin6_scope_id = 0;
@@ -168,6 +168,32 @@ int create_listen_socket6(const char *ip, int port, int listen_backlog)
168
}
169
170
171
+int create_listen_socket(void) {
172
+ listen_backlog = (int) config_get_number("global", "http port listen backlog", LISTEN_BACKLOG);
173
+
174
+ listen_port = (int) config_get_number("global", "port", LISTEN_PORT);
175
+ if(listen_port < 1 || listen_port > 65535) {
176
+ error("Invalid listen port %d given. Defaulting to %d.", listen_port, LISTEN_PORT);
177
+ listen_port = LISTEN_PORT;
178
+ }
179
+ else debug(D_OPTIONS, "Listen port set to %d.", listen_port);
180
+
181
+ int ip = 0;
182
+ char *ipv = config_get("global", "ip version", "any");
183
+ if(!strcmp(ipv, "any") || !strcmp(ipv, "both") || !strcmp(ipv, "all")) ip = 0;
184
+ else if(!strcmp(ipv, "ipv4") || !strcmp(ipv, "IPV4") || !strcmp(ipv, "IPv4") || !strcmp(ipv, "4")) ip = 4;
185
+ else if(!strcmp(ipv, "ipv6") || !strcmp(ipv, "IPV6") || !strcmp(ipv, "IPv6") || !strcmp(ipv, "6")) ip = 6;
186
+ else error("Cannot understand ip version '%s'. Assuming 'any'.", ipv);
187
+
188
+ if(ip == 0 || ip == 6) listen_fd = create_listen_socket6(config_get("global", "bind socket to IP", "*"), listen_port, listen_backlog);
189
+ if(listen_fd < 0) {
190
+ listen_fd = create_listen_socket4(config_get("global", "bind socket to IP", "*"), listen_port, listen_backlog);
191
+ // if(listen_fd >= 0 && ip != 4) info("Managed to open an IPv4 socket on port %d.", listen_port);
192
+ }
193
+
194
+ return listen_fd;
195
+}
196
+
197
// --------------------------------------------------------------------------------------
198
// the main socket listener
199
@@ -176,15 +202,14 @@ int create_listen_socket6(const char *ip, int port, int listen_backlog)
202
// 3. spawns a new pthread to serve the client (this is optimal for keep-alive clients)
203
// 4. cleans up old web_clients that their pthreads have been exited
204
179
-void *socket_listen_main(void *ptr)
180
-{
205
+void *socket_listen_main_multi_threaded(void *ptr) {
206
if(ptr) { ; }
207
183
- info("WEB SERVER thread created with task id %d", gettid());
208
+ info("Multi-threaded WEB SERVER thread created with task id %d", gettid());
209
210
struct web_client *w;
211
struct timeval tv;
187
- int retval;
212
+ int retval, failures = 0;
213
214
if(ptr) { ; }
215
@@ -199,32 +224,40 @@ void *socket_listen_main(void *ptr)
224
225
if(listen_fd < 0) fatal("LISTENER: Listen socket is not ready.");
226
202
- fd_set ifds, ofds, efds;
203
- int fdmax = listen_fd;
204
-
227
+ fd_set ifds;
228
FD_ZERO (&ifds);
206
- FD_ZERO (&ofds);
207
- FD_ZERO (&efds);
229
230
for(;;) {
231
tv.tv_sec = 0;
232
tv.tv_usec = 200000;
233
213
- if(listen_fd >= 0) {
234
+ if(likely(listen_fd >= 0))
235
FD_SET(listen_fd, &ifds);
215
- FD_SET(listen_fd, &efds);
216
- }
236
237
// debug(D_WEB_CLIENT, "LISTENER: Waiting...");
219
- retval = select(fdmax+1, &ifds, &ofds, &efds, &tv);
238
+ retval = select(listen_fd + 1, &ifds, NULL, NULL, &tv);
239
221
- if(retval == -1) {
240
+ if(unlikely(retval == -1)) {
241
error("LISTENER: select() failed.");
242
+ failures++;
243
+
244
+ if(failures > 10) {
245
+ error("LISTENER: our listen port %d seems dead. Re-opening it.", listen_fd);
246
+ close(listen_fd);
247
+ listen_fd = -1;
248
+ sleep(5);
249
+ create_listen_socket();
250
+ if(listen_fd < 0)
251
+ fatal("Cannot listen for web clients (connected clients %llu).", global_statistics.connected_clients);
252
+
253
+ failures = 0;
254
+ }
255
+
256
continue;
257
}
225
- else if(retval) {
258
+ else if(likely(retval)) {
259
// check for new incoming connections
227
- if(FD_ISSET(listen_fd, &ifds)) {
260
+ if(likely(FD_ISSET(listen_fd, &ifds))) {
261
w = web_client_create(listen_fd);
262
if(unlikely(!w)) {
263
// no need for error log - web_client_create already logged the error
@@ -247,16 +280,20 @@ void *socket_listen_main(void *ptr)
280
// debug(D_WEB_CLIENT, "LISTENER: select() timeout.");
281
//}
282
283
+ failures = 0;
284
+
285
// cleanup unused clients
251
- for(w = web_clients; w ; w = w?w->next:NULL) {
252
- if(w->obsolete) {
286
+ for (w = web_clients; w; ) {
287
+ if (w->obsolete) {
288
debug(D_WEB_CLIENT, "%llu: Removing client.", w->id);
254
- // pthread_join(w->thread, NULL);
289
+ // pthread_cancel(w->thread);
290
+ // pthread_join(w->thread, NULL);
291
w = web_client_free(w);
292
#ifdef NETDATA_INTERNAL_CHECKS
293
log_allocations();
294
#endif
295
}
296
+ else w = w->next;
297
}
298
}
299
@@ -268,3 +305,190 @@ void *socket_listen_main(void *ptr)
305
return NULL;
306
}
307
308
+void *socket_listen_main_single_threaded(void *ptr) {
309
+ if(ptr) { ; }
310
+
311
+ info("Single threaded WEB SERVER thread created with task id %d", gettid());
312
+
313
+ struct web_client *w;
314
+ int retval, failures = 0;
315
+
316
+ if(ptr) { ; }
317
+
318
+ if(pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL) != 0)
319
+ error("Cannot set pthread cancel type to DEFERRED.");
320
+
321
+ if(pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL) != 0)
322
+ error("Cannot set pthread cancel state to ENABLE.");
323
+
324
+ web_client_timeout = (int) config_get_number("global", "disconnect idle web clients after seconds", DEFAULT_DISCONNECT_IDLE_WEB_CLIENTS_AFTER_SECONDS);
325
+ web_enable_gzip = config_get_boolean("global", "enable web responses gzip compression", web_enable_gzip);
326
+
327
+ if(listen_fd < 0) fatal("LISTENER: Listen socket is not ready.");
328
+
329
+ fd_set ifds, ofds, efds;
330
+ int fdmax = listen_fd;
331
+
332
+ for(;;) {
333
+ int has_obsolete = 0;
334
+ FD_ZERO (&ifds);
335
+ FD_ZERO (&ofds);
336
+ FD_ZERO (&efds);
337
+
338
+ if(listen_fd >= 0) {
339
+ // debug(D_WEB_CLIENT_ACCESS, "LISTENER: adding listen socket %d to ifds, efds", listen_fd);
340
+ FD_SET(listen_fd, &ifds);
341
+ FD_SET(listen_fd, &efds);
342
+ }
343
+
344
+ for(w = web_clients; w ; w = w->next) {
345
+ if(unlikely(w->dead)) {
346
+ error("%llu: client is dead.");
347
+ w->obsolete = 1;
348
+ }
349
+ else if(unlikely(!w->wait_receive && !w->wait_send)) {
350
+ error("%llu: client is not set for neither receiving nor sending data.");
351
+ w->obsolete = 1;
352
+ }
353
+
354
+ if(unlikely(w->obsolete)) {
355
+ has_obsolete++;
356
+ continue;
357
+ }
358
+
359
+ // debug(D_WEB_CLIENT_ACCESS, "%llu: adding input socket %d to efds", w->id, w->ifd);
360
+ FD_SET(w->ifd, &efds);
361
+ if(w->ifd > fdmax) fdmax = w->ifd;
362
+
363
+ if(w->ifd != w->ofd) {
364
+ // debug(D_WEB_CLIENT_ACCESS, "%llu: adding output socket %d to efds", w->id, w->ofd);
365
+ FD_SET(w->ofd, &efds);
366
+ if(w->ofd > fdmax) fdmax = w->ofd;
367
+ }
368
+
369
+ if (w->wait_receive) {
370
+ // debug(D_WEB_CLIENT_ACCESS, "%llu: adding input socket %d to ifds", w->id, w->ifd);
371
+ FD_SET(w->ifd, &ifds);
372
+ if(w->ifd > fdmax) fdmax = w->ifd;
373
+ }
374
+
375
+ if (w->wait_send) {
376
+ // debug(D_WEB_CLIENT_ACCESS, "%llu: adding output socket %d to ofds", w->id, w->ofd);
377
+ FD_SET(w->ofd, &ofds);
378
+ if(w->ofd > fdmax) fdmax = w->ofd;
379
+ }
380
+ }
381
+
382
+ // cleanup unused clients
383
+ if(unlikely(has_obsolete)) {
384
+ for (w = web_clients; w; ) {
385
+ if (w->obsolete) {
386
+ debug(D_WEB_CLIENT, "%llu: Removing client.", w->id);
387
+ w = web_client_free(w);
388
+#ifdef NETDATA_INTERNAL_CHECKS
389
+ log_allocations();
390
+#endif
391
+ }
392
+ else w = w->next;
393
+ }
394
+ }
395
+
396
+ debug(D_WEB_CLIENT_ACCESS, "LISTENER: Waiting...");
397
+ struct timeval tv = { .tv_sec = 1, .tv_usec = 0 };
398
+ errno = 0;
399
+ retval = select(fdmax+1, &ifds, &ofds, &efds, &tv);
400
+
401
+ if(retval == -1) {
402
+ error("LISTENER: select() failed.");
403
+
404
+ if(errno != EAGAIN) {
405
+ // debug(D_WEB_CLIENT_ACCESS, "LISTENER: select() failed.");
406
+ error("REMOVING ALL %lu WEB CLIENTS !", global_statistics.connected_clients);
407
+ while (web_clients) web_client_free(web_clients);
408
+ }
409
+
410
+ failures++;
411
+ if(failures > 10) {
412
+ error("LISTENER: our listen port %d seems dead. Re-opening it.", listen_fd);
413
+ close(listen_fd);
414
+ listen_fd = -1;
415
+ sleep(5);
416
+ create_listen_socket();
417
+ if(listen_fd < 0)
418
+ fatal("Cannot listen for web clients (connected clients %llu).", global_statistics.connected_clients);
419
+
420
+ failures = 0;
421
+ }
422
+
423
+ continue;
424
+ }
425
+ else if(retval) {
426
+ for(w = web_clients; w ; w = w->next) {
427
+ if (unlikely(w->obsolete)) continue;
428
+
429
+ if (unlikely(FD_ISSET(w->ifd, &efds))) {
430
+ debug(D_WEB_CLIENT_ACCESS, "%llu: Received error on input socket.", w->id);
431
+ web_client_reset(w);
432
+ w->obsolete = 1;
433
+ continue;
434
+ }
435
+
436
+ if (unlikely(FD_ISSET(w->ofd, &efds))) {
437
+ debug(D_WEB_CLIENT_ACCESS, "%llu: Received error on output socket.", w->id);
438
+ web_client_reset(w);
439
+ w->obsolete = 1;
440
+ continue;
441
+ }
442
+
443
+ if (unlikely(w->wait_receive && FD_ISSET(w->ifd, &ifds))) {
444
+ long bytes;
445
+ if (unlikely((bytes = web_client_receive(w)) < 0)) {
446
+ debug(D_WEB_CLIENT, "%llu: Cannot receive data from client. Closing client.", w->id);
447
+ errno = 0;
448
+ web_client_reset(w);
449
+ w->obsolete = 1;
450
+ continue;
451
+ }
452
+
453
+ if (w->mode == WEB_CLIENT_MODE_NORMAL) {
454
+ debug(D_WEB_CLIENT, "%llu: Processing received data (%ld bytes).", w->id, bytes);
455
+ // info("%llu: Attempting to process received data (%ld bytes).", w->id, bytes);
456
+ web_client_process(w);
457
+ }
458
+ else {
459
+ debug(D_WEB_CLIENT, "%llu: NO Processing for received data (%ld bytes).", w->id, bytes);
460
+ }
461
+ }
462
+
463
+ if (unlikely(w->wait_send && FD_ISSET(w->ofd, &ofds))) {
464
+ ssize_t bytes;
465
+ if (unlikely((bytes = web_client_send(w)) < 0)) {
466
+ debug(D_WEB_CLIENT, "%llu: Cannot send data to client. Closing client.", w->id);
467
+ errno = 0;
468
+ web_client_reset(w);
469
+ w->obsolete = 1;
470
+ continue;
471
+ }
472
+ }
473
+ }
474
+
475
+ // check for new incoming connections
476
+ if(FD_ISSET(listen_fd, &ifds)) {
477
+ debug(D_WEB_CLIENT_ACCESS, "LISTENER: new connection.");
478
+ web_client_create(listen_fd);
479
+ }
480
+ }
481
+ else {
482
+ debug(D_WEB_CLIENT_ACCESS, "LISTENER: timeout.");
483
+ }
484
+
485
+ failures = 0;
486
+ }
487
+
488
+ error("LISTENER: exit!");
489
+
490
+ if(listen_fd >= 0) close(listen_fd);
491
+ exit(2);
492
+
493
+ return NULL;
494
+}
src/web_server.h
+3
-1
@@ -15,6 +15,8 @@ extern int listen_port;
15
16
extern int create_listen_socket4(const char *ip, int port, int listen_backlog);
17
extern int create_listen_socket6(const char *ip, int port, int listen_backlog);
18
-extern void *socket_listen_main(void *ptr);
18
+extern void *socket_listen_main_multi_threaded(void *ptr);
19
+extern void *socket_listen_main_single_threaded(void *ptr);
20
+extern int create_listen_socket(void);
21
22
#endif /* NETDATA_WEB_SERVER_H */