Removing timer_callbacks global, ev_timer allocation

Found out that ev_timer watchers have a void* data that allows currying information to the libev callbacks! That means I could stick the Command* in there and completely get rid of the global timer_callbacks map that was needed before. Additionally, we can move ev_timer into the Command object instead of separately allocating it on the heap. Simplifies management. Implemented a synchronous loop speed test based on the latest and greatest command_blocking(). Happy to see it is around 30k/s, which is 100% network delays.

Removing timer_callbacks global, ev_timer allocation
Found out that ev_timer watchers have a void* data that allows currying information to the libev callbacks! That means I could stick the Command* in there and completely get rid of the global timer_callbacks map that was needed before. Additionally, we can move ev_timer into the Command object instead of separately allocating it on the heap. Simplifies management. Implemented a synchronous loop speed test based on the latest and greatest command_blocking(). Happy to see it is around 30k/s, which is 100% network delays.
Hayk Martirosyan
1 parent 78585079
Showing 6 changed files with 82 additions and 34 deletions
CMakeLists.txt
examples/simple_loop.cpp
examples/simple_sync_loop.cpp
src/command.hpp
src/redisx.cpp
src/redisx.hpp
@@ -42,3 +42,6 @@ set(LIB_ALL ${LIB_REDIS})
 add_executable(simple_loop examples/simple_loop.cpp ${SRC_ALL})
 target_link_libraries(simple_loop ${LIB_REDIS})
+
+add_executable(simple_sync_loop examples/simple_sync_loop.cpp ${SRC_ALL})
+target_link_libraries(simple_sync_loop ${LIB_REDIS})
@@ -17,10 +17,12 @@ int main(int argc, char* argv[]) {
   Redis rdx = {"localhost", 6379};
   rdx.run();
-
-  Command<int>* del_cmd = rdx.command_blocking<int>("DEL simple_loop:count");
-  cout << "deleted key, reply: " << del_cmd->reply() << endl;
-  del_cmd->free();
+//
+//  Command<int>* del_cmd = rdx.command_blocking<int>("DEL simple_loop:count");
+//  cout << "deleted key, reply: " << del_cmd->reply() << endl;
+//  del_cmd->free();
+  if(rdx.command_blocking("DEL simple_loop:count")) cout << "Deleted simple_loop:count" << endl;
+  else cerr << "Failed to delete simple_loop:count" << endl;
   Command<char*>* set_cmd = rdx.command_blocking<char*>("SET simple_loop:count 0");
   cout << "set key, reply: " << set_cmd->reply() << endl;
+/**
+* Basic asynchronous calls using redisx.
+*/
+
+#include <iostream>
+#include "../src/redisx.hpp"
+
+using namespace std;
+using namespace redisx;
+
+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[]) {
+
+  Redis 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;
+
+  string cmd_str = "INCR simple_loop:count";
+
+  int count = 50000;
+  double t0 = time_s();
+
+  cout << "Running \"" << cmd_str << "\" " << count << " times." << endl;
+
+  for(int i = 0; i < count; i++) {
+    Command<int>* c = rdx.command_blocking<int>(cmd_str);
+    if(c->status() != REDIS_OK) cerr << "Bad reply, code: " << c->status() << endl;
+  }
+
+  cout << "At the end, simple_loop:count = "
+    << rdx.command_blocking<string>("GET simple_loop:count")->reply() << endl;
+
+  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;
+
+  return 0;
+}
@@ -74,12 +74,12 @@ private:
   std::atomic_bool completed;
-  ev_timer* timer;
+  ev_timer timer;
   std::mutex timer_guard;
   ev_timer* get_timer() {
     std::lock_guard<std::mutex> lg(timer_guard);
-    return timer;
+    return &timer;
   }
 };
