1 //
2 //
3 // Copyright 2015 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 "test/core/test_util/port_server_client.h"
20
21 #include <grpc/credentials.h>
22 #include <grpc/grpc.h>
23 #include <grpc/grpc_security.h>
24 #include <grpc/support/alloc.h>
25 #include <grpc/support/port_platform.h>
26 #include <grpc/support/sync.h>
27 #include <grpc/support/time.h>
28 #include <stdint.h>
29 #include <stdlib.h>
30 #include <string.h>
31
32 #include <cmath>
33 #include <memory>
34 #include <string>
35 #include <utility>
36
37 #include "absl/log/check.h"
38 #include "absl/log/log.h"
39 #include "absl/status/statusor.h"
40 #include "absl/strings/str_format.h"
41 #include "src/core/lib/iomgr/closure.h"
42 #include "src/core/lib/iomgr/error.h"
43 #include "src/core/lib/iomgr/exec_ctx.h"
44 #include "src/core/lib/iomgr/iomgr_fwd.h"
45 #include "src/core/lib/iomgr/polling_entity.h"
46 #include "src/core/lib/iomgr/pollset.h"
47 #include "src/core/lib/security/credentials/credentials.h"
48 #include "src/core/util/http_client/httpcli.h"
49 #include "src/core/util/http_client/parser.h"
50 #include "src/core/util/orphanable.h"
51 #include "src/core/util/ref_counted_ptr.h"
52 #include "src/core/util/status_helper.h"
53 #include "src/core/util/time.h"
54 #include "src/core/util/uri.h"
55
56 typedef struct freereq {
57 gpr_mu* mu = nullptr;
58 grpc_polling_entity pops = {};
59 int done = 0;
60 } freereq;
61
destroy_pops_and_shutdown(void * p,grpc_error_handle)62 static void destroy_pops_and_shutdown(void* p, grpc_error_handle /*error*/) {
63 grpc_pollset* pollset =
64 grpc_polling_entity_pollset(static_cast<grpc_polling_entity*>(p));
65 grpc_pollset_destroy(pollset);
66 gpr_free(pollset);
67 }
68
freed_port_from_server(void * arg,grpc_error_handle)69 static void freed_port_from_server(void* arg, grpc_error_handle /*error*/) {
70 freereq* pr = static_cast<freereq*>(arg);
71 gpr_mu_lock(pr->mu);
72 pr->done = 1;
73 GRPC_LOG_IF_ERROR(
74 "pollset_kick",
75 grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
76 gpr_mu_unlock(pr->mu);
77 }
78
grpc_free_port_using_server(int port)79 void grpc_free_port_using_server(int port) {
80 grpc_http_request req;
81 grpc_http_response rsp;
82 freereq pr;
83 grpc_closure* shutdown_closure;
84
85 grpc_init();
86 {
87 grpc_core::ExecCtx exec_ctx;
88
89 pr = {};
90 memset(&req, 0, sizeof(req));
91 rsp = {};
92
93 grpc_pollset* pollset =
94 static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
95 grpc_pollset_init(pollset, &pr.mu);
96 pr.pops = grpc_polling_entity_create_from_pollset(pollset);
97 shutdown_closure = GRPC_CLOSURE_CREATE(destroy_pops_and_shutdown, &pr.pops,
98 grpc_schedule_on_exec_ctx);
99
100 std::string path = absl::StrFormat("/drop/%d", port);
101 auto uri = grpc_core::URI::Create("https", GRPC_PORT_SERVER_ADDRESS, path,
102 {} /* query params */, "" /* fragment */);
103 CHECK_OK(uri);
104 auto http_request = grpc_core::HttpRequest::Get(
105 std::move(*uri), nullptr /* channel args */, &pr.pops, &req,
106 grpc_core::Timestamp::Now() + grpc_core::Duration::Seconds(30),
107 GRPC_CLOSURE_CREATE(freed_port_from_server, &pr,
108 grpc_schedule_on_exec_ctx),
109 &rsp,
110 grpc_core::RefCountedPtr<grpc_channel_credentials>(
111 grpc_insecure_credentials_create()));
112 http_request->Start();
113 grpc_core::ExecCtx::Get()->Flush();
114 gpr_mu_lock(pr.mu);
115 while (!pr.done) {
116 grpc_pollset_worker* worker = nullptr;
117 if (!GRPC_LOG_IF_ERROR(
118 "pollset_work",
119 grpc_pollset_work(grpc_polling_entity_pollset(&pr.pops), &worker,
120 grpc_core::Timestamp::Now() +
121 grpc_core::Duration::Seconds(1)))) {
122 pr.done = 1;
123 }
124 }
125 gpr_mu_unlock(pr.mu);
126
127 grpc_pollset_shutdown(grpc_polling_entity_pollset(&pr.pops),
128 shutdown_closure);
129
130 grpc_http_response_destroy(&rsp);
131 }
132 grpc_shutdown();
133 }
134
135 typedef struct portreq {
136 gpr_mu* mu = nullptr;
137 grpc_polling_entity pops = {};
138 int port = 0;
139 int retries = 0;
140 char* server = nullptr;
141 grpc_http_response response = {};
142 grpc_core::OrphanablePtr<grpc_core::HttpRequest> http_request;
143 } portreq;
144
got_port_from_server(void * arg,grpc_error_handle error)145 static void got_port_from_server(void* arg, grpc_error_handle error) {
146 size_t i;
147 int port = 0;
148 portreq* pr = static_cast<portreq*>(arg);
149 pr->http_request.reset();
150 int failed = 0;
151 grpc_http_response* response = &pr->response;
152
153 if (!error.ok()) {
154 failed = 1;
155 VLOG(2) << "failed port pick from server: retrying ["
156 << grpc_core::StatusToString(error) << "]";
157 } else if (response->status != 200) {
158 failed = 1;
159 VLOG(2) << "failed port pick from server: status=" << response->status;
160 }
161
162 if (failed) {
163 grpc_http_request req;
164 memset(&req, 0, sizeof(req));
165 if (pr->retries >= 5) {
166 gpr_mu_lock(pr->mu);
167 pr->port = 0;
168 GRPC_LOG_IF_ERROR(
169 "pollset_kick",
170 grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
171 gpr_mu_unlock(pr->mu);
172 return;
173 }
174 CHECK(pr->retries < 10);
175 gpr_sleep_until(gpr_time_add(
176 gpr_now(GPR_CLOCK_REALTIME),
177 gpr_time_from_millis(
178 static_cast<int64_t>(
179 1000.0 * (1 + pow(1.3, pr->retries) * rand() / RAND_MAX)),
180 GPR_TIMESPAN)));
181 pr->retries++;
182 grpc_http_response_destroy(&pr->response);
183 pr->response = {};
184 auto uri = grpc_core::URI::Create("http", pr->server, "/get",
185 {} /* query params */, "" /* fragment */);
186 CHECK_OK(uri);
187 pr->http_request = grpc_core::HttpRequest::Get(
188 std::move(*uri), nullptr /* channel args */, &pr->pops, &req,
189 grpc_core::Timestamp::Now() + grpc_core::Duration::Seconds(30),
190 GRPC_CLOSURE_CREATE(got_port_from_server, pr,
191 grpc_schedule_on_exec_ctx),
192 &pr->response,
193 grpc_core::RefCountedPtr<grpc_channel_credentials>(
194 grpc_insecure_credentials_create()));
195 pr->http_request->Start();
196 return;
197 }
198 CHECK(response);
199 CHECK_EQ(response->status, 200);
200 for (i = 0; i < response->body_length; i++) {
201 CHECK(response->body[i] >= '0');
202 CHECK(response->body[i] <= '9');
203 port = port * 10 + response->body[i] - '0';
204 }
205 CHECK(port > 1024);
206 gpr_mu_lock(pr->mu);
207 pr->port = port;
208 GRPC_LOG_IF_ERROR(
209 "pollset_kick",
210 grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
211 gpr_mu_unlock(pr->mu);
212 }
213
grpc_pick_port_using_server(void)214 int grpc_pick_port_using_server(void) {
215 grpc_http_request req;
216 portreq pr;
217 grpc_closure* shutdown_closure;
218
219 grpc_init();
220 {
221 grpc_core::ExecCtx exec_ctx;
222 pr = {};
223 memset(&req, 0, sizeof(req));
224 grpc_pollset* pollset =
225 static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
226 grpc_pollset_init(pollset, &pr.mu);
227 pr.pops = grpc_polling_entity_create_from_pollset(pollset);
228 shutdown_closure = GRPC_CLOSURE_CREATE(destroy_pops_and_shutdown, &pr.pops,
229 grpc_schedule_on_exec_ctx);
230 pr.port = -1;
231 pr.server = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
232 auto uri = grpc_core::URI::Create("http", GRPC_PORT_SERVER_ADDRESS, "/get",
233 {} /* query params */, "" /* fragment */);
234 CHECK_OK(uri);
235 auto http_request = grpc_core::HttpRequest::Get(
236 std::move(*uri), nullptr /* channel args */, &pr.pops, &req,
237 grpc_core::Timestamp::Now() + grpc_core::Duration::Seconds(30),
238 GRPC_CLOSURE_CREATE(got_port_from_server, &pr,
239 grpc_schedule_on_exec_ctx),
240 &pr.response,
241 grpc_core::RefCountedPtr<grpc_channel_credentials>(
242 grpc_insecure_credentials_create()));
243 http_request->Start();
244 grpc_core::ExecCtx::Get()->Flush();
245 gpr_mu_lock(pr.mu);
246 while (pr.port == -1) {
247 grpc_pollset_worker* worker = nullptr;
248 if (!GRPC_LOG_IF_ERROR(
249 "pollset_work",
250 grpc_pollset_work(grpc_polling_entity_pollset(&pr.pops), &worker,
251 grpc_core::Timestamp::Now() +
252 grpc_core::Duration::Seconds(1)))) {
253 pr.port = 0;
254 }
255 }
256 gpr_mu_unlock(pr.mu);
257
258 grpc_http_response_destroy(&pr.response);
259 grpc_pollset_shutdown(grpc_polling_entity_pollset(&pr.pops),
260 shutdown_closure);
261
262 grpc_core::ExecCtx::Get()->Flush();
263 }
264 grpc_shutdown();
265
266 return pr.port;
267 }
268