From 8d8b6e5bfab1993718580205612e5397215a1bd6 Mon Sep 17 00:00:00 2001
From: Philipp Moritz <pcmoritz@gmail.com>
Date: Mon, 22 Oct 2018 22:31:13 -0700
Subject: [PATCH] Retry connections to redis for async and subscribe contexts
 (#3105)

This is fixing a problem that @devin-petersohn observed on the windows subsystem for linux.

In theory, redis should be up once the async connect is happening and there should be no retries needed for the async connect. However on the windows subsystem for linux, the async connect was failing even though the synchronous one was working. Maybe windows has a different semantics here than linux.
---
 src/ray/gcs/redis_context.cc | 36 ++++++++++++++++++------------------
 1 file changed, 18 insertions(+), 18 deletions(-)

diff --git a/src/ray/gcs/redis_context.cc b/src/ray/gcs/redis_context.cc
index 1a8111963..f9e24cd93 100644
--- a/src/ray/gcs/redis_context.cc
+++ b/src/ray/gcs/redis_context.cc
@@ -157,27 +157,35 @@ Status AuthenticateRedis(redisAsyncContext *context, const std::string &password
   return Status::OK();
 }
 
-Status RedisContext::Connect(const std::string &address, int port, bool sharding,
-                             const std::string &password = "") {
+template <typename RedisContext, typename RedisConnectFunction>
+Status ConnectWithRetries(const std::string &address, int port,
+                          const RedisConnectFunction &connect_function,
+                          RedisContext **context) {
   int connection_attempts = 0;
-  context_ = redisConnect(address.c_str(), port);
-  while (context_ == nullptr || context_->err) {
+  *context = connect_function(address.c_str(), port);
+  while (*context == nullptr || (*context)->err) {
     if (connection_attempts >= RayConfig::instance().redis_db_connect_retries()) {
-      if (context_ == nullptr) {
+      if (*context == nullptr) {
         RAY_LOG(FATAL) << "Could not allocate redis context.";
       }
-      if (context_->err) {
+      if ((*context)->err) {
         RAY_LOG(FATAL) << "Could not establish connection to redis " << address << ":"
-                       << port;
+                       << port << " (context.err = " << (*context)->err << ")";
       }
       break;
     }
     RAY_LOG(WARNING) << "Failed to connect to Redis, retrying.";
     // Sleep for a little.
     usleep(RayConfig::instance().redis_db_connect_wait_milliseconds() * 1000);
-    context_ = redisConnect(address.c_str(), port);
+    *context = connect_function(address.c_str(), port);
     connection_attempts += 1;
   }
+  return Status::OK();
+}
+
+Status RedisContext::Connect(const std::string &address, int port, bool sharding,
+                             const std::string &password = "") {
+  RAY_CHECK_OK(ConnectWithRetries(address, port, redisConnect, &context_));
   RAY_CHECK_OK(AuthenticateRedis(context_, password));
 
   redisReply *reply = reinterpret_cast<redisReply *>(
@@ -186,19 +194,11 @@ Status RedisContext::Connect(const std::string &address, int port, bool sharding
   freeReplyObject(reply);
 
   // Connect to async context
-  async_context_ = redisAsyncConnect(address.c_str(), port);
-  if (async_context_ == nullptr || async_context_->err) {
-    RAY_LOG(FATAL) << "Could not establish connection to redis " << address << ":"
-                   << port;
-  }
+  RAY_CHECK_OK(ConnectWithRetries(address, port, redisAsyncConnect, &async_context_));
   RAY_CHECK_OK(AuthenticateRedis(async_context_, password));
 
   // Connect to subscribe context
-  subscribe_context_ = redisAsyncConnect(address.c_str(), port);
-  if (subscribe_context_ == nullptr || subscribe_context_->err) {
-    RAY_LOG(FATAL) << "Could not establish subscribe connection to redis " << address
-                   << ":" << port;
-  }
+  RAY_CHECK_OK(ConnectWithRetries(address, port, redisAsyncConnect, &subscribe_context_));
   RAY_CHECK_OK(AuthenticateRedis(subscribe_context_, password));
 
   return Status::OK();