8340760366

Committed 19 Mar 2024 09:40AM UTC coverage: 59.238% (-0.005%) from 59.243%

Build # 8340760366

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #13834 from rgacogne/ddist-fix-first-lazy-health-check-failure

dnsdist: Properly handle a failure of the first lazy health-check

Run Details

32517 of 87978 branches covered (36.96%)

Branch coverage included in aggregate %.

8 of 14 new or added lines in 3 files covered. (57.14%)

52 existing lines in 11 files now uncovered.

113616 of 158712 relevant lines covered (71.59%)

3477602.24 hits per line

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

82.21

/pdns/dnsdistdist/dnsdist-async.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-async.hh"
#include "dnsdist-internal-queries.hh"
#include "dolog.hh"
#include "threadname.hh"

namespace dnsdist
{

AsynchronousHolder::Data::Data(bool failOpen) :
  d_failOpen(failOpen)
{
  auto [notifier, waiter] = pdns::channel::createNotificationQueue(true);
  // NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer): how I am supposed to do that?
  d_waiter = std::move(waiter);
  // NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer): how I am supposed to do that?
  d_notifier = std::move(notifier);
}

AsynchronousHolder::AsynchronousHolder(bool failOpen) :
  d_data(std::make_shared<Data>(failOpen))
{
  std::thread main([data = this->d_data] { mainThread(data); });
  main.detach();
}

AsynchronousHolder::~AsynchronousHolder()
{
  try {
    stop();
  }
  catch (...) {
  }
}

bool AsynchronousHolder::notify() const
{
  return d_data->d_notifier.notify();
}

bool AsynchronousHolder::wait(AsynchronousHolder::Data& data, FDMultiplexer& mplexer, std::vector<int>& readyFDs, int atMostMs)
{
  readyFDs.clear();
  mplexer.getAvailableFDs(readyFDs, atMostMs);
  if (readyFDs.empty()) {
    /* timeout */
    return true;
  }

  data.d_waiter.clear();
  return false;
}

void AsynchronousHolder::stop()
{
  {
    auto content = d_data->d_content.lock();
    d_data->d_done = true;
  }

  notify();
}

// NOLINTNEXTLINE(performance-unnecessary-value-param): this is a long-lived thread, and we want to make sure the reference count of the shared pointer has been increased
void AsynchronousHolder::mainThread(std::shared_ptr<Data> data)
{
  setThreadName("dnsdist/async");
  struct timeval now
  {
  };
  std::list<std::pair<uint16_t, std::unique_ptr<CrossProtocolQuery>>> expiredEvents;

  auto mplexer = std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent(1));
  mplexer->addReadFD(data->d_waiter.getDescriptor(), [](int, FDMultiplexer::funcparam_t&) {});
  std::vector<int> readyFDs;

  while (true) {
    bool shouldWait = true;
    int timeout = -1;
    {
      auto content = data->d_content.lock();
      if (data->d_done) {
        return;
      }

      if (!content->empty()) {
        gettimeofday(&now, nullptr);
        struct timeval next = getNextTTD(*content);
        if (next <= now) {
          pickupExpired(*content, now, expiredEvents);
          shouldWait = false;
        }
        else {
          auto remainingUsec = uSec(next - now);
          timeout = static_cast<int>(std::round(static_cast<double>(remainingUsec) / 1000.0));
          if (timeout == 0 && remainingUsec > 0) {
            /* if we have less than 1 ms, let's wait at least 1 ms */
            timeout = 1;
          }
        }
      }
    }

    if (shouldWait) {
      auto timedOut = wait(*data, *mplexer, readyFDs, timeout);
      if (timedOut) {
        auto content = data->d_content.lock();
        gettimeofday(&now, nullptr);
        pickupExpired(*content, now, expiredEvents);
      }
    }

    while (!expiredEvents.empty()) {
      auto [queryID, query] = std::move(expiredEvents.front());
      expiredEvents.pop_front();
      if (!data->d_failOpen) {
        vinfolog("Asynchronous query %d has expired at %d.%d, notifying the sender", queryID, now.tv_sec, now.tv_usec);
        auto sender = query->getTCPQuerySender();
        if (sender) {
          TCPResponse tresponse(std::move(query->query));
          sender->notifyIOError(now, std::move(tresponse));
        }
      }
      else {
        vinfolog("Asynchronous query %d has expired at %d.%d, resuming", queryID, now.tv_sec, now.tv_usec);
        resumeQuery(std::move(query));
      }
    }
  }
}

