1 /*
2 * Copyright (c) 2023 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 #define LOG_TAG "SchedulerManager"
16
17 #include "scheduler_manager.h"
18
19 #include "log_print.h"
20 #include "uri_utils.h"
21 #include "utils/anonymous.h"
22
23 namespace OHOS::DataShare {
24 static constexpr int64_t MAX_MILLISECONDS = 31536000000; // 365 days
GetInstance()25 SchedulerManager &SchedulerManager::GetInstance()
26 {
27 static SchedulerManager instance;
28 return instance;
29 }
30
Execute(const std::string & uri,const int32_t userId,const std::string & rdbDir,int version)31 void SchedulerManager::Execute(const std::string &uri, const int32_t userId, const std::string &rdbDir, int version)
32 {
33 if (!URIUtils::IsDataProxyURI(uri)) {
34 return;
35 }
36 auto delegate = DBDelegate::Create(rdbDir, version, true);
37 if (delegate == nullptr) {
38 ZLOGE("malloc fail %{public}s", DistributedData::Anonymous::Change(uri).c_str());
39 return;
40 }
41 std::vector<Key> keys = RdbSubscriberManager::GetInstance().GetKeysByUri(uri);
42 for (auto const &key : keys) {
43 ExecuteSchedulerSQL(rdbDir, userId, version, key, delegate);
44 }
45 }
46
Execute(const Key & key,const int32_t userId,const std::string & rdbDir,int version)47 void SchedulerManager::Execute(const Key &key, const int32_t userId, const std::string &rdbDir, int version)
48 {
49 auto delegate = DBDelegate::Create(rdbDir, version, true);
50 if (delegate == nullptr) {
51 ZLOGE("malloc fail %{public}s", DistributedData::Anonymous::Change(key.uri).c_str());
52 return;
53 }
54 ExecuteSchedulerSQL(rdbDir, userId, version, key, delegate);
55 }
56
SetTimer(const std::string & dbPath,const int32_t userId,int version,const Key & key,int64_t reminderTime)57 void SchedulerManager::SetTimer(
58 const std::string &dbPath, const int32_t userId, int version, const Key &key, int64_t reminderTime)
59 {
60 std::lock_guard<std::mutex> lock(mutex_);
61 if (executor_ == nullptr) {
62 ZLOGE("executor_ is nullptr");
63 return;
64 }
65 int64_t now = std::chrono::duration_cast<std::chrono::milliseconds>(
66 std::chrono::system_clock::now().time_since_epoch()).count();
67 // reminder time must is in future
68 if (reminderTime <= now || reminderTime - now >= MAX_MILLISECONDS) {
69 ZLOGE("reminderTime is not in future, %{public}" PRId64 "%{public}" PRId64, reminderTime, now);
70 return;
71 }
72 auto duration = std::chrono::milliseconds(reminderTime - now);
73 ZLOGI("reminderTime will notify in %{public}" PRId64 " milliseconds", reminderTime - now);
74 auto it = timerCache_.find(key);
75 if (it != timerCache_.end()) {
76 // has current timer, reset time
77 ZLOGD("has current taskId, uri is %{private}s, subscriberId is %{public}" PRId64 ", bundleName is %{public}s",
78 key.uri.c_str(), key.subscriberId, key.bundleName.c_str());
79 executor_->Reset(it->second, duration);
80 return;
81 }
82 // not find task in map, create new timer
83 auto taskId = executor_->Schedule(duration, [key, dbPath, version, userId, this]() {
84 ZLOGI("schedule notify start, uri is %{private}s, subscriberId is %{public}" PRId64 ", bundleName is "
85 "%{public}s", key.uri.c_str(), key.subscriberId, key.bundleName.c_str());
86 {
87 std::lock_guard<std::mutex> lock(mutex_);
88 timerCache_.erase(key);
89 }
90 // 1. execute schedulerSQL in next time
91 Execute(key, userId, dbPath, version);
92 // 2. notify
93 RdbSubscriberManager::GetInstance().EmitByKey(key, userId, dbPath, version);
94 });
95 if (taskId == ExecutorPool::INVALID_TASK_ID) {
96 ZLOGE("create timer failed, over the max capacity");
97 return;
98 }
99 ZLOGI("create new task success, uri is %{public}s, subscriberId is %{public}" PRId64 ", bundleName is %{public}s",
100 DistributedData::Anonymous::Change(key.uri).c_str(), key.subscriberId, key.bundleName.c_str());
101 timerCache_.emplace(key, taskId);
102 }
103
ExecuteSchedulerSQL(const std::string & rdbDir,const int32_t userId,int version,const Key & key,std::shared_ptr<DBDelegate> delegate)104 void SchedulerManager::ExecuteSchedulerSQL(const std::string &rdbDir, const int32_t userId, int version, const Key &key,
105 std::shared_ptr<DBDelegate> delegate)
106 {
107 Template tpl;
108 if (!TemplateManager::GetInstance().Get(key, userId, tpl)) {
109 ZLOGE("template undefined, %{public}s, %{public}" PRId64 ", %{public}s",
110 DistributedData::Anonymous::Change(key.uri).c_str(), key.subscriberId, key.bundleName.c_str());
111 return;
112 }
113 if (tpl.scheduler_.empty()) {
114 ZLOGW("template scheduler_ empty, %{public}s, %{public}" PRId64 ", %{public}s",
115 DistributedData::Anonymous::Change(key.uri).c_str(), key.subscriberId, key.bundleName.c_str());
116 return;
117 }
118 GenRemindTimerFuncParams(rdbDir, userId, version, key, tpl.scheduler_);
119 auto resultSet = delegate->QuerySql(tpl.scheduler_);
120 if (resultSet == nullptr) {
121 ZLOGE("resultSet is nullptr, %{public}s, %{public}" PRId64 ", %{public}s",
122 DistributedData::Anonymous::Change(key.uri).c_str(), key.subscriberId, key.bundleName.c_str());
123 return;
124 }
125 int count;
126 int errCode = resultSet->GetRowCount(count);
127 if (errCode != E_OK || count == 0) {
128 ZLOGE("GetRowCount error, %{public}s, %{public}" PRId64 ", %{public}s, errorCode is %{public}d, count is "
129 "%{public}d",
130 DistributedData::Anonymous::Change(key.uri).c_str(), key.subscriberId, key.bundleName.c_str(), errCode,
131 count);
132 return;
133 }
134 errCode = resultSet->GoToFirstRow();
135 if (errCode != E_OK) {
136 ZLOGE("GoToFirstRow error, %{public}s, %{public}" PRId64 ", %{public}s, errCode is %{public}d",
137 DistributedData::Anonymous::Change(key.uri).c_str(), key.subscriberId, key.bundleName.c_str(), errCode);
138 }
139 }
140
GenRemindTimerFuncParams(const std::string & rdbDir,const int32_t userId,int version,const Key & key,std::string & schedulerSQL)141 void SchedulerManager::GenRemindTimerFuncParams(
142 const std::string &rdbDir, const int32_t userId, int version, const Key &key, std::string &schedulerSQL)
143 {
144 auto index = schedulerSQL.find(REMIND_TIMER_FUNC);
145 if (index == std::string::npos) {
146 ZLOGW("not find remindTimer, sql is %{public}s", schedulerSQL.c_str());
147 return;
148 }
149 index += REMIND_TIMER_FUNC_LEN;
150 std::string keyStr = "'" + rdbDir + "', " + std::to_string(version) + ", '" + key.uri + "', " +
151 std::to_string(key.subscriberId) + ", '" + key.bundleName + "', " + std::to_string(userId) +
152 ", ";
153 schedulerSQL.insert(index, keyStr);
154 return;
155 }
156
RemoveTimer(const Key & key)157 void SchedulerManager::RemoveTimer(const Key &key)
158 {
159 std::lock_guard<std::mutex> lock(mutex_);
160 if (executor_ == nullptr) {
161 ZLOGE("executor_ is nullptr");
162 return;
163 }
164 auto it = timerCache_.find(key);
165 if (it != timerCache_.end()) {
166 ZLOGW("RemoveTimer %{public}s %{public}s %{public}" PRId64,
167 DistributedData::Anonymous::Change(key.uri).c_str(), key.bundleName.c_str(), key.subscriberId);
168 executor_->Remove(it->second);
169 timerCache_.erase(key);
170 }
171 }
172
ClearTimer()173 void SchedulerManager::ClearTimer()
174 {
175 ZLOGI("Clear all timer");
176 std::lock_guard<std::mutex> lock(mutex_);
177 if (executor_ == nullptr) {
178 ZLOGE("executor_ is nullptr");
179 return;
180 }
181 auto it = timerCache_.begin();
182 while (it != timerCache_.end()) {
183 executor_->Remove(it->second);
184 it = timerCache_.erase(it);
185 }
186 }
187
SetExecutorPool(std::shared_ptr<ExecutorPool> executor)188 void SchedulerManager::SetExecutorPool(std::shared_ptr<ExecutorPool> executor)
189 {
190 executor_ = executor;
191 }
192
ReExecuteAll()193 void SchedulerManager::ReExecuteAll()
194 {
195 std::lock_guard<std::mutex> lock(mutex_);
196 for (auto &item : timerCache_) {
197 // restart in 200ms
198 executor_->Reset(item.second, std::chrono::milliseconds(200));
199 }
200 }
201 } // namespace OHOS::DataShare
202
203