Line data    Source code

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