1 // Copyright 2019 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
5 #include "platform/impl/socket_handle_waiter.h"
6
7 #include <algorithm>
8 #include <atomic>
9
10 #include "absl/algorithm/container.h"
11 #include "util/osp_logging.h"
12
13 namespace openscreen {
14
SocketHandleWaiter(ClockNowFunctionPtr now_function)15 SocketHandleWaiter::SocketHandleWaiter(ClockNowFunctionPtr now_function)
16 : now_function_(now_function) {}
17
Subscribe(Subscriber * subscriber,SocketHandleRef handle)18 void SocketHandleWaiter::Subscribe(Subscriber* subscriber,
19 SocketHandleRef handle) {
20 std::lock_guard<std::mutex> lock(mutex_);
21 if (handle_mappings_.find(handle) == handle_mappings_.end()) {
22 handle_mappings_.emplace(handle, SocketSubscription{subscriber});
23 }
24 }
25
Unsubscribe(Subscriber * subscriber,SocketHandleRef handle)26 void SocketHandleWaiter::Unsubscribe(Subscriber* subscriber,
27 SocketHandleRef handle) {
28 std::lock_guard<std::mutex> lock(mutex_);
29 auto iterator = handle_mappings_.find(handle);
30 if (handle_mappings_.find(handle) != handle_mappings_.end()) {
31 handle_mappings_.erase(iterator);
32 }
33 }
34
UnsubscribeAll(Subscriber * subscriber)35 void SocketHandleWaiter::UnsubscribeAll(Subscriber* subscriber) {
36 std::lock_guard<std::mutex> lock(mutex_);
37 for (auto it = handle_mappings_.begin(); it != handle_mappings_.end();) {
38 if (it->second.subscriber == subscriber) {
39 it = handle_mappings_.erase(it);
40 } else {
41 it++;
42 }
43 }
44 }
45
OnHandleDeletion(Subscriber * subscriber,SocketHandleRef handle,bool disable_locking_for_testing)46 void SocketHandleWaiter::OnHandleDeletion(Subscriber* subscriber,
47 SocketHandleRef handle,
48 bool disable_locking_for_testing) {
49 std::unique_lock<std::mutex> lock(mutex_);
50 auto it = handle_mappings_.find(handle);
51 if (it != handle_mappings_.end()) {
52 handle_mappings_.erase(it);
53 if (!disable_locking_for_testing) {
54 handles_being_deleted_.push_back(handle);
55
56 OSP_DVLOG << "Starting to block for handle deletion";
57 // This code will allow us to block completion of the socket destructor
58 // (and subsequent invalidation of pointers to this socket) until we no
59 // longer are waiting on a SELECT(...) call to it, since we only signal
60 // this condition variable's wait(...) to proceed outside of SELECT(...).
61 handle_deletion_block_.wait(lock, [this, handle]() {
62 return std::find(handles_being_deleted_.begin(),
63 handles_being_deleted_.end(),
64 handle) == handles_being_deleted_.end();
65 });
66 OSP_DVLOG << "\tDone blocking for handle deletion!";
67 }
68 }
69 }
70
ProcessReadyHandles(std::vector<HandleWithSubscription> * handles,Clock::duration timeout)71 void SocketHandleWaiter::ProcessReadyHandles(
72 std::vector<HandleWithSubscription>* handles,
73 Clock::duration timeout) {
74 if (handles->empty()) {
75 return;
76 }
77
78 Clock::time_point start_time = now_function_();
79 // Process the stalest handles one by one until we hit our timeout.
80 do {
81 Clock::time_point oldest_time = Clock::time_point::max();
82 HandleWithSubscription& oldest_handle = handles->at(0);
83 for (HandleWithSubscription& handle : *handles) {
84 // Skip already processed handles.
85 if (handle.subscription->last_updated >= start_time) {
86 continue;
87 }
88
89 // Select the oldest handle.
90 if (handle.subscription->last_updated < oldest_time) {
91 oldest_time = handle.subscription->last_updated;
92 oldest_handle = handle;
93 }
94 }
95
96 // Already processed all handles.
97 if (oldest_time == Clock::time_point::max()) {
98 return;
99 }
100
101 // Process the oldest handle.
102 oldest_handle.subscription->last_updated = now_function_();
103 oldest_handle.subscription->subscriber->ProcessReadyHandle(
104 oldest_handle.ready_handle.handle, oldest_handle.ready_handle.flags);
105 } while (now_function_() - start_time <= timeout);
106 }
107
ProcessHandles(Clock::duration timeout)108 Error SocketHandleWaiter::ProcessHandles(Clock::duration timeout) {
109 Clock::time_point start_time = now_function_();
110 std::vector<SocketHandleRef> handles;
111 {
112 std::lock_guard<std::mutex> lock(mutex_);
113 handles_being_deleted_.clear();
114 handle_deletion_block_.notify_all();
115 handles.reserve(handle_mappings_.size());
116 for (const auto& pair : handle_mappings_) {
117 handles.push_back(pair.first);
118 }
119 }
120
121 Clock::time_point current_time = now_function_();
122 Clock::duration remaining_timeout = timeout - (current_time - start_time);
123 ErrorOr<std::vector<ReadyHandle>> changed_handles =
124 AwaitSocketsReadable(handles, remaining_timeout);
125
126 std::vector<HandleWithSubscription> ready_handles;
127 {
128 std::lock_guard<std::mutex> lock(mutex_);
129 handles_being_deleted_.clear();
130 handle_deletion_block_.notify_all();
131 if (changed_handles) {
132 auto& ch = changed_handles.value();
133 ready_handles.reserve(ch.size());
134 for (const auto& handle : ch) {
135 auto mapping_it = handle_mappings_.find(handle.handle);
136 if (mapping_it != handle_mappings_.end()) {
137 ready_handles.push_back(
138 HandleWithSubscription{handle, &(mapping_it->second)});
139 }
140 }
141 }
142
143 if (changed_handles.is_error()) {
144 return changed_handles.error();
145 }
146
147 current_time = now_function_();
148 remaining_timeout = timeout - (current_time - start_time);
149 ProcessReadyHandles(&ready_handles, remaining_timeout);
150 }
151 return Error::None();
152 }
153
154 } // namespace openscreen
155