Working loop command implementation

Uses libev timer watchers to queue commands on a regular interval. Very fast. Still need to handle deactivation of timers from the user side.

Working loop command implementation
Uses libev timer watchers to queue commands on a regular interval. Very fast. Still need to handle deactivation of timers from the user side.
Hayk Martirosyan
1 parent 2833f2e8
Showing 4 changed files with 111 additions and 40 deletions
CMakeLists.txt
examples/simple_loop.cpp
src/redisx.cpp
src/redisx.hpp
 cmake_minimum_required(VERSION 2.8.4)
 project(redisx)
-#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)
 # ---------------------------------------------------------
@@ -34,8 +34,11 @@ 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(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})
+/**
+* Basic asynchronous calls using redisx.
+*/
+
+#include <iostream>
+#include "../src/redisx.hpp"
+
+using namespace std;
+
+double time_s() {
+  unsigned long ms = chrono::system_clock::now().time_since_epoch() / chrono::milliseconds(1);
+  return (double)ms / 1000;
+}
+
+int main(int argc, char* argv[]) {
+
+  redisx::Redis rdx = {"localhost", 6379};
+  rdx.run();
+
+  string cmd_str = "SET alaska rules!";
+
+  double freq = 10000; // Hz
+  double t_end = 5;
+
+  double dt = 1 / freq;
+  double t0 = time_s();
+  int count = 0;
+
+  cout << "Running \"" << cmd_str << "\" at dt = " << dt
+      << "s for " << t_end << "s..." << endl;
+
+  rdx.command<const string &>(
+      cmd_str,
+      [&count, &rdx, t0, t_end](const string &cmd, const string &value) {
+        count++;
+        if(time_s() - t0 >= t_end) rdx.stop();
+      },
+      dt,
+      dt
+  );
+
+  rdx.block_until_stopped();
+  double actual_freq = (double)count / t_end;
+  cout << "Sent " << count << " commands in " << t_end<< "s, "
+       << "that's " << actual_freq << " commands/s." << endl;
+
+  return 0;
+}
@@ -3,20 +3,20 @@
 */
 #include <signal.h>
