9936897972

Committed 12 Jul 2024 11:28AM UTC coverage: 55.776% (-0.01%) from 55.79%

Build # 9936897972

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #14436 from rgacogne/ddist19-xsk-backend-race

dnsdist-1.9.x: Backport 14429 - Fix a race in the XSK/AF_XDP backend handling code

Run Details

13824 of 45312 branches covered (30.51%)

Branch coverage included in aggregate %.

1 of 71 new or added lines in 5 files covered. (1.41%)

14 existing lines in 5 files now uncovered.

47988 of 65509 relevant lines covered (73.25%)

3591675.17 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

4.18

/pdns/dnsdistdist/dnsdist-xsk.cc

/*
 * This file is part of PowerDNS or dnsdist.
 * Copyright -- PowerDNS.COM B.V. and its contributors
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of version 2 of the GNU General Public License as
 * published by the Free Software Foundation.
 *
 * In addition, for the avoidance of any doubt, permission is granted to
 * link this program with OpenSSL and to (re)distribute the binaries
 * produced as the result of such linking.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */
#include "dnsdist.hh"
#include "dnsdist-xsk.hh"

#ifdef HAVE_XSK
#include <sys/poll.h>

#include "dolog.hh"
#include "dnsdist-metrics.hh"
#include "dnsdist-proxy-protocol.hh"
#include "threadname.hh"
#include "xsk.hh"

