Add RedisContext ping before returning it to the pool

src/atomdb/redis_mongodb/RedisContext.cc

@@ -39,6 +39,13 @@ redisReply* RedisContext::execute(const char* command) {
     }
 }
 
+bool RedisContext::ping() {
+    redisReply* reply = this->execute("PING");
+    if (reply == NULL) return false;
+    if (reply->type != REDIS_REPLY_STATUS) return false;
+    return string(reply->str) == string("PONG");
+}
+
 void RedisContext::append_command(const char* command) {
     if (cluster_flag) {
         if (redisClusterAppendCommand(cluster_ctx, command) == REDIS_OK) {

src/atomdb/redis_mongodb/RedisContext.h

@@ -17,6 +17,7 @@ class RedisContext {
     void set_context(redisClusterContext* ctx);
 
     redisReply* execute(const char* command);
+    bool ping();
     void append_command(const char* command);
     void flush_commands();
     bool has_error() const;

src/atomdb/redis_mongodb/RedisContextPool.cc

@@ -32,7 +32,13 @@ shared_ptr<RedisContext> RedisContextPool::acquire() {
     if (!pool.empty()) {
         auto ctx = pool.front();
         pool.pop();
-        return ctx;
+        if (ctx->ping()) {
+            LOG_DEBUG("Context is alive, returning to the caller.");
+            return shared_ptr<RedisContext>(ctx.get(),
+                                            [this, ctx](RedisContext* ptr) { this->release(ctx); });
+        }
+        LOG_DEBUG("Context is not alive, creating a new one.");
+        total_contexts--;
     }
 
     // Create a new context if the pool is empty
@@ -83,9 +89,9 @@ shared_ptr<RedisContext> RedisContextPool::acquire() {
 
 void RedisContextPool::release(shared_ptr<RedisContext> ctx) {
     if (!ctx) return;
-
     unique_lock<mutex> lock(this->pool_mutex);
     pool.push(ctx);
-    total_contexts--;
-    pool_cond_var.notify_one();
+    LOG_DEBUG("Context added to the pool (size: " << pool.size()
+                                                  << ", contexts: " << to_string(total_contexts) << ")");
+    pool_cond_var.notify_all();
 }

src/atomdb/redis_mongodb/RedisMongoDB.cc

@@ -375,8 +375,15 @@ uint RedisMongoDB::get_next_score(const string& key) {
 
 void RedisMongoDB::set_next_score(const string& key, uint score) {
     if (SKIP_REDIS) return;
-
     auto ctx = this->redis_pool->acquire();
+    set_next_score_with_context(ctx, key, score);
+}
+
+void RedisMongoDB::set_next_score_with_context(shared_ptr<RedisContext> ctx,
+                                               const string& key,
+                                               uint score) {
+    if (SKIP_REDIS) return;
+
     string command = "SET " + key + " " + to_string(score);
     redisReply* reply = ctx->execute(command.c_str());
     if (reply == NULL) Utils::error("Redis error at set_next_score: <" + command + ">");
@@ -1012,8 +1019,11 @@ vector<string> RedisMongoDB::add_links(const vector<atoms::Link*>& links,
 
     LOG_DEBUG("Setting next scores: patterns=" + to_string(this->patterns_next_score.load()) +
               ", incoming=" + to_string(this->incoming_set_next_score.load()));
-    set_next_score(REDIS_PATTERNS_PREFIX + ":next_score", this->patterns_next_score.load());
-    set_next_score(REDIS_INCOMING_PREFIX + ":next_score", this->incoming_set_next_score.load());
+
+    set_next_score_with_context(
+        ctx, REDIS_PATTERNS_PREFIX + ":next_score", this->patterns_next_score.load());
+    set_next_score_with_context(
+        ctx, REDIS_INCOMING_PREFIX + ":next_score", this->incoming_set_next_score.load());
 
     if (is_transactional) {
         lock_guard<mutex> composite_type_hashes_map_lock(this->composite_type_hashes_map_mutex);

src/atomdb/redis_mongodb/RedisMongoDB.h

@@ -170,6 +170,7 @@ class RedisMongoDB : public AtomDB {
 
     uint get_next_score(const string& key);
     void set_next_score(const string& key, uint score);
+    void set_next_score_with_context(shared_ptr<RedisContext> ctx, const string& key, uint score);
     void reset_scores();
 
     void add_pattern(const string& handle, const string& pattern_handle);