wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-08-15 12:45:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix resubscribe miss callback index bug (#8604)

Browse Source
This commit is contained in:
fangfengbin
2020-05-27 11:55:17 +08:00
committed by GitHub
parent 99dd6a581d
commit b0cf781152
2 changed files with 14 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/gcs/pubsub/gcs_pub_sub.cc
+13 -4
View File
@@ -45,8 +45,15 @@ Status GcsPubSub::SubscribeAll(const std::string &channel, const Callback &subsc
}
Status GcsPubSub::Unsubscribe(const std::string &channel, const std::string &id) {
return redis_client_->GetPrimaryContext()->PUnsubscribeAsync(
GenChannelPattern(channel, boost::optional<std::string>(id)));
std::string pattern = GenChannelPattern(channel, id);
{
absl::MutexLock lock(&mutex_);
auto it = subscribe_callback_index_.find(pattern);
RAY_CHECK(it != subscribe_callback_index_.end());
unsubscribe_callback_index_[pattern] = it->second;
subscribe_callback_index_.erase(it);
}
return redis_client_->GetPrimaryContext()->PUnsubscribeAsync(pattern);
}
Status GcsPubSub::SubscribeInternal(const std::string &channel, const Callback &subscribe,
@@ -58,8 +65,8 @@ Status GcsPubSub::SubscribeInternal(const std::string &channel, const Callback &
if (reply->IsUnsubscribeCallback()) {
absl::MutexLock lock(&mutex_);
ray::gcs::RedisCallbackManager::instance().remove(
subscribe_callback_index_[pattern]);
subscribe_callback_index_.erase(pattern);
unsubscribe_callback_index_[pattern]);
unsubscribe_callback_index_.erase(pattern);
} else if (reply->IsSubscribeCallback()) {
if (done) {
done(Status::OK());
@@ -80,6 +87,8 @@ Status GcsPubSub::SubscribeInternal(const std::string &channel, const Callback &
&out_callback_index);
if (id) {
absl::MutexLock lock(&mutex_);
// If the same pattern has been subscribed more than once, the last subscription takes
// effect.
subscribe_callback_index_[pattern] = out_callback_index;
}
return status;
src/ray/gcs/pubsub/gcs_pub_sub.h
+1
View File
@@ -101,6 +101,7 @@ class GcsPubSub {
absl::Mutex mutex_;
std::unordered_map<std::string, int64_t> subscribe_callback_index_ GUARDED_BY(mutex_);
std::unordered_map<std::string, int64_t> unsubscribe_callback_index_ GUARDED_BY(mutex_);
};
} // namespace gcs