11663687396

Committed 04 Nov 2024 11:54AM UTC coverage: 64.645% (+0.03%) from 64.617%

Build # 11663687396

Build Type

Pull #14796

github

Committed by

web-flow

Commit Message

Merge bb3a8b078 into 5305f15a1

Pull Request Pull Request #14796: rec: provide a few more Lua functions, most importantly record cache dump and load

Run Details

37295 of 88476 branches covered (42.15%)

Branch coverage included in aggregate %.

333 of 431 new or added lines in 5 files covered. (77.26%)

683 existing lines in 19 files now uncovered.

125284 of 163018 relevant lines covered (76.85%)

4755082.38 hits per line

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

94.33

/pdns/recursordist/mtasker.hh

/*
 * 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.
 */
#pragma once
#include <cstdint>
#include <ctime>
#include <queue>
#include <memory>
#include <stack>

#include <boost/multi_index_container.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/key_extractors.hpp>

#include "namespaces.hh"
#include "misc.hh"
#include "mtasker_context.hh"

// #define MTASKERTIMING 1

//! The main MTasker class
/** The main MTasker class. See the main page for more information.
    \tparam EventKey Type of the key with which events are to be identified. Defaults to int.
    \tparam EventVal Type of the content or value of an event. Defaults to int. Cannot be set to void.
    \note The EventKey needs to have an operator< defined because it is used as the key of an associative array
*/

template <class EventKey = int, class EventVal = int, class Cmp = std::less<EventKey>>
class MTasker
{
public:
  struct Waiter
  {
    EventKey key;
    std::shared_ptr<pdns_ucontext_t> context;
    struct timeval ttd
    {
    };
    int tid{};
  };
  struct KeyTag
  {
  };

  using waiters_t = boost::multi_index::multi_index_container<
    Waiter,
    boost::multi_index::indexed_by<
      boost::multi_index::ordered_unique<boost::multi_index::member<Waiter, EventKey, &Waiter::key>, Cmp>,
      boost::multi_index::ordered_non_unique<boost::multi_index::tag<KeyTag>, boost::multi_index::member<Waiter, struct timeval, &Waiter::ttd>>>>;

  //! Constructor
  /** Constructor with a small default stacksize. If any of your threads exceeds this stack, your application will crash.
      This limit applies solely to the stack, the heap is not limited in any way. If threads need to allocate a lot of data,
      the use of new/delete is suggested.
   */
  MTasker(size_t stacksize = static_cast<size_t>(16 * 8192), size_t stackCacheSize = 0) :
    d_stacksize(stacksize), d_maxCachedStacks(stackCacheSize), d_waitstatus(Error)
  {
    initMainStackBounds();

    // make sure our stack is 16-byte aligned to make all the architectures happy
    d_stacksize = d_stacksize >> 4 << 4;
  }

  using tfunc_t = void(void*); //!< type of the pointer that starts a thread
  uint64_t nextWaiterDelayUsec(uint64_t defusecs);
  int waitEvent(EventKey& key, EventVal* val = nullptr, unsigned int timeoutMsec = 0, const struct timeval* now = nullptr);
  void yield();
  int sendEvent(const EventKey& key, const EventVal* val = nullptr);
  void makeThread(tfunc_t* start, void* val);
  bool schedule(const struct timeval& now);

  const waiters_t& getWaiters() const
  {
    return d_waiters;
  }

  //! gives access to the list of Events threads are waiting for
  /** The kernel can call this to get a list of Events threads are waiting for. This is very useful
      to setup 'select' or 'poll' or 'aio' events needed to satisfy these requests.
      getEvents clears the events parameter before filling it.

      \param events Vector which is to be filled with keys threads are waiting for
  */
  void getEvents(std::vector<EventKey>& events) const
  {
    events.clear();
    for (const auto& waiter : d_waiters) {
      events.emplace_back(waiter.key);
    }
  }

  //! returns true if there are no processes
  /** Call this to check if no processes are running anymore
      \return true if no processes are left
  */
  [[nodiscard]] bool noProcesses() const
  {
    return d_threadsCount == 0;
  }

  //! returns the number of processes running
  /** Call this to perhaps limit activities if too many threads are running
      \return number of processes running
  */
  [[nodiscard]] unsigned int numProcesses() const
  {
    return d_threadsCount;
  }

  //! Returns the current Thread ID (tid)
  /** Processes can call this to get a numerical representation of their current thread ID.
      This can be useful for logging purposes.
  */
  [[nodiscard]] int getTid() const
  {
    return d_tid;
  }

  //! Returns the maximum stack usage so far of this MThread
  [[nodiscard]] uint64_t getMaxStackUsage() const
  {
    return d_threads.at(d_tid).startOfStack - d_threads.at(d_tid).highestStackSeen;
  }

