1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "single_ver_sync_task_context.h"
17
18 #include <algorithm>
19 #include "db_common.h"
20 #include "db_dfx_adapter.h"
21 #include "db_errno.h"
22 #include "isyncer.h"
23 #include "log_print.h"
24 #include "single_ver_sync_state_machine.h"
25 #include "single_ver_sync_target.h"
26 #include "sync_types.h"
27
28 namespace DistributedDB {
SingleVerSyncTaskContext()29 SingleVerSyncTaskContext::SingleVerSyncTaskContext()
30 : SyncTaskContext(),
31 token_(nullptr),
32 endMark_(0),
33 needClearRemoteStaleData_(false)
34 {}
35
~SingleVerSyncTaskContext()36 SingleVerSyncTaskContext::~SingleVerSyncTaskContext()
37 {
38 token_ = nullptr;
39 subManager_ = nullptr;
40 }
41
Initialize(const std::string & deviceId,ISyncInterface * syncInterface,std::shared_ptr<Metadata> & metadata,ICommunicator * communicator)42 int SingleVerSyncTaskContext::Initialize(const std::string &deviceId,
43 ISyncInterface *syncInterface, std::shared_ptr<Metadata> &metadata, ICommunicator *communicator)
44 {
45 if (deviceId.empty() || syncInterface == nullptr || metadata == nullptr ||
46 communicator == nullptr) {
47 return -E_INVALID_ARGS;
48 }
49 stateMachine_ = new (std::nothrow) SingleVerSyncStateMachine;
50 if (stateMachine_ == nullptr) {
51 return -E_OUT_OF_MEMORY;
52 }
53 deviceId_ = deviceId;
54 std::vector<uint8_t> dbIdentifier = syncInterface->GetIdentifier();
55 dbIdentifier.resize(3); // only show 3 bytes
56 syncActionName_ = DBDfxAdapter::SYNC_ACTION + "_" +
57 DBCommon::VectorToHexString(dbIdentifier) + "_" + deviceId_.c_str();
58 TimerAction timeOutCallback;
59 int errCode = stateMachine_->Initialize(this, syncInterface, metadata, communicator);
60 if (errCode != E_OK) {
61 LOGE("[SingleVerSyncTaskContext] stateMachine Initialize failed, err %d.", errCode);
62 goto ERROR_OUT;
63 }
64
65 timeHelper_ = std::make_unique<TimeHelper>();
66 errCode = timeHelper_->Initialize(syncInterface, metadata);
67 if (errCode != E_OK) {
68 LOGE("[SingleVerSyncTaskContext] timeHelper Initialize failed, err %d.", errCode);
69 goto ERROR_OUT;
70 }
71 timeOutCallback = std::bind(&SyncStateMachine::TimeoutCallback,
72 static_cast<SingleVerSyncStateMachine *>(stateMachine_),
73 std::placeholders::_1);
74 SetTimeoutCallback(timeOutCallback);
75
76 syncInterface_ = syncInterface;
77 communicator_ = communicator;
78 taskExecStatus_ = INIT;
79 OnKill([this]() { this->KillWait(); });
80 {
81 std::lock_guard<std::mutex> lock(synTaskContextSetLock_);
82 synTaskContextSet_.insert(this);
83 }
84 return errCode;
85
86 ERROR_OUT:
87 delete stateMachine_;
88 stateMachine_ = nullptr;
89 return errCode;
90 }
91
AddSyncOperation(SyncOperation * operation)92 int SingleVerSyncTaskContext::AddSyncOperation(SyncOperation *operation)
93 {
94 if (operation == nullptr) {
95 return -E_INVALID_ARGS;
96 }
97
98 // If auto sync, just update the end watermark
99 if (operation->IsAutoSync()) {
100 std::lock_guard<std::mutex> lock(targetQueueLock_);
101 bool isQuerySync = operation->IsQuerySync();
102 std::string queryId = operation->GetQueryId();
103 auto iter = std::find_if(requestTargetQueue_.begin(), requestTargetQueue_.end(),
104 [isQuerySync, queryId](const ISyncTarget *target) {
105 if (target == nullptr) {
106 return false;
107 }
108 if (isQuerySync) {
109 SyncOperation *tmpOperation = nullptr;
110 target->GetSyncOperation(tmpOperation);
111 return (tmpOperation != nullptr && tmpOperation->GetQueryId() == queryId) && target->IsAutoSync();
112 }
113 return target->IsAutoSync();
114 });
115 if (iter != requestTargetQueue_.end()) {
116 static_cast<SingleVerSyncTarget *>(*iter)->SetEndWaterMark(timeHelper_->GetTime());
117 operation->SetStatus(deviceId_, SyncOperation::OP_FINISHED_ALL);
118 return E_OK;
119 }
120 }
121
122 auto *newTarget = new (std::nothrow) SingleVerSyncTarget;
123 if (newTarget == nullptr) {
124 return -E_OUT_OF_MEMORY;
125 }
126 newTarget->SetSyncOperation(operation);
127 Timestamp timstamp = timeHelper_->GetTime();
128 newTarget->SetEndWaterMark(timstamp);
129 newTarget->SetTaskType(ISyncTarget::REQUEST);
130 AddSyncTarget(newTarget);
131 return E_OK;
132 }
133
SetEndMark(WaterMark endMark)134 void SingleVerSyncTaskContext::SetEndMark(WaterMark endMark)
135 {
136 endMark_ = endMark;
137 }
138
GetEndMark() const139 WaterMark SingleVerSyncTaskContext::GetEndMark() const
140 {
141 return endMark_;
142 }
143
GetContinueToken(ContinueToken & outToken) const144 void SingleVerSyncTaskContext::GetContinueToken(ContinueToken &outToken) const
145 {
146 outToken = token_;
147 }
148
SetContinueToken(ContinueToken token)149 void SingleVerSyncTaskContext::SetContinueToken(ContinueToken token)
150 {
151 token_ = token;
152 return;
153 }
154
ReleaseContinueToken()155 void SingleVerSyncTaskContext::ReleaseContinueToken()
156 {
157 if (token_ != nullptr) {
158 static_cast<SyncGenericInterface *>(syncInterface_)->ReleaseContinueToken(token_);
159 token_ = nullptr;
160 }
161 }
162
PopResponseTarget(SingleVerSyncTarget & target)163 int SingleVerSyncTaskContext::PopResponseTarget(SingleVerSyncTarget &target)
164 {
165 std::lock_guard<std::mutex> lock(targetQueueLock_);
166 LOGD("[SingleVerSyncTaskContext] GetFrontExtWaterMarak size = %zu", responseTargetQueue_.size());
167 if (!responseTargetQueue_.empty()) {
168 ISyncTarget *tmpTarget = responseTargetQueue_.front();
169 responseTargetQueue_.pop_front();
170 target = *(static_cast<SingleVerSyncTarget *>(tmpTarget));
171 delete tmpTarget;
172 tmpTarget = nullptr;
173 return E_OK;
174 }
175 return -E_LENGTH_ERROR;
176 }
177
GetRspTargetQueueSize() const178 int SingleVerSyncTaskContext::GetRspTargetQueueSize() const
179 {
180 std::lock_guard<std::mutex> lock(targetQueueLock_);
181 return responseTargetQueue_.size();
182 }
183
SetResponseSessionId(uint32_t responseSessionId)184 void SingleVerSyncTaskContext::SetResponseSessionId(uint32_t responseSessionId)
185 {
186 responseSessionId_ = responseSessionId;
187 }
188
GetResponseSessionId() const189 uint32_t SingleVerSyncTaskContext::GetResponseSessionId() const
190 {
191 return responseSessionId_;
192 }
193
CopyTargetData(const ISyncTarget * target,const TaskParam & taskParam)194 void SingleVerSyncTaskContext::CopyTargetData(const ISyncTarget *target, const TaskParam &taskParam)
195 {
196 const SingleVerSyncTarget *targetTmp = static_cast<const SingleVerSyncTarget *>(target);
197 SyncTaskContext::CopyTargetData(target, taskParam);
198 mode_ = targetTmp->GetMode();
199 endMark_ = targetTmp->GetEndWaterMark();
200 if (mode_ == SyncModeType::RESPONSE_PULL) {
201 responseSessionId_ = targetTmp->GetResponseSessionId();
202 }
203 query_ = targetTmp->GetQuery();
204 isQuerySync_ = targetTmp->IsQuerySync();
205 }
206
Clear()207 void SingleVerSyncTaskContext::Clear()
208 {
209 retryTime_ = 0;
210 ClearSyncOperation();
211 SyncTaskContext::Clear();
212 SetMode(SyncModeType::INVALID_MODE);
213 syncId_ = 0;
214 isAutoSync_ = false;
215 SetOperationStatus(SyncOperation::OP_WAITING);
216 SetEndMark(0);
217 SetResponseSessionId(0);
218 query_ = QuerySyncObject();
219 isQuerySync_ = false;
220 }
221
Abort(int status)222 void SingleVerSyncTaskContext::Abort(int status)
223 {
224 {
225 std::lock_guard<std::mutex> lock(operationLock_);
226 if (syncOperation_ != nullptr) {
227 syncOperation_->SetStatus(deviceId_, status);
228 if ((status >= SyncOperation::OP_FINISHED_ALL)) {
229 UnlockObj();
230 if (syncOperation_->CheckIsAllFinished()) {
231 syncOperation_->Finished();
232 }
233 LockObj();
234 }
235 }
236 }
237 StopFeedDogForSync(SyncDirectionFlag::SEND);
238 StopFeedDogForSync(SyncDirectionFlag::RECEIVE);
239 Clear();
240 }
241
ClearAllSyncTask()242 void SingleVerSyncTaskContext::ClearAllSyncTask()
243 {
244 // clear request queue sync task and responsequeue first.
245 std::list<ISyncTarget *> targetQueue;
246 {
247 std::lock_guard<std::mutex> lock(targetQueueLock_);
248 LOGI("[SingleVerSyncTaskContext] request taskcount=%zu, responsecount=%zu", requestTargetQueue_.size(),
249 responseTargetQueue_.size());
250 while (!requestTargetQueue_.empty()) {
251 ISyncTarget *tmpTarget = nullptr;
252 tmpTarget = requestTargetQueue_.front();
253 requestTargetQueue_.pop_front();
254 targetQueue.push_back(tmpTarget);
255 }
256 while (!responseTargetQueue_.empty()) {
257 ISyncTarget *tmpTarget = nullptr;
258 tmpTarget = responseTargetQueue_.front();
259 responseTargetQueue_.pop_front();
260 delete tmpTarget;
261 tmpTarget = nullptr;
262 }
263 }
264 while (!targetQueue.empty()) {
265 ISyncTarget *target = nullptr;
266 target = targetQueue.front();
267 targetQueue.pop_front();
268 SyncOperation *tmpOperation = nullptr;
269 target->GetSyncOperation(tmpOperation);
270 if (tmpOperation == nullptr) {
271 LOGE("[ClearAllSyncTask] tmpOperation is nullptr");
272 continue; // not exit this scene
273 }
274 LOGI("[SingleVerSyncTaskContext] killing syncId=%d,dev=%s", tmpOperation->GetSyncId(), STR_MASK(deviceId_));
275 if (target->IsAutoSync()) {
276 tmpOperation->SetStatus(deviceId_, SyncOperation::OP_FINISHED_ALL);
277 } else {
278 tmpOperation->SetStatus(deviceId_, SyncOperation::OP_COMM_ABNORMAL);
279 }
280 if (tmpOperation->CheckIsAllFinished()) {
281 tmpOperation->Finished();
282 }
283 delete target;
284 target = nullptr;
285 }
286 if (GetTaskExecStatus() == SyncTaskContext::RUNNING) {
287 // clear syncing task.
288 stateMachine_->CommErrAbort();
289 }
290 // reset last push status for sync merge
291 ResetLastPushTaskStatus();
292 }
293
EnableClearRemoteStaleData(bool enable)294 void SingleVerSyncTaskContext::EnableClearRemoteStaleData(bool enable)
295 {
296 needClearRemoteStaleData_ = enable;
297 }
298
IsNeedClearRemoteStaleData() const299 bool SingleVerSyncTaskContext::IsNeedClearRemoteStaleData() const
300 {
301 return needClearRemoteStaleData_;
302 }
303
StartFeedDogForSync(uint32_t time,SyncDirectionFlag flag)304 bool SingleVerSyncTaskContext::StartFeedDogForSync(uint32_t time, SyncDirectionFlag flag)
305 {
306 return stateMachine_->StartFeedDogForSync(time, flag);
307 }
308
StopFeedDogForSync(SyncDirectionFlag flag)309 void SingleVerSyncTaskContext::StopFeedDogForSync(SyncDirectionFlag flag)
310 {
311 stateMachine_->StopFeedDogForSync(flag);
312 }
313
HandleDataRequestRecv(const Message * msg)314 int SingleVerSyncTaskContext::HandleDataRequestRecv(const Message *msg)
315 {
316 return static_cast<SingleVerSyncStateMachine *>(stateMachine_)->HandleDataRequestRecv(msg);
317 }
318
IsReceiveWaterMarkErr() const319 bool SingleVerSyncTaskContext::IsReceiveWaterMarkErr() const
320 {
321 return isReceiveWaterMarkErr_;
322 }
323
SetReceiveWaterMarkErr(bool isErr)324 void SingleVerSyncTaskContext::SetReceiveWaterMarkErr(bool isErr)
325 {
326 isReceiveWaterMarkErr_ = isErr;
327 }
328
SetRemoteSeccurityOption(SecurityOption secOption)329 void SingleVerSyncTaskContext::SetRemoteSeccurityOption(SecurityOption secOption)
330 {
331 remoteSecOption_ = secOption;
332 }
333
GetRemoteSeccurityOption() const334 SecurityOption SingleVerSyncTaskContext::GetRemoteSeccurityOption() const
335 {
336 return remoteSecOption_;
337 }
338
SetReceivcPermitCheck(bool isChecked)339 void SingleVerSyncTaskContext::SetReceivcPermitCheck(bool isChecked)
340 {
341 isReceivcPermitChecked_ = isChecked;
342 }
343
GetReceivcPermitCheck() const344 bool SingleVerSyncTaskContext::GetReceivcPermitCheck() const
345 {
346 return isReceivcPermitChecked_;
347 }
348
SetSendPermitCheck(bool isChecked)349 void SingleVerSyncTaskContext::SetSendPermitCheck(bool isChecked)
350 {
351 isSendPermitChecked_ = isChecked;
352 }
353
GetSendPermitCheck() const354 bool SingleVerSyncTaskContext::GetSendPermitCheck() const
355 {
356 return isSendPermitChecked_;
357 }
358
SetIsSchemaSync(bool isSchemaSync)359 void SingleVerSyncTaskContext::SetIsSchemaSync(bool isSchemaSync)
360 {
361 isSchemaSync_ = isSchemaSync;
362 }
363
GetIsSchemaSync() const364 bool SingleVerSyncTaskContext::GetIsSchemaSync() const
365 {
366 return isSchemaSync_;
367 }
368
IsSkipTimeoutError(int errCode) const369 bool SingleVerSyncTaskContext::IsSkipTimeoutError(int errCode) const
370 {
371 if (errCode == -E_TIMEOUT && IsSyncTaskNeedRetry() && (GetRetryTime() < GetSyncRetryTimes())) {
372 LOGE("[SingleVerSyncTaskContext] send message timeout error occurred");
373 return true;
374 } else {
375 return false;
376 }
377 }
378
FindResponseSyncTarget(uint32_t responseSessionId) const379 bool SingleVerSyncTaskContext::FindResponseSyncTarget(uint32_t responseSessionId) const
380 {
381 std::lock_guard<std::mutex> lock(targetQueueLock_);
382 auto iter = std::find_if(responseTargetQueue_.begin(), responseTargetQueue_.end(),
383 [responseSessionId](const ISyncTarget *target) {
384 return target->GetResponseSessionId() == responseSessionId;
385 });
386 if (iter == responseTargetQueue_.end()) {
387 return false;
388 }
389 return true;
390 }
391
SetQuery(const QuerySyncObject & query)392 void SingleVerSyncTaskContext::SetQuery(const QuerySyncObject &query)
393 {
394 query_ = query;
395 }
396
GetQuery() const397 const QuerySyncObject &SingleVerSyncTaskContext::GetQuery() const
398 {
399 return query_;
400 }
401
SetQuerySync(bool isQuerySync)402 void SingleVerSyncTaskContext::SetQuerySync(bool isQuerySync)
403 {
404 isQuerySync_ = isQuerySync;
405 }
406
IsQuerySync() const407 bool SingleVerSyncTaskContext::IsQuerySync() const
408 {
409 return isQuerySync_;
410 }
411
GetRemoteCompressAlgo() const412 std::set<CompressAlgorithm> SingleVerSyncTaskContext::GetRemoteCompressAlgo() const
413 {
414 std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
415 std::set<CompressAlgorithm> compressAlgoSet;
416 for (const auto &algo : SyncConfig::COMPRESSALGOMAP) {
417 if (remoteDbAbility_.GetAbilityItem(algo.second) == SUPPORT_MARK) {
418 compressAlgoSet.insert(static_cast<CompressAlgorithm>(algo.first));
419 }
420 }
421 return compressAlgoSet;
422 }
423
GetRemoteCompressAlgoStr() const424 std::string SingleVerSyncTaskContext::GetRemoteCompressAlgoStr() const
425 {
426 static std::map<CompressAlgorithm, std::string> algoMap = {{CompressAlgorithm::ZLIB, "zlib"}};
427 std::set<CompressAlgorithm> remoteCompressAlgoSet = GetRemoteCompressAlgo();
428 if (remoteCompressAlgoSet.size() == 0) {
429 return "none";
430 }
431 std::string currentAlgoStr;
432 for (const auto &algo : remoteCompressAlgoSet) {
433 auto iter = algoMap.find(algo);
434 if (iter != algoMap.end()) {
435 currentAlgoStr += algoMap[algo] + ",";
436 }
437 }
438 return currentAlgoStr.substr(0, currentAlgoStr.length() - 1);
439 }
440
SetDbAbility(DbAbility & remoteDbAbility)441 void SingleVerSyncTaskContext::SetDbAbility(DbAbility &remoteDbAbility)
442 {
443 {
444 std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
445 remoteDbAbility_ = remoteDbAbility;
446 }
447 LOGI("[SingleVerSyncTaskContext] set dev=%s compressAlgo=%s, IsSupAllPredicateQuery=%u,"
448 "IsSupSubscribeQuery=%u, inKeys=%u",
449 STR_MASK(GetDeviceId()), GetRemoteCompressAlgoStr().c_str(),
450 remoteDbAbility.GetAbilityItem(SyncConfig::ALLPREDICATEQUERY),
451 remoteDbAbility.GetAbilityItem(SyncConfig::SUBSCRIBEQUERY),
452 remoteDbAbility.GetAbilityItem(SyncConfig::INKEYS_QUERY));
453 }
454
ChooseCompressAlgo() const455 CompressAlgorithm SingleVerSyncTaskContext::ChooseCompressAlgo() const
456 {
457 std::set<CompressAlgorithm> remoteAlgo = GetRemoteCompressAlgo();
458 if (remoteAlgo.size() == 0) {
459 return CompressAlgorithm::NONE;
460 }
461 std::set<CompressAlgorithm> localAlgorithmSet;
462 (void)(static_cast<SyncGenericInterface *>(syncInterface_))->GetCompressionAlgo(localAlgorithmSet);
463 std::set<CompressAlgorithm> algoIntersection;
464 set_intersection(remoteAlgo.begin(), remoteAlgo.end(), localAlgorithmSet.begin(), localAlgorithmSet.end(),
465 inserter(algoIntersection, algoIntersection.begin()));
466 if (algoIntersection.size() == 0) {
467 return CompressAlgorithm::NONE;
468 }
469 return *(algoIntersection.begin());
470 }
471
IsNotSupportAbility(const AbilityItem & abilityItem) const472 bool SingleVerSyncTaskContext::IsNotSupportAbility(const AbilityItem &abilityItem) const
473 {
474 std::lock_guard<std::mutex> autoLock(remoteDbAbilityLock_);
475 return remoteDbAbility_.GetAbilityItem(abilityItem) != SUPPORT_MARK;
476 }
477
SetSubscribeManager(std::shared_ptr<SubscribeManager> & subManager)478 void SingleVerSyncTaskContext::SetSubscribeManager(std::shared_ptr<SubscribeManager> &subManager)
479 {
480 subManager_ = subManager;
481 }
482
GetSubscribeManager() const483 std::shared_ptr<SubscribeManager> SingleVerSyncTaskContext::GetSubscribeManager() const
484 {
485 return subManager_;
486 }
DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncTaskContext)487 DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncTaskContext)
488
489 bool SingleVerSyncTaskContext::IsCurrentSyncTaskCanBeSkipped() const
490 {
491 SyncOperation *operation = GetAndIncSyncOperation();
492 bool res = IsCurrentSyncTaskCanBeSkippedInner(operation);
493 RefObject::DecObjRef(operation);
494 return res;
495 }
496
SaveLastPushTaskExecStatus(int finalStatus)497 void SingleVerSyncTaskContext::SaveLastPushTaskExecStatus(int finalStatus)
498 {
499 if (IsTargetQueueEmpty()) {
500 LOGD("sync que is empty, reset last push status");
501 ResetLastPushTaskStatus();
502 return;
503 }
504 if (mode_ == SyncModeType::PUSH || mode_ == SyncModeType::PUSH_AND_PULL || mode_ == SyncModeType::RESPONSE_PULL) {
505 lastFullSyncTaskStatus_ = finalStatus;
506 } else if (mode_ == SyncModeType::QUERY_PUSH || mode_ == SyncModeType::QUERY_PUSH_PULL) {
507 std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
508 lastQuerySyncTaskStatusMap_[syncOperation_->GetQueryId()] = finalStatus;
509 }
510 }
511
GetCorrectedSendWaterMarkForCurrentTask(const SyncOperation * operation,uint64_t & waterMark) const512 int SingleVerSyncTaskContext::GetCorrectedSendWaterMarkForCurrentTask(const SyncOperation *operation,
513 uint64_t &waterMark) const
514 {
515 if (operation != nullptr && operation->IsQuerySync()) {
516 LOGD("Is QuerySync");
517 int errCode = static_cast<SingleVerSyncStateMachine *>(stateMachine_)->GetSendQueryWaterMark(
518 operation->GetQueryId(), deviceId_,
519 lastFullSyncTaskStatus_ == SyncOperation::OP_FINISHED_ALL, waterMark);
520 if (errCode != E_OK) {
521 return errCode;
522 }
523 } else {
524 LOGD("Not QuerySync");
525 static_cast<SingleVerSyncStateMachine *>(stateMachine_)->GetLocalWaterMark(deviceId_, waterMark);
526 }
527 return E_OK;
528 }
529
ResetLastPushTaskStatus()530 void SingleVerSyncTaskContext::ResetLastPushTaskStatus()
531 {
532 lastFullSyncTaskStatus_ = SyncOperation::OP_WAITING;
533 std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
534 lastQuerySyncTaskStatusMap_.clear();
535 }
536
SetCommNormal(bool isCommNormal)537 void SingleVerSyncTaskContext::SetCommNormal(bool isCommNormal)
538 {
539 isCommNormal_ = isCommNormal;
540 }
541
IsCurrentSyncTaskCanBeSkippedInner(const SyncOperation * operation) const542 bool SingleVerSyncTaskContext::IsCurrentSyncTaskCanBeSkippedInner(const SyncOperation *operation) const
543 {
544 if (mode_ == SyncModeType::PUSH) {
545 if (lastFullSyncTaskStatus_ != SyncOperation::OP_FINISHED_ALL) {
546 return false;
547 }
548 if (operation == nullptr) {
549 return true;
550 }
551 } else if (mode_ == SyncModeType::QUERY_PUSH) {
552 if (operation == nullptr) {
553 return true;
554 }
555 std::lock_guard<std::mutex> autoLock(queryTaskStatusMutex_);
556 auto it = lastQuerySyncTaskStatusMap_.find(operation->GetQueryId());
557 if (it == lastQuerySyncTaskStatusMap_.end()) {
558 // no last query_push and push
559 if (lastFullSyncTaskStatus_ != SyncOperation::OP_FINISHED_ALL) {
560 LOGD("no prev query push or successful prev push");
561 return false;
562 }
563 } else {
564 if (it->second != SyncOperation::OP_FINISHED_ALL) {
565 LOGD("last query push status = %d.", it->second);
566 return false;
567 }
568 }
569 } else {
570 return false;
571 }
572
573 Timestamp maxTimestampInDb;
574 syncInterface_->GetMaxTimestamp(maxTimestampInDb);
575 uint64_t localWaterMark = 0;
576 int errCode = GetCorrectedSendWaterMarkForCurrentTask(operation, localWaterMark);
577 if (errCode != E_OK) {
578 LOGE("GetLocalWaterMark in state machine failed: %d", errCode);
579 return false;
580 }
581 if (localWaterMark > maxTimestampInDb) {
582 LOGI("skip current push task, deviceId_ = %s, localWaterMark = %" PRIu64 ", maxTimestampInDb = %" PRIu64,
583 STR_MASK(deviceId_), localWaterMark, maxTimestampInDb);
584 return true;
585 }
586 return false;
587 }
588
StartFeedDogForGetData(uint32_t sessionId)589 void SingleVerSyncTaskContext::StartFeedDogForGetData(uint32_t sessionId)
590 {
591 stateMachine_->StartFeedDogForGetData(sessionId);
592 }
593
StopFeedDogForGetData()594 void SingleVerSyncTaskContext::StopFeedDogForGetData()
595 {
596 stateMachine_->StopFeedDogForGetData();
597 }
598 } // namespace DistributedDB
599