1 /*
2 * Copyright (c) 2021, The OpenThread Authors.
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are met:
7 * 1. Redistributions of source code must retain the above copyright
8 * notice, this list of conditions and the following disclaimer.
9 * 2. Redistributions in binary form must reproduce the above copyright
10 * notice, this list of conditions and the following disclaimer in the
11 * documentation and/or other materials provided with the distribution.
12 * 3. Neither the name of the copyright holder nor the
13 * names of its contributors may be used to endorse or promote products
14 * derived from this software without specific prior written permission.
15 *
16 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
17 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
18 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
19 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
20 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
21 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
22 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
23 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
24 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
25 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
26 * POSSIBILITY OF SUCH DAMAGE.
27 */
28
29 /**
30 * @file
31 * This file implements the Task Runner that executes tasks on the mainloop.
32 */
33
34 #include "common/task_runner.hpp"
35
36 #include <algorithm>
37
38 #include <fcntl.h>
39 #include <unistd.h>
40
41 #include "common/code_utils.hpp"
42
43 namespace otbr {
44
TaskRunner(void)45 TaskRunner::TaskRunner(void)
46 : mTaskQueue(DelayedTask::Comparator{})
47 {
48 int flags;
49
50 // We do not handle failures when creating a pipe, simply die.
51 VerifyOrDie(pipe(mEventFd) != -1, strerror(errno));
52
53 flags = fcntl(mEventFd[kRead], F_GETFL, 0);
54 VerifyOrDie(fcntl(mEventFd[kRead], F_SETFL, flags | O_NONBLOCK) != -1, strerror(errno));
55 flags = fcntl(mEventFd[kWrite], F_GETFL, 0);
56 VerifyOrDie(fcntl(mEventFd[kWrite], F_SETFL, flags | O_NONBLOCK) != -1, strerror(errno));
57 }
58
~TaskRunner(void)59 TaskRunner::~TaskRunner(void)
60 {
61 if (mEventFd[kRead] != -1)
62 {
63 close(mEventFd[kRead]);
64 mEventFd[kRead] = -1;
65 }
66 if (mEventFd[kWrite] != -1)
67 {
68 close(mEventFd[kWrite]);
69 mEventFd[kWrite] = -1;
70 }
71 }
72
Post(Task<void> aTask)73 void TaskRunner::Post(Task<void> aTask)
74 {
75 Post(Milliseconds::zero(), std::move(aTask));
76 }
77
Post(Milliseconds aDelay,Task<void> aTask)78 TaskRunner::TaskId TaskRunner::Post(Milliseconds aDelay, Task<void> aTask)
79 {
80 return PushTask(aDelay, std::move(aTask));
81 }
82
Update(MainloopContext & aMainloop)83 void TaskRunner::Update(MainloopContext &aMainloop)
84 {
85 aMainloop.AddFdToReadSet(mEventFd[kRead]);
86
87 {
88 std::lock_guard<std::mutex> _(mTaskQueueMutex);
89
90 if (!mTaskQueue.empty())
91 {
92 auto now = Clock::now();
93 auto &task = mTaskQueue.top();
94 auto delay = std::chrono::duration_cast<Microseconds>(task.GetTimeExecute() - now);
95 auto timeout = FromTimeval<Microseconds>(aMainloop.mTimeout);
96
97 if (task.GetTimeExecute() < now)
98 {
99 delay = Microseconds::zero();
100 }
101
102 if (delay <= timeout)
103 {
104 aMainloop.mTimeout.tv_sec = delay.count() / 1000000;
105 aMainloop.mTimeout.tv_usec = delay.count() % 1000000;
106 }
107 }
108 }
109 }
110
Process(const MainloopContext & aMainloop)111 void TaskRunner::Process(const MainloopContext &aMainloop)
112 {
113 OTBR_UNUSED_VARIABLE(aMainloop);
114
115 ssize_t rval;
116
117 // Read any data in the pipe.
118 do
119 {
120 uint8_t n;
121
122 rval = read(mEventFd[kRead], &n, sizeof(n));
123 } while (rval > 0 || (rval == -1 && errno == EINTR));
124
125 // Critical error happens, simply die.
126 VerifyOrDie(errno == EAGAIN || errno == EWOULDBLOCK, strerror(errno));
127
128 PopTasks();
129 }
130
PushTask(Milliseconds aDelay,Task<void> aTask)131 TaskRunner::TaskId TaskRunner::PushTask(Milliseconds aDelay, Task<void> aTask)
132 {
133 ssize_t rval;
134 const uint8_t kOne = 1;
135 TaskId taskId;
136
137 {
138 std::lock_guard<std::mutex> _(mTaskQueueMutex);
139
140 taskId = mNextTaskId++;
141
142 mActiveTaskIds.insert(taskId);
143 mTaskQueue.emplace(taskId, aDelay, std::move(aTask));
144 }
145
146 do
147 {
148 rval = write(mEventFd[kWrite], &kOne, sizeof(kOne));
149 } while (rval == -1 && errno == EINTR);
150
151 VerifyOrExit(rval == -1);
152
153 // Critical error happens, simply die.
154 VerifyOrDie(errno == EAGAIN || errno == EWOULDBLOCK, strerror(errno));
155
156 // We are blocked because there are already data (written by other concurrent callers in
157 // different threads) in the pipe, and the mEventFd[kRead] should be readable now.
158 otbrLogWarning("Failed to write fd %d: %s", mEventFd[kWrite], strerror(errno));
159
160 exit:
161 return taskId;
162 }
163
Cancel(TaskRunner::TaskId aTaskId)164 void TaskRunner::Cancel(TaskRunner::TaskId aTaskId)
165 {
166 std::lock_guard<std::mutex> _(mTaskQueueMutex);
167
168 mActiveTaskIds.erase(aTaskId);
169 }
170
PopTasks(void)171 void TaskRunner::PopTasks(void)
172 {
173 while (true)
174 {
175 Task<void> task;
176 bool canceled;
177
178 // The braces here are necessary for auto-releasing of the mutex.
179 {
180 std::lock_guard<std::mutex> _(mTaskQueueMutex);
181
182 if (!mTaskQueue.empty() && mTaskQueue.top().GetTimeExecute() <= Clock::now())
183 {
184 const DelayedTask &top = mTaskQueue.top();
185 TaskId taskId = top.mTaskId;
186
187 task = std::move(top.mTask);
188 mTaskQueue.pop();
189 canceled = (mActiveTaskIds.erase(taskId) == 0);
190 }
191 else
192 {
193 break;
194 }
195 }
196
197 if (!canceled)
198 {
199 task();
200 }
201 }
202 }
203
204 } // namespace otbr
205