message_pump_epoll.cc 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332
  1. // Copyright 2022 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include "base/message_loop/message_pump_epoll.h"
  5. #include <sys/epoll.h>
  6. #include <sys/eventfd.h>
  7. #include <algorithm>
  8. #include <cstddef>
  9. #include <cstdint>
  10. #include <utility>
  11. #include "base/auto_reset.h"
  12. #include "base/check_op.h"
  13. #include "base/memory/ref_counted.h"
  14. #include "base/posix/eintr_wrapper.h"
  15. #include "base/threading/thread_checker.h"
  16. #include "base/trace_event/base_tracing.h"
  17. #include "third_party/abseil-cpp/absl/types/optional.h"
  18. namespace base {
  19. MessagePumpEpoll::MessagePumpEpoll() {
  20. epoll_.reset(epoll_create(/*ignored_but_must_be_positive=*/1));
  21. PCHECK(epoll_.is_valid());
  22. wake_event_.reset(eventfd(0, EFD_NONBLOCK));
  23. PCHECK(wake_event_.is_valid());
  24. epoll_event wake{.events = EPOLLIN, .data = {.ptr = &wake_event_}};
  25. int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_ADD, wake_event_.get(), &wake);
  26. PCHECK(rv == 0);
  27. }
  28. MessagePumpEpoll::~MessagePumpEpoll() = default;
  29. bool MessagePumpEpoll::WatchFileDescriptor(int fd,
  30. bool persistent,
  31. int mode,
  32. FdWatchController* controller,
  33. FdWatcher* watcher) {
  34. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  35. TRACE_EVENT("base", "MessagePumpEpoll::WatchFileDescriptor", "fd", fd,
  36. "persistent", persistent, "watch_read", mode & WATCH_READ,
  37. "watch_write", mode & WATCH_WRITE);
  38. const InterestParams params{
  39. .fd = fd,
  40. .read = (mode == WATCH_READ || mode == WATCH_READ_WRITE),
  41. .write = (mode == WATCH_WRITE || mode == WATCH_READ_WRITE),
  42. .one_shot = !persistent,
  43. };
  44. auto [it, is_new_fd_entry] = entries_.emplace(fd, fd);
  45. EpollEventEntry& entry = it->second;
  46. scoped_refptr<Interest> existing_interest = controller->epoll_interest();
  47. if (existing_interest && existing_interest->params().IsEqual(params)) {
  48. // WatchFileDescriptor() has already been called for this controller at
  49. // least once before, and as in the most common cases, it is now being
  50. // called again with the same parameters.
  51. //
  52. // We don't need to allocate and register a new Interest in this case, but
  53. // we can instead reactivate the existing (presumably deactivated,
  54. // non-persistent) Interest.
  55. existing_interest->set_active(true);
  56. } else {
  57. entry.interests->push_back(controller->AssignEpollInterest(params));
  58. if (existing_interest) {
  59. UnregisterInterest(existing_interest);
  60. }
  61. }
  62. if (is_new_fd_entry) {
  63. AddEpollEvent(entry);
  64. } else {
  65. UpdateEpollEvent(entry);
  66. }
  67. controller->set_epoll_pump(weak_ptr_factory_.GetWeakPtr());
  68. controller->set_watcher(watcher);
  69. return true;
  70. }
  71. void MessagePumpEpoll::Run(Delegate* delegate) {
  72. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  73. RunState run_state(delegate);
  74. AutoReset<RunState*> auto_reset_run_state(&run_state_, &run_state);
  75. for (;;) {
  76. // Do some work and see if the next task is ready right away.
  77. Delegate::NextWorkInfo next_work_info = delegate->DoWork();
  78. const bool immediate_work_available = next_work_info.is_immediate();
  79. if (run_state.should_quit) {
  80. break;
  81. }
  82. // Process any immediately ready IO event, but don't wait for more yet.
  83. const bool processed_events = WaitForEpollEvent(TimeDelta());
  84. if (run_state.should_quit) {
  85. break;
  86. }
  87. if (immediate_work_available || processed_events) {
  88. continue;
  89. }
  90. const bool did_idle_work = delegate->DoIdleWork();
  91. if (run_state.should_quit) {
  92. break;
  93. }
  94. if (did_idle_work) {
  95. continue;
  96. }
  97. TimeDelta timeout = TimeDelta::Max();
  98. DCHECK(!next_work_info.delayed_run_time.is_null());
  99. if (!next_work_info.delayed_run_time.is_max()) {
  100. timeout = next_work_info.remaining_delay();
  101. }
  102. delegate->BeforeWait();
  103. WaitForEpollEvent(timeout);
  104. if (run_state.should_quit) {
  105. break;
  106. }
  107. }
  108. }
  109. void MessagePumpEpoll::Quit() {
  110. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  111. DCHECK(run_state_) << "Quit() called outside of Run()";
  112. run_state_->should_quit = true;
  113. }
  114. void MessagePumpEpoll::ScheduleWork() {
  115. const uint64_t value = 1;
  116. ssize_t n = HANDLE_EINTR(write(wake_event_.get(), &value, sizeof(value)));
  117. // EAGAIN here implies that the write() would overflow of the event counter,
  118. // which is a condition we can safely ignore. It implies that the event
  119. // counter is non-zero and therefore readable, which is enough to ensure that
  120. // any pending wait eventually wakes up.
  121. DPCHECK(n == sizeof(value) || errno == EAGAIN);
  122. }
  123. void MessagePumpEpoll::ScheduleDelayedWork(
  124. const Delegate::NextWorkInfo& next_work_info) {
  125. // Nothing to do. This can only be called from the same thread as Run(), so
  126. // the pump must be in between waits. The scheduled work therefore will be
  127. // seen in time for the next wait.
  128. }
  129. void MessagePumpEpoll::AddEpollEvent(EpollEventEntry& entry) {
  130. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  131. const uint32_t events = entry.ComputeActiveEvents();
  132. epoll_event event{.events = events, .data = {.ptr = &entry}};
  133. int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_ADD, entry.fd, &event);
  134. DPCHECK(rv == 0);
  135. entry.registered_events = events;
  136. }
  137. void MessagePumpEpoll::UpdateEpollEvent(EpollEventEntry& entry) {
  138. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  139. const uint32_t events = entry.ComputeActiveEvents();
  140. if (events == entry.registered_events && !(events & EPOLLONESHOT)) {
  141. // Persistent events don't need to be modified if no bits are changing.
  142. return;
  143. }
  144. epoll_event event{.events = events, .data = {.ptr = &entry}};
  145. int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_MOD, entry.fd, &event);
  146. DPCHECK(rv == 0);
  147. entry.registered_events = events;
  148. }
  149. void MessagePumpEpoll::UnregisterInterest(
  150. const scoped_refptr<Interest>& interest) {
  151. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  152. const int fd = interest->params().fd;
  153. auto entry_it = entries_.find(fd);
  154. DCHECK(entry_it != entries_.end());
  155. EpollEventEntry& entry = entry_it->second;
  156. auto& interests = entry.interests.container();
  157. auto it = std::find(interests.begin(), interests.end(), interest);
  158. DCHECK(it != interests.end());
  159. interests.erase(it);
  160. if (interests.empty()) {
  161. entries_.erase(entry_it);
  162. int rv = epoll_ctl(epoll_.get(), EPOLL_CTL_DEL, fd, nullptr);
  163. DPCHECK(rv == 0);
  164. } else {
  165. UpdateEpollEvent(entry);
  166. }
  167. }
  168. bool MessagePumpEpoll::WaitForEpollEvent(TimeDelta timeout) {
  169. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  170. const int epoll_timeout =
  171. timeout.is_max() ? -1 : saturated_cast<int>(timeout.InMilliseconds());
  172. epoll_event event;
  173. const int epoll_result =
  174. epoll_wait(epoll_.get(), &event, /*maxevents=*/1, epoll_timeout);
  175. if (epoll_result < 0) {
  176. DPCHECK(errno == EINTR);
  177. return false;
  178. }
  179. if (epoll_result == 0) {
  180. return false;
  181. }
  182. DPCHECK(epoll_result == 1);
  183. OnEpollEvent(event);
  184. return true;
  185. }
  186. void MessagePumpEpoll::OnEpollEvent(const epoll_event& e) {
  187. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  188. if (e.data.ptr == &wake_event_) {
  189. HandleWakeUp();
  190. return;
  191. }
  192. const bool readable = (e.events & EPOLLIN) != 0;
  193. const bool writable = (e.events & EPOLLOUT) != 0;
  194. // Under different circumstances, peer closure may raise both/either EPOLLHUP
  195. // and/or EPOLLERR. Treat them as equivalent.
  196. const bool disconnected = (e.events & (EPOLLHUP | EPOLLERR)) != 0;
  197. DCHECK(e.data.ptr);
  198. auto& entry = *static_cast<EpollEventEntry*>(e.data.ptr);
  199. // Copy the set of Interests, since interests may be added to or removed from
  200. // `entry` during the loop below. This copy is inexpensive in practice
  201. // because the size of this vector is expected to be very small (<= 2).
  202. auto interests = entry.interests;
  203. for (const auto& interest : interests.container()) {
  204. if (!interest->active()) {
  205. continue;
  206. }
  207. const bool can_read = (readable || disconnected) && interest->params().read;
  208. const bool can_write = writable && interest->params().write;
  209. if (!can_read && !can_write) {
  210. // If this Interest is active but not watching for whichever event was
  211. // raised here, there's nothing to do. This can occur if a descriptor has
  212. // multiple active interests, since only one interest needs to be
  213. // satisfied in order for us to process an epoll event.
  214. continue;
  215. }
  216. if (interest->params().one_shot) {
  217. // This is a one-shot event watch which is about to be triggered. We
  218. // deactivate the interest and update epoll immediately. The event handler
  219. // may reactivate it.
  220. interest->set_active(false);
  221. UpdateEpollEvent(entry);
  222. }
  223. HandleEvent(entry.fd, can_read, can_write, interest->controller());
  224. }
  225. }
  226. void MessagePumpEpoll::HandleEvent(int fd,
  227. bool can_read,
  228. bool can_write,
  229. FdWatchController* controller) {
  230. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  231. // Make the MessagePumpDelegate aware of this other form of "DoWork". Skip if
  232. // HandleNotification() is called outside of Run() (e.g. in unit tests).
  233. Delegate::ScopedDoWorkItem scoped_do_work_item;
  234. if (run_state_) {
  235. scoped_do_work_item = run_state_->delegate->BeginWorkItem();
  236. }
  237. // Trace events must begin after the above BeginWorkItem() so that the
  238. // ensuing "ThreadController active" outscopes all the events under it.
  239. TRACE_EVENT("toplevel", "EpollEvent", "controller_created_from",
  240. controller->created_from_location(), "fd", fd, "can_read",
  241. can_read, "can_write", can_write, "context",
  242. static_cast<void*>(controller));
  243. TRACE_HEAP_PROFILER_API_SCOPED_TASK_EXECUTION heap_profiler_scope(
  244. controller->created_from_location().file_name());
  245. if (can_read && can_write) {
  246. bool controller_was_destroyed = false;
  247. controller->was_destroyed_ = &controller_was_destroyed;
  248. controller->OnFdWritable();
  249. if (!controller_was_destroyed) {
  250. controller->OnFdReadable();
  251. }
  252. if (!controller_was_destroyed) {
  253. controller->was_destroyed_ = nullptr;
  254. }
  255. } else if (can_write) {
  256. controller->OnFdWritable();
  257. } else if (can_read) {
  258. controller->OnFdReadable();
  259. }
  260. }
  261. void MessagePumpEpoll::HandleWakeUp() {
  262. DCHECK_CALLED_ON_VALID_THREAD(thread_checker_);
  263. uint64_t value;
  264. ssize_t n = HANDLE_EINTR(read(wake_event_.get(), &value, sizeof(value)));
  265. DPCHECK(n == sizeof(value));
  266. }
  267. MessagePumpEpoll::EpollEventEntry::EpollEventEntry(int fd) : fd(fd) {}
  268. MessagePumpEpoll::EpollEventEntry::~EpollEventEntry() = default;
  269. uint32_t MessagePumpEpoll::EpollEventEntry::ComputeActiveEvents() {
  270. uint32_t events = 0;
  271. bool one_shot = true;
  272. for (const auto& interest : interests.container()) {
  273. if (!interest->active()) {
  274. continue;
  275. }
  276. const InterestParams& params = interest->params();
  277. events |= (params.read ? EPOLLIN : 0) | (params.write ? EPOLLOUT : 0);
  278. one_shot &= params.one_shot;
  279. }
  280. if (events != 0 && one_shot) {
  281. return events | EPOLLONESHOT;
  282. }
  283. return events;
  284. }
  285. } // namespace base