summaryrefslogtreecommitdiffstats
path: root/libnetdata/socket
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2022-06-09 04:52:39 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2022-06-09 04:52:39 +0000
commit89f3604407aff8f4cb2ed958252c61e23c767e24 (patch)
tree7fbf408102cab051557d38193524d8c6e991d070 /libnetdata/socket
parentAdding upstream version 1.34.1. (diff)
downloadnetdata-89f3604407aff8f4cb2ed958252c61e23c767e24.tar.xz
netdata-89f3604407aff8f4cb2ed958252c61e23c767e24.zip
Adding upstream version 1.35.0.upstream/1.35.0
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'libnetdata/socket')
-rw-r--r--libnetdata/socket/socket.c384
1 files changed, 231 insertions, 153 deletions
diff --git a/libnetdata/socket/socket.c b/libnetdata/socket/socket.c
index 73eb8e662..df6d3148b 100644
--- a/libnetdata/socket/socket.c
+++ b/libnetdata/socket/socket.c
@@ -1386,175 +1386,142 @@ static void poll_events_cleanup(void *data) {
freez(p->inf);
}
-static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, short int revents, time_t now) {
- short int events = pf->events;
- int fd = pf->fd;
- pf->revents = 0;
- size_t i = pi->slot;
+static int poll_process_error(POLLINFO *pi, struct pollfd *pf, short int revents) {
+ error("POLLFD: LISTENER: received %s %s %s on socket at slot %zu (fd %d) client '%s' port '%s' expecting %s %s %s, having %s %s %s"
+ , revents & POLLERR ? "POLLERR" : ""
+ , revents & POLLHUP ? "POLLHUP" : ""
+ , revents & POLLNVAL ? "POLLNVAL" : ""
+ , pi->slot
+ , pi->fd
+ , pi->client_ip ? pi->client_ip : "<undefined-ip>"
+ , pi->client_port ? pi->client_port : "<undefined-port>"
+ , pf->events & POLLIN ? "POLLIN" : "", pf->events & POLLOUT ? "POLLOUT" : "", pf->events & POLLPRI ? "POLLPRI" : ""
+ , revents & POLLIN ? "POLLIN" : "", revents & POLLOUT ? "POLLOUT" : "", revents & POLLPRI ? "POLLPRI" : ""
+ );
- if(unlikely(fd == -1)) {
- debug(D_POLLFD, "POLLFD: LISTENER: ignoring slot %zu, it does not have an fd", i);
- return;
- }
+ pf->events = 0;
+ poll_close_fd(pi);
+ return 1;
+}
- debug(D_POLLFD, "POLLFD: LISTENER: processing events for slot %zu (events = %d, revents = %d)", i, events, revents);
+static inline int poll_process_send(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, time_t now) {
+ pi->last_sent_t = now;
+ pi->send_count++;
- if(revents & POLLIN || revents & POLLPRI) {
- // receiving data
+ debug(D_POLLFD, "POLLFD: LISTENER: sending data to socket on slot %zu (fd %d)", pi->slot, pf->fd);
- pi->last_received_t = now;
- pi->recv_count++;
+ pf->events = 0;
- if(likely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET)) {
- // read data from client TCP socket
- debug(D_POLLFD, "POLLFD: LISTENER: reading data from TCP client slot %zu (fd %d)", i, fd);
+ // remember the slot, in case we need to close it later
+ // the callback may manipulate the socket list and our pf and pi pointers may be invalid after that call
+ size_t slot = pi->slot;
- pf->events = 0;
- if (pi->rcv_callback(pi, &pf->events) == -1) {
- poll_close_fd(&p->inf[i]);
- return;
- }
- pf = &p->fds[i];
- pi = &p->inf[i];
-
-#ifdef NETDATA_INTERNAL_CHECKS
- // this is common - it is used for web server file copies
- if(unlikely(!(pf->events & (POLLIN|POLLOUT)))) {
- error("POLLFD: LISTENER: after reading, client slot %zu (fd %d) from %s port %s was left without expecting input or output. ", i, fd, pi->client_ip?pi->client_ip:"<undefined-ip>", pi->client_port?pi->client_port:"<undefined-port>");
- //poll_close_fd(pi);
- //return;
- }
-#endif
- }
- else if(likely(pi->flags & POLLINFO_FLAG_SERVER_SOCKET)) {
- // new connection
- // debug(D_POLLFD, "POLLFD: LISTENER: accepting connections from slot %zu (fd %d)", i, fd);
-
- switch(pi->socktype) {
- case SOCK_STREAM: {
- // a TCP socket
- // we accept the connection
-
- int nfd;
- do {
- char client_ip[INET6_ADDRSTRLEN];
- char client_port[NI_MAXSERV];
- char client_host[NI_MAXHOST];
- client_host[0] = 0;
- client_ip[0] = 0;
- client_port[0] = 0;
-
- debug(D_POLLFD, "POLLFD: LISTENER: calling accept4() slot %zu (fd %d)", i, fd);
- nfd = accept_socket(fd, SOCK_NONBLOCK, client_ip, INET6_ADDRSTRLEN, client_port, NI_MAXSERV,
- client_host, NI_MAXHOST, p->access_list, p->allow_dns);
- if (unlikely(nfd < 0)) {
- // accept failed
-
- debug(D_POLLFD, "POLLFD: LISTENER: accept4() slot %zu (fd %d) failed.", i, fd);
-
- if(unlikely(errno == EMFILE)) {
- error("POLLFD: LISTENER: too many open files - sleeping for 1ms - used by this thread %zu, max for this thread %zu", p->used, p->limit);
- usleep(1000); // 10ms
- }
- else if(unlikely(errno != EWOULDBLOCK && errno != EAGAIN))
- error("POLLFD: LISTENER: accept() failed.");
-
- break;
- }
- else {
- // accept ok
- // info("POLLFD: LISTENER: client '[%s]:%s' connected to '%s' on fd %d", client_ip, client_port, sockets->fds_names[i], nfd);
- poll_add_fd(p
- , nfd
- , SOCK_STREAM
- , pi->port_acl
- , POLLINFO_FLAG_CLIENT_SOCKET
- , client_ip
- , client_port
- , client_host
- , p->add_callback
- , p->del_callback
- , p->rcv_callback
- , p->snd_callback
- , NULL
- );
+ if (unlikely(pi->snd_callback(pi, &pf->events) == -1))
+ poll_close_fd(&p->inf[slot]);
- // it may have reallocated them, so refresh our pointers
- pf = &p->fds[i];
- pi = &p->inf[i];
- }
- } while (nfd >= 0 && (!p->limit || p->used < p->limit));
- break;
- }
+ // IMPORTANT:
+ // pf and pi may be invalid below this point, they may have been reallocated.
- case SOCK_DGRAM: {
- // a UDP socket
- // we read data from the server socket
+ return 1;
+}
- debug(D_POLLFD, "POLLFD: LISTENER: reading data from UDP slot %zu (fd %d)", i, fd);
+static inline int poll_process_tcp_read(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, time_t now) {
+ pi->last_received_t = now;
+ pi->recv_count++;
- // TODO: access_list is not applied to UDP
- // but checking the access list on every UDP packet will destroy
- // performance, especially for statsd.
+ debug(D_POLLFD, "POLLFD: LISTENER: reading data from TCP client slot %zu (fd %d)", pi->slot, pf->fd);
- pf->events = 0;
- pi->rcv_callback(pi, &pf->events);
- break;
- }
+ pf->events = 0;
- default: {
- error("POLLFD: LISTENER: Unknown socktype %d on slot %zu", pi->socktype, pi->slot);
- break;
- }
- }
- }
- }
+ // remember the slot, in case we need to close it later
+ // the callback may manipulate the socket list and our pf and pi pointers may be invalid after that call
+ size_t slot = pi->slot;
- if(unlikely(revents & POLLOUT)) {
- // sending data
- debug(D_POLLFD, "POLLFD: LISTENER: sending data to socket on slot %zu (fd %d)", i, fd);
+ if (pi->rcv_callback(pi, &pf->events) == -1)
+ poll_close_fd(&p->inf[slot]);
- pi->last_sent_t = now;
- pi->send_count++;
+ // IMPORTANT:
+ // pf and pi may be invalid below this point, they may have been reallocated.
- pf->events = 0;
- if (pi->snd_callback(pi, &pf->events) == -1) {
- poll_close_fd(&p->inf[i]);
- return;
- }
- pf = &p->fds[i];
- pi = &p->inf[i];
-
-#ifdef NETDATA_INTERNAL_CHECKS
- // this is common - it is used for streaming
- if(unlikely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET && !(pf->events & (POLLIN|POLLOUT)))) {
- error("POLLFD: LISTENER: after sending, client slot %zu (fd %d) from %s port %s was left without expecting input or output. ", i, fd, pi->client_ip?pi->client_ip:"<undefined-ip>", pi->client_port?pi->client_port:"<undefined-port>");
- //poll_close_fd(pi);
- //return;
+ return 1;
+}
+
+static inline int poll_process_udp_read(POLLINFO *pi, struct pollfd *pf, time_t now __maybe_unused) {
+ pi->last_received_t = now;
+ pi->recv_count++;
+
+ debug(D_POLLFD, "POLLFD: LISTENER: reading data from UDP slot %zu (fd %d)", pi->slot, pf->fd);
+
+ // TODO: access_list is not applied to UDP
+ // but checking the access list on every UDP packet will destroy
+ // performance, especially for statsd.
+
+ pf->events = 0;
+ if(pi->rcv_callback(pi, &pf->events) == -1)
+ return 0;
+
+ // IMPORTANT:
+ // pf and pi may be invalid below this point, they may have been reallocated.
+
+ return 1;
+}
+
+static int poll_process_new_tcp_connection(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, time_t now) {
+ pi->last_received_t = now;
+ pi->recv_count++;
+
+ debug(D_POLLFD, "POLLFD: LISTENER: accepting connections from slot %zu (fd %d)", pi->slot, pf->fd);
+
+ char client_ip[INET6_ADDRSTRLEN] = "";
+ char client_port[NI_MAXSERV] = "";
+ char client_host[NI_MAXHOST] = "";
+
+ debug(D_POLLFD, "POLLFD: LISTENER: calling accept4() slot %zu (fd %d)", pi->slot, pf->fd);
+
+ int nfd = accept_socket(
+ pf->fd,SOCK_NONBLOCK,
+ client_ip, INET6_ADDRSTRLEN, client_port,NI_MAXSERV, client_host, NI_MAXHOST,
+ p->access_list, p->allow_dns
+ );
+
+ if (unlikely(nfd < 0)) {
+ // accept failed
+
+ debug(D_POLLFD, "POLLFD: LISTENER: accept4() slot %zu (fd %d) failed.", pi->slot, pf->fd);
+
+ if(unlikely(errno == EMFILE)) {
+ error("POLLFD: LISTENER: too many open files - sleeping for 1ms - used by this thread %zu, max for this thread %zu", p->used, p->limit);
+ usleep(1000); // 1ms
}
-#endif
- }
+ else if(unlikely(errno != EWOULDBLOCK && errno != EAGAIN))
+ error("POLLFD: LISTENER: accept() failed.");
- if(unlikely(revents & POLLERR)) {
- error("POLLFD: LISTENER: processing POLLERR events for slot %zu fd %d (events = %d, revents = %d)", i, events, revents, fd);
- pf->events = 0;
- poll_close_fd(pi);
- return;
}
+ else {
+ // accept ok
+
+ poll_add_fd(p
+ , nfd
+ , SOCK_STREAM
+ , pi->port_acl
+ , POLLINFO_FLAG_CLIENT_SOCKET
+ , client_ip
+ , client_port
+ , client_host
+ , p->add_callback
+ , p->del_callback
+ , p->rcv_callback
+ , p->snd_callback
+ , NULL
+ );
- if(unlikely(revents & POLLHUP)) {
- error("POLLFD: LISTENER: processing POLLHUP events for slot %zu fd %d (events = %d, revents = %d)", i, events, revents, fd);
- pf->events = 0;
- poll_close_fd(pi);
- return;
- }
+ // IMPORTANT:
+ // pf and pi may be invalid below this point, they may have been reallocated.
- if(unlikely(revents & POLLNVAL)) {
- error("POLLFD: LISTENER: processing POLLNVAL events for slot %zu fd %d (events = %d, revents = %d)", i, events, revents, fd);
- pf->events = 0;
- poll_close_fd(pi);
- return;
+ return 1;
}
+
+ return 0;
}
void poll_events(LISTEN_SOCKETS *sockets
@@ -1687,18 +1654,129 @@ void poll_events(LISTEN_SOCKETS *sockets
debug(D_POLLFD, "POLLFD: LISTENER: poll() timeout.");
}
else {
+ POLLINFO *pi;
+ struct pollfd *pf;
+ size_t idx, processed = 0;
+ short int revents;
+
+ // keep fast lookup arrays per function
+ // to avoid looping through the entire list every time
+ size_t sends[p.max + 1], sends_max = 0;
+ size_t reads[p.max + 1], reads_max = 0;
+ size_t conns[p.max + 1], conns_max = 0;
+ size_t udprd[p.max + 1], udprd_max = 0;
+
for (i = 0; i <= p.max; i++) {
- struct pollfd *pf = &p.fds[i];
- short int revents = pf->revents;
- if (unlikely(revents))
- poll_events_process(&p, &p.inf[i], pf, revents, now);
+ pi = &p.inf[i];
+ pf = &p.fds[i];
+ revents = pf->revents;
+
+ if(unlikely(revents == 0 || pf->fd == -1))
+ continue;
+
+ if (unlikely(revents & (POLLERR|POLLHUP|POLLNVAL))) {
+ // something is wrong to one of our sockets
+
+ pf->revents = 0;
+ processed += poll_process_error(pi, pf, revents);
+ }
+ else if (likely(revents & POLLOUT)) {
+ // a client is ready to receive data
+
+ sends[sends_max++] = i;
+ }
+ else if (likely(revents & (POLLIN|POLLPRI))) {
+ if (pi->flags & POLLINFO_FLAG_CLIENT_SOCKET) {
+ // a client sent data to us
+
+ reads[reads_max++] = i;
+ }
+ else if (pi->flags & POLLINFO_FLAG_SERVER_SOCKET) {
+ // something is coming to our server sockets
+
+ if(pi->socktype == SOCK_DGRAM) {
+ // UDP receive, directly on our listening socket
+
+ udprd[udprd_max++] = i;
+ }
+ else if(pi->socktype == SOCK_STREAM) {
+ // new TCP connection
+
+ conns[conns_max++] = i;
+ }
+ else
+ error("POLLFD: LISTENER: server slot %zu (fd %d) connection from %s port %s using unhandled socket type %d."
+ , i
+ , pi->fd
+ , pi->client_ip ? pi->client_ip : "<undefined-ip>"
+ , pi->client_port ? pi->client_port : "<undefined-port>"
+ , pi->socktype
+ );
+ }
+ else
+ error("POLLFD: LISTENER: client slot %zu (fd %d) data from %s port %s using flags %08X is neither client nor server."
+ , i
+ , pi->fd
+ , pi->client_ip ? pi->client_ip : "<undefined-ip>"
+ , pi->client_port ? pi->client_port : "<undefined-port>"
+ , pi->flags
+ );
+ }
+ else
+ error("POLLFD: LISTENER: socket slot %zu (fd %d) client %s port %s unhandled event id %d."
+ , i
+ , pi->fd
+ , pi->client_ip ? pi->client_ip : "<undefined-ip>"
+ , pi->client_port ? pi->client_port : "<undefined-port>"
+ , revents
+ );
+ }
+
+ // process sends
+ for (idx = 0; idx < sends_max; idx++) {
+ i = sends[idx];
+ pi = &p.inf[i];
+ pf = &p.fds[i];
+ pf->revents = 0;
+ processed += poll_process_send(&p, pi, pf, now);
+ }
+
+ // process UDP reads
+ for (idx = 0; idx < udprd_max; idx++) {
+ i = udprd[idx];
+ pi = &p.inf[i];
+ pf = &p.fds[i];
+ pf->revents = 0;
+ processed += poll_process_udp_read(pi, pf, now);
+ }
+
+ // process TCP reads
+ for (idx = 0; idx < reads_max; idx++) {
+ i = reads[idx];
+ pi = &p.inf[i];
+ pf = &p.fds[i];
+ pf->revents = 0;
+ processed += poll_process_tcp_read(&p, pi, pf, now);
+ }
+
+ if(!processed && (!p.limit || p.used < p.limit)) {
+ // nothing processed above (rcv, snd) and we have room for another TCP connection
+ // so, accept one TCP connection
+ for (idx = 0; idx < conns_max; idx++) {
+ i = conns[idx];
+ pi = &p.inf[i];
+ pf = &p.fds[i];
+ pf->revents = 0;
+ if (poll_process_new_tcp_connection(&p, pi, pf, now))
+ break;
+ }
}
}
if(unlikely(p.checks_every > 0 && now - last_check > p.checks_every)) {
last_check = now;
- // security checks
+ // cleanup old sockets
for(i = 0; i <= p.max; i++) {
POLLINFO *pi = &p.inf[i];