[thirdparty/pdns.git] / pdns / dnsdist.hh

#pragma once
#include "ext/luawrapper/include/LuaContext.hpp"
#include <time.h>
#include "misc.hh"
#include "iputils.hh"
#include "dnsname.hh"
#include <atomic>
#include <boost/circular_buffer.hpp>
#include <boost/program_options.hpp>
#include <mutex>
#include <thread>
#include "sholder.hh"


struct StopWatch
{
#ifndef CLOCK_MONOTONIC_RAW
#define CLOCK_MONOTONIC_RAW CLOCK_MONOTONIC
#endif
  struct timespec d_start{0,0};
  void start() {  
    if(clock_gettime(CLOCK_MONOTONIC_RAW, &d_start) < 0)
      unixDie("Getting timestamp");
    
  }
  
  double udiff() const {
    struct timespec now;
    if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
      unixDie("Getting timestamp");
    
    return 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
  }

  double udiffAndSet() {
    struct timespec now;
    if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
      unixDie("Getting timestamp");
    
    auto ret= 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
    d_start = now;
    return ret;
  }

};

class QPSLimiter
{
public:
  QPSLimiter()
  {
  }

  QPSLimiter(unsigned int rate, unsigned int burst) : d_rate(rate), d_burst(burst), d_tokens(burst)
  {
    d_passthrough=false;
    d_prev.start();
  }

  unsigned int getRate() const
  {
    return d_passthrough? 0 : d_rate;
  }

  int getPassed() const
  {
    return d_passed;
  }
  int getBlocked() const
  {
    return d_blocked;
  }

  bool check() const // this is not quite fair
  {
    if(d_passthrough)
      return true;
    auto delta = d_prev.udiffAndSet();
  
    d_tokens += 1.0*d_rate * (delta/1000000.0);

    if(d_tokens > d_burst)
      d_tokens = d_burst;

    bool ret=false;
    if(d_tokens >= 1.0) { // we need this because burst=1 is weird otherwise
      ret=true;
      --d_tokens;
      d_passed++;
    }
    else
      d_blocked++;

    return ret; 
  }
private:
  bool d_passthrough{true};
  unsigned int d_rate;
  unsigned int d_burst;
  mutable double d_tokens;
  mutable StopWatch d_prev;
  mutable unsigned int d_passed{0};
  mutable unsigned int d_blocked{0};
};


struct IDState
{
  IDState() : origFD(-1) { origDest.sin4.sin_family = 0;}
  IDState(const IDState& orig)
  {
    origFD = orig.origFD;
    origID = orig.origID;
    origRemote = orig.origRemote;
    origDest = orig.origDest;
    age.store(orig.age.load());
  }

  int origFD;  // set to <0 to indicate this state is empty   // 4

  ComboAddress origRemote;                                    // 28
  ComboAddress origDest;                                      // 28
  StopWatch sentTime;                                         // 16
  DNSName qname;                                              // 80
  std::atomic<uint16_t> age;                                  // 4
  uint16_t qtype;                                             // 2
  uint16_t origID;                                            // 2
};

struct Rings {
  Rings()
  {
    clientRing.set_capacity(10000);
    queryRing.set_capacity(10000);
    respRing.set_capacity(10000);
  }
  boost::circular_buffer<ComboAddress> clientRing;
  boost::circular_buffer<DNSName> queryRing;
  struct Response
  {
    DNSName name;
    uint16_t qtype;
    uint8_t rcode;
    unsigned int usec;
  };
  boost::circular_buffer<Response> respRing;
  std::mutex respMutex;
};

extern Rings g_rings; // XXX locking for this is still substandard, queryRing and clientRing need RW lock

struct DownstreamState
{
  DownstreamState(const ComboAddress& remote_);

  int fd;            
  std::thread tid;
  ComboAddress remote;
  QPSLimiter qps;
  vector<IDState> idStates;
  std::atomic<uint64_t> idOffset{0};
  std::atomic<uint64_t> sendErrors{0};
  std::atomic<uint64_t> outstanding{0};
  std::atomic<uint64_t> reuseds{0};
  std::atomic<uint64_t> queries{0};
  struct {
    std::atomic<uint64_t> sendErrors{0};
    std::atomic<uint64_t> reuseds{0};
    std::atomic<uint64_t> queries{0};
  } prev;
  double queryLoad{0.0};
  double dropRate{0.0};
  double latencyUsec{0.0};
  int order{1};
  int weight{1};
  StopWatch sw;
  set<string> pools;
  enum class Availability { Up, Down, Auto} availability{Availability::Auto};
  bool upStatus{false};
  bool isUp() const
  {
    if(availability == Availability::Down)
      return false;
    if(availability == Availability::Up)
      return true;
    return upStatus;
  }
  void setUp() { availability = Availability::Up; }
  void setDown() { availability = Availability::Down; }
  void setAuto() { availability = Availability::Auto; }
};
using servers_t =vector<std::shared_ptr<DownstreamState>>;
typedef std::function<shared_ptr<DownstreamState>(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh)> policy_t;


