Add more guards, rearrange logic to be more bulletproof

Fixed a couple of bugs found with a test case of 100 parallel asynchronous clients. As of now, there are no known memory leaks, segfaults, or deadlocks in Redox.

Add more guards, rearrange logic to be more bulletproof
Fixed a couple of bugs found with a test case of 100 parallel asynchronous clients. As of now, there are no known memory leaks, segfaults, or deadlocks in Redox.
Hayk Martirosyan
1 parent 5d3828dd
Showing 8 changed files with 115 additions and 109 deletions
CMakeLists.txt
README.md
examples/basic.cpp
examples/basic_threaded.cpp
src/command.cpp
src/command.hpp
src/redox.cpp
src/redox.hpp
@@ -29,8 +29,8 @@ set(LIB_ALL ${LIB_REDIS})
 # Examples
 # ---------------------------------------------------------
  
-#add_executable(basic examples/basic.cpp ${SRC_ALL})
-#target_link_libraries(basic ${LIB_REDIS})
+add_executable(basic examples/basic.cpp ${SRC_ALL})
+target_link_libraries(basic ${LIB_REDIS})
  
 #add_executable(progressive examples/progressive.cpp ${SRC_ALL})
 #target_link_libraries(progressive ${LIB_REDIS})
 redox
 ======
  
-High-level, asynchronous, and wicked fast C++11 bindings for Redis
+Modern, asynchronous, and wicked fast C++11 bindings for Redis
  
 Work in progress, details coming soon.
 /**
-* Basic asynchronous calls using redisx.
+* Basic use of Redox to set and get a Redis key.
 */
  
 #include <iostream>
-#include "../src/redisx.hpp"
+#include "../src/redox.hpp"
  
 using namespace std;
  
 int main(int argc, char* argv[]) {
  
-  redisx::Redis rdx = {"localhost", 6379};
+  redox::Redox rdx = {"localhost", 6379};
  
-  rdx.command<const string &>("SET alaska rules!", [](const string &cmd, const string &value) {
+  rdx.command<string>("SET alaska rules!", [](const string &cmd, const string &value) {
     cout << cmd << ": " << value << endl;
   });
  
-  rdx.command<const string &>("GET alaska", [](const string &cmd, const string &value) {
+  rdx.command<string>("GET alaska", [](const string &cmd, const string &value) {
     cout << cmd << ": " << value << endl;
   });
  
@@ -20,6 +20,7 @@ int main(int argc, char* argv[]) {
       rdx.command<int>("INCR counter");
       this_thread::sleep_for(chrono::milliseconds(1));
     }
+    cout << "Setter thread exiting." << endl;
   });
  
   thread getter([]() {
@@ -32,6 +33,7 @@ int main(int argc, char* argv[]) {
       );
       this_thread::sleep_for(chrono::milliseconds(1000));
     }
+    cout << "Getter thread exiting." << endl;
   });
  
   setter.join();
@@ -64,7 +64,7 @@ void Command&lt;char*&gt;::invoke_callback() {
  
 template<>
 void Command<int>::invoke_callback() {
-//  std::cout << "invoking int callback" << std::endl;
+
   if(is_error_reply()) invoke_error(REDOX_ERROR_REPLY);
   else if(is_nil_reply()) invoke_error(REDOX_NIL_REPLY);
  
@@ -74,6 +74,13 @@ public:
  
   void process_reply();
  
+  ev_timer* get_timer() {
+    std::lock_guard<std::mutex> lg(timer_guard);
+    return &timer;
+  }
+
+  static void free_command(Command<ReplyT>* c);
+
 private:
  
   const std::function<void(const std::string&, const ReplyT&)> callback;
@@ -89,16 +96,11 @@ private:
   ev_timer timer;
   std::mutex timer_guard;
  
-  ev_timer* get_timer() {
-    std::lock_guard<std::mutex> lg(timer_guard);
-    return &timer;
-  }
  
   // Make sure we don't free resources until details taken care of
   std::mutex free_guard;
  
   void free_reply_object();
-  static void free_command(Command<ReplyT>* c);
  
   void invoke_callback();
   bool is_error_reply();
@@ -122,11 +124,6 @@ Command&lt;ReplyT&gt;::Command(
 template<class ReplyT>
 void Command<ReplyT>::process_reply() {
  
-  if(cmd == "GET simple_loop:count") {
-    std::cout << "In process_reply, cmd = " << cmd << ", reply_obj = " << reply_obj << std::endl;
-    std::cout << "reply int: " << reply_obj->integer << std::endl;
-    std::cout << "reply str: " << reply_obj->str << std::endl;
-  }
   free_guard.lock();
  
   invoke_callback();
@@ -134,20 +131,44 @@ void Command&lt;ReplyT&gt;::process_reply() {
   pending--;
  
   if(!free_memory) {
-    std::cout << "Command memory not being freed, free_memory = " << free_memory << std::endl;
     // Allow free() method to free memory
+//    std::cout << "Command memory not being freed, free_memory = " << free_memory << std::endl;
     free_guard.unlock();
     return;
   }
  
-
   free_reply_object();
  
-  if((pending == 0) && (repeat == 0)) {
-    free_command(this);
-  } else {
-    free_guard.unlock();
+//  // Free memory when all pending callbacks are received
+//  if((repeat != 0) && (pending == 0) && ((long)(get_timer()->data) == 0)) {
+//    std::cout << "Freeing command, timer stopped and pending is 0." << std::endl;
+//    free_command(this);
+//  }
+//
+//  if((pending == 0) && (repeat == 0)) {
+//    free_command(this);
+//  } else {
+//    free_guard.unlock();
+//  }
+
+  // Handle memory if all pending replies have arrived
+  if(pending == 0) {
+
+    // Just free non-repeating commands
+    if (repeat == 0) {
+      free_command(this);
+      return;
+
+    // Free repeating commands if timer is stopped
+    } else {
+      if((long)(get_timer()->data) == 0) {
+        free_command(this);
+        return;
+      }
+    }
   }
+
+  free_guard.unlock();
 }
  
 template<class ReplyT>
@@ -174,9 +195,8 @@ void Command&lt;ReplyT&gt;::free_reply_object() {
  
 template<class ReplyT>
 void Command<ReplyT>::free_command(Command<ReplyT>* c) {
-  c->rdx->commands_deleted += 1;
   c->rdx->template remove_active_command<ReplyT>(c->id);
-//  std::cout << "[INFO] Deleted Command " << c->rdx->commands_created << " at " << c << std::endl;
+//  std::cout << "[INFO] Deleted Command " << c->id << " at " << c << std::endl;
   delete c;
 }
  
@@ -66,14 +66,6 @@ Redox::Redox(const string&amp; host, const int port)
  
 Redox::~Redox() {
  
-//  cout << "Queue sizes: " << endl;
-//  cout << commands_redis_reply.size() << endl;
-//  cout << commands_string_r.size() << endl;
-//  cout << commands_char_p.size() << endl;
-//  cout << commands_int.size() << endl;
-//  cout << commands_long_long_int.size() << endl;
-//  cout << commands_null.size() << endl;
-
   redisAsyncDisconnect(ctx);
  
   stop();
@@ -105,14 +97,14 @@ void Redox::run_blocking() {
   cout << "[INFO] Stop signal detected." << endl;
  
   // Run a few more times to clear out canceled events
-//  for(int i = 0; i < 100; i++) {
-//    ev_run(evloop, EVRUN_NOWAIT);
-//  }
-
-  // Run until all commands are processed
-  do {
+  for(int i = 0; i < 100; i++) {
     ev_run(evloop, EVRUN_NOWAIT);
-  } while(commands_created != commands_deleted);
+  }
+
+  if(commands_created != commands_deleted) {
+    cerr << "[ERROR] All commands were not freed! "
+         << commands_created << "/" << commands_deleted << endl;
+  }
  
   exited = true;
  
@@ -145,36 +137,30 @@ void Redox::stop() {
 }
  
 template<class ReplyT>
+Command<ReplyT>* Redox::find_command(long id) {
+
+  lock_guard<mutex> lg(command_map_guard);
+
+  auto& command_map = get_command_map<ReplyT>();
+  auto it = command_map.find(id);
+  if(it == command_map.end()) return nullptr;
+  return it->second;
+}
+
+template<class ReplyT>
 void command_callback(redisAsyncContext *ctx, void *r, void *privdata) {
  
   Redox* rdx = (Redox*) ctx->data;
   long id = (long)privdata;
   redisReply* reply_obj = (redisReply*) r;
  
-  auto& command_map = rdx->get_command_map<ReplyT>();
-  auto it = command_map.find(id);
-  if(it == command_map.end()) {
-    cout << "[ERROR] Couldn't find Command " << id << " in command_map." << endl;
-    freeReplyObject(r);
-    return;
-  };
-  Command<ReplyT>* c = it->second;
-
-  if(!rdx->is_active_command(c->id)) {
-    std::cout << "[INFO] Ignoring callback, command " << c << " was freed." << std::endl;
-    freeReplyObject(r);
+  Command<ReplyT>* c = rdx->find_command<ReplyT>(id);
+  if(c == nullptr) {
+//    cout << "[WARNING] Couldn't find Command " << id << " in command_map (command_callback)." << endl;
+    freeReplyObject(reply_obj);
     return;
   }
  
-  if(c->cmd == "GET simple_loop:count") {
-    std::cout << "In command_callback = " << c->cmd << " at " << c << ", reply_obj = " << reply_obj << std::endl;
-    std::cout << "reply type: " << reply_obj->type << std::endl;
-    std::cout << "reply int: " << reply_obj->integer << std::endl;
-    std::cout << "reply str: " << reply_obj->str << std::endl;
-//    std::string s(reply_obj->str);
-//    std::cout << "string object: " << s << std::endl;
-  }
-
   c->reply_obj = reply_obj;
   c->process_reply();
  
@@ -188,9 +174,6 @@ void command_callback(redisAsyncContext *ctx, void *r, void *privdata) {
 */
 template<class ReplyT>
 bool submit_to_server(Command<ReplyT>* c) {
-  if(c->cmd == "GET simple_loop:count") {
-    std::cout << "submit_to_server for cmd at " << c << ": " << c->cmd << std::endl;
-  }
   c->pending++;
   if (redisAsyncCommand(c->rdx->ctx, command_callback<ReplyT>, (void*)c->id, c->cmd.c_str()) != REDIS_OK) {
     cerr << "[ERROR] Could not send \"" << c->cmd << "\": " << c->rdx->ctx->errstr << endl;
@@ -206,24 +189,24 @@ void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents)
   Redox* rdx = (Redox*) ev_userdata(loop);
   long id = (long)timer->data;
  
-  auto& command_map = rdx->get_command_map<ReplyT>();
-  auto it = command_map.find(id);
-  if(it == command_map.end()) {
-    cout << "[ERROR] Couldn't find Command " << id << " in command_map." << endl;
+  Command<ReplyT>* c = rdx->find_command<ReplyT>(id);
+  if(c == nullptr) {
+    cout << "[ERROR] Couldn't find Command " << id
+         << " in command_map (submit_command_callback)." << endl;
     return;
-  };
-  Command<ReplyT>* c = it->second;
+  }
  
   if(c->is_completed()) {
  
-    cerr << "[INFO] Command " << c << " is completed, stopping event timer." << endl;
+//    cout << "[INFO] Command " << c << " is completed, stopping event timer." << endl;
  
     c->timer_guard.lock();
     if((c->repeat != 0) || (c->after != 0))
       ev_timer_stop(loop, &c->timer);
     c->timer_guard.unlock();
  
-    Command<ReplyT>::free_command(c);
+    // Mark for memory to be freed when all callbacks are received
+    c->timer.data = (void*)0;
  
     return;
   }
@@ -234,15 +217,8 @@ void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents)
 template<class ReplyT>
 bool Redox::process_queued_command(long id) {
  
-  auto& command_map = get_command_map<ReplyT>();
-
-  auto it = command_map.find(id);
-  if(it == command_map.end()) return false;
-  Command<ReplyT>* c = it->second;
-
-  if(c->cmd == "GET simple_loop:count") {
-    std::cout << "process_queued_command for cmd at " << c << ": " << c->cmd << std::endl;
-  }
+  Command<ReplyT>* c = find_command<ReplyT>(id);
+  if(c == nullptr) return false;
  
   if((c->repeat == 0) && (c->after == 0)) {
     submit_to_server<ReplyT>(c);
@@ -266,6 +242,8 @@ void Redox::process_queued_commands() {
   while(!command_queue.empty()) {
  
     long id = command_queue.front();
+    command_queue.pop();
+
     if(process_queued_command<redisReply*>(id)) {}
     else if(process_queued_command<string>(id)) {}
     else if(process_queued_command<char*>(id)) {}
@@ -273,30 +251,38 @@ void Redox::process_queued_commands() {
     else if(process_queued_command<long long int>(id)) {}
     else if(process_queued_command<nullptr_t>(id)) {}
     else throw runtime_error("[FATAL] Command pointer not found in any queue!");
-
-    command_queue.pop();
   }
 }
  
 // ----------------------------
  
 template<> unordered_map<long, Command<redisReply*>*>&
-Redox::get_command_map() { return commands_redis_reply; }
+Redox::get_command_map<redisReply*>() {
+//  cout << "redis reply command map at " << &commands_redis_reply << endl;
+  return commands_redis_reply; }
  
 template<> unordered_map<long, Command<string>*>&
-Redox::get_command_map() { return commands_string_r; }
+Redox::get_command_map<string>() {
+//  cout << "string command map at " << &commands_string_r << endl;
+  return commands_string_r; }
  
 template<> unordered_map<long, Command<char*>*>&
-Redox::get_command_map() { return commands_char_p; }
+Redox::get_command_map<char*>() {
+//  cout << "char* command map at " << &commands_char_p << endl;
+  return commands_char_p; }
  
 template<> unordered_map<long, Command<int>*>&
-Redox::get_command_map() { return commands_int; }
+Redox::get_command_map<int>() {
+//  cout << "int command map at " << &commands_int << " has size: " << commands_int.size() << endl;
+  return commands_int; }
  
 template<> unordered_map<long, Command<long long int>*>&
-Redox::get_command_map() { return commands_long_long_int; }
+Redox::get_command_map<long long int>() {
+//  cout << "long long int command map at " << &commands_long_long_int << endl;
+  return commands_long_long_int; }
  
 template<> unordered_map<long, Command<nullptr_t>*>&
-Redox::get_command_map() { return commands_null; }
+Redox::get_command_map<nullptr_t>() { return commands_null; }
  
 // ----------------------------
 // Helpers
@@ -73,17 +73,17 @@ public:
   std::atomic_long commands_created = {0};
   std::atomic_long commands_deleted = {0};
  
-  bool is_active_command(const long id) {
-    return active_commands.find(id) != active_commands.end();
-  }
-
   template<class ReplyT>
   void remove_active_command(const long id) {
-    active_commands.erase(id);
+    std::lock_guard<std::mutex> lg1(command_map_guard);
     get_command_map<ReplyT>().erase(id);
+    commands_deleted += 1;
   }
  
   template<class ReplyT>
+  Command<ReplyT>* find_command(long id);
+
+  template<class ReplyT>
   std::unordered_map<long, Command<ReplyT>*>& get_command_map();
  
 private:
@@ -114,6 +114,7 @@ private:
   std::unordered_map<long, Command<int>*> commands_int;
   std::unordered_map<long, Command<long long int>*> commands_long_long_int;
   std::unordered_map<long, Command<std::nullptr_t>*> commands_null;
+  std::mutex command_map_guard;
  
   std::queue<long> command_queue;
   std::mutex queue_guard;
@@ -121,9 +122,6 @@ private:
  
   template<class ReplyT>
   bool process_queued_command(long id);
-
-  // Commands created but not yet deleted (stored by id)
-  std::unordered_set<long> active_commands;
 };
  
 // ---------------------------
@@ -137,19 +135,19 @@ Command&lt;ReplyT&gt;* Redox::command(
   double after,
   bool free_memory
 ) {
-  std::lock_guard<std::mutex> lg(queue_guard);
  
   commands_created += 1;
   auto* c = new Command<ReplyT>(this, commands_created, cmd,
     callback, error_callback, repeat, after, free_memory);
  
+  std::lock_guard<std::mutex> lg(queue_guard);
+  std::lock_guard<std::mutex> lg2(command_map_guard);
+
   get_command_map<ReplyT>()[c->id] = c;
-  active_commands.insert(c->id);
   command_queue.push(c->id);
-  std::cout << "[DEBUG] Created Command " << c->id << " at " << c << std::endl;
-  if(cmd == "GET simple_loop:count") {
-    std::cout << "Command created at " << c << ": " << c->cmd << std::endl;
-  }
+
+//  std::cout << "[DEBUG] Created Command " << c->id << " at " << c << std::endl;
+
   return c;
 }
  
@@ -161,7 +159,7 @@ bool Redox::cancel(Command&lt;ReplyT&gt;* c) {
     return false;
   }
  
-  std::cout << "[INFO] Canceling command " << c->id << " at " << c << std::endl;
+//  std::cout << "[INFO] Canceling command " << c->id << " at " << c << std::endl;
   c->completed = true;
  
   return true;
@@ -181,7 +179,7 @@ Command&lt;ReplyT&gt;* Redox::command_blocking(const std::string&amp; cmd) {
   Command<ReplyT>* c = command<ReplyT>(cmd,
     [&val, &status, &m, &cv](const std::string& cmd_str, const ReplyT& reply) {
       std::unique_lock<std::mutex> ul(m);
-      std::cout << "success callback: " << cmd_str << std::endl;
+
       val = reply;
       status = REDOX_OK;
       ul.unlock();
@@ -195,10 +193,10 @@ Command&lt;ReplyT&gt;* Redox::command_blocking(const std::string&amp; cmd) {
     },
     0, 0, false // No repeats, don't free memory
   );
-  std::cout << "command blocking cv wait starting" << std::endl;
+
   // Wait until a callback is invoked
   cv.wait(lk, [&status] { return status != REDOX_UNINIT; });
-  std::cout << "command blocking cv wait over" << std::endl;
+
   c->reply_val = val;
   c->reply_status = status;