1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
|
#pragma once
#include "dolog.hh"
#include "dnsdist-tcp.hh"
class TCPClientThreadData
{
public:
TCPClientThreadData():
localRespRuleActions(g_respruleactions.getLocal()), localCacheInsertedRespRuleActions(g_cacheInsertedRespRuleActions.getLocal()), mplexer(std::unique_ptr<FDMultiplexer>(FDMultiplexer::getMultiplexerSilent()))
{
}
LocalHolders holders;
LocalStateHolder<vector<DNSDistResponseRuleAction>> localRespRuleActions;
LocalStateHolder<vector<DNSDistResponseRuleAction>> localCacheInsertedRespRuleActions;
std::unique_ptr<FDMultiplexer> mplexer{nullptr};
int crossProtocolResponsesPipe{-1};
};
class IncomingTCPConnectionState : public TCPQuerySender, public std::enable_shared_from_this<IncomingTCPConnectionState>
{
public:
IncomingTCPConnectionState(ConnectionInfo&& ci, TCPClientThreadData& threadData, const struct timeval& now): d_buffer(s_maxPacketCacheEntrySize), d_ci(std::move(ci)), d_handler(d_ci.fd, timeval{g_tcpRecvTimeout,0}, d_ci.cs->tlsFrontend ? d_ci.cs->tlsFrontend->getContext() : nullptr, now.tv_sec), d_connectionStartTime(now), d_ioState(make_unique<IOStateHandler>(*threadData.mplexer, d_ci.fd)), d_threadData(threadData), d_creatorThreadID(std::this_thread::get_id())
{
d_origDest.reset();
d_origDest.sin4.sin_family = d_ci.remote.sin4.sin_family;
socklen_t socklen = d_origDest.getSocklen();
if (getsockname(d_ci.fd, reinterpret_cast<sockaddr*>(&d_origDest), &socklen)) {
d_origDest = d_ci.cs->local;
}
/* belongs to the handler now */
d_ci.fd = -1;
d_proxiedDestination = d_origDest;
d_proxiedRemote = d_ci.remote;
/* we manage the release of the downstream connection ourselves */
d_releaseConnection = false;
}
IncomingTCPConnectionState(const IncomingTCPConnectionState& rhs) = delete;
IncomingTCPConnectionState& operator=(const IncomingTCPConnectionState& rhs) = delete;
~IncomingTCPConnectionState();
void resetForNewQuery();
boost::optional<struct timeval> getClientReadTTD(struct timeval now) const
{
if (g_maxTCPConnectionDuration == 0 && g_tcpRecvTimeout == 0) {
return boost::none;
}
if (g_maxTCPConnectionDuration > 0) {
auto elapsed = now.tv_sec - d_connectionStartTime.tv_sec;
if (elapsed < 0 || (static_cast<size_t>(elapsed) >= g_maxTCPConnectionDuration)) {
return now;
}
auto remaining = g_maxTCPConnectionDuration - elapsed;
if (g_tcpRecvTimeout == 0 || remaining <= static_cast<size_t>(g_tcpRecvTimeout)) {
now.tv_sec += remaining;
return now;
}
}
now.tv_sec += g_tcpRecvTimeout;
return now;
}
boost::optional<struct timeval> getClientWriteTTD(const struct timeval& now) const
{
if (g_maxTCPConnectionDuration == 0 && g_tcpSendTimeout == 0) {
return boost::none;
}
struct timeval res = now;
if (g_maxTCPConnectionDuration > 0) {
auto elapsed = res.tv_sec - d_connectionStartTime.tv_sec;
if (elapsed < 0 || static_cast<size_t>(elapsed) >= g_maxTCPConnectionDuration) {
return res;
}
auto remaining = g_maxTCPConnectionDuration - elapsed;
if (g_tcpSendTimeout == 0 || remaining <= static_cast<size_t>(g_tcpSendTimeout)) {
res.tv_sec += remaining;
return res;
}
}
res.tv_sec += g_tcpSendTimeout;
return res;
}
bool maxConnectionDurationReached(unsigned int maxConnectionDuration, const struct timeval& now)
{
if (maxConnectionDuration) {
time_t curtime = now.tv_sec;
unsigned int elapsed = 0;
if (curtime > d_connectionStartTime.tv_sec) { // To prevent issues when time goes backward
elapsed = curtime - d_connectionStartTime.tv_sec;
}
if (elapsed >= maxConnectionDuration) {
return true;
}
}
return false;
}
std::shared_ptr<TCPConnectionToBackend> getOwnedDownstreamConnection(const std::shared_ptr<DownstreamState>& ds, const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs);
std::shared_ptr<TCPConnectionToBackend> getDownstreamConnection(std::shared_ptr<DownstreamState>& ds, const std::unique_ptr<std::vector<ProxyProtocolValue>>& tlvs, const struct timeval& now);
void registerOwnedDownstreamConnection(std::shared_ptr<TCPConnectionToBackend>& conn);
static size_t clearAllDownstreamConnections();
static void handleIO(std::shared_ptr<IncomingTCPConnectionState>& conn, const struct timeval& now);
static void handleIOCallback(int fd, FDMultiplexer::funcparam_t& param);
static void handleAsyncReady(int fd, FDMultiplexer::funcparam_t& param);
static void updateIO(std::shared_ptr<IncomingTCPConnectionState>& state, IOState newState, const struct timeval& now);
static IOState sendResponse(std::shared_ptr<IncomingTCPConnectionState>& state, const struct timeval& now, TCPResponse&& response);
static void queueResponse(std::shared_ptr<IncomingTCPConnectionState>& state, const struct timeval& now, TCPResponse&& response);
static void handleTimeout(std::shared_ptr<IncomingTCPConnectionState>& state, bool write);
/* we take a copy of a shared pointer, not a reference, because the initial shared pointer might be released during the handling of the response */
void handleResponse(const struct timeval& now, TCPResponse&& response) override;
void handleXFRResponse(const struct timeval& now, TCPResponse&& response) override;
void notifyIOError(InternalQueryState&& query, const struct timeval& now) override;
void handleCrossProtocolResponse(const struct timeval& now, TCPResponse&& response);
void terminateClientConnection();
void queueQuery(TCPQuery&& query);
bool canAcceptNewQueries(const struct timeval& now);
bool active() const override
{
return d_ioState != nullptr;
}
std::string toString() const
{
ostringstream o;
o << "Incoming TCP connection from "<<d_ci.remote.toStringWithPort()<<" over FD "<<d_handler.getDescriptor()<<", state is "<<(int)d_state<<", io state is "<<(d_ioState ? d_ioState->getState() : "empty")<<", queries count is "<<d_queriesCount<<", current queries count is "<<d_currentQueriesCount<<", "<<d_queuedResponses.size()<<" queued responses, "<<d_ownedConnectionsToBackend.size()<<" owned connections to a backend";
return o.str();
}
enum class State : uint8_t { doingHandshake, readingProxyProtocolHeader, waitingForQuery, readingQuerySize, readingQuery, sendingResponse, idle /* in case of XFR, we stop processing queries */ };
TCPResponse d_currentResponse;
std::map<std::shared_ptr<DownstreamState>, std::deque<std::shared_ptr<TCPConnectionToBackend>>> d_ownedConnectionsToBackend;
std::deque<TCPResponse> d_queuedResponses;
PacketBuffer d_buffer;
ConnectionInfo d_ci;
ComboAddress d_origDest;
ComboAddress d_proxiedRemote;
ComboAddress d_proxiedDestination;
TCPIOHandler d_handler;
struct timeval d_connectionStartTime;
struct timeval d_handshakeDoneTime;
struct timeval d_firstQuerySizeReadTime;
struct timeval d_querySizeReadTime;
struct timeval d_queryReadTime;
std::unique_ptr<IOStateHandler> d_ioState{nullptr};
std::unique_ptr<std::vector<ProxyProtocolValue>> d_proxyProtocolValues{nullptr};
TCPClientThreadData& d_threadData;
size_t d_currentPos{0};
size_t d_proxyProtocolNeed{0};
size_t d_queriesCount{0};
size_t d_currentQueriesCount{0};
std::thread::id d_creatorThreadID;
uint16_t d_querySize{0};
State d_state{State::doingHandshake};
bool d_isXFR{false};
bool d_proxyProtocolPayloadHasTLV{false};
bool d_lastIOBlocked{false};
bool d_hadErrors{false};
};
|
