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 KVSTORE_SYNC_MANAGER_H
17 #define KVSTORE_SYNC_MANAGER_H
18 
19 #include <atomic>
20 #include <list>
21 #include <map>
22 
23 #include "executor_pool.h"
24 #include "kv_store_nb_delegate.h"
25 #include "types.h"
26 
27 namespace OHOS {
28 namespace DistributedKv {
29 class API_EXPORT KvStoreSyncManager {
30 public:
31     static constexpr uint32_t SYNC_DEFAULT_DELAY_MS = 1000;
32     static constexpr uint32_t SYNC_MIN_DELAY_MS = 100;
33     static constexpr uint32_t SYNC_MAX_DELAY_MS = 1000 * 3600 * 24; // 24hours
34     static constexpr uint32_t SYNC_RETRY_MAX_COUNT = 3;
GetInstance()35     static KvStoreSyncManager *GetInstance()
36     {
37         static KvStoreSyncManager syncManager;
38         return &syncManager;
39     }
40     using TimePoint = std::chrono::steady_clock::time_point;
41     using SyncEnd = std::function<void(const std::map<std::string, DistributedDB::DBStatus> &)>;
42     using SyncFunc = std::function<Status(const SyncEnd &)>;
43 
44     struct KvSyncOperation {
45         uintptr_t syncId = 0;
46         uint32_t opSeq = 0;
47         uint32_t delayMs = 0;
48         SyncFunc syncFunc;
49         SyncEnd syncEnd;
50         TimePoint beginTime;
51     };
52     using OpPred = std::function<bool(KvSyncOperation &)>;
53     void SetThreadPool(std::shared_ptr<ExecutorPool> executors);
54     Status AddSyncOperation(uintptr_t syncId, uint32_t delayMs, const SyncFunc &syncFunc, const SyncEnd &syncEnd);
55     Status RemoveSyncOperation(uintptr_t syncId);
56 
57 private:
58     KvStoreSyncManager();
59     ~KvStoreSyncManager();
60 
61     uint32_t GetExpireTimeRange(uint32_t delayMs) const;
62     uint32_t DoRemoveSyncingOp(OpPred pred, std::list<KvSyncOperation> &syncingOps);
63     Status RemoveSyncingOp(uint32_t opSeq, std::list<KvSyncOperation> &syncingOps);
64     void AddTimer(const TimePoint &expireTime);
65     bool GetTimeoutSyncOps(const TimePoint &currentTime, std::list<KvSyncOperation> &syncOps);
66     void DoCheckSyncingTimeout(std::list<KvSyncOperation> &syncingOps);
67     void Schedule(const TimePoint &time);
68 
69     static constexpr uint32_t SYNCING_TIMEOUT_MS = 5000;
70     static constexpr uint32_t REALTIME_PRIOR_SYNCING_MS = 300;
71     static constexpr uint32_t DELAY_TIME_RANGE_DIVISOR = 4;
72 
73     mutable std::mutex syncOpsMutex_;
74     std::list<KvSyncOperation> realtimeSyncingOps_;
75     std::list<KvSyncOperation> delaySyncingOps_;
76     std::multimap<TimePoint, KvSyncOperation> scheduleSyncOps_;
77     std::shared_ptr<ExecutorPool> executors_;
78 
79     TimePoint nextScheduleTime_;
80     std::atomic_uint32_t syncOpSeq_ = 0;
81 };
82 } // namespace DistributedKv
83 } // namespace OHOS
84 #endif // KVSTORE_SYNC_MANAGER_H
85