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