namespace dnsdist::xsk
{
std::vector<std::shared_ptr<XskSocket>> g_xsk;

void XskResponderThread(std::shared_ptr<DownstreamState> dss, std::shared_ptr<XskWorker> xskInfo)
{
  try {
    setThreadName("dnsdist/XskResp");
    auto localRespRuleActions = g_respruleactions.getLocal();
    auto localCacheInsertedRespRuleActions = g_cacheInsertedRespRuleActions.getLocal();
    auto pollfds = getPollFdsForWorker(*xskInfo);
    while (!dss->isStopped()) {
      poll(pollfds.data(), pollfds.size(), -1);
      bool needNotify = false;
      if ((pollfds[0].revents & POLLIN) != 0) {
        needNotify = true;
        xskInfo->cleanSocketNotification();
        xskInfo->processIncomingFrames([&](XskPacket& packet) {
          if (packet.getDataLen() < sizeof(dnsheader)) {
            xskInfo->markAsFree(packet);
            return;
          }
          const dnsheader_aligned dnsHeader(packet.getPayloadData());
          const auto queryId = dnsHeader->id;
          auto ids = dss->getState(queryId);
          if (ids) {
            if (!ids->isXSK()) {
              dss->restoreState(queryId, std::move(*ids));
              ids = std::nullopt;
            }
          }
          if (!ids) {
            xskInfo->markAsFree(packet);
            return;
          }
          auto response = packet.clonePacketBuffer();
          if (response.size() > packet.getCapacity()) {
            /* fallback to sending the packet via normal socket */
            ids->xskPacketHeader.clear();
          }
          if (!processResponderPacket(dss, response, *localRespRuleActions, *localCacheInsertedRespRuleActions, std::move(*ids))) {
            xskInfo->markAsFree(packet);
            vinfolog("XSK packet dropped because processResponderPacket failed");
            return;
          }
          if (response.size() > packet.getCapacity()) {
            /* fallback to sending the packet via normal socket */
            sendUDPResponse(ids->cs->udpFD, response, ids->delayMsec, ids->hopLocal, ids->hopRemote);
            infolog("XSK packet falling back because packet is too large");
            xskInfo->markAsFree(packet);
            return;
          }
          packet.setHeader(ids->xskPacketHeader);
          if (!packet.setPayload(response)) {
            infolog("Unable to set XSK payload !");
          }
          if (ids->delayMsec > 0) {
            packet.addDelay(ids->delayMsec);
          }
          packet.updatePacket();
          xskInfo->pushToSendQueue(packet);
        });
      }
      if (needNotify) {
        xskInfo->notifyXskSocket();
      }
    }
  }
  catch (const std::exception& e) {
    errlog("XSK responder thread died because of exception: %s", e.what());
  }
  catch (const PDNSException& e) {
    errlog("XSK responder thread died because of PowerDNS exception: %s", e.reason);
  }
  catch (...) {
    errlog("XSK responder thread died because of an exception: %s", "unknown");
  }
}

bool XskIsQueryAcceptable(const XskPacket& packet, ClientState& clientState, LocalHolders& holders, bool& expectProxyProtocol)
{
  const auto& from = packet.getFromAddr();
  expectProxyProtocol = expectProxyProtocolFrom(from);
  if (!holders.acl->match(from) && !expectProxyProtocol) {
    vinfolog("Query from %s dropped because of ACL", from.toStringWithPort());
    ++dnsdist::metrics::g_stats.aclDrops;
    return false;
  }
  clientState.queries++;
  ++dnsdist::metrics::g_stats.queries;

  return true;
}

void XskRouter(std::shared_ptr<XskSocket> xsk)
{
  setThreadName("dnsdist/XskRouter");
  uint32_t failed = 0;
  // packets to be submitted for sending
  vector<XskPacket> fillInTx;
  const auto& fds = xsk->getDescriptors();
  // list of workers that need to be notified
  std::set<int> needNotify;
  while (true) {
    try {
      auto ready = xsk->wait(-1);
      // descriptor 0 gets incoming AF_XDP packets
      if ((fds.at(0).revents & POLLIN) != 0) {
        auto packets = xsk->recv(64, &failed);
        dnsdist::metrics::g_stats.nonCompliantQueries += failed;
        for (auto& packet : packets) {
          const auto dest = packet.getToAddr();
          auto worker = xsk->getWorkerByDestination(dest);
          if (!worker) {
            xsk->markAsFree(packet);
            continue;
          }
          worker->pushToProcessingQueue(packet);
          needNotify.insert(worker->workerWaker.getHandle());
        }
        for (auto socket : needNotify) {
          uint64_t value = 1;
          auto written = write(socket, &value, sizeof(value));
          if (written != sizeof(value)) {
            // oh, well, the worker is clearly overloaded
            // but there is nothing we can do about it,
            // and hopefully the queue will be processed eventually
          }
        }
        needNotify.clear();
        ready--;
      }
      for (size_t fdIndex = 1; fdIndex < fds.size() && ready > 0; fdIndex++) {
        if ((fds.at(fdIndex).revents & POLLIN) != 0) {
          ready--;
          const auto& info = xsk->getWorkerByDescriptor(fds.at(fdIndex).fd);
          info->processOutgoingFrames([&](XskPacket& packet) {
            if ((packet.getFlags() & XskPacket::UPDATE) == 0) {
              xsk->markAsFree(packet);
              return;
            }
            if ((packet.getFlags() & XskPacket::DELAY) != 0) {
              xsk->pushDelayed(packet);
              return;
            }
            fillInTx.push_back(packet);
          });
          info->cleanWorkerNotification();
        }
      }
      xsk->pickUpReadyPacket(fillInTx);
      xsk->recycle(4096);
      xsk->fillFq();
      xsk->send(fillInTx);
    }
    catch (...) {
      vinfolog("Exception in XSK router loop");
    }
  }
}

void XskClientThread(ClientState* clientState)
{
  setThreadName("dnsdist/xskClient");
  auto xskInfo = clientState->xskInfo;
  LocalHolders holders;

  for (;;) {
    while (!xskInfo->hasIncomingFrames()) {
      xskInfo->waitForXskSocket();
    }
    xskInfo->processIncomingFrames([&](XskPacket& packet) {
      if (XskProcessQuery(*clientState, holders, packet)) {
        packet.updatePacket();
        xskInfo->pushToSendQueue(packet);
      }
      else {
        xskInfo->markAsFree(packet);
      }
    });
    xskInfo->notifyXskSocket();
  }
}

static std::string getDestinationMap(bool isV6)
{
  return !isV6 ? "/sys/fs/bpf/dnsdist/xsk-destinations-v4" : "/sys/fs/bpf/dnsdist/xsk-destinations-v6";
}

void addDestinationAddress(const ComboAddress& addr)
{
  auto map = getDestinationMap(addr.isIPv6());
  XskSocket::addDestinationAddress(map, addr);
}

void removeDestinationAddress(const ComboAddress& addr)
{
  auto map = getDestinationMap(addr.isIPv6());
  XskSocket::removeDestinationAddress(map, addr);
}

void clearDestinationAddresses()
{
  auto map = getDestinationMap(false);
  XskSocket::clearDestinationMap(map, false);
  map = getDestinationMap(true);
  XskSocket::clearDestinationMap(map, true);
}

}
#endif /* HAVE_XSK */