-#include <iostream>
-#include <thread>
-#include <hiredis/adapters/libev.h>
-#include <ev.h>
-#include <event2/thread.h>
-#include <vector>
+//#include <event2/thread.h>
 #include "redisx.hpp"
 using namespace std;
 namespace redisx {
+// Global mutex to manage waiting for connected state
 mutex connected_lock;
+// Map of ev_timer events to pointers to CommandAsync objects
+// Used to get the object back from the timer watcher callback
+unordered_map<ev_timer*, void*> timer_callbacks;
+
 void connected(const redisAsyncContext *c, int status) {
   if (status != REDIS_OK) {
     cerr << "[ERROR] Connecting to Redis: " << c->errstr << endl;
@@ -94,17 +94,27 @@ void Redis::block_until_stopped() {
   exit_waiter.wait(ul, [this]() { return to_exit.load(); });
 }
+/**
+* Submit an asynchronous command to the Redis server. Return
+* true if succeeded, false otherwise.
+*/
 template<class ReplyT>
-bool Redis::submit_to_server(const CommandAsync<ReplyT>* cmd_obj) {
-  if (redisAsyncCommand(c, command_callback<ReplyT>, (void*)cmd_obj, cmd_obj->cmd.c_str()) != REDIS_OK) {
-    cerr << "[ERROR] Async command \"" << cmd_obj->cmd << "\": " << c->errstr << endl;
-    delete cmd_obj;
+bool submit_to_server(CommandAsync<ReplyT>* cmd_obj) {
+  if (redisAsyncCommand(cmd_obj->c, command_callback<ReplyT>, (void*)cmd_obj, cmd_obj->cmd.c_str()) != REDIS_OK) {
+    cerr << "[ERROR] Async command \"" << cmd_obj->cmd << "\": " << cmd_obj->c->errstr << endl;
+    cmd_obj->free_if_done();
     return false;
   }
   return true;
 }
 template<class ReplyT>
+void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents) {
+  auto cmd_obj = (CommandAsync<ReplyT>*)timer_callbacks.at(timer);
+  submit_to_server<ReplyT>(cmd_obj);
+}
+
+template<class ReplyT>
 bool Redis::process_queued_command(void* cmd_ptr) {
   auto& command_map = get_command_map<ReplyT>();
@@ -114,7 +124,14 @@ bool Redis::process_queued_command(void* cmd_ptr) {
   CommandAsync<ReplyT>* cmd_obj = it->second;
   command_map.erase(cmd_ptr);
-  submit_to_server<ReplyT>(cmd_obj);
+  if((cmd_obj->repeat == 0) && (cmd_obj->after == 0)) {
+    submit_to_server<ReplyT>(cmd_obj);
+  } else {
+    cmd_obj->timer = new ev_timer();
+    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);
+  }
   return true;
 }
@@ -147,13 +164,13 @@ long Redis::num_commands_processed() {
 template<> unordered_map<void*, CommandAsync<const redisReply*>*>& Redis::get_command_map() { return commands_redis_reply; }
 template<>
-void invoke_callback(const CommandAsync<const redisReply*>* cmd_obj, redisReply* reply) {
+void invoke_callback(CommandAsync<const redisReply*>* cmd_obj, redisReply* reply) {
   cmd_obj->invoke(reply);
 }
 template<> unordered_map<void*, CommandAsync<const string&>*>& Redis::get_command_map() { return commands_string_r; }
 template<>
-void invoke_callback(const CommandAsync<const string&>* cmd_obj, redisReply* reply) {
+void invoke_callback(CommandAsync<const string&>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_STRING && reply->type != REDIS_REPLY_STATUS) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-string reply." << endl;
     return;
@@ -164,7 +181,7 @@ void invoke_callback(const CommandAsync&lt;const string&amp;&gt;* cmd_obj, redisReply* rep
 template<> unordered_map<void*, CommandAsync<const char*>*>& Redis::get_command_map() { return commands_char_p; }
 template<>
-void invoke_callback(const CommandAsync<const char*>* cmd_obj, redisReply* reply) {
+void invoke_callback(CommandAsync<const char*>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_STRING && reply->type != REDIS_REPLY_STATUS) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-string reply." << endl;
     return;
@@ -174,7 +191,7 @@ void invoke_callback(const CommandAsync&lt;const char*&gt;* cmd_obj, redisReply* reply
 template<> unordered_map<void*, CommandAsync<int>*>& Redis::get_command_map() { return commands_int; }
 template<>
-void invoke_callback(const CommandAsync<int>* cmd_obj, redisReply* reply) {
+void invoke_callback(CommandAsync<int>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_INTEGER) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-integer reply." << endl;
     return;
@@ -184,7 +201,7 @@ void invoke_callback(const CommandAsync&lt;int&gt;* cmd_obj, redisReply* reply) {
 template<> unordered_map<void*, CommandAsync<long long int>*>& Redis::get_command_map() { return commands_long_long_int; }
 template<>
-void invoke_callback(const CommandAsync<long long int>* cmd_obj, redisReply* reply) {
+void invoke_callback(CommandAsync<long long int>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_INTEGER) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-integer reply." << endl;
     return;
@@ -18,7 +18,7 @@
 #include <hiredis/hiredis.h>
 #include <hiredis/async.h>
-#include <condition_variable>
+#include <hiredis/adapters/libev.h>
 namespace redisx {
@@ -26,17 +26,30 @@ template&lt;class ReplyT&gt;
 class CommandAsync {
 public:
   CommandAsync(
+      redisAsyncContext* c,
       const std::string& cmd,
       const std::function<void(const std::string&, ReplyT)>& callback,
       double repeat, double after
-  ) : cmd(cmd), callback(callback), repeat(repeat), after(after) {}
+  ) : c(c), cmd(cmd), callback(callback), repeat(repeat), after(after), done(false) {}
+  redisAsyncContext* c;
   const std::string cmd;
   const std::function<void(const std::string&, ReplyT)> callback;
   double repeat;
   double after;
+  bool done;
+  ev_timer* timer;
-  void invoke(ReplyT reply) const {if(callback != NULL) callback(cmd, reply); }
+  void invoke(ReplyT reply) {
+    if(callback != NULL) callback(cmd, reply);
+    if((repeat == 0)) done = true;
+  }
+  void free_if_done() {
+    if(done) {
+      std::cout << "Deleting CommandAsync: " << cmd << std::endl;
+      delete this;
+    };
+  }
 };
 class Redis {
@@ -63,8 +76,6 @@ public:
   long num_commands_processed();
-//  struct event* command_loop(const char* command, long interval_s, long interval_us);
-
 //  void get(const char* key, std::function<void(const std::string&, const char*)> callback);
 //
 //  void set(const char* key, const char* value);
@@ -109,20 +120,13 @@ private:
   template<class ReplyT>
   bool process_queued_command(void* cmd_ptr);
-
-  /**
-  * Submit an asynchronous command to the Redis server. Return
-  * true if succeeded, false otherwise.
-  */
-  template<class ReplyT>
-  bool submit_to_server(const CommandAsync<ReplyT>* cmd_obj);
 };
 // ---------------------------
 template<class ReplyT>
 void invoke_callback(
-    const CommandAsync<ReplyT>* cmd_obj,
+    CommandAsync<ReplyT>* cmd_obj,
     redisReply* reply
 );
@@ -134,18 +138,18 @@ void command_callback(redisAsyncContext *c, void *r, void *privdata) {
   if (reply->type == REDIS_REPLY_ERROR) {
     std::cerr << "[ERROR] " << cmd_obj->cmd << ": " << reply->str << std::endl;
-    delete cmd_obj;
+    cmd_obj->free_if_done();
     return;
   }
   if(reply->type == REDIS_REPLY_NIL) {
     std::cerr << "[WARNING] " << cmd_obj->cmd << ": Nil reply." << std::endl;
-    delete cmd_obj;
+    cmd_obj->free_if_done();
     return; // cmd_obj->invoke(NULL);
   }
   invoke_callback<ReplyT>(cmd_obj, reply);
-  delete cmd_obj;
+  cmd_obj->free_if_done();
 }
 template<class ReplyT>
@@ -157,10 +161,9 @@ void Redis::command(
 ) {
   std::lock_guard<std::mutex> lg(queue_guard);
-  auto* cmd_obj = new CommandAsync<ReplyT>(cmd, callback, repeat, after);
+  auto* cmd_obj = new CommandAsync<ReplyT>(c, cmd, callback, repeat, after);
   get_command_map<ReplyT>()[(void*)cmd_obj] = cmd_obj;
   command_queue.push((void*)cmd_obj);
 }
-
 } // End namespace redis