1 //
2 //
3 // Copyright 2015-2016 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 <limits.h>
20 #include <string.h>
21
22 #include <algorithm>
23 #include <memory>
24 #include <string>
25 #include <utility>
26 #include <vector>
27
28 #include <grpc/grpc.h>
29 #include <grpc/impl/channel_arg_names.h>
30 #include <grpc/impl/compression_types.h>
31 #include <grpc/support/log.h>
32 #include <grpc/support/sync.h>
33 #include <grpc/support/workaround_list.h>
34 #include <grpcpp/completion_queue.h>
35 #include <grpcpp/impl/server_builder_option.h>
36 #include <grpcpp/impl/server_builder_plugin.h>
37 #include <grpcpp/impl/service_type.h>
38 #include <grpcpp/resource_quota.h>
39 #include <grpcpp/security/authorization_policy_provider.h>
40 #include <grpcpp/security/server_credentials.h>
41 #include <grpcpp/server.h>
42 #include <grpcpp/server_builder.h>
43 #include <grpcpp/server_context.h>
44 #include <grpcpp/server_interface.h>
45 #include <grpcpp/support/channel_arguments.h>
46 #include <grpcpp/support/server_interceptor.h>
47
48 #include "src/core/lib/gpr/string.h"
49 #include "src/core/lib/gpr/useful.h"
50 #include "src/cpp/server/external_connection_acceptor_impl.h"
51
52 namespace grpc {
53
54 static std::vector<std::unique_ptr<ServerBuilderPlugin> (*)()>*
55 g_plugin_factory_list;
56 static gpr_once once_init_plugin_list = GPR_ONCE_INIT;
57
do_plugin_list_init(void)58 static void do_plugin_list_init(void) {
59 g_plugin_factory_list =
60 new std::vector<std::unique_ptr<ServerBuilderPlugin> (*)()>();
61 }
62
ServerBuilder()63 ServerBuilder::ServerBuilder()
64 : max_receive_message_size_(INT_MIN),
65 max_send_message_size_(INT_MIN),
66 sync_server_settings_(SyncServerSettings()),
67 resource_quota_(nullptr) {
68 gpr_once_init(&once_init_plugin_list, do_plugin_list_init);
69 for (const auto& value : *g_plugin_factory_list) {
70 plugins_.emplace_back(value());
71 }
72
73 // all compression algorithms enabled by default.
74 enabled_compression_algorithms_bitset_ =
75 (1u << GRPC_COMPRESS_ALGORITHMS_COUNT) - 1;
76 memset(&maybe_default_compression_level_, 0,
77 sizeof(maybe_default_compression_level_));
78 memset(&maybe_default_compression_algorithm_, 0,
79 sizeof(maybe_default_compression_algorithm_));
80 }
81
~ServerBuilder()82 ServerBuilder::~ServerBuilder() {
83 if (resource_quota_ != nullptr) {
84 grpc_resource_quota_unref(resource_quota_);
85 }
86 }
87
AddCompletionQueue(bool is_frequently_polled)88 std::unique_ptr<grpc::ServerCompletionQueue> ServerBuilder::AddCompletionQueue(
89 bool is_frequently_polled) {
90 grpc::ServerCompletionQueue* cq = new grpc::ServerCompletionQueue(
91 GRPC_CQ_NEXT,
92 is_frequently_polled ? GRPC_CQ_DEFAULT_POLLING : GRPC_CQ_NON_LISTENING,
93 nullptr);
94 cqs_.push_back(cq);
95 return std::unique_ptr<grpc::ServerCompletionQueue>(cq);
96 }
97
RegisterService(Service * service)98 ServerBuilder& ServerBuilder::RegisterService(Service* service) {
99 services_.emplace_back(new NamedService(service));
100 return *this;
101 }
102
RegisterService(const std::string & host,Service * service)103 ServerBuilder& ServerBuilder::RegisterService(const std::string& host,
104 Service* service) {
105 services_.emplace_back(new NamedService(host, service));
106 return *this;
107 }
108
RegisterAsyncGenericService(AsyncGenericService * service)109 ServerBuilder& ServerBuilder::RegisterAsyncGenericService(
110 AsyncGenericService* service) {
111 if (generic_service_ || callback_generic_service_) {
112 gpr_log(GPR_ERROR,
113 "Adding multiple generic services is unsupported for now. "
114 "Dropping the service %p",
115 service);
116 } else {
117 generic_service_ = service;
118 }
119 return *this;
120 }
121
RegisterCallbackGenericService(CallbackGenericService * service)122 ServerBuilder& ServerBuilder::RegisterCallbackGenericService(
123 CallbackGenericService* service) {
124 if (generic_service_ || callback_generic_service_) {
125 gpr_log(GPR_ERROR,
126 "Adding multiple generic services is unsupported for now. "
127 "Dropping the service %p",
128 service);
129 } else {
130 callback_generic_service_ = service;
131 }
132 return *this;
133 }
134
SetContextAllocator(std::unique_ptr<grpc::ContextAllocator> context_allocator)135 ServerBuilder& ServerBuilder::SetContextAllocator(
136 std::unique_ptr<grpc::ContextAllocator> context_allocator) {
137 context_allocator_ = std::move(context_allocator);
138 return *this;
139 }
140
141 std::unique_ptr<grpc::experimental::ExternalConnectionAcceptor>
AddExternalConnectionAcceptor(experimental_type::ExternalConnectionType type,std::shared_ptr<ServerCredentials> creds)142 ServerBuilder::experimental_type::AddExternalConnectionAcceptor(
143 experimental_type::ExternalConnectionType type,
144 std::shared_ptr<ServerCredentials> creds) {
145 std::string name_prefix("external:");
146 char count_str[GPR_LTOA_MIN_BUFSIZE];
147 gpr_ltoa(static_cast<long>(builder_->acceptors_.size()), count_str);
148 builder_->acceptors_.emplace_back(
149 std::make_shared<grpc::internal::ExternalConnectionAcceptorImpl>(
150 name_prefix.append(count_str), type, creds));
151 return builder_->acceptors_.back()->GetAcceptor();
152 }
153
SetAuthorizationPolicyProvider(std::shared_ptr<experimental::AuthorizationPolicyProviderInterface> provider)154 void ServerBuilder::experimental_type::SetAuthorizationPolicyProvider(
155 std::shared_ptr<experimental::AuthorizationPolicyProviderInterface>
156 provider) {
157 builder_->authorization_provider_ = std::move(provider);
158 }
159
EnableCallMetricRecording(experimental::ServerMetricRecorder * server_metric_recorder)160 void ServerBuilder::experimental_type::EnableCallMetricRecording(
161 experimental::ServerMetricRecorder* server_metric_recorder) {
162 builder_->AddChannelArgument(GRPC_ARG_SERVER_CALL_METRIC_RECORDING, 1);
163 GPR_ASSERT(builder_->server_metric_recorder_ == nullptr);
164 builder_->server_metric_recorder_ = server_metric_recorder;
165 }
166
SetOption(std::unique_ptr<ServerBuilderOption> option)167 ServerBuilder& ServerBuilder::SetOption(
168 std::unique_ptr<ServerBuilderOption> option) {
169 options_.push_back(std::move(option));
170 return *this;
171 }
172
SetSyncServerOption(ServerBuilder::SyncServerOption option,int val)173 ServerBuilder& ServerBuilder::SetSyncServerOption(
174 ServerBuilder::SyncServerOption option, int val) {
175 switch (option) {
176 case NUM_CQS:
177 sync_server_settings_.num_cqs = val;
178 break;
179 case MIN_POLLERS:
180 sync_server_settings_.min_pollers = val;
181 break;
182 case MAX_POLLERS:
183 sync_server_settings_.max_pollers = val;
184 break;
185 case CQ_TIMEOUT_MSEC:
186 sync_server_settings_.cq_timeout_msec = val;
187 break;
188 }
189 return *this;
190 }
191
SetCompressionAlgorithmSupportStatus(grpc_compression_algorithm algorithm,bool enabled)192 ServerBuilder& ServerBuilder::SetCompressionAlgorithmSupportStatus(
193 grpc_compression_algorithm algorithm, bool enabled) {
194 if (enabled) {
195 grpc_core::SetBit(&enabled_compression_algorithms_bitset_, algorithm);
196 } else {
197 grpc_core::ClearBit(&enabled_compression_algorithms_bitset_, algorithm);
198 }
199 return *this;
200 }
201
SetDefaultCompressionLevel(grpc_compression_level level)202 ServerBuilder& ServerBuilder::SetDefaultCompressionLevel(
203 grpc_compression_level level) {
204 maybe_default_compression_level_.is_set = true;
205 maybe_default_compression_level_.level = level;
206 return *this;
207 }
208
SetDefaultCompressionAlgorithm(grpc_compression_algorithm algorithm)209 ServerBuilder& ServerBuilder::SetDefaultCompressionAlgorithm(
210 grpc_compression_algorithm algorithm) {
211 maybe_default_compression_algorithm_.is_set = true;
212 maybe_default_compression_algorithm_.algorithm = algorithm;
213 return *this;
214 }
215
SetResourceQuota(const grpc::ResourceQuota & resource_quota)216 ServerBuilder& ServerBuilder::SetResourceQuota(
217 const grpc::ResourceQuota& resource_quota) {
218 if (resource_quota_ != nullptr) {
219 grpc_resource_quota_unref(resource_quota_);
220 }
221 resource_quota_ = resource_quota.c_resource_quota();
222 grpc_resource_quota_ref(resource_quota_);
223 return *this;
224 }
225
AddListeningPort(const std::string & addr_uri,std::shared_ptr<ServerCredentials> creds,int * selected_port)226 ServerBuilder& ServerBuilder::AddListeningPort(
227 const std::string& addr_uri, std::shared_ptr<ServerCredentials> creds,
228 int* selected_port) {
229 const std::string uri_scheme = "dns:";
230 std::string addr = addr_uri;
231 if (addr_uri.compare(0, uri_scheme.size(), uri_scheme) == 0) {
232 size_t pos = uri_scheme.size();
233 while (addr_uri[pos] == '/') ++pos; // Skip slashes.
234 addr = addr_uri.substr(pos);
235 }
236 Port port = {addr, std::move(creds), selected_port};
237 ports_.push_back(port);
238 return *this;
239 }
240
BuildChannelArgs()241 ChannelArguments ServerBuilder::BuildChannelArgs() {
242 ChannelArguments args;
243 if (max_receive_message_size_ >= -1) {
244 args.SetInt(GRPC_ARG_MAX_RECEIVE_MESSAGE_LENGTH, max_receive_message_size_);
245 }
246 if (max_send_message_size_ >= -1) {
247 args.SetInt(GRPC_ARG_MAX_SEND_MESSAGE_LENGTH, max_send_message_size_);
248 }
249 for (const auto& option : options_) {
250 option->UpdateArguments(&args);
251 option->UpdatePlugins(&plugins_);
252 }
253 args.SetInt(GRPC_COMPRESSION_CHANNEL_ENABLED_ALGORITHMS_BITSET,
254 enabled_compression_algorithms_bitset_);
255 if (maybe_default_compression_level_.is_set) {
256 args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_LEVEL,
257 maybe_default_compression_level_.level);
258 }
259 if (maybe_default_compression_algorithm_.is_set) {
260 args.SetInt(GRPC_COMPRESSION_CHANNEL_DEFAULT_ALGORITHM,
261 maybe_default_compression_algorithm_.algorithm);
262 }
263 if (resource_quota_ != nullptr) {
264 args.SetPointerWithVtable(GRPC_ARG_RESOURCE_QUOTA, resource_quota_,
265 grpc_resource_quota_arg_vtable());
266 }
267 for (const auto& plugin : plugins_) {
268 plugin->UpdateServerBuilder(this);
269 plugin->UpdateChannelArguments(&args);
270 }
271 if (authorization_provider_ != nullptr) {
272 args.SetPointerWithVtable(GRPC_ARG_AUTHORIZATION_POLICY_PROVIDER,
273 authorization_provider_->c_provider(),
274 grpc_authorization_policy_provider_arg_vtable());
275 }
276 return args;
277 }
278
BuildAndStart()279 std::unique_ptr<grpc::Server> ServerBuilder::BuildAndStart() {
280 ChannelArguments args = BuildChannelArgs();
281
282 // == Determine if the server has any syncrhonous methods ==
283 bool has_sync_methods = false;
284 for (const auto& value : services_) {
285 if (value->service->has_synchronous_methods()) {
286 has_sync_methods = true;
287 break;
288 }
289 }
290
291 if (!has_sync_methods) {
292 for (const auto& value : plugins_) {
293 if (value->has_sync_methods()) {
294 has_sync_methods = true;
295 break;
296 }
297 }
298 }
299
300 // If this is a Sync server, i.e a server expositing sync API, then the server
301 // needs to create some completion queues to listen for incoming requests.
302 // 'sync_server_cqs' are those internal completion queues.
303 //
304 // This is different from the completion queues added to the server via
305 // ServerBuilder's AddCompletionQueue() method (those completion queues
306 // are in 'cqs_' member variable of ServerBuilder object)
307 std::shared_ptr<std::vector<std::unique_ptr<grpc::ServerCompletionQueue>>>
308 sync_server_cqs(
309 std::make_shared<
310 std::vector<std::unique_ptr<grpc::ServerCompletionQueue>>>());
311
312 bool has_frequently_polled_cqs = false;
313 for (const auto& cq : cqs_) {
314 if (cq->IsFrequentlyPolled()) {
315 has_frequently_polled_cqs = true;
316 break;
317 }
318 }
319
320 // == Determine if the server has any callback methods ==
321 bool has_callback_methods = false;
322 for (const auto& service : services_) {
323 if (service->service->has_callback_methods()) {
324 has_callback_methods = true;
325 has_frequently_polled_cqs = true;
326 break;
327 }
328 }
329
330 if (callback_generic_service_ != nullptr) {
331 has_frequently_polled_cqs = true;
332 }
333
334 const bool is_hybrid_server = has_sync_methods && has_frequently_polled_cqs;
335
336 if (has_sync_methods) {
337 grpc_cq_polling_type polling_type =
338 is_hybrid_server ? GRPC_CQ_NON_POLLING : GRPC_CQ_DEFAULT_POLLING;
339
340 // Create completion queues to listen to incoming rpc requests
341 for (int i = 0; i < sync_server_settings_.num_cqs; i++) {
342 sync_server_cqs->emplace_back(
343 new grpc::ServerCompletionQueue(GRPC_CQ_NEXT, polling_type, nullptr));
344 }
345 }
346
347 // TODO(vjpai): Add a section here for plugins once they can support callback
348 // methods
349
350 if (has_sync_methods) {
351 // This is a Sync server
352 gpr_log(GPR_INFO,
353 "Synchronous server. Num CQs: %d, Min pollers: %d, Max Pollers: "
354 "%d, CQ timeout (msec): %d",
355 sync_server_settings_.num_cqs, sync_server_settings_.min_pollers,
356 sync_server_settings_.max_pollers,
357 sync_server_settings_.cq_timeout_msec);
358 }
359
360 if (has_callback_methods) {
361 gpr_log(GPR_INFO, "Callback server.");
362 }
363
364 std::unique_ptr<grpc::Server> server(new grpc::Server(
365 &args, sync_server_cqs, sync_server_settings_.min_pollers,
366 sync_server_settings_.max_pollers, sync_server_settings_.cq_timeout_msec,
367 std::move(acceptors_), server_config_fetcher_, resource_quota_,
368 std::move(interceptor_creators_), server_metric_recorder_));
369
370 ServerInitializer* initializer = server->initializer();
371
372 // Register all the completion queues with the server. i.e
373 // 1. sync_server_cqs: internal completion queues created IF this is a sync
374 // server
375 // 2. cqs_: Completion queues added via AddCompletionQueue() call
376
377 for (const auto& cq : *sync_server_cqs) {
378 grpc_server_register_completion_queue(server->server_, cq->cq(), nullptr);
379 has_frequently_polled_cqs = true;
380 }
381
382 if (has_callback_methods || callback_generic_service_ != nullptr) {
383 auto* cq = server->CallbackCQ();
384 grpc_server_register_completion_queue(server->server_, cq->cq(), nullptr);
385 }
386
387 // cqs_ contains the completion queue added by calling the ServerBuilder's
388 // AddCompletionQueue() API. Some of them may not be frequently polled (i.e by
389 // calling Next() or AsyncNext()) and hence are not safe to be used for
390 // listening to incoming channels. Such completion queues must be registered
391 // as non-listening queues. In debug mode, these should have their server list
392 // tracked since these are provided the user and must be Shutdown by the user
393 // after the server is shutdown.
394 for (const auto& cq : cqs_) {
395 grpc_server_register_completion_queue(server->server_, cq->cq(), nullptr);
396 cq->RegisterServer(server.get());
397 }
398
399 if (!has_frequently_polled_cqs) {
400 gpr_log(GPR_ERROR,
401 "At least one of the completion queues must be frequently polled");
402 return nullptr;
403 }
404
405 server->RegisterContextAllocator(std::move(context_allocator_));
406
407 for (const auto& value : services_) {
408 if (!server->RegisterService(value->host.get(), value->service)) {
409 return nullptr;
410 }
411 }
412
413 for (const auto& value : plugins_) {
414 value->InitServer(initializer);
415 }
416
417 if (generic_service_) {
418 server->RegisterAsyncGenericService(generic_service_);
419 } else if (callback_generic_service_) {
420 server->RegisterCallbackGenericService(callback_generic_service_);
421 } else {
422 for (const auto& value : services_) {
423 if (value->service->has_generic_methods()) {
424 gpr_log(GPR_ERROR,
425 "Some methods were marked generic but there is no "
426 "generic service registered.");
427 return nullptr;
428 }
429 }
430 }
431
432 bool added_port = false;
433 for (auto& port : ports_) {
434 int r = server->AddListeningPort(port.addr, port.creds.get());
435 if (!r) {
436 if (added_port) server->Shutdown();
437 return nullptr;
438 }
439 added_port = true;
440 if (port.selected_port != nullptr) {
441 *port.selected_port = r;
442 }
443 }
444
445 auto cqs_data = cqs_.empty() ? nullptr : &cqs_[0];
446 server->Start(cqs_data, cqs_.size());
447
448 for (const auto& value : plugins_) {
449 value->Finish(initializer);
450 }
451
452 return server;
453 }
454
InternalAddPluginFactory(std::unique_ptr<ServerBuilderPlugin> (* CreatePlugin)())455 void ServerBuilder::InternalAddPluginFactory(
456 std::unique_ptr<ServerBuilderPlugin> (*CreatePlugin)()) {
457 gpr_once_init(&once_init_plugin_list, do_plugin_list_init);
458 (*g_plugin_factory_list).push_back(CreatePlugin);
459 }
460
EnableWorkaround(grpc_workaround_list id)461 ServerBuilder& ServerBuilder::EnableWorkaround(grpc_workaround_list id) {
462 switch (id) {
463 case GRPC_WORKAROUND_ID_CRONET_COMPRESSION:
464 return AddChannelArgument(GRPC_ARG_WORKAROUND_CRONET_COMPRESSION, 1);
465 default:
466 gpr_log(GPR_ERROR, "Workaround %u does not exist or is obsolete.", id);
467 return *this;
468 }
469 }
470
471 } // namespace grpc
472