};
uint64_t g_maxTCPQueuedConnections{1000};
+size_t g_maxTCPQueriesPerConn{0};
+size_t g_maxTCPConnectionDuration{0};
+size_t g_maxTCPConnectionsPerClient{0};
+static std::mutex tcpClientsCountMutex;
+static std::map<ComboAddress,size_t,ComboAddress::addressOnlyLessThan> tcpClientsCount;
+
void* tcpClientThread(int pipefd);
-// Should not be called simultaneously!
-void TCPClientCollection::addTCPClientThread()
+static void decrementTCPClientCount(const ComboAddress& client)
{
- if (d_numthreads >= d_tcpclientthreads.capacity()) {
- warnlog("Adding a new TCP client thread would exceed the vector capacity (%d/%d), skipping", d_numthreads.load(), d_tcpclientthreads.capacity());
- return;
+ if (g_maxTCPConnectionsPerClient) {
+ std::lock_guard<std::mutex> lock(tcpClientsCountMutex);
+ tcpClientsCount[client]--;
+ if (tcpClientsCount[client] == 0) {
+ tcpClientsCount.erase(client);
+ }
}
+}
+void TCPClientCollection::addTCPClientThread()
+{
vinfolog("Adding TCP Client thread");
int pipefds[2] = { -1, -1};
return;
}
- try {
- thread t1(tcpClientThread, pipefds[0]);
- t1.detach();
- }
- catch(const std::runtime_error& e) {
- /* the thread creation failed, don't leak */
- errlog("Error creating a TCP thread: %s", e.what());
- close(pipefds[0]);
- close(pipefds[1]);
- return;
+ {
+ std::lock_guard<std::mutex> lock(d_mutex);
+
+ if (d_numthreads >= d_tcpclientthreads.capacity()) {
+ warnlog("Adding a new TCP client thread would exceed the vector capacity (%d/%d), skipping", d_numthreads.load(), d_tcpclientthreads.capacity());
+ close(pipefds[0]);
+ close(pipefds[1]);
+ return;
+ }
+
+ try {
+ thread t1(tcpClientThread, pipefds[0]);
+ t1.detach();
+ }
+ catch(const std::runtime_error& e) {
+ /* the thread creation failed, don't leak */
+ errlog("Error creating a TCP thread: %s", e.what());
+ close(pipefds[0]);
+ close(pipefds[1]);
+ return;
+ }
+
+ d_tcpclientthreads.push_back(pipefds[1]);
}
- d_tcpclientthreads.push_back(pipefds[1]);
++d_numthreads;
}
return true;
}
+static bool maxConnectionDurationReached(unsigned int maxConnectionDuration, time_t start, unsigned int& remainingTime)
+{
+ if (maxConnectionDuration) {
+ time_t elapsed = time(NULL) - start;
+ if (elapsed >= maxConnectionDuration) {
+ return true;
+ }
+ remainingTime = maxConnectionDuration - elapsed;
+ }
+ return false;
+}
+
std::shared_ptr<TCPClientCollection> g_tcpclientthreads;
void* tcpClientThread(int pipefd)
throw std::runtime_error("Error reading from TCP acceptor pipe (" + std::to_string(pipefd) + ") in " + std::string(isNonBlocking(pipefd) ? "non-blocking" : "blocking") + " mode: " + e.what());
}
- --g_tcpclientthreads->d_queued;
+ g_tcpclientthreads->decrementQueuedCount();
ci=*citmp;
delete citmp;
memset(&dest, 0, sizeof(dest));
dest.sin4.sin_family = ci.remote.sin4.sin_family;
socklen_t len = dest.getSocklen();
+ size_t queriesCount = 0;
+ time_t connectionStartTime = time(NULL);
+
if (!setNonBlocking(ci.fd))
goto drop;
try {
for(;;) {
+ unsigned int remainingTime = 0;
ds = nullptr;
outstanding = false;
ci.cs->queries++;
g_stats.queries++;
+ queriesCount++;
+
+ if (g_maxTCPQueriesPerConn && queriesCount > g_maxTCPQueriesPerConn) {
+ vinfolog("Terminating TCP connection from %s because it reached the maximum number of queries per conn (%d / %d)", ci.remote.toStringWithPort(), queriesCount, g_maxTCPQueriesPerConn);
+ break;
+ }
+
+ if (maxConnectionDurationReached(g_maxTCPConnectionDuration, connectionStartTime, remainingTime)) {
+ vinfolog("Terminating TCP connection from %s because it reached the maximum TCP connection duration", ci.remote.toStringWithPort());
+ break;
+ }
+
if (qlen < sizeof(dnsheader)) {
g_stats.nonCompliantQueries++;
break;
size_t querySize = qlen <= 4096 ? qlen + 512 : qlen;
char queryBuffer[querySize];
const char* query = queryBuffer;
- readn2WithTimeout(ci.fd, queryBuffer, qlen, g_tcpRecvTimeout);
+ readn2WithTimeout(ci.fd, queryBuffer, qlen, g_tcpRecvTimeout, remainingTime);
#ifdef HAVE_DNSCRYPT
std::shared_ptr<DnsCryptQuery> dnsCryptQuery = 0;
outstanding = false;
--ds->outstanding;
}
+ decrementTCPClientCount(ci.remote);
}
return 0;
}
-
/* spawn as many of these as required, they call Accept on a socket on which they will accept queries, and
they will hand off to worker threads & spawn more of them if required
*/
void* tcpAcceptorThread(void* p)
{
ClientState* cs = (ClientState*) p;
-
+ bool tcpClientCountIncremented = false;
ComboAddress remote;
remote.sin4.sin_family = cs->local.sin4.sin_family;
for(;;) {
bool queuedCounterIncremented = false;
ConnectionInfo* ci = nullptr;
+ tcpClientCountIncremented = false;
try {
ci = new ConnectionInfo;
ci->cs = cs;
continue;
}
- if(g_maxTCPQueuedConnections > 0 && g_tcpclientthreads->d_queued >= g_maxTCPQueuedConnections) {
+ if(g_maxTCPQueuedConnections > 0 && g_tcpclientthreads->getQueuedCount() >= g_maxTCPQueuedConnections) {
close(ci->fd);
delete ci;
ci=nullptr;
continue;
}
+ if (g_maxTCPConnectionsPerClient) {
+ std::lock_guard<std::mutex> lock(tcpClientsCountMutex);
+
+ if (tcpClientsCount[remote] >= g_maxTCPConnectionsPerClient) {
+ close(ci->fd);
+ delete ci;
+ ci=nullptr;
+ vinfolog("Dropping TCP connection from %s because we have too many from this client already", remote.toStringWithPort());
+ continue;
+ }
+ tcpClientsCount[remote]++;
+ tcpClientCountIncremented = true;
+ }
+
vinfolog("Got TCP connection from %s", remote.toStringWithPort());
-
+
ci->remote = remote;
int pipe = g_tcpclientthreads->getThread();
if (pipe >= 0) {
writen2WithTimeout(pipe, &ci, sizeof(ci), 0);
}
else {
- --g_tcpclientthreads->d_queued;
+ g_tcpclientthreads->decrementQueuedCount();
queuedCounterIncremented = false;
close(ci->fd);
delete ci;
ci=nullptr;
+ if(tcpClientCountIncremented) {
+ decrementTCPClientCount(remote);
+ }
}
}
catch(std::exception& e) {
errlog("While reading a TCP question: %s", e.what());
if(ci && ci->fd >= 0)
close(ci->fd);
+ if(tcpClientCountIncremented) {
+ decrementTCPClientCount(remote);
+ }
delete ci;
ci = nullptr;
if (queuedCounterIncremented) {
- --g_tcpclientthreads->d_queued;
+ g_tcpclientthreads->decrementQueuedCount();
}
}
catch(...){}
projects / thirdparty / pdns.git / blobdiff