void AsynchronousHolder::push(uint16_t asyncID, uint16_t queryID, const struct timeval& ttd, std::unique_ptr<CrossProtocolQuery>&& query)
{
  bool needNotify = false;
  {
    auto content = d_data->d_content.lock();
    if (!content->empty()) {
      /* the thread is already waiting on a TTD expiry in addition to notifications,
         let's not wake it unless our TTD comes before the current one */
      const struct timeval next = getNextTTD(*content);
      if (ttd < next) {
        needNotify = true;
      }
    }
    else {
      /* the thread is currently only waiting for a notify */
      needNotify = true;
    }
    content->insert({std::move(query), ttd, asyncID, queryID});
  }

  if (needNotify) {
    notify();
  }
}

std::unique_ptr<CrossProtocolQuery> AsynchronousHolder::get(uint16_t asyncID, uint16_t queryID)
{
  /* no need to notify, worst case the thread wakes up for nothing because this was the next TTD */
  auto content = d_data->d_content.lock();
  auto contentIt = content->find(std::tie(queryID, asyncID));
  if (contentIt == content->end()) {
    struct timeval now
    {
    };
    gettimeofday(&now, nullptr);
    vinfolog("Asynchronous object %d not found at %d.%d", queryID, now.tv_sec, now.tv_usec);
    return nullptr;
  }

  auto result = std::move(contentIt->d_query);
  content->erase(contentIt);
  return result;
}

void AsynchronousHolder::pickupExpired(content_t& content, const struct timeval& now, std::list<std::pair<uint16_t, std::unique_ptr<CrossProtocolQuery>>>& events)
{
  auto& idx = content.get<TTDTag>();
  for (auto contentIt = idx.begin(); contentIt != idx.end() && contentIt->d_ttd < now;) {
    events.emplace_back(contentIt->d_queryID, std::move(contentIt->d_query));
    contentIt = idx.erase(contentIt);
  }
}

struct timeval AsynchronousHolder::getNextTTD(const content_t& content)
{
  if (content.empty()) {
    throw std::runtime_error("AsynchronousHolder::getNextTTD() called on an empty holder");
  }

  return content.get<TTDTag>().begin()->d_ttd;
}

bool AsynchronousHolder::empty()
{
  return d_data->d_content.read_only_lock()->empty();
}

static bool resumeResponse(std::unique_ptr<CrossProtocolQuery>&& response)
{
  try {
    auto& ids = response->query.d_idstate;
    DNSResponse dnsResponse = response->getDR();

    LocalHolders holders;
    auto result = processResponseAfterRules(response->query.d_buffer, *holders.cacheInsertedRespRuleActions, dnsResponse, ids.cs->muted);
    if (!result) {
      /* easy */
      return true;
    }

    auto sender = response->getTCPQuerySender();
    if (sender) {
      struct timeval now
      {
      };
      gettimeofday(&now, nullptr);

      TCPResponse resp(std::move(response->query.d_buffer), std::move(response->query.d_idstate), nullptr, response->downstream);
      resp.d_async = true;
      sender->handleResponse(now, std::move(resp));
    }
  }
  catch (const std::exception& e) {
    vinfolog("Got exception while resuming cross-protocol response: %s", e.what());
    return false;
  }

  return true;
}

static LockGuarded<std::deque<std::unique_ptr<CrossProtocolQuery>>> s_asynchronousEventsQueue;

bool queueQueryResumptionEvent(std::unique_ptr<CrossProtocolQuery>&& query)
{
  s_asynchronousEventsQueue.lock()->push_back(std::move(query));
  return true;
}

