prevent possible callback interleaving issue and unlock mutex in io_service when calling callbacks

Author: Cylix

includes/cpp_redis/network/io_service.hpp

@@ -75,10 +75,12 @@ class io_service {
 private:
   //! select fds sets handling (init, rd/wr handling)
   int init_sets(fd_set* rd_set, fd_set* wr_set);
-  void read_fd(int fd);
-  void write_fd(int fd);
   void process_sets(fd_set* rd_set, fd_set* wr_set);
 
+  typedef std::function<void()> callback_t;
+  callback_t read_fd(int fd);
+  callback_t write_fd(int fd);
+
 private:
   //! whether the worker should terminate or not
   std::atomic_bool m_should_stop;

includes/cpp_redis/redis_client.hpp

@@ -55,6 +55,7 @@ class redis_client {
 
   //! thread safety
   std::mutex m_callbacks_mutex;
+  std::mutex m_send_mutex;
 };
 
 } //! cpp_redis

sources/network/io_service.cpp

@@ -59,13 +59,13 @@ io_service::init_sets(fd_set* rd_set, fd_set* wr_set) {
   return max_fd;
 }
 
-void
+io_service::callback_t
 io_service::read_fd(int fd) {
   std::lock_guard<std::recursive_mutex> lock(m_fds_mutex);
 
   auto fd_it = m_fds.find(fd);
   if (fd_it == m_fds.end())
-    return ;
+    return nullptr;
 
   auto& buffer = *fd_it->second.read_buffer;
   int original_buffer_size = buffer.size();
@@ -78,30 +78,35 @@ 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);
-    fd_it->second.read_callback(nb_bytes_read);
+
+    return std::bind(fd_it->second.read_callback, nb_bytes_read);
   }
 }
 
-void
+io_service::callback_t
 io_service::write_fd(int fd) {
   std::lock_guard<std::recursive_mutex> lock(m_fds_mutex);
 
   auto fd_it = m_fds.find(fd);
   if (fd_it == m_fds.end())
-    return ;
+    return nullptr;
 
   int nb_bytes_written = send(fd_it->first, fd_it->second.write_buffer.data(), fd_it->second.write_size, 0);
   fd_it->second.async_write = false;
 
   if (nb_bytes_written <= 0) {
     fd_it->second.disconnection_handler(*this);
     m_fds.erase(fd_it);
+
+    return nullptr;
   }
   else
-    fd_it->second.write_callback(nb_bytes_written);
+    return std::bind(fd_it->second.write_callback, nb_bytes_written);
 }
 
 void
@@ -123,8 +128,16 @@ io_service::process_sets(fd_set* rd_set, fd_set* wr_set) {
     }
   }
 
-  for (int fd : fds_to_read) { read_fd(fd); }
-  for (int fd : fds_to_write) { write_fd(fd); }
+  for (int fd : fds_to_read) {
+    auto callback = read_fd(fd);
+    if (callback)
+      callback();
+  }
+  for (int fd : fds_to_write) {
+    auto callback = write_fd(fd);
+    if (callback)
+      callback();
+  }
 
   if (FD_ISSET(m_notif_pipe_fds[0], rd_set)) {
     char buf[1024];

sources/redis_client.cpp

@@ -31,9 +31,9 @@ redis_client::is_connected(void) {
 
 redis_client&
 redis_client::send(const std::vector<std::string>& redis_cmd, const reply_callback_t& callback) {
-  m_client.send(redis_cmd);
+  std::lock_guard<std::mutex> lock_callback(m_callbacks_mutex);
 
-  std::lock_guard<std::mutex> lock(m_callbacks_mutex);
+  m_client.send(redis_cmd);
   m_callbacks.push(callback);
 
   return *this;
@@ -49,15 +49,19 @@ redis_client::commit(void) {
 
 void
 redis_client::connection_receive_handler(network::redis_connection&, reply& reply) {
-  std::lock_guard<std::mutex> lock(m_callbacks_mutex);
+  reply_callback_t callback;
 
-  if (not m_callbacks.size())
-    return ;
+  {
+    std::lock_guard<std::mutex> lock(m_callbacks_mutex);
 
-  if (m_callbacks.front())
-    m_callbacks.front()(reply);
+    if (m_callbacks.size()) {
+      callback = m_callbacks.front();
+      m_callbacks.pop();
+    }
+  }
 
-  m_callbacks.pop();
+  if (callback)
+    callback(reply);
 }
 
 void