@@ -11,10 +11,8 @@ using namespace std;
 namespace redisx {
-// Default construct the static map
-std::unordered_map<ev_timer*, void*> Redis::timer_callbacks;
-
 // Global mutex to manage waiting for connected state
+// TODO get rid of this as the only global variable?
 mutex connected_lock;
 /**
@@ -119,12 +117,14 @@ bool submit_to_server(Command&lt;ReplyT&gt;* cmd_obj) {
 template<class ReplyT>
 void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents) {
-  auto cmd_obj = (Command<ReplyT>*)Redis::timer_callbacks.at(timer);
-  if(cmd_obj == NULL) {
+
+  // Check if canceled
+  if(timer->data == NULL) {
     cerr << "[WARNING] Skipping event, has been canceled." << endl;
-    Redis::timer_callbacks.erase(timer);
     return;
   }
+
+  auto cmd_obj = (Command<ReplyT>*)timer->data;
   submit_to_server<ReplyT>(cmd_obj);
 }
@@ -141,14 +141,11 @@ bool Redis::process_queued_command(void* cmd_ptr) {
   if((cmd_obj->repeat == 0) && (cmd_obj->after == 0)) {
     submit_to_server<ReplyT>(cmd_obj);
   } else {
-    // TODO manage memory somehow
-    cmd_obj->timer = new ev_timer();
-    // TODO use cmd_obj->timer->data instead of timer callbacks!!!!!
+    cmd_obj->timer.data = (void*)cmd_obj;
-    timer_callbacks[cmd_obj->timer] = (void*)cmd_obj;
-    ev_timer_init(cmd_obj->timer, submit_command_callback<ReplyT>, cmd_obj->after, cmd_obj->repeat);
-    ev_timer_start(EV_DEFAULT_ cmd_obj->timer);
+    ev_timer_init(&cmd_obj->timer, submit_command_callback<ReplyT>, cmd_obj->after, cmd_obj->repeat);
+    ev_timer_start(EV_DEFAULT_ &cmd_obj->timer);
     cmd_obj->timer_guard.unlock();
   }
@@ -242,8 +239,11 @@ void Redis::command(const string&amp; cmd) {
   command<redisReply*>(cmd, NULL);
 }
-void Redis::command_blocking(const string& cmd) {
-  command_blocking<redisReply*>(cmd);
+bool Redis::command_blocking(const string& cmd) {
+  Command<redisReply*>* c = command_blocking<redisReply*>(cmd);
+  bool succeeded = (c->status() == REDISX_OK);
+  c->free();
+  return succeeded;
 }
 } // End namespace redis
@@ -49,12 +49,11 @@ public:
   template<class ReplyT>
   bool cancel(Command<ReplyT>* cmd_obj);
-  void command(const std::string& command);
-
   template<class ReplyT>
   Command<ReplyT>* command_blocking(const std::string& cmd);
-  void command_blocking(const std::string& command);
+  void command(const std::string& command);
+  bool command_blocking(const std::string& command);
   long num_commands_processed();
@@ -62,10 +61,6 @@ public:
 //  void subscribe(std::string channel, std::function<void(std::string channel, std::string msg)> callback);
 //  void unsubscribe(std::string channel);
-  // Map of ev_timer events to pointers to Command objects
-  // Used to get the object back from the timer watcher callback
-  static std::unordered_map<ev_timer*, void*> timer_callbacks;
-
 private:
   // Redis server
@@ -154,13 +149,12 @@ bool Redis::cancel(Command&lt;ReplyT&gt;* cmd_obj) {
     return false;
   }
-  timer_callbacks.at(cmd_obj->timer) = NULL;
+  cmd_obj->timer.data = NULL;
   std::lock_guard<std::mutex> lg(cmd_obj->timer_guard);
   if((cmd_obj->repeat != 0) || (cmd_obj->after != 0))
-    ev_timer_stop(EV_DEFAULT_ cmd_obj->timer);
+    ev_timer_stop(EV_DEFAULT_ &cmd_obj->timer);
-  delete cmd_obj->timer;
   cmd_obj->completed = true;
   return true;
@@ -179,21 +173,22 @@ Command&lt;ReplyT&gt;* Redis::command_blocking(const std::string&amp; cmd) {
   Command<ReplyT>* cmd_obj = command<ReplyT>(cmd,
     [&val, &status, &m, &cv](const std::string& cmd_str, const ReplyT& reply) {
-      std::unique_lock<std::mutex> lk(m);
+      std::unique_lock<std::mutex> ul(m);
       val = reply;
       status = REDISX_OK;
-      lk.unlock();
+      ul.unlock();
       cv.notify_one();
     },
     [&status, &m, &cv](const std::string& cmd_str, int error) {
-      std::unique_lock<std::mutex> lk(m);
+      std::unique_lock<std::mutex> ul(m);
       status = error;
-      lk.unlock();
+      ul.unlock();
       cv.notify_one();
     },
     0, 0, false // No repeats, don't free memory
   );
+  // Wait until a callback is invoked
   cv.wait(lk, [&status] { return status != REDISX_UNINIT; });
   cmd_obj->reply_val = val;