1	/*
2	* This file is part of PowerDNS or dnsdist.
3	* Copyright -- PowerDNS.COM B.V. and its contributors
4	*
5	* This program is free software; you can redistribute it and/or modify
6	* it under the terms of version 2 of the GNU General Public License as
7	* published by the Free Software Foundation.
8	*
9	* In addition, for the avoidance of any doubt, permission is granted to
10	* link this program with OpenSSL and to (re)distribute the binaries
11	* produced as the result of such linking.
12	*
13	* This program is distributed in the hope that it will be useful,
14	* but WITHOUT ANY WARRANTY; without even the implied warranty of
15	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16	* GNU General Public License for more details.
17	*
18	* You should have received a copy of the GNU General Public License
19	* along with this program; if not, write to the Free Software
20	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
21	*/
22	#include "dnsdist.hh"
23	#include "dnsdist-xsk.hh"
24
25	#ifdef HAVE_XSK
26	#include <sys/poll.h>
27
28	#include "dolog.hh"
29	#include "dnsdist-metrics.hh"
30	#include "dnsdist-proxy-protocol.hh"
31	#include "threadname.hh"
32	#include "xsk.hh"
33
34	namespace dnsdist::xsk
35	{
36	std::vector<std::shared_ptr<XskSocket>> g_xsk;
37
38	void XskResponderThread(std::shared_ptr<DownstreamState> dss, std::shared_ptr<XskWorker> xskInfo)
39	{	×
40	try {	×
41	setThreadName("dnsdist/XskResp");	×
42	auto localRespRuleActions = g_respruleactions.getLocal();	×
43	auto localCacheInsertedRespRuleActions = g_cacheInsertedRespRuleActions.getLocal();	×
44	auto pollfds = getPollFdsForWorker(*xskInfo);	×
45	while (!dss->isStopped()) {	×
46	poll(pollfds.data(), pollfds.size(), -1);	×
47	bool needNotify = false;	×
48	if ((pollfds[0].revents & POLLIN) != 0) {	×
49	needNotify = true;	×
50	xskInfo->cleanSocketNotification();	×
NEW 51	xskInfo->processIncomingFrames([&](XskPacket& packet) {	×
UNCOV 52	if (packet.getDataLen() < sizeof(dnsheader)) {	×
53	xskInfo->markAsFree(packet);	×
54	return;	×
55	}	×
56	const dnsheader_aligned dnsHeader(packet.getPayloadData());	×
57	const auto queryId = dnsHeader->id;	×
58	auto ids = dss->getState(queryId);	×
59	if (ids) {	×
60	if (!ids->isXSK()) {	×
61	dss->restoreState(queryId, std::move(*ids));	×
62	ids = std::nullopt;	×
63	}	×
64	}	×
65	if (!ids) {	×
66	xskInfo->markAsFree(packet);	×
67	return;	×
68	}	×
69	auto response = packet.clonePacketBuffer();	×
70	if (response.size() > packet.getCapacity()) {	×
71	/* fallback to sending the packet via normal socket */
72	ids->xskPacketHeader.clear();	×
73	}	×
74	if (!processResponderPacket(dss, response, localRespRuleActions, localCacheInsertedRespRuleActions, std::move(*ids))) {	×
75	xskInfo->markAsFree(packet);	×
NEW 76	vinfolog("XSK packet dropped because processResponderPacket failed");	×
77	return;	×
78	}	×
79	if (response.size() > packet.getCapacity()) {	×
80	/* fallback to sending the packet via normal socket */
81	sendUDPResponse(ids->cs->udpFD, response, ids->delayMsec, ids->hopLocal, ids->hopRemote);	×
82	infolog("XSK packet falling back because packet is too large");	×
83	xskInfo->markAsFree(packet);	×
84	return;	×
85	}	×
86	packet.setHeader(ids->xskPacketHeader);	×
87	if (!packet.setPayload(response)) {	×
88	infolog("Unable to set XSK payload !");	×
89	}	×
90	if (ids->delayMsec > 0) {	×
91	packet.addDelay(ids->delayMsec);	×
92	}	×
93	packet.updatePacket();	×
94	xskInfo->pushToSendQueue(packet);	×
95	});	×
96	}	×
97	if (needNotify) {	×
98	xskInfo->notifyXskSocket();	×
99	}	×
100	}	×
101	}	×
102	catch (const std::exception& e) {	×
103	errlog("XSK responder thread died because of exception: %s", e.what());	×
104	}	×
105	catch (const PDNSException& e) {	×
106	errlog("XSK responder thread died because of PowerDNS exception: %s", e.reason);	×
107	}	×
108	catch (...) {	×
109	errlog("XSK responder thread died because of an exception: %s", "unknown");	×
110	}	×
111	}	×
112
113	bool XskIsQueryAcceptable(const XskPacket& packet, ClientState& clientState, LocalHolders& holders, bool& expectProxyProtocol)
114	{	×
115	const auto& from = packet.getFromAddr();	×
116	expectProxyProtocol = expectProxyProtocolFrom(from);	×
117	if (!holders.acl->match(from) && !expectProxyProtocol) {	×
118	vinfolog("Query from %s dropped because of ACL", from.toStringWithPort());	×
119	++dnsdist::metrics::g_stats.aclDrops;	×
120	return false;	×
121	}	×
122	clientState.queries++;	×
123	++dnsdist::metrics::g_stats.queries;	×
124
125	return true;	×
126	}	×
127
128	void XskRouter(std::shared_ptr<XskSocket> xsk)
129	{	×
130	setThreadName("dnsdist/XskRouter");	×
131	uint32_t failed = 0;	×
132	// packets to be submitted for sending
133	vector<XskPacket> fillInTx;	×
134	const auto& fds = xsk->getDescriptors();	×
135	// list of workers that need to be notified
136	std::set<int> needNotify;	×
137	while (true) {	×
138	try {	×
139	auto ready = xsk->wait(-1);	×
140	// descriptor 0 gets incoming AF_XDP packets
141	if ((fds.at(0).revents & POLLIN) != 0) {	×
142	auto packets = xsk->recv(64, &failed);	×
143	dnsdist::metrics::g_stats.nonCompliantQueries += failed;	×
144	for (auto& packet : packets) {	×
145	const auto dest = packet.getToAddr();	×
146	auto worker = xsk->getWorkerByDestination(dest);	×
147	if (!worker) {	×
148	xsk->markAsFree(packet);	×
149	continue;	×
150	}	×
151	worker->pushToProcessingQueue(packet);	×
152	needNotify.insert(worker->workerWaker.getHandle());	×
153	}	×
154	for (auto socket : needNotify) {	×
155	uint64_t value = 1;	×
156	auto written = write(socket, &value, sizeof(value));	×
157	if (written != sizeof(value)) {	×
158	// oh, well, the worker is clearly overloaded
159	// but there is nothing we can do about it,
160	// and hopefully the queue will be processed eventually
161	}	×
162	}	×
163	needNotify.clear();	×
164	ready--;	×
165	}	×
166	for (size_t fdIndex = 1; fdIndex < fds.size() && ready > 0; fdIndex++) {	×
167	if ((fds.at(fdIndex).revents & POLLIN) != 0) {	×
168	ready--;	×
169	const auto& info = xsk->getWorkerByDescriptor(fds.at(fdIndex).fd);	×
NEW 170	info->processOutgoingFrames([&](XskPacket& packet) {	×
UNCOV 171	if ((packet.getFlags() & XskPacket::UPDATE) == 0) {	×
172	xsk->markAsFree(packet);	×
173	return;	×
174	}	×
175	if ((packet.getFlags() & XskPacket::DELAY) != 0) {	×
176	xsk->pushDelayed(packet);	×
177	return;	×
178	}	×
179	fillInTx.push_back(packet);	×
180	});	×
181	info->cleanWorkerNotification();	×
182	}	×
183	}	×
184	xsk->pickUpReadyPacket(fillInTx);	×
185	xsk->recycle(4096);	×
186	xsk->fillFq();	×
187	xsk->send(fillInTx);	×
188	}	×
189	catch (...) {	×
190	vinfolog("Exception in XSK router loop");	×
191	}	×
192	}	×
193	}	×
194
195	void XskClientThread(ClientState* clientState)
196	{	×
197	setThreadName("dnsdist/xskClient");	×
198	auto xskInfo = clientState->xskInfo;	×
199	LocalHolders holders;	×
200
201	for (;;) {	×
NEW 202	while (!xskInfo->hasIncomingFrames()) {	×
UNCOV 203	xskInfo->waitForXskSocket();	×
204	}	×
NEW 205	xskInfo->processIncomingFrames([&](XskPacket& packet) {	×
UNCOV 206	if (XskProcessQuery(*clientState, holders, packet)) {	×
207	packet.updatePacket();	×
208	xskInfo->pushToSendQueue(packet);	×
209	}	×
210	else {	×
211	xskInfo->markAsFree(packet);	×
212	}	×
213	});	×
214	xskInfo->notifyXskSocket();	×
215	}	×
216	}	×
217
218	static std::string getDestinationMap(bool isV6)
219	{	631✔
220	return !isV6 ? "/sys/fs/bpf/dnsdist/xsk-destinations-v4" : "/sys/fs/bpf/dnsdist/xsk-destinations-v6";	631!
221	}	631✔
222
223	void addDestinationAddress(const ComboAddress& addr)
224	{	×
225	auto map = getDestinationMap(addr.isIPv6());	×
226	XskSocket::addDestinationAddress(map, addr);	×
227	}	×
228
229	void removeDestinationAddress(const ComboAddress& addr)
230	{	×
231	auto map = getDestinationMap(addr.isIPv6());	×
232	XskSocket::removeDestinationAddress(map, addr);	×
233	}	×
234
235	void clearDestinationAddresses()
236	{	631✔
237	auto map = getDestinationMap(false);	631✔
238	XskSocket::clearDestinationMap(map, false);	631✔
239	map = getDestinationMap(true);	631✔
240	XskSocket::clearDestinationMap(map, true);	631✔
241	}	631✔
242
243	}
244	#endif /* HAVE_XSK */

PowerDNS / pdns / 9936897972

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous