From 57886001aaf3a094e31d9d1289a708ce7e74b177 Mon Sep 17 00:00:00 2001
From: Binbin <binloveplay1314@qq.com>
Date: Sat, 2 Sep 2023 14:26:32 +0800
Subject: [PATCH] Split Inter in ZSet::InterStore into a separate function

CI TSAN show ZINTERSTORE may has a deadlock after introducing
locks to DEL in #1712. In ZSet::InterStore if the dst key was
inside the source key list we may have a deadlock since the
OverWrite function will also lock the dst key.

In this PR, we split Inter in ZSet::InterStore into a separate
function, just like the Set apis.

After this PR, after the CI verification in #1712, it can pass
the CI verification now.
---
 src/types/redis_zset.cc | 19 ++++++++++++++-----
 src/types/redis_zset.h  |  2 ++
 2 files changed, 16 insertions(+), 5 deletions(-)

diff --git a/src/types/redis_zset.cc b/src/types/redis_zset.cc
index 09495151be7..80d34c51dfd 100644
--- a/src/types/redis_zset.cc
+++ b/src/types/redis_zset.cc
@@ -626,6 +626,16 @@ rocksdb::Status ZSet::Overwrite(const Slice &user_key, const MemberScores &mscor
 
 rocksdb::Status ZSet::InterStore(const Slice &dst, const std::vector<KeyWeight> &keys_weights,
                                  AggregateMethod aggregate_method, uint64_t *saved_cnt) {
+  *saved_cnt = 0;
+  std::vector<MemberScore> members;
+  auto s = Inter(keys_weights, aggregate_method, saved_cnt, &members);
+  if (!s.ok()) return s;
+  *saved_cnt = members.size();
+  return Overwrite(dst, members);
+}
+
+rocksdb::Status ZSet::Inter(const std::vector<KeyWeight> &keys_weights, AggregateMethod aggregate_method,
+                            uint64_t *saved_cnt, std::vector<MemberScore> *members) {
   std::vector<std::string> lock_keys;
   lock_keys.reserve(keys_weights.size());
   for (const auto &key_weight : keys_weights) {
@@ -680,14 +690,13 @@ rocksdb::Status ZSet::InterStore(const Slice &dst, const std::vector<KeyWeight>
       }
     }
   }
-  if (!dst_zset.empty()) {
-    std::vector<MemberScore> mscores;
+  if (members && !dst_zset.empty()) {
+    members->reserve(dst_zset.size());
     for (const auto &iter : dst_zset) {
       if (member_counters[iter.first] != keys_weights.size()) continue;
-      mscores.emplace_back(MemberScore{iter.first, iter.second});
+      members->emplace_back(MemberScore{iter.first, iter.second});
     }
-    if (saved_cnt) *saved_cnt = mscores.size();
-    Overwrite(dst, mscores);
+    if (saved_cnt) *saved_cnt = members->size();
   }
 
   return rocksdb::Status::OK();
diff --git a/src/types/redis_zset.h b/src/types/redis_zset.h
index 1ea0b192f06..6ca4780e76b 100644
--- a/src/types/redis_zset.h
+++ b/src/types/redis_zset.h
@@ -109,6 +109,8 @@ class ZSet : public SubKeyScanner {
   rocksdb::Status Overwrite(const Slice &user_key, const MemberScores &mscores);
   rocksdb::Status InterStore(const Slice &dst, const std::vector<KeyWeight> &keys_weights,
                              AggregateMethod aggregate_method, uint64_t *saved_cnt);
+  rocksdb::Status Inter(const std::vector<KeyWeight> &keys_weights, AggregateMethod aggregate_method,
+                        uint64_t *saved_cnt, std::vector<MemberScore> *members);
   rocksdb::Status UnionStore(const Slice &dst, const std::vector<KeyWeight> &keys_weights,
                              AggregateMethod aggregate_method, uint64_t *saved_cnt);
   rocksdb::Status Union(const std::vector<KeyWeight> &keys_weights, AggregateMethod aggregate_method,