struct ServerPolicy
{
  string name;
  policy_t policy;
};

void* responderThread(std::shared_ptr<DownstreamState> state);
extern std::mutex g_luamutex;
extern LuaContext g_lua;
extern std::string g_outputBuffer; // locking for this is ok, as locked by g_luamutex

extern GlobalStateHolder<ServerPolicy> g_policy;
extern GlobalStateHolder<servers_t> g_dstates;
extern GlobalStateHolder<vector<pair<boost::variant<SuffixMatchNode,NetmaskGroup>, QPSLimiter> >> g_limiters;
extern GlobalStateHolder<vector<pair<boost::variant<SuffixMatchNode,NetmaskGroup>, std::string> >> g_poolrules;
extern GlobalStateHolder<SuffixMatchNode> g_suffixMatchNodeFilter;
extern GlobalStateHolder<NetmaskGroup> g_ACL;

extern ComboAddress g_serverControl; // not changed during runtime

extern std::vector<ComboAddress> g_locals; // not changed at runtime
extern std::string g_key; // in theory needs locking

struct dnsheader;

void controlThread(int fd, ComboAddress local);
vector<std::function<void(void)>> setupLua(bool client);


namespace po = boost::program_options;
extern po::variables_map g_vm;

std::shared_ptr<DownstreamState> firstAvailable(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
std::shared_ptr<DownstreamState> leastOutstanding(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
std::shared_ptr<DownstreamState> wrandom(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
std::shared_ptr<DownstreamState> roundrobin(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
Commit	Line	Data
df111b53	1	#pragma once
	2	#include "ext/luawrapper/include/LuaContext.hpp"
	3	#include <time.h>
	4	#include "misc.hh"
	5	#include "iputils.hh"
	6	#include "dnsname.hh"
	7	#include <atomic>
	8	#include <boost/circular_buffer.hpp>
	9	#include <boost/program_options.hpp>
df111b53	10	#include <mutex>
df111b53	11	#include <thread>
ecbe9133	12	#include "sholder.hh"
638184e9	13
638184e9	14
df111b53	15	struct StopWatch
	16	{
	17	#ifndef CLOCK_MONOTONIC_RAW
	18	#define CLOCK_MONOTONIC_RAW CLOCK_MONOTONIC
	19	#endif
	20	struct timespec d_start{0,0};
	21	void start() {
	22	if(clock_gettime(CLOCK_MONOTONIC_RAW, &d_start) < 0)
	23	unixDie("Getting timestamp");
	24
	25	}
	26
	27	double udiff() const {
	28	struct timespec now;
	29	if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
	30	unixDie("Getting timestamp");
	31
	32	return 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
	33	}
	34
	35	double udiffAndSet() {
	36	struct timespec now;
	37	if(clock_gettime(CLOCK_MONOTONIC_RAW, &now) < 0)
	38	unixDie("Getting timestamp");
	39
	40	auto ret= 1000000.0*(now.tv_sec - d_start.tv_sec) + (now.tv_nsec - d_start.tv_nsec)/1000.0;
	41	d_start = now;
	42	return ret;
	43	}
	44
	45	};
	46
	47	class QPSLimiter
	48	{
	49	public:
	50	QPSLimiter()
	51	{
	52	}
	53
	54	QPSLimiter(unsigned int rate, unsigned int burst) : d_rate(rate), d_burst(burst), d_tokens(burst)
	55	{
	56	d_passthrough=false;
	57	d_prev.start();
	58	}
	59
	60	unsigned int getRate() const
	61	{
	62	return d_passthrough? 0 : d_rate;
	63	}
	64
	65	int getPassed() const
	66	{
	67	return d_passed;
	68	}
	69	int getBlocked() const
	70	{
	71	return d_blocked;
	72	}
	73
ecbe9133	74	bool check() const // this is not quite fair
df111b53	75	{
	76	if(d_passthrough)
	77	return true;
	78	auto delta = d_prev.udiffAndSet();
	79
	80	d_tokens += 1.0d_rate (delta/1000000.0);
	81
	82	if(d_tokens > d_burst)
	83	d_tokens = d_burst;
	84
	85	bool ret=false;
	86	if(d_tokens >= 1.0) { // we need this because burst=1 is weird otherwise
	87	ret=true;
	88	--d_tokens;
	89	d_passed++;
	90	}
	91	else
	92	d_blocked++;
	93
	94	return ret;
	95	}
	96	private:
	97	bool d_passthrough{true};
	98	unsigned int d_rate;
	99	unsigned int d_burst;
ecbe9133	100	mutable double d_tokens;
	101	mutable StopWatch d_prev;
	102	mutable unsigned int d_passed{0};
	103	mutable unsigned int d_blocked{0};
df111b53	104	};
	105
	106
	107	struct IDState
	108	{
549d63c9	109	IDState() : origFD(-1) { origDest.sin4.sin_family = 0;}
df111b53	110	IDState(const IDState& orig)
	111	{
	112	origFD = orig.origFD;
	113	origID = orig.origID;
	114	origRemote = orig.origRemote;
549d63c9	115	origDest = orig.origDest;
df111b53	116	age.store(orig.age.load());
	117	}
	118
2bf26975	119	int origFD; // set to <0 to indicate this state is empty // 4
	120
	121	ComboAddress origRemote; // 28
549d63c9	122	ComboAddress origDest; // 28
2bf26975	123	StopWatch sentTime; // 16
	124	DNSName qname; // 80
	125	std::atomic<uint16_t> age; // 4
	126	uint16_t qtype; // 2
	127	uint16_t origID; // 2
df111b53	128	};
	129
	130	struct Rings {
	131	Rings()
	132	{
	133	clientRing.set_capacity(10000);
	134	queryRing.set_capacity(10000);
	135	respRing.set_capacity(10000);
	136	}
	137	boost::circular_buffer<ComboAddress> clientRing;
	138	boost::circular_buffer<DNSName> queryRing;
	139	struct Response
	140	{
	141	DNSName name;
	142	uint16_t qtype;
	143	uint8_t rcode;
	144	unsigned int usec;
	145	};
	146	boost::circular_buffer<Response> respRing;
	147	std::mutex respMutex;
	148	};
	149
ecbe9133	150	extern Rings g_rings; // XXX locking for this is still substandard, queryRing and clientRing need RW lock
df111b53	151
	152	struct DownstreamState
	153	{
	154	DownstreamState(const ComboAddress& remote_);
	155
	156	int fd;
	157	std::thread tid;
	158	ComboAddress remote;
	159	QPSLimiter qps;
	160	vector<IDState> idStates;
	161	std::atomic<uint64_t> idOffset{0};
	162	std::atomic<uint64_t> sendErrors{0};
	163	std::atomic<uint64_t> outstanding{0};
	164	std::atomic<uint64_t> reuseds{0};
	165	std::atomic<uint64_t> queries{0};
	166	struct {
	167	std::atomic<uint64_t> sendErrors{0};
	168	std::atomic<uint64_t> reuseds{0};
	169	std::atomic<uint64_t> queries{0};
	170	} prev;
	171	double queryLoad{0.0};
	172	double dropRate{0.0};
	173	double latencyUsec{0.0};
	174	int order{1};
	175	int weight{1};
	176	StopWatch sw;
	177	set<string> pools;
	178	enum class Availability { Up, Down, Auto} availability{Availability::Auto};
	179	bool upStatus{false};
	180	bool isUp() const
	181	{
	182	if(availability == Availability::Down)
	183	return false;
	184	if(availability == Availability::Up)
	185	return true;
	186	return upStatus;
	187	}
	188	void setUp() { availability = Availability::Up; }
	189	void setDown() { availability = Availability::Down; }
	190	void setAuto() { availability = Availability::Auto; }
	191	};
	192	using servers_t =vector<std::shared_ptr<DownstreamState>>;
	193	typedef std::function<shared_ptr<DownstreamState>(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh)> policy_t;
	194
	195
	196	struct ServerPolicy
	197	{
	198	string name;
	199	policy_t policy;
	200	};
	201
	202	void* responderThread(std::shared_ptr<DownstreamState> state);
	203	extern std::mutex g_luamutex;
	204	extern LuaContext g_lua;
ecbe9133	205	extern std::string g_outputBuffer; // locking for this is ok, as locked by g_luamutex
df111b53	206
ecbe9133	207	extern GlobalStateHolder<ServerPolicy> g_policy;
	208	extern GlobalStateHolder<servers_t> g_dstates;
	209	extern GlobalStateHolder<vector<pair<boost::variant<SuffixMatchNode,NetmaskGroup>, QPSLimiter> >> g_limiters;
	210	extern GlobalStateHolder<vector<pair<boost::variant<SuffixMatchNode,NetmaskGroup>, std::string> >> g_poolrules;
	211	extern GlobalStateHolder<SuffixMatchNode> g_suffixMatchNodeFilter;
638184e9	212	extern GlobalStateHolder<NetmaskGroup> g_ACL;
2e72cc0e	213
ecbe9133	214	extern ComboAddress g_serverControl; // not changed during runtime
	215
	216	extern std::vector<ComboAddress> g_locals; // not changed at runtime
	217	extern std::string g_key; // in theory needs locking
	218
	219	struct dnsheader;
	220
	221	void controlThread(int fd, ComboAddress local);
2e72cc0e	222	vector<std::function<void(void)>> setupLua(bool client);
ecbe9133	223
ecbe9133	224
df111b53	225	namespace po = boost::program_options;
df111b53	226	extern po::variables_map g_vm;
ecbe9133	227
	228	std::shared_ptr<DownstreamState> firstAvailable(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
	229	std::shared_ptr<DownstreamState> leastOutstanding(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
	230	std::shared_ptr<DownstreamState> wrandom(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);
	231	std::shared_ptr<DownstreamState> roundrobin(const servers_t& servers, const ComboAddress& remote, const DNSName& qname, uint16_t qtype, dnsheader* dh);