Unsafe use of detached threads can lead to access violations

[nkochakian]

There are several cases where client::clear_callbacks is called, one of which is from client::disconnect. clear_callbacks will copy a list of callbacks to execute on a detached thread and update the state of the callback condition variable through several member variables. There are a few issues that need to be addressed with this approach:

1. The client can be deleted while the detached thread is still running, which will cause access violations when the thread tries to modify member variables of a deleted instance.

   • It would be a good idea to block on the condition variable inside the destructor and wait for the callback count to reach 0, which will ensure that the zero or more outstanding detached threads have finished.

2. There is no user accessible function that can be used to wait for existing callback functions that are running in the background to finish. This can be important when using disconnect since it will start a detached thread to reject commands that haven't been sent. sync_commit comes close since the end of the function waits on the condition variable, but that will never happen if try_commit is called on a disconnected client and it throws an exception.

The documentation of client::disconnect should also be changed since it currently states

wait_for_removal when sets to true, disconnect blocks until the underlying TCP client has
been effectively removed from the io_service and that all the underlying callbacks have completed.

I'm assuming it's referring to internal I/O callbacks, but the wording makes it seem like it will block until command callbacks have been completed.

This is a Windows example that can be used to reproduce the access violation on delete:

#include <cpp_redis/cpp_redis>
#include <iostream>
#include <memory>

int main(void)
{
   WORD version = MAKEWORD(2, 2);
   WSADATA data;

   if (WSAStartup(version, &data) != 0)
      return 1;

   std::unique_ptr<cpp_redis::client> client = std::make_unique<cpp_redis::client>();

   client->connect("localhost",
                   6379,
                   [] (const std::string &, std::size_t, cpp_redis::client::connect_state status)
                   {
                      switch (status)
                      {
                         case cpp_redis::client::connect_state::ok:
                            std::cout << "Connected" << std::endl;
                         break;

                         case cpp_redis::client::connect_state::dropped:
                            std::cout << "Disconnected" << std::endl;
                         break;
                      }
                   });

   client->get("1",
               [] (cpp_redis::reply &)
               {
                  Sleep(4000);
               });

   client->get("2",
               [] (cpp_redis::reply &)
               {
                  std::cout << "Callback 2" << std::endl;
                  Sleep(4000);
               });

   std::cout << "Disconnecting" << std::endl;

   client->disconnect(true);

   std::cout << "Deleting" << std::endl;

   client.reset();

   std::cout << "Waiting" << std::endl;

   Sleep(INFINITE);

   return 0;
}