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 #ifndef SYNC_STATE_MACHINE_H
17 #define SYNC_STATE_MACHINE_H
18 
19 #include <mutex>
20 
21 #include "isync_interface.h"
22 #include "isync_state_machine.h"
23 
24 namespace DistributedDB {
25 // the 1st uint8_t is event, the 2nd uint8_t is out state
26 using EventToState = std::map<uint8_t, uint8_t>;
27 
28 // The StateSwitchTable with the SyncProctolVersion
29 struct StateSwitchTable {
30     uint32_t version = 0;
31     std::map<uint8_t, EventToState> switchTable; // the 1st uint8_t is current state
32 };
33 
34 struct WatchDogController {
35     TimerId feedDogTimerId = 0;
36     uint8_t feedDogCnt = 0;
37     uint8_t feedDogUpperLimit = 0;
38     /* this variable will +1 when call StartFeedDogForSync, -1 when recv one ack,
39     when it become <= 0, we stop the watch dog. */
40     int refCount = 0;
41 };
42 
43 class SyncStateMachine : public ISyncStateMachine {
44 public:
45     SyncStateMachine();
46     ~SyncStateMachine() override;
47 
48     // Init the SingleVerSyncStateMachine
49     int Initialize(ISyncTaskContext *context, ISyncInterface *syncInterface, const std::shared_ptr<Metadata> &metadata,
50         ICommunicator *communicator) override;
51 
52     // start a sync step
53     int StartSync() override;
54 
55     // call when timeout
56     int TimeoutCallback(TimerId timerId) override;
57 
58     // Force stop the state machine
59     void Abort() override;
60 
61     // Force stop the state machine now
62     void AbortImmediately() override;
63 
64     // Force stop current task with sessionId
65     void InnerErrorAbort(uint32_t sessionId) override;
66 
67     // start a timer to ResetWatchDog when sync data one (key,value) size bigger than mtu
68     bool StartFeedDogForSync(uint32_t time, SyncDirectionFlag flag) override;
69 
70     uint8_t GetFeedDogTimeout(int timeoutCount) const;
71 
72     // stop timer to ResetWatchDog when sync data one (key,value) size bigger than mtu
73     void StopFeedDogForSync(SyncDirectionFlag flag) override;
74 
75     // start a timer to ResetWatchDog when get data and send notify ack if need
76     void StartFeedDogForGetData(uint32_t sessionId) override;
77 
78     // start a timer to ResetWatchDog when get data and stop send notify ack if need
79     void StopFeedDogForGetData() override;
80 
81     // Notify machine is closing, should release some lock
82     void NotifyClosing() override;
83 
84     void SchemaChange() override;
85 
86     void TimeChange() override;
87 protected:
88 
89     // SyncOperation is timeout, step to timeout state
90     virtual void StepToTimeout(TimerId timerId) = 0;
91 
92     // Step the SingleVerSyncStateMachine
93     virtual void SyncStep() = 0;
94 
95     // Called by SyncStep, Sub class should realize this function to do machine step
96     virtual void SyncStepInnerLocked() = 0;
97 
98     // Do state machine step with no lock, for inner use
99     virtual void SyncStepInner() = 0;
100 
101     // Called by StartSync, Sub class should realize this function to start statemachine
102     virtual int StartSyncInner() = 0;
103 
104     // Called by Abort, Sub class should realize this function to force abort statemachine
105     virtual void AbortInner() = 0;
106 
107     // while currentstate could not be found, should called, Sub class should realize this function.
108     virtual void SetCurStateErrStatus();
109 
110     // Used to get instance class' stateSwitchTables
111     virtual const std::vector<StateSwitchTable> &GetStateSwitchTables() const = 0;
112 
113     // Called by ExecNextTask, Sub class should realize this function to do some thing for run next sync task
114     virtual int PrepareNextSyncTask() = 0;
115 
116     // Called by StartSaveDataNotifyTimer, Sub class should realize this function to send a heartbeet packet
117     virtual void SendNotifyPacket(uint32_t sessionId, uint32_t sequenceId, uint32_t inMsgId) = 0;
118 
119     // Used to parse state table to switch machine state, this function must be called in stateMachineLock
120     int SwitchMachineState(uint8_t event);
121 
122     // Do state switch with the event, and do syncstep
123     virtual void SwitchStateAndStep(uint8_t event);
124 
125     // To Exec next sync task in context targetQueue
126     int ExecNextTask();
127 
128     // Start a watchdog used for manual sync, when begin a manual sync
129     int StartWatchDog();
130 
131     // Reset the watchdog used for manual sync
132     int ResetWatchDog();
133 
134     // stop a watchdog used for manual sync, call when sync finished,
135     void StopWatchDog();
136 
137     // Start a timer to send data notify packet to keep remote device not timeout
138     bool StartSaveDataNotify(uint32_t sessionId, uint32_t sequenceId, uint32_t inMsgId);
139 
140     // Stop send save data notify
141     void StopSaveDataNotify();
142 
143     // Stop send save data notify without lock
144     void StopSaveDataNotifyNoLock();
145 
146     // stop a timer to ResetWatchDog when sync data bigger than mtu  without lock
147     void StopFeedDogForSyncNoLock(SyncDirectionFlag flag);
148 
149     void DecRefCountOfFeedDogTimer(SyncDirectionFlag flag);
150 
151     virtual void DoSaveDataNotify(uint32_t sessionId, uint32_t sequenceId, uint32_t inMsgId);
152 
153     void DoFeedDogForSync(SyncDirectionFlag flag);
154 
155     void DoGetAndSendDataNotify(uint32_t sessionId);
156 
157     void StopFeedDogForGetDataInner(TimerId timerId);
158 
159     DISABLE_COPY_ASSIGN_MOVE(SyncStateMachine);
160 
161     ISyncTaskContext *syncContext_;
162     ISyncInterface *storageInterface_;
163     ICommunicator *communicator_;
164     std::shared_ptr<Metadata> metadata_;
165     std::mutex stateMachineLock_;
166     uint8_t currentState_;
167     volatile bool watchDogStarted_;
168     volatile uint32_t currentSyncProctolVersion_;
169 
170     // For save data notify
171     static const int DATA_NOTIFY_INTERVAL = 2000; // 2s for save/get data notify
172     static const int MAX_DATA_NOTIFY_COUNT = 15; // only notify 15 times
173     static const int SYNC_DIRECTION_NUM = 2; // send receive
174     std::mutex saveDataNotifyLock_;
175     TimerId saveDataNotifyTimerId_;
176     volatile uint8_t saveDataNotifyCount_;
177     volatile bool waitingResetLockBySaveData_;
178     volatile uint32_t saveDataNotifyRefCount_;
179 
180     std::mutex getDataNotifyLock_;
181     TimerId getDataNotifyTimerId_;
182     volatile uint8_t getDataNotifyCount_;
183 
184     // used for one (key,value) bigger than mtu size, in this case, send packet need more longger time
185     std::mutex feedDogLock_[SYNC_DIRECTION_NUM];
186     WatchDogController watchDogController_[SYNC_DIRECTION_NUM] = {{0}, {0}};
187 };
188 } // namespace DistributedDB
189 #endif // SYNC_STATE_MACHINE_H
190