fix(backup): use existing block_service_manager when add backup policy

src/block_service/block_service_manager.cpp

@@ -33,55 +33,51 @@ block_service_manager::block_service_manager()
 {
 }
 
-block_filesystem *block_service_manager::get_block_filesystem(const std::string &provider)
+block_service_manager::~block_service_manager()
 {
-    {
-        zauto_read_lock l(_fs_lock);
-        auto iter = _fs_map.find(provider);
-        if (iter != _fs_map.end())
-            return iter->second.get();
+    ddebug("close block service manager.");
+    zauto_write_lock l(_fs_lock);
+    _fs_map.clear();
+}
+
+block_filesystem *block_service_manager::get_or_create_block_filesystem(const std::string &provider)
+{
+    zauto_write_lock l(_fs_lock);
+    auto iter = _fs_map.find(provider);
+    if (iter != _fs_map.end()) {
+        return iter->second.get();
     }
 
-    {
-        zauto_write_lock l(_fs_lock);
-        auto iter = _fs_map.find(provider);
-        if (iter != _fs_map.end())
-            return iter->second.get();
-
-        const char *provider_type = dsn_config_get_value_string(
-            (std::string("block_service.") + provider).c_str(), "type", "", "block service type");
-
-        block_filesystem *fs =
-            utils::factory_store<block_filesystem>::create(provider_type, PROVIDER_TYPE_MAIN);
-        if (fs == nullptr) {
-            derror("acquire block filesystem failed, provider = %s, provider_type = %s",
-                   provider.c_str(),
-                   provider_type);
-            return nullptr;
-        }
+    const char *provider_type = dsn_config_get_value_string(
+        (std::string("block_service.") + provider).c_str(), "type", "", "block service type");
 
-        const char *arguments =
-            dsn_config_get_value_string((std::string("block_service.") + provider).c_str(),
-                                        "args",
-                                        "",
-                                        "args for block_service");
-
-        std::vector<std::string> args;
-        utils::split_args(arguments, args);
-        dsn::error_code err = fs->initialize(args);
-
-        if (dsn::ERR_OK == err) {
-            ddebug("create block filesystem ok for provider(%s)", provider.c_str());
-            _fs_map.emplace(provider, std::unique_ptr<block_filesystem>(fs));
-            return fs;
-        } else {
-            derror("create block file system err(%s) for provider(%s)",
-                   err.to_string(),
-                   provider.c_str());
-            delete fs;
-            return nullptr;
-        }
+    block_filesystem *fs =
+        utils::factory_store<block_filesystem>::create(provider_type, PROVIDER_TYPE_MAIN);
+    if (fs == nullptr) {
+        derror_f("acquire block filesystem failed, provider = {}, provider_type = {}",
+                 provider,
+                 std::string(provider_type));
+        return nullptr;
+    }
+
+    const char *arguments = dsn_config_get_value_string(
+        (std::string("block_service.") + provider).c_str(), "args", "", "args for block_service");
+
+    std::vector<std::string> args;
+    utils::split_args(arguments, args);
+    dsn::error_code err = fs->initialize(args);
+
+    if (dsn::ERR_OK == err) {
+        ddebug_f("create block filesystem ok for provider {}", provider);
+        _fs_map.emplace(provider, std::unique_ptr<block_filesystem>(fs));
+    } else {
+        derror_f("create block file system err {} for provider {}",
+                 std::string(err.to_string()),
+                 provider);
+        delete fs;
+        fs = nullptr;
     }
+    return fs;
 }
 
 static create_file_response create_block_file_sync(const std::string &remote_file_path,

src/block_service/block_service_manager.h

@@ -26,7 +26,8 @@ class block_service_manager
 {
 public:
     block_service_manager();
-    block_filesystem *get_block_filesystem(const std::string &provider);
+    ~block_service_manager();
+    block_filesystem *get_or_create_block_filesystem(const std::string &provider);
 
     // download files from remote file system
     // \return  ERR_FILE_OPERATION_FAILED: local file system error

src/meta/meta_backup_service.cpp

@@ -865,32 +865,16 @@ bool policy_context::is_under_backuping()
     return false;
 }
 
-void policy_context::set_policy(policy &&p)
-{
-    zauto_lock l(_lock);
-
-    const std::string old_backup_provider_type = _policy.backup_provider_type;
-    _policy = std::move(p);
-    if (_policy.backup_provider_type != old_backup_provider_type) {
-        _block_service =
-            _backup_service->get_meta_service()->get_block_service_manager().get_block_filesystem(
-                _policy.backup_provider_type);
-    }
-    dassert(_block_service,
-            "can't initialize block filesystem by provider (%s)",
-            _policy.backup_provider_type.c_str());
-}
-
 void policy_context::set_policy(const policy &p)
 {
     zauto_lock l(_lock);
 
     const std::string old_backup_provider_type = _policy.backup_provider_type;
     _policy = p;
     if (_policy.backup_provider_type != old_backup_provider_type) {
-        _block_service =
-            _backup_service->get_meta_service()->get_block_service_manager().get_block_filesystem(
-                _policy.backup_provider_type);
+        _block_service = _backup_service->get_meta_service()
+                             ->get_block_service_manager()
+                             .get_or_create_block_filesystem(_policy.backup_provider_type);
     }
     dassert(_block_service,
             "can't initialize block filesystem by provider (%s)",
@@ -1279,14 +1263,11 @@ void backup_service::add_backup_policy(dsn::message_ex *msg)
             }
         }
 
-        {
-            dist::block_service::block_service_manager _block_service_manager;
-            if (_block_service_manager.get_block_filesystem(request.backup_provider_type) ==
-                nullptr) {
-                derror("invalid backup_provider_type(%s)", request.backup_provider_type.c_str());
-                response.err = ERR_INVALID_PARAMETERS;
-                should_create_new_policy = false;
-            }
+        if (_meta_svc->get_block_service_manager().get_or_create_block_filesystem(
+                request.backup_provider_type) == nullptr) {
+            derror("invalid backup_provider_type(%s)", request.backup_provider_type.c_str());
+            response.err = ERR_INVALID_PARAMETERS;
+            should_create_new_policy = false;
         }
 
         if (should_create_new_policy) {

src/meta/meta_backup_service.h

@@ -230,7 +230,6 @@ class policy_context
     }
     mock_virtual ~policy_context() {}
 
-    void set_policy(policy &&p);
     void set_policy(const policy &p);
     policy get_policy();
     void add_backup_history(const backup_info &info);

src/meta/meta_bulk_load_service.cpp

@@ -95,7 +95,7 @@ error_code bulk_load_service::check_bulk_load_request_params(const std::string &
 
     // check file provider
     dsn::dist::block_service::block_filesystem *blk_fs =
-        _meta_svc->get_block_service_manager().get_block_filesystem(file_provider);
+        _meta_svc->get_block_service_manager().get_or_create_block_filesystem(file_provider);
     if (blk_fs == nullptr) {
         derror_f("invalid remote file provider type: {}", file_provider);
         hint_msg = "invalid file_provider";

src/meta/server_state_restore.cpp

@@ -47,7 +47,8 @@ void server_state::sync_app_from_backup_media(
             LPC_RESTORE_BACKGROUND, std::move(callback), 0));
 
     block_filesystem *blk_fs =
-        _meta_svc->get_block_service_manager().get_block_filesystem(request.backup_provider_name);
+        _meta_svc->get_block_service_manager().get_or_create_block_filesystem(
+            request.backup_provider_name);
     if (blk_fs == nullptr) {
         derror("acquire block_filesystem(%s) failed", request.backup_provider_name.c_str());
         callback_tsk->enqueue_with(ERR_INVALID_PARAMETERS, dsn::blob());

src/replica/bulk_load/replica_bulk_loader.cpp

@@ -382,7 +382,7 @@ error_code replica_bulk_loader::download_sst_files(const std::string &app_name,
     const std::string remote_dir =
         get_remote_bulk_load_dir(app_name, cluster_name, get_gpid().get_partition_index());
     dist::block_service::block_filesystem *fs =
-        _stub->_block_service_manager.get_block_filesystem(provider_name);
+        _stub->_block_service_manager.get_or_create_block_filesystem(provider_name);
 
     // download metadata file synchronously
     uint64_t file_size = 0;

src/replica/replica_backup.cpp

@@ -43,7 +43,7 @@ void replica::on_cold_backup(const backup_request &request, /*out*/ backup_respo
         } else {
             /// TODO: policy may change provider
             dist::block_service::block_filesystem *block_service =
-                _stub->_block_service_manager.get_block_filesystem(
+                _stub->_block_service_manager.get_or_create_block_filesystem(
                     request.policy.backup_provider_type);
             if (block_service == nullptr) {
                 derror("%s: create cold backup block service failed, provider_type = %s, response "

src/replica/replica_restore.cpp

@@ -99,7 +99,7 @@ error_code replica::download_checkpoint(const configuration_restore_request &req
                                         const std::string &local_chkpt_dir)
 {
     block_filesystem *fs =
-        _stub->_block_service_manager.get_block_filesystem(req.backup_provider_name);
+        _stub->_block_service_manager.get_or_create_block_filesystem(req.backup_provider_name);
 
     // download metadata file and parse it into cold_backup_meta
     cold_backup_metadata backup_metadata;
@@ -230,7 +230,7 @@ dsn::error_code replica::find_valid_checkpoint(const configuration_restore_reque
     std::string manifest_file = cold_backup::get_current_chkpt_file(
         backup_root, policy_name, req.app_name, old_gpid, backup_id);
     block_filesystem *fs =
-        _stub->_block_service_manager.get_block_filesystem(req.backup_provider_name);
+        _stub->_block_service_manager.get_or_create_block_filesystem(req.backup_provider_name);
     dassert(fs,
             "%s: get block filesystem by provider(%s) failed",
             name(),