forwarders_manager.cc 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133
  1. // Copyright 2013 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 "tools/android/forwarder2/forwarders_manager.h"
  5. #include <stddef.h>
  6. #include <sys/select.h>
  7. #include <unistd.h>
  8. #include <algorithm>
  9. #include <iterator>
  10. #include <utility>
  11. #include "base/bind.h"
  12. #include "base/callback_helpers.h"
  13. #include "base/location.h"
  14. #include "base/logging.h"
  15. #include "base/memory/ptr_util.h"
  16. #include "base/posix/eintr_wrapper.h"
  17. #include "tools/android/forwarder2/forwarder.h"
  18. #include "tools/android/forwarder2/socket.h"
  19. namespace forwarder2 {
  20. ForwardersManager::ForwardersManager() : thread_("ForwardersManagerThread") {
  21. thread_.Start();
  22. WaitForEventsOnInternalThreadSoon();
  23. }
  24. ForwardersManager::~ForwardersManager() {
  25. deletion_notifier_.Notify();
  26. }
  27. void ForwardersManager::CreateAndStartNewForwarder(
  28. std::unique_ptr<Socket> socket1,
  29. std::unique_ptr<Socket> socket2) {
  30. // Note that the internal Forwarder vector is populated on the internal thread
  31. // which is the only thread from which it's accessed.
  32. thread_.task_runner()->PostTask(
  33. FROM_HERE,
  34. base::BindOnce(&ForwardersManager::CreateNewForwarderOnInternalThread,
  35. base::Unretained(this), std::move(socket1),
  36. std::move(socket2)));
  37. // Guarantees that the CreateNewForwarderOnInternalThread callback posted to
  38. // the internal thread gets executed immediately.
  39. wakeup_notifier_.Notify();
  40. }
  41. void ForwardersManager::CreateNewForwarderOnInternalThread(
  42. std::unique_ptr<Socket> socket1,
  43. std::unique_ptr<Socket> socket2) {
  44. DCHECK(thread_.task_runner()->RunsTasksInCurrentSequence());
  45. forwarders_.push_back(
  46. std::make_unique<Forwarder>(std::move(socket1), std::move(socket2)));
  47. }
  48. void ForwardersManager::WaitForEventsOnInternalThreadSoon() {
  49. thread_.task_runner()->PostTask(
  50. FROM_HERE,
  51. base::BindOnce(&ForwardersManager::WaitForEventsOnInternalThread,
  52. base::Unretained(this)));
  53. }
  54. void ForwardersManager::WaitForEventsOnInternalThread() {
  55. DCHECK(thread_.task_runner()->RunsTasksInCurrentSequence());
  56. fd_set read_fds;
  57. fd_set write_fds;
  58. FD_ZERO(&read_fds);
  59. FD_ZERO(&write_fds);
  60. // Populate the file descriptor sets.
  61. int max_fd = -1;
  62. for (const auto& forwarder : forwarders_)
  63. forwarder->RegisterFDs(&read_fds, &write_fds, &max_fd);
  64. const int notifier_fds[] = {
  65. wakeup_notifier_.receiver_fd(),
  66. deletion_notifier_.receiver_fd(),
  67. };
  68. for (size_t i = 0; i < std::size(notifier_fds); ++i) {
  69. const int notifier_fd = notifier_fds[i];
  70. DCHECK_GT(notifier_fd, -1);
  71. FD_SET(notifier_fd, &read_fds);
  72. max_fd = std::max(max_fd, notifier_fd);
  73. }
  74. const int ret = HANDLE_EINTR(
  75. select(max_fd + 1, &read_fds, &write_fds, NULL, NULL));
  76. if (ret < 0) {
  77. PLOG(ERROR) << "select";
  78. return;
  79. }
  80. const bool must_shutdown = FD_ISSET(
  81. deletion_notifier_.receiver_fd(), &read_fds);
  82. if (must_shutdown && forwarders_.empty())
  83. return;
  84. base::ScopedClosureRunner wait_for_events_soon(
  85. base::BindOnce(&ForwardersManager::WaitForEventsOnInternalThreadSoon,
  86. base::Unretained(this)));
  87. if (FD_ISSET(wakeup_notifier_.receiver_fd(), &read_fds)) {
  88. // Note that the events on FDs other than the wakeup notifier one, if any,
  89. // will be processed upon the next select().
  90. wakeup_notifier_.Reset();
  91. return;
  92. }
  93. // Notify the Forwarder instances and remove the ones that are closed.
  94. for (size_t i = 0; i < forwarders_.size(); ) {
  95. Forwarder* const forwarder = forwarders_[i].get();
  96. forwarder->ProcessEvents(read_fds, write_fds);
  97. if (must_shutdown)
  98. forwarder->Shutdown();
  99. if (!forwarder->IsClosed()) {
  100. ++i;
  101. continue;
  102. }
  103. std::swap(forwarders_[i], forwarders_.back());
  104. forwarders_.pop_back();
  105. }
  106. }
  107. } // namespace forwarder2