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 "single_ver_kv_syncer.h"
17 
18 #include <functional>
19 #include <map>
20 #include <mutex>
21 
22 #include "db_common.h"
23 #include "ikvdb_sync_interface.h"
24 #include "log_print.h"
25 #include "meta_data.h"
26 #include "single_ver_sync_engine.h"
27 #include "sqlite_single_ver_natural_store.h"
28 
29 namespace DistributedDB {
SingleVerKVSyncer()30 SingleVerKVSyncer::SingleVerKVSyncer()
31     : autoSyncEnable_(false), triggerSyncTask_(true)
32 {
33 }
34 
~SingleVerKVSyncer()35 SingleVerKVSyncer::~SingleVerKVSyncer()
36 {
37 }
38 
EnableAutoSync(bool enable)39 void SingleVerKVSyncer::EnableAutoSync(bool enable)
40 {
41     LOGI("[Syncer] EnableAutoSync enable = %d, Label=%s", enable, label_.c_str());
42     if (autoSyncEnable_ == enable) {
43         return;
44     }
45 
46     autoSyncEnable_ = enable;
47     if (!enable) {
48         return;
49     }
50 
51     if (!initialized_) {
52         LOGI("[Syncer] Syncer has not Init");
53         return;
54     }
55 
56     std::vector<std::string> devices;
57     GetOnlineDevices(devices);
58     if (devices.empty()) {
59         LOGI("[Syncer] EnableAutoSync no online devices");
60         return;
61     }
62     int errCode = Sync(devices, SyncModeType::AUTO_PUSH, nullptr, nullptr, false);
63     if (errCode != E_OK) {
64         LOGE("[Syncer] sync start by EnableAutoSync failed err %d", errCode);
65     }
66 }
67 
68 // Local data changed callback
LocalDataChanged(int notifyEvent)69 void SingleVerKVSyncer::LocalDataChanged(int notifyEvent)
70 {
71     if (!initialized_) {
72         LOGE("[Syncer] Syncer has not Init");
73         return;
74     }
75 
76     if (notifyEvent != static_cast<int>(SQLiteGeneralNSNotificationEventType::SQLITE_GENERAL_FINISH_MIGRATE_EVENT) &&
77         notifyEvent != static_cast<int>(SQLiteGeneralNSNotificationEventType::SQLITE_GENERAL_NS_PUT_EVENT)) {
78         LOGD("[Syncer] ignore event:%d", notifyEvent);
79         return;
80     }
81     if (!triggerSyncTask_) {
82         LOGI("[Syncer] some sync task is scheduling");
83         return;
84     }
85     triggerSyncTask_ = false;
86     std::vector<std::string> devices;
87     GetOnlineDevices(devices);
88     if (devices.empty()) {
89         LOGI("[Syncer] LocalDataChanged no online standard devices, Label=%s", label_.c_str());
90         triggerSyncTask_ = true;
91         return;
92     }
93     ISyncEngine *engine = syncEngine_;
94     ISyncInterface *storage = syncInterface_;
95     RefObject::IncObjRef(engine);
96     storage->IncRefCount();
97     // To avoid many task were produced and waiting in the queue. For example, put value in a loop.
98     // It will consume thread pool resources, so other task will delay until these task finish.
99     // In extreme situation, 10 thread run the localDataChanged task and 1 task waiting in queue.
100     int errCode = RuntimeContext::GetInstance()->ScheduleTask([this, devices, engine, storage] {
101         triggerSyncTask_ = true;
102         if (!TryFullSync(devices)) {
103             TriggerSubQuerySync(devices);
104         }
105         RefObject::DecObjRef(engine);
106         storage->DecRefCount();
107     });
108     // if task schedule failed, but triggerSyncTask_ is not set to true, other thread may skip the schedule time
109     // when task schedule failed, it means unormal status, it is unable to schedule next time probably
110     // so it is ok if other thread skip the schedule if last task schedule failed
111     if (errCode != E_OK) {
112         triggerSyncTask_ = true;
113         LOGE("[TriggerSync] LocalDataChanged retCode:%d", errCode);
114         RefObject::DecObjRef(engine);
115         storage->DecRefCount();
116     }
117 }
118 
119 // remote device online callback
RemoteDataChanged(const std::string & device)120 void SingleVerKVSyncer::RemoteDataChanged(const std::string &device)
121 {
122     LOGI("[SingleVerKVSyncer] device online dev %s", STR_MASK(device));
123     if (!initialized_) {
124         LOGE("[Syncer] Syncer has not Init");
125         return;
126     }
127     std::string userId = syncInterface_->GetDbProperties().GetStringProp(KvDBProperties::USER_ID, "");
128     std::string appId = syncInterface_->GetDbProperties().GetStringProp(KvDBProperties::APP_ID, "");
129     std::string storeId = syncInterface_->GetDbProperties().GetStringProp(KvDBProperties::STORE_ID, "");
130     RuntimeContext::GetInstance()->NotifyDatabaseStatusChange(userId, appId, storeId, device, true);
131     SingleVerSyncer::RemoteDataChanged(device);
132     if (autoSyncEnable_) {
133         RefObject::IncObjRef(syncEngine_);
134         syncInterface_->IncRefCount();
135         int retCode = RuntimeContext::GetInstance()->ScheduleTask([this, userId, appId, storeId, device] {
136             std::vector<std::string> devices;
137             devices.push_back(device);
138             int errCode = E_OK;
139             if (RuntimeContext::GetInstance()->IsNeedAutoSync(userId, appId, storeId, device)) {
140                 errCode = Sync(devices, SyncModeType::AUTO_PUSH, nullptr, nullptr, false);
141             }
142             if (errCode != E_OK) {
143                 LOGE("[SingleVerKVSyncer] sync start by RemoteDataChanged failed err %d", errCode);
144             }
145             RefObject::DecObjRef(syncEngine_);
146             syncInterface_->DecRefCount();
147         });
148         if (retCode != E_OK) {
149             LOGE("[AutoLaunch] RemoteDataChanged triggler sync retCode:%d", retCode);
150             RefObject::DecObjRef(syncEngine_);
151             syncInterface_->DecRefCount();
152         }
153     }
154     // db online again ,trigger subscribe
155     // if remote device online, subscribequery num is 0
156     std::vector<QuerySyncObject> syncQueries;
157     static_cast<SingleVerSyncEngine *>(syncEngine_)->GetLocalSubscribeQueries(device, syncQueries);
158     if (syncQueries.empty()) {
159         LOGI("no need to trigger auto subscribe");
160         return;
161     }
162     LOGI("[SingleVerKVSyncer] trigger local subscribe sync, queryNums=%zu", syncQueries.size());
163     for (const auto &query : syncQueries) {
164         TriggerSubscribe(device, query);
165     }
166     static_cast<SingleVerSyncEngine *>(syncEngine_)->PutUnfinishedSubQueries(device, syncQueries);
167 }
168 
SyncConditionCheck(const SyncParma & param,const ISyncEngine * engine,ISyncInterface * storage) const169 int SingleVerKVSyncer::SyncConditionCheck(const SyncParma &param, const ISyncEngine *engine,
170     ISyncInterface *storage) const
171 {
172     if (!param.isQuerySync) {
173         return E_OK;
174     }
175     QuerySyncObject query = param.syncQuery;
176     int errCode = static_cast<SingleVerKvDBSyncInterface *>(storage)->CheckAndInitQueryCondition(query);
177     if (errCode != E_OK) {
178         LOGE("[SingleVerKVSyncer] QuerySyncObject check failed");
179         return errCode;
180     }
181     if (param.mode != SUBSCRIBE_QUERY) {
182         return E_OK;
183     }
184     if (query.HasLimit() || query.HasOrderBy()) {
185         LOGE("[SingleVerKVSyncer] subscribe query not support limit,offset or orderby");
186         return -E_NOT_SUPPORT;
187     }
188     if (param.devices.size() > MAX_DEVICES_NUM) {
189         LOGE("[SingleVerKVSyncer] devices is overlimit");
190         return -E_MAX_LIMITS;
191     }
192     return engine->SubscribeLimitCheck(param.devices, query);
193 }
194 
TriggerSubscribe(const std::string & device,const QuerySyncObject & query)195 void SingleVerKVSyncer::TriggerSubscribe(const std::string &device, const QuerySyncObject &query)
196 {
197     if (!initialized_) { // LCOV_EXCL_BR_LINE
198         LOGE("[Syncer] Syncer has not Init");
199         return;
200     }
201     RefObject::IncObjRef(syncEngine_);
202     int retCode = RuntimeContext::GetInstance()->ScheduleTask([this, device, query] {
203         std::vector<std::string> devices;
204         devices.push_back(device);
205         SyncParma param;
206         param.devices = devices;
207         param.mode = SyncModeType::AUTO_SUBSCRIBE_QUERY;
208         param.onComplete = nullptr;
209         param.onFinalize = nullptr;
210         param.wait = false;
211         param.isQuerySync = true;
212         param.syncQuery = query;
213         int errCode = Sync(param);
214         if (errCode != E_OK) { // LCOV_EXCL_BR_LINE
215             LOGE("[SingleVerKVSyncer] subscribe start by RemoteDataChanged failed err %d", errCode);
216         }
217         RefObject::DecObjRef(syncEngine_);
218     });
219     if (retCode != E_OK) { // LCOV_EXCL_BR_LINE
220         LOGE("[Syncer] triggler query subscribe start failed err %d", retCode);
221         RefObject::DecObjRef(syncEngine_);
222     }
223 }
224 
TryFullSync(const std::vector<std::string> & devices)225 bool SingleVerKVSyncer::TryFullSync(const std::vector<std::string> &devices)
226 {
227     if (!initialized_) {
228         LOGE("[Syncer] Syncer has not Init");
229         return true;
230     }
231     if (!autoSyncEnable_) {
232         LOGD("[Syncer] autoSync no enable");
233         return false;
234     }
235     int errCode = Sync(devices, SyncModeType::AUTO_PUSH, nullptr, nullptr, false);
236     if (errCode != E_OK) {
237         LOGE("[Syncer] sync start by RemoteDataChanged failed err %d", errCode);
238         return false;
239     }
240     return true;
241 }
242 
TriggerSubQuerySync(const std::vector<std::string> & devices)243 void SingleVerKVSyncer::TriggerSubQuerySync(const std::vector<std::string> &devices)
244 {
245     if (!initialized_) {
246         LOGE("[Syncer] Syncer has not Init");
247         return;
248     }
249     std::shared_ptr<Metadata> metadata = nullptr;
250     ISyncInterface *syncInterface = nullptr;
251     {
252         std::lock_guard<std::mutex> lock(syncerLock_);
253         if (metadata_ == nullptr || syncInterface_ == nullptr) {
254             return;
255         }
256         metadata = metadata_;
257         syncInterface = syncInterface_;
258         syncInterface->IncRefCount();
259     }
260     int errCode;
261     for (auto &device : devices) {
262         std::vector<QuerySyncObject> queries;
263         static_cast<SingleVerSyncEngine *>(syncEngine_)->GetRemoteSubscribeQueries(device, queries);
264         for (auto &query : queries) {
265             std::string queryId = query.GetIdentify();
266             WaterMark queryWaterMark = 0;
267             uint64_t lastTimestamp = metadata->GetQueryLastTimestamp(device, queryId);
268             errCode = metadata->GetSendQueryWaterMark(queryId, device, queryWaterMark, false);
269             if (errCode != E_OK) {
270                 LOGE("[Syncer] get queryId=%s,dev=%s watermark failed", STR_MASK(queryId), STR_MASK(device));
271                 continue;
272             }
273             if (lastTimestamp < queryWaterMark || lastTimestamp == 0) {
274                 continue;
275             }
276             LOGD("[Syncer] lastTime=%" PRIu64 " vs WaterMark=%" PRIu64 ",trigger queryId=%s,dev=%s", lastTimestamp,
277                 queryWaterMark, STR_MASK(queryId), STR_MASK(device));
278             InternalSyncParma param;
279             std::vector<std::string> targetDevices;
280             targetDevices.push_back(device);
281             param.devices = targetDevices;
282             param.mode = SyncModeType::AUTO_PUSH;
283             param.isQuerySync = true;
284             param.syncQuery = query;
285             QueryAutoSync(param);
286         }
287     }
288     syncInterface->DecRefCount();
289 }
290 
DumpSyncerBasicInfo()291 SyncerBasicInfo SingleVerKVSyncer::DumpSyncerBasicInfo()
292 {
293     SyncerBasicInfo basicInfo = GenericSyncer::DumpSyncerBasicInfo();
294     basicInfo.isAutoSync = autoSyncEnable_;
295     return basicInfo;
296 }
297 
InitSyncEngine(DistributedDB::ISyncInterface * syncInterface)298 int SingleVerKVSyncer::InitSyncEngine(DistributedDB::ISyncInterface *syncInterface)
299 {
300     int errCode = GenericSyncer::InitSyncEngine(syncInterface);
301     if (errCode != E_OK) {
302         return errCode;
303     }
304     TriggerAddSubscribeAsync(syncInterface);
305     return E_OK;
306 }
307 
TriggerAddSubscribeAsync(ISyncInterface * syncInterface)308 void SingleVerKVSyncer::TriggerAddSubscribeAsync(ISyncInterface *syncInterface)
309 {
310     if (syncInterface == nullptr || syncEngine_ == nullptr) {
311         return;
312     }
313     if (syncInterface->GetInterfaceType() != ISyncInterface::SYNC_SVD) {
314         return;
315     }
316     DBInfo dbInfo;
317     auto storage = static_cast<SyncGenericInterface *>(syncInterface);
318     storage->GetDBInfo(dbInfo);
319     std::map<std::string, std::vector<QuerySyncObject>> subscribeQuery;
320     RuntimeContext::GetInstance()->GetSubscribeQuery(dbInfo, subscribeQuery);
321     if (subscribeQuery.empty()) {
322         LOGD("[SingleVerKVSyncer][TriggerAddSubscribeAsync] Subscribe cache is empty");
323         return;
324     }
325     storage->IncRefCount();
326     ISyncEngine *engine = syncEngine_;
327     RefObject::IncObjRef(engine);
328     int errCode = RuntimeContext::GetInstance()->ScheduleTask([this, engine, storage, subscribeQuery]() {
329         engine->AddSubscribe(storage, subscribeQuery);
330         // try to trigger query sync after add trigger
331         LocalDataChanged(static_cast<int>(SQLiteGeneralNSNotificationEventType::SQLITE_GENERAL_NS_PUT_EVENT));
332         RefObject::DecObjRef(engine);
333         storage->DecRefCount();
334     });
335     if (errCode != E_OK) {
336         LOGW("[SingleVerKVSyncer] TriggerAddSubscribeAsync failed errCode = %d", errCode);
337         syncInterface->DecRefCount();
338         RefObject::DecObjRef(engine);
339     }
340 }
341 } // namespace DistributedDB
342