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