1 //
2 //
3 // Copyright 2019 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 <mutex>
20 #include <thread>
21
22 #include <gtest/gtest.h>
23
24 #include <grpc/grpc.h>
25 #include <grpc/support/alloc.h>
26 #include <grpc/support/log.h>
27 #include <grpc/support/time.h>
28 #include <grpcpp/channel.h>
29 #include <grpcpp/client_context.h>
30 #include <grpcpp/create_channel.h>
31 #include <grpcpp/security/credentials.h>
32 #include <grpcpp/security/server_credentials.h>
33 #include <grpcpp/server.h>
34 #include <grpcpp/server_builder.h>
35 #include <grpcpp/server_context.h>
36
37 #include "src/core/lib/gprpp/crash.h"
38 #include "src/core/lib/gprpp/env.h"
39 #include "src/core/lib/iomgr/endpoint.h"
40 #include "src/core/lib/iomgr/exec_ctx.h"
41 #include "src/core/lib/iomgr/pollset.h"
42 #include "src/core/lib/iomgr/port.h"
43 #include "src/core/lib/iomgr/tcp_server.h"
44 #include "src/core/lib/security/credentials/credentials.h"
45 #include "src/proto/grpc/testing/echo.grpc.pb.h"
46 #include "test/core/util/port.h"
47 #include "test/core/util/test_config.h"
48 #include "test/core/util/test_tcp_server.h"
49 #include "test/cpp/end2end/test_service_impl.h"
50 #include "test/cpp/util/test_credentials_provider.h"
51
52 #ifdef GRPC_POSIX_SOCKET_TCP_SERVER
53
54 #include "src/core/lib/iomgr/tcp_posix.h"
55
56 namespace grpc {
57 namespace testing {
58 namespace {
59
60 class TestScenario {
61 public:
TestScenario(bool server_port,bool pending_data,const std::string & creds_type)62 TestScenario(bool server_port, bool pending_data,
63 const std::string& creds_type)
64 : server_has_port(server_port),
65 queue_pending_data(pending_data),
66 credentials_type(creds_type) {}
67 void Log() const;
68 // server has its own port or not
69 bool server_has_port;
70 // whether tcp server should read some data before handoff
71 bool queue_pending_data;
72 const std::string credentials_type;
73 };
74
operator <<(std::ostream & out,const TestScenario & scenario)75 std::ostream& operator<<(std::ostream& out, const TestScenario& scenario) {
76 return out << "TestScenario{server_has_port="
77 << (scenario.server_has_port ? "true" : "false")
78 << ", queue_pending_data="
79 << (scenario.queue_pending_data ? "true" : "false")
80 << ", credentials='" << scenario.credentials_type << "'}";
81 }
82
Log() const83 void TestScenario::Log() const {
84 std::ostringstream out;
85 out << *this;
86 gpr_log(GPR_ERROR, "%s", out.str().c_str());
87 }
88
89 // Set up a test tcp server which is in charge of accepting connections and
90 // handing off the connections as fds.
91 class TestTcpServer {
92 public:
TestTcpServer()93 TestTcpServer()
94 : shutdown_(false),
95 queue_data_(false),
96 port_(grpc_pick_unused_port_or_die()) {
97 std::ostringstream server_address;
98 server_address << "localhost:" << port_;
99 address_ = server_address.str();
100 test_tcp_server_init(&tcp_server_, &TestTcpServer::OnConnect, this);
101 GRPC_CLOSURE_INIT(&on_fd_released_, &TestTcpServer::OnFdReleased, this,
102 grpc_schedule_on_exec_ctx);
103 }
104
~TestTcpServer()105 ~TestTcpServer() {
106 running_thread_.join();
107 test_tcp_server_destroy(&tcp_server_);
108 grpc_recycle_unused_port(port_);
109 }
110
111 // Read some data before handing off the connection.
SetQueueData()112 void SetQueueData() { queue_data_ = true; }
113
Start()114 void Start() {
115 test_tcp_server_start(&tcp_server_, port_);
116 gpr_log(GPR_INFO, "Test TCP server started at %s", address_.c_str());
117 }
118
address()119 const std::string& address() { return address_; }
120
SetAcceptor(std::unique_ptr<experimental::ExternalConnectionAcceptor> acceptor)121 void SetAcceptor(
122 std::unique_ptr<experimental::ExternalConnectionAcceptor> acceptor) {
123 connection_acceptor_ = std::move(acceptor);
124 }
125
Run()126 void Run() {
127 running_thread_ = std::thread([this]() {
128 while (true) {
129 {
130 std::lock_guard<std::mutex> lock(mu_);
131 if (shutdown_) {
132 return;
133 }
134 }
135 test_tcp_server_poll(&tcp_server_, 1);
136 }
137 });
138 }
139
Shutdown()140 void Shutdown() {
141 std::lock_guard<std::mutex> lock(mu_);
142 shutdown_ = true;
143 }
144
OnConnect(void * arg,grpc_endpoint * tcp,grpc_pollset * accepting_pollset,grpc_tcp_server_acceptor * acceptor)145 static void OnConnect(void* arg, grpc_endpoint* tcp,
146 grpc_pollset* accepting_pollset,
147 grpc_tcp_server_acceptor* acceptor) {
148 auto* self = static_cast<TestTcpServer*>(arg);
149 self->OnConnect(tcp, accepting_pollset, acceptor);
150 }
151
OnFdReleased(void * arg,grpc_error_handle err)152 static void OnFdReleased(void* arg, grpc_error_handle err) {
153 auto* self = static_cast<TestTcpServer*>(arg);
154 self->OnFdReleased(err);
155 }
156
157 private:
OnConnect(grpc_endpoint * tcp,grpc_pollset *,grpc_tcp_server_acceptor * acceptor)158 void OnConnect(grpc_endpoint* tcp, grpc_pollset* /*accepting_pollset*/,
159 grpc_tcp_server_acceptor* acceptor) {
160 std::string peer(grpc_endpoint_get_peer(tcp));
161 gpr_log(GPR_INFO, "Got incoming connection! from %s", peer.c_str());
162 EXPECT_FALSE(acceptor->external_connection);
163 listener_fd_ = grpc_tcp_server_port_fd(
164 acceptor->from_server, acceptor->port_index, acceptor->fd_index);
165 gpr_free(acceptor);
166 grpc_tcp_destroy_and_release_fd(tcp, &fd_, &on_fd_released_);
167 }
168
OnFdReleased(grpc_error_handle err)169 void OnFdReleased(grpc_error_handle err) {
170 EXPECT_EQ(absl::OkStatus(), err);
171 experimental::ExternalConnectionAcceptor::NewConnectionParameters p;
172 p.listener_fd = listener_fd_;
173 p.fd = fd_;
174 if (queue_data_) {
175 char buf[1024];
176 ssize_t read_bytes = 0;
177 while (read_bytes <= 0) {
178 read_bytes = read(fd_, buf, 1024);
179 }
180 Slice data(buf, read_bytes);
181 p.read_buffer = ByteBuffer(&data, 1);
182 }
183 gpr_log(GPR_INFO, "Handing off fd %d with data size %d from listener fd %d",
184 fd_, static_cast<int>(p.read_buffer.Length()), listener_fd_);
185 connection_acceptor_->HandleNewConnection(&p);
186 }
187
188 std::mutex mu_;
189 bool shutdown_;
190
191 int listener_fd_ = -1;
192 int fd_ = -1;
193 bool queue_data_ = false;
194
195 grpc_closure on_fd_released_;
196 std::thread running_thread_;
197 int port_ = -1;
198 std::string address_;
199 std::unique_ptr<experimental::ExternalConnectionAcceptor>
200 connection_acceptor_;
201 test_tcp_server tcp_server_;
202 };
203
204 class PortSharingEnd2endTest : public ::testing::TestWithParam<TestScenario> {
205 protected:
PortSharingEnd2endTest()206 PortSharingEnd2endTest() : is_server_started_(false), first_picked_port_(0) {
207 GetParam().Log();
208 }
209
SetUp()210 void SetUp() override {
211 if (GetParam().queue_pending_data) {
212 tcp_server1_.SetQueueData();
213 tcp_server2_.SetQueueData();
214 }
215 tcp_server1_.Start();
216 tcp_server2_.Start();
217 ServerBuilder builder;
218 if (GetParam().server_has_port) {
219 int port = grpc_pick_unused_port_or_die();
220 first_picked_port_ = port;
221 server_address_ << "localhost:" << port;
222 auto creds = GetCredentialsProvider()->GetServerCredentials(
223 GetParam().credentials_type);
224 builder.AddListeningPort(server_address_.str(), creds);
225 gpr_log(GPR_INFO, "gRPC server listening on %s",
226 server_address_.str().c_str());
227 }
228 auto server_creds = GetCredentialsProvider()->GetServerCredentials(
229 GetParam().credentials_type);
230 auto acceptor1 = builder.experimental().AddExternalConnectionAcceptor(
231 ServerBuilder::experimental_type::ExternalConnectionType::FROM_FD,
232 server_creds);
233 tcp_server1_.SetAcceptor(std::move(acceptor1));
234 auto acceptor2 = builder.experimental().AddExternalConnectionAcceptor(
235 ServerBuilder::experimental_type::ExternalConnectionType::FROM_FD,
236 server_creds);
237 tcp_server2_.SetAcceptor(std::move(acceptor2));
238 builder.RegisterService(&service_);
239 server_ = builder.BuildAndStart();
240 is_server_started_ = true;
241
242 tcp_server1_.Run();
243 tcp_server2_.Run();
244 }
245
TearDown()246 void TearDown() override {
247 tcp_server1_.Shutdown();
248 tcp_server2_.Shutdown();
249 if (is_server_started_) {
250 server_->Shutdown();
251 }
252 if (first_picked_port_ > 0) {
253 grpc_recycle_unused_port(first_picked_port_);
254 }
255 }
256
ResetStubs()257 void ResetStubs() {
258 EXPECT_TRUE(is_server_started_);
259 ChannelArguments args;
260 args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
261 auto channel_creds = GetCredentialsProvider()->GetChannelCredentials(
262 GetParam().credentials_type, &args);
263 channel_handoff1_ =
264 CreateCustomChannel(tcp_server1_.address(), channel_creds, args);
265 stub_handoff1_ = EchoTestService::NewStub(channel_handoff1_);
266 channel_handoff2_ =
267 CreateCustomChannel(tcp_server2_.address(), channel_creds, args);
268 stub_handoff2_ = EchoTestService::NewStub(channel_handoff2_);
269 if (GetParam().server_has_port) {
270 ChannelArguments direct_args;
271 direct_args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
272 auto direct_creds = GetCredentialsProvider()->GetChannelCredentials(
273 GetParam().credentials_type, &direct_args);
274 channel_direct_ =
275 CreateCustomChannel(server_address_.str(), direct_creds, direct_args);
276 stub_direct_ = EchoTestService::NewStub(channel_direct_);
277 }
278 }
279
280 bool is_server_started_;
281 // channel/stub to the test tcp server, the connection will be handed to the
282 // grpc server.
283 std::shared_ptr<Channel> channel_handoff1_;
284 std::unique_ptr<EchoTestService::Stub> stub_handoff1_;
285 std::shared_ptr<Channel> channel_handoff2_;
286 std::unique_ptr<EchoTestService::Stub> stub_handoff2_;
287 // channel/stub to talk to the grpc server directly, if applicable.
288 std::shared_ptr<Channel> channel_direct_;
289 std::unique_ptr<EchoTestService::Stub> stub_direct_;
290 std::unique_ptr<Server> server_;
291 std::ostringstream server_address_;
292 TestServiceImpl service_;
293 TestTcpServer tcp_server1_;
294 TestTcpServer tcp_server2_;
295 int first_picked_port_;
296 };
297
SendRpc(EchoTestService::Stub * stub,int num_rpcs)298 void SendRpc(EchoTestService::Stub* stub, int num_rpcs) {
299 EchoRequest request;
300 EchoResponse response;
301 request.set_message("Hello hello hello hello");
302
303 for (int i = 0; i < num_rpcs; ++i) {
304 ClientContext context;
305 Status s = stub->Echo(&context, request, &response);
306 EXPECT_EQ(response.message(), request.message());
307 EXPECT_TRUE(s.ok());
308 }
309 }
310
CreateTestScenarios()311 std::vector<TestScenario> CreateTestScenarios() {
312 std::vector<TestScenario> scenarios;
313 std::vector<std::string> credentials_types;
314
315 #if TARGET_OS_IPHONE
316 // Workaround Apple CFStream bug
317 grpc_core::SetEnv("grpc_cfstream", "0");
318 #endif
319
320 credentials_types = GetCredentialsProvider()->GetSecureCredentialsTypeList();
321 // Only allow insecure credentials type when it is registered with the
322 // provider. User may create providers that do not have insecure.
323 if (GetCredentialsProvider()->GetChannelCredentials(kInsecureCredentialsType,
324 nullptr) != nullptr) {
325 credentials_types.push_back(kInsecureCredentialsType);
326 }
327
328 GPR_ASSERT(!credentials_types.empty());
329 for (const auto& cred : credentials_types) {
330 for (auto server_has_port : {true, false}) {
331 for (auto queue_pending_data : {true, false}) {
332 scenarios.emplace_back(server_has_port, queue_pending_data, cred);
333 }
334 }
335 }
336 return scenarios;
337 }
338
TEST_P(PortSharingEnd2endTest,HandoffAndDirectCalls)339 TEST_P(PortSharingEnd2endTest, HandoffAndDirectCalls) {
340 ResetStubs();
341 SendRpc(stub_handoff1_.get(), 5);
342 if (GetParam().server_has_port) {
343 SendRpc(stub_direct_.get(), 5);
344 }
345 }
346
TEST_P(PortSharingEnd2endTest,MultipleHandoff)347 TEST_P(PortSharingEnd2endTest, MultipleHandoff) {
348 for (int i = 0; i < 3; i++) {
349 ResetStubs();
350 SendRpc(stub_handoff2_.get(), 1);
351 }
352 }
353
TEST_P(PortSharingEnd2endTest,TwoHandoffPorts)354 TEST_P(PortSharingEnd2endTest, TwoHandoffPorts) {
355 for (int i = 0; i < 3; i++) {
356 ResetStubs();
357 SendRpc(stub_handoff1_.get(), 5);
358 SendRpc(stub_handoff2_.get(), 5);
359 }
360 }
361
362 INSTANTIATE_TEST_SUITE_P(PortSharingEnd2end, PortSharingEnd2endTest,
363 ::testing::ValuesIn(CreateTestScenarios()));
364
365 } // namespace
366 } // namespace testing
367 } // namespace grpc
368
369 #endif // GRPC_POSIX_SOCKET_TCP_SERVER
370
main(int argc,char ** argv)371 int main(int argc, char** argv) {
372 grpc::testing::TestEnvironment env(&argc, argv);
373 ::testing::InitGoogleTest(&argc, argv);
374 return RUN_ALL_TESTS();
375 }
376