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_state_machine.h"
17
18 #include <cmath>
19 #include <climits>
20 #include <algorithm>
21
22 #include "db_errno.h"
23 #include "log_print.h"
24 #include "sync_operation.h"
25 #include "message_transform.h"
26 #include "sync_types.h"
27 #include "db_common.h"
28 #include "runtime_context.h"
29 #include "performance_analysis.h"
30 #include "single_ver_sync_target.h"
31 #include "single_ver_data_sync.h"
32 #include "single_ver_data_sync_utils.h"
33
34 namespace DistributedDB {
35 using Event = SingleVerSyncStateMachine::Event;
36 using State = SingleVerSyncStateMachine::State;
37 namespace {
38 // used for state switch table
39 const int CURRENT_STATE_INDEX = 0;
40 const int EVENT_INDEX = 1;
41 const int OUTPUT_STATE_INDEX = 2;
42
43 // drop v1 and v2 table by one optimize, dataSend mode in all version go with slide window mode.
44 // State switch table v3, has three columns, CurrentState, Event, and OutSate
45 const std::vector<std::vector<uint8_t>> STATE_SWITCH_TABLE_V3 = {
46 {State::IDLE, Event::START_SYNC_EVENT, State::TIME_SYNC},
47
48 // In TIME_SYNC state
49 {State::TIME_SYNC, Event::TIME_SYNC_FINISHED_EVENT, State::ABILITY_SYNC},
50 {State::TIME_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
51 {State::TIME_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
52
53 // In ABILITY_SYNC state, compare version num and schema
54 {State::ABILITY_SYNC, Event::VERSION_NOT_SUPPOR_EVENT, State::INNER_ERR},
55 {State::ABILITY_SYNC, Event::ABILITY_SYNC_FINISHED_EVENT, State::START_INITIACTIVE_DATA_SYNC},
56 {State::ABILITY_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
57 {State::ABILITY_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
58 {State::ABILITY_SYNC, Event::CONTROL_CMD_EVENT, State::SYNC_CONTROL_CMD},
59
60 // In START_INITIACTIVE_DATA_SYNC state, send a sync request, and send first packt of data sync
61 {State::START_INITIACTIVE_DATA_SYNC, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
62 {State::START_INITIACTIVE_DATA_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
63 {State::START_INITIACTIVE_DATA_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
64 {State::START_INITIACTIVE_DATA_SYNC, Event::SEND_FINISHED_EVENT, State::START_PASSIVE_DATA_SYNC},
65 {State::START_INITIACTIVE_DATA_SYNC, Event::RE_SEND_DATA_EVENT, State::START_INITIACTIVE_DATA_SYNC},
66 {State::START_INITIACTIVE_DATA_SYNC, Event::NEED_TIME_SYNC_EVENT, State::TIME_SYNC},
67
68 // In START_PASSIVE_DATA_SYNC state, do response pull request, and send first packt of data sync
69 {State::START_PASSIVE_DATA_SYNC, Event::SEND_FINISHED_EVENT, State::START_PASSIVE_DATA_SYNC},
70 {State::START_PASSIVE_DATA_SYNC, Event::RESPONSE_TASK_FINISHED_EVENT, State::WAIT_FOR_RECEIVE_DATA_FINISH},
71 {State::START_PASSIVE_DATA_SYNC, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
72 {State::START_PASSIVE_DATA_SYNC, Event::INNER_ERR_EVENT, State::INNER_ERR},
73 {State::START_PASSIVE_DATA_SYNC, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
74 {State::START_PASSIVE_DATA_SYNC, Event::RE_SEND_DATA_EVENT, State::START_PASSIVE_DATA_SYNC},
75 {State::START_PASSIVE_DATA_SYNC, Event::NEED_TIME_SYNC_EVENT, State::TIME_SYNC},
76
77 // In WAIT_FOR_RECEIVE_DATA_FINISH,
78 {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::RECV_FINISHED_EVENT, State::SYNC_TASK_FINISHED},
79 {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::START_PULL_RESPONSE_EVENT, State::START_PASSIVE_DATA_SYNC},
80 {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
81 {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::INNER_ERR_EVENT, State::INNER_ERR},
82 {State::WAIT_FOR_RECEIVE_DATA_FINISH, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
83
84 {State::SYNC_CONTROL_CMD, Event::SEND_FINISHED_EVENT, State::SYNC_TASK_FINISHED},
85 {State::SYNC_CONTROL_CMD, Event::TIME_OUT_EVENT, State::SYNC_TIME_OUT},
86 {State::SYNC_CONTROL_CMD, Event::INNER_ERR_EVENT, State::INNER_ERR},
87 {State::SYNC_CONTROL_CMD, Event::NEED_ABILITY_SYNC_EVENT, State::ABILITY_SYNC},
88
89 // In SYNC_TASK_FINISHED,
90 {State::SYNC_TASK_FINISHED, Event::ALL_TASK_FINISHED_EVENT, State::IDLE},
91 {State::SYNC_TASK_FINISHED, Event::START_SYNC_EVENT, State::TIME_SYNC},
92
93 // SYNC_TIME_OUT and INNE_ERR state, just do some exception resolve
94 {State::SYNC_TIME_OUT, Event::ANY_EVENT, State::SYNC_TASK_FINISHED},
95 {State::INNER_ERR, Event::ANY_EVENT, State::SYNC_TASK_FINISHED},
96 };
97 }
98
99 std::mutex SingleVerSyncStateMachine::stateSwitchTableLock_;
100 std::vector<StateSwitchTable> SingleVerSyncStateMachine::stateSwitchTables_;
101 bool SingleVerSyncStateMachine::isStateSwitchTableInited_ = false;
102
SingleVerSyncStateMachine()103 SingleVerSyncStateMachine::SingleVerSyncStateMachine()
104 : context_(nullptr),
105 syncInterface_(nullptr),
106 timeSync_(nullptr),
107 abilitySync_(nullptr),
108 dataSync_(nullptr),
109 currentRemoteVersionId_(0)
110 {
111 }
112
~SingleVerSyncStateMachine()113 SingleVerSyncStateMachine::~SingleVerSyncStateMachine()
114 {
115 LOGD("~SingleVerSyncStateMachine");
116 Clear();
117 }
118
Initialize(ISyncTaskContext * context,ISyncInterface * syncInterface,const std::shared_ptr<Metadata> & metaData,ICommunicator * communicator)119 int SingleVerSyncStateMachine::Initialize(ISyncTaskContext *context, ISyncInterface *syncInterface,
120 const std::shared_ptr<Metadata> &metaData, ICommunicator *communicator)
121 {
122 if ((context == nullptr) || (syncInterface == nullptr) || (metaData == nullptr) || (communicator == nullptr)) {
123 return -E_INVALID_ARGS;
124 }
125
126 int errCode = SyncStateMachine::Initialize(context, syncInterface, metaData, communicator);
127 if (errCode != E_OK) {
128 return errCode;
129 }
130
131 timeSync_ = std::make_shared<TimeSync>();
132 dataSync_ = std::make_shared<SingleVerDataSync>();
133 abilitySync_ = std::make_unique<AbilitySync>();
134 if ((timeSync_ == nullptr) || (dataSync_ == nullptr) || (abilitySync_ == nullptr)) {
135 timeSync_ = nullptr;
136 dataSync_ = nullptr;
137 abilitySync_ = nullptr;
138 return -E_OUT_OF_MEMORY;
139 }
140
141 errCode = timeSync_->Initialize(communicator, metaData, syncInterface, context->GetDeviceId());
142 if (errCode != E_OK) {
143 goto ERROR_OUT;
144 }
145 errCode = dataSync_->Initialize(syncInterface, communicator, metaData, context->GetDeviceId());
146 if (errCode != E_OK) {
147 goto ERROR_OUT;
148 }
149 errCode = abilitySync_->Initialize(communicator, syncInterface, metaData, context->GetDeviceId());
150 if (errCode != E_OK) {
151 goto ERROR_OUT;
152 }
153 abilitySync_->InitAbilitySyncFinishStatus(*context);
154
155 currentState_ = IDLE;
156 context_ = static_cast<SingleVerSyncTaskContext *>(context);
157 syncInterface_ = static_cast<SyncGenericInterface *>(syncInterface);
158
159 InitStateSwitchTables();
160 InitStateMapping();
161 return E_OK;
162
163 ERROR_OUT:
164 Clear();
165 return errCode;
166 }
167
SyncStep()168 void SingleVerSyncStateMachine::SyncStep()
169 {
170 RefObject::IncObjRef(context_);
171 RefObject::IncObjRef(communicator_);
172 int errCode = RuntimeContext::GetInstance()->ScheduleTask([this] { SyncStepInnerLocked(); });
173 if (errCode != E_OK) {
174 LOGE("[StateMachine][SyncStep] Schedule SyncStep failed");
175 RefObject::DecObjRef(communicator_);
176 RefObject::DecObjRef(context_);
177 }
178 }
179
ReceiveMessageCallback(Message * inMsg)180 int SingleVerSyncStateMachine::ReceiveMessageCallback(Message *inMsg)
181 {
182 int errCode = MessageCallbackPre(inMsg);
183 if (errCode != E_OK) {
184 LOGE("[StateMachine] message pre check failed");
185 return errCode;
186 }
187 switch (inMsg->GetMessageId()) {
188 case TIME_SYNC_MESSAGE:
189 errCode = TimeMarkSyncRecv(inMsg);
190 break;
191 case ABILITY_SYNC_MESSAGE:
192 errCode = AbilitySyncRecv(inMsg);
193 break;
194 case DATA_SYNC_MESSAGE:
195 case QUERY_SYNC_MESSAGE:
196 errCode = DataPktRecv(inMsg);
197 break;
198 case CONTROL_SYNC_MESSAGE:
199 errCode = ControlPktRecv(inMsg);
200 break;
201 default:
202 errCode = -E_NOT_SUPPORT;
203 }
204 return errCode;
205 }
206
SyncStepInnerLocked()207 void SingleVerSyncStateMachine::SyncStepInnerLocked()
208 {
209 if (context_->IncUsedCount() != E_OK) {
210 goto SYNC_STEP_OUT;
211 }
212 {
213 std::lock_guard<std::mutex> lock(stateMachineLock_);
214 SyncStepInner();
215 }
216 context_->SafeExit();
217
218 SYNC_STEP_OUT:
219 RefObject::DecObjRef(communicator_);
220 RefObject::DecObjRef(context_);
221 }
222
SyncStepInner()223 void SingleVerSyncStateMachine::SyncStepInner()
224 {
225 Event event = INNER_ERR_EVENT;
226 do {
227 auto iter = stateMapping_.find(currentState_);
228 if (iter != stateMapping_.end()) {
229 event = static_cast<Event>(iter->second());
230 } else {
231 LOGE("[StateMachine][SyncStepInner] can not find state=%d,label=%s,dev=%s", currentState_,
232 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
233 break;
234 }
235 } while (event != Event::WAIT_ACK_EVENT && SwitchMachineState(event) == E_OK && currentState_ != IDLE);
236 }
237
SetCurStateErrStatus()238 void SingleVerSyncStateMachine::SetCurStateErrStatus()
239 {
240 currentState_ = State::INNER_ERR;
241 }
242
StartSyncInner()243 int SingleVerSyncStateMachine::StartSyncInner()
244 {
245 PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
246 if (performance != nullptr) {
247 performance->StepTimeRecordStart(PT_TEST_RECORDS::RECORD_MACHINE_START_TO_PUSH_SEND);
248 }
249 int errCode = PrepareNextSyncTask();
250 if (errCode == E_OK) {
251 SwitchStateAndStep(Event::START_SYNC_EVENT);
252 }
253 return errCode;
254 }
255
AbortInner()256 void SingleVerSyncStateMachine::AbortInner()
257 {
258 LOGE("[StateMachine][AbortInner] error occurred,abort,label=%s,dev=%s", dataSync_->GetLabel().c_str(),
259 STR_MASK(context_->GetDeviceId()));
260 if (context_->IsKilled()) {
261 dataSync_->ClearDataMsg();
262 }
263 dataSync_->ClearSyncStatus();
264 ContinueToken token;
265 context_->GetContinueToken(token);
266 if (token != nullptr) {
267 syncInterface_->ReleaseContinueToken(token);
268 }
269 context_->SetContinueToken(nullptr);
270 context_->Clear();
271 }
272
GetStateSwitchTables() const273 const std::vector<StateSwitchTable> &SingleVerSyncStateMachine::GetStateSwitchTables() const
274 {
275 return stateSwitchTables_;
276 }
277
PrepareNextSyncTask()278 int SingleVerSyncStateMachine::PrepareNextSyncTask()
279 {
280 int errCode = StartWatchDog();
281 if (errCode != E_OK && errCode != -E_UNEXPECTED_DATA) {
282 LOGE("[StateMachine][PrepareNextSyncTask] WatchDog start failed,err=%d", errCode);
283 return errCode;
284 }
285 if (errCode == -E_UNEXPECTED_DATA) {
286 LOGI("[PrepareNextSyncTask] timer already exists, reset the timer.");
287 (void)ResetWatchDog();
288 }
289
290 if (currentState_ != State::IDLE && currentState_ != State::SYNC_TASK_FINISHED) {
291 LOGW("[StateMachine][PrepareNextSyncTask] PreSync may get an err, state=%" PRIu8 ",dev=%s",
292 currentState_, STR_MASK(context_->GetDeviceId()));
293 currentState_ = State::IDLE;
294 }
295 return E_OK;
296 }
297
SendNotifyPacket(uint32_t sessionId,uint32_t sequenceId,uint32_t inMsgId)298 void SingleVerSyncStateMachine::SendNotifyPacket(uint32_t sessionId, uint32_t sequenceId, uint32_t inMsgId)
299 {
300 dataSync_->SendSaveDataNotifyPacket(context_,
301 std::min(context_->GetRemoteSoftwareVersion(), SOFTWARE_VERSION_CURRENT), sessionId, sequenceId, inMsgId);
302 }
303
CommErrAbort(uint32_t sessionId)304 void SingleVerSyncStateMachine::CommErrAbort(uint32_t sessionId)
305 {
306 std::lock_guard<std::mutex> lock(stateMachineLock_);
307 uint32_t requestSessionId = context_->GetRequestSessionId();
308 if ((sessionId != 0) && ((requestSessionId == 0) || (sessionId != requestSessionId))) {
309 return;
310 }
311 context_->SetCommNormal(false);
312 if (SwitchMachineState(Event::INNER_ERR_EVENT) == E_OK) {
313 SyncStep();
314 }
315 }
316
InitStateSwitchTables()317 void SingleVerSyncStateMachine::InitStateSwitchTables()
318 {
319 if (isStateSwitchTableInited_) {
320 return;
321 }
322
323 std::lock_guard<std::mutex> lock(stateSwitchTableLock_);
324 if (isStateSwitchTableInited_) {
325 return;
326 }
327
328 InitStateSwitchTable(SINGLE_VER_SYNC_PROCTOL_V3, STATE_SWITCH_TABLE_V3);
329 std::sort(stateSwitchTables_.begin(), stateSwitchTables_.end(),
330 [](const auto &tableA, const auto &tableB) {
331 return tableA.version > tableB.version;
332 }); // descending
333 isStateSwitchTableInited_ = true;
334 }
335
InitStateSwitchTable(uint32_t version,const std::vector<std::vector<uint8_t>> & switchTable)336 void SingleVerSyncStateMachine::InitStateSwitchTable(uint32_t version,
337 const std::vector<std::vector<uint8_t>> &switchTable)
338 {
339 StateSwitchTable table;
340 table.version = version;
341 for (const auto &stateSwitch : switchTable) {
342 if (stateSwitch.size() <= OUTPUT_STATE_INDEX) {
343 LOGE("[StateMachine][InitSwitchTable] stateSwitch size err,size=%zu", stateSwitch.size());
344 return;
345 }
346 if (table.switchTable.count(stateSwitch[CURRENT_STATE_INDEX]) == 0) {
347 EventToState eventToState; // new EventToState
348 eventToState[stateSwitch[EVENT_INDEX]] = stateSwitch[OUTPUT_STATE_INDEX];
349 table.switchTable[stateSwitch[CURRENT_STATE_INDEX]] = eventToState;
350 } else { // key stateSwitch[CURRENT_STATE_INDEX] already has EventToState
351 EventToState &eventToState = table.switchTable[stateSwitch[CURRENT_STATE_INDEX]];
352 eventToState[stateSwitch[EVENT_INDEX]] = stateSwitch[OUTPUT_STATE_INDEX];
353 }
354 }
355 stateSwitchTables_.push_back(table);
356 }
357
InitStateMapping()358 void SingleVerSyncStateMachine::InitStateMapping()
359 {
360 stateMapping_[TIME_SYNC] = [this] { return DoTimeSync(); };
361 stateMapping_[ABILITY_SYNC] = [this] { return DoAbilitySync(); };
362 stateMapping_[WAIT_FOR_RECEIVE_DATA_FINISH] = [this] { return DoWaitForDataRecv(); };
363 stateMapping_[SYNC_TASK_FINISHED] = [this] { return DoSyncTaskFinished(); };
364 stateMapping_[SYNC_TIME_OUT] = [this] { return DoTimeout(); };
365 stateMapping_[INNER_ERR] = [this] { return DoInnerErr(); };
366 stateMapping_[START_INITIACTIVE_DATA_SYNC] = [this] { return DoInitiactiveDataSyncWithSlidingWindow(); };
367 stateMapping_[START_PASSIVE_DATA_SYNC] = [this] { return DoPassiveDataSyncWithSlidingWindow(); };
368 stateMapping_[SYNC_CONTROL_CMD] = [this] { return DoInitiactiveControlSync(); };
369 }
370
DoInitiactiveDataSyncWithSlidingWindow() const371 Event SingleVerSyncStateMachine::DoInitiactiveDataSyncWithSlidingWindow() const
372 {
373 LOGD("[StateMachine][activeDataSync] mode=%d,label=%s,dev=%s", context_->GetMode(),
374 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
375 int errCode = E_OK;
376 switch (context_->GetMode()) {
377 case SyncModeType::PUSH:
378 case SyncModeType::QUERY_PUSH:
379 context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
380 errCode = dataSync_->SyncStart(context_->GetMode(), context_);
381 break;
382 case SyncModeType::PULL:
383 case SyncModeType::QUERY_PULL:
384 context_->SetOperationStatus(SyncOperation::OP_SEND_FINISHED);
385 errCode = dataSync_->SyncStart(context_->GetMode(), context_);
386 break;
387 case SyncModeType::PUSH_AND_PULL:
388 case SyncModeType::QUERY_PUSH_PULL:
389 errCode = dataSync_->SyncStart(context_->GetMode(), context_);
390 break;
391 case SyncModeType::RESPONSE_PULL:
392 errCode = dataSync_->SyncStart(context_->GetMode(), context_);
393 break;
394 default:
395 errCode = -E_NOT_SUPPORT;
396 break;
397 }
398 if (errCode == E_OK) {
399 return Event::WAIT_ACK_EVENT;
400 }
401 // once E_EKEYREVOKED error occurred, PUSH_AND_PULL mode should wait for ack to pull remote data.
402 if (SyncOperation::TransferSyncMode(context_->GetMode()) == SyncModeType::PUSH_AND_PULL &&
403 errCode == -E_EKEYREVOKED) {
404 return Event::WAIT_ACK_EVENT;
405 }
406
407 // when response task step dataSync again while request task is running, ignore the errCode
408 bool ignoreInnerErr = (context_->GetResponseSessionId() != 0 && context_->GetRequestSessionId() != 0);
409 Event event = TransformErrCodeToEvent(errCode);
410 return (ignoreInnerErr && event == INNER_ERR_EVENT) ? SEND_FINISHED_EVENT : event;
411 }
412
DoPassiveDataSyncWithSlidingWindow()413 Event SingleVerSyncStateMachine::DoPassiveDataSyncWithSlidingWindow()
414 {
415 {
416 RefObject::AutoLock lock(context_);
417 if (context_->GetRspTargetQueueSize() != 0) {
418 PreStartPullResponse();
419 } else if (context_->GetResponseSessionId() == 0 ||
420 context_->GetRetryStatus() == static_cast<int>(SyncTaskContext::NO_NEED_RETRY)) {
421 return RESPONSE_TASK_FINISHED_EVENT;
422 }
423 }
424 int errCode = dataSync_->SyncStart(SyncModeType::RESPONSE_PULL, context_);
425 if (errCode != E_OK) {
426 LOGE("[SingleVerSyncStateMachine][DoPassiveDataSyncWithSlidingWindow] response pull send failed[%d]", errCode);
427 return RESPONSE_TASK_FINISHED_EVENT;
428 }
429 return Event::WAIT_ACK_EVENT;
430 }
431
DoInitiactiveControlSync() const432 Event SingleVerSyncStateMachine::DoInitiactiveControlSync() const
433 {
434 LOGD("[StateMachine][activeControlSync] mode=%d,label=%s,dev=%s", context_->GetMode(),
435 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
436 context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
437 int errCode = dataSync_->ControlCmdStart(context_);
438 if (errCode == E_OK) {
439 return Event::WAIT_ACK_EVENT;
440 }
441 context_->SetTaskErrCode(errCode);
442 return TransformErrCodeToEvent(errCode);
443 }
444
HandleControlAckRecv(const Message * inMsg)445 int SingleVerSyncStateMachine::HandleControlAckRecv(const Message *inMsg)
446 {
447 std::lock_guard<std::mutex> lock(stateMachineLock_);
448 if (IsNeedResetWatchdog(inMsg)) {
449 (void)ResetWatchDog();
450 }
451 int errCode = dataSync_->ControlCmdAckRecv(context_, inMsg);
452 ControlAckRecvErrCodeHandle(errCode);
453 SwitchStateAndStep(TransformErrCodeToEvent(errCode));
454 return E_OK;
455 }
456
DoWaitForDataRecv() const457 Event SingleVerSyncStateMachine::DoWaitForDataRecv() const
458 {
459 if (context_->GetRspTargetQueueSize() != 0) {
460 return START_PULL_RESPONSE_EVENT;
461 }
462 if (context_->GetOperationStatus() == SyncOperation::OP_FINISHED_ALL) {
463 return RECV_FINISHED_EVENT;
464 }
465 if (SyncOperation::TransferSyncMode(context_->GetMode()) == SyncModeType::PUSH_AND_PULL &&
466 context_->GetOperationStatus() == SyncOperation::OP_EKEYREVOKED_FAILURE &&
467 context_->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_2_0) {
468 return RECV_FINISHED_EVENT;
469 }
470 return Event::WAIT_ACK_EVENT;
471 }
472
DoTimeSync() const473 Event SingleVerSyncStateMachine::DoTimeSync() const
474 {
475 timeSync_->SetTimeSyncFinishIfNeed();
476 if (timeSync_->IsNeedSync()) {
477 CommErrHandler handler = nullptr;
478 // Auto sync need do retry don't use errHandler to return.
479 if (!context_->IsAutoSync()) {
480 handler = [this, context = context_,
481 requestSessionId = context_->GetRequestSessionId()](int ret, bool isDirectEnd) {
482 SyncTaskContext::CommErrHandlerFunc(ret, context, requestSessionId, isDirectEnd);
483 };
484 }
485 int errCode = timeSync_->SyncStart(handler, context_->GetRequestSessionId());
486 if (errCode == E_OK) {
487 return Event::WAIT_ACK_EVENT;
488 }
489 context_->SetTaskErrCode(errCode);
490 return TransformErrCodeToEvent(errCode);
491 }
492
493 return Event::TIME_SYNC_FINISHED_EVENT;
494 }
495
DoAbilitySync() const496 Event SingleVerSyncStateMachine::DoAbilitySync() const
497 {
498 uint16_t remoteCommunicatorVersion = 0;
499 int errCode = communicator_->GetRemoteCommunicatorVersion(context_->GetDeviceId(), remoteCommunicatorVersion);
500 if (errCode != E_OK && errCode != -E_NOT_FOUND) {
501 LOGE("[StateMachine][DoAbilitySync] Get RemoteCommunicatorVersion errCode=%d", errCode);
502 return Event::INNER_ERR_EVENT;
503 }
504 // Fistr version, not support AbilitySync
505 if (remoteCommunicatorVersion == 0 && errCode != -E_NOT_FOUND) {
506 context_->SetRemoteSoftwareVersion(SOFTWARE_VERSION_EARLIEST);
507 return GetEventAfterTimeSync(context_->GetMode());
508 }
509 if (abilitySync_->GetAbilitySyncFinishedStatus()) {
510 return GetEventAfterTimeSync(context_->GetMode());
511 }
512
513 CommErrHandler handler = [this, context = context_,
514 requestSessionId = context_->GetRequestSessionId()](int ret, bool isDirectEnd) {
515 SyncTaskContext::CommErrHandlerFunc(ret, context, requestSessionId, isDirectEnd);
516 };
517 LOGI("[StateMachine][AbilitySync] start abilitySync,label=%s,dev=%s", dataSync_->GetLabel().c_str(),
518 STR_MASK(context_->GetDeviceId()));
519 errCode = abilitySync_->SyncStart(context_->GetRequestSessionId(), context_->GetSequenceId(),
520 remoteCommunicatorVersion, handler, context_);
521 if (errCode != E_OK) {
522 LOGE("[StateMachine][DoAbilitySync] ability sync start failed,errCode=%d", errCode);
523 return TransformErrCodeToEvent(errCode);
524 }
525 return Event::WAIT_ACK_EVENT;
526 }
527
GetEventAfterTimeSync(int mode) const528 Event SingleVerSyncStateMachine::GetEventAfterTimeSync(int mode) const
529 {
530 if (mode == SyncModeType::SUBSCRIBE_QUERY || mode == SyncModeType::UNSUBSCRIBE_QUERY) {
531 return Event::CONTROL_CMD_EVENT;
532 }
533 return Event::ABILITY_SYNC_FINISHED_EVENT;
534 }
535
DoSyncTaskFinished()536 Event SingleVerSyncStateMachine::DoSyncTaskFinished()
537 {
538 StopWatchDog();
539 dataSync_->ClearSyncStatus();
540 RefObject::AutoLock lock(syncContext_);
541 int errCode = ExecNextTask();
542 if (errCode == E_OK) {
543 return Event::START_SYNC_EVENT;
544 }
545 return TransformErrCodeToEvent(errCode);
546 }
547
DoTimeout()548 Event SingleVerSyncStateMachine::DoTimeout()
549 {
550 RefObject::AutoLock lock(context_);
551 if (context_->GetMode() == SyncModeType::SUBSCRIBE_QUERY) {
552 std::shared_ptr<SubscribeManager> subManager = context_->GetSubscribeManager();
553 if (subManager != nullptr) {
554 subManager->DeleteLocalSubscribeQuery(context_->GetDeviceId(), context_->GetQuery());
555 }
556 }
557 context_->Abort(SyncOperation::OP_TIMEOUT);
558 context_->Clear();
559 AbortInner();
560 return Event::ANY_EVENT;
561 }
562
DoInnerErr()563 Event SingleVerSyncStateMachine::DoInnerErr()
564 {
565 RefObject::AutoLock lock(context_);
566 if (!context_->IsCommNormal()) {
567 if (context_->GetMode() == SyncModeType::SUBSCRIBE_QUERY) {
568 std::shared_ptr<SubscribeManager> subManager = context_->GetSubscribeManager();
569 if (subManager != nullptr) {
570 subManager->DeleteLocalSubscribeQuery(context_->GetDeviceId(), context_->GetQuery());
571 }
572 }
573 context_->Abort(SyncOperation::OP_COMM_ABNORMAL);
574 } else {
575 int status = GetSyncOperationStatus(context_->GetTaskErrCode());
576 context_->Abort(status);
577 }
578 context_->Clear();
579 AbortInner();
580 return Event::ANY_EVENT;
581 }
582
AbilitySyncRecv(const Message * inMsg)583 int SingleVerSyncStateMachine::AbilitySyncRecv(const Message *inMsg)
584 {
585 if (inMsg->GetMessageType() == TYPE_REQUEST) {
586 int errCode = abilitySync_->RequestRecv(inMsg, context_);
587 if (errCode != E_OK && inMsg->GetSessionId() == context_->GetResponseSessionId()) {
588 context_->SetTaskErrCode(errCode);
589 std::lock_guard<std::mutex> lock(stateMachineLock_);
590 SwitchStateAndStep(Event::INNER_ERR_EVENT);
591 }
592 return E_OK;
593 }
594 return AbilitySyncResponseRecv(inMsg);
595 }
596
AbilitySyncResponseRecv(const Message * inMsg)597 int SingleVerSyncStateMachine::AbilitySyncResponseRecv(const Message *inMsg)
598 {
599 if (inMsg->GetMessageType() == TYPE_RESPONSE && AbilityMsgSessionIdCheck(inMsg)) {
600 std::lock_guard<std::mutex> lock(stateMachineLock_);
601 int errCode = abilitySync_->AckRecv(inMsg, context_);
602 (void)ResetWatchDog();
603 if (errCode == -E_ABILITY_SYNC_FINISHED) {
604 LOGI("[StateMachine][AbilitySyncRecv] ability sync finished with both kv,label=%s,dev=%s",
605 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
606 abilitySync_->SetAbilitySyncFinishedStatus(true, *context_);
607 JumpStatusAfterAbilitySync(context_->GetMode());
608 } else if (errCode != E_OK) {
609 LOGE("[StateMachine][AbilitySyncRecv] handle ackRecv failed,errCode=%d", errCode);
610 SwitchStateAndStep(TransformErrCodeToEvent(errCode));
611 } else if (context_->GetRemoteSoftwareVersion() <= SOFTWARE_VERSION_RELEASE_2_0) {
612 abilitySync_->SetAbilitySyncFinishedStatus(true, *context_);
613 LOGI("[StateMachine][AbilitySyncRecv] ability Sync Finished,label=%s,dev=%s",
614 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
615 currentRemoteVersionId_ = context_->GetRemoteSoftwareVersionId();
616 JumpStatusAfterAbilitySync(context_->GetMode());
617 }
618 return E_OK;
619 }
620 if (inMsg->GetMessageType() == TYPE_NOTIFY) {
621 return AbilitySyncNotifyRecv(inMsg);
622 }
623 LOGE("[StateMachine][AbilitySyncRecv] msg type invalid");
624 return -E_NOT_SUPPORT;
625 }
626
HandleDataRequestRecv(const Message * inMsg)627 int SingleVerSyncStateMachine::HandleDataRequestRecv(const Message *inMsg)
628 {
629 TimeOffset offset = 0;
630 auto [systemOffset, senderLocalOffset] = SingleVerDataSyncUtils::GetTimeOffsetFromRequestMsg(inMsg);
631 int errCode = timeSync_->GenerateTimeOffsetIfNeed(systemOffset, senderLocalOffset);
632 if (errCode != E_OK) {
633 (void)dataSync_->SendDataAck(context_, inMsg, errCode, 0);
634 return errCode;
635 }
636 uint32_t timeout = communicator_->GetTimeout(context_->GetDeviceId());
637 // If message is data sync request, we should check timeoffset.
638 errCode = timeSync_->GetTimeOffset(offset, timeout);
639 if (errCode != E_OK) {
640 LOGE("[StateMachine][HandleDataRequestRecv] GetTimeOffset err! errCode=%d", errCode);
641 return errCode;
642 }
643 context_->SetTimeOffset(offset);
644 PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
645 if (performance != nullptr) {
646 performance->StepTimeRecordStart(PT_TEST_RECORDS::RECORD_DATA_REQUEST_RECV_TO_SEND_ACK);
647 }
648 DecRefCountOfFeedDogTimer(SyncDirectionFlag::RECEIVE);
649
650 // RequestRecv will save data, it may cost a long time.
651 // So we need to send save data notify to keep remote alive.
652 bool isNeedStop = StartSaveDataNotify(inMsg->GetSessionId(), inMsg->GetSequenceId(), inMsg->GetMessageId());
653 {
654 std::lock_guard<std::mutex> lockWatchDog(stateMachineLock_);
655 if (IsNeedResetWatchdog(inMsg)) {
656 (void)ResetWatchDog();
657 }
658 }
659 WaterMark pullEndWaterkark = 0;
660 errCode = dataSync_->DataRequestRecv(context_, inMsg, pullEndWaterkark);
661 if (performance != nullptr) {
662 performance->StepTimeRecordEnd(PT_TEST_RECORDS::RECORD_DATA_REQUEST_RECV_TO_SEND_ACK);
663 }
664 // only higher than 102 version receive this errCode here.
665 // while both RequestSessionId is not equal,but get this errCode;slwr would seem to handle first secquencid.
666 // so while receive the same secquencid after abiitysync it wouldn't handle.
667 if (errCode == -E_NEED_ABILITY_SYNC) {
668 if (isNeedStop) {
669 StopSaveDataNotify();
670 }
671 return errCode;
672 }
673 {
674 std::lock_guard<std::mutex> lock(stateMachineLock_);
675 DataRecvErrCodeHandle(inMsg->GetSessionId(), errCode);
676 if (pullEndWaterkark > 0) {
677 AddPullResponseTarget(inMsg, pullEndWaterkark);
678 }
679 }
680 if (isNeedStop) {
681 StopSaveDataNotify();
682 }
683 return E_OK;
684 }
685
HandleDataAckRecvWithSlidingWindow(int errCode,const Message * inMsg,bool ignoreInnerErr)686 void SingleVerSyncStateMachine::HandleDataAckRecvWithSlidingWindow(int errCode, const Message *inMsg,
687 bool ignoreInnerErr)
688 {
689 if (errCode == -E_RE_SEND_DATA) { // LOCAL_WATER_MARK_NOT_INIT
690 dataSync_->ClearSyncStatus();
691 }
692 if (errCode == -E_NO_DATA_SEND || errCode == -E_SEND_DATA) {
693 int ret = dataSync_->TryContinueSync(context_, inMsg);
694 if (ret == -E_FINISHED) {
695 SwitchStateAndStep(Event::SEND_FINISHED_EVENT);
696 return;
697 } else if (ret == E_OK) { // do nothing and waiting for all ack receive
698 return;
699 }
700 errCode = ret;
701 }
702 ResponsePullError(errCode, ignoreInnerErr);
703 }
704
NeedAbilitySyncHandle()705 void SingleVerSyncStateMachine::NeedAbilitySyncHandle()
706 {
707 // if the remote device version num is overdue,
708 // mean the version num has been reset when syncing data,
709 // there should not clear the new version cache again.
710 if (currentRemoteVersionId_ == context_->GetRemoteSoftwareVersionId()) {
711 LOGI("[StateMachine] set remote version 0, currentRemoteVersionId_ = %" PRIu64, currentRemoteVersionId_);
712 context_->SetRemoteSoftwareVersion(0);
713 } else {
714 currentRemoteVersionId_ = context_->GetRemoteSoftwareVersionId();
715 }
716 abilitySync_->SetAbilitySyncFinishedStatus(false, *context_);
717 }
718
HandleDataAckRecv(const Message * inMsg)719 int SingleVerSyncStateMachine::HandleDataAckRecv(const Message *inMsg)
720 {
721 if (inMsg->GetMessageType() == TYPE_RESPONSE) {
722 DecRefCountOfFeedDogTimer(SyncDirectionFlag::SEND);
723 }
724 std::lock_guard<std::mutex> lock(stateMachineLock_);
725 // Unfortunately we use stateMachineLock_ in many sync process
726 // So a bad ack will check before the lock and wait
727 // And then another process is running, it will get the lock.After this process, the ack became invalid.
728 // If we don't check ack again, it will be delivered to dataSyncer.
729 if (!IsPacketValid(inMsg)) {
730 return -E_INVALID_ARGS;
731 }
732 if (IsNeedResetWatchdog(inMsg)) {
733 (void)ResetWatchDog();
734 }
735 if (context_->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_2_0 && !dataSync_->AckPacketIdCheck(inMsg)) {
736 // packetId not match but sequence id matched scene, means resend map has be rebuilt
737 // this is old ack, should be dropped and wait for the same packetId sequence.
738 return E_OK;
739 }
740 // AckRecv will save meta data, it may cost a long time. if another thread is saving data
741 // So we need to send save data notify to keep remote alive.
742 // e.g. remote do pull sync
743 bool isNeedStop = StartSaveDataNotify(inMsg->GetSessionId(), inMsg->GetSequenceId(), inMsg->GetMessageId());
744 int errCode = dataSync_->AckRecv(context_, inMsg);
745 if (isNeedStop) {
746 StopSaveDataNotify();
747 }
748 if (errCode == -E_NEED_ABILITY_SYNC || errCode == -E_RE_SEND_DATA || errCode == -E_NEED_TIME_SYNC) {
749 StopFeedDogForSync(SyncDirectionFlag::SEND);
750 } else if (errCode == -E_SAVE_DATA_NOTIFY) {
751 return errCode;
752 }
753 // when this msg is from response task while request task is running, ignore the errCode
754 bool ignoreInnerErr = inMsg->GetSessionId() == context_->GetResponseSessionId() &&
755 context_->GetRequestSessionId() != 0;
756 DataAckRecvErrCodeHandle(errCode, !ignoreInnerErr);
757 HandleDataAckRecvWithSlidingWindow(errCode, inMsg, ignoreInnerErr);
758 return errCode;
759 }
760
DataPktRecv(Message * inMsg)761 int SingleVerSyncStateMachine::DataPktRecv(Message *inMsg)
762 {
763 PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
764 int errCode = E_OK;
765 switch (inMsg->GetMessageType()) {
766 case TYPE_REQUEST:
767 ScheduleMsgAndHandle(inMsg);
768 errCode = -E_NOT_NEED_DELETE_MSG;
769 break;
770 case TYPE_RESPONSE:
771 case TYPE_NOTIFY:
772 if (performance != nullptr) {
773 performance->StepTimeRecordEnd(PT_TEST_RECORDS::RECORD_DATA_SEND_REQUEST_TO_ACK_RECV);
774 performance->StepTimeRecordStart(PT_TEST_RECORDS::RECORD_ACK_RECV_TO_USER_CALL_BACK);
775 }
776 errCode = HandleDataAckRecv(inMsg);
777 break;
778 default:
779 errCode = -E_INVALID_ARGS;
780 break;
781 }
782 return errCode;
783 }
784
ScheduleMsgAndHandle(Message * inMsg)785 void SingleVerSyncStateMachine::ScheduleMsgAndHandle(Message *inMsg)
786 {
787 dataSync_->PutDataMsg(inMsg);
788 while (true) {
789 bool isNeedHandle = true;
790 bool isNeedContinue = true;
791 Message *msg = dataSync_->MoveNextDataMsg(context_, isNeedHandle, isNeedContinue);
792 if (!isNeedContinue) {
793 break;
794 }
795 if (msg == nullptr) {
796 if (dataSync_->IsNeedReloadQueue()) {
797 continue;
798 }
799 dataSync_->ScheduleInfoHandle(false, false, nullptr);
800 break;
801 }
802 bool isNeedClearMap = false;
803 if (isNeedHandle) {
804 int errCode = HandleDataRequestRecv(msg);
805 if (context_->IsReceiveWaterMarkErr() || errCode == -E_NEED_ABILITY_SYNC || errCode == -E_NEED_TIME_SYNC) {
806 isNeedClearMap = true;
807 }
808 if (errCode == -E_TIMEOUT) {
809 isNeedHandle = false;
810 }
811 } else {
812 dataSync_->SendFinishedDataAck(context_, msg);
813 }
814 if (context_->GetRemoteSoftwareVersion() < SOFTWARE_VERSION_RELEASE_3_0) {
815 // for lower version, no need to handle map schedule info, just reset schedule working status
816 isNeedHandle = false;
817 }
818 dataSync_->ScheduleInfoHandle(isNeedHandle, isNeedClearMap, msg);
819 delete msg;
820 }
821 }
822
ControlPktRecv(Message * inMsg)823 int SingleVerSyncStateMachine::ControlPktRecv(Message *inMsg)
824 {
825 int errCode = E_OK;
826 switch (inMsg->GetMessageType()) {
827 case TYPE_REQUEST:
828 errCode = dataSync_->ControlCmdRequestRecv(context_, inMsg);
829 break;
830 case TYPE_RESPONSE:
831 errCode = HandleControlAckRecv(inMsg);
832 break;
833 default:
834 errCode = -E_INVALID_ARGS;
835 break;
836 }
837 return errCode;
838 }
839
StepToTimeout(TimerId timerId)840 void SingleVerSyncStateMachine::StepToTimeout(TimerId timerId)
841 {
842 std::lock_guard<std::mutex> lock(stateMachineLock_);
843 TimerId timer = syncContext_->GetTimerId();
844 if (timer != timerId) {
845 return;
846 }
847 SwitchStateAndStep(Event::TIME_OUT_EVENT);
848 }
849
850 namespace {
851 struct StateNode {
852 int errCode = 0;
853 SyncOperation::Status status = SyncOperation::OP_WAITING;
854 };
855 }
GetSyncOperationStatus(int errCode) const856 int SingleVerSyncStateMachine::GetSyncOperationStatus(int errCode) const
857 {
858 static const StateNode stateNodes[] = {
859 { -E_SCHEMA_MISMATCH, SyncOperation::OP_SCHEMA_INCOMPATIBLE },
860 { -E_EKEYREVOKED, SyncOperation::OP_EKEYREVOKED_FAILURE },
861 { -E_SECURITY_OPTION_CHECK_ERROR, SyncOperation::OP_SECURITY_OPTION_CHECK_FAILURE },
862 { -E_BUSY, SyncOperation::OP_BUSY_FAILURE },
863 { -E_NOT_PERMIT, SyncOperation::OP_PERMISSION_CHECK_FAILED },
864 { -E_TIMEOUT, SyncOperation::OP_TIMEOUT },
865 { -E_INVALID_QUERY_FORMAT, SyncOperation::OP_QUERY_FORMAT_FAILURE },
866 { -E_INVALID_QUERY_FIELD, SyncOperation::OP_QUERY_FIELD_FAILURE },
867 { -E_FEEDBACK_UNKNOWN_MESSAGE, SyncOperation::OP_NOT_SUPPORT },
868 { -E_FEEDBACK_COMMUNICATOR_NOT_FOUND, SyncOperation::OP_COMM_ABNORMAL },
869 { -E_NOT_SUPPORT, SyncOperation::OP_NOT_SUPPORT },
870 { -E_INTERCEPT_DATA_FAIL, SyncOperation::OP_INTERCEPT_DATA_FAIL },
871 { -E_MAX_LIMITS, SyncOperation::OP_MAX_LIMITS },
872 { -E_DISTRIBUTED_SCHEMA_CHANGED, SyncOperation::OP_SCHEMA_CHANGED },
873 { -E_NOT_REGISTER, SyncOperation::OP_NOT_SUPPORT },
874 { -E_DENIED_SQL, SyncOperation::OP_DENIED_SQL },
875 { -E_REMOTE_OVER_SIZE, SyncOperation::OP_MAX_LIMITS },
876 { -E_INVALID_PASSWD_OR_CORRUPTED_DB, SyncOperation::OP_NOTADB_OR_CORRUPTED }
877 };
878 const auto &result = std::find_if(std::begin(stateNodes), std::end(stateNodes), [errCode](const auto &node) {
879 return node.errCode == errCode;
880 });
881 return result == std::end(stateNodes) ? SyncOperation::OP_FAILED : static_cast<int>((*result).status);
882 }
883
TimeMarkSyncRecv(const Message * inMsg)884 int SingleVerSyncStateMachine::TimeMarkSyncRecv(const Message *inMsg)
885 {
886 LOGD("[StateMachine][TimeMarkSyncRecv] type=%d,label=%s,dev=%s", inMsg->GetMessageType(),
887 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
888 {
889 std::lock_guard<std::mutex> lock(stateMachineLock_);
890 (void)ResetWatchDog();
891 }
892 if (inMsg->GetMessageType() == TYPE_REQUEST) {
893 return timeSync_->RequestRecv(inMsg);
894 } else if (inMsg->GetMessageType() == TYPE_RESPONSE) {
895 int errCode = timeSync_->AckRecv(inMsg, context_->GetRequestSessionId());
896 if (errCode != E_OK) {
897 LOGE("[StateMachine][TimeMarkSyncRecv] AckRecv failed errCode=%d", errCode);
898 if (inMsg->GetSessionId() != 0 && inMsg->GetSessionId() == context_->GetRequestSessionId()) {
899 context_->SetTaskErrCode(errCode);
900 InnerErrorAbort(inMsg->GetSessionId());
901 }
902 return errCode;
903 }
904 std::lock_guard<std::mutex> lock(stateMachineLock_);
905 SwitchStateAndStep(TIME_SYNC_FINISHED_EVENT);
906 return E_OK;
907 } else {
908 return -E_INVALID_ARGS;
909 }
910 }
911
Clear()912 void SingleVerSyncStateMachine::Clear()
913 {
914 dataSync_ = nullptr;
915 timeSync_ = nullptr;
916 abilitySync_ = nullptr;
917 context_ = nullptr;
918 syncInterface_ = nullptr;
919 }
920
IsPacketValid(const Message * inMsg) const921 bool SingleVerSyncStateMachine::IsPacketValid(const Message *inMsg) const
922 {
923 if (inMsg == nullptr) {
924 return false;
925 }
926
927 if ((inMsg->GetMessageId() < TIME_SYNC_MESSAGE) || (inMsg->GetMessageId() >= UNKNOW_MESSAGE)) {
928 LOGE("[StateMachine][IsPacketValid] Message is invalid, id=%d", inMsg->GetMessageId());
929 return false;
930 }
931 // filter invalid ack at first
932 bool isResponseType = (inMsg->GetMessageType() == TYPE_RESPONSE);
933 if (isResponseType && (inMsg->GetMessageId() == CONTROL_SYNC_MESSAGE) &&
934 (inMsg->GetSessionId() != context_->GetRequestSessionId())) {
935 LOGE("[StateMachine][IsPacketValid] Control Message is invalid, label=%s, dev=%s",
936 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
937 return false;
938 }
939 if (isResponseType && (inMsg->GetMessageId() != TIME_SYNC_MESSAGE) &&
940 (inMsg->GetSessionId() != context_->GetRequestSessionId()) &&
941 (inMsg->GetSessionId() != context_->GetResponseSessionId())) {
942 LOGE("[StateMachine][IsPacketValid] Data Message is invalid, label=%s, dev=%s",
943 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
944 return false;
945 }
946
947 // timeSync and abilitySync don't need to check sequenceId
948 if (inMsg->GetMessageId() == TIME_SYNC_MESSAGE || inMsg->GetMessageId() == ABILITY_SYNC_MESSAGE ||
949 inMsg->GetMessageId() == CONTROL_SYNC_MESSAGE) {
950 return true;
951 }
952 // sequenceId will be checked in dataSync
953 return true;
954 }
955
PreStartPullResponse()956 void SingleVerSyncStateMachine::PreStartPullResponse()
957 {
958 SingleVerSyncTarget target;
959 context_->PopResponseTarget(target);
960 context_->SetEndMark(target.GetEndWaterMark());
961 context_->SetResponseSessionId(target.GetResponseSessionId());
962 context_->SetMode(SyncModeType::RESPONSE_PULL);
963 context_->ReSetSequenceId();
964 context_->SetQuerySync(target.IsQuerySync());
965 context_->SetQuery(target.GetQuery());
966 }
967
CheckIsStartPullResponse() const968 bool SingleVerSyncStateMachine::CheckIsStartPullResponse() const
969 {
970 // Other state will step to do pull response, only this statem we need to step the statemachine
971 if (currentState_ == WAIT_FOR_RECEIVE_DATA_FINISH) {
972 return true;
973 }
974 return false;
975 }
976
MessageCallbackPre(const Message * inMsg)977 int SingleVerSyncStateMachine::MessageCallbackPre(const Message *inMsg)
978 {
979 RefObject::AutoLock lock(context_);
980 if (context_->IsKilled()) {
981 return -E_OBJ_IS_KILLED;
982 }
983
984 if (!IsPacketValid(inMsg)) {
985 return -E_INVALID_ARGS;
986 }
987 return E_OK;
988 }
989
AddPullResponseTarget(const Message * inMsg,WaterMark pullEndWatermark)990 void SingleVerSyncStateMachine::AddPullResponseTarget(const Message *inMsg, WaterMark pullEndWatermark)
991 {
992 int messageType = static_cast<int>(inMsg->GetMessageId());
993 uint32_t sessionId = inMsg->GetSessionId();
994 if (pullEndWatermark == 0) {
995 LOGE("[StateMachine][AddPullResponseTarget] pullEndWatermark is 0!");
996 return;
997 }
998 if (context_->GetResponseSessionId() == sessionId || context_->FindResponseSyncTarget(sessionId)) {
999 LOGI("[StateMachine][AddPullResponseTarget] task is already running");
1000 return;
1001 }
1002 const DataRequestPacket *packet = inMsg->GetObject<DataRequestPacket>();
1003 if (packet == nullptr) {
1004 LOGE("[AddPullResponseTarget] get packet object failed");
1005 return;
1006 }
1007 SingleVerSyncTarget *targetTmp = new (std::nothrow) SingleVerSyncTarget;
1008 if (targetTmp == nullptr) {
1009 LOGE("[StateMachine][AddPullResponseTarget] add failed, may oom");
1010 return;
1011 }
1012 targetTmp->SetTaskType(ISyncTarget::RESPONSE);
1013 if (messageType == QUERY_SYNC_MESSAGE) {
1014 targetTmp->SetQuery(packet->GetQuery());
1015 targetTmp->SetQuerySync(true);
1016 }
1017 targetTmp->SetMode(SyncModeType::RESPONSE_PULL);
1018 targetTmp->SetEndWaterMark(pullEndWatermark);
1019 targetTmp->SetResponseSessionId(sessionId);
1020 if (context_->AddSyncTarget(targetTmp) != E_OK) {
1021 delete targetTmp;
1022 return;
1023 }
1024 if (CheckIsStartPullResponse()) {
1025 SwitchStateAndStep(TransformErrCodeToEvent(-E_NEED_PULL_REPONSE));
1026 }
1027 }
1028
TransformErrCodeToEvent(int errCode) const1029 Event SingleVerSyncStateMachine::TransformErrCodeToEvent(int errCode) const
1030 {
1031 switch (errCode) {
1032 case -E_TIMEOUT:
1033 return TransforTimeOutErrCodeToEvent();
1034 case -static_cast<int>(VERSION_NOT_SUPPOR_EVENT):
1035 return Event::VERSION_NOT_SUPPOR_EVENT;
1036 case -E_SEND_DATA:
1037 return Event::SEND_DATA_EVENT;
1038 case -E_NO_DATA_SEND:
1039 return Event::SEND_FINISHED_EVENT;
1040 case -E_RECV_FINISHED:
1041 return Event::RECV_FINISHED_EVENT;
1042 case -E_NEED_ABILITY_SYNC:
1043 return Event::NEED_ABILITY_SYNC_EVENT;
1044 case -E_NO_SYNC_TASK:
1045 return Event::ALL_TASK_FINISHED_EVENT;
1046 case -E_NEED_PULL_REPONSE:
1047 return Event::START_PULL_RESPONSE_EVENT;
1048 case -E_RE_SEND_DATA:
1049 return Event::RE_SEND_DATA_EVENT;
1050 case -E_NEED_TIME_SYNC:
1051 return Event::NEED_TIME_SYNC_EVENT;
1052 default:
1053 return Event::INNER_ERR_EVENT;
1054 }
1055 }
1056
IsNeedResetWatchdog(const Message * inMsg) const1057 bool SingleVerSyncStateMachine::IsNeedResetWatchdog(const Message *inMsg) const
1058 {
1059 if (inMsg == nullptr) {
1060 return false;
1061 }
1062
1063 if (IsNeedErrCodeHandle(inMsg->GetSessionId())) {
1064 return true;
1065 }
1066
1067 int msgType = inMsg->GetMessageType();
1068 if (msgType == TYPE_RESPONSE || msgType == TYPE_NOTIFY) {
1069 if (inMsg->GetSessionId() == context_->GetResponseSessionId()) {
1070 // Pull response ack also should reset watchdog
1071 return true;
1072 }
1073 }
1074
1075 return false;
1076 }
1077
TransforTimeOutErrCodeToEvent() const1078 Event SingleVerSyncStateMachine::TransforTimeOutErrCodeToEvent() const
1079 {
1080 if (syncContext_->IsSyncTaskNeedRetry() && (syncContext_->GetRetryTime() < syncContext_->GetSyncRetryTimes())) {
1081 return Event::WAIT_TIME_OUT_EVENT;
1082 } else {
1083 return Event::TIME_OUT_EVENT;
1084 }
1085 }
1086
IsNeedErrCodeHandle(uint32_t sessionId) const1087 bool SingleVerSyncStateMachine::IsNeedErrCodeHandle(uint32_t sessionId) const
1088 {
1089 // omit to set sessionId so version_3 should skip to compare sessionid.
1090 if (sessionId == context_->GetRequestSessionId() ||
1091 context_->GetRemoteSoftwareVersion() == SOFTWARE_VERSION_RELEASE_2_0) {
1092 return true;
1093 }
1094 return false;
1095 }
1096
PushPullDataRequestEvokeErrHandle()1097 void SingleVerSyncStateMachine::PushPullDataRequestEvokeErrHandle()
1098 {
1099 // the pushpull sync task should wait for send finished after remote dev get data occur E_EKEYREVOKED error.
1100 if (context_->GetRemoteSoftwareVersion() > SOFTWARE_VERSION_RELEASE_2_0 &&
1101 SyncOperation::TransferSyncMode(context_->GetMode()) == SyncModeType::PUSH_AND_PULL) { // LCOV_EXCL_BR_LINE
1102 LOGI("data request errCode = %d, wait for send finished", -E_EKEYREVOKED);
1103 context_->SetTaskErrCode(-E_EKEYREVOKED);
1104 context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
1105 SwitchStateAndStep(Event::RECV_FINISHED_EVENT);
1106 } else {
1107 context_->SetTaskErrCode(-E_EKEYREVOKED);
1108 SwitchStateAndStep(Event::INNER_ERR_EVENT);
1109 }
1110 }
1111
DataRecvErrCodeHandle(uint32_t sessionId,int errCode)1112 void SingleVerSyncStateMachine::DataRecvErrCodeHandle(uint32_t sessionId, int errCode)
1113 {
1114 if (IsNeedErrCodeHandle(sessionId)) {
1115 switch (errCode) {
1116 case E_OK:
1117 break;
1118 case -E_NOT_PERMIT:
1119 context_->SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
1120 break;
1121 case -E_RECV_FINISHED:
1122 context_->SetOperationStatus(SyncOperation::OP_RECV_FINISHED);
1123 SwitchStateAndStep(Event::RECV_FINISHED_EVENT);
1124 break;
1125 case -E_EKEYREVOKED:
1126 PushPullDataRequestEvokeErrHandle();
1127 break;
1128 case -E_BUSY:
1129 case -E_DISTRIBUTED_SCHEMA_CHANGED:
1130 case -E_DISTRIBUTED_SCHEMA_NOT_FOUND:
1131 case -E_FEEDBACK_COMMUNICATOR_NOT_FOUND:
1132 case -E_FEEDBACK_UNKNOWN_MESSAGE:
1133 case -E_INTERCEPT_DATA_FAIL:
1134 case -E_INVALID_PASSWD_OR_CORRUPTED_DB:
1135 case -E_INVALID_QUERY_FIELD:
1136 case -E_INVALID_QUERY_FORMAT:
1137 case -E_MAX_LIMITS:
1138 case -E_NOT_REGISTER:
1139 case -E_NOT_SUPPORT:
1140 case -E_SECURITY_OPTION_CHECK_ERROR:
1141 context_->SetTaskErrCode(errCode);
1142 SwitchStateAndStep(Event::INNER_ERR_EVENT);
1143 break;
1144 default:
1145 SwitchStateAndStep(Event::INNER_ERR_EVENT);
1146 break;
1147 }
1148 }
1149 }
1150
AbilityMsgSessionIdCheck(const Message * inMsg)1151 bool SingleVerSyncStateMachine::AbilityMsgSessionIdCheck(const Message *inMsg)
1152 {
1153 if (inMsg != nullptr && inMsg->GetSessionId() == context_->GetRequestSessionId()) { // LCOV_EXCL_BR_LINE
1154 return true;
1155 }
1156 LOGE("[AbilitySync] session check failed,dev=%s", STR_MASK(context_->GetDeviceId()));
1157 return false;
1158 }
1159
GetSyncType(uint32_t messageId) const1160 SyncType SingleVerSyncStateMachine::GetSyncType(uint32_t messageId) const
1161 {
1162 if (messageId == QUERY_SYNC_MESSAGE) { // LCOV_EXCL_BR_LINE
1163 return SyncType::QUERY_SYNC_TYPE;
1164 }
1165 return SyncType::MANUAL_FULL_SYNC_TYPE;
1166 }
1167
DataAckRecvErrCodeHandle(int errCode,bool handleError)1168 void SingleVerSyncStateMachine::DataAckRecvErrCodeHandle(int errCode, bool handleError)
1169 {
1170 switch (errCode) {
1171 case -E_NEED_ABILITY_SYNC:
1172 NeedAbilitySyncHandle();
1173 break;
1174 case -E_NEED_TIME_SYNC:
1175 timeSync_->ClearTimeSyncFinish();
1176 break;
1177 case -E_NOT_PERMIT:
1178 if (handleError) {
1179 context_->SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
1180 }
1181 break;
1182 case -E_BUSY:
1183 case -E_DISTRIBUTED_SCHEMA_CHANGED:
1184 case -E_DISTRIBUTED_SCHEMA_NOT_FOUND:
1185 case -E_EKEYREVOKED:
1186 case -E_FEEDBACK_COMMUNICATOR_NOT_FOUND:
1187 case -E_FEEDBACK_UNKNOWN_MESSAGE:
1188 case -E_INTERCEPT_DATA_FAIL:
1189 case -E_INVALID_PASSWD_OR_CORRUPTED_DB:
1190 case -E_INVALID_QUERY_FIELD:
1191 case -E_INVALID_QUERY_FORMAT:
1192 case -E_MAX_LIMITS:
1193 case -E_NOT_REGISTER:
1194 case -E_NOT_SUPPORT:
1195 case -E_SECURITY_OPTION_CHECK_ERROR:
1196 if (handleError) {
1197 context_->SetTaskErrCode(errCode);
1198 }
1199 break;
1200 default:
1201 break;
1202 }
1203 }
1204
IsNeedTriggerQueryAutoSync(Message * inMsg,QuerySyncObject & query)1205 bool SingleVerSyncStateMachine::IsNeedTriggerQueryAutoSync(Message *inMsg, QuerySyncObject &query)
1206 {
1207 return SingleVerDataSyncUtils::IsNeedTriggerQueryAutoSync(inMsg, query);
1208 }
1209
JumpStatusAfterAbilitySync(int mode)1210 void SingleVerSyncStateMachine::JumpStatusAfterAbilitySync(int mode)
1211 {
1212 if ((mode == SyncModeType::SUBSCRIBE_QUERY) || (mode == SyncModeType::UNSUBSCRIBE_QUERY)) {
1213 SwitchStateAndStep(CONTROL_CMD_EVENT);
1214 } else {
1215 SwitchStateAndStep(ABILITY_SYNC_FINISHED_EVENT);
1216 }
1217 }
1218
ControlAckRecvErrCodeHandle(int errCode)1219 void SingleVerSyncStateMachine::ControlAckRecvErrCodeHandle(int errCode)
1220 {
1221 switch (errCode) {
1222 case -E_NEED_ABILITY_SYNC:
1223 NeedAbilitySyncHandle();
1224 break;
1225 case -E_NO_DATA_SEND:
1226 context_->SetOperationStatus(SyncOperation::OP_SEND_FINISHED);
1227 break;
1228 case -E_NOT_PERMIT:
1229 context_->SetOperationStatus(SyncOperation::OP_PERMISSION_CHECK_FAILED);
1230 break;
1231 // other errCode use default
1232 default:
1233 context_->SetTaskErrCode(errCode);
1234 break;
1235 }
1236 }
1237
GetLocalWaterMark(const DeviceID & deviceId,uint64_t & outValue)1238 void SingleVerSyncStateMachine::GetLocalWaterMark(const DeviceID &deviceId, uint64_t &outValue)
1239 {
1240 metadata_->GetLocalWaterMark(deviceId, outValue);
1241 }
1242
GetSendQueryWaterMark(const std::string & queryId,const DeviceID & deviceId,bool isAutoLift,uint64_t & outValue)1243 int SingleVerSyncStateMachine::GetSendQueryWaterMark(const std::string &queryId, const DeviceID &deviceId,
1244 bool isAutoLift, uint64_t &outValue)
1245 {
1246 return metadata_->GetSendQueryWaterMark(queryId, deviceId, outValue, isAutoLift);
1247 }
1248
ResponsePullError(int errCode,bool ignoreInnerErr)1249 void SingleVerSyncStateMachine::ResponsePullError(int errCode, bool ignoreInnerErr)
1250 {
1251 Event event = TransformErrCodeToEvent(errCode);
1252 if (event == INNER_ERR_EVENT) {
1253 if (ignoreInnerErr) {
1254 event = RESPONSE_TASK_FINISHED_EVENT;
1255 } else if (context_ != nullptr) {
1256 context_->SetTaskErrCode(errCode);
1257 }
1258 }
1259 SwitchStateAndStep(event);
1260 }
1261
InnerErrorAbort(uint32_t sessionId)1262 void SingleVerSyncStateMachine::InnerErrorAbort(uint32_t sessionId)
1263 {
1264 std::lock_guard<std::mutex> lock(stateMachineLock_);
1265 uint32_t requestSessionId = context_->GetRequestSessionId();
1266 if (sessionId != requestSessionId) {
1267 LOGD("[SingleVerSyncStateMachine][InnerErrorAbort] Ignore abort by different sessionId");
1268 return;
1269 }
1270 if (SwitchMachineState(Event::INNER_ERR_EVENT) == E_OK) {
1271 SyncStep();
1272 }
1273 }
1274
NotifyClosing()1275 void SingleVerSyncStateMachine::NotifyClosing()
1276 {
1277 if (timeSync_ != nullptr) {
1278 timeSync_->Close();
1279 }
1280 }
1281
AbilitySyncNotifyRecv(const Message * inMsg)1282 int SingleVerSyncStateMachine::AbilitySyncNotifyRecv(const Message *inMsg)
1283 {
1284 const AbilitySyncAckPacket *packet = inMsg->GetObject<AbilitySyncAckPacket>();
1285 if (packet == nullptr) {
1286 return -E_INVALID_ARGS;
1287 }
1288 int ackCode = packet->GetAckCode();
1289 if (ackCode != AbilitySync::CHECK_SUCCESS && ackCode != AbilitySync::LAST_NOTIFY) {
1290 LOGE("[StateMachine][AbilitySyncRecv] ackCode check failed,ackCode=%d", ackCode);
1291 context_->SetTaskErrCode(ackCode);
1292 std::lock_guard<std::mutex> lock(stateMachineLock_);
1293 SwitchStateAndStep(Event::INNER_ERR_EVENT);
1294 return E_OK;
1295 }
1296 if (ackCode == AbilitySync::LAST_NOTIFY && AbilityMsgSessionIdCheck(inMsg)) {
1297 abilitySync_->SetAbilitySyncFinishedStatus(true, *context_);
1298 // while recv last notify means ability sync finished,it is better to reset watchDog to avoid timeout.
1299 LOGI("[StateMachine][AbilitySyncRecv] ability sync finished,label=%s,dev=%s",
1300 dataSync_->GetLabel().c_str(), STR_MASK(context_->GetDeviceId()));
1301 context_->SetRemoteSoftwareVersion(packet->GetSoftwareVersion());
1302 currentRemoteVersionId_ = context_->GetRemoteSoftwareVersionId();
1303 std::lock_guard<std::mutex> lock(stateMachineLock_);
1304 (void)ResetWatchDog();
1305 JumpStatusAfterAbilitySync(context_->GetMode());
1306 } else if (ackCode != AbilitySync::LAST_NOTIFY) {
1307 abilitySync_->AckNotifyRecv(inMsg, context_);
1308 }
1309 return E_OK;
1310 }
1311
SchemaChange()1312 void SingleVerSyncStateMachine::SchemaChange()
1313 {
1314 abilitySync_->SetAbilitySyncFinishedStatus(false, *context_);
1315 }
1316
TimeChange()1317 void SingleVerSyncStateMachine::TimeChange()
1318 {
1319 if (timeSync_ == nullptr) {
1320 LOGW("[SingleVerSyncStateMachine] time sync is null when time change");
1321 return;
1322 }
1323 timeSync_->ClearTimeSyncFinish();
1324 }
1325 } // namespace DistributedDB
1326