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_task_context.h"
17 
18 #include <algorithm>
19 #include <cmath>
20 
21 #include "db_constant.h"
22 #include "db_dump_helper.h"
23 #include "db_dfx_adapter.h"
24 #include "db_errno.h"
25 #include "hash.h"
26 #include "isync_state_machine.h"
27 #include "log_print.h"
28 #include "time_helper.h"
29 #include "version.h"
30 
31 namespace DistributedDB {
32 std::mutex SyncTaskContext::synTaskContextSetLock_;
33 std::set<ISyncTaskContext *> SyncTaskContext::synTaskContextSet_;
34 
35 namespace {
36     constexpr int NEGOTIATION_LIMIT = 2;
37     constexpr uint32_t SESSION_ID_MAX_VALUE = 0x8fffffffu;
38 }
39 
SyncTaskContext()40 SyncTaskContext::SyncTaskContext()
41     : syncOperation_(nullptr),
42       syncId_(0),
43       mode_(0),
44       isAutoSync_(false),
45       status_(0),
46       taskExecStatus_(0),
47       syncInterface_(nullptr),
48       communicator_(nullptr),
49       stateMachine_(nullptr),
50       requestSessionId_(0),
51       lastRequestSessionId_(0),
52       timeHelper_(nullptr),
53       remoteSoftwareVersion_(0),
54       remoteSoftwareVersionId_(0),
55       isCommNormal_(true),
56       taskErrCode_(E_OK),
57       commErrCode_(E_OK),
58       syncTaskRetryStatus_(false),
59       isSyncRetry_(false),
60       negotiationCount_(0),
61       isAutoSubscribe_(false)
62 {
63 }
64 
~SyncTaskContext()65 SyncTaskContext::~SyncTaskContext()
66 {
67     if (stateMachine_ != nullptr) {
68         delete stateMachine_;
69         stateMachine_ = nullptr;
70     }
71     SyncTaskContext::ClearSyncOperation();
72     ClearSyncTarget();
73     syncInterface_ = nullptr;
74     communicator_ = nullptr;
75 }
76 
AddSyncTarget(ISyncTarget * target)77 int SyncTaskContext::AddSyncTarget(ISyncTarget *target)
78 {
79     if (target == nullptr) {
80         return -E_INVALID_ARGS;
81     }
82     int targetMode = target->GetMode();
83     auto syncId = static_cast<uint32_t>(target->GetSyncId());
84     {
85         std::lock_guard<std::mutex> lock(targetQueueLock_);
86         if (target->GetTaskType() == ISyncTarget::REQUEST) {
87             requestTargetQueue_.push_back(target);
88         } else if (target->GetTaskType() == ISyncTarget::RESPONSE) {
89             responseTargetQueue_.push_back(target);
90         } else {
91             return -E_INVALID_ARGS;
92         }
93     }
94     RefObject::IncObjRef(this);
95     int errCode = RuntimeContext::GetInstance()->ScheduleTask([this, targetMode, syncId]() {
96         CancelCurrentSyncRetryIfNeed(targetMode, syncId);
97         RefObject::DecObjRef(this);
98     });
99     if (errCode != E_OK) {
100         RefObject::DecObjRef(this);
101     }
102     if (onSyncTaskAdd_) {
103         RefObject::IncObjRef(this);
104         errCode = RuntimeContext::GetInstance()->ScheduleTask([this]() {
105             onSyncTaskAdd_();
106             RefObject::DecObjRef(this);
107         });
108         if (errCode != E_OK) {
109             RefObject::DecObjRef(this);
110         }
111     }
112     return E_OK;
113 }
114 
SetOperationStatus(int status)115 void SyncTaskContext::SetOperationStatus(int status)
116 {
117     std::lock_guard<std::mutex> lock(operationLock_);
118     if (syncOperation_ == nullptr) {
119         LOGD("[SyncTaskContext][SetStatus] syncOperation is null");
120         return;
121     }
122     int finalStatus = status;
123 
124     int operationStatus = syncOperation_->GetStatus(deviceId_);
125     if (status == SyncOperation::OP_SEND_FINISHED && operationStatus == SyncOperation::OP_RECV_FINISHED) {
126         if (GetTaskErrCode() == -E_EKEYREVOKED) { // LCOV_EXCL_BR_LINE
127             finalStatus = SyncOperation::OP_EKEYREVOKED_FAILURE;
128         } else {
129             finalStatus = SyncOperation::OP_FINISHED_ALL;
130         }
131     } else if (status == SyncOperation::OP_RECV_FINISHED && operationStatus == SyncOperation::OP_SEND_FINISHED) {
132         if (GetTaskErrCode() == -E_EKEYREVOKED) {
133             finalStatus = SyncOperation::OP_EKEYREVOKED_FAILURE;
134         } else {
135             finalStatus = SyncOperation::OP_FINISHED_ALL;
136         }
137     }
138     syncOperation_->SetStatus(deviceId_, finalStatus);
139     if (finalStatus >= SyncOperation::OP_FINISHED_ALL) {
140         SaveLastPushTaskExecStatus(finalStatus);
141     }
142     if (syncOperation_->CheckIsAllFinished()) {
143         syncOperation_->Finished();
144     }
145 }
146 
SaveLastPushTaskExecStatus(int finalStatus)147 void SyncTaskContext::SaveLastPushTaskExecStatus(int finalStatus)
148 {
149     (void)finalStatus;
150 }
151 
Clear()152 void SyncTaskContext::Clear()
153 {
154     StopTimer();
155     retryTime_ = 0;
156     sequenceId_ = 1;
157     syncId_ = 0;
158     isAutoSync_ = false;
159     requestSessionId_ = 0;
160     isNeedRetry_ = NO_NEED_RETRY;
161     mode_ = SyncModeType::INVALID_MODE;
162     status_ = SyncOperation::OP_WAITING;
163     taskErrCode_ = E_OK;
164     packetId_ = 0;
165     isAutoSubscribe_ = false;
166 }
167 
RemoveSyncOperation(int syncId)168 int SyncTaskContext::RemoveSyncOperation(int syncId)
169 {
170     std::lock_guard<std::mutex> lock(targetQueueLock_);
171     auto iter = std::find_if(requestTargetQueue_.begin(), requestTargetQueue_.end(),
172         [syncId](const ISyncTarget *target) {
173             if (target == nullptr) {
174                 return false;
175             }
176             return target->GetSyncId() == syncId;
177         });
178     if (iter != requestTargetQueue_.end()) {
179         if (*iter != nullptr) {
180             delete *iter;
181             *iter = nullptr;
182         }
183         requestTargetQueue_.erase(iter);
184         return E_OK;
185     }
186     return -E_INVALID_ARGS;
187 }
188 
ClearSyncTarget()189 void SyncTaskContext::ClearSyncTarget()
190 {
191     std::lock_guard<std::mutex> lock(targetQueueLock_);
192     for (auto &requestTarget : requestTargetQueue_) {
193         if (requestTarget != nullptr) {
194             delete requestTarget;
195             requestTarget = nullptr;
196         }
197     }
198     requestTargetQueue_.clear();
199 
200     for (auto &responseTarget : responseTargetQueue_) {
201         if (responseTarget != nullptr) { // LCOV_EXCL_BR_LINE
202             delete responseTarget;
203             responseTarget = nullptr;
204         }
205     }
206     responseTargetQueue_.clear();
207 }
208 
IsTargetQueueEmpty() const209 bool SyncTaskContext::IsTargetQueueEmpty() const
210 {
211     std::lock_guard<std::mutex> lock(targetQueueLock_);
212     return requestTargetQueue_.empty() && responseTargetQueue_.empty();
213 }
214 
GetOperationStatus() const215 int SyncTaskContext::GetOperationStatus() const
216 {
217     std::lock_guard<std::mutex> lock(operationLock_);
218     if (syncOperation_ == nullptr) {
219         return SyncOperation::OP_FINISHED_ALL;
220     }
221     return syncOperation_->GetStatus(deviceId_);
222 }
223 
SetMode(int mode)224 void SyncTaskContext::SetMode(int mode)
225 {
226     mode_ = mode;
227 }
228 
GetMode() const229 int SyncTaskContext::GetMode() const
230 {
231     return mode_;
232 }
233 
MoveToNextTarget()234 void SyncTaskContext::MoveToNextTarget()
235 {
236     ClearSyncOperation();
237     TaskParam param;
238     // call other system api without lock
239     param.timeout = communicator_->GetTimeout(deviceId_);
240     std::lock_guard<std::mutex> lock(targetQueueLock_);
241     while (!requestTargetQueue_.empty() || !responseTargetQueue_.empty()) {
242         ISyncTarget *tmpTarget = nullptr;
243         if (!requestTargetQueue_.empty()) {
244             tmpTarget = requestTargetQueue_.front();
245             requestTargetQueue_.pop_front();
246         } else {
247             tmpTarget = responseTargetQueue_.front();
248             responseTargetQueue_.pop_front();
249         }
250         if (tmpTarget == nullptr) {
251             LOGE("[SyncTaskContext][MoveToNextTarget] currentTarget is null skip!");
252             continue;
253         }
254         SyncOperation *tmpOperation = nullptr;
255         tmpTarget->GetSyncOperation(tmpOperation);
256         if ((tmpOperation != nullptr) && tmpOperation->IsKilled()) {
257             // if killed skip this syncOperation_.
258             delete tmpTarget;
259             tmpTarget = nullptr;
260             continue;
261         }
262         CopyTargetData(tmpTarget, param);
263         delete tmpTarget;
264         tmpTarget = nullptr;
265         break;
266     }
267 }
268 
GetNextTarget()269 int SyncTaskContext::GetNextTarget()
270 {
271     MoveToNextTarget();
272     int checkErrCode = RunPermissionCheck(GetPermissionCheckFlag(IsAutoSync(), GetMode()));
273     if (checkErrCode != E_OK) {
274         SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
275         return checkErrCode;
276     }
277     return E_OK;
278 }
279 
GetSyncId() const280 uint32_t SyncTaskContext::GetSyncId() const
281 {
282     return syncId_;
283 }
284 
285 // Get the current task deviceId.
GetDeviceId() const286 std::string SyncTaskContext::GetDeviceId() const
287 {
288     return deviceId_;
289 }
290 
SetTaskExecStatus(int status)291 void SyncTaskContext::SetTaskExecStatus(int status)
292 {
293     taskExecStatus_ = status;
294 }
295 
GetTaskExecStatus() const296 int SyncTaskContext::GetTaskExecStatus() const
297 {
298     return taskExecStatus_;
299 }
300 
IsAutoSync() const301 bool SyncTaskContext::IsAutoSync() const
302 {
303     return isAutoSync_;
304 }
305 
StartTimer()306 int SyncTaskContext::StartTimer()
307 {
308     std::lock_guard<std::mutex> lockGuard(timerLock_);
309     if (timerId_ > 0) {
310         return -E_UNEXPECTED_DATA;
311     }
312     TimerId timerId = 0;
313     RefObject::IncObjRef(this);
314     TimerAction timeOutCallback = [this](TimerId id) { return TimeOut(id); };
315     int errCode = RuntimeContext::GetInstance()->SetTimer(timeout_, timeOutCallback,
316         [this]() {
317             int ret = RuntimeContext::GetInstance()->ScheduleTask([this]() { RefObject::DecObjRef(this); });
318             if (ret != E_OK) {
319                 LOGE("[SyncTaskContext] timer finalizer ScheduleTask, errCode %d", ret);
320             }
321         }, timerId);
322     if (errCode != E_OK) {
323         RefObject::DecObjRef(this);
324         return errCode;
325     }
326     timerId_ = timerId;
327     return errCode;
328 }
329 
StopTimer()330 void SyncTaskContext::StopTimer()
331 {
332     TimerId timerId;
333     {
334         std::lock_guard<std::mutex> lockGuard(timerLock_);
335         timerId = timerId_;
336         if (timerId_ == 0) {
337             return;
338         }
339         timerId_ = 0;
340     }
341     RuntimeContext::GetInstance()->RemoveTimer(timerId);
342 }
343 
ModifyTimer(int milliSeconds)344 int SyncTaskContext::ModifyTimer(int milliSeconds)
345 {
346     std::lock_guard<std::mutex> lockGuard(timerLock_);
347     if (timerId_ == 0) {
348         return -E_UNEXPECTED_DATA;
349     }
350     return RuntimeContext::GetInstance()->ModifyTimer(timerId_, milliSeconds);
351 }
352 
SetRetryTime(int retryTime)353 void SyncTaskContext::SetRetryTime(int retryTime)
354 {
355     retryTime_ = retryTime;
356 }
357 
GetRetryTime() const358 int SyncTaskContext::GetRetryTime() const
359 {
360     return retryTime_;
361 }
362 
SetRetryStatus(int isNeedRetry)363 void SyncTaskContext::SetRetryStatus(int isNeedRetry)
364 {
365     isNeedRetry_ = isNeedRetry;
366 }
367 
GetRetryStatus() const368 int SyncTaskContext::GetRetryStatus() const
369 {
370     return isNeedRetry_;
371 }
372 
GetTimerId() const373 TimerId SyncTaskContext::GetTimerId() const
374 {
375     return timerId_;
376 }
377 
GetRequestSessionId() const378 uint32_t SyncTaskContext::GetRequestSessionId() const
379 {
380     return requestSessionId_;
381 }
382 
IncSequenceId()383 void SyncTaskContext::IncSequenceId()
384 {
385     sequenceId_++;
386 }
387 
GetSequenceId() const388 uint32_t SyncTaskContext::GetSequenceId() const
389 {
390     return sequenceId_;
391 }
392 
ReSetSequenceId()393 void SyncTaskContext::ReSetSequenceId()
394 {
395     sequenceId_ = 1;
396 }
397 
IncPacketId()398 void SyncTaskContext::IncPacketId()
399 {
400     packetId_++;
401 }
402 
GetPacketId() const403 uint64_t SyncTaskContext::GetPacketId() const
404 {
405     return packetId_;
406 }
407 
GetTimeoutTime() const408 int SyncTaskContext::GetTimeoutTime() const
409 {
410     return timeout_;
411 }
412 
SetTimeoutCallback(const TimerAction & timeOutCallback)413 void SyncTaskContext::SetTimeoutCallback(const TimerAction &timeOutCallback)
414 {
415     timeOutCallback_ = timeOutCallback;
416 }
417 
SetTimeOffset(TimeOffset offset)418 void SyncTaskContext::SetTimeOffset(TimeOffset offset)
419 {
420     timeOffset_ = offset;
421 }
422 
GetTimeOffset() const423 TimeOffset SyncTaskContext::GetTimeOffset() const
424 {
425     return timeOffset_;
426 }
427 
StartStateMachine()428 int SyncTaskContext::StartStateMachine()
429 {
430     return stateMachine_->StartSync();
431 }
432 
ReceiveMessageCallback(Message * inMsg)433 int SyncTaskContext::ReceiveMessageCallback(Message *inMsg)
434 {
435     if (GetRemoteSoftwareVersion() <= SOFTWARE_VERSION_BASE && inMsg->GetMessageId() != ABILITY_SYNC_MESSAGE) {
436         uint16_t remoteVersion = 0;
437         (void)communicator_->GetRemoteCommunicatorVersion(deviceId_, remoteVersion);
438         SetRemoteSoftwareVersion(SOFTWARE_VERSION_EARLIEST + remoteVersion);
439     }
440     int errCode = E_OK;
441     if (IncUsedCount() == E_OK) {
442         errCode = stateMachine_->ReceiveMessageCallback(inMsg);
443         SafeExit();
444     }
445     return errCode;
446 }
447 
RegOnSyncTask(const std::function<int (void)> & callback)448 void SyncTaskContext::RegOnSyncTask(const std::function<int(void)> &callback)
449 {
450     onSyncTaskAdd_ = callback;
451 }
452 
IncUsedCount()453 int SyncTaskContext::IncUsedCount()
454 {
455     AutoLock lock(this);
456     if (IsKilled()) {
457         LOGI("[SyncTaskContext] IncUsedCount isKilled");
458         return -E_OBJ_IS_KILLED;
459     }
460     usedCount_++;
461     return E_OK;
462 }
463 
SafeExit()464 void SyncTaskContext::SafeExit()
465 {
466     AutoLock lock(this);
467     usedCount_--;
468     if (usedCount_ < 1) {
469         safeKill_.notify_one();
470     }
471 }
472 
GetCurrentLocalTime() const473 Timestamp SyncTaskContext::GetCurrentLocalTime() const
474 {
475     if (timeHelper_ == nullptr) {
476         return TimeHelper::INVALID_TIMESTAMP;
477     }
478     return timeHelper_->GetTime();
479 }
480 
Abort(int status)481 void SyncTaskContext::Abort(int status)
482 {
483     (void)status;
484     Clear();
485 }
486 
CommErrHandlerFunc(int errCode,ISyncTaskContext * context,int32_t sessionId,bool isDirectEnd)487 void SyncTaskContext::CommErrHandlerFunc(int errCode, ISyncTaskContext *context, int32_t sessionId, bool isDirectEnd)
488 {
489     {
490         std::lock_guard<std::mutex> lock(synTaskContextSetLock_);
491         if (synTaskContextSet_.count(context) == 0) {
492             LOGI("[SyncTaskContext][CommErrHandle] context has been killed");
493             return;
494         }
495         // IncObjRef to maker sure context not been killed. after the lock_guard
496         RefObject::IncObjRef(context);
497     }
498 
499     static_cast<SyncTaskContext *>(context)->CommErrHandlerFuncInner(errCode, static_cast<uint32_t>(sessionId),
500         isDirectEnd);
501     RefObject::DecObjRef(context);
502 }
503 
SetRemoteSoftwareVersion(uint32_t version)504 void SyncTaskContext::SetRemoteSoftwareVersion(uint32_t version)
505 {
506     std::lock_guard<std::mutex> lock(remoteSoftwareVersionLock_);
507     remoteSoftwareVersion_ = version;
508     remoteSoftwareVersionId_++;
509 }
510 
GetRemoteSoftwareVersion() const511 uint32_t SyncTaskContext::GetRemoteSoftwareVersion() const
512 {
513     std::lock_guard<std::mutex> lock(remoteSoftwareVersionLock_);
514     return remoteSoftwareVersion_;
515 }
516 
GetRemoteSoftwareVersionId() const517 uint64_t SyncTaskContext::GetRemoteSoftwareVersionId() const
518 {
519     std::lock_guard<std::mutex> lock(remoteSoftwareVersionLock_);
520     return remoteSoftwareVersionId_;
521 }
522 
IsCommNormal() const523 bool SyncTaskContext::IsCommNormal() const
524 {
525     return isCommNormal_;
526 }
527 
CommErrHandlerFuncInner(int errCode,uint32_t sessionId,bool isDirectEnd)528 void SyncTaskContext::CommErrHandlerFuncInner(int errCode, uint32_t sessionId, bool isDirectEnd)
529 {
530     {
531         RefObject::AutoLock lock(this);
532         if ((sessionId != requestSessionId_) || (requestSessionId_ == 0)) {
533             return;
534         }
535 
536         if (errCode == E_OK) {
537             SetCommFailErrCode(errCode);
538             // when communicator sent message failed, the state machine will get the error and exit this sync task
539             // it seems unnecessary to change isCommNormal_ value, so just return here
540             return;
541         }
542     }
543     LOGE("[SyncTaskContext][CommErr] errCode %d, isDirectEnd %d", errCode, static_cast<int>(isDirectEnd));
544     if (!isDirectEnd) {
545         SetErrCodeWhenWaitTimeOut(errCode);
546         return;
547     }
548     if (errCode > 0) {
549         SetCommFailErrCode(static_cast<int>(COMM_FAILURE));
550     } else {
551         SetCommFailErrCode(errCode);
552     }
553     stateMachine_->CommErrAbort(sessionId);
554 }
555 
TimeOut(TimerId id)556 int SyncTaskContext::TimeOut(TimerId id)
557 {
558     if (!timeOutCallback_) {
559         return E_OK;
560     }
561     IncObjRef(this);
562     int errCode = RuntimeContext::GetInstance()->ScheduleTask([this, id]() {
563         timeOutCallback_(id);
564         DecObjRef(this);
565     });
566     if (errCode != E_OK) {
567         LOGW("[SyncTaskContext][TimeOut] Trigger TimeOut Async Failed! TimerId=" PRIu64 " errCode=%d", id, errCode);
568         DecObjRef(this);
569     }
570     return E_OK;
571 }
572 
CopyTargetData(const ISyncTarget * target,const TaskParam & taskParam)573 void SyncTaskContext::CopyTargetData(const ISyncTarget *target, const TaskParam &taskParam)
574 {
575     retryTime_ = 0;
576     mode_ = target->GetMode();
577     status_ = SyncOperation::OP_SYNCING;
578     isNeedRetry_ = SyncTaskContext::NO_NEED_RETRY;
579     taskErrCode_ = E_OK;
580     packetId_ = 0;
581     isCommNormal_ = true; // reset comm status here
582     commErrCode_ = E_OK;
583     syncTaskRetryStatus_ = isSyncRetry_;
584     timeout_ = static_cast<int>(taskParam.timeout);
585     negotiationCount_ = 0;
586     target->GetSyncOperation(syncOperation_);
587     ReSetSequenceId();
588 
589     if (syncOperation_ != nullptr) {
590         // IncRef for syncOperation_ to make sure syncOperation_ is valid, when setStatus
591         RefObject::IncObjRef(syncOperation_);
592         syncId_ = syncOperation_->GetSyncId();
593         isAutoSync_ = syncOperation_->IsAutoSync();
594         isAutoSubscribe_ = syncOperation_->IsAutoControlCmd();
595         if (isAutoSync_ || mode_ == SUBSCRIBE_QUERY || mode_ == UNSUBSCRIBE_QUERY) {
596             syncTaskRetryStatus_ = true;
597         }
598         requestSessionId_ = GenerateRequestSessionId();
599         LOGI("[SyncTaskContext][copyTarget] mode=%d,syncId=%d,isAutoSync=%d,isRetry=%d,dev=%s{private}",
600             mode_, syncId_, isAutoSync_, syncTaskRetryStatus_, deviceId_.c_str());
601         DBDfxAdapter::StartAsyncTrace(syncActionName_, static_cast<int>(syncId_));
602     } else {
603         isAutoSync_ = false;
604         LOGI("[SyncTaskContext][copyTarget] for response data dev %s{private},isRetry=%d", deviceId_.c_str(),
605             syncTaskRetryStatus_);
606     }
607 }
608 
KillWait()609 void SyncTaskContext::KillWait()
610 {
611     StopTimer();
612     UnlockObj();
613     stateMachine_->NotifyClosing();
614     stateMachine_->AbortImmediately();
615     LockObj();
616     LOGW("[SyncTaskContext] Try to kill a context, now wait.");
617     bool noDeadLock = WaitLockedUntil(
618         safeKill_,
619         [this]() {
620             if (usedCount_ < 1) {
621                 return true;
622             }
623             return false;
624         },
625         KILL_WAIT_SECONDS);
626     if (!noDeadLock) { // LCOV_EXCL_BR_LINE
627         LOGE("[SyncTaskContext] Dead lock may happen, we stop waiting the task exit.");
628     } else {
629         LOGW("[SyncTaskContext] Wait the task exit ok.");
630     }
631     std::lock_guard<std::mutex> lock(synTaskContextSetLock_);
632     synTaskContextSet_.erase(this);
633 }
634 
ClearSyncOperation()635 void SyncTaskContext::ClearSyncOperation()
636 {
637     std::lock_guard<std::mutex> lock(operationLock_);
638     if (syncOperation_ != nullptr) {
639         DBDfxAdapter::FinishAsyncTrace(syncActionName_, static_cast<int>(syncId_));
640         RefObject::DecObjRef(syncOperation_);
641         syncOperation_ = nullptr;
642     }
643 }
644 
CancelCurrentSyncRetryIfNeed(int newTargetMode,uint32_t syncId)645 void SyncTaskContext::CancelCurrentSyncRetryIfNeed(int newTargetMode, uint32_t syncId)
646 {
647     AutoLock lock(this);
648     if (!isAutoSync_) {
649         return;
650     }
651     if (syncId_ >= syncId) {
652         return;
653     }
654     int mode = SyncOperation::TransferSyncMode(newTargetMode);
655     if (newTargetMode == mode_ || mode == SyncModeType::PUSH_AND_PULL) {
656         SetRetryTime(AUTO_RETRY_TIMES);
657         ModifyTimer(timeout_);
658     }
659 }
660 
GetTaskErrCode() const661 int SyncTaskContext::GetTaskErrCode() const
662 {
663     return taskErrCode_;
664 }
665 
SetTaskErrCode(int errCode)666 void SyncTaskContext::SetTaskErrCode(int errCode)
667 {
668     taskErrCode_ = errCode;
669 }
670 
IsSyncTaskNeedRetry() const671 bool SyncTaskContext::IsSyncTaskNeedRetry() const
672 {
673     return syncTaskRetryStatus_;
674 }
675 
SetSyncRetry(bool isRetry)676 void SyncTaskContext::SetSyncRetry(bool isRetry)
677 {
678     isSyncRetry_ = isRetry;
679 }
680 
GetSyncRetryTimes() const681 int SyncTaskContext::GetSyncRetryTimes() const
682 {
683     if (IsAutoSync() || mode_ == SUBSCRIBE_QUERY || mode_ == UNSUBSCRIBE_QUERY) {
684         return AUTO_RETRY_TIMES;
685     }
686     return MANUAL_RETRY_TIMES;
687 }
688 
GetSyncRetryTimeout(int retryTime) const689 int SyncTaskContext::GetSyncRetryTimeout(int retryTime) const
690 {
691     int timeoutTime = GetTimeoutTime();
692     if (IsAutoSync()) {
693         // set the new timeout value with 2 raised to the power of retryTime.
694         return timeoutTime * (1u << retryTime);
695     }
696     return timeoutTime;
697 }
698 
ClearAllSyncTask()699 void SyncTaskContext::ClearAllSyncTask()
700 {
701 }
702 
IsAutoLiftWaterMark() const703 bool SyncTaskContext::IsAutoLiftWaterMark() const
704 {
705     return negotiationCount_ < NEGOTIATION_LIMIT;
706 }
707 
IncNegotiationCount()708 void SyncTaskContext::IncNegotiationCount()
709 {
710     negotiationCount_++;
711 }
712 
IsNeedTriggerQueryAutoSync(Message * inMsg,QuerySyncObject & query)713 bool SyncTaskContext::IsNeedTriggerQueryAutoSync(Message *inMsg, QuerySyncObject &query)
714 {
715     return stateMachine_->IsNeedTriggerQueryAutoSync(inMsg, query);
716 }
717 
IsAutoSubscribe() const718 bool SyncTaskContext::IsAutoSubscribe() const
719 {
720     return isAutoSubscribe_;
721 }
722 
IsCurrentSyncTaskCanBeSkipped() const723 bool SyncTaskContext::IsCurrentSyncTaskCanBeSkipped() const
724 {
725     return false;
726 }
727 
ResetLastPushTaskStatus()728 void SyncTaskContext::ResetLastPushTaskStatus()
729 {
730 }
731 
SchemaChange()732 void SyncTaskContext::SchemaChange()
733 {
734     if (stateMachine_ != nullptr) {
735         stateMachine_->SchemaChange();
736     }
737 }
738 
Dump(int fd)739 void SyncTaskContext::Dump(int fd)
740 {
741     size_t totalSyncTaskCount = 0u;
742     size_t autoSyncTaskCount = 0u;
743     size_t reponseTaskCount = 0u;
744     {
745         std::lock_guard<std::mutex> lock(targetQueueLock_);
746         totalSyncTaskCount = requestTargetQueue_.size() + responseTargetQueue_.size();
747         for (const auto &target : requestTargetQueue_) {
748             if (target->IsAutoSync()) { // LCOV_EXCL_BR_LINE
749                 autoSyncTaskCount++;
750             }
751         }
752         reponseTaskCount = responseTargetQueue_.size();
753     }
754     DBDumpHelper::Dump(fd, "\t\ttarget = %s, total sync task count = %zu, auto sync task count = %zu,"
755         " response task count = %zu\n",
756         deviceId_.c_str(), totalSyncTaskCount, autoSyncTaskCount, reponseTaskCount);
757 }
758 
RunPermissionCheck(uint8_t flag) const759 int SyncTaskContext::RunPermissionCheck(uint8_t flag) const
760 {
761     std::string appId = syncInterface_->GetDbProperties().GetStringProp(DBProperties::APP_ID, "");
762     std::string userId = syncInterface_->GetDbProperties().GetStringProp(DBProperties::USER_ID, "");
763     std::string storeId = syncInterface_->GetDbProperties().GetStringProp(DBProperties::STORE_ID, "");
764     int32_t instanceId = syncInterface_->GetDbProperties().GetIntProp(DBProperties::INSTANCE_ID, 0);
765     int errCode = RuntimeContext::GetInstance()->RunPermissionCheck(
766         { userId, appId, storeId, deviceId_, instanceId }, flag);
767     if (errCode != E_OK) {
768         LOGE("[SyncTaskContext] RunPermissionCheck not pass errCode:%d, flag:%d, %s{private}",
769             errCode, flag, deviceId_.c_str());
770     }
771     return errCode;
772 }
773 
GetPermissionCheckFlag(bool isAutoSync,int syncMode)774 uint8_t SyncTaskContext::GetPermissionCheckFlag(bool isAutoSync, int syncMode)
775 {
776     uint8_t flag = 0;
777     int mode = SyncOperation::TransferSyncMode(syncMode);
778     if (mode == SyncModeType::PUSH || mode == SyncModeType::RESPONSE_PULL) {
779         flag = CHECK_FLAG_SEND;
780     } else if (mode == SyncModeType::PULL) {
781         flag = CHECK_FLAG_RECEIVE;
782     } else if (mode == SyncModeType::PUSH_AND_PULL) {
783         flag = CHECK_FLAG_SEND | CHECK_FLAG_RECEIVE;
784     }
785     if (isAutoSync) {
786         flag = flag | CHECK_FLAG_AUTOSYNC;
787     }
788     if (mode != SyncModeType::RESPONSE_PULL) {
789         // it means this sync is started by local
790         flag = flag | CHECK_FLAG_SPONSOR;
791     }
792     return flag;
793 }
794 
AbortMachineIfNeed(uint32_t syncId)795 void SyncTaskContext::AbortMachineIfNeed(uint32_t syncId)
796 {
797     uint32_t sessionId = 0u;
798     {
799         RefObject::AutoLock autoLock(this);
800         if (syncId_ != syncId) {
801             return;
802         }
803         sessionId = requestSessionId_;
804     }
805     stateMachine_->InnerErrorAbort(sessionId);
806 }
807 
GetAndIncSyncOperation() const808 SyncOperation *SyncTaskContext::GetAndIncSyncOperation() const
809 {
810     std::lock_guard<std::mutex> lock(operationLock_);
811     if (syncOperation_ == nullptr) {
812         return nullptr;
813     }
814     RefObject::IncObjRef(syncOperation_);
815     return syncOperation_;
816 }
817 
GenerateRequestSessionId()818 uint32_t SyncTaskContext::GenerateRequestSessionId()
819 {
820     uint32_t sessionId = lastRequestSessionId_ != 0 ? lastRequestSessionId_ + 1 : 0;
821     // make sure sessionId is between 0x01 and 0x8fffffff
822     if (sessionId > SESSION_ID_MAX_VALUE || sessionId == 0) {
823         sessionId = Hash::Hash32Func(deviceId_ + std::to_string(syncId_) +
824             std::to_string(TimeHelper::GetSysCurrentTime()));
825     }
826     lastRequestSessionId_ = sessionId;
827     return sessionId;
828 }
829 
IsSchemaCompatible() const830 bool SyncTaskContext::IsSchemaCompatible() const
831 {
832     return true;
833 }
834 
835 void SyncTaskContext::SetDbAbility([[gnu::unused]] DbAbility &remoteDbAbility)
836 {
837 }
838 
TimeChange()839 void SyncTaskContext::TimeChange()
840 {
841     if (stateMachine_ == nullptr) {
842         LOGW("[SyncTaskContext] machine is null when time change");
843         return;
844     }
845     stateMachine_->TimeChange();
846 }
847 
GetResponseTaskCount()848 int32_t SyncTaskContext::GetResponseTaskCount()
849 {
850     std::lock_guard<std::mutex> autoLock(targetQueueLock_);
851     return static_cast<int32_t>(responseTargetQueue_.size());
852 }
853 
GetCommErrCode() const854 int SyncTaskContext::GetCommErrCode() const
855 {
856     return commErrCode_;
857 }
858 
SetCommFailErrCode(int errCode)859 void SyncTaskContext::SetCommFailErrCode(int errCode)
860 {
861     commErrCode_ = errCode;
862 }
863 
SetErrCodeWhenWaitTimeOut(int errCode)864 void SyncTaskContext::SetErrCodeWhenWaitTimeOut(int errCode)
865 {
866     if (errCode > 0) {
867         SetCommFailErrCode(static_cast<int>(TIME_OUT));
868     } else {
869         SetCommFailErrCode(errCode);
870     }
871 }
872 } // namespace DistributedDB
873