  //! Returns the maximum stack usage so far of this MThread
  [[nodiscard]] unsigned int getUsec() const
  {
#ifdef MTASKERTIMING
    return d_threads.at(d_tid).totTime + d_threads.at(d_tid).dt.ndiff() / 1000;
#else
    return 0;
#endif
  }

private:
  EventKey d_eventkey; // for waitEvent, contains exact key it was awoken for
  EventVal d_waitval;

  pdns_ucontext_t d_kernel;
  std::queue<int> d_runQueue;
  std::queue<int> d_zombiesQueue;

  struct ThreadInfo
  {
    std::shared_ptr<pdns_ucontext_t> context;
    std::function<void(void)> start;
    const char* startOfStack{};
    const char* highestStackSeen{};
#ifdef MTASKERTIMING
    CPUTime dt;
    unsigned int totTime;
#endif
  };

  using pdns_mtasker_stack_t = std::vector<char, lazy_allocator<char>>;
  using mthreads_t = std::map<int, ThreadInfo>;

  mthreads_t d_threads;
  std::stack<pdns_mtasker_stack_t> d_cachedStacks;
  waiters_t d_waiters;
  size_t d_stacksize;
  size_t d_threadsCount{0};
  size_t d_maxCachedStacks{0};
  int d_tid{0};
  int d_maxtid{0};

  enum waitstatusenum : int8_t
  {
    Error = -1,
    TimeOut = 0,
    Answer
  } d_waitstatus;

  std::shared_ptr<pdns_ucontext_t> getUContext();

  void initMainStackBounds()
  {
#ifdef HAVE_FIBER_SANITIZER

#ifdef HAVE_PTHREAD_GETATTR_NP
    pthread_attr_t attr;
    pthread_attr_init(&attr);
    pthread_getattr_np(pthread_self(), &attr);
    pthread_attr_getstack(&attr, &t_mainStack, &t_mainStackSize);
    pthread_attr_destroy(&attr);
#elif defined(HAVE_PTHREAD_GET_STACKSIZE_NP) && defined(HAVE_PTHREAD_GET_STACKADDR_NP)
    t_mainStack = pthread_get_stackaddr_np(pthread_self());
    t_mainStackSize = pthread_get_stacksize_np(pthread_self());
#else
#error Cannot determine stack size and base on this platform
#endif

#endif /* HAVE_FIBER_SANITIZER */
  }
};

#ifdef PDNS_USE_VALGRIND
#include <valgrind/valgrind.h>
#endif /* PDNS_USE_VALGRIND */

template <class EventKey, class EventVal, class Cmp>
uint64_t MTasker<EventKey, EventVal, Cmp>::nextWaiterDelayUsec(uint64_t defusecs)
{
  if (d_waiters.empty()) {
    // no waiters
    return defusecs;
  }
  auto& ttdindex = boost::multi_index::get<KeyTag>(d_waiters);
  auto iter = ttdindex.begin();
  timeval rnow{};
  gettimeofday(&rnow, nullptr);
  if (iter->ttd.tv_sec != 0) {
    // we have a waiter with a timeout specified
    if (rnow < iter->ttd) {
      // we should not wait longer than the default timeout
      return std::min(defusecs, uSec(iter->ttd - rnow));
    }
    // already expired
    return 0;
  }
  return defusecs;
}

//! puts a thread to sleep waiting until a specified event arrives
/** Threads can call waitEvent to register that they are waiting on an event with a certain key.
    If so desired, the event can carry data which is returned in val in case that is non-zero.

    Furthermore, a timeout can be specified in seconds.

    Only one thread can be waiting on a key, results of trying to have more threads
    waiting on the same key are undefined.

    \param key Event key to wait for. Needs to match up to a key reported to sendEvent
    \param val If non-zero, the value of the event will be stored in *val
    \param timeout If non-zero, number of seconds to wait for an event.

    \return returns -1 in case of error, 0 in case of timeout, 1 in case of an answer
*/
template <class EventKey, class EventVal, class Cmp>
int MTasker<EventKey, EventVal, Cmp>::waitEvent(EventKey& key, EventVal* val, unsigned int timeoutMsec, const struct timeval* now)
{
  if (d_waiters.count(key)) { // there was already an exact same waiter
    return -1;
  }

  Waiter waiter;
  waiter.context = std::make_shared<pdns_ucontext_t>();
  waiter.ttd.tv_sec = 0;
  waiter.ttd.tv_usec = 0;
  if (timeoutMsec != 0) {
    struct timeval increment
    {
    };
    increment.tv_sec = timeoutMsec / 1000;
    increment.tv_usec = static_cast<decltype(increment.tv_usec)>(1000 * (timeoutMsec % 1000));
    if (now != nullptr) {
      waiter.ttd = increment + *now;
    }
    else {
      struct timeval realnow
      {
      };
      gettimeofday(&realnow, nullptr);
      waiter.ttd = increment + realnow;
    }
  }

  waiter.tid = d_tid;
  waiter.key = key;

  d_waiters.insert(waiter);
#ifdef MTASKERTIMING
  unsigned int diff = d_threads[d_tid].dt.ndiff() / 1000;
  d_threads[d_tid].totTime += diff;
#endif
  notifyStackSwitchToKernel();
  pdns_swapcontext(*d_waiters.find(key)->context, d_kernel); // 'A' will return here when 'key' has arrived, hands over control to kernel first
  notifyStackSwitchDone();
#ifdef MTASKERTIMING
  d_threads[d_tid].dt.start();
#endif
  if (val && d_waitstatus == Answer) {
    *val = d_waitval;
  }
  d_tid = waiter.tid;
  if ((char*)&waiter < d_threads[d_tid].highestStackSeen) {
    d_threads[d_tid].highestStackSeen = (char*)&waiter;
  }
  key = d_eventkey;
  return d_waitstatus;
}

