patch issue concerning clients destroyed before the end of io_service callback execution. Handle subscribe/psubscribe ackowledgement reply. Patch randomly failing test (by using the new ackowledgement feature).

Author: Cylix

includes/cpp_redis/network/unix/io_service.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <atomic>
+#include <condition_variable>
 #include <mutex>
 #include <thread>
 #include <unordered_map>
@@ -65,6 +66,9 @@ class io_service {
     std::vector<char> write_buffer;
     std::size_t write_size;
     write_callback_t write_callback;
+
+    std::atomic_bool callback_running;
+    std::condition_variable_any callback_notification;
   };
 
 private:
@@ -79,9 +83,8 @@ class io_service {
   unsigned int init_sets(struct pollfd* fds);
   void process_sets(struct pollfd* fds, unsigned int nfds);
 
-  typedef std::function<void()> callback_t;
-  callback_t read_fd(int fd);
-  callback_t write_fd(int fd);
+  void read_fd(int fd);
+  void write_fd(int fd);
 
 private:
   //! whether the worker should terminate or not

includes/cpp_redis/redis_subscriber.hpp

@@ -29,28 +29,38 @@ class redis_subscriber {
 
   //! subscribe - unsubscribe
   typedef std::function<void(const std::string&, const std::string&)> subscribe_callback_t;
-  redis_subscriber& subscribe(const std::string& channel, const subscribe_callback_t& callback);
-  redis_subscriber& psubscribe(const std::string& pattern, const subscribe_callback_t& callback);
+  typedef std::function<void(int)> acknowledgement_callback_t;
+  redis_subscriber& subscribe(const std::string& channel, const subscribe_callback_t& callback, const acknowledgement_callback_t& acknowledgement_callback = nullptr);
+  redis_subscriber& psubscribe(const std::string& pattern, const subscribe_callback_t& callback, const acknowledgement_callback_t& acknowledgement_callback = nullptr);
   redis_subscriber& unsubscribe(const std::string& channel);
   redis_subscriber& punsubscribe(const std::string& pattern);
 
   //! commit pipelined transaction
   redis_subscriber& commit(void);
 
+private:
+  struct callback_holder {
+    subscribe_callback_t subscribe_callback;
+    acknowledgement_callback_t acknowledgement_callback;
+  };
+
 private:
   void connection_receive_handler(network::redis_connection&, reply& reply);
   void connection_disconnection_handler(network::redis_connection&);
 
+  void handle_acknowledgement_reply(const std::vector<reply>& reply);
   void handle_subscribe_reply(const std::vector<reply>& reply);
   void handle_psubscribe_reply(const std::vector<reply>& reply);
 
+  void call_acknowledgement_callback(const std::string& channel, const std::map<std::string, callback_holder>& channels, std::mutex& channels_mtx, int nb_chans);
+
 private:
   //! redis connection
   network::redis_connection m_client;
 
   //! (p)subscribed channels and their associated channels
-  std::map<std::string, subscribe_callback_t> m_subscribed_channels;
-  std::map<std::string, subscribe_callback_t> m_psubscribed_channels;
+  std::map<std::string, callback_holder> m_subscribed_channels;
+  std::map<std::string, callback_holder> m_psubscribed_channels;
 
   //! disconnection handler
   disconnection_handler_t m_disconnection_handler;

sources/network/redis_connection.cpp

@@ -13,6 +13,7 @@ redis_connection::redis_connection(const std::shared_ptr<io_service>& IO)
 }
 
 redis_connection::~redis_connection(void) {
+  m_client.disconnect();
   __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection destroyed");
 }
 

sources/network/unix/io_service.cpp

@@ -72,16 +72,16 @@ io_service::init_sets(struct pollfd* fds) {
   return nfds;
 }
 
-io_service::callback_t
+void
 io_service::read_fd(int fd) {
-  std::lock_guard<std::recursive_mutex> lock(m_fds_mutex);
+  std::unique_lock<std::recursive_mutex> lock(m_fds_mutex);
 
   __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service non-blocking read available for fd #" + std::to_string(fd));
 
   auto fd_it = m_fds.find(fd);
   if (fd_it == m_fds.end()) {
     __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service does not track fd #" + std::to_string(fd));
-    return nullptr;
+    return;
   }
 
   auto& buffer             = *fd_it->second.read_buffer;
@@ -97,26 +97,30 @@ io_service::read_fd(int fd) {
     buffer.resize(original_buffer_size);
     fd_it->second.disconnection_handler(*this);
     m_fds.erase(fd_it);
-
-    return nullptr;
   }
   else {
     buffer.resize(original_buffer_size + nb_bytes_read);
 
-    return std::bind(fd_it->second.read_callback, nb_bytes_read);
+    __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service calling read callback for fd #" + std::to_string(fd));
+
+    fd_it->second.callback_running = true;
+    lock.unlock();
+    fd_it->second.read_callback(nb_bytes_read);
+    fd_it->second.callback_running = false;
+    fd_it->second.callback_notification.notify_all();
   }
 }
 
-io_service::callback_t
+void
 io_service::write_fd(int fd) {
-  std::lock_guard<std::recursive_mutex> lock(m_fds_mutex);
+  std::unique_lock<std::recursive_mutex> lock(m_fds_mutex);
 
   __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service non-blocking write available for fd #" + std::to_string(fd));
 
   auto fd_it = m_fds.find(fd);
   if (fd_it == m_fds.end()) {
     __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service does not track fd #" + std::to_string(fd));
-    return nullptr;
+    return;
   }
 
   __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service writing data for fd #" + std::to_string(fd));
@@ -127,11 +131,16 @@ io_service::write_fd(int fd) {
     __CPP_REDIS_LOG(error, "cpp_redis::network::io_service write error for fd #" + std::to_string(fd));
     fd_it->second.disconnection_handler(*this);
     m_fds.erase(fd_it);
+  }
+  else {
+    __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service calling write callback for fd #" + std::to_string(fd));
 
-    return nullptr;
+    fd_it->second.callback_running = true;
+    lock.unlock();
+    fd_it->second.write_callback(nb_bytes_written);
+    fd_it->second.callback_running = false;
+    fd_it->second.callback_notification.notify_all();
   }
-  else
-    return std::bind(fd_it->second.write_callback, nb_bytes_written);
 }
 
 void
@@ -153,20 +162,8 @@ io_service::process_sets(struct pollfd* fds, unsigned int nfds) {
     }
   }
 
