1 //
2 //
3 // Copyright 2017 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18
19 #include <grpc/credentials.h>
20 #include <grpc/grpc.h>
21 #include <grpc/grpc_security.h>
22 #include <grpc/impl/channel_arg_names.h>
23 #include <grpc/slice.h>
24 #include <grpc/slice_buffer.h>
25 #include <grpc/support/alloc.h>
26 #include <grpc/support/atm.h>
27 #include <grpc/support/sync.h>
28 #include <grpc/support/time.h>
29 #include <inttypes.h>
30
31 #include <functional>
32 #include <memory>
33 #include <string>
34 #include <thread>
35 #include <vector>
36
37 #include "absl/log/check.h"
38 #include "absl/log/log.h"
39 #include "absl/status/status.h"
40 #include "absl/status/statusor.h"
41 #include "absl/strings/str_cat.h"
42 #include "gtest/gtest.h"
43 #include "src/core/config/core_configuration.h"
44 #include "src/core/lib/channel/channel_args_preconditioning.h"
45 #include "src/core/lib/event_engine/channel_args_endpoint_config.h"
46 #include "src/core/lib/iomgr/closure.h"
47 #include "src/core/lib/iomgr/endpoint.h"
48 #include "src/core/lib/iomgr/error.h"
49 #include "src/core/lib/iomgr/exec_ctx.h"
50 #include "src/core/lib/iomgr/iomgr_fwd.h"
51 #include "src/core/lib/iomgr/pollset.h"
52 #include "src/core/lib/iomgr/pollset_set.h"
53 #include "src/core/lib/iomgr/resolve_address.h"
54 #include "src/core/lib/iomgr/resolved_address.h"
55 #include "src/core/lib/iomgr/tcp_client.h"
56 #include "src/core/lib/resource_quota/api.h"
57 #include "src/core/util/status_helper.h"
58 #include "src/core/util/time.h"
59 #include "test/core/test_util/port.h"
60 #include "test/core/test_util/test_config.h"
61
62 namespace grpc_core {
63 namespace test {
64 namespace {
65
66 // A gRPC server, running in its own thread.
67 class ServerThread {
68 public:
ServerThread(const char * address)69 explicit ServerThread(const char* address) : address_(address) {}
70
Start()71 void Start() {
72 // Start server with 1-second handshake timeout.
73 grpc_arg a[2];
74 a[0].type = GRPC_ARG_INTEGER;
75 a[0].key = const_cast<char*>(GRPC_ARG_SERVER_HANDSHAKE_TIMEOUT_MS);
76 a[0].value.integer = 1000;
77 a[1].key = const_cast<char*>(GRPC_ARG_RESOURCE_QUOTA);
78 a[1].type = GRPC_ARG_POINTER;
79 a[1].value.pointer.p = grpc_resource_quota_create("test");
80 a[1].value.pointer.vtable = grpc_resource_quota_arg_vtable();
81 grpc_channel_args args = {2, a};
82 server_ = grpc_server_create(&args, nullptr);
83 grpc_server_credentials* server_creds =
84 grpc_insecure_server_credentials_create();
85 ASSERT_TRUE(grpc_server_add_http2_port(server_, address_, server_creds));
86 grpc_server_credentials_release(server_creds);
87 cq_ = grpc_completion_queue_create_for_next(nullptr);
88 grpc_server_register_completion_queue(server_, cq_, nullptr);
89 grpc_server_start(server_);
90 thread_ =
91 std::make_unique<std::thread>(std::bind(&ServerThread::Serve, this));
92 grpc_resource_quota_unref(
93 static_cast<grpc_resource_quota*>(a[1].value.pointer.p));
94 }
95
Shutdown()96 void Shutdown() {
97 grpc_completion_queue* shutdown_cq =
98 grpc_completion_queue_create_for_pluck(nullptr);
99 grpc_server_shutdown_and_notify(server_, shutdown_cq, nullptr);
100 CHECK(grpc_completion_queue_pluck(shutdown_cq, nullptr,
101 grpc_timeout_seconds_to_deadline(1),
102 nullptr)
103 .type == GRPC_OP_COMPLETE);
104 grpc_completion_queue_destroy(shutdown_cq);
105 grpc_server_destroy(server_);
106 grpc_completion_queue_destroy(cq_);
107 thread_->join();
108 }
109
110 private:
Serve()111 void Serve() {
112 // The completion queue should not return anything other than shutdown.
113 grpc_event ev = grpc_completion_queue_next(
114 cq_, gpr_inf_future(GPR_CLOCK_MONOTONIC), nullptr);
115 ASSERT_EQ(GRPC_QUEUE_SHUTDOWN, ev.type);
116 }
117
118 const char* address_; // Do not own.
119 grpc_server* server_ = nullptr;
120 grpc_completion_queue* cq_ = nullptr;
121 std::unique_ptr<std::thread> thread_;
122 };
123
124 // A TCP client that connects to the server, reads data until the server
125 // closes, and then terminates.
126 class Client {
127 public:
Client(const char * server_address)128 explicit Client(const char* server_address)
129 : server_address_(server_address) {}
130
Connect()131 void Connect() {
132 ExecCtx exec_ctx;
133 absl::StatusOr<std::vector<grpc_resolved_address>> addresses_or =
134 GetDNSResolver()->LookupHostnameBlocking(server_address_, "80");
135 ASSERT_EQ(absl::OkStatus(), addresses_or.status())
136 << addresses_or.status().ToString();
137 ASSERT_GE(addresses_or->size(), 1UL);
138 pollset_ = static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
139 grpc_pollset_init(pollset_, &mu_);
140 grpc_pollset_set* pollset_set = grpc_pollset_set_create();
141 grpc_pollset_set_add_pollset(pollset_set, pollset_);
142 EventState state;
143 auto args = CoreConfiguration::Get()
144 .channel_args_preconditioning()
145 .PreconditionChannelArgs(nullptr);
146 grpc_tcp_client_connect(
147 state.closure(), &endpoint_, pollset_set,
148 grpc_event_engine::experimental::ChannelArgsEndpointConfig(args),
149 addresses_or->data(), Timestamp::Now() + Duration::Seconds(1));
150 ASSERT_TRUE(PollUntilDone(&state, Timestamp::InfFuture()));
151 ASSERT_EQ(absl::OkStatus(), state.error());
152 grpc_pollset_set_destroy(pollset_set);
153 grpc_endpoint_add_to_pollset(endpoint_, pollset_);
154 }
155
156 // Reads until an error is returned.
157 // Returns true if an error was encountered before the deadline.
ReadUntilError()158 bool ReadUntilError() {
159 ExecCtx exec_ctx;
160 grpc_slice_buffer read_buffer;
161 grpc_slice_buffer_init(&read_buffer);
162 bool retval = true;
163 // Use a deadline of 3 seconds, which is a lot more than we should
164 // need for a 1-second timeout, but this helps avoid flakes.
165 Timestamp deadline = Timestamp::Now() + Duration::Seconds(3);
166 while (true) {
167 EventState state;
168 grpc_endpoint_read(endpoint_, &read_buffer, state.closure(),
169 /*urgent=*/true, /*min_progress_size=*/1);
170 if (!PollUntilDone(&state, deadline)) {
171 retval = false;
172 break;
173 }
174 if (state.error() != absl::OkStatus()) break;
175 LOG(INFO) << "client read " << read_buffer.length << " bytes";
176 grpc_slice_buffer_reset_and_unref(&read_buffer);
177 }
178 grpc_endpoint_destroy(endpoint_);
179 endpoint_ = nullptr;
180 grpc_slice_buffer_destroy(&read_buffer);
181 return retval;
182 }
183
Shutdown()184 void Shutdown() {
185 ExecCtx exec_ctx;
186 if (endpoint_ != nullptr) grpc_endpoint_destroy(endpoint_);
187 grpc_pollset_shutdown(pollset_,
188 GRPC_CLOSURE_CREATE(&Client::PollsetDestroy, pollset_,
189 grpc_schedule_on_exec_ctx));
190 }
191
192 private:
193 // State used to wait for an I/O event.
194 class EventState {
195 public:
EventState()196 EventState() {
197 GRPC_CLOSURE_INIT(&closure_, &EventState::OnEventDone, this,
198 grpc_schedule_on_exec_ctx);
199 }
200
~EventState()201 ~EventState() {}
202
closure()203 grpc_closure* closure() { return &closure_; }
204
done() const205 bool done() const { return gpr_atm_acq_load(&done_atm_) != 0; }
206
207 // Caller does NOT take ownership of the error.
error() const208 grpc_error_handle error() const { return error_; }
209
210 private:
OnEventDone(void * arg,grpc_error_handle error)211 static void OnEventDone(void* arg, grpc_error_handle error) {
212 LOG(INFO) << "OnEventDone(): " << StatusToString(error);
213 EventState* state = static_cast<EventState*>(arg);
214 state->error_ = error;
215 gpr_atm_rel_store(&state->done_atm_, 1);
216 }
217
218 grpc_closure closure_;
219 gpr_atm done_atm_ = 0;
220 grpc_error_handle error_;
221 };
222
223 // Returns true if done, or false if deadline exceeded.
PollUntilDone(EventState * state,Timestamp deadline)224 bool PollUntilDone(EventState* state, Timestamp deadline) {
225 while (true) {
226 grpc_pollset_worker* worker = nullptr;
227 gpr_mu_lock(mu_);
228 GRPC_LOG_IF_ERROR(
229 "grpc_pollset_work",
230 grpc_pollset_work(pollset_, &worker,
231 Timestamp::Now() + Duration::Milliseconds(100)));
232 // Flushes any work scheduled before or during polling.
233 ExecCtx::Get()->Flush();
234 gpr_mu_unlock(mu_);
235 if (state != nullptr && state->done()) return true;
236 if (Timestamp::Now() >= deadline) return false;
237 }
238 }
239
PollsetDestroy(void * arg,grpc_error_handle)240 static void PollsetDestroy(void* arg, grpc_error_handle /*error*/) {
241 grpc_pollset* pollset = static_cast<grpc_pollset*>(arg);
242 grpc_pollset_destroy(pollset);
243 gpr_free(pollset);
244 }
245
246 const char* server_address_; // Do not own.
247 grpc_endpoint* endpoint_;
248 gpr_mu* mu_;
249 grpc_pollset* pollset_;
250 };
251
TEST(SettingsTimeout,Basic)252 TEST(SettingsTimeout, Basic) {
253 // Construct server address string.
254 const int server_port = grpc_pick_unused_port_or_die();
255 std::string server_address_string = absl::StrCat("localhost:", server_port);
256 // Start server.
257 LOG(INFO) << "starting server on " << server_address_string;
258 ServerThread server_thread(server_address_string.c_str());
259 server_thread.Start();
260 // Create client and connect to server.
261 LOG(INFO) << "starting client connect";
262 Client client(server_address_string.c_str());
263 client.Connect();
264 // Client read. Should fail due to server dropping connection.
265 LOG(INFO) << "starting client read";
266 EXPECT_TRUE(client.ReadUntilError());
267 // Shut down client.
268 LOG(INFO) << "shutting down client";
269 client.Shutdown();
270 // Shut down server.
271 LOG(INFO) << "shutting down server";
272 server_thread.Shutdown();
273 // Clean up.
274 }
275
276 } // namespace
277 } // namespace test
278 } // namespace grpc_core
279
main(int argc,char ** argv)280 int main(int argc, char** argv) {
281 ::testing::InitGoogleTest(&argc, argv);
282 grpc::testing::TestEnvironment env(&argc, argv);
283 grpc_init();
284 int result = RUN_ALL_TESTS();
285 grpc_shutdown();
286 return result;
287 }
288