//! yields control to the kernel or other threads
/** Hands over control to the kernel, allowing other processes to run, or events to arrive */

template <class Key, class Val, class Cmp>
void MTasker<Key, Val, Cmp>::yield()
{
  d_runQueue.push(d_tid);
  notifyStackSwitchToKernel();
  pdns_swapcontext(*d_threads[d_tid].context, d_kernel); // give control to the kernel
  notifyStackSwitchDone();
}

//! reports that an event took place for which threads may be waiting
/** From the kernel loop, sendEvent can be called to report that something occurred for which there may be waiters.
    \param key Key of the event for which threads may be waiting
    \param val If non-zero, pointer to the content of the event
    \return Returns -1 in case of error, 0 if there were no waiters, 1 if a thread was woken up.

    WARNING: when passing val as zero, d_waitval is undefined, and hence waitEvent will return undefined!
*/
template <class EventKey, class EventVal, class Cmp>
int MTasker<EventKey, EventVal, Cmp>::sendEvent(const EventKey& key, const EventVal* val)
{
  typename waiters_t::iterator waiter = d_waiters.find(key);

  if (waiter == d_waiters.end()) {
    // cerr<<"Event sent nobody was waiting for! " <<key << endl;
    return 0;
  }
  d_waitstatus = Answer;
  if (val) {
    d_waitval = *val;
  }
  d_tid = waiter->tid; // set tid
  d_eventkey = waiter->key; // pass waitEvent the exact key it was woken for
  auto userspace = std::move(waiter->context);
  d_waiters.erase(waiter); // removes the waitpoint
  notifyStackSwitch(d_threads[d_tid].startOfStack, d_stacksize);
  try {
    pdns_swapcontext(d_kernel, *userspace); // swaps back to the above point 'A'
  }
  catch (...) {
    notifyStackSwitchDone();
    throw;
  }
  notifyStackSwitchDone();
  return 1;
}

template <class Key, class Val, class Cmp>
std::shared_ptr<pdns_ucontext_t> MTasker<Key, Val, Cmp>::getUContext()
{
  auto ucontext = std::make_shared<pdns_ucontext_t>();
  if (d_cachedStacks.empty()) {
    ucontext->uc_stack.resize(d_stacksize + 1);
  }
  else {
    ucontext->uc_stack = std::move(d_cachedStacks.top());
    d_cachedStacks.pop();
  }

  ucontext->uc_link = &d_kernel; // come back to kernel after dying

#ifdef PDNS_USE_VALGRIND
  ucontext->valgrind_id = VALGRIND_STACK_REGISTER(&ucontext->uc_stack[0], &ucontext->uc_stack[ucontext->uc_stack.size() - 1]);
#endif /* PDNS_USE_VALGRIND */

  return ucontext;
}

//! launches a new thread
/** The kernel can call this to make a new thread, which starts at the function start and gets passed the val void pointer.
    \param start Pointer to the function which will form the start of the thread
    \param val A void pointer that can be used to pass data to the thread
*/
template <class Key, class Val, class Cmp>
void MTasker<Key, Val, Cmp>::makeThread(tfunc_t* start, void* val)
{
  auto ucontext = getUContext();

  ++d_threadsCount;
  auto& thread = d_threads[d_maxtid];
  // we will get a better approximation when the task is executed, but that prevents notifying a stack at nullptr
  // on the first invocation
  d_threads[d_maxtid].startOfStack = &ucontext->uc_stack[ucontext->uc_stack.size() - 1];
  thread.start = [start, val, this]() {
    char dummy{};
    d_threads[d_tid].startOfStack = d_threads[d_tid].highestStackSeen = &dummy;
    auto const tid = d_tid;
    start(val);
    d_zombiesQueue.push(tid);
  };
  pdns_makecontext(*ucontext, thread.start);

  thread.context = std::move(ucontext);
  d_runQueue.push(d_maxtid++); // will run at next schedule invocation
}