-  for (int fd : fds_to_read) {
-    auto callback = read_fd(fd);
-    if (callback) {
-      __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service calling read callback for fd #" + std::to_string(fd));
-      callback();
-    }
-  }
-  for (int fd : fds_to_write) {
-    auto callback = write_fd(fd);
-    if (callback) {
-      __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service calling write callback for fd #" + std::to_string(fd));
-      callback();
-    }
-  }
+  for (int fd : fds_to_read) { read_fd(fd); }
+  for (int fd : fds_to_write) { write_fd(fd); }
 
   if (fds[0].revents & POLLIN) {
     char buf[1024];
@@ -210,8 +207,22 @@ io_service::track(int fd, const disconnection_handler_t& handler) {
 
 void
 io_service::untrack(int fd) {
-  std::lock_guard<std::recursive_mutex> lock(m_fds_mutex);
-  m_fds.erase(fd);
+  std::unique_lock<std::recursive_mutex> lock(m_fds_mutex);
+
+  __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service requests to untrack fd #" + std::to_string(fd));
+
+  auto fd_it = m_fds.find(fd);
+  if (fd_it == m_fds.end()) {
+    __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service does not track fd #" + std::to_string(fd));
+    return;
+  }
+
+  if (fd_it->second.callback_running) {
+    __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service waits for callbacks to complete before untracking fd #" + std::to_string(fd));
+    fd_it->second.callback_notification.wait(lock, [=] { return !fd_it->second.callback_running; });
+  }
+
+  m_fds.erase(fd_it);
 
   __CPP_REDIS_LOG(debug, "cpp_redis::network::io_service now untracks fd #" + std::to_string(fd));
 }

sources/redis_client.cpp

@@ -9,6 +9,7 @@ redis_client::redis_client(const std::shared_ptr<network::io_service>& IO)
 }
 
 redis_client::~redis_client(void) {
+  m_client.disconnect();
   __CPP_REDIS_LOG(debug, "cpp_redis::redis_client destroyed");
 }
 

sources/redis_subscriber.cpp

@@ -10,6 +10,7 @@ redis_subscriber::redis_subscriber(const std::shared_ptr<network::io_service>& I
 }
 
 redis_subscriber::~redis_subscriber(void) {
+  m_client.disconnect();
   __CPP_REDIS_LOG(debug, "cpp_redis::redis_subscriber destroyed");
 }
 
@@ -40,23 +41,23 @@ redis_subscriber::is_connected(void) {
 }
 
 redis_subscriber&
-redis_subscriber::subscribe(const std::string& channel, const subscribe_callback_t& callback) {
+redis_subscriber::subscribe(const std::string& channel, const subscribe_callback_t& callback, const acknowledgement_callback_t& acknowledgement_callback) {
   std::lock_guard<std::mutex> lock(m_subscribed_channels_mutex);
 
   __CPP_REDIS_LOG(debug, "cpp_redis::redis_subscriber attemps to subscribe to channel " + channel);
-  m_subscribed_channels[channel] = callback;
+  m_subscribed_channels[channel] = {callback, acknowledgement_callback};
   m_client.send({"SUBSCRIBE", channel});
   __CPP_REDIS_LOG(info, "cpp_redis::redis_subscriber subscribed to channel " + channel);
 
   return *this;
 }
 
 redis_subscriber&
-redis_subscriber::psubscribe(const std::string& pattern, const subscribe_callback_t& callback) {
+redis_subscriber::psubscribe(const std::string& pattern, const subscribe_callback_t& callback, const acknowledgement_callback_t& acknowledgement_callback) {
   std::lock_guard<std::mutex> lock(m_psubscribed_channels_mutex);
 
   __CPP_REDIS_LOG(debug, "cpp_redis::redis_subscriber attemps to psubscribe to channel " + pattern);
-  m_psubscribed_channels[pattern] = callback;
+  m_psubscribed_channels[pattern] = {callback, acknowledgement_callback};
   m_client.send({"PSUBSCRIBE", pattern});
   __CPP_REDIS_LOG(info, "cpp_redis::redis_subscriber psubscribed to channel " + pattern);
 
@@ -114,6 +115,40 @@ redis_subscriber::commit(void) {
   return *this;
 }
 
+void
+redis_subscriber::call_acknowledgement_callback(const std::string& channel, const std::map<std::string, callback_holder>& channels, std::mutex& channels_mtx, int nb_chans) {
+  std::lock_guard<std::mutex> lock(channels_mtx);
+
+  auto it = channels.find(channel);
+  if (it == channels.end())
+    return;
+
+  if (it->second.acknowledgement_callback) {
+    __CPP_REDIS_LOG(debug, "cpp_redis::redis_subscriber executes acknowledgement callback for channel " + channel);
+    it->second.acknowledgement_callback(nb_chans);
+  }
+}
+
+void
+redis_subscriber::handle_acknowledgement_reply(const std::vector<reply>& reply) {
+  if (reply.size() != 3)
+    return;
+
+  const auto& title    = reply[0];
+  const auto& channel  = reply[1];
+  const auto& nb_chans = reply[2];
+
+  if (!title.is_string()
+      || !channel.is_string()
+      || !nb_chans.is_integer())
+    return;
+
+  if (title.as_string() == "subscribe")
+    call_acknowledgement_callback(channel.as_string(), m_subscribed_channels, m_subscribed_channels_mutex, nb_chans.as_integer());
+  else if (title.as_string() == "psubscribe")
+    call_acknowledgement_callback(channel.as_string(), m_psubscribed_channels, m_psubscribed_channels_mutex, nb_chans.as_integer());
+}
+
 void
 redis_subscriber::handle_subscribe_reply(const std::vector<reply>& reply) {
   if (reply.size() != 3)
@@ -138,7 +173,7 @@ redis_subscriber::handle_subscribe_reply(const std::vector<reply>& reply) {
     return;
 
   __CPP_REDIS_LOG(debug, "cpp_redis::redis_subscriber executes subscribe callback for channel " + channel.as_string());
-  it->second(channel.as_string(), message.as_string());
+  it->second.subscribe_callback(channel.as_string(), message.as_string());
 }
 
 void
@@ -167,7 +202,7 @@ redis_subscriber::handle_psubscribe_reply(const std::vector<reply>& reply) {
     return;
 
   __CPP_REDIS_LOG(debug, "cpp_redis::redis_subscriber executes psubscribe callback for channel " + channel.as_string());
-  it->second(channel.as_string(), message.as_string());
+  it->second.subscribe_callback(channel.as_string(), message.as_string());
 }
 
 void
@@ -180,10 +215,13 @@ redis_subscriber::connection_receive_handler(network::redis_connection&, reply&
 
   auto& array = reply.as_array();
 
-  //! Array size of 3 -> SUBSCRIBE
+  //! Array size of 3 -> SUBSCRIBE if array[2] is a string
+  //! Array size of 3 -> AKNOWLEDGEMENT if array[2] is an integer
   //! Array size of 4 -> PSUBSCRIBE
   //! Otherwise -> unexepcted reply
-  if (array.size() == 3)
+  if (array.size() == 3 && array[2].is_integer())
+    handle_acknowledgement_reply(array);
+  else if (array.size() == 3 && array[2].is_string())
     handle_subscribe_reply(array);
   else if (array.size() == 4)
     handle_psubscribe_reply(array);

tests/sources/main.cpp

@@ -1,7 +1,15 @@
 #include <gtest/gtest.h>
 
+//! For debugging purpose, uncomment
+// #include <memory>
+// #include <cpp_redis/cpp_redis>
+
 int
 main(int argc, char** argv) {
+  //! For debugging purpose, uncomment
+  // cpp_redis::active_logger = std::unique_ptr<cpp_redis::logger>(new cpp_redis::logger(cpp_redis::logger::log_level::debug));
+
   ::testing::InitGoogleTest(&argc, argv);
+
   return RUN_ALL_TESTS();
 }

tests/sources/spec/redis_client_spec.cpp

@@ -234,28 +234,40 @@ TEST(RedisClient, MultipleSendPipeline) {
 
 TEST(RedisClient, DisconnectionHandlerWithQuit) {
   cpp_redis::redis_client client;
+  std::condition_variable cv;
 
   std::atomic_bool disconnection_handler_called(false);
   client.connect("127.0.0.1", 6379, [&](cpp_redis::redis_client&) {
     disconnection_handler_called = true;
+    cv.notify_all();
   });
 
   client.send({"QUIT"});
   client.sync_commit();
-  std::this_thread::sleep_for(std::chrono::seconds(1));
+
+  std::mutex mutex;
+  std::unique_lock<std::mutex> lock(mutex);
+  cv.wait_for(lock, std::chrono::seconds(2));
+
   EXPECT_TRUE(disconnection_handler_called);
 }
 
 TEST(RedisClient, DisconnectionHandlerWithoutQuit) {
   cpp_redis::redis_client client;
+  std::condition_variable cv;
 
   std::atomic_bool disconnection_handler_called(false);
   client.connect("127.0.0.1", 6379, [&](cpp_redis::redis_client&) {
     disconnection_handler_called = true;
+    cv.notify_all();
   });
 
   client.sync_commit();
-  std::this_thread::sleep_for(std::chrono::seconds(1));
+
+  std::mutex mutex;
+  std::unique_lock<std::mutex> lock(mutex);
+  cv.wait_for(lock, std::chrono::seconds(2));
+
   EXPECT_FALSE(disconnection_handler_called);
 }