1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "storage_engine_manager.h"
17 #include "log_print.h"
18 #include "db_errno.h"
19 #include "runtime_context.h"
20 #include "rd_single_ver_storage_engine.h"
21 #include "sqlite_single_ver_storage_engine.h"
22
23 namespace DistributedDB {
24 bool StorageEngineManager::isRegLockStatusListener_ = false;
25 std::mutex StorageEngineManager::instanceLock_;
26 std::atomic<StorageEngineManager *> StorageEngineManager::instance_{nullptr};
27 std::mutex StorageEngineManager::storageEnginesLock_;
28
29 namespace {
GetIdentifier(const KvDBProperties & property)30 std::string GetIdentifier(const KvDBProperties &property)
31 {
32 return property.GetStringProp(KvDBProperties::IDENTIFIER_DATA, "");
33 }
34
GetDatabaseType(const KvDBProperties & property)35 int GetDatabaseType(const KvDBProperties &property)
36 {
37 return property.GetIntProp(KvDBProperties::DATABASE_TYPE, KvDBProperties::LOCAL_TYPE_SQLITE);
38 }
39
IsSingleVerType(int databaseType)40 int IsSingleVerType(int databaseType)
41 {
42 return databaseType == KvDBProperties::SINGLE_VER_TYPE_SQLITE ||
43 databaseType == KvDBProperties::SINGLE_VER_TYPE_RD_KERNAL;
44 }
45 }
46
StorageEngineManager()47 StorageEngineManager::StorageEngineManager() : lockStatusListener_(nullptr)
48 {}
49
~StorageEngineManager()50 StorageEngineManager::~StorageEngineManager()
51 {
52 if (lockStatusListener_ != nullptr) {
53 lockStatusListener_->Drop(true);
54 }
55 }
56
GetStorageEngine(const KvDBProperties & property,int & errCode)57 StorageEngine *StorageEngineManager::GetStorageEngine(const KvDBProperties &property, int &errCode)
58 {
59 StorageEngineManager *manager = GetInstance();
60 if (manager == nullptr) {
61 LOGE("[StorageEngineManager] GetInstance failed");
62 errCode = -E_OUT_OF_MEMORY;
63 return nullptr;
64 }
65 std::string identifier = GetIdentifier(property);
66 manager->EnterGetEngineProcess(identifier);
67 auto storageEngine = manager->FindStorageEngine(identifier);
68 if (storageEngine == nullptr) {
69 storageEngine = manager->CreateStorageEngine(property, errCode);
70 if (errCode == E_OK) {
71 manager->InsertStorageEngine(identifier, storageEngine);
72 }
73 } else {
74 errCode = storageEngine->CheckEngineOption(property);
75 if (errCode != E_OK) {
76 LOGE("kvdb property mismatch engine option! errCode = [%d]", errCode);
77 storageEngine = nullptr;
78 }
79 }
80
81 manager->ExitGetEngineProcess(identifier);
82 return storageEngine;
83 }
84
ReleaseStorageEngine(StorageEngine * storageEngine)85 int StorageEngineManager::ReleaseStorageEngine(StorageEngine *storageEngine)
86 {
87 if (storageEngine == nullptr) {
88 LOGE("[StorageEngineManager] The engine to be released is nullptr");
89 return -E_INVALID_ARGS;
90 }
91
92 // Clear commit notify callback function.
93 storageEngine->SetNotifiedCallback(nullptr);
94
95 // If the cacheDB is valid, the storageEngine is not released to prevent the cache mechanism failed
96 bool isRelease = storageEngine->IsNeedTobeReleased();
97 if (!isRelease) {
98 LOGW("[StorageEngineManager] storageEngine do not need to be released.");
99 return E_OK;
100 }
101
102 StorageEngineManager *manager = GetInstance();
103 if (manager == nullptr) {
104 LOGE("[StorageEngineManager] Release GetInstance failed");
105 return -E_OUT_OF_MEMORY;
106 }
107
108 LOGD("[StorageEngineManager] storageEngine to be released.");
109 return manager->ReleaseEngine(storageEngine);
110 }
111
ForceReleaseStorageEngine(const std::string & identifier)112 int StorageEngineManager::ForceReleaseStorageEngine(const std::string &identifier)
113 {
114 StorageEngineManager *manager = GetInstance();
115 if (manager == nullptr) {
116 LOGE("[StorageEngineManager] Force release GetInstance failed");
117 return -E_OUT_OF_MEMORY;
118 }
119
120 LOGD("[StorageEngineManager] Force release engine.");
121 manager->ReleaseResources(identifier);
122 return E_OK;
123 }
124
ExecuteMigration(StorageEngine * storageEngine)125 int StorageEngineManager::ExecuteMigration(StorageEngine *storageEngine)
126 {
127 if (storageEngine == nullptr) {
128 LOGE("storage engine is nullptr can not execute migration!");
129 return -E_INVALID_ARGS;
130 }
131 if (storageEngine->IsExistConnection()) {
132 return storageEngine->ExecuteMigrate();
133 }
134 LOGI("connection is not existed, not need execute migration!");
135 return -E_INVALID_DB;
136 }
137
GetInstance()138 StorageEngineManager *StorageEngineManager::GetInstance()
139 {
140 // For Double-Checked Locking, we need check instance_ twice
141 if (instance_ == nullptr) {
142 std::lock_guard<std::mutex> lockGuard(instanceLock_);
143 if (instance_ == nullptr) {
144 instance_ = new (std::nothrow) StorageEngineManager();
145 if (instance_ == nullptr) {
146 LOGE("[StorageEngineManager] Failed to alloc the engine manager!");
147 return nullptr;
148 }
149 }
150 }
151
152 if (!isRegLockStatusListener_) {
153 int errCode = (instance_.load())->RegisterLockStatusListener();
154 if (errCode == E_OK) {
155 isRegLockStatusListener_ = true;
156 LOGW("[StorageEngineManager] Register lock status listener.");
157 }
158 }
159 return instance_;
160 }
161
RegisterLockStatusListener()162 int StorageEngineManager::RegisterLockStatusListener()
163 {
164 int errCode = E_OK;
165 lockStatusListener_ = RuntimeContext::GetInstance()->RegisterLockStatusLister(
166 [this](void *lockStatus) {
167 if (lockStatus == nullptr) {
168 return;
169 }
170 bool isLocked = *static_cast<bool *>(lockStatus);
171 LOGD("[StorageEngineManager] Lock status to %d", isLocked);
172 if (isLocked) {
173 return;
174 }
175 int taskErrCode = RuntimeContext::GetInstance()->ScheduleTask(
176 [this, isLocked] { LockStatusNotifier(isLocked); });
177 if (taskErrCode != E_OK) {
178 LOGE("[StorageEngineManager] LockStatusNotifier ScheduleTask failed : %d", taskErrCode);
179 }
180 }, errCode);
181 return errCode;
182 }
183
LockStatusNotifier(bool isAccessControlled)184 void StorageEngineManager::LockStatusNotifier(bool isAccessControlled)
185 {
186 (void)isAccessControlled;
187 std::lock_guard<std::mutex> lockGuard(storageEnginesLock_);
188 StorageEngine *storageEngine = nullptr;
189 for (const auto &item : storageEngines_) {
190 storageEngine = item.second;
191 LOGD("Begin to migrate for lock status change");
192 (void)ExecuteMigration(storageEngine);
193 }
194 }
195
CreateSingleVerStorageEngine(int databaseType)196 StorageEngine *CreateSingleVerStorageEngine(int databaseType)
197 {
198 if (databaseType == KvDBProperties::SINGLE_VER_TYPE_SQLITE) {
199 return new (std::nothrow) SQLiteSingleVerStorageEngine();
200 } else {
201 return new (std::nothrow) RdSingleVerStorageEngine();
202 }
203 return nullptr;
204 }
205
CreateStorageEngine(const KvDBProperties & property,int & errCode)206 StorageEngine *StorageEngineManager::CreateStorageEngine(const KvDBProperties &property, int &errCode)
207 {
208 int databaseType = GetDatabaseType(property);
209 if (!IsSingleVerType(databaseType)) {
210 LOGE("[StorageEngineManager] Database type error : %d", databaseType);
211 errCode = -E_NOT_SUPPORT;
212 return nullptr;
213 }
214
215 auto storageEngine = CreateSingleVerStorageEngine(databaseType);
216 if (storageEngine == nullptr) {
217 LOGE("[StorageEngineManager] Create storage engine failed");
218 errCode = -E_OUT_OF_MEMORY;
219 return nullptr;
220 }
221 errCode = E_OK;
222 return storageEngine;
223 }
224
FindStorageEngine(const std::string & identifier)225 StorageEngine *StorageEngineManager::FindStorageEngine(const std::string &identifier)
226 {
227 std::lock_guard<std::mutex> lockGuard(storageEnginesLock_);
228 auto iter = storageEngines_.find(identifier);
229 if (iter != storageEngines_.end()) {
230 auto storageEngine = iter->second;
231 if (storageEngine == nullptr) {
232 LOGE("[StorageEngineManager] storageEngine in cache is nullptr");
233 storageEngines_.erase(identifier);
234 return nullptr;
235 }
236
237 return storageEngine;
238 }
239
240 return nullptr;
241 }
242
InsertStorageEngine(const std::string & identifier,StorageEngine * & storageEngine)243 void StorageEngineManager::InsertStorageEngine(const std::string &identifier, StorageEngine *&storageEngine)
244 {
245 std::lock_guard<std::mutex> lockGuard(storageEnginesLock_);
246 storageEngines_.insert(std::pair<std::string, StorageEngine *>(identifier, storageEngine));
247 }
248
EraseStorageEngine(const std::string & identifier)249 void StorageEngineManager::EraseStorageEngine(const std::string &identifier)
250 {
251 std::lock_guard<std::mutex> lockGuard(storageEnginesLock_);
252 storageEngines_.erase(identifier);
253 }
254
ReleaseResources(const std::string & identifier)255 void StorageEngineManager::ReleaseResources(const std::string &identifier)
256 {
257 StorageEngine *storageEngine = nullptr;
258
259 {
260 std::lock_guard<std::mutex> lockGuard(storageEnginesLock_);
261 auto iter = storageEngines_.find(identifier);
262 if (iter != storageEngines_.end()) {
263 storageEngine = iter->second;
264 storageEngines_.erase(identifier);
265 }
266 }
267
268 if (storageEngine != nullptr) {
269 LOGI("[StorageEngineManager] Release storage engine");
270 RefObject::KillAndDecObjRef(storageEngine);
271 }
272 }
273
ReleaseEngine(StorageEngine * releaseEngine)274 int StorageEngineManager::ReleaseEngine(StorageEngine *releaseEngine)
275 {
276 const std::string identifier = releaseEngine->GetIdentifier();
277 StorageEngine *cacheEngine = nullptr;
278
279 {
280 std::lock_guard<std::mutex> lockGuard(storageEnginesLock_);
281 auto iter = storageEngines_.find(identifier);
282 if (iter != storageEngines_.end()) {
283 cacheEngine = iter->second;
284 storageEngines_.erase(identifier);
285 }
286 }
287
288 if (cacheEngine == nullptr) {
289 LOGE("[StorageEngineManager] cache engine is null");
290 return -E_ALREADY_RELEASE;
291 }
292 if (cacheEngine != releaseEngine) {
293 LOGE("[StorageEngineManager] cache engine is not equal the input engine");
294 return -E_INVALID_ARGS;
295 }
296
297 RefObject::KillAndDecObjRef(releaseEngine);
298 return E_OK;
299 }
300
EnterGetEngineProcess(const std::string & identifier)301 void StorageEngineManager::EnterGetEngineProcess(const std::string &identifier)
302 {
303 std::unique_lock<std::mutex> lock(getEngineMutex_);
304 getEngineCondition_.wait(lock, [this, &identifier]() {
305 return this->getEngineSet_.count(identifier) == 0;
306 });
307 (void)getEngineSet_.insert(identifier);
308 }
309
ExitGetEngineProcess(const std::string & identifier)310 void StorageEngineManager::ExitGetEngineProcess(const std::string &identifier)
311 {
312 std::unique_lock<std::mutex> lock(getEngineMutex_);
313 (void)getEngineSet_.erase(identifier);
314 getEngineCondition_.notify_all();
315 }
316 } // namespace DistributedDB
317