@cryptotaxi247 / netdata-1 / commits / 619f9964a

optimize poll_events() to spread the work over the threads more evenly (#12975)

* optimize poll_events() to spread the work over the threads more evenly * fixed typos, code cleanup * better error handling * prevent crash in case callbacks manipulate the sockets arrays - added warnings

Costa Tsaousis committed May 21, 2022 at 09:49 UTC 619f9964a69b93301b66b9d2842354f1d36b82f3
1 file changed +231 -153
libnetdata/socket/socket.c
+231 -153
@@ -1386,175 +1386,142 @@ static void poll_events_cleanup(void *data) {
1386 freez(p->inf);
1387 }
1388
1389 -static void poll_events_process(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, short int revents, time_t now) {
1390 - short int events = pf->events;
1391 - int fd = pf->fd;
1392 - pf->revents = 0;
1393 - size_t i = pi->slot;
1389 +static int poll_process_error(POLLINFO *pi, struct pollfd *pf, short int revents) {
1390 + 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"
1391 + , revents & POLLERR ? "POLLERR" : ""
1392 + , revents & POLLHUP ? "POLLHUP" : ""
1393 + , revents & POLLNVAL ? "POLLNVAL" : ""
1394 + , pi->slot
1395 + , pi->fd
1396 + , pi->client_ip ? pi->client_ip : "<undefined-ip>"
1397 + , pi->client_port ? pi->client_port : "<undefined-port>"
1398 + , pf->events & POLLIN ? "POLLIN" : "", pf->events & POLLOUT ? "POLLOUT" : "", pf->events & POLLPRI ? "POLLPRI" : ""
1399 + , revents & POLLIN ? "POLLIN" : "", revents & POLLOUT ? "POLLOUT" : "", revents & POLLPRI ? "POLLPRI" : ""
1400 + );
1401
1395 - if(unlikely(fd == -1)) {
1396 - debug(D_POLLFD, "POLLFD: LISTENER: ignoring slot %zu, it does not have an fd", i);
1397 - return;
1398 - }
1402 + pf->events = 0;
1403 + poll_close_fd(pi);
1404 + return 1;
1405 +}
1406
1400 - debug(D_POLLFD, "POLLFD: LISTENER: processing events for slot %zu (events = %d, revents = %d)", i, events, revents);
1407 +static inline int poll_process_send(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, time_t now) {
1408 + pi->last_sent_t = now;
1409 + pi->send_count++;
1410
1402 - if(revents & POLLIN || revents & POLLPRI) {
1403 - // receiving data
1411 + debug(D_POLLFD, "POLLFD: LISTENER: sending data to socket on slot %zu (fd %d)", pi->slot, pf->fd);
1412
1405 - pi->last_received_t = now;
1406 - pi->recv_count++;
1413 + pf->events = 0;
1414
1408 - if(likely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET)) {
1409 - // read data from client TCP socket
1410 - debug(D_POLLFD, "POLLFD: LISTENER: reading data from TCP client slot %zu (fd %d)", i, fd);
1415 + // remember the slot, in case we need to close it later
1416 + // the callback may manipulate the socket list and our pf and pi pointers may be invalid after that call
1417 + size_t slot = pi->slot;
1418
1412 - pf->events = 0;
1413 - if (pi->rcv_callback(pi, &pf->events) == -1) {
1414 - poll_close_fd(&p->inf[i]);
1415 - return;
1416 - }
1417 - pf = &p->fds[i];
1418 - pi = &p->inf[i];
1419 -
1420 -#ifdef NETDATA_INTERNAL_CHECKS
1421 - // this is common - it is used for web server file copies
1422 - if(unlikely(!(pf->events & (POLLIN|POLLOUT)))) {
1423 - 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>");
1424 - //poll_close_fd(pi);
1425 - //return;
1426 - }
1427 -#endif
1428 - }
1429 - else if(likely(pi->flags & POLLINFO_FLAG_SERVER_SOCKET)) {
1430 - // new connection
1431 - // debug(D_POLLFD, "POLLFD: LISTENER: accepting connections from slot %zu (fd %d)", i, fd);
1432 -
1433 - switch(pi->socktype) {
1434 - case SOCK_STREAM: {
1435 - // a TCP socket
1436 - // we accept the connection
1437 -
1438 - int nfd;
1439 - do {
1440 - char client_ip[INET6_ADDRSTRLEN];
1441 - char client_port[NI_MAXSERV];
1442 - char client_host[NI_MAXHOST];
1443 - client_host[0] = 0;
1444 - client_ip[0] = 0;
1445 - client_port[0] = 0;
1446 -
1447 - debug(D_POLLFD, "POLLFD: LISTENER: calling accept4() slot %zu (fd %d)", i, fd);
1448 - nfd = accept_socket(fd, SOCK_NONBLOCK, client_ip, INET6_ADDRSTRLEN, client_port, NI_MAXSERV,
1449 - client_host, NI_MAXHOST, p->access_list, p->allow_dns);
1450 - if (unlikely(nfd < 0)) {
1451 - // accept failed
1452 -
1453 - debug(D_POLLFD, "POLLFD: LISTENER: accept4() slot %zu (fd %d) failed.", i, fd);
1454 -
1455 - if(unlikely(errno == EMFILE)) {
1456 - error("POLLFD: LISTENER: too many open files - sleeping for 1ms - used by this thread %zu, max for this thread %zu", p->used, p->limit);
1457 - usleep(1000); // 10ms
1458 - }
1459 - else if(unlikely(errno != EWOULDBLOCK && errno != EAGAIN))
1460 - error("POLLFD: LISTENER: accept() failed.");
1461 -
1462 - break;
1463 - }
1464 - else {
1465 - // accept ok
1466 - // info("POLLFD: LISTENER: client '[%s]:%s' connected to '%s' on fd %d", client_ip, client_port, sockets->fds_names[i], nfd);
1467 - poll_add_fd(p
1468 - , nfd
1469 - , SOCK_STREAM
1470 - , pi->port_acl
1471 - , POLLINFO_FLAG_CLIENT_SOCKET
1472 - , client_ip
1473 - , client_port
1474 - , client_host
1475 - , p->add_callback
1476 - , p->del_callback
1477 - , p->rcv_callback
1478 - , p->snd_callback
1479 - , NULL
1480 - );
1419 + if (unlikely(pi->snd_callback(pi, &pf->events) == -1))
1420 + poll_close_fd(&p->inf[slot]);
1421
1482 - // it may have reallocated them, so refresh our pointers
1483 - pf = &p->fds[i];
1484 - pi = &p->inf[i];
1485 - }
1486 - } while (nfd >= 0 && (!p->limit || p->used < p->limit));
1487 - break;
1488 - }
1422 + // IMPORTANT:
1423 + // pf and pi may be invalid below this point, they may have been reallocated.
1424
1490 - case SOCK_DGRAM: {
1491 - // a UDP socket
1492 - // we read data from the server socket
1425 + return 1;
1426 +}
1427
1494 - debug(D_POLLFD, "POLLFD: LISTENER: reading data from UDP slot %zu (fd %d)", i, fd);
1428 +static inline int poll_process_tcp_read(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, time_t now) {
1429 + pi->last_received_t = now;
1430 + pi->recv_count++;
1431
1496 - // TODO: access_list is not applied to UDP
1497 - // but checking the access list on every UDP packet will destroy
1498 - // performance, especially for statsd.
1432 + debug(D_POLLFD, "POLLFD: LISTENER: reading data from TCP client slot %zu (fd %d)", pi->slot, pf->fd);
1433
1500 - pf->events = 0;
1501 - pi->rcv_callback(pi, &pf->events);
1502 - break;
1503 - }
1434 + pf->events = 0;
1435
1505 - default: {
1506 - error("POLLFD: LISTENER: Unknown socktype %d on slot %zu", pi->socktype, pi->slot);
1507 - break;
1508 - }
1509 - }
1510 - }
1511 - }
1436 + // remember the slot, in case we need to close it later
1437 + // the callback may manipulate the socket list and our pf and pi pointers may be invalid after that call
1438 + size_t slot = pi->slot;
1439
1513 - if(unlikely(revents & POLLOUT)) {
1514 - // sending data
1515 - debug(D_POLLFD, "POLLFD: LISTENER: sending data to socket on slot %zu (fd %d)", i, fd);
1440 + if (pi->rcv_callback(pi, &pf->events) == -1)
1441 + poll_close_fd(&p->inf[slot]);
1442
1517 - pi->last_sent_t = now;
1518 - pi->send_count++;
1443 + // IMPORTANT:
1444 + // pf and pi may be invalid below this point, they may have been reallocated.
1445
1520 - pf->events = 0;
1521 - if (pi->snd_callback(pi, &pf->events) == -1) {
1522 - poll_close_fd(&p->inf[i]);
1523 - return;
1524 - }
1525 - pf = &p->fds[i];
1526 - pi = &p->inf[i];
1527 -
1528 -#ifdef NETDATA_INTERNAL_CHECKS
1529 - // this is common - it is used for streaming
1530 - if(unlikely(pi->flags & POLLINFO_FLAG_CLIENT_SOCKET && !(pf->events & (POLLIN|POLLOUT)))) {
1531 - 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>");
1532 - //poll_close_fd(pi);
1533 - //return;
1446 + return 1;
1447 +}
1448 +
1449 +static inline int poll_process_udp_read(POLLINFO *pi, struct pollfd *pf, time_t now __maybe_unused) {
1450 + pi->last_received_t = now;
1451 + pi->recv_count++;
1452 +
1453 + debug(D_POLLFD, "POLLFD: LISTENER: reading data from UDP slot %zu (fd %d)", pi->slot, pf->fd);
1454 +
1455 + // TODO: access_list is not applied to UDP
1456 + // but checking the access list on every UDP packet will destroy
1457 + // performance, especially for statsd.
1458 +
1459 + pf->events = 0;
1460 + if(pi->rcv_callback(pi, &pf->events) == -1)
1461 + return 0;
1462 +
1463 + // IMPORTANT:
1464 + // pf and pi may be invalid below this point, they may have been reallocated.
1465 +
1466 + return 1;
1467 +}
1468 +
1469 +static int poll_process_new_tcp_connection(POLLJOB *p, POLLINFO *pi, struct pollfd *pf, time_t now) {
1470 + pi->last_received_t = now;
1471 + pi->recv_count++;
1472 +
1473 + debug(D_POLLFD, "POLLFD: LISTENER: accepting connections from slot %zu (fd %d)", pi->slot, pf->fd);
1474 +
1475 + char client_ip[INET6_ADDRSTRLEN] = "";
1476 + char client_port[NI_MAXSERV] = "";
1477 + char client_host[NI_MAXHOST] = "";
1478 +
1479 + debug(D_POLLFD, "POLLFD: LISTENER: calling accept4() slot %zu (fd %d)", pi->slot, pf->fd);
1480 +
1481 + int nfd = accept_socket(
1482 + pf->fd,SOCK_NONBLOCK,
1483 + client_ip, INET6_ADDRSTRLEN, client_port,NI_MAXSERV, client_host, NI_MAXHOST,
1484 + p->access_list, p->allow_dns
1485 + );
1486 +
1487 + if (unlikely(nfd < 0)) {
1488 + // accept failed
1489 +
1490 + debug(D_POLLFD, "POLLFD: LISTENER: accept4() slot %zu (fd %d) failed.", pi->slot, pf->fd);
1491 +
1492 + if(unlikely(errno == EMFILE)) {
1493 + error("POLLFD: LISTENER: too many open files - sleeping for 1ms - used by this thread %zu, max for this thread %zu", p->used, p->limit);
1494 + usleep(1000); // 1ms
1495 }
1535 -#endif
1536 - }
1496 + else if(unlikely(errno != EWOULDBLOCK && errno != EAGAIN))
1497 + error("POLLFD: LISTENER: accept() failed.");
1498
1538 - if(unlikely(revents & POLLERR)) {
1539 - error("POLLFD: LISTENER: processing POLLERR events for slot %zu fd %d (events = %d, revents = %d)", i, events, revents, fd);
1540 - pf->events = 0;
1541 - poll_close_fd(pi);
1542 - return;
1499 }
1500 + else {
1501 + // accept ok
1502 +
1503 + poll_add_fd(p
1504 + , nfd
1505 + , SOCK_STREAM
1506 + , pi->port_acl
1507 + , POLLINFO_FLAG_CLIENT_SOCKET
1508 + , client_ip
1509 + , client_port
1510 + , client_host
1511 + , p->add_callback
1512 + , p->del_callback
1513 + , p->rcv_callback
1514 + , p->snd_callback
1515 + , NULL
1516 + );
1517
1545 - if(unlikely(revents & POLLHUP)) {
1546 - error("POLLFD: LISTENER: processing POLLHUP events for slot %zu fd %d (events = %d, revents = %d)", i, events, revents, fd);
1547 - pf->events = 0;
1548 - poll_close_fd(pi);
1549 - return;
1550 - }
1518 + // IMPORTANT:
1519 + // pf and pi may be invalid below this point, they may have been reallocated.
1520
1552 - if(unlikely(revents & POLLNVAL)) {
1553 - error("POLLFD: LISTENER: processing POLLNVAL events for slot %zu fd %d (events = %d, revents = %d)", i, events, revents, fd);
1554 - pf->events = 0;
1555 - poll_close_fd(pi);
1556 - return;
1521 + return 1;
1522 }
1523 +
1524 + return 0;
1525 }
1526
1527 void poll_events(LISTEN_SOCKETS *sockets
@@ -1687,18 +1654,129 @@ void poll_events(LISTEN_SOCKETS *sockets
1654 debug(D_POLLFD, "POLLFD: LISTENER: poll() timeout.");
1655 }
1656 else {
1657 + POLLINFO *pi;
1658 + struct pollfd *pf;
1659 + size_t idx, processed = 0;
1660 + short int revents;
1661 +
1662 + // keep fast lookup arrays per function
1663 + // to avoid looping through the entire list every time
1664 + size_t sends[p.max + 1], sends_max = 0;
1665 + size_t reads[p.max + 1], reads_max = 0;
1666 + size_t conns[p.max + 1], conns_max = 0;
1667 + size_t udprd[p.max + 1], udprd_max = 0;
1668 +
1669 for (i = 0; i <= p.max; i++) {
1691 - struct pollfd *pf = &p.fds[i];
1692 - short int revents = pf->revents;
1693 - if (unlikely(revents))
1694 - poll_events_process(&p, &p.inf[i], pf, revents, now);
1670 + pi = &p.inf[i];
1671 + pf = &p.fds[i];
1672 + revents = pf->revents;
1673 +
1674 + if(unlikely(revents == 0 || pf->fd == -1))
1675 + continue;
1676 +
1677 + if (unlikely(revents & (POLLERR|POLLHUP|POLLNVAL))) {
1678 + // something is wrong to one of our sockets
1679 +
1680 + pf->revents = 0;
1681 + processed += poll_process_error(pi, pf, revents);
1682 + }
1683 + else if (likely(revents & POLLOUT)) {
1684 + // a client is ready to receive data
1685 +
1686 + sends[sends_max++] = i;
1687 + }
1688 + else if (likely(revents & (POLLIN|POLLPRI))) {
1689 + if (pi->flags & POLLINFO_FLAG_CLIENT_SOCKET) {
1690 + // a client sent data to us
1691 +
1692 + reads[reads_max++] = i;
1693 + }
1694 + else if (pi->flags & POLLINFO_FLAG_SERVER_SOCKET) {
1695 + // something is coming to our server sockets
1696 +
1697 + if(pi->socktype == SOCK_DGRAM) {
1698 + // UDP receive, directly on our listening socket
1699 +
1700 + udprd[udprd_max++] = i;
1701 + }
1702 + else if(pi->socktype == SOCK_STREAM) {
1703 + // new TCP connection
1704 +
1705 + conns[conns_max++] = i;
1706 + }
1707 + else
1708 + error("POLLFD: LISTENER: server slot %zu (fd %d) connection from %s port %s using unhandled socket type %d."
1709 + , i
1710 + , pi->fd
1711 + , pi->client_ip ? pi->client_ip : "<undefined-ip>"
1712 + , pi->client_port ? pi->client_port : "<undefined-port>"
1713 + , pi->socktype
1714 + );
1715 + }
1716 + else
1717 + error("POLLFD: LISTENER: client slot %zu (fd %d) data from %s port %s using flags %08X is neither client nor server."
1718 + , i
1719 + , pi->fd
1720 + , pi->client_ip ? pi->client_ip : "<undefined-ip>"
1721 + , pi->client_port ? pi->client_port : "<undefined-port>"
1722 + , pi->flags
1723 + );
1724 + }
1725 + else
1726 + error("POLLFD: LISTENER: socket slot %zu (fd %d) client %s port %s unhandled event id %d."
1727 + , i
1728 + , pi->fd
1729 + , pi->client_ip ? pi->client_ip : "<undefined-ip>"
1730 + , pi->client_port ? pi->client_port : "<undefined-port>"
1731 + , revents
1732 + );
1733 + }
1734 +
1735 + // process sends
1736 + for (idx = 0; idx < sends_max; idx++) {
1737 + i = sends[idx];
1738 + pi = &p.inf[i];
1739 + pf = &p.fds[i];
1740 + pf->revents = 0;
1741 + processed += poll_process_send(&p, pi, pf, now);
1742 + }
1743 +
1744 + // process UDP reads
1745 + for (idx = 0; idx < udprd_max; idx++) {
1746 + i = udprd[idx];
1747 + pi = &p.inf[i];
1748 + pf = &p.fds[i];
1749 + pf->revents = 0;
1750 + processed += poll_process_udp_read(pi, pf, now);
1751 + }
1752 +
1753 + // process TCP reads
1754 + for (idx = 0; idx < reads_max; idx++) {
1755 + i = reads[idx];
1756 + pi = &p.inf[i];
1757 + pf = &p.fds[i];
1758 + pf->revents = 0;
1759 + processed += poll_process_tcp_read(&p, pi, pf, now);
1760 + }
1761 +
1762 + if(!processed && (!p.limit || p.used < p.limit)) {
1763 + // nothing processed above (rcv, snd) and we have room for another TCP connection
1764 + // so, accept one TCP connection
1765 + for (idx = 0; idx < conns_max; idx++) {
1766 + i = conns[idx];
1767 + pi = &p.inf[i];
1768 + pf = &p.fds[i];
1769 + pf->revents = 0;
1770 + if (poll_process_new_tcp_connection(&p, pi, pf, now))
1771 + break;
1772 + }
1773 }
1774 }
1775
1776 if(unlikely(p.checks_every > 0 && now - last_check > p.checks_every)) {
1777 last_check = now;
1778
1701 - // security checks
1779 + // cleanup old sockets
1780 for(i = 0; i <= p.max; i++) {
1781 POLLINFO *pi = &p.inf[i];
1782