Connection state management implemented

Created an atomic_int connect_state that keeps track of not yet connected, connected, disconnected, and errors. Used to implement good error behavior when the server is down and you start() a Redox instance, or when the server goes down in the middle of running commands. Now, nothing should hang, but should return errors (or throw exceptions) when the server goes down. Also added hooks for a user connect/disconnect callback in the constructor, which is helpful for client programs. Added an example with three Redox clients in one thread.

Connection state management implemented
Created an atomic_int connect_state that keeps track of not yet connected, connected, disconnected, and errors. Used to implement good error behavior when the server is down and you start() a Redox instance, or when the server goes down in the middle of running commands. Now, nothing should hang, but should return errors (or throw exceptions) when the server goes down. Also added hooks for a user connect/disconnect callback in the constructor, which is helpful for client programs. Added an example with three Redox clients in one thread.
Hayk Martirosyan
1 parent 2e36078f
Showing 6 changed files with 145 additions and 49 deletions
CMakeLists.txt
README.md
examples/basic.cpp
examples/multi-client.cpp
src/redox.cpp
src/redox.hpp
 cmake_minimum_required(VERSION 2.8.4)
 project(redox)
  
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -fPIC -Wall -O3")
-#set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -fPIC -Wall -g  -fno-omit-frame-pointer")
+#set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -fPIC -Wall -O3")
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -fPIC -Wall -g  -fno-omit-frame-pointer")
 # set(CMAKE_VERBOSE_MAKEFILE ON)
  
 # ---------------------------------------------------------
@@ -52,3 +52,6 @@ target_link_libraries(speed_test_sync ${LIB_REDIS})
  
 #add_executable(string_v_char examples/string_vs_charp.cpp ${SRC_ALL})
 #target_link_libraries(string_v_char ${LIB_REDIS})
+
+add_executable(multi_client examples/multi-client.cpp ${SRC_ALL})
+target_link_libraries(multi_client ${LIB_REDIS})
@@ -28,6 +28,22 @@ During these tests, Redox communicated with a local Redis server over TCP.
  * One repeating asynchronous command (`speed_test_async`): **195,159 commands/s**
  * One blocking command in a loop (`speed_test_sync`): **23,609 commands/s**
  
+## Build from source
+Instructions provided are for Ubuntu, but Redox is fully platform-independent.
+
+Get the build environment:
+
+    sudo apt-get install git cmake build-essential
+
+Get the dependencies:
+
+    sudo apt-get install libhiredis-dev libev-dev
+
+Build Redox and examples using CMake (a helper script is provided):
+
+    cd redox
+    ./make.sh
+
 ## Tutorial
 Coming soon. For now, look at the example programs located in `examples/`, and the snippets
 posted below.