//! needs to be called periodically so threads can run and housekeeping can be performed
/** The kernel should call this function every once in a while. It makes sense
    to call this function if you:
    - reported an event
    - called makeThread
    - want to have threads running waitEvent() to get a timeout if enough time passed

    \return Returns if there is more work scheduled and recalling schedule now would be useful

*/
template <class Key, class Val, class Cmp>
bool MTasker<Key, Val, Cmp>::schedule(const struct timeval& now)
{
  if (!d_runQueue.empty()) {
    d_tid = d_runQueue.front();
#ifdef MTASKERTIMING
    d_threads[d_tid].dt.start();
#endif
    notifyStackSwitch(d_threads[d_tid].startOfStack, d_stacksize);
    try {
      pdns_swapcontext(d_kernel, *d_threads[d_tid].context);
    }
    catch (...) {
      notifyStackSwitchDone();
      // It is not clear if the d_runQueue.pop() should be done in this case
      throw;
    }
    notifyStackSwitchDone();

    d_runQueue.pop();
    return true;
  }
  if (!d_zombiesQueue.empty()) {
    auto zombi = d_zombiesQueue.front();
    if (d_cachedStacks.size() < d_maxCachedStacks) {
      auto thread = d_threads.find(zombi);
      if (thread != d_threads.end()) {
        d_cachedStacks.push(std::move(thread->second.context->uc_stack));
      }
      d_threads.erase(thread);
    }
    else {
      d_threads.erase(zombi);
    }
    --d_threadsCount;
    d_zombiesQueue.pop();
    return true;
  }
  if (!d_waiters.empty()) {
    typedef typename waiters_t::template index<KeyTag>::type waiters_by_ttd_index_t;
    //    waiters_by_ttd_index_t& ttdindex=d_waiters.template get<KeyTag>();
    waiters_by_ttd_index_t& ttdindex = boost::multi_index::get<KeyTag>(d_waiters);

    for (typename waiters_by_ttd_index_t::iterator i = ttdindex.begin(); i != ttdindex.end();) {
      if (i->ttd.tv_sec && i->ttd < now) {
        d_waitstatus = TimeOut;
        d_eventkey = i->key; // pass waitEvent the exact key it was woken for
        auto ucontext = i->context;
        d_tid = i->tid;
        ttdindex.erase(i++); // removes the waitpoint

        notifyStackSwitch(d_threads[d_tid].startOfStack, d_stacksize);
        try {
          pdns_swapcontext(d_kernel, *ucontext); // swaps back to the above point 'A'
        }
        catch (...) {
          notifyStackSwitchDone();
          throw;
        }
        notifyStackSwitchDone();
      }
      else if (i->ttd.tv_sec != 0) {
        break;
      }
      else {
        ++i;
      }
    }
  }
  return false;
}

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	#pragma once
23	#include <cstdint>
24	#include <ctime>
25	#include <queue>
26	#include <memory>
27	#include <stack>
28
29	#include <boost/multi_index_container.hpp>
30	#include <boost/multi_index/ordered_index.hpp>
31	#include <boost/multi_index/key_extractors.hpp>
32
33	#include "namespaces.hh"
34	#include "misc.hh"
35	#include "mtasker_context.hh"
36
37	// #define MTASKERTIMING 1
38
39	//! The main MTasker class
40	/** The main MTasker class. See the main page for more information.
41	\tparam EventKey Type of the key with which events are to be identified. Defaults to int.
42	\tparam EventVal Type of the content or value of an event. Defaults to int. Cannot be set to void.
43	\note The EventKey needs to have an operator< defined because it is used as the key of an associative array
44	*/
45
46	template <class EventKey = int, class EventVal = int, class Cmp = std::less<EventKey>>
47	class MTasker
48	{
49	public:
50	struct Waiter
51	{
52	EventKey key;
53	std::shared_ptr<pdns_ucontext_t> context;
54	struct timeval ttd
55	{
56	};
57	int tid{};
58	};
59	struct KeyTag
60	{
61	};
62
63	using waiters_t = boost::multi_index::multi_index_container<
64	Waiter,
65	boost::multi_index::indexed_by<
66	boost::multi_index::ordered_unique<boost::multi_index::member<Waiter, EventKey, &Waiter::key>, Cmp>,
67	boost::multi_index::ordered_non_unique<boost::multi_index::tag<KeyTag>, boost::multi_index::member<Waiter, struct timeval, &Waiter::ttd>>>>;
68
69	//! Constructor
70	/** Constructor with a small default stacksize. If any of your threads exceeds this stack, your application will crash.
71	This limit applies solely to the stack, the heap is not limited in any way. If threads need to allocate a lot of data,
72	the use of new/delete is suggested.
73	*/
74	MTasker(size_t stacksize = static_cast<size_t>(16 * 8192), size_t stackCacheSize = 0) :
75	d_stacksize(stacksize), d_maxCachedStacks(stackCacheSize), d_waitstatus(Error)
76	{	802✔
77	initMainStackBounds();	802✔
78
79	// make sure our stack is 16-byte aligned to make all the architectures happy
80	d_stacksize = d_stacksize >> 4 << 4;	802✔
81	}	802✔
82
83	using tfunc_t = void(void*); //!< type of the pointer that starts a thread
84	uint64_t nextWaiterDelayUsec(uint64_t defusecs);
85	int waitEvent(EventKey& key, EventVal* val = nullptr, unsigned int timeoutMsec = 0, const struct timeval* now = nullptr);
86	void yield();
87	int sendEvent(const EventKey& key, const EventVal* val = nullptr);
88	void makeThread(tfunc_t* start, void* val);
89	bool schedule(const struct timeval& now);
90
91	const waiters_t& getWaiters() const
92	{	25,341✔
93	return d_waiters;	25,341✔
94	}	25,341✔
95
96	//! gives access to the list of Events threads are waiting for
97	/** The kernel can call this to get a list of Events threads are waiting for. This is very useful
98	to setup 'select' or 'poll' or 'aio' events needed to satisfy these requests.
99	getEvents clears the events parameter before filling it.
100
101	\param events Vector which is to be filled with keys threads are waiting for
102	*/
103	void getEvents(std::vector<EventKey>& events) const
104	{	2✔
105	events.clear();	2✔
106	for (const auto& waiter : d_waiters) {	2!
107	events.emplace_back(waiter.key);
108	}
109	}	2✔
110
111	//! returns true if there are no processes
112	/** Call this to check if no processes are running anymore
113	\return true if no processes are left
114	*/
115	[[nodiscard]] bool noProcesses() const
116	{	8✔
117	return d_threadsCount == 0;	8✔
118	}	8✔
119
120	//! returns the number of processes running
121	/** Call this to perhaps limit activities if too many threads are running
122	\return number of processes running
123	*/
124	[[nodiscard]] unsigned int numProcesses() const
125	{	14,149✔
126	return d_threadsCount;	14,149✔
127	}	14,149✔
128
129	//! Returns the current Thread ID (tid)
130	/** Processes can call this to get a numerical representation of their current thread ID.
131	This can be useful for logging purposes.
132	*/
133	[[nodiscard]] int getTid() const
134	{	4,657✔
135	return d_tid;	4,657✔
136	}	4,657✔
137
138	//! Returns the maximum stack usage so far of this MThread
139	[[nodiscard]] uint64_t getMaxStackUsage() const
140	{	4,638✔
141	return d_threads.at(d_tid).startOfStack - d_threads.at(d_tid).highestStackSeen;	4,638✔
142	}	4,638✔
143
144	//! Returns the maximum stack usage so far of this MThread
145	[[nodiscard]] unsigned int getUsec() const
146	{
147	#ifdef MTASKERTIMING
148	return d_threads.at(d_tid).totTime + d_threads.at(d_tid).dt.ndiff() / 1000;
149	#else
150	return 0;
151	#endif
152	}
153
154	private:
155	EventKey d_eventkey; // for waitEvent, contains exact key it was awoken for
156	EventVal d_waitval;
157
158	pdns_ucontext_t d_kernel;
159	std::queue<int> d_runQueue;
160	std::queue<int> d_zombiesQueue;
161
162	struct ThreadInfo
163	{
164	std::shared_ptr<pdns_ucontext_t> context;
165	std::function<void(void)> start;
166	const char* startOfStack{};
167	const char* highestStackSeen{};
168	#ifdef MTASKERTIMING
169	CPUTime dt;
170	unsigned int totTime;
171	#endif
172	};
173
174	using pdns_mtasker_stack_t = std::vector<char, lazy_allocator<char>>;
175	using mthreads_t = std::map<int, ThreadInfo>;
176
177	mthreads_t d_threads;
178	std::stack<pdns_mtasker_stack_t> d_cachedStacks;
179	waiters_t d_waiters;
180	size_t d_stacksize;
181	size_t d_threadsCount{0};
182	size_t d_maxCachedStacks{0};
183	int d_tid{0};
184	int d_maxtid{0};
185
186	enum waitstatusenum : int8_t
187	{
188	Error = -1,
189	TimeOut = 0,
190	Answer
191	} d_waitstatus;
192
193	std::shared_ptr<pdns_ucontext_t> getUContext();
194
195	void initMainStackBounds()
196	{	801✔
197	#ifdef HAVE_FIBER_SANITIZER	801✔
198
199	#ifdef HAVE_PTHREAD_GETATTR_NP	801✔
200	pthread_attr_t attr;	801✔
201	pthread_attr_init(&attr);	801✔
202	pthread_getattr_np(pthread_self(), &attr);	801✔
203	pthread_attr_getstack(&attr, &t_mainStack, &t_mainStackSize);	801✔
204	pthread_attr_destroy(&attr);	801✔
205	#elif defined(HAVE_PTHREAD_GET_STACKSIZE_NP) && defined(HAVE_PTHREAD_GET_STACKADDR_NP)
206	t_mainStack = pthread_get_stackaddr_np(pthread_self());
207	t_mainStackSize = pthread_get_stacksize_np(pthread_self());
208	#else
209	#error Cannot determine stack size and base on this platform
210	#endif
211
212	#endif /* HAVE_FIBER_SANITIZER */	801✔
213	}	801✔
214	};
215
216	#ifdef PDNS_USE_VALGRIND
217	#include <valgrind/valgrind.h>
218	#endif /* PDNS_USE_VALGRIND */
219
220	template <class EventKey, class EventVal, class Cmp>
221	uint64_t MTasker<EventKey, EventVal, Cmp>::nextWaiterDelayUsec(uint64_t defusecs)
222	{	28,260✔
223	if (d_waiters.empty()) {	28,260✔
224	// no waiters
225	return defusecs;	15,371✔
226	}	15,371✔
227	auto& ttdindex = boost::multi_index::get<KeyTag>(d_waiters);	12,889✔
228	auto iter = ttdindex.begin();	12,889✔
229	timeval rnow{};	12,889✔
230	gettimeofday(&rnow, nullptr);	12,889✔
231	if (iter->ttd.tv_sec != 0) {	12,891✔
232	// we have a waiter with a timeout specified
233	if (rnow < iter->ttd) {	12,824✔
234	// we should not wait longer than the default timeout
235	return std::min(defusecs, uSec(iter->ttd - rnow));	12,804✔
236	}	12,804✔
237	// already expired
238	return 0;	2,147,483,655✔
239	}	12,822✔
240	return defusecs;	2,147,483,649✔
241	}	12,889✔
242
243	//! puts a thread to sleep waiting until a specified event arrives
244	/** Threads can call waitEvent to register that they are waiting on an event with a certain key.
245	If so desired, the event can carry data which is returned in val in case that is non-zero.
246
247	Furthermore, a timeout can be specified in seconds.
248
249	Only one thread can be waiting on a key, results of trying to have more threads
250	waiting on the same key are undefined.
251
252	\param key Event key to wait for. Needs to match up to a key reported to sendEvent
253	\param val If non-zero, the value of the event will be stored in *val
254	\param timeout If non-zero, number of seconds to wait for an event.
255
256	\return returns -1 in case of error, 0 in case of timeout, 1 in case of an answer
257	*/
258	template <class EventKey, class EventVal, class Cmp>
259	int MTasker<EventKey, EventVal, Cmp>::waitEvent(EventKey& key, EventVal* val, unsigned int timeoutMsec, const struct timeval* now)
260	{	9,425✔
261	if (d_waiters.count(key)) { // there was already an exact same waiter	9,425!
262	return -1;	×
263	}	×
264
265	Waiter waiter;	9,425✔
266	waiter.context = std::make_shared<pdns_ucontext_t>();	9,425✔
267	waiter.ttd.tv_sec = 0;	9,425✔
268	waiter.ttd.tv_usec = 0;	9,425✔
269	if (timeoutMsec != 0) {	9,425✔
270	struct timeval increment	9,417✔
271	{	9,417✔
272	};	9,417✔
273	increment.tv_sec = timeoutMsec / 1000;	9,417✔
274	increment.tv_usec = static_cast<decltype(increment.tv_usec)>(1000 * (timeoutMsec % 1000));	9,417✔
275	if (now != nullptr) {	9,419✔
276	waiter.ttd = increment + *now;	9,383✔
277	}	9,383✔
278	else {	2,147,483,677✔
279	struct timeval realnow	2,147,483,677✔
280	{	2,147,483,677✔
281	};	2,147,483,677✔
282	gettimeofday(&realnow, nullptr);	2,147,483,677✔
283	waiter.ttd = increment + realnow;	2,147,483,677✔
284	}	2,147,483,677✔
285	}	9,417✔
286
287	waiter.tid = d_tid;	9,425✔
288	waiter.key = key;	9,425✔
289
290	d_waiters.insert(waiter);	9,425✔
291	#ifdef MTASKERTIMING
292	unsigned int diff = d_threads[d_tid].dt.ndiff() / 1000;
293	d_threads[d_tid].totTime += diff;
294	#endif
295	notifyStackSwitchToKernel();	9,425✔
296	pdns_swapcontext(*d_waiters.find(key)->context, d_kernel); // 'A' will return here when 'key' has arrived, hands over control to kernel first	9,425✔
297	notifyStackSwitchDone();	9,425✔
298	#ifdef MTASKERTIMING
299	d_threads[d_tid].dt.start();
300	#endif
301	if (val && d_waitstatus == Answer) {	9,425✔
302	*val = d_waitval;	9,408✔
303	}	9,408✔
304	d_tid = waiter.tid;	9,425✔
305	if ((char*)&waiter < d_threads[d_tid].highestStackSeen) {	9,425✔
306	d_threads[d_tid].highestStackSeen = (char*)&waiter;	3,907✔
307	}	3,907✔
308	key = d_eventkey;	9,425✔
309	return d_waitstatus;	9,425✔
310	}	9,425✔
311
312	//! yields control to the kernel or other threads
313	/** Hands over control to the kernel, allowing other processes to run, or events to arrive */
314
315	template <class Key, class Val, class Cmp>
316	void MTasker<Key, Val, Cmp>::yield()
317	{
318	d_runQueue.push(d_tid);
319	notifyStackSwitchToKernel();
320	pdns_swapcontext(*d_threads[d_tid].context, d_kernel); // give control to the kernel
321	notifyStackSwitchDone();
322	}
323
324	//! reports that an event took place for which threads may be waiting
325	/** From the kernel loop, sendEvent can be called to report that something occurred for which there may be waiters.
326	\param key Key of the event for which threads may be waiting
327	\param val If non-zero, pointer to the content of the event
328	\return Returns -1 in case of error, 0 if there were no waiters, 1 if a thread was woken up.
329
330	WARNING: when passing val as zero, d_waitval is undefined, and hence waitEvent will return undefined!
331	*/
332	template <class EventKey, class EventVal, class Cmp>
333	int MTasker<EventKey, EventVal, Cmp>::sendEvent(const EventKey& key, const EventVal* val)
334	{	9,409✔
335	typename waiters_t::iterator waiter = d_waiters.find(key);	9,409✔
336
337	if (waiter == d_waiters.end()) {	9,409!
338	// cerr<<"Event sent nobody was waiting for! " <<key << endl;
339	return 0;	×
340	}	×
341	d_waitstatus = Answer;	9,409✔
342	if (val) {	9,409!
343	d_waitval = *val;	9,409✔
344	}	9,409✔
345	d_tid = waiter->tid; // set tid	9,409✔
346	d_eventkey = waiter->key; // pass waitEvent the exact key it was woken for	9,409✔
347	auto userspace = std::move(waiter->context);	9,409✔
348	d_waiters.erase(waiter); // removes the waitpoint	9,409✔
349	notifyStackSwitch(d_threads[d_tid].startOfStack, d_stacksize);	9,409✔
350	try {	9,409✔
351	pdns_swapcontext(d_kernel, *userspace); // swaps back to the above point 'A'	9,409✔
352	}	9,409✔
353	catch (...) {	9,409✔
354	notifyStackSwitchDone();	×
355	throw;	×
356	}	×
357	notifyStackSwitchDone();	9,409✔
358	return 1;	9,409✔
359	}	9,409✔
360
361	template <class Key, class Val, class Cmp>
362	std::shared_ptr<pdns_ucontext_t> MTasker<Key, Val, Cmp>::getUContext()
363	{	3,725✔
364	auto ucontext = std::make_shared<pdns_ucontext_t>();	3,725✔
365	if (d_cachedStacks.empty()) {	3,725✔
366	ucontext->uc_stack.resize(d_stacksize + 1);	1,767✔
367	}	1,767✔
368	else {	1,958✔
369	ucontext->uc_stack = std::move(d_cachedStacks.top());	1,958✔
370	d_cachedStacks.pop();	1,958✔
371	}	1,958✔
372
373	ucontext->uc_link = &d_kernel; // come back to kernel after dying	3,725✔
374
375	#ifdef PDNS_USE_VALGRIND
376	ucontext->valgrind_id = VALGRIND_STACK_REGISTER(&ucontext->uc_stack[0], &ucontext->uc_stack[ucontext->uc_stack.size() - 1]);
377	#endif /* PDNS_USE_VALGRIND */
378
379	return ucontext;	3,725✔
380	}	3,725✔
381
382	//! launches a new thread
383	/** The kernel can call this to make a new thread, which starts at the function start and gets passed the val void pointer.
384	\param start Pointer to the function which will form the start of the thread
385	\param val A void pointer that can be used to pass data to the thread
386	*/
387	template <class Key, class Val, class Cmp>
388	void MTasker<Key, Val, Cmp>::makeThread(tfunc_t* start, void* val)
389	{	3,726✔
390	auto ucontext = getUContext();	3,726✔
391
392	++d_threadsCount;	3,726✔
393	auto& thread = d_threads[d_maxtid];	3,726✔
394	// we will get a better approximation when the task is executed, but that prevents notifying a stack at nullptr
395	// on the first invocation
396	d_threads[d_maxtid].startOfStack = &ucontext->uc_stack[ucontext->uc_stack.size() - 1];	3,726✔
397	thread.start = [start, val, this]() {	3,726✔
398	char dummy{};	3,596✔
399	d_threads[d_tid].startOfStack = d_threads[d_tid].highestStackSeen = &dummy;	3,596✔
400	auto const tid = d_tid;	3,596✔
401	start(val);	3,596✔
402	d_zombiesQueue.push(tid);	3,596✔
403	};	3,596✔
404	pdns_makecontext(*ucontext, thread.start);	3,726✔
405
406	thread.context = std::move(ucontext);	3,726✔
407	d_runQueue.push(d_maxtid++); // will run at next schedule invocation	3,726✔
408	}	3,726✔
409
410	//! needs to be called periodically so threads can run and housekeeping can be performed
411	/** The kernel should call this function every once in a while. It makes sense
412	to call this function if you:
413	- reported an event
414	- called makeThread
415	- want to have threads running waitEvent() to get a timeout if enough time passed
416
417	\return Returns if there is more work scheduled and recalling schedule now would be useful
418
419	*/
420	template <class Key, class Val, class Cmp>
421	bool MTasker<Key, Val, Cmp>::schedule(const struct timeval& now)
422	{	35,371✔
423	if (!d_runQueue.empty()) {	35,371✔
424	d_tid = d_runQueue.front();	3,599✔
425	#ifdef MTASKERTIMING
426	d_threads[d_tid].dt.start();
427	#endif
428	notifyStackSwitch(d_threads[d_tid].startOfStack, d_stacksize);	3,599✔
429	try {	3,599✔
430	pdns_swapcontext(d_kernel, *d_threads[d_tid].context);	3,599✔
431	}	3,599✔
432	catch (...) {	3,599✔
433	notifyStackSwitchDone();	2✔
434	// It is not clear if the d_runQueue.pop() should be done in this case
435	throw;	2✔
436	}	2✔
437	notifyStackSwitchDone();	3,599✔
438
439	d_runQueue.pop();	3,599✔
440	return true;	3,599✔
441	}	3,599✔
442	if (!d_zombiesQueue.empty()) {	31,772✔
443	auto zombi = d_zombiesQueue.front();	3,597✔
444	if (d_cachedStacks.size() < d_maxCachedStacks) {	3,598✔
445	auto thread = d_threads.find(zombi);	3,490✔
446	if (thread != d_threads.end()) {	3,495✔
447	d_cachedStacks.push(std::move(thread->second.context->uc_stack));	3,495✔
448	}	3,495✔
449	d_threads.erase(thread);	3,490✔
450	}	3,490✔
451	else {	2,147,483,751✔
452	d_threads.erase(zombi);	2,147,483,751✔
453	}	2,147,483,751✔
454	--d_threadsCount;	3,597✔
455	d_zombiesQueue.pop();	3,597✔
456	return true;	3,597✔
457	}	3,597✔
458	if (!d_waiters.empty()) {	28,175✔
459	typedef typename waiters_t::template index<KeyTag>::type waiters_by_ttd_index_t;	12,826✔
460	// waiters_by_ttd_index_t& ttdindex=d_waiters.template get<KeyTag>();
461	waiters_by_ttd_index_t& ttdindex = boost::multi_index::get<KeyTag>(d_waiters);	12,826✔
462
463	for (typename waiters_by_ttd_index_t::iterator i = ttdindex.begin(); i != ttdindex.end();) {	12,850✔
464	if (i->ttd.tv_sec && i->ttd < now) {	12,834✔
465	d_waitstatus = TimeOut;	16✔
466	d_eventkey = i->key; // pass waitEvent the exact key it was woken for	16✔
467	auto ucontext = i->context;	16✔
468	d_tid = i->tid;	16✔
469	ttdindex.erase(i++); // removes the waitpoint	16✔
470
471	notifyStackSwitch(d_threads[d_tid].startOfStack, d_stacksize);	16✔
472	try {	16✔
473	pdns_swapcontext(d_kernel, *ucontext); // swaps back to the above point 'A'	16✔
474	}	16✔
475	catch (...) {	16✔
UNCOV 476	notifyStackSwitchDone();	×
477	throw;	×
478	}	×
479	notifyStackSwitchDone();	16✔
480	}	16✔
481	else if (i->ttd.tv_sec != 0) {	12,816✔
482	break;	12,809✔
483	}	12,809✔
484	else {	2,147,483,651✔
485	++i;	2,147,483,651✔
486	}	2,147,483,651✔
487	}	12,830✔
488	}	12,826✔
489	return false;	28,175✔
490	}	28,175✔

PowerDNS / pdns / 11663687396

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