Clean Valgrind reports, improved speed examples

Take that, hours of debugging memory leaks! Did lots of structural tweaks to more smartly keep track of commands and make sure everything is evenutally freed from the heap.

Clean Valgrind reports, improved speed examples
Take that, hours of debugging memory leaks! Did lots of structural tweaks to more smartly keep track of commands and make sure everything is evenutally freed from the heap.
Hayk Martirosyan
1 parent 46752c1e
Showing 9 changed files with 332 additions and 140 deletions
CMakeLists.txt
examples/basic_threaded.cpp
examples/simple_loop.cpp → examples/speed_test_async.cpp
examples/speed_test_async_multi.cpp
examples/simple_sync_loop.cpp → examples/speed_test_sync.cpp
src/command.cpp
src/command.hpp
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)
 # ---------------------------------------------------------
@@ -35,14 +35,17 @@ set(LIB_ALL ${LIB_REDIS})
 #add_executable(progressive examples/progressive.cpp ${SRC_ALL})
 #target_link_libraries(progressive ${LIB_REDIS})
-#add_executable(basic_threaded examples/basic_threaded.cpp ${SRC_ALL})
-#target_link_libraries(basic_threaded ${LIB_REDIS})
+add_executable(basic_threaded examples/basic_threaded.cpp ${SRC_ALL})
+target_link_libraries(basic_threaded ${LIB_REDIS})
 #add_executable(lpush_benchmark examples/lpush_benchmark.cpp ${SRC_ALL})
 #target_link_libraries(lpush_benchmark ${LIB_REDIS})
-add_executable(simple_loop examples/simple_loop.cpp ${SRC_ALL})
-target_link_libraries(simple_loop ${LIB_REDIS})
+add_executable(speed_test_async examples/speed_test_async.cpp ${SRC_ALL})
+target_link_libraries(speed_test_async ${LIB_REDIS})
-add_executable(simple_sync_loop examples/simple_sync_loop.cpp ${SRC_ALL})
-target_link_libraries(simple_sync_loop ${LIB_REDIS})
+add_executable(speed_test_sync examples/speed_test_sync.cpp ${SRC_ALL})
+target_link_libraries(speed_test_sync ${LIB_REDIS})
+
+add_executable(speed_test_async_multi examples/speed_test_async_multi.cpp ${SRC_ALL})
+target_link_libraries(speed_test_async_multi ${LIB_REDIS})
 /**
-* Basic asynchronous calls using redisx.
+*
 */
 #include <iostream>
-#include <thread>
 #include <chrono>
-#include "../src/redisx.hpp"
+#include <thread>
+#include "../src/redox.hpp"
 using namespace std;