void handleQueuedAsynchronousEvents()
{
  while (true) {
    std::unique_ptr<CrossProtocolQuery> query;
    {
      // we do not want to hold the lock while resuming
      auto queue = s_asynchronousEventsQueue.lock();
      if (queue->empty()) {
        return;
      }

      query = std::move(queue->front());
      queue->pop_front();
    }
    if (query && !resumeQuery(std::move(query))) {
      vinfolog("Unable to resume asynchronous query event");
    }
  }
}

bool resumeQuery(std::unique_ptr<CrossProtocolQuery>&& query)
{
  if (query->d_isResponse) {
    return resumeResponse(std::move(query));
  }

  DNSQuestion dnsQuestion = query->getDQ();
  LocalHolders holders;

  auto result = processQueryAfterRules(dnsQuestion, holders, query->downstream);
  if (result == ProcessQueryResult::Drop) {
    /* easy */
    return true;
  }
  if (result == ProcessQueryResult::PassToBackend) {
    if (query->downstream == nullptr) {
      return false;
    }

#ifdef HAVE_DNS_OVER_HTTPS
    if (dnsQuestion.ids.du != nullptr) {
      dnsQuestion.ids.du->downstream = query->downstream;
    }
#endif

    if (query->downstream->isTCPOnly() || !(dnsQuestion.getProtocol().isUDP() || dnsQuestion.getProtocol() == dnsdist::Protocol::DoH)) {
      query->downstream->passCrossProtocolQuery(std::move(query));
      return true;
    }

    auto queryID = dnsQuestion.getHeader()->id;
    /* at this point 'du', if it is not nullptr, is owned by the DoHCrossProtocolQuery
       which will stop existing when we return, so we need to increment the reference count
    */
    return assignOutgoingUDPQueryToBackend(query->downstream, queryID, dnsQuestion, query->query.d_buffer);
  }
  if (result == ProcessQueryResult::SendAnswer) {
    auto sender = query->getTCPQuerySender();
    if (!sender) {
      return false;
    }

    struct timeval now
    {
    };
    gettimeofday(&now, nullptr);

    TCPResponse response(std::move(query->query.d_buffer), std::move(query->query.d_idstate), nullptr, query->downstream);
    response.d_async = true;
    response.d_idstate.selfGenerated = true;

    try {
      sender->handleResponse(now, std::move(response));
      return true;
    }
    catch (const std::exception& e) {
      vinfolog("Got exception while resuming cross-protocol self-answered query: %s", e.what());
      return false;
    }
  }
  if (result == ProcessQueryResult::Asynchronous) {
    /* nope */
    errlog("processQueryAfterRules returned 'asynchronous' while trying to resume an already asynchronous query");
    return false;
  }

  return false;
}

bool suspendQuery(DNSQuestion& dnsQuestion, uint16_t asyncID, uint16_t queryID, uint32_t timeoutMs)
{
  if (!g_asyncHolder) {
    return false;
  }

  struct timeval now
  {
  };
  gettimeofday(&now, nullptr);
  struct timeval ttd = now;
  ttd.tv_sec += timeoutMs / 1000;
  ttd.tv_usec += static_cast<decltype(ttd.tv_usec)>((timeoutMs % 1000) * 1000);
  normalizeTV(ttd);

  vinfolog("Suspending asynchronous query %d at %d.%d until %d.%d", queryID, now.tv_sec, now.tv_usec, ttd.tv_sec, ttd.tv_usec);
  auto query = getInternalQueryFromDQ(dnsQuestion, false);

  g_asyncHolder->push(asyncID, queryID, ttd, std::move(query));
  return true;
}

bool suspendResponse(DNSResponse& dnsResponse, uint16_t asyncID, uint16_t queryID, uint32_t timeoutMs)
{
  if (!g_asyncHolder) {
    return false;
  }

  struct timeval now
  {
  };
  gettimeofday(&now, nullptr);
  struct timeval ttd = now;
  ttd.tv_sec += timeoutMs / 1000;
  ttd.tv_usec += static_cast<decltype(ttd.tv_usec)>((timeoutMs % 1000) * 1000);
  normalizeTV(ttd);

  vinfolog("Suspending asynchronous response %d at %d.%d until %d.%d", queryID, now.tv_sec, now.tv_usec, ttd.tv_sec, ttd.tv_usec);
  auto query = getInternalQueryFromDQ(dnsResponse, true);
  query->d_isResponse = true;
  query->downstream = dnsResponse.d_downstream;

  g_asyncHolder->push(asyncID, queryID, ttd, std::move(query));
  return true;
}

std::unique_ptr<AsynchronousHolder> g_asyncHolder;
}

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-async.hh"
23	#include "dnsdist-internal-queries.hh"
24	#include "dolog.hh"
25	#include "threadname.hh"
26
27	namespace dnsdist
28	{
29
30	AsynchronousHolder::Data::Data(bool failOpen) :
31	d_failOpen(failOpen)
32	{	308✔
33	auto [notifier, waiter] = pdns::channel::createNotificationQueue(true);	308✔
34	// NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer): how I am supposed to do that?
35	d_waiter = std::move(waiter);	308✔
36	// NOLINTNEXTLINE(cppcoreguidelines-prefer-member-initializer): how I am supposed to do that?
37	d_notifier = std::move(notifier);	308✔
38	}	308✔
39
40	AsynchronousHolder::AsynchronousHolder(bool failOpen) :
41	d_data(std::make_shared<Data>(failOpen))
42	{	308✔
43	std::thread main([data = this->d_data] { mainThread(data); });	308✔
44	main.detach();	308✔
45	}	308✔
46
47	AsynchronousHolder::~AsynchronousHolder()
48	{	6✔
49	try {	6✔
50	stop();	6✔
51	}	6✔
52	catch (...) {	6✔
53	}	×
54	}	6✔
55
56	bool AsynchronousHolder::notify() const
57	{	386✔
58	return d_data->d_notifier.notify();	386✔
59	}	386✔
60
61	bool AsynchronousHolder::wait(AsynchronousHolder::Data& data, FDMultiplexer& mplexer, std::vector<int>& readyFDs, int atMostMs)
62	{	742✔
63	readyFDs.clear();	742✔
64	mplexer.getAvailableFDs(readyFDs, atMostMs);	742✔
65	if (readyFDs.empty()) {	742✔
66	/* timeout */
67	return true;	60✔
68	}	60✔
69
70	data.d_waiter.clear();	682✔
71	return false;	682✔
72	}	742✔
73
74	void AsynchronousHolder::stop()
75	{	12✔
76	{	12✔
77	auto content = d_data->d_content.lock();	12✔
78	d_data->d_done = true;	12✔
79	}	12✔
80
81	notify();	12✔
82	}	12✔
83
84	// NOLINTNEXTLINE(performance-unnecessary-value-param): this is a long-lived thread, and we want to make sure the reference count of the shared pointer has been increased
85	void AsynchronousHolder::mainThread(std::shared_ptr<Data> data)
86	{	308✔
87	setThreadName("dnsdist/async");	308✔
88	struct timeval now	308✔
89	{	308✔
90	};	308✔
91	std::list<std::pair<uint16_t, std::unique_ptr<CrossProtocolQuery>>> expiredEvents;	308✔
92
93	auto mplexer = std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent(1));	308✔
94	mplexer->addReadFD(data->d_waiter.getDescriptor(), [](int, FDMultiplexer::funcparam_t&) {});	308✔
95	std::vector<int> readyFDs;	308✔
96
97	while (true) {	1,052✔
98	bool shouldWait = true;	750✔
99	int timeout = -1;	750✔
100	{	750✔
101	auto content = data->d_content.lock();	750✔
102	if (data->d_done) {	750✔
103	return;	6✔
104	}	6✔
105
106	if (!content->empty()) {	744✔
107	gettimeofday(&now, nullptr);	373✔
108	struct timeval next = getNextTTD(*content);	373✔
109	if (next <= now) {	373✔
110	pickupExpired(*content, now, expiredEvents);	2✔
111	shouldWait = false;	2✔
112	}	2✔
113	else {	371✔
114	auto remainingUsec = uSec(next - now);	371✔
115	timeout = static_cast<int>(std::round(static_cast<double>(remainingUsec) / 1000.0));	371✔
116	if (timeout == 0 && remainingUsec > 0) {	371!
117	/* if we have less than 1 ms, let's wait at least 1 ms */
UNCOV 118	timeout = 1;	×
UNCOV 119	}	×
120	}	371✔
121	}	373✔
122	}	744✔
123
124	if (shouldWait) {	744✔
125	auto timedOut = wait(data, mplexer, readyFDs, timeout);	742✔
126	if (timedOut) {	742✔
127	auto content = data->d_content.lock();	60✔
128	gettimeofday(&now, nullptr);	60✔
129	pickupExpired(*content, now, expiredEvents);	60✔
130	}	60✔
131	}	742✔
132
133	while (!expiredEvents.empty()) {	800✔
134	auto [queryID, query] = std::move(expiredEvents.front());	56✔
135	expiredEvents.pop_front();	56✔
136	if (!data->d_failOpen) {	56✔
137	vinfolog("Asynchronous query %d has expired at %d.%d, notifying the sender", queryID, now.tv_sec, now.tv_usec);	4!
138	auto sender = query->getTCPQuerySender();	4✔
139	if (sender) {	4!
140	TCPResponse tresponse(std::move(query->query));	4✔
141	sender->notifyIOError(now, std::move(tresponse));	4✔
142	}	4✔
143	}	4✔
144	else {	52✔
145	vinfolog("Asynchronous query %d has expired at %d.%d, resuming", queryID, now.tv_sec, now.tv_usec);	52!
146	resumeQuery(std::move(query));	52✔
147	}	52✔
148	}	56✔
149	}	744✔
150	}	308✔
151
152	void AsynchronousHolder::push(uint16_t asyncID, uint16_t queryID, const struct timeval& ttd, std::unique_ptr<CrossProtocolQuery>&& query)
153	{	374✔
154	bool needNotify = false;	374✔
155	{	374✔
156	auto content = d_data->d_content.lock();	374✔
157	if (!content->empty()) {	374!
158	/* the thread is already waiting on a TTD expiry in addition to notifications,
159	let's not wake it unless our TTD comes before the current one */
160	const struct timeval next = getNextTTD(*content);	×
161	if (ttd < next) {	×
162	needNotify = true;	×
163	}	×
164	}	×
165	else {	374✔
166	/* the thread is currently only waiting for a notify */
167	needNotify = true;	374✔
168	}	374✔
169	content->insert({std::move(query), ttd, asyncID, queryID});	374✔
170	}	374✔
171
172	if (needNotify) {	374!
173	notify();	374✔
174	}	374✔
175	}	374✔
176
177	std::unique_ptr<CrossProtocolQuery> AsynchronousHolder::get(uint16_t asyncID, uint16_t queryID)
178	{	324✔
179	/* no need to notify, worst case the thread wakes up for nothing because this was the next TTD */
180	auto content = d_data->d_content.lock();	324✔
181	auto contentIt = content->find(std::tie(queryID, asyncID));	324✔
182	if (contentIt == content->end()) {	324✔
183	struct timeval now	6✔
184	{	6✔
185	};	6✔
186	gettimeofday(&now, nullptr);	6✔
187	vinfolog("Asynchronous object %d not found at %d.%d", queryID, now.tv_sec, now.tv_usec);	6!
188	return nullptr;	6✔
189	}	6✔
190
191	auto result = std::move(contentIt->d_query);	318✔
192	content->erase(contentIt);	318✔
193	return result;	318✔
194	}	324✔
195
196	void AsynchronousHolder::pickupExpired(content_t& content, const struct timeval& now, std::list<std::pair<uint16_t, std::unique_ptr<CrossProtocolQuery>>>& events)
197	{	62✔
198	auto& idx = content.get<TTDTag>();	62✔
199	for (auto contentIt = idx.begin(); contentIt != idx.end() && contentIt->d_ttd < now;) {	118!
200	events.emplace_back(contentIt->d_queryID, std::move(contentIt->d_query));	56✔
201	contentIt = idx.erase(contentIt);	56✔
202	}	56✔
203	}	62✔
204
205	struct timeval AsynchronousHolder::getNextTTD(const content_t& content)
206	{	373✔
207	if (content.empty()) {	373!
208	throw std::runtime_error("AsynchronousHolder::getNextTTD() called on an empty holder");	×
209	}	×
210
211	return content.get<TTDTag>().begin()->d_ttd;	373✔
212	}	373✔
213
214	bool AsynchronousHolder::empty()
215	{	22✔
216	return d_data->d_content.read_only_lock()->empty();	22✔
217	}	22✔
218
219	static bool resumeResponse(std::unique_ptr<CrossProtocolQuery>&& response)
220	{	142✔
221	try {	142✔
222	auto& ids = response->query.d_idstate;	142✔
223	DNSResponse dnsResponse = response->getDR();	142✔
224
225	LocalHolders holders;	142✔
226	auto result = processResponseAfterRules(response->query.d_buffer, *holders.cacheInsertedRespRuleActions, dnsResponse, ids.cs->muted);	142✔
227	if (!result) {	142!
228	/* easy */
229	return true;	×
230	}	×
231
232	auto sender = response->getTCPQuerySender();	142✔
233	if (sender) {	142!
234	struct timeval now	142✔
235	{	142✔
236	};	142✔
237	gettimeofday(&now, nullptr);	142✔
238
239	TCPResponse resp(std::move(response->query.d_buffer), std::move(response->query.d_idstate), nullptr, response->downstream);	142✔
240	resp.d_async = true;	142✔
241	sender->handleResponse(now, std::move(resp));	142✔
242	}	142✔
243	}	142✔
244	catch (const std::exception& e) {	142✔
245	vinfolog("Got exception while resuming cross-protocol response: %s", e.what());	×
246	return false;	×
247	}	×
248
249	return true;	142✔
250	}	142✔
251
252	static LockGuarded<std::deque<std::unique_ptr<CrossProtocolQuery>>> s_asynchronousEventsQueue;
253
254	bool queueQueryResumptionEvent(std::unique_ptr<CrossProtocolQuery>&& query)
255	{	282✔
256	s_asynchronousEventsQueue.lock()->push_back(std::move(query));	282✔
257	return true;	282✔
258	}	282✔
259
260	void handleQueuedAsynchronousEvents()
261	{	966✔
262	while (true) {	1,248✔
263	std::unique_ptr<CrossProtocolQuery> query;	1,248✔
264	{	1,248✔
265	// we do not want to hold the lock while resuming
266	auto queue = s_asynchronousEventsQueue.lock();	1,248✔
267	if (queue->empty()) {	1,248✔
268	return;	966✔
269	}	966✔
270
271	query = std::move(queue->front());	282✔
272	queue->pop_front();	282✔
273	}	282✔
274	if (query && !resumeQuery(std::move(query))) {	282!
275	vinfolog("Unable to resume asynchronous query event");	×
276	}	×
277	}	282✔
278	}	966✔
279
280	bool resumeQuery(std::unique_ptr<CrossProtocolQuery>&& query)
281	{	334✔
282	if (query->d_isResponse) {	334✔
283	return resumeResponse(std::move(query));	142✔
284	}	142✔
285
286	DNSQuestion dnsQuestion = query->getDQ();	192✔
287	LocalHolders holders;	192✔
288
289	auto result = processQueryAfterRules(dnsQuestion, holders, query->downstream);	192✔
290	if (result == ProcessQueryResult::Drop) {	192✔
291	/* easy */
292	return true;	12✔
293	}	12✔
294	if (result == ProcessQueryResult::PassToBackend) {	180✔
295	if (query->downstream == nullptr) {	156!
296	return false;	×
297	}	×
298
299	#ifdef HAVE_DNS_OVER_HTTPS	156✔
300	if (dnsQuestion.ids.du != nullptr) {	156✔
301	dnsQuestion.ids.du->downstream = query->downstream;	54✔
302	}	54✔
303	#endif	156✔
304
305	if (query->downstream->isTCPOnly() \|\| !(dnsQuestion.getProtocol().isUDP() \|\| dnsQuestion.getProtocol() == dnsdist::Protocol::DoH)) {	156✔
306	query->downstream->passCrossProtocolQuery(std::move(query));	111✔
307	return true;	111✔
308	}	111✔
309
310	auto queryID = dnsQuestion.getHeader()->id;	45✔
311	/* at this point 'du', if it is not nullptr, is owned by the DoHCrossProtocolQuery
312	which will stop existing when we return, so we need to increment the reference count
313	*/
314	return assignOutgoingUDPQueryToBackend(query->downstream, queryID, dnsQuestion, query->query.d_buffer);	45✔
315	}	156✔
316	if (result == ProcessQueryResult::SendAnswer) {	24!
317	auto sender = query->getTCPQuerySender();	24✔
318	if (!sender) {	24!
319	return false;	×
320	}	×
321
322	struct timeval now	24✔
323	{	24✔
324	};	24✔
325	gettimeofday(&now, nullptr);	24✔
326
327	TCPResponse response(std::move(query->query.d_buffer), std::move(query->query.d_idstate), nullptr, query->downstream);	24✔
328	response.d_async = true;	24✔
329	response.d_idstate.selfGenerated = true;	24✔
330
331	try {	24✔
332	sender->handleResponse(now, std::move(response));	24✔
333	return true;	24✔
334	}	24✔
335	catch (const std::exception& e) {	24✔
336	vinfolog("Got exception while resuming cross-protocol self-answered query: %s", e.what());	×
337	return false;	×
338	}	×
339	}	24✔
340	if (result == ProcessQueryResult::Asynchronous) {	×
341	/* nope */
342	errlog("processQueryAfterRules returned 'asynchronous' while trying to resume an already asynchronous query");	×
343	return false;	×
344	}	×
345
346	return false;	×
347	}	×
348
349	bool suspendQuery(DNSQuestion& dnsQuestion, uint16_t asyncID, uint16_t queryID, uint32_t timeoutMs)
350	{	202✔
351	if (!g_asyncHolder) {	202!
352	return false;	×
353	}	×
354
355	struct timeval now	202✔
356	{	202✔
357	};	202✔
358	gettimeofday(&now, nullptr);	202✔
359	struct timeval ttd = now;	202✔
360	ttd.tv_sec += timeoutMs / 1000;	202✔
361	ttd.tv_usec += static_cast<decltype(ttd.tv_usec)>((timeoutMs % 1000) * 1000);	202✔
362	normalizeTV(ttd);	202✔
363
364	vinfolog("Suspending asynchronous query %d at %d.%d until %d.%d", queryID, now.tv_sec, now.tv_usec, ttd.tv_sec, ttd.tv_usec);	202!
365	auto query = getInternalQueryFromDQ(dnsQuestion, false);	202✔
366
367	g_asyncHolder->push(asyncID, queryID, ttd, std::move(query));	202✔
368	return true;	202✔
369	}	202✔
370
371	bool suspendResponse(DNSResponse& dnsResponse, uint16_t asyncID, uint16_t queryID, uint32_t timeoutMs)
372	{	166✔
373	if (!g_asyncHolder) {	166!
374	return false;	×
375	}	×
376
377	struct timeval now	166✔
378	{	166✔
379	};	166✔
380	gettimeofday(&now, nullptr);	166✔
381	struct timeval ttd = now;	166✔
382	ttd.tv_sec += timeoutMs / 1000;	166✔
383	ttd.tv_usec += static_cast<decltype(ttd.tv_usec)>((timeoutMs % 1000) * 1000);	166✔
384	normalizeTV(ttd);	166✔
385
386	vinfolog("Suspending asynchronous response %d at %d.%d until %d.%d", queryID, now.tv_sec, now.tv_usec, ttd.tv_sec, ttd.tv_usec);	166!
387	auto query = getInternalQueryFromDQ(dnsResponse, true);	166✔
388	query->d_isResponse = true;	166✔
389	query->downstream = dnsResponse.d_downstream;	166✔
390
391	g_asyncHolder->push(asyncID, queryID, ttd, std::move(query));	166✔
392	return true;	166✔
393	}	166✔
394
395	std::unique_ptr<AsynchronousHolder> g_asyncHolder;
396	}

PowerDNS / pdns / 8340760366

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