@@ -10,7 +10,7 @@ using namespace std;
 int main(int argc, char* argv[]) {
  
   redox::Redox rdx = {"localhost", 6379}; // Initialize Redox
-  rdx.start(); // Start the event loop
+  if(!rdx.start()) return 1; // Start the event loop
  
   rdx.del("occupation");
  
+/**
+* Redox example with multiple clients.
+*/
+
+#include <iostream>
+#include "../src/redox.hpp"
+
+using namespace std;
+
+int main(int argc, char* argv[]) {
+
+  redox::Redox rdx1, rdx2, rdx3;
+
+  if(!rdx1.start() || !rdx2.start() || !rdx3.start()) return 1;
+
+  rdx1.del("occupation");
+
+  if(!rdx2.set("occupation", "carpenter")) // Set a key, check if succeeded
+    cerr << "Failed to set key!" << endl;
+
+  cout << "key = occupation, value = \"" << rdx3.get("occupation") << "\"" << endl;
+
+  rdx1.stop();
+  rdx2.stop();
+  rdx3.stop();
+}
@@ -9,40 +9,51 @@ using namespace std;
  
 namespace redox {
  
-void Redox::connected(const redisAsyncContext *ctx, int status) {
+void Redox::connected_callback(const redisAsyncContext *ctx, int status) {
+
+  Redox* rdx = (Redox*) ctx->data;
  
   if (status != REDIS_OK) {
     cerr << "[ERROR] Connecting to Redis: " << ctx->errstr << endl;
+    rdx->connect_state = REDOX_CONNECT_ERROR;
+    rdx->connect_waiter.notify_all();
     return;
   }
  
   // Disable hiredis automatically freeing reply objects
   ctx->c.reader->fn->freeObject = [](void* reply) {};
  
-  Redox* rdx = (Redox*) ctx->data;
-  rdx->connected_lock.unlock();
+  rdx->connect_state = REDOX_CONNECTED;
+  rdx->connect_waiter.notify_all();
  
   cout << "[INFO] Connected to Redis." << endl;
+
+  if(rdx->user_connect_callback) rdx->user_connect_callback();
 }
  
-void Redox::disconnected(const redisAsyncContext *ctx, int status) {
+void Redox::disconnected_callback(const redisAsyncContext *ctx, int status) {
+
+  Redox* rdx = (Redox*) ctx->data;
  
   if (status != REDIS_OK) {
     cerr << "[ERROR] Disconnecting from Redis: " << ctx->errstr << endl;
-    return;
+    rdx->connect_state = REDOX_DISCONNECT_ERROR;
+  } else {
+    cout << "[INFO] Disconnected from Redis as planned." << endl;
+    rdx->connect_state = REDOX_DISCONNECTED;
   }
  
-  // Re-enable hiredis automatically freeing reply objects
-  ctx->c.reader->fn->freeObject = freeReplyObject;
+  rdx->stop_signal();
+  rdx->connect_waiter.notify_all();
  
-  Redox* rdx = (Redox*) ctx->data;
-  rdx->connected_lock.unlock();
-
-  cout << "[INFO] Disconnected from Redis." << endl;
+  if(rdx->user_disconnect_callback) rdx->user_disconnect_callback();
 }
  
-Redox::Redox(const string& host, const int port)
-    : host(host), port(port) {
+Redox::Redox(
+  const string& host, const int port,
+  std::function<void(void)> connected,
+  std::function<void(void)> disconnected
+) : host(host), port(port), user_connect_callback(connected), user_disconnect_callback(disconnected) {
  
   // Required by libev
   signal(SIGPIPE, SIG_IGN);
@@ -59,30 +70,12 @@ Redox::Redox(const string&amp; host, const int port)
   redisLibevAttach(evloop, ctx);
  
   // Set the callbacks to be invoked on server connection/disconnection
-  redisAsyncSetConnectCallback(ctx, Redox::connected);
-  redisAsyncSetDisconnectCallback(ctx, Redox::disconnected);
+  redisAsyncSetConnectCallback(ctx, Redox::connected_callback);
+  redisAsyncSetDisconnectCallback(ctx, Redox::disconnected_callback);
  
   // Set back references to this Redox object (for use in callbacks)
   ev_set_userdata(evloop, (void*)this);
   ctx->data = (void*)this;
-
-  // Lock this mutex until the connected callback is invoked
-  connected_lock.lock();
-}
-
-Redox::~Redox() {
-
-  redisAsyncDisconnect(ctx);
-
-  stop();
-
-  if(event_loop_thread.joinable())
-    event_loop_thread.join();
-
-  ev_loop_destroy(evloop);
-
-  std::cout << "[INFO] Redox created " << commands_created
-            << " Commands and freed " << commands_deleted << "." << std::endl;
 }
  
 void Redox::run_event_loop() {
@@ -90,9 +83,16 @@ void Redox::run_event_loop() {
   // Events to connect to Redox
   ev_run(evloop, EVRUN_NOWAIT);
  
-  // Block until connected to Redis
-  connected_lock.lock();
-  connected_lock.unlock();
+  // Block until connected to Redis, or error
+  unique_lock<mutex> ul(connect_lock);
+  connect_waiter.wait(ul, [this] { return connect_state != REDOX_NOT_YET_CONNECTED; });
+
+  // Handle connection error
+  if(connect_state != REDOX_CONNECTED) {
+    cout << "[INFO] Did not connect, event loop exiting." << endl;
+    running_waiter.notify_one();
+    return;
+  }
  
   // Set up asynchronous watcher which we signal every
   // time we add a command
@@ -128,13 +128,19 @@ void Redox::run_event_loop() {
   cout << "[INFO] Event thread exited." << endl;
 }
  
-void Redox::start() {
+bool Redox::start() {
  
   event_loop_thread = thread([this] { run_event_loop(); });
  
-  // Block until connected and running the event loop
+  // Block until connected and running the event loop, or until
+  // a connection error happens and the event loop exits
   unique_lock<mutex> ul(running_waiter_lock);
-  running_waiter.wait(ul, [this] { return running.load(); });
+  running_waiter.wait(ul, [this] {
+    return running.load() || connect_state == REDOX_CONNECT_ERROR;
+  });
+
+  // Return if succeeded
+  return connect_state == REDOX_CONNECTED;
 }
  
 void Redox::stop_signal() {
@@ -152,6 +158,27 @@ void Redox::stop() {
   block();
 }
  
+void Redox::disconnect() {
+  stop_signal();
+  if(connect_state == REDOX_CONNECTED) {
+    redisAsyncDisconnect(ctx);
+    block();
+  }
+}
+
+Redox::~Redox() {
+
+  disconnect();
+
+  if(event_loop_thread.joinable())
+    event_loop_thread.join();
+
+  ev_loop_destroy(evloop);
+
+  std::cout << "[INFO] Redox created " << commands_created
+    << " Commands and freed " << commands_deleted << "." << std::endl;
+}
+
 template<class ReplyT>
 Command<ReplyT>* Redox::find_command(long id) {
  
@@ -29,6 +29,13 @@ namespace redox {
 static const std::string REDIS_DEFAULT_HOST = "localhost";
 static const int REDIS_DEFAULT_PORT = 6379;
  
+// Connection status
+static const int REDOX_NOT_YET_CONNECTED = 0;
+static const int REDOX_CONNECTED = 1;
+static const int REDOX_DISCONNECTED = 2;
+static const int REDOX_CONNECT_ERROR = 3;
+static const int REDOX_DISCONNECT_ERROR = 4;
+
 class Redox {
  
 public:
@@ -36,14 +43,19 @@ public:
   /**
   * Initialize everything, connect over TCP to a Redis server.
   */
-  Redox(const std::string& host = REDIS_DEFAULT_HOST, const int port = REDIS_DEFAULT_PORT);
+  Redox(
+    const std::string& host = REDIS_DEFAULT_HOST,
+    const int port = REDIS_DEFAULT_PORT,
+    std::function<void(void)> connected = nullptr,
+    std::function<void(void)> disconnected = nullptr
+  );
   ~Redox();
  
   /**
   * Connect to Redis and start the event loop in a separate thread. Returns
-  * once everything is ready to go.
+  * true if and when everything is ready to go, or false on failure.
   */
-  void start();
+  bool start();
  
   /**
   * Signal the event loop to stop processing commands and shut down.
@@ -107,6 +119,12 @@ public:
   // Hiredis context, left public to allow low-level access
   redisAsyncContext *ctx;
  
+  /**
+  * If connected, disconnect from the Redis server. Usually not necessary to invoke
+  * manually, as it is called in the destructor.
+  */
+  void disconnect();
+
   // ------------------------------------------------
   // Wrapper methods for convenience only
   // ------------------------------------------------
@@ -160,8 +178,14 @@ private:
   std::string host;
   int port;
  
-  // Block run() until redis is connected
-  std::mutex connected_lock;
+  // Manage connection state
+  std::atomic_int connect_state = {REDOX_NOT_YET_CONNECTED};
+  std::mutex connect_lock;
+  std::condition_variable connect_waiter;
+
+  // User connect/disconnect callbacks
+  std::function<void(void)> user_connect_callback;
+  std::function<void(void)> user_disconnect_callback;
  
   // Dynamically allocated libev event loop
   struct ev_loop* evloop;
@@ -218,8 +242,8 @@ private:
   void run_event_loop();
  
   // Callbacks invoked on server connection/disconnection
-  static void connected(const redisAsyncContext *c, int status);
-  static void disconnected(const redisAsyncContext *c, int status);
+  static void connected_callback(const redisAsyncContext *c, int status);
+  static void disconnected_callback(const redisAsyncContext *c, int status);
  
   template<class ReplyT>
   static void command_callback(redisAsyncContext *ctx, void *r, void *privdata);