-redisx::Redis rdx = {"localhost", 6379};
+redox::Redox rdx = {"localhost", 6379};
 int main(int argc, char* argv[]) {
   rdx.run();
   thread setter([]() {
-    while(true) {
-      rdx.command("INCR counter");
+    for(int i = 0; i < 5000; i++) {
+      rdx.command<int>("INCR counter");
       this_thread::sleep_for(chrono::milliseconds(1));
     }
   });
   thread getter([]() {
-    while(true) {
-      rdx.command<const string &>(
+    for(int i = 0; i < 5; i++) {
+      rdx.command<string>(
           "GET counter",
           [](const string& cmd, const string& value) {
             cout << cmd << ": " << value << endl;
@@ -37,5 +37,7 @@ int main(int argc, char* argv[]) {
   setter.join();
   getter.join();
+  rdx.stop();
+
   return 0;
 };
 /**
-* Basic asynchronous calls using redox.
+* Redox test
+* ----------
+* Increment a key on Redis using an asynchronous command on a timer.
 */
 #include <iostream>
@@ -18,48 +20,39 @@ int main(int argc, char* argv[]) {
   Redox rdx = {"localhost", 6379};
   rdx.run();
-  if(rdx.command_blocking("DEL simple_loop:count")) cout << "Deleted simple_loop:count" << endl;
-  else cerr << "Failed to delete simple_loop:count" << endl;
-
-  Command<nullptr_t>* null_cmd = rdx.command_blocking<nullptr_t>("GET WFEOIEFJ");
-  if(null_cmd->status() == REDOX_OK) cout << "got nonexistent key." << endl;
-  else cerr << "error with null cmd: " << null_cmd->status() << endl;
-  null_cmd->free();
-
-  Command<char*>* set_cmd = rdx.command_blocking<char*>("SET simple_loop:count 0");
-  cout << "set key, reply: " << set_cmd->reply() << endl;
-  set_cmd->free();
-
-  Command<char*>* count_cmd = rdx.command_blocking<char*>("GET simple_loop:count");
-  if(count_cmd->status() == REDOX_OK) {
-    cout << "At the start, simple_loop:count = " << count_cmd->reply() << endl;
+  if(rdx.command_blocking("SET simple_loop:count 0")) {
+    cout << "Reset the counter to zero." << endl;
+  } else {
+    cerr << "Failed to reset counter." << endl;
+    return 1;
   }
-  count_cmd->free();
   string cmd_str = "INCR simple_loop:count";
-
-  double freq = 10000; // Hz
+  double freq = 400000; // Hz
   double dt = 1 / freq; // s
-  double t = 3; // s
+  double t = 5; // s
-  cout << "Running \"" << cmd_str << "\" at dt = " << dt
-      << "s for " << t << "s..." << endl;
+  cout << "Sending \"" << cmd_str << "\" asynchronously every "
+       << dt << "s for " << t << "s..." << endl;
+  double t0 = time_s();
   atomic_int count(0);
+
   Command<int>* c = rdx.command<int>(
       cmd_str,
-      [&count](const string &cmd, const int& value) { count++; },
-      NULL,
-      dt,
-      0
+      [&count, &rdx](const string &cmd, const int& value) { count++; },
+      [](const string& cmd, int status) { cerr << "Bad reply: " << status << endl; },
+      dt
   );
-  double t0 = time_s();
+  // Wait for t time, then stop the command.
   this_thread::sleep_for(chrono::microseconds((int)(t*1e6)));
   rdx.cancel(c);
-  cout << "At the end, simple_loop:count = "
-       << rdx.command_blocking<string>("GET simple_loop:count")->reply() << endl;
+  // Get the final value of the counter
+  auto get_cmd = rdx.command_blocking<string>("GET simple_loop:count");
+  long final_count = stol(get_cmd->reply());
+  get_cmd->free();
   rdx.stop();
@@ -69,6 +62,7 @@ int main(int argc, char* argv[]) {
   cout << "Sent " << count << " commands in " << t_elapsed << "s, "
        << "that's " << actual_freq << " commands/s." << endl;
-  cout << "rdx.num_commands_processed() = " << rdx.num_commands_processed() << endl;
+  cout << "Final value of counter: " << final_count << endl;
+
   return 0;
 }
+/**
+* Redox test
+* ----------
+* Increment a key on Redis using an asynchronous command on a timer.
+*/
+
+#include <iostream>
+#include <vector>
+#include "../src/redox.hpp"
+
+using namespace std;
+using namespace redox;
+
+double time_s() {
+  unsigned long ms = chrono::system_clock::now().time_since_epoch() / chrono::microseconds(1);
+  return (double)ms / 1e6;
+}
+
+int main(int argc, char* argv[]) {
+
+  Redox rdx = {"localhost", 6379};
+  rdx.run();
+
+  if(rdx.command_blocking("SET simple_loop:count 0")) {
+    cout << "Reset the counter to zero." << endl;
+  } else {
+    cerr << "Failed to reset counter." << endl;
+    return 1;
+  }
+
+  string cmd_str = "INCR simple_loop:count";
+  double freq = 10000; // Hz
+  double dt = 1 / freq; // s
+  double t = 5; // s
+  int parallel = 100;
+
+  cout << "Sending \"" << cmd_str << "\" asynchronously every "
+       << dt << "s for " << t << "s..." << endl;
+
+  double t0 = time_s();
+  atomic_int count(0);
+
+  vector<Command<int>*> commands;
+  for(int i = 0; i < parallel; i++) {
+    commands.push_back(rdx.command<int>(
+      cmd_str,
+      [&count, &rdx](const string &cmd, const int& value) { count++; },
+      [](const string& cmd, int status) { cerr << "Bad reply: " << status << endl; },
+      dt
+    ));
+  }
+
+  // Wait for t time, then stop the command.
+  this_thread::sleep_for(chrono::microseconds((int)(t*1e6)));
+  for(auto c : commands) rdx.cancel(c);
+
+  // Get the final value of the counter
+  auto get_cmd = rdx.command_blocking<string>("GET simple_loop:count");
+  long final_count = stol(get_cmd->reply());
+  get_cmd->free();
+
+  rdx.stop();
+
+  double t_elapsed = time_s() - t0;
+  double actual_freq = (double)count / t_elapsed;
+
+  cout << "Sent " << count << " commands in " << t_elapsed << "s, "
+       << "that's " << actual_freq << " commands/s." << endl;
+
+  cout << "Final value of counter: " << final_count << endl;
+
+  return 0;
+}
 /**
-* Basic synchronous calls using redox.
+* Redox test
+* ----------
+* Increment a key on Redis using synchronous commands in a loop.
 */
 #include <iostream>
@@ -18,23 +20,32 @@ int main(int argc, char* argv[]) {
   Redox rdx = {"localhost", 6379};
   rdx.run();
-  if(rdx.command_blocking("DEL simple_loop:count")) cout << "Deleted simple_loop:count" << endl;
-  else cerr << "Failed to delete simple_loop:count" << endl;
+  if(rdx.command_blocking("SET simple_loop:count 0")) {
+    cout << "Reset the counter to zero." << endl;
+  } else {
+    cerr << "Failed to reset counter." << endl;
+    return 1;
+  }
   string cmd_str = "INCR simple_loop:count";
+  double t = 5; // s
-  int count = 50000;
-  double t0 = time_s();
+  cout << "Sending \"" << cmd_str << "\" synchronously for " << t << "s..." << endl;
-  cout << "Running \"" << cmd_str << "\" " << count << " times." << endl;
+  double t0 = time_s();
+  double t_end = t0 + t;
+  int count = 0;
-  for(int i = 0; i < count; i++) {
+  while(time_s() < t_end) {
     Command<int>* c = rdx.command_blocking<int>(cmd_str);
     if(c->status() != REDOX_OK) cerr << "Bad reply, code: " << c->status() << endl;
+    c->free();
+    count++;
   }
-  cout << "At the end, simple_loop:count = "
-    << rdx.command_blocking<string>("GET simple_loop:count")->reply() << endl;
+  auto get_cmd = rdx.command_blocking<string>("GET simple_loop:count");
+  long final_count = stol(get_cmd->reply());
+  get_cmd->free();
   rdx.stop();
@@ -42,8 +53,9 @@ int main(int argc, char* argv[]) {
   double actual_freq = (double)count / t_elapsed;
   cout << "Sent " << count << " commands in " << t_elapsed << "s, "
-    << "that's " << actual_freq << " commands/s." << endl;
+       << "that's " << actual_freq << " commands/s." << endl;
+
+  cout << "Final value of counter: " << final_count << endl;
-  cout << "rdx.num_commands_processed() = " << rdx.num_commands_processed() << endl;
   return 0;
 }
@@ -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);
@@ -26,10 +26,15 @@ static const int REDOX_TIMEOUT = 5;
 class Redox;
 template<class ReplyT>
+void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents);
+
+template<class ReplyT>
 class Command {
 friend class Redox;
+friend void submit_command_callback<ReplyT>(struct ev_loop* loop, ev_timer* timer, int revents);
+
 public:
   Command(
     Redox* rdx,
@@ -48,15 +53,16 @@ public:
   const bool free_memory;
-  redisReply* reply_obj;
+  redisReply* reply_obj = nullptr;
-  std::atomic_int pending;
+  std::atomic_int pending = {0};
   void invoke(const ReplyT& reply);
   void invoke_error(int status);
   const ReplyT& reply();
   int status() { return reply_status; };
+  bool is_completed() { return completed; }
   /**
   * Called by the user to free the redisReply object, when the free_memory
@@ -64,7 +70,7 @@ public:
   */
   void free();
-  static void command_callback(redisAsyncContext *c, void *r, void *privdata);
+  void process_reply();
 private:
@@ -76,7 +82,7 @@ private:
   ReplyT reply_val;
   int reply_status;
-  std::atomic_bool completed;
+  std::atomic_bool completed = {false};
   ev_timer timer;
   std::mutex timer_guard;
@@ -86,7 +92,11 @@ private:
     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();
@@ -100,74 +110,74 @@ Command&lt;ReplyT&gt;::Command(
     const std::function<void(const std::string&, const ReplyT&)>& callback,
     const std::function<void(const std::string&, int status)>& error_callback,
     double repeat, double after, bool free_memory
-) : rdx(rdx), cmd(cmd), repeat(repeat), after(after), free_memory(free_memory), reply_obj(NULL),
-    pending(0), callback(callback), error_callback(error_callback), completed(false)
+) : rdx(rdx), cmd(cmd), repeat(repeat), after(after), free_memory(free_memory),
+    callback(callback), error_callback(error_callback)
 {
   timer_guard.lock();
 }
 template<class ReplyT>
-void Command<ReplyT>::command_callback(redisAsyncContext *ctx, void *r, void *privdata) {
+void Command<ReplyT>::process_reply() {
+
+  free_guard.lock();
+
+  invoke_callback();
-  auto *c = (Command<ReplyT> *) privdata;
-  c->reply_obj = (redisReply *) r;
-  c->invoke_callback();
+  pending--;
+
+  if(!free_memory) {
+    // Allow free() method to free memory
+    free_guard.unlock();
+    return;
+  }
-  // Free the reply object unless told not to
-  if(c->free_memory) c->free_reply_object();
+  free_reply_object();
-  // Increment the Redox object command counter
-  c->rdx->incr_cmd_count();
+  if((pending == 0) && (repeat == 0)) {
+    free_command(this);
+  } else {
+    free_guard.unlock();
+  }
 }
 template<class ReplyT>
 void Command<ReplyT>::invoke(const ReplyT& r) {
-
   if(callback) callback(cmd, r);
-
-  pending--;
-  if(!free_memory) return;
-  if(pending != 0) return;
-  if(completed || (repeat == 0)) {
-//    std::cout << cmd << ": suicide!" << std::endl;
-    delete this;
-  }
 }
 template<class ReplyT>
 void Command<ReplyT>::invoke_error(int status) {
-
   if(error_callback) error_callback(cmd, status);
-
-  pending--;
-  if(!free_memory) return;
-  if(pending != 0) return;
-  if(completed || (repeat == 0)) {
-//    std::cout << cmd << ": suicide!" << std::endl;
-    delete this;
-  }
 }
 template<class ReplyT>
 void Command<ReplyT>::free_reply_object() {
-  if(reply_obj == NULL) {
-    std::cerr << "[ERROR] " << cmd << ": Attempting to double free reply object!" << std::endl;
+  if(reply_obj == nullptr) {
+    std::cerr << "[ERROR] " << cmd << ": Attempting to double free reply object." << std::endl;
     return;
   }
   freeReplyObject(reply_obj);
-  reply_obj = NULL;
+  reply_obj = nullptr;
+}
+
+template<class ReplyT>
+void Command<ReplyT>::free_command(Command<ReplyT>* c) {
+  c->rdx->commands_deleted += 1;
+  c->rdx->remove_active_command(c);
+//  std::cout << "[INFO] Deleted Command " << c->rdx->commands_created << " at " << c << std::endl;
+  delete c;
 }
 template<class ReplyT>
 void Command<ReplyT>::free() {
+  free_guard.lock();
   free_reply_object();
+  free_guard.unlock();
-  // Commit suicide
-//  std::cout << cmd << ": suicide, by calling free()!" << std::endl;
-  delete this;
+  free_command(this);
 }
 template<class ReplyT>
@@ -3,48 +3,47 @@
 */
 #include <signal.h>
-#include <string.h>
 #include "redox.hpp"
 using namespace std;
 namespace redox {
-// Global mutex to manage waiting for connected state
-// TODO get rid of this as the only global variable?
-mutex connected_lock;
+void Redox::connected(const redisAsyncContext *ctx, int status) {
-void Redox::connected(const redisAsyncContext *c, int status) {
   if (status != REDIS_OK) {
-    cerr << "[ERROR] Connecting to Redis: " << c->errstr << endl;
+    cerr << "[ERROR] Connecting to Redis: " << ctx->errstr << endl;
     return;
   }
   // Disable hiredis automatically freeing reply objects
-  c->c.reader->fn->freeObject = [](void* reply) {};
+  ctx->c.reader->fn->freeObject = [](void* reply) {};
-  cout << "Connected to Redis." << endl;
-  connected_lock.unlock();
+  Redox* rdx = (Redox*) ctx->data;
+  rdx->connected_lock.unlock();
+
+  cout << "[INFO] Connected to Redis." << endl;
 }
-void Redox::disconnected(const redisAsyncContext *c, int status) {
+void Redox::disconnected(const redisAsyncContext *ctx, int status) {
   if (status != REDIS_OK) {
-    cerr << "[ERROR] Disconnecting from Redis: " << c->errstr << endl;
+    cerr << "[ERROR] Disconnecting from Redis: " << ctx->errstr << endl;
     return;
   }
   // Re-enable hiredis automatically freeing reply objects
-  c->c.reader->fn->freeObject = freeReplyObject;
+  ctx->c.reader->fn->freeObject = freeReplyObject;
-  cout << "Disconnected from Redis." << endl;
-  connected_lock.lock();
+  Redox* rdx = (Redox*) ctx->data;
+  rdx->connected_lock.unlock();
+
+  cout << "[INFO] Disconnected from Redis." << endl;
 }
 Redox::Redox(const string& host, const int port)
-    : host(host), port(port), cmd_count(0), to_exit(false) {
+    : host(host), port(port) {
   lock_guard<mutex> lg(queue_guard);
-  connected_lock.lock();
   signal(SIGPIPE, SIG_IGN);
@@ -54,48 +53,115 @@ Redox::Redox(const string&amp; host, const int port)
     return;
   }
-  redisLibevAttach(EV_DEFAULT_ ctx);
+  evloop = ev_loop_new(EVFLAG_AUTO);
+  ev_set_userdata(evloop, (void*)this);
+
+  redisLibevAttach(evloop, ctx);
   redisAsyncSetConnectCallback(ctx, Redox::connected);
   redisAsyncSetDisconnectCallback(ctx, Redox::disconnected);
+
+  ctx->data = (void*)this;
+  connected_lock.lock();
 }
 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();
+
+  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_blocking() {
   // Events to connect to Redox
-  ev_run(EV_DEFAULT_ EVRUN_NOWAIT);
-  lock_guard<mutex> lg(connected_lock);
+  ev_run(evloop, EVRUN_NOWAIT);
+
+  // Block until connected to Redis
+  connected_lock.lock();
+  connected_lock.unlock();
   // Continuously create events and handle them
   while (!to_exit) {
     process_queued_commands();
-    ev_run(EV_DEFAULT_ EVRUN_NOWAIT);
+    ev_run(evloop, EVRUN_NOWAIT);
   }
-  // Handle exit events
-  ev_run(EV_DEFAULT_ EVRUN_NOWAIT);
+  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 {
+    ev_run(evloop, EVRUN_NOWAIT);
+  } while(commands_created != commands_deleted);
+
+  exited = true;
   // Let go for block_until_stopped method
   exit_waiter.notify_one();
+
+  cout << "[INFO] Event thread exited." << endl;
 }
 void Redox::run() {
   event_loop_thread = thread([this] { run_blocking(); });
-  event_loop_thread.detach();
+
+  // Don't return until connected
+  lock_guard<mutex> lg(connected_lock);
 }
-void Redox::stop() {
+void Redox::stop_signal() {
   to_exit = true;
 }
 void Redox::block() {
   unique_lock<mutex> ul(exit_waiter_lock);
-  exit_waiter.wait(ul, [this]() { return to_exit.load(); });
+  exit_waiter.wait(ul, [this] { return exited.load(); });
+}
+
+void Redox::stop() {
+  stop_signal();
+  block();
+}
+
+template<class ReplyT>
+void command_callback(redisAsyncContext *ctx, void *r, void *privdata) {
+
+  Command<ReplyT>* c = (Command<ReplyT>*) privdata;
+  redisReply* reply_obj = (redisReply*) r;
+  Redox* rdx = (Redox*) ctx->data;
+
+  if(!rdx->is_active_command(c)) {
+    std::cout << "[INFO] Ignoring callback, command " << c << " was freed." << std::endl;
+    freeReplyObject(r);
+    return;
+  }
+
+  c->reply_obj = reply_obj;
+  c->process_reply();
+
+  // Increment the Redox object command counter
+  rdx->incr_cmd_count();
 }
 /**
@@ -105,7 +171,7 @@ void Redox::block() {
 template<class ReplyT>
 bool submit_to_server(Command<ReplyT>* c) {
   c->pending++;
-  if (redisAsyncCommand(c->rdx->ctx, c->command_callback, (void*)c, c->cmd.c_str()) != REDIS_OK) {
+  if (redisAsyncCommand(c->rdx->ctx, command_callback<ReplyT>, (void*)c, c->cmd.c_str()) != REDIS_OK) {
     cerr << "[ERROR] Could not send \"" << c->cmd << "\": " << c->rdx->ctx->errstr << endl;
     c->invoke_error(REDOX_SEND_ERROR);
     return false;
@@ -116,13 +182,22 @@ bool submit_to_server(Command&lt;ReplyT&gt;* c) {
 template<class ReplyT>
 void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents) {
-  // Check if canceled
-  if(timer->data == NULL) {
-    cerr << "[WARNING] Skipping event, has been canceled." << endl;
+  auto c = (Command<ReplyT>*)timer->data;
+
+  if(c->is_completed()) {
+
+    cerr << "[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);
+
     return;
   }
-  auto c = (Command<ReplyT>*)timer->data;
   submit_to_server<ReplyT>(c);
 }
@@ -143,7 +218,7 @@ bool Redox::process_queued_command(void* c_ptr) {
     c->timer.data = (void*)c;
     ev_timer_init(&c->timer, submit_command_callback<ReplyT>, c->after, c->repeat);
-    ev_timer_start(EV_DEFAULT_ &c->timer);
+    ev_timer_start(evloop, &c->timer);
     c->timer_guard.unlock();
   }
@@ -195,7 +270,7 @@ Redox::get_command_map() { return commands_null; }
 // ----------------------------
 void Redox::command(const string& cmd) {
-  command<redisReply*>(cmd, NULL);
+  command<redisReply*>(cmd);
 }
 bool Redox::command_blocking(const string& cmd) {
@@ -15,6 +15,7 @@
 #include <string>
 #include <queue>
 #include <unordered_map>
+#include <unordered_set>
 #include <hiredis/hiredis.h>
 #include <hiredis/async.h>
@@ -35,8 +36,10 @@ public:
   void run();
   void run_blocking();
-  void stop();
+
+  void stop_signal();
   void block();
+  void stop();
   template<class ReplyT>
   Command<ReplyT>* command(
@@ -67,16 +70,34 @@ public:
 //  void subscribe(std::string channel, std::function<void(std::string channel, std::string msg)> callback);
 //  void unsubscribe(std::string channel);
+  std::atomic_int commands_created = {0};
+  std::atomic_int commands_deleted = {0};
+
+  bool is_active_command(void* c_ptr) {
+    return active_commands.find(c_ptr) != active_commands.end();
+  }
+
+  void remove_active_command(void* c_ptr) {
+    active_commands.erase(c_ptr);
+  }
+
 private:
   // Redox server
   std::string host;
   int port;
+  // Block run() until redis is connected
+  std::mutex connected_lock;
+
+  // Dynamically allocated libev event loop
+  struct ev_loop* evloop;
+
   // Number of commands processed
-  std::atomic_long cmd_count;
+  std::atomic_long cmd_count = {0};
-  std::atomic_bool to_exit;
+  std::atomic_bool to_exit = {false}; // Signal to exit
+  std::atomic_bool exited = {false}; // Event thread exited
   std::mutex exit_waiter_lock;
   std::condition_variable exit_waiter;
@@ -98,6 +119,9 @@ private:
   template<class ReplyT>
   bool process_queued_command(void* cmd_ptr);
+
+  // Commands created but not yet deleted
+  std::unordered_set<void*> active_commands;
 };
 // ---------------------------
@@ -113,25 +137,24 @@ Command&lt;ReplyT&gt;* Redox::command(
 ) {
   std::lock_guard<std::mutex> lg(queue_guard);
   auto* c = new Command<ReplyT>(this, cmd, callback, error_callback, repeat, after, free_memory);
-  get_command_map<ReplyT>()[(void*)c] = c;
-  command_queue.push((void*)c);
+  void* c_ptr = (void*)c;
+  get_command_map<ReplyT>()[c_ptr] = c;
+  command_queue.push(c_ptr);
+  active_commands.insert(c_ptr);
+  commands_created += 1;
+//  std::cout << "[DEBUG] Created Command " << commands_created << " at " << c << std::endl;
   return c;
 }
 template<class ReplyT>
 bool Redox::cancel(Command<ReplyT>* c) {
-  if(c == NULL) {
+  if(c == nullptr) {
     std::cerr << "[ERROR] Canceling null command." << std::endl;
     return false;
   }
-  c->timer.data = NULL;
-
-  std::lock_guard<std::mutex> lg(c->timer_guard);
-  if((c->repeat != 0) || (c->after != 0))
-    ev_timer_stop(EV_DEFAULT_ &c->timer);
-
+  std::cout << "[INFO] Canceling command at " << c << std::endl;
   c->completed = true;
   return true;