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 "sync_engine.h"
17 
18 #include <algorithm>
19 #include <deque>
20 #include <functional>
21 
22 #include "ability_sync.h"
23 #include "db_common.h"
24 #include "db_dump_helper.h"
25 #include "db_errno.h"
26 #include "device_manager.h"
27 #include "hash.h"
28 #include "isync_state_machine.h"
29 #include "log_print.h"
30 #include "runtime_context.h"
31 #include "single_ver_serialize_manager.h"
32 #include "subscribe_manager.h"
33 #include "time_sync.h"
34 
35 #ifndef OMIT_MULTI_VER
36 #include "commit_history_sync.h"
37 #include "multi_ver_data_sync.h"
38 #include "value_slice_sync.h"
39 #endif
40 
41 namespace DistributedDB {
42 int SyncEngine::queueCacheSize_ = 0;
43 int SyncEngine::maxQueueCacheSize_ = DEFAULT_CACHE_SIZE;
44 unsigned int SyncEngine::discardMsgNum_ = 0;
45 std::mutex SyncEngine::queueLock_;
46 
SyncEngine()47 SyncEngine::SyncEngine()
48     : syncInterface_(nullptr),
49       communicator_(nullptr),
50       deviceManager_(nullptr),
51       metadata_(nullptr),
52       execTaskCount_(0),
53       isSyncRetry_(false),
54       communicatorProxy_(nullptr),
55       isActive_(false),
56       remoteExecutor_(nullptr)
57 {
58 }
59 
~SyncEngine()60 SyncEngine::~SyncEngine()
61 {
62     LOGD("[SyncEngine] ~SyncEngine!");
63     ClearInnerResource();
64     equalIdentifierMap_.clear();
65     subManager_ = nullptr;
66     LOGD("[SyncEngine] ~SyncEngine ok!");
67 }
68 
Initialize(ISyncInterface * syncInterface,const std::shared_ptr<Metadata> & metadata,const InitCallbackParam & callbackParam)69 int SyncEngine::Initialize(ISyncInterface *syncInterface, const std::shared_ptr<Metadata> &metadata,
70     const InitCallbackParam &callbackParam)
71 {
72     if ((syncInterface == nullptr) || (metadata == nullptr)) {
73         return -E_INVALID_ARGS;
74     }
75     int errCode = StartAutoSubscribeTimer(*syncInterface);
76     if (errCode != E_OK) {
77         return errCode;
78     }
79 
80     errCode = InitComunicator(syncInterface);
81     if (errCode != E_OK) {
82         LOGE("[SyncEngine] Init Communicator failed");
83         // There need to set nullptr. other wise, syncInterface will be
84         // DecRef in th destroy-method.
85         StopAutoSubscribeTimer();
86         return errCode;
87     }
88     onRemoteDataChanged_ = callbackParam.onRemoteDataChanged;
89     offlineChanged_ = callbackParam.offlineChanged;
90     queryAutoSyncCallback_ = callbackParam.queryAutoSyncCallback;
91     errCode = InitInnerSource(callbackParam.onRemoteDataChanged, callbackParam.offlineChanged, syncInterface);
92     if (errCode != E_OK) {
93         // reset ptr if initialize device manager failed
94         StopAutoSubscribeTimer();
95         return errCode;
96     }
97     SetSyncInterface(syncInterface);
98     if (subManager_ == nullptr) {
99         subManager_ = std::make_shared<SubscribeManager>();
100     }
101     metadata_ = metadata;
102     isActive_ = true;
103     LOGI("[SyncEngine] Engine [%s] init ok", label_.c_str());
104     return E_OK;
105 }
106 
Close()107 int SyncEngine::Close()
108 {
109     LOGI("[SyncEngine] SyncEngine [%s] close enter!", label_.c_str());
110     isActive_ = false;
111     UnRegCommunicatorsCallback();
112     StopAutoSubscribeTimer();
113     std::vector<ISyncTaskContext *> decContext;
114     // Clear SyncContexts
115     {
116         std::unique_lock<std::mutex> lock(contextMapLock_);
117         for (auto &iter : syncTaskContextMap_) {
118             decContext.push_back(iter.second);
119             iter.second = nullptr;
120         }
121         syncTaskContextMap_.clear();
122     }
123     for (auto &iter : decContext) {
124         RefObject::KillAndDecObjRef(iter);
125         iter = nullptr;
126     }
127     WaitingExecTaskExist();
128     ReleaseCommunicators();
129     {
130         std::lock_guard<std::mutex> msgLock(queueLock_);
131         while (!msgQueue_.empty()) {
132             Message *inMsg = msgQueue_.front();
133             msgQueue_.pop_front();
134             if (inMsg != nullptr) { // LCOV_EXCL_BR_LINE
135                 queueCacheSize_ -= GetMsgSize(inMsg);
136                 delete inMsg;
137                 inMsg = nullptr;
138             }
139         }
140     }
141     // close db, rekey or import scene, need clear all remote query info
142     // local query info will destroy with syncEngine destruct
143     if (subManager_ != nullptr) {
144         subManager_->ClearAllRemoteQuery();
145     }
146 
147     RemoteExecutor *executor = GetAndIncRemoteExector();
148     if (executor != nullptr) {
149         executor->Close();
150         RefObject::DecObjRef(executor);
151         executor = nullptr;
152     }
153     ClearInnerResource();
154     LOGI("[SyncEngine] SyncEngine [%s] closed!", label_.c_str());
155     return E_OK;
156 }
157 
AddSyncOperation(SyncOperation * operation)158 int SyncEngine::AddSyncOperation(SyncOperation *operation)
159 {
160     if (operation == nullptr) {
161         LOGE("[SyncEngine] operation is nullptr");
162         return -E_INVALID_ARGS;
163     }
164 
165     std::vector<std::string> devices = operation->GetDevices();
166     std::string localDeviceId;
167     int errCode = GetLocalDeviceId(localDeviceId);
168     for (const auto &deviceId : devices) {
169         if (errCode != E_OK) {
170             operation->SetStatus(deviceId, errCode == -E_BUSY ?
171                 SyncOperation::OP_BUSY_FAILURE : SyncOperation::OP_FAILED);
172             continue;
173         }
174         if (!CheckDeviceIdValid(deviceId, localDeviceId)) {
175             operation->SetStatus(deviceId, SyncOperation::OP_INVALID_ARGS);
176             continue;
177         }
178         operation->SetStatus(deviceId, SyncOperation::OP_WAITING);
179         if (AddSyncOperForContext(deviceId, operation) != E_OK) {
180             operation->SetStatus(deviceId, SyncOperation::OP_FAILED);
181         }
182     }
183     return E_OK;
184 }
185 
RemoveSyncOperation(int syncId)186 void SyncEngine::RemoveSyncOperation(int syncId)
187 {
188     std::lock_guard<std::mutex> lock(contextMapLock_);
189     for (auto &iter : syncTaskContextMap_) {
190         ISyncTaskContext *context = iter.second;
191         if (context != nullptr) {
192             context->RemoveSyncOperation(syncId);
193         }
194     }
195 }
196 
197 #ifndef OMIT_MULTI_VER
BroadCastDataChanged() const198 void SyncEngine::BroadCastDataChanged() const
199 {
200     if (deviceManager_ != nullptr) {
201         (void)deviceManager_->SendBroadCast(LOCAL_DATA_CHANGED);
202     }
203 }
204 #endif // OMIT_MULTI_VER
205 
StartCommunicator()206 void SyncEngine::StartCommunicator()
207 {
208     if (communicator_ == nullptr) {
209         LOGE("[SyncEngine][StartCommunicator] communicator is not set!");
210         return;
211     }
212     LOGD("[SyncEngine][StartCommunicator] RegOnConnectCallback");
213     int errCode = communicator_->RegOnConnectCallback(
214         [this, deviceManager = deviceManager_](const std::string &targetDev, bool isConnect) {
215             deviceManager->OnDeviceConnectCallback(targetDev, isConnect);
216         }, nullptr);
217     if (errCode != E_OK) {
218         LOGE("[SyncEngine][StartCommunicator] register failed, auto sync can not use! err %d", errCode);
219         return;
220     }
221     communicator_->Activate();
222 }
223 
GetOnlineDevices(std::vector<std::string> & devices) const224 void SyncEngine::GetOnlineDevices(std::vector<std::string> &devices) const
225 {
226     devices.clear();
227     if (deviceManager_ != nullptr) {
228         deviceManager_->GetOnlineDevices(devices);
229     }
230 }
231 
InitInnerSource(const std::function<void (std::string)> & onRemoteDataChanged,const std::function<void (std::string)> & offlineChanged,ISyncInterface * syncInterface)232 int SyncEngine::InitInnerSource(const std::function<void(std::string)> &onRemoteDataChanged,
233     const std::function<void(std::string)> &offlineChanged, ISyncInterface *syncInterface)
234 {
235     deviceManager_ = new (std::nothrow) DeviceManager();
236     if (deviceManager_ == nullptr) {
237         LOGE("[SyncEngine] deviceManager alloc failed!");
238         return -E_OUT_OF_MEMORY;
239     }
240     auto executor = new (std::nothrow) RemoteExecutor();
241     if (executor == nullptr) {
242         LOGE("[SyncEngine] remoteExecutor alloc failed!");
243         delete deviceManager_;
244         deviceManager_ = nullptr;
245         return -E_OUT_OF_MEMORY;
246     }
247 
248     int errCode = E_OK;
249     do {
250         CommunicatorProxy *comProxy = nullptr;
251         {
252             std::lock_guard<std::mutex> lock(communicatorProxyLock_);
253             comProxy = communicatorProxy_;
254             RefObject::IncObjRef(comProxy);
255         }
256         errCode = deviceManager_->Initialize(comProxy, onRemoteDataChanged, offlineChanged);
257         RefObject::DecObjRef(comProxy);
258         if (errCode != E_OK) {
259             LOGE("[SyncEngine] deviceManager init failed! err %d", errCode);
260             break;
261         }
262         errCode = executor->Initialize(syncInterface, communicator_);
263     } while (false);
264     if (errCode != E_OK) {
265         delete deviceManager_;
266         deviceManager_ = nullptr;
267         delete executor;
268         executor = nullptr;
269     } else {
270         SetRemoteExector(executor);
271     }
272     return errCode;
273 }
274 
InitComunicator(const ISyncInterface * syncInterface)275 int SyncEngine::InitComunicator(const ISyncInterface *syncInterface)
276 {
277     ICommunicatorAggregator *communicatorAggregator = nullptr;
278     int errCode = RuntimeContext::GetInstance()->GetCommunicatorAggregator(communicatorAggregator);
279     if (communicatorAggregator == nullptr) {
280         LOGE("[SyncEngine] Get ICommunicatorAggregator error when init the sync engine err = %d", errCode);
281         return errCode;
282     }
283     std::vector<uint8_t> label = syncInterface->GetIdentifier();
284     bool isSyncDualTupleMode = syncInterface->GetDbProperties().GetBoolProp(DBProperties::SYNC_DUAL_TUPLE_MODE, false);
285     if (isSyncDualTupleMode) {
286         std::vector<uint8_t> dualTuplelabel = syncInterface->GetDualTupleIdentifier();
287         LOGI("[SyncEngine] dual tuple mode, original identifier=%.3s, target identifier=%.3s", VEC_TO_STR(label),
288             VEC_TO_STR(dualTuplelabel));
289         communicator_ = communicatorAggregator->AllocCommunicator(dualTuplelabel, errCode);
290     } else {
291         communicator_ = communicatorAggregator->AllocCommunicator(label, errCode);
292     }
293     if (communicator_ == nullptr) {
294         LOGE("[SyncEngine] AllocCommunicator error when init the sync engine! err = %d", errCode);
295         return errCode;
296     }
297 
298     errCode = communicator_->RegOnMessageCallback(
299         [this](const std::string &targetDev, Message *inMsg) { MessageReciveCallback(targetDev, inMsg); }, []() {});
300     if (errCode != E_OK) {
301         LOGE("[SyncEngine] SyncRequestCallback register failed! err = %d", errCode);
302         communicatorAggregator->ReleaseCommunicator(communicator_);
303         communicator_ = nullptr;
304         return errCode;
305     }
306     {
307         std::lock_guard<std::mutex> lock(communicatorProxyLock_);
308         communicatorProxy_ = new (std::nothrow) CommunicatorProxy();
309         if (communicatorProxy_ == nullptr) {
310             communicatorAggregator->ReleaseCommunicator(communicator_);
311             communicator_ = nullptr;
312             return -E_OUT_OF_MEMORY;
313         }
314         communicatorProxy_->SetMainCommunicator(communicator_);
315     }
316     label.resize(3); // only show 3 Bytes enough
317     label_ = DBCommon::VectorToHexString(label);
318     LOGD("[SyncEngine] RegOnConnectCallback");
319     return errCode;
320 }
321 
AddSyncOperForContext(const std::string & deviceId,SyncOperation * operation)322 int SyncEngine::AddSyncOperForContext(const std::string &deviceId, SyncOperation *operation)
323 {
324     int errCode = E_OK;
325     ISyncTaskContext *context = nullptr;
326     {
327         std::lock_guard<std::mutex> lock(contextMapLock_);
328         context = FindSyncTaskContext(deviceId);
329         if (context == nullptr) {
330             if (!IsKilled()) {
331                 context = GetSyncTaskContext(deviceId, errCode);
332             }
333             if (context == nullptr) {
334                 return errCode;
335             }
336         }
337         if (context->IsKilled()) { // LCOV_EXCL_BR_LINE
338             return -E_OBJ_IS_KILLED;
339         }
340         // IncRef for SyncEngine to make sure context is valid, to avoid a big lock
341         RefObject::IncObjRef(context);
342     }
343 
344     errCode = context->AddSyncOperation(operation);
345     if (operation != nullptr) {
346         operation->SetSyncContext(context); // make the life cycle of context and operation are same
347     }
348     RefObject::DecObjRef(context);
349     return errCode;
350 }
351 
MessageReciveCallbackTask(ISyncTaskContext * context,const ICommunicator * communicator,Message * inMsg)352 void SyncEngine::MessageReciveCallbackTask(ISyncTaskContext *context, const ICommunicator *communicator,
353     Message *inMsg)
354 {
355     std::string deviceId = context->GetDeviceId();
356 
357     if (inMsg->GetMessageId() != LOCAL_DATA_CHANGED) {
358         int errCode = context->ReceiveMessageCallback(inMsg);
359         if (errCode == -E_NOT_NEED_DELETE_MSG) {
360             goto MSG_CALLBACK_OUT_NOT_DEL;
361         }
362         // add auto sync here while recv subscribe request
363         QuerySyncObject syncObject;
364         if (errCode == E_OK && context->IsNeedTriggerQueryAutoSync(inMsg, syncObject)) {
365             InternalSyncParma param;
366             GetQueryAutoSyncParam(deviceId, syncObject, param);
367             queryAutoSyncCallback_(param);
368         }
369     }
370 
371     delete inMsg;
372     inMsg = nullptr;
373 MSG_CALLBACK_OUT_NOT_DEL:
374     ScheduleTaskOut(context, communicator);
375 }
376 
RemoteDataChangedTask(ISyncTaskContext * context,const ICommunicator * communicator,Message * inMsg)377 void SyncEngine::RemoteDataChangedTask(ISyncTaskContext *context, const ICommunicator *communicator, Message *inMsg)
378 {
379     std::string deviceId = context->GetDeviceId();
380     if (onRemoteDataChanged_ && deviceManager_->IsDeviceOnline(deviceId)) {
381         onRemoteDataChanged_(deviceId);
382     } else {
383         LOGE("[SyncEngine] onRemoteDataChanged is null!");
384     }
385     delete inMsg;
386     inMsg = nullptr;
387     ScheduleTaskOut(context, communicator);
388 }
389 
ScheduleTaskOut(ISyncTaskContext * context,const ICommunicator * communicator)390 void SyncEngine::ScheduleTaskOut(ISyncTaskContext *context, const ICommunicator *communicator)
391 {
392     (void)DealMsgUtilQueueEmpty();
393     DecExecTaskCount();
394     RefObject::DecObjRef(communicator);
395     RefObject::DecObjRef(context);
396 }
397 
DealMsgUtilQueueEmpty()398 int SyncEngine::DealMsgUtilQueueEmpty()
399 {
400     if (!isActive_) {
401         return -E_BUSY; // db is closing just return
402     }
403     int errCode = E_OK;
404     Message *inMsg = nullptr;
405     {
406         std::lock_guard<std::mutex> lock(queueLock_);
407         if (msgQueue_.empty()) {
408             return errCode;
409         }
410         inMsg = msgQueue_.front();
411         msgQueue_.pop_front();
412         queueCacheSize_ -= GetMsgSize(inMsg);
413     }
414 
415     IncExecTaskCount();
416     // it will deal with the first message in queue, we should increase object reference counts and sure that resources
417     // could be prevented from destroying by other threads.
418     do {
419         ISyncTaskContext *nextContext = GetContextForMsg(inMsg->GetTarget(), errCode);
420         if (errCode != E_OK) {
421             break;
422         }
423         errCode = ScheduleDealMsg(nextContext, inMsg);
424         if (errCode != E_OK) {
425             RefObject::DecObjRef(nextContext);
426         }
427     } while (false);
428     if (errCode != E_OK) {
429         delete inMsg;
430         inMsg = nullptr;
431         DecExecTaskCount();
432     }
433     return errCode;
434 }
435 
GetContextForMsg(const std::string & targetDev,int & errCode)436 ISyncTaskContext *SyncEngine::GetContextForMsg(const std::string &targetDev, int &errCode)
437 {
438     ISyncTaskContext *context = nullptr;
439     {
440         std::lock_guard<std::mutex> lock(contextMapLock_);
441         context = FindSyncTaskContext(targetDev);
442         if (context != nullptr) { // LCOV_EXCL_BR_LINE
443             if (context->IsKilled()) {
444                 errCode = -E_OBJ_IS_KILLED;
445                 return nullptr;
446             }
447         } else {
448             if (IsKilled()) {
449                 errCode = -E_OBJ_IS_KILLED;
450                 return nullptr;
451             }
452             context = GetSyncTaskContext(targetDev, errCode);
453             if (context == nullptr) {
454                 return nullptr;
455             }
456         }
457         // IncRef for context to make sure context is valid, when task run another thread
458         RefObject::IncObjRef(context);
459     }
460     return context;
461 }
462 
ScheduleDealMsg(ISyncTaskContext * context,Message * inMsg)463 int SyncEngine::ScheduleDealMsg(ISyncTaskContext *context, Message *inMsg)
464 {
465     if (inMsg == nullptr) {
466         LOGE("[SyncEngine] MessageReciveCallback inMsg is null!");
467         DecExecTaskCount();
468         return E_OK;
469     }
470     CommunicatorProxy *comProxy = nullptr;
471     {
472         std::lock_guard<std::mutex> lock(communicatorProxyLock_);
473         comProxy = communicatorProxy_;
474         RefObject::IncObjRef(comProxy);
475     }
476     int errCode = E_OK;
477     // deal remote local data changed message
478     if (inMsg->GetMessageId() == LOCAL_DATA_CHANGED) {
479         RemoteDataChangedTask(context, comProxy, inMsg);
480     } else {
481         errCode = RuntimeContext::GetInstance()->ScheduleTask(
482             [this, context, comProxy, inMsg] { MessageReciveCallbackTask(context, comProxy, inMsg); });
483     }
484 
485     if (errCode != E_OK) {
486         LOGE("[SyncEngine] MessageReciveCallbackTask Schedule failed err %d", errCode);
487         RefObject::DecObjRef(comProxy);
488     }
489     return errCode;
490 }
491 
MessageReciveCallback(const std::string & targetDev,Message * inMsg)492 void SyncEngine::MessageReciveCallback(const std::string &targetDev, Message *inMsg)
493 {
494     IncExecTaskCount();
495     int errCode = MessageReciveCallbackInner(targetDev, inMsg);
496     if (errCode != E_OK) {
497         delete inMsg;
498         inMsg = nullptr;
499         DecExecTaskCount();
500         LOGE("[SyncEngine] MessageReciveCallback failed!");
501     }
502 }
503 
MessageReciveCallbackInner(const std::string & targetDev,Message * inMsg)504 int SyncEngine::MessageReciveCallbackInner(const std::string &targetDev, Message *inMsg)
505 {
506     if (targetDev.empty() || inMsg == nullptr) {
507         LOGE("[SyncEngine][MessageReciveCallback] from a invalid device or inMsg is null ");
508         return -E_INVALID_ARGS;
509     }
510     if (!isActive_) {
511         LOGE("[SyncEngine] engine is closing, ignore msg");
512         return -E_BUSY;
513     }
514     if (inMsg->GetMessageId() == REMOTE_EXECUTE_MESSAGE) {
515         return HandleRemoteExecutorMsg(targetDev, inMsg);
516     }
517 
518     int msgSize = 0;
519     if (!IsSkipCalculateLen(inMsg)) {
520         msgSize = GetMsgSize(inMsg);
521         if (msgSize <= 0) {
522             LOGE("[SyncEngine] GetMsgSize makes a mistake");
523             return -E_NOT_SUPPORT;
524         }
525     }
526 
527     {
528         std::lock_guard<std::mutex> lock(queueLock_);
529         if ((queueCacheSize_ + msgSize) > maxQueueCacheSize_) {
530             LOGE("[SyncEngine] The size of message queue is beyond maximum");
531             discardMsgNum_++;
532             return -E_BUSY;
533         }
534 
535         if (execTaskCount_ > MAX_EXEC_NUM) {
536             PutMsgIntoQueue(targetDev, inMsg, msgSize);
537             // task dont exec here
538             DecExecTaskCount();
539             return E_OK;
540         }
541     }
542 
543     int errCode = E_OK;
544     ISyncTaskContext *nextContext = GetContextForMsg(targetDev, errCode);
545     if (errCode != E_OK) {
546         return errCode;
547     }
548     LOGD("[SyncEngine] MessageReciveCallback MSG ID = %d", inMsg->GetMessageId());
549     return ScheduleDealMsg(nextContext, inMsg);
550 }
551 
PutMsgIntoQueue(const std::string & targetDev,Message * inMsg,int msgSize)552 void SyncEngine::PutMsgIntoQueue(const std::string &targetDev, Message *inMsg, int msgSize)
553 {
554     if (inMsg->GetMessageId() == LOCAL_DATA_CHANGED) {
555         auto iter = std::find_if(msgQueue_.begin(), msgQueue_.end(),
556             [&targetDev](const Message *msg) {
557                 return targetDev == msg->GetTarget() && msg->GetMessageId() == LOCAL_DATA_CHANGED;
558             });
559         if (iter != msgQueue_.end()) { // LCOV_EXCL_BR_LINE
560             delete inMsg;
561             inMsg = nullptr;
562             return;
563         }
564     }
565     inMsg->SetTarget(targetDev);
566     msgQueue_.push_back(inMsg);
567     queueCacheSize_ += msgSize;
568     LOGW("[SyncEngine] The quantity of executing threads is beyond maximum. msgQueueSize = %zu", msgQueue_.size());
569 }
570 
GetMsgSize(const Message * inMsg) const571 int SyncEngine::GetMsgSize(const Message *inMsg) const
572 {
573     switch (inMsg->GetMessageId()) {
574         case TIME_SYNC_MESSAGE:
575             return TimeSync::CalculateLen(inMsg);
576         case ABILITY_SYNC_MESSAGE:
577             return AbilitySync::CalculateLen(inMsg);
578         case DATA_SYNC_MESSAGE:
579         case QUERY_SYNC_MESSAGE:
580         case CONTROL_SYNC_MESSAGE:
581             return SingleVerSerializeManager::CalculateLen(inMsg);
582 #ifndef OMIT_MULTI_VER
583         case COMMIT_HISTORY_SYNC_MESSAGE:
584             return CommitHistorySync::CalculateLen(inMsg);
585         case MULTI_VER_DATA_SYNC_MESSAGE:
586             return MultiVerDataSync::CalculateLen(inMsg);
587         case VALUE_SLICE_SYNC_MESSAGE:
588             return ValueSliceSync::CalculateLen(inMsg);
589 #endif
590         case LOCAL_DATA_CHANGED:
591             return DeviceManager::CalculateLen();
592         default:
593             LOGE("[SyncEngine] GetMsgSize not support msgId:%u", inMsg->GetMessageId());
594             return -E_NOT_SUPPORT;
595     }
596 }
597 
FindSyncTaskContext(const std::string & deviceId)598 ISyncTaskContext *SyncEngine::FindSyncTaskContext(const std::string &deviceId)
599 {
600     auto iter = syncTaskContextMap_.find(deviceId);
601     if (iter != syncTaskContextMap_.end()) {
602         ISyncTaskContext *context = iter->second;
603         return context;
604     }
605     return nullptr;
606 }
607 
GetSyncTaskContextAndInc(const std::string & deviceId)608 ISyncTaskContext *SyncEngine::GetSyncTaskContextAndInc(const std::string &deviceId)
609 {
610     ISyncTaskContext *context = nullptr;
611     std::lock_guard<std::mutex> lock(contextMapLock_);
612     context = FindSyncTaskContext(deviceId);
613     if (context == nullptr) {
614         LOGI("[SyncEngine] dev=%s, context is null, no need to clear sync operation", STR_MASK(deviceId));
615         return nullptr;
616     }
617     if (context->IsKilled()) { // LCOV_EXCL_BR_LINE
618         LOGI("[SyncEngine] context is killing");
619         return nullptr;
620     }
621     RefObject::IncObjRef(context);
622     return context;
623 }
624 
GetSyncTaskContext(const std::string & deviceId,int & errCode)625 ISyncTaskContext *SyncEngine::GetSyncTaskContext(const std::string &deviceId, int &errCode)
626 {
627     auto storage = GetAndIncSyncInterface();
628     if (storage == nullptr) {
629         errCode = -E_INVALID_DB;
630         LOGE("[SyncEngine] SyncTaskContext alloc failed with null db");
631         return nullptr;
632     }
633     ISyncTaskContext *context = CreateSyncTaskContext(*storage);
634     if (context == nullptr) {
635         errCode = -E_OUT_OF_MEMORY;
636         LOGE("[SyncEngine] SyncTaskContext alloc failed, may be no memory available!");
637         return nullptr;
638     }
639     errCode = context->Initialize(deviceId, storage, metadata_, communicatorProxy_);
640     if (errCode != E_OK) {
641         LOGE("[SyncEngine] context init failed err %d, dev %s", errCode, STR_MASK(deviceId));
642         RefObject::DecObjRef(context);
643         storage->DecRefCount();
644         context = nullptr;
645         return nullptr;
646     }
647     syncTaskContextMap_.insert(std::pair<std::string, ISyncTaskContext *>(deviceId, context));
648     // IncRef for SyncEngine to make sure SyncEngine is valid when context access
649     RefObject::IncObjRef(this);
650     context->OnLastRef([this, deviceId, storage]() {
651         LOGD("[SyncEngine] SyncTaskContext for id %s finalized", STR_MASK(deviceId));
652         RefObject::DecObjRef(this);
653         storage->DecRefCount();
654     });
655     context->RegOnSyncTask([this, context] { return ExecSyncTask(context); });
656     return context;
657 }
658 
ExecSyncTask(ISyncTaskContext * context)659 int SyncEngine::ExecSyncTask(ISyncTaskContext *context)
660 {
661     if (IsKilled()) {
662         return -E_OBJ_IS_KILLED;
663     }
664     AutoLock lockGuard(context);
665     int status = context->GetTaskExecStatus();
666     if ((status == SyncTaskContext::RUNNING) || context->IsKilled()) {
667         return -E_NOT_SUPPORT;
668     }
669     context->SetTaskExecStatus(ISyncTaskContext::RUNNING);
670     while (!context->IsTargetQueueEmpty()) {
671         int errCode = context->GetNextTarget();
672         if (errCode != E_OK) {
673             // current task execute failed, try next task
674             context->ClearSyncOperation();
675             continue;
676         }
677         if (context->IsCurrentSyncTaskCanBeSkipped()) { // LCOV_EXCL_BR_LINE
678             context->SetOperationStatus(SyncOperation::OP_FINISHED_ALL);
679             context->ClearSyncOperation();
680             continue;
681         }
682         context->UnlockObj();
683         errCode = context->StartStateMachine();
684         context->LockObj();
685         if (errCode != E_OK) {
686             // machine start failed because timer start failed, try to execute next task
687             LOGW("[SyncEngine] machine StartSync failed");
688             context->SetOperationStatus(SyncOperation::OP_FAILED);
689             context->ClearSyncOperation();
690             continue;
691         }
692         // now task is running just return here
693         return errCode;
694     }
695     LOGD("[SyncEngine] ExecSyncTask finished");
696     context->SetTaskExecStatus(ISyncTaskContext::FINISHED);
697     return E_OK;
698 }
699 
GetQueueCacheSize() const700 int SyncEngine::GetQueueCacheSize() const
701 {
702     return queueCacheSize_;
703 }
704 
GetDiscardMsgNum() const705 unsigned int SyncEngine::GetDiscardMsgNum() const
706 {
707     return discardMsgNum_;
708 }
709 
GetMaxExecNum() const710 unsigned int SyncEngine::GetMaxExecNum() const
711 {
712     return MAX_EXEC_NUM;
713 }
714 
SetMaxQueueCacheSize(int value)715 void SyncEngine::SetMaxQueueCacheSize(int value)
716 {
717     maxQueueCacheSize_ = value;
718 }
719 
GetLabel() const720 std::string SyncEngine::GetLabel() const
721 {
722     return label_;
723 }
724 
GetSyncRetry() const725 bool SyncEngine::GetSyncRetry() const
726 {
727     return isSyncRetry_;
728 }
729 
SetSyncRetry(bool isRetry)730 void SyncEngine::SetSyncRetry(bool isRetry)
731 {
732     if (isSyncRetry_ == isRetry) {
733         LOGI("sync retry is equal, syncTry=%d, no need to set.", isRetry);
734         return;
735     }
736     isSyncRetry_ = isRetry;
737     LOGI("[SyncEngine] SetSyncRetry:%d ok", isRetry);
738     std::lock_guard<std::mutex> lock(contextMapLock_);
739     for (auto &iter : syncTaskContextMap_) {
740         ISyncTaskContext *context = iter.second;
741         if (context != nullptr) { // LCOV_EXCL_BR_LINE
742             context->SetSyncRetry(isRetry);
743         }
744     }
745 }
746 
SetEqualIdentifier(const std::string & identifier,const std::vector<std::string> & targets)747 int SyncEngine::SetEqualIdentifier(const std::string &identifier, const std::vector<std::string> &targets)
748 {
749     if (!isActive_) {
750         LOGI("[SyncEngine] engine is closed, just put into map");
751         return E_OK;
752     }
753     ICommunicator *communicator = nullptr;
754     {
755         std::lock_guard<std::mutex> lock(equalCommunicatorsLock_);
756         if (equalCommunicators_.count(identifier) != 0) {
757             communicator = equalCommunicators_[identifier];
758         } else {
759             int errCode = E_OK;
760             communicator = AllocCommunicator(identifier, errCode);
761             if (communicator == nullptr) {
762                 return errCode;
763             }
764             equalCommunicators_[identifier] = communicator;
765         }
766     }
767     std::string targetDevices;
768     for (const auto &dev : targets) {
769         targetDevices += DBCommon::StringMasking(dev) + ",";
770     }
771     LOGI("[SyncEngine] set equal identifier=%.3s, original=%.3s, targetDevices=%s",
772         DBCommon::TransferStringToHex(identifier).c_str(), label_.c_str(),
773         targetDevices.substr(0, (targetDevices.size() > 0 ? targetDevices.size() - 1 : 0)).c_str());
774     {
775         std::lock_guard<std::mutex> lock(communicatorProxyLock_);
776         if (communicatorProxy_ == nullptr) {
777             return -E_INTERNAL_ERROR;
778         }
779         communicatorProxy_->SetEqualCommunicator(communicator, identifier, targets);
780     }
781     communicator->Activate();
782     return E_OK;
783 }
784 
SetEqualIdentifier()785 void SyncEngine::SetEqualIdentifier()
786 {
787     std::map<std::string, std::vector<std::string>> equalIdentifier; // key: equalIdentifier value: devices
788     for (auto &item : equalIdentifierMap_) {
789         if (equalIdentifier.find(item.second) == equalIdentifier.end()) { // LCOV_EXCL_BR_LINE
790             equalIdentifier[item.second] = {item.first};
791         } else {
792             equalIdentifier[item.second].push_back(item.first);
793         }
794     }
795     for (const auto &item : equalIdentifier) {
796         SetEqualIdentifier(item.first, item.second);
797     }
798 }
799 
SetEqualIdentifierMap(const std::string & identifier,const std::vector<std::string> & targets)800 void SyncEngine::SetEqualIdentifierMap(const std::string &identifier, const std::vector<std::string> &targets)
801 {
802     for (auto iter = equalIdentifierMap_.begin(); iter != equalIdentifierMap_.end();) {
803         if (identifier == iter->second) {
804             iter = equalIdentifierMap_.erase(iter);
805             continue;
806         }
807         iter++;
808     }
809     for (const auto &device : targets) {
810         equalIdentifierMap_[device] = identifier;
811     }
812 }
813 
OfflineHandleByDevice(const std::string & deviceId,ISyncInterface * storage)814 void SyncEngine::OfflineHandleByDevice(const std::string &deviceId, ISyncInterface *storage)
815 {
816     if (!isActive_) {
817         LOGD("[SyncEngine][OfflineHandleByDevice] ignore offline because not init");
818         return;
819     }
820     RemoteExecutor *executor = GetAndIncRemoteExector();
821     if (executor != nullptr) {
822         executor->NotifyDeviceOffline(deviceId);
823         RefObject::DecObjRef(executor);
824         executor = nullptr;
825     }
826     // db closed or device is offline
827     // clear remote subscribe and trigger
828     std::vector<std::string> remoteQueryId;
829     subManager_->GetRemoteSubscribeQueryIds(deviceId, remoteQueryId);
830     subManager_->ClearRemoteSubscribeQuery(deviceId);
831     for (const auto &queryId: remoteQueryId) {
832         if (!subManager_->IsQueryExistSubscribe(queryId)) {
833             static_cast<SingleVerKvDBSyncInterface *>(storage)->RemoveSubscribe(queryId);
834         }
835     }
836     DBInfo dbInfo;
837     static_cast<SyncGenericInterface *>(storage)->GetDBInfo(dbInfo);
838     RuntimeContext::GetInstance()->RemoveRemoteSubscribe(dbInfo, deviceId);
839     // get context and Inc context if context is not nullptr
840     ISyncTaskContext *context = GetSyncTaskContextAndInc(deviceId);
841     {
842         std::lock_guard<std::mutex> lock(communicatorProxyLock_);
843         if (communicatorProxy_ == nullptr) {
844             return;
845         }
846         if (communicatorProxy_->IsDeviceOnline(deviceId)) { // LCOV_EXCL_BR_LINE
847             LOGI("[SyncEngine] target dev=%s is online, no need to clear task.", STR_MASK(deviceId));
848             RefObject::DecObjRef(context);
849             return;
850         }
851     }
852     // means device is offline, clear local subscribe
853     subManager_->ClearLocalSubscribeQuery(deviceId);
854     // clear sync task
855     if (context != nullptr) {
856         context->ClearAllSyncTask();
857         RefObject::DecObjRef(context);
858     }
859 }
860 
GetLocalSubscribeQueries(const std::string & device,std::vector<QuerySyncObject> & subscribeQueries)861 void SyncEngine::GetLocalSubscribeQueries(const std::string &device, std::vector<QuerySyncObject> &subscribeQueries)
862 {
863     subManager_->GetLocalSubscribeQueries(device, subscribeQueries);
864 }
865 
GetRemoteSubscribeQueryIds(const std::string & device,std::vector<std::string> & subscribeQueryIds)866 void SyncEngine::GetRemoteSubscribeQueryIds(const std::string &device, std::vector<std::string> &subscribeQueryIds)
867 {
868     subManager_->GetRemoteSubscribeQueryIds(device, subscribeQueryIds);
869 }
870 
GetRemoteSubscribeQueries(const std::string & device,std::vector<QuerySyncObject> & subscribeQueries)871 void SyncEngine::GetRemoteSubscribeQueries(const std::string &device, std::vector<QuerySyncObject> &subscribeQueries)
872 {
873     subManager_->GetRemoteSubscribeQueries(device, subscribeQueries);
874 }
875 
PutUnfinishedSubQueries(const std::string & device,const std::vector<QuerySyncObject> & subscribeQueries)876 void SyncEngine::PutUnfinishedSubQueries(const std::string &device,
877     const std::vector<QuerySyncObject> &subscribeQueries)
878 {
879     subManager_->PutLocalUnFinishedSubQueries(device, subscribeQueries);
880 }
881 
GetAllUnFinishSubQueries(std::map<std::string,std::vector<QuerySyncObject>> & allSyncQueries)882 void SyncEngine::GetAllUnFinishSubQueries(std::map<std::string, std::vector<QuerySyncObject>> &allSyncQueries)
883 {
884     subManager_->GetAllUnFinishSubQueries(allSyncQueries);
885 }
886 
AllocCommunicator(const std::string & identifier,int & errCode)887 ICommunicator *SyncEngine::AllocCommunicator(const std::string &identifier, int &errCode)
888 {
889     ICommunicatorAggregator *communicatorAggregator = nullptr;
890     errCode = RuntimeContext::GetInstance()->GetCommunicatorAggregator(communicatorAggregator);
891     if (communicatorAggregator == nullptr) {
892         LOGE("[SyncEngine] Get ICommunicatorAggregator error when SetEqualIdentifier err = %d", errCode);
893         return nullptr;
894     }
895     std::vector<uint8_t> identifierVect(identifier.begin(), identifier.end());
896     auto communicator = communicatorAggregator->AllocCommunicator(identifierVect, errCode);
897     if (communicator == nullptr) {
898         LOGE("[SyncEngine] AllocCommunicator error when SetEqualIdentifier! err = %d", errCode);
899         return communicator;
900     }
901 
902     errCode = communicator->RegOnMessageCallback(
903         [this](const std::string &targetDev, Message *inMsg) { MessageReciveCallback(targetDev, inMsg); }, []() {});
904     if (errCode != E_OK) {
905         LOGE("[SyncEngine] SyncRequestCallback register failed in SetEqualIdentifier! err = %d", errCode);
906         communicatorAggregator->ReleaseCommunicator(communicator);
907         return nullptr;
908     }
909 
910     errCode = communicator->RegOnConnectCallback(
911         [this, deviceManager = deviceManager_](const std::string &targetDev, bool isConnect) {
912             deviceManager->OnDeviceConnectCallback(targetDev, isConnect);
913         }, nullptr);
914     if (errCode != E_OK) {
915         LOGE("[SyncEngine][RegConnCB] register failed in SetEqualIdentifier! err %d", errCode);
916         communicator->RegOnMessageCallback(nullptr, nullptr);
917         communicatorAggregator->ReleaseCommunicator(communicator);
918         return nullptr;
919     }
920 
921     return communicator;
922 }
923 
UnRegCommunicatorsCallback()924 void SyncEngine::UnRegCommunicatorsCallback()
925 {
926     if (communicator_ != nullptr) {
927         communicator_->RegOnMessageCallback(nullptr, nullptr);
928         communicator_->RegOnConnectCallback(nullptr, nullptr);
929         communicator_->RegOnSendableCallback(nullptr, nullptr);
930     }
931     std::lock_guard<std::mutex> lock(equalCommunicatorsLock_);
932     for (const auto &iter : equalCommunicators_) {
933         iter.second->RegOnMessageCallback(nullptr, nullptr);
934         iter.second->RegOnConnectCallback(nullptr, nullptr);
935         iter.second->RegOnSendableCallback(nullptr, nullptr);
936     }
937 }
938 
ReleaseCommunicators()939 void SyncEngine::ReleaseCommunicators()
940 {
941     {
942         std::lock_guard<std::mutex> lock(communicatorProxyLock_);
943         RefObject::KillAndDecObjRef(communicatorProxy_);
944         communicatorProxy_ = nullptr;
945     }
946     ICommunicatorAggregator *communicatorAggregator = nullptr;
947     int errCode = RuntimeContext::GetInstance()->GetCommunicatorAggregator(communicatorAggregator);
948     if (communicatorAggregator == nullptr) {
949         LOGF("[SyncEngine] ICommunicatorAggregator get failed when fialize SyncEngine err %d", errCode);
950         return;
951     }
952 
953     if (communicator_ != nullptr) {
954         communicatorAggregator->ReleaseCommunicator(communicator_);
955         communicator_ = nullptr;
956     }
957 
958     std::lock_guard<std::mutex> lock(equalCommunicatorsLock_);
959     for (auto &iter : equalCommunicators_) {
960         communicatorAggregator->ReleaseCommunicator(iter.second);
961     }
962     equalCommunicators_.clear();
963 }
964 
IsSkipCalculateLen(const Message * inMsg)965 bool SyncEngine::IsSkipCalculateLen(const Message *inMsg)
966 {
967     if (inMsg->IsFeedbackError()) {
968         LOGE("[SyncEngine] Feedback Message with errorNo=%u.", inMsg->GetErrorNo());
969         return true;
970     }
971     return false;
972 }
973 
GetSubscribeSyncParam(const std::string & device,const QuerySyncObject & query,InternalSyncParma & outParam)974 void SyncEngine::GetSubscribeSyncParam(const std::string &device, const QuerySyncObject &query,
975     InternalSyncParma &outParam)
976 {
977     outParam.devices = { device };
978     outParam.mode = SyncModeType::AUTO_SUBSCRIBE_QUERY;
979     outParam.isQuerySync = true;
980     outParam.syncQuery = query;
981 }
982 
GetQueryAutoSyncParam(const std::string & device,const QuerySyncObject & query,InternalSyncParma & outParam)983 void SyncEngine::GetQueryAutoSyncParam(const std::string &device, const QuerySyncObject &query,
984     InternalSyncParma &outParam)
985 {
986     outParam.devices = { device };
987     outParam.mode = SyncModeType::AUTO_PUSH;
988     outParam.isQuerySync = true;
989     outParam.syncQuery = query;
990 }
991 
992 int SyncEngine::StartAutoSubscribeTimer([[gnu::unused]] const ISyncInterface &syncInterface)
993 {
994     return E_OK;
995 }
996 
StopAutoSubscribeTimer()997 void SyncEngine::StopAutoSubscribeTimer()
998 {
999 }
1000 
SubscribeLimitCheck(const std::vector<std::string> & devices,QuerySyncObject & query) const1001 int SyncEngine::SubscribeLimitCheck(const std::vector<std::string> &devices, QuerySyncObject &query) const
1002 {
1003     return subManager_->LocalSubscribeLimitCheck(devices, query);
1004 }
1005 
1006 
ClearInnerResource()1007 void SyncEngine::ClearInnerResource()
1008 {
1009     ClearSyncInterface();
1010     if (deviceManager_ != nullptr) {
1011         delete deviceManager_;
1012         deviceManager_ = nullptr;
1013     }
1014     communicator_ = nullptr;
1015     metadata_ = nullptr;
1016     onRemoteDataChanged_ = nullptr;
1017     offlineChanged_ = nullptr;
1018     queryAutoSyncCallback_ = nullptr;
1019     std::lock_guard<std::mutex> autoLock(remoteExecutorLock_);
1020     if (remoteExecutor_ != nullptr) {
1021         RefObject::KillAndDecObjRef(remoteExecutor_);
1022         remoteExecutor_ = nullptr;
1023     }
1024 }
1025 
IsEngineActive() const1026 bool SyncEngine::IsEngineActive() const
1027 {
1028     return isActive_;
1029 }
1030 
SchemaChange()1031 void SyncEngine::SchemaChange()
1032 {
1033     std::vector<ISyncTaskContext *> tmpContextVec;
1034     {
1035         std::lock_guard<std::mutex> lock(contextMapLock_);
1036         for (const auto &entry : syncTaskContextMap_) {
1037             auto context = entry.second;
1038             if (context == nullptr || context->IsKilled()) { // LCOV_EXCL_BR_LINE
1039                 continue;
1040             }
1041             RefObject::IncObjRef(context);
1042             tmpContextVec.push_back(context);
1043         }
1044     }
1045     for (const auto &entryContext : tmpContextVec) {
1046         entryContext->SchemaChange();
1047         RefObject::DecObjRef(entryContext);
1048     }
1049 }
1050 
IncExecTaskCount()1051 void SyncEngine::IncExecTaskCount()
1052 {
1053     std::lock_guard<std::mutex> incLock(execTaskCountLock_);
1054     execTaskCount_++;
1055 }
1056 
DecExecTaskCount()1057 void SyncEngine::DecExecTaskCount()
1058 {
1059     {
1060         std::lock_guard<std::mutex> decLock(execTaskCountLock_);
1061         execTaskCount_--;
1062     }
1063     execTaskCv_.notify_all();
1064 }
1065 
Dump(int fd)1066 void SyncEngine::Dump(int fd)
1067 {
1068     {
1069         std::lock_guard<std::mutex> lock(communicatorProxyLock_);
1070         std::string communicatorLabel;
1071         if (communicatorProxy_ != nullptr) {
1072             communicatorProxy_->GetLocalIdentity(communicatorLabel);
1073         }
1074         DBDumpHelper::Dump(fd, "\tcommunicator label = %s, equalIdentify Info [\n", communicatorLabel.c_str());
1075         if (communicatorProxy_ != nullptr) {
1076             communicatorProxy_->GetLocalIdentity(communicatorLabel);
1077             communicatorProxy_->Dump(fd);
1078         }
1079     }
1080     DBDumpHelper::Dump(fd, "\t]\n\tcontext info [\n");
1081     // dump context info
1082     std::lock_guard<std::mutex> autoLock(contextMapLock_);
1083     for (const auto &entry : syncTaskContextMap_) {
1084         if (entry.second != nullptr) {
1085             entry.second->Dump(fd);
1086         }
1087     }
1088     DBDumpHelper::Dump(fd, "\t]\n\n");
1089 }
1090 
RemoteQuery(const std::string & device,const RemoteCondition & condition,uint64_t timeout,uint64_t connectionId,std::shared_ptr<ResultSet> & result)1091 int SyncEngine::RemoteQuery(const std::string &device, const RemoteCondition &condition,
1092     uint64_t timeout, uint64_t connectionId, std::shared_ptr<ResultSet> &result)
1093 {
1094     RemoteExecutor *executor = GetAndIncRemoteExector();
1095     if (!isActive_ || executor == nullptr) {
1096         RefObject::DecObjRef(executor);
1097         return -E_BUSY; // db is closing just return
1098     }
1099     int errCode = executor->RemoteQuery(device, condition, timeout, connectionId, result);
1100     RefObject::DecObjRef(executor);
1101     return errCode;
1102 }
1103 
NotifyConnectionClosed(uint64_t connectionId)1104 void SyncEngine::NotifyConnectionClosed(uint64_t connectionId)
1105 {
1106     RemoteExecutor *executor = GetAndIncRemoteExector();
1107     if (!isActive_ || executor == nullptr) {
1108         RefObject::DecObjRef(executor);
1109         return; // db is closing just return
1110     }
1111     executor->NotifyConnectionClosed(connectionId);
1112     RefObject::DecObjRef(executor);
1113 }
1114 
NotifyUserChange()1115 void SyncEngine::NotifyUserChange()
1116 {
1117     RemoteExecutor *executor = GetAndIncRemoteExector();
1118     if (!isActive_ || executor == nullptr) {
1119         RefObject::DecObjRef(executor);
1120         return; // db is closing just return
1121     }
1122     executor->NotifyUserChange();
1123     RefObject::DecObjRef(executor);
1124 }
1125 
GetAndIncRemoteExector()1126 RemoteExecutor *SyncEngine::GetAndIncRemoteExector()
1127 {
1128     std::lock_guard<std::mutex> autoLock(remoteExecutorLock_);
1129     RefObject::IncObjRef(remoteExecutor_);
1130     return remoteExecutor_;
1131 }
1132 
SetRemoteExector(RemoteExecutor * executor)1133 void SyncEngine::SetRemoteExector(RemoteExecutor *executor)
1134 {
1135     std::lock_guard<std::mutex> autoLock(remoteExecutorLock_);
1136     remoteExecutor_ = executor;
1137 }
1138 
CheckDeviceIdValid(const std::string & checkDeviceId,const std::string & localDeviceId)1139 bool SyncEngine::CheckDeviceIdValid(const std::string &checkDeviceId, const std::string &localDeviceId)
1140 {
1141     if (checkDeviceId.empty()) {
1142         return false;
1143     }
1144     if (checkDeviceId.length() > DBConstant::MAX_DEV_LENGTH) {
1145         LOGE("[SyncEngine] dev is too long len=%zu", checkDeviceId.length());
1146         return false;
1147     }
1148     return localDeviceId != checkDeviceId;
1149 }
1150 
GetLocalDeviceId(std::string & deviceId)1151 int SyncEngine::GetLocalDeviceId(std::string &deviceId)
1152 {
1153     if (!isActive_ || communicator_ == nullptr) {
1154         // db is closing
1155         return -E_BUSY;
1156     }
1157     auto communicator = communicator_;
1158     RefObject::IncObjRef(communicator);
1159     int errCode = communicator->GetLocalIdentity(deviceId);
1160     RefObject::DecObjRef(communicator);
1161     return errCode;
1162 }
1163 
AbortMachineIfNeed(uint32_t syncId)1164 void SyncEngine::AbortMachineIfNeed(uint32_t syncId)
1165 {
1166     std::vector<ISyncTaskContext *> abortContexts;
1167     {
1168         std::lock_guard<std::mutex> lock(contextMapLock_);
1169         for (const auto &entry : syncTaskContextMap_) {
1170             auto context = entry.second;
1171             if (context == nullptr || context->IsKilled()) { // LCOV_EXCL_BR_LINE
1172                 continue;
1173             }
1174             RefObject::IncObjRef(context);
1175             if (context->GetSyncId() == syncId) {
1176                 RefObject::IncObjRef(context);
1177                 abortContexts.push_back(context);
1178             }
1179             RefObject::DecObjRef(context);
1180         }
1181     }
1182     for (const auto &abortContext : abortContexts) {
1183         abortContext->AbortMachineIfNeed(static_cast<uint32_t>(syncId));
1184         RefObject::DecObjRef(abortContext);
1185     }
1186 }
1187 
WaitingExecTaskExist()1188 void SyncEngine::WaitingExecTaskExist()
1189 {
1190     std::unique_lock<std::mutex> closeLock(execTaskCountLock_);
1191     bool isTimeout = execTaskCv_.wait_for(closeLock, std::chrono::milliseconds(DBConstant::MIN_TIMEOUT),
1192         [this]() { return execTaskCount_ == 0; });
1193     if (!isTimeout) { // LCOV_EXCL_BR_LINE
1194         LOGD("SyncEngine Close with executing task!");
1195     }
1196 }
1197 
HandleRemoteExecutorMsg(const std::string & targetDev,Message * inMsg)1198 int SyncEngine::HandleRemoteExecutorMsg(const std::string &targetDev, Message *inMsg)
1199 {
1200     RemoteExecutor *executor = GetAndIncRemoteExector();
1201     int errCode = E_OK;
1202     if (executor != nullptr) {
1203         errCode = executor->ReceiveMessage(targetDev, inMsg);
1204     } else {
1205         errCode = -E_BUSY;
1206     }
1207     DecExecTaskCount();
1208     RefObject::DecObjRef(executor);
1209     return errCode;
1210 }
1211 
AddSubscribe(SyncGenericInterface * storage,const std::map<std::string,std::vector<QuerySyncObject>> & subscribeQuery)1212 void SyncEngine::AddSubscribe(SyncGenericInterface *storage,
1213     const std::map<std::string, std::vector<QuerySyncObject>> &subscribeQuery)
1214 {
1215     for (const auto &[device, queryList]: subscribeQuery) {
1216         for (const auto &query: queryList) {
1217             AddQuerySubscribe(storage, device, query);
1218         }
1219     }
1220 }
1221 
AddQuerySubscribe(SyncGenericInterface * storage,const std::string & device,const QuerySyncObject & query)1222 void SyncEngine::AddQuerySubscribe(SyncGenericInterface *storage, const std::string &device,
1223     const QuerySyncObject &query)
1224 {
1225     int errCode = storage->AddSubscribe(query.GetIdentify(), query, true);
1226     if (errCode != E_OK) {
1227         LOGW("[SyncEngine][AddSubscribe] Add trigger failed dev = %s queryId = %s",
1228             STR_MASK(device), STR_MASK(query.GetIdentify()));
1229         return;
1230     }
1231     errCode = subManager_->ReserveRemoteSubscribeQuery(device, query);
1232     if (errCode != E_OK) {
1233         if (!subManager_->IsQueryExistSubscribe(query.GetIdentify())) { // LCOV_EXCL_BR_LINE
1234             (void)storage->RemoveSubscribe(query.GetIdentify());
1235         }
1236         return;
1237     }
1238     subManager_->ActiveRemoteSubscribeQuery(device, query);
1239 }
1240 
TimeChange()1241 void SyncEngine::TimeChange()
1242 {
1243     std::vector<ISyncTaskContext *> decContext;
1244     {
1245         // copy context
1246         std::lock_guard<std::mutex> lock(contextMapLock_);
1247         for (const auto &iter : syncTaskContextMap_) {
1248             RefObject::IncObjRef(iter.second);
1249             decContext.push_back(iter.second);
1250         }
1251     }
1252     for (auto &iter : decContext) {
1253         iter->TimeChange();
1254         RefObject::DecObjRef(iter);
1255     }
1256 }
1257 
GetResponseTaskCount()1258 int32_t SyncEngine::GetResponseTaskCount()
1259 {
1260     std::vector<ISyncTaskContext *> decContext;
1261     {
1262         // copy context
1263         std::lock_guard<std::mutex> lock(contextMapLock_);
1264         for (const auto &iter : syncTaskContextMap_) {
1265             RefObject::IncObjRef(iter.second);
1266             decContext.push_back(iter.second);
1267         }
1268     }
1269     int32_t taskCount = 0;
1270     for (auto &iter : decContext) {
1271         taskCount += iter->GetResponseTaskCount();
1272         RefObject::DecObjRef(iter);
1273     }
1274     {
1275         std::lock_guard<std::mutex> decLock(execTaskCountLock_);
1276         taskCount += static_cast<int32_t>(execTaskCount_);
1277     }
1278     return taskCount;
1279 }
1280 
ClearSyncInterface()1281 void SyncEngine::ClearSyncInterface()
1282 {
1283     ISyncInterface *syncInterface = nullptr;
1284     {
1285         std::lock_guard<std::mutex> autoLock(storageMutex_);
1286         if (syncInterface_ == nullptr) {
1287             return;
1288         }
1289         syncInterface = syncInterface_;
1290         syncInterface_ = nullptr;
1291     }
1292     syncInterface->DecRefCount();
1293 }
1294 
GetAndIncSyncInterface()1295 ISyncInterface *SyncEngine::GetAndIncSyncInterface()
1296 {
1297     std::lock_guard<std::mutex> autoLock(storageMutex_);
1298     if (syncInterface_ == nullptr) {
1299         return nullptr;
1300     }
1301     syncInterface_->IncRefCount();
1302     return syncInterface_;
1303 }
1304 
SetSyncInterface(ISyncInterface * syncInterface)1305 void SyncEngine::SetSyncInterface(ISyncInterface *syncInterface)
1306 {
1307     std::lock_guard<std::mutex> autoLock(storageMutex_);
1308     syncInterface_ = syncInterface;
1309 }
1310 } // namespace DistributedDB
1311