fix free service after service instance data expired

Author: lambdaliu

polaris/plugin/local_registry/local_registry.cpp

@@ -138,20 +138,38 @@ void InMemoryRegistry::RunGcTask() {
   service_circuit_breaker_config_data_.CheckGc(min_gc_time);
 }
 
-Service* InMemoryRegistry::GetOrCreateServiceInLock(const ServiceKey& service_key) {
+Service* InMemoryRegistry::CreateServiceInLock(const ServiceKey& service_key) {
   Service* service = NULL;
   pthread_rwlock_wrlock(&rwlock_);
   std::map<ServiceKey, Service*>::iterator service_it = service_cache_.find(service_key);
-  if (service_it == service_cache_.end()) {
-    service                     = new Service(service_key, ++next_service_id_);
-    service_cache_[service_key] = service;
-  } else {
+  POLARIS_ASSERT(service_it == service_cache_.end())
+  service                     = new Service(service_key, ++next_service_id_);
+  service_cache_[service_key] = service;
+  pthread_rwlock_unlock(&rwlock_);
+  return service;
+}
+
+Service* InMemoryRegistry::GetServiceInLock(const ServiceKey& service_key) {
+  Service* service = NULL;
+  pthread_rwlock_wrlock(&rwlock_);
+  std::map<ServiceKey, Service*>::iterator service_it = service_cache_.find(service_key);
+  if (service_it != service_cache_.end()) {
     service = service_it->second;
   }
   pthread_rwlock_unlock(&rwlock_);
   return service;
 }
 
+void InMemoryRegistry::DeleteServiceInLock(const ServiceKey& service_key) {
+  pthread_rwlock_wrlock(&rwlock_);
+  std::map<ServiceKey, Service*>::iterator service_it = service_cache_.find(service_key);
+  if (service_it != service_cache_.end()) {
+    delete service_it->second;
+    service_cache_.erase(service_it);
+  }
+  pthread_rwlock_unlock(&rwlock_);
+}
+
 void InMemoryRegistry::CheckExpireServiceData(uint64_t min_access_time,
                                               RcuMap<ServiceKey, ServiceData>& rcu_cache,
                                               ServiceDataType service_data_type) {
@@ -166,17 +184,16 @@ void InMemoryRegistry::CheckExpireServiceData(uint64_t min_access_time,
     if (service_data_notify_map_.erase(service_key_with_type) > 0) {  // 有通知对象表示注册过handler
       context_->GetServerConnector()->DeregisterEventHandler(expired_services[i],
                                                              service_data_type);
-    } else {  // 没有通知对象，表示未注册过handler，从磁盘加载后从未访问过的数据，直接删除数据
-      rcu_cache.Delete(expired_services[i]);
-      context_impl->GetServiceRecord()->ServiceDataDelete(expired_services[i], service_data_type);
-      context_impl->GetCacheManager()->GetCachePersist().PersistServiceData(expired_services[i],
-                                                                            service_data_type, "");
     }
-    pthread_rwlock_unlock(&notify_rwlock_);
     if (service_data_type == kServiceDataInstances) {  // 清除实例数据时对应的服务级别插件也删除
       context_impl->DeleteServiceContext(expired_services[i]);
       DeleteServiceInLock(expired_services[i]);
     }
+    rcu_cache.Delete(expired_services[i]);
+    context_impl->GetServiceRecord()->ServiceDataDelete(expired_services[i], service_data_type);
+    context_impl->GetCacheManager()->GetCachePersist().PersistServiceData(expired_services[i],
+                                                                          service_data_type, "");
+    pthread_rwlock_unlock(&notify_rwlock_);
   }
 }
 
@@ -249,6 +266,9 @@ ReturnCode InMemoryRegistry::LoadServiceDataWithNotify(const ServiceKey& service
     if (interval_it != service_interval_map_.end()) {
       refresh_interval = interval_it->second;
     }
+    if (data_type == kServiceDataInstances) {
+      CreateServiceInLock(service_key);
+    }
     // 先加载磁盘缓存数据
     CachePersist& cache_persist = context_->GetContextImpl()->GetCacheManager()->GetCachePersist();
     ServiceData* disk_service_data = cache_persist.LoadServiceData(service_key, data_type);
@@ -269,25 +289,21 @@ ReturnCode InMemoryRegistry::LoadServiceDataWithNotify(const ServiceKey& service
   return kReturnOk;
 }
 
-void InMemoryRegistry::DeleteServiceInLock(const ServiceKey& service_key) {
-  pthread_rwlock_wrlock(&rwlock_);
-  std::map<ServiceKey, Service*>::iterator service_it = service_cache_.find(service_key);
-  if (service_it != service_cache_.end()) {
-    delete service_it->second;
-    service_cache_.erase(service_it);
-  }
-  pthread_rwlock_unlock(&rwlock_);
-}
-
 ReturnCode InMemoryRegistry::UpdateServiceData(const ServiceKey& service_key,
                                                ServiceDataType data_type,
                                                ServiceData* service_data) {
-  if (service_data != NULL) {  // 更新服务数据指向服务
-    Service* service = GetOrCreateServiceInLock(service_key);
+  Service* service = GetServiceInLock(service_key);
+  if (service != NULL) {  // 更新服务数据指向服务
     service->UpdateData(service_data);
   }
   ContextImpl* context_impl = context_->GetContextImpl();
   if (data_type == kServiceDataInstances) {
+    if (service == NULL) {  // 服务被反注册了
+      if (service_data != NULL) {
+        service_data->DecrementRef();
+      }
+      return kReturnOk;
+    }
     ServiceData* old_service_data = service_instances_data_.Get(service_key);
     if (old_service_data != NULL) {
       PluginManager::Instance().OnPreUpdateServiceData(old_service_data, service_data);
@@ -308,14 +324,6 @@ ReturnCode InMemoryRegistry::UpdateServiceData(const ServiceKey& service_key,
     POLARIS_ASSERT(false);
   }
   if (service_data == NULL) {  // Server Connector反注册Handler触发更新为NULL
-    if (data_type == kServiceDataInstances) {  // 删除服务实例数据时，同时删除服务
-      context_impl->DeleteServiceContext(service_key);
-      DeleteServiceInLock(service_key);
-    }
-    context_impl->GetServiceRecord()->ServiceDataDelete(service_key,
-                                                        data_type);  // 同步记录Service数据删除
-    context_impl->GetCacheManager()->GetCachePersist().PersistServiceData(
-        service_key, data_type, "");  // 异步删除磁盘服务数据
     return kReturnOk;
   }
   context_impl->GetServiceRecord()->ServiceDataUpdate(service_data);  // 同步记录Service版本变化

polaris/plugin/local_registry/local_registry.h

@@ -109,7 +109,9 @@ class InMemoryRegistry : public LocalRegistry {
   ServiceDataNotify* GetOrCreateDataNotify(const ServiceKey& service_key, ServiceDataType data_type,
                                            bool& new_create);
 
-  Service* GetOrCreateServiceInLock(const ServiceKey& service_key);
+  Service* CreateServiceInLock(const ServiceKey& service_key);
+
+  Service* GetServiceInLock(const ServiceKey& service_key);
 
   void DeleteServiceInLock(const ServiceKey& service_key);
 

test/integration/set_circuit_breaker/set_circuit_breaker_test.cpp

@@ -68,6 +68,7 @@ class SetCircuitBreakerTest : public IntegrationBase {
     ASSERT_TRUE(consumer_ != NULL);
 
     SetUpServiceData();
+    TryDoRoute();
   }
 
   virtual void TearDown() {
@@ -358,6 +359,17 @@ class SetCircuitBreakerTest : public IntegrationBase {
         << total << " " << percent << " " << err_rate << " " << count;
   }
 
+  void TryDoRoute() {
+    polaris::GetOneInstanceRequest req(service_key_);
+    polaris::ServiceInfo service_info;
+    service_info.service_key_.name_      = "test2";
+    service_info.service_key_.namespace_ = "Test";
+    service_info.metadata_["f"]          = "fv1";
+    req.SetSourceService(service_info);
+    polaris::Instance instance;
+    ASSERT_EQ(consumer_->GetOneInstance(req, instance), kReturnOk);
+  }
+
 protected:
   polaris::ServiceKey service_key_;
   polaris::ConsumerApi* consumer_;
@@ -391,19 +403,6 @@ class SetCircuitBreakerTest : public IntegrationBase {
   std::string ins3_id_;
 };
 
-TEST_F(SetCircuitBreakerTest, TestRoute) {
-  ReturnCode ret;
-  polaris::GetOneInstanceRequest req(service_key_);
-  polaris::ServiceInfo service_info;
-  service_info.service_key_.name_      = "test2";
-  service_info.service_key_.namespace_ = "Test";
-  service_info.metadata_["f"]          = "fv1";
-  req.SetSourceService(service_info);
-  polaris::Instance instance;
-  ret = consumer_->GetOneInstance(req, instance);
-  ASSERT_EQ(ret, kReturnOk);
-}
-
 TEST_F(SetCircuitBreakerTest, ErrRateOpen) {
   polaris::ServiceCallResult err_result;
   err_result.SetServiceNamespace(service_key_.namespace_);

test/plugin/load_balancer/ring_hash_test.cpp

@@ -97,6 +97,10 @@ TEST_F(RingHashCstLbTest, TestSelectWithInstancesUpdate) {
   LocalRegistry *local_registry = context_->GetLocalRegistry();
   local_registry->GetServiceDataWithRef(service_key_, kServiceDataInstances, service_data);
   ASSERT_TRUE(service_data == NULL);
+  ServiceDataNotify *notify = NULL;
+  local_registry->LoadServiceDataWithNotify(service_key_, kServiceDataInstances, service_data,
+                                            notify);
+  ASSERT_TRUE(notify != NULL);
   v1::DiscoverResponse response;
   CreateInstancesResponse(response);
 

test/plugin/local_registry/local_registry_test.cpp

@@ -203,7 +203,7 @@ TEST_F(InMemoryLocalRegistryTest, TestUpdateServiceData) {
   std::set<ServiceKey> service_key_set;
   ret = local_registry_->GetAllServiceKey(service_key_set);
   ASSERT_EQ(ret, kReturnOk);
-  ASSERT_EQ(service_key_set.size(), 1);
+  ASSERT_EQ(service_key_set.size(), 0);
 
   delete mock_server_connector_->saved_handler_;
   mock_server_connector_->saved_handler_ = NULL;