Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 :
6 : #include "redis_connection.h"
7 :
8 : #include <cstdlib>
9 : #include <mutex>
10 :
11 : #include <boost/bind/bind.hpp>
12 : #include "base/util.h"
13 : #include "base/address_util.h"
14 : #include "base/logging.h"
15 : #include "base/parse_object.h"
16 : #include "hiredis/hiredis.h"
17 : #include "hiredis/hiredis_ssl.h"
18 : #include "hiredis/base64.h"
19 : #include "hiredis/boostasio.hpp"
20 :
21 : using std::string;
22 : using std::vector;
23 : using namespace boost::placeholders;
24 :
25 : const int RedisAsyncConnection::RedisReconnectTime;
26 : RedisAsyncConnection::RAC_CbFnsMap RedisAsyncConnection::rac_cb_fns_map_;
27 : std::mutex RedisAsyncConnection::rac_cb_fns_map_mutex_;
28 :
29 220 : RedisAsyncConnection::RedisAsyncConnection(EventManager *evm, const std::string & redis_ip,
30 : unsigned short redis_port, ClientConnectCbFn client_connect_cb,
31 : ClientDisconnectCbFn client_disconnect_cb,
32 : const bool redis_ssl_enable, const string & redis_keyfile,
33 220 : const string & redis_certfile, const string & redis_ca_cert) :
34 220 : evm_(evm),
35 220 : hostname_(redis_ip),
36 220 : port_(redis_port),
37 220 : redis_ssl_enable_(redis_ssl_enable),
38 220 : redis_keyfile_(redis_keyfile),
39 220 : redis_certfile_(redis_certfile),
40 220 : redis_ca_cert_(redis_ca_cert),
41 220 : callDisconnected_(0),
42 220 : callFailed_(0),
43 220 : callSucceeded_(0),
44 220 : callbackNull_(0),
45 220 : callbackFailed_(0),
46 220 : callbackSucceeded_(0),
47 220 : context_(NULL),
48 220 : state_(REDIS_ASYNC_CONNECTION_INIT),
49 220 : reconnect_timer_(*evm->io_service()),
50 220 : client_connect_cb_(client_connect_cb),
51 660 : client_disconnect_cb_(client_disconnect_cb) {
52 220 : boost::system::error_code ec;
53 : boost::asio::ip::address redis_addr(
54 220 : AddressFromString(hostname_, &ec));
55 220 : endpoint_ = boost::asio::ip::tcp::endpoint(redis_addr, redis_port);
56 220 : }
57 :
58 0 : RedisAsyncConnection::~RedisAsyncConnection() {
59 0 : boost::system::error_code ec;
60 :
61 0 : if (client_connect_cb_)
62 0 : client_connect_cb_ = NULL;
63 :
64 0 : if (client_disconnect_cb_)
65 0 : client_disconnect_cb_ = NULL;
66 :
67 0 : if (context_) {
68 : {
69 0 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
70 :
71 : RedisAsyncConnection::RAC_CbFnsMap& fns_map =
72 0 : RedisAsyncConnection::rac_cb_fns_map();
73 0 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(context_);
74 0 : assert(it != fns_map.end());
75 0 : fns_map.erase(it);
76 0 : }
77 0 : redisAsyncFree(context_);
78 : }
79 0 : reconnect_timer_.cancel(ec);
80 0 : }
81 :
82 40 : void RedisAsyncConnection::RAC_Reconnect(const boost::system::error_code &error) {
83 40 : if (error) {
84 0 : LOG(INFO, "RAC_Reconnect error: " << error.message());
85 0 : if (error.value() != boost::system::errc::operation_canceled) {
86 0 : LOG(INFO, "OpServerProxy::OpServerImpl::RAC_Reconnect error: "
87 : << error.category().name()
88 : << " " << error.message());
89 : } else {
90 0 : return;
91 : }
92 : } else {
93 40 : LOG(INFO, "RedisAsyncConnection::RAC_Reconnect initiated " << this);
94 : }
95 :
96 40 : if (!RAC_Connect()) {
97 0 : assert(0);
98 : }
99 : }
100 :
101 275 : void RedisAsyncConnection::RAC_ConnectCallbackProcess(const struct redisAsyncContext *c, int status) {
102 275 : LOG(DEBUG, "RAC_Connect status: " << status << " " << this);
103 275 : if (status != REDIS_OK) {
104 45 : if (context_) {
105 45 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
106 :
107 : RedisAsyncConnection::RAC_CbFnsMap& fns_map =
108 45 : RedisAsyncConnection::rac_cb_fns_map();
109 45 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(context_);
110 45 : if (it != fns_map.end()) {
111 45 : fns_map.erase(it);
112 : }
113 45 : context_ = NULL;
114 45 : client_.reset();
115 45 : }
116 :
117 45 : boost::system::error_code ec;
118 45 : reconnect_timer_.expires_from_now(boost::posix_time::seconds(RedisAsyncConnection::RedisReconnectTime), ec);
119 45 : reconnect_timer_.async_wait(boost::bind(&RedisAsyncConnection::RAC_Reconnect, this, boost::asio::placeholders::error));
120 45 : return;
121 : }
122 230 : state_ = REDIS_ASYNC_CONNECTION_CONNECTED;
123 230 : LOG(DEBUG, "Connected to REDIS...\n");
124 :
125 230 : if (client_connect_cb_)
126 230 : client_connect_cb_();
127 : }
128 :
129 0 : void RedisAsyncConnection::RAC_StatUpdate(const redisReply *reply) {
130 0 : if (reply == NULL) {
131 0 : callbackNull_++;
132 0 : return;
133 : }
134 0 : if (reply->type == REDIS_REPLY_ERROR) {
135 0 : callbackFailed_++;
136 : } else {
137 0 : callbackSucceeded_++;
138 : }
139 : }
140 :
141 275 : void RedisAsyncConnection::RAC_ConnectCallback(const struct redisAsyncContext *c, int status) {
142 275 : RedisAsyncConnection::RAC_ConnectCbFn fn;
143 :
144 : {
145 275 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
146 275 : RedisAsyncConnection::RAC_CbFnsMap& fns_map = RedisAsyncConnection::rac_cb_fns_map();
147 275 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(c);
148 275 : if (it == fns_map.end())
149 0 : assert(0);
150 :
151 275 : RedisAsyncConnection::RAC_CbFns *fns = it->second;
152 275 : fn = fns->connect_cbfn_;
153 275 : fns->connect_cbfn_ = NULL;
154 275 : }
155 :
156 275 : if (!fn) {
157 0 : assert(0);
158 : }
159 275 : (fn)(c, status);
160 275 : }
161 :
162 15 : void RedisAsyncConnection::RAC_DisconnectCallbackProcess(const struct redisAsyncContext *c, int status) {
163 15 : LOG(DEBUG, "RAC_Disconnect status: " << status << " " << this);
164 : {
165 15 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
166 15 : RedisAsyncConnection::RAC_CbFnsMap& fns_map = RedisAsyncConnection::rac_cb_fns_map();
167 15 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(context_);
168 15 : if (it != fns_map.end())
169 15 : fns_map.erase(it);
170 15 : state_ = REDIS_ASYNC_CONNECTION_DISCONNECTED;
171 15 : context_ = NULL;
172 15 : client_.reset();
173 15 : }
174 :
175 15 : if (client_disconnect_cb_)
176 15 : client_disconnect_cb_();
177 15 : }
178 :
179 15 : void RedisAsyncConnection::RAC_DisconnectCallback(const struct redisAsyncContext *c, int status) {
180 15 : RedisAsyncConnection::RAC_DisconnectCbFn fn;
181 :
182 : {
183 15 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
184 15 : RedisAsyncConnection::RAC_CbFnsMap& fns_map = RedisAsyncConnection::rac_cb_fns_map();
185 15 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(c);
186 15 : if (it == fns_map.end()) {
187 0 : return;
188 : }
189 :
190 15 : RedisAsyncConnection::RAC_CbFns *fns = it->second;
191 15 : fn = fns->disconnect_cbfn_;
192 15 : fns->disconnect_cbfn_ = NULL;
193 15 : }
194 :
195 15 : if (!fn) {
196 0 : assert(0);
197 : }
198 :
199 15 : (fn)(c, status);
200 15 : }
201 :
202 275 : bool RedisAsyncConnection::RAC_Connect(void) {
203 275 : std::scoped_lock lock(mutex_);
204 :
205 275 : assert(!context_);
206 275 : context_ = redisAsyncConnect(hostname_.c_str(), port_);
207 275 : if (context_->err) {
208 0 : LOG(DEBUG, "RAC_Connect: redisAsyncConnect() failed:" << context_->errstr);
209 0 : boost::system::error_code ec;
210 0 : reconnect_timer_.expires_from_now(boost::posix_time::seconds(RedisAsyncConnection::RedisReconnectTime), ec);
211 0 : reconnect_timer_.async_wait(boost::bind(&RedisAsyncConnection::RAC_Reconnect, this, boost::asio::placeholders::error));
212 0 : context_ = NULL;
213 0 : return true;
214 : }
215 :
216 : /* Secure the connection if SSL enabled */
217 275 : if (redis_ssl_enable_) {
218 0 : redisSSLContextError ssl_error = REDIS_SSL_CTX_NONE;
219 0 : redisSSLContext *ssl_ctx = redisCreateSSLContext(
220 : redis_ca_cert_.c_str(),
221 : NULL,
222 : redis_certfile_.c_str(),
223 : redis_keyfile_.c_str(),
224 : "sni",
225 : &ssl_error);
226 0 : if (!ssl_ctx || ssl_error != REDIS_SSL_CTX_NONE) {
227 0 : LOG(DEBUG, "RAC_Connect: redisCreateSSLContext() failed: "
228 : << redisSSLContextGetError(ssl_error));
229 0 : if (ssl_ctx) redisFreeSSLContext(ssl_ctx);
230 0 : boost::system::error_code ec;
231 0 : reconnect_timer_.expires_from_now(
232 0 : boost::posix_time::seconds(RedisAsyncConnection::RedisReconnectTime), ec);
233 0 : reconnect_timer_.async_wait(boost::bind(&RedisAsyncConnection::RAC_Reconnect, this,
234 : boost::asio::placeholders::error));
235 0 : context_ = NULL;
236 0 : return true;
237 : }
238 :
239 0 : int rc = redisInitiateSSLWithContext(&context_->c, ssl_ctx);
240 0 : if (rc != REDIS_OK) {
241 0 : LOG(DEBUG, "RAC_Connect: redisInitiateSSLWithContext() failed: "
242 : << (&context_->c)->errstr);
243 0 : redisFreeSSLContext(ssl_ctx);
244 0 : boost::system::error_code ec;
245 0 : reconnect_timer_.expires_from_now(
246 0 : boost::posix_time::seconds(RedisAsyncConnection::RedisReconnectTime), ec);
247 0 : reconnect_timer_.async_wait(boost::bind(&RedisAsyncConnection::RAC_Reconnect, this,
248 : boost::asio::placeholders::error));
249 0 : context_ = NULL;
250 0 : return true;
251 : }
252 0 : redisFreeSSLContext(ssl_ctx);
253 : }
254 :
255 275 : client_.reset(new redisBoostClient(*evm_->io_service(), context_, mutex_));
256 :
257 275 : std::scoped_lock fns_lock(rac_cb_fns_map_mutex_);
258 :
259 275 : assert(redisAsyncSetConnectCallback(context_, RedisAsyncConnection::RAC_ConnectCallback) == REDIS_OK);
260 275 : RedisAsyncConnection::RAC_CbFnsMap& fns_map = RedisAsyncConnection::rac_cb_fns_map();
261 275 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(context_);
262 275 : if (it == fns_map.end()) {
263 275 : const redisAsyncContext *c_context = context_;
264 275 : it = (fns_map.insert(c_context, new RAC_CbFns)).first;
265 : } else {
266 0 : assert(0);
267 : }
268 275 : it->second->stat_cbfn_ = boost::bind(&RedisAsyncConnection::RAC_StatUpdate, this, _1);
269 275 : it->second->connect_cbfn_ = boost::bind(&RedisAsyncConnection::RAC_ConnectCallbackProcess, this, _1, _2);
270 :
271 275 : assert(redisAsyncSetDisconnectCallback(context_, RedisAsyncConnection::RAC_DisconnectCallback) == REDIS_OK);
272 275 : it->second->disconnect_cbfn_ = boost::bind(&RedisAsyncConnection::RAC_DisconnectCallbackProcess, this, _1, _2);
273 :
274 275 : state_ = REDIS_ASYNC_CONNECTION_PENDING;
275 275 : return true;
276 275 : }
277 :
278 3567 : void RedisAsyncConnection::RAC_AsyncCmdCallback(redisAsyncContext *c, void *r, void *privdata) {
279 3567 : ClientAsyncCmdCbFn cbfn;
280 3567 : RAC_StatCbFn stat_fn;
281 : {
282 3567 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
283 3567 : RedisAsyncConnection::RAC_CbFnsMap& fns_map = RedisAsyncConnection::rac_cb_fns_map();
284 3567 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(c);
285 3567 : if (it == fns_map.end()) {
286 0 : return;
287 : }
288 3567 : cbfn = it->second->client_async_cmd_cbfn_;
289 3567 : stat_fn = it->second->stat_cbfn_;
290 3567 : }
291 3567 : if (cbfn) {
292 3567 : (cbfn)(c, r, privdata);
293 3567 : return;
294 : }
295 :
296 0 : redisReply *reply = (redisReply*)r;
297 0 : assert(stat_fn);
298 0 : (stat_fn)(reply);
299 :
300 : #if 0
301 :
302 : if (reply->type == REDIS_REPLY_ARRAY) {
303 : LOG(DEBUG, __func__ << "REDIS_REPLY_ARRAY == " << reply->elements);
304 : int i;
305 : for (i = 0; i < (int)reply->elements; i++) {
306 : if (reply->element[i]->type == REDIS_REPLY_STRING) {
307 : LOG(DEBUG, __func__ << "Element" << i << "== " << reply->element[i]->str);
308 : } else {
309 : LOG(DEBUG, __func__ << "Element" << i << " type == " << reply->element[i]->type);
310 : }
311 : }
312 : } else if (reply->type == REDIS_REPLY_STRING) {
313 : LOG(DEBUG, __func__ << "REDIS_REPLY_STRING == " << reply->str);
314 : } else if (reply->type == REDIS_REPLY_INTEGER) {
315 : LOG(DEBUG, __func__ << "REDIS_REPLY_INTEGER == " << reply->type);
316 : } else {
317 : LOG(DEBUG, __func__ << "reply->type == " << reply->type);
318 : }
319 : #endif
320 7134 : }
321 :
322 460 : bool RedisAsyncConnection::SetClientAsyncCmdCb(ClientAsyncCmdCbFn cb_fn) {
323 460 : std::scoped_lock lock(rac_cb_fns_map_mutex_);
324 460 : RedisAsyncConnection::RAC_CbFnsMap& fns_map = rac_cb_fns_map();
325 460 : RedisAsyncConnection::RAC_CbFnsMap::iterator it = fns_map.find(context_);
326 :
327 460 : if (it == fns_map.end()) {
328 0 : assert(0);
329 : } else {
330 460 : it->second->client_async_cmd_cbfn_ = cb_fn;
331 : }
332 460 : return true;
333 460 : }
334 :
335 :
336 :
337 3584 : bool RedisAsyncConnection::RedisAsyncArgCmd(void *rpi,
338 : const vector<string> &args) {
339 :
340 3584 : std::scoped_lock lock(mutex_);
341 :
342 3609 : if (state_ != REDIS_ASYNC_CONNECTION_CONNECTED) {
343 0 : callDisconnected_++;
344 0 : return false;
345 : }
346 :
347 3609 : int argc = args.size();
348 3609 : const char** argv = new const char* [argc];
349 17010 : for (uint i=0; i < args.size(); i++) {
350 13401 : argv[i] = args[i].c_str();
351 : }
352 3609 : bool status = false;
353 : int ret;
354 :
355 3609 : ret = redisAsyncCommandArgv(context_,
356 : RedisAsyncConnection::RAC_AsyncCmdCallback,
357 : rpi,
358 : argc,
359 : argv,
360 : NULL);
361 :
362 3610 : delete[] argv;
363 :
364 3610 : if (REDIS_ERR == ret) {
365 0 : LOG(INFO, "Could NOT apply " << args[0] << " to Redis : ");
366 0 : callFailed_++;
367 : } else {
368 3610 : status = true;
369 3610 : callSucceeded_++;
370 : }
371 3610 : return status;
372 3610 : }
373 :
374 :
375 0 : bool RedisAsyncConnection::RedisAsyncCommand(void *rpi, const char *format, ...) {
376 0 : std::scoped_lock lock(mutex_);
377 :
378 0 : if (state_ != REDIS_ASYNC_CONNECTION_CONNECTED) {
379 0 : callDisconnected_++;
380 0 : return false;
381 : }
382 :
383 0 : bool status = false;
384 : int ret;
385 : va_list ap;
386 0 : va_start(ap,format);
387 :
388 0 : ret = redisvAsyncCommand(context_,
389 : RedisAsyncConnection::RAC_AsyncCmdCallback,
390 : rpi,
391 : format,
392 : ap);
393 :
394 0 : if (REDIS_ERR == ret) {
395 0 : LOG(INFO, "Could NOT apply " << format << " to Redis : ");
396 0 : callFailed_++;
397 : } else {
398 0 : status = true;
399 0 : callSucceeded_++;
400 : }
401 0 : va_end(ap);
402 0 : return status;
403 0 : }
|
