aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorGravatar Nicolas "Pixel" Noble <pixel@nobis-crew.org>2016-01-29 23:07:31 +0100
committerGravatar Nicolas "Pixel" Noble <pixel@nobis-crew.org>2016-01-29 23:07:31 +0100
commitefa1f167271f09179ba04f7c453f6302e2c0cbd2 (patch)
tree7c490ea409b9528442fd9138bb886cfca8057d81 /src
parenteade6e02d2ae59cb4aeb3b718eebbc41bf7b738f (diff)
parent86d8f5ca71ec176682281f63421852f0c16ebe91 (diff)
Merge branch 'master' of github.com:grpc/grpc into make-ruby-installable
Diffstat (limited to 'src')
-rw-r--r--src/compiler/cpp_generator.cc350
-rw-r--r--src/compiler/python_generator.cc1
-rw-r--r--src/core/census/grpc_filter.c4
-rw-r--r--src/core/channel/channel_args.c7
-rw-r--r--src/core/channel/client_channel.c23
-rw-r--r--src/core/channel/client_uchannel.c6
-rw-r--r--src/core/channel/compress_filter.c6
-rw-r--r--src/core/channel/http_client_filter.c4
-rw-r--r--src/core/channel/http_server_filter.c4
-rw-r--r--src/core/channel/subchannel_call_holder.c18
-rw-r--r--src/core/client_config/lb_policies/pick_first.c18
-rw-r--r--src/core/client_config/lb_policies/round_robin.c14
-rw-r--r--src/core/client_config/resolvers/dns_resolver.c6
-rw-r--r--src/core/client_config/resolvers/sockaddr_resolver.c6
-rw-r--r--src/core/client_config/resolvers/zookeeper_resolver.c4
-rw-r--r--src/core/client_config/subchannel.c18
-rw-r--r--src/core/httpcli/httpcli.c10
-rw-r--r--src/core/iomgr/closure.c8
-rw-r--r--src/core/iomgr/closure.h9
-rw-r--r--src/core/iomgr/exec_ctx.c18
-rw-r--r--src/core/iomgr/exec_ctx.h17
-rw-r--r--src/core/iomgr/executor.c8
-rw-r--r--src/core/iomgr/executor.h4
-rw-r--r--src/core/iomgr/fd_posix.c6
-rw-r--r--src/core/iomgr/iocp_windows.c6
-rw-r--r--src/core/iomgr/pollset_multipoller_with_epoll.c6
-rw-r--r--src/core/iomgr/pollset_posix.c13
-rw-r--r--src/core/iomgr/pollset_windows.c4
-rw-r--r--src/core/iomgr/resolve_address_posix.c4
-rw-r--r--src/core/iomgr/resolve_address_windows.c4
-rw-r--r--src/core/iomgr/tcp_client_posix.c14
-rw-r--r--src/core/iomgr/tcp_client_windows.c8
-rw-r--r--src/core/iomgr/tcp_posix.c14
-rw-r--r--src/core/iomgr/tcp_server_posix.c9
-rw-r--r--src/core/iomgr/tcp_server_windows.c6
-rw-r--r--src/core/iomgr/tcp_windows.c22
-rw-r--r--src/core/iomgr/timer.c6
-rw-r--r--src/core/iomgr/udp_server.c12
-rw-r--r--src/core/iomgr/udp_server.h10
-rw-r--r--src/core/iomgr/workqueue.h6
-rw-r--r--src/core/iomgr/workqueue_posix.c6
-rw-r--r--src/core/security/credentials.c4
-rw-r--r--src/core/security/google_default_credentials.c4
-rw-r--r--src/core/security/handshake.c11
-rw-r--r--src/core/security/secure_endpoint.c10
-rw-r--r--src/core/security/server_auth_filter.c4
-rw-r--r--src/core/security/server_secure_chttp2.c2
-rw-r--r--src/core/surface/call.c22
-rw-r--r--src/core/surface/channel.c6
-rw-r--r--src/core/surface/channel_connectivity.c6
-rw-r--r--src/core/surface/channel_create.c6
-rw-r--r--src/core/surface/channel_ping.c4
-rw-r--r--src/core/surface/completion_queue.c6
-rw-r--r--src/core/surface/lame_client.c6
-rw-r--r--src/core/surface/secure_channel_create.c6
-rw-r--r--src/core/surface/server.c36
-rw-r--r--src/core/surface/server_chttp2.c2
-rw-r--r--src/core/surface/server_create.c5
-rw-r--r--src/core/transport/chttp2/internal.h2
-rw-r--r--src/core/transport/chttp2/writing.c4
-rw-r--r--src/core/transport/chttp2_transport.c34
-rw-r--r--src/core/transport/connectivity_state.c12
-rw-r--r--src/core/transport/transport.c10
-rw-r--r--src/core/transport/transport.h15
-rw-r--r--src/cpp/client/channel.cc17
-rw-r--r--src/cpp/client/client_context.cc1
-rw-r--r--src/cpp/client/create_channel.cc6
-rw-r--r--src/cpp/client/credentials.cc8
-rw-r--r--src/cpp/client/secure_credentials.cc15
-rw-r--r--src/cpp/client/secure_credentials.h16
-rw-r--r--src/cpp/codegen/grpc_library.cc40
-rw-r--r--src/cpp/common/alarm.cc11
-rw-r--r--src/cpp/common/completion_queue.cc8
-rw-r--r--src/cpp/server/server.cc84
-rw-r--r--src/cpp/server/server_builder.cc57
-rw-r--r--src/cpp/server/server_context.cc2
-rw-r--r--src/node/performance/benchmark_server.js2
-rw-r--r--src/node/performance/worker_service_impl.js4
-rw-r--r--src/node/src/grpc_extension.js4
-rw-r--r--src/node/test/surface_test.js8
-rw-r--r--src/objective-c/BoringSSL.podspec4
-rw-r--r--src/proto/grpc/testing/control.proto2
-rw-r--r--src/python/README.md100
-rw-r--r--src/ruby/ext/grpc/rb_call.c3
84 files changed, 689 insertions, 639 deletions
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc
index 3c8ca8ab45..d3d6224de2 100644
--- a/src/compiler/cpp_generator.cc
+++ b/src/compiler/cpp_generator.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -112,18 +112,17 @@ grpc::string GetHeaderPrologue(const grpc::protobuf::FileDescriptor *file,
grpc::string GetHeaderIncludes(const grpc::protobuf::FileDescriptor *file,
const Parameters &params) {
grpc::string temp =
- "#include <grpc++/support/async_stream.h>\n"
- "#include <grpc++/impl/rpc_method.h>\n"
- "#include <grpc++/impl/proto_utils.h>\n"
- "#include <grpc++/impl/service_type.h>\n"
- "#include <grpc++/support/async_unary_call.h>\n"
- "#include <grpc++/support/status.h>\n"
- "#include <grpc++/support/stub_options.h>\n"
- "#include <grpc++/support/sync_stream.h>\n"
+ "#include <grpc++/impl/codegen/async_stream.h>\n"
+ "#include <grpc++/impl/codegen/async_unary_call.h>\n"
+ "#include <grpc++/impl/codegen/proto_utils.h>\n"
+ "#include <grpc++/impl/codegen/rpc_method.h>\n"
+ "#include <grpc++/impl/codegen/service_type.h>\n"
+ "#include <grpc++/impl/codegen/status.h>\n"
+ "#include <grpc++/impl/codegen/stub_options.h>\n"
+ "#include <grpc++/impl/codegen/sync_stream.h>\n"
"\n"
"namespace grpc {\n"
"class CompletionQueue;\n"
- "class Channel;\n"
"class RpcService;\n"
"class ServerCompletionQueue;\n"
"class ServerContext;\n"
@@ -491,39 +490,186 @@ void PrintHeaderServerMethodAsync(
grpc_cpp_generator::ClassName(method->input_type(), true);
(*vars)["Response"] =
grpc_cpp_generator::ClassName(method->output_type(), true);
+ printer->Print(*vars, "template <class BaseClass>\n");
+ printer->Print(*vars,
+ "class WithAsyncMethod_$Method$ : public BaseClass {\n");
+ printer->Print(
+ " private:\n"
+ " void BaseClassMustBeDerivedFromService(Service *service) {}\n");
+ printer->Print(" public:\n");
+ printer->Indent();
+ printer->Print(*vars,
+ "WithAsyncMethod_$Method$() {\n"
+ " ::grpc::Service::MarkMethodAsync($Idx$);\n"
+ "}\n");
+ printer->Print(*vars,
+ "~WithAsyncMethod_$Method$() GRPC_OVERRIDE {\n"
+ " BaseClassMustBeDerivedFromService(this);\n"
+ "}\n");
if (NoStreaming(method)) {
printer->Print(
*vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, const $Request$* request, "
+ "$Response$* response) GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ printer->Print(
+ *vars,
"void Request$Method$("
"::grpc::ServerContext* context, $Request$* request, "
"::grpc::ServerAsyncResponseWriter< $Response$>* response, "
"::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag);\n");
+ "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
+ printer->Print(*vars,
+ " ::grpc::Service::RequestAsyncUnary($Idx$, context, "
+ "request, response, new_call_cq, notification_cq, tag);\n");
+ printer->Print("}\n");
} else if (ClientOnlyStreaming(method)) {
printer->Print(
*vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerReader< $Request$>* reader, "
+ "$Response$* response) GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ printer->Print(
+ *vars,
"void Request$Method$("
"::grpc::ServerContext* context, "
"::grpc::ServerAsyncReader< $Response$, $Request$>* reader, "
"::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag);\n");
+ "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
+ printer->Print(*vars,
+ " ::grpc::Service::RequestAsyncClientStreaming($Idx$, "
+ "context, reader, new_call_cq, notification_cq, tag);\n");
+ printer->Print("}\n");
} else if (ServerOnlyStreaming(method)) {
printer->Print(
*vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, const $Request$* request, "
+ "::grpc::ServerWriter< $Response$>* writer) GRPC_FINAL GRPC_OVERRIDE "
+ "{\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ printer->Print(
+ *vars,
"void Request$Method$("
"::grpc::ServerContext* context, $Request$* request, "
"::grpc::ServerAsyncWriter< $Response$>* writer, "
"::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag);\n");
+ "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
+ printer->Print(
+ *vars,
+ " ::grpc::Service::RequestAsyncServerStreaming($Idx$, "
+ "context, request, writer, new_call_cq, notification_cq, tag);\n");
+ printer->Print("}\n");
} else if (BidiStreaming(method)) {
printer->Print(
*vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerReaderWriter< $Response$, $Request$>* stream) "
+ "GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ printer->Print(
+ *vars,
"void Request$Method$("
"::grpc::ServerContext* context, "
"::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
"::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag);\n");
+ "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
+ printer->Print(*vars,
+ " ::grpc::Service::RequestAsyncBidiStreaming($Idx$, "
+ "context, stream, new_call_cq, notification_cq, tag);\n");
+ printer->Print("}\n");
+ }
+ printer->Outdent();
+ printer->Print(*vars, "};\n");
+}
+
+void PrintHeaderServerMethodGeneric(
+ grpc::protobuf::io::Printer *printer,
+ const grpc::protobuf::MethodDescriptor *method,
+ std::map<grpc::string, grpc::string> *vars) {
+ (*vars)["Method"] = method->name();
+ (*vars)["Request"] =
+ grpc_cpp_generator::ClassName(method->input_type(), true);
+ (*vars)["Response"] =
+ grpc_cpp_generator::ClassName(method->output_type(), true);
+ printer->Print(*vars, "template <class BaseClass>\n");
+ printer->Print(*vars,
+ "class WithGenericMethod_$Method$ : public BaseClass {\n");
+ printer->Print(
+ " private:\n"
+ " void BaseClassMustBeDerivedFromService(Service *service) {}\n");
+ printer->Print(" public:\n");
+ printer->Indent();
+ printer->Print(*vars,
+ "WithGenericMethod_$Method$() {\n"
+ " ::grpc::Service::MarkMethodGeneric($Idx$);\n"
+ "}\n");
+ printer->Print(*vars,
+ "~WithGenericMethod_$Method$() GRPC_OVERRIDE {\n"
+ " BaseClassMustBeDerivedFromService(this);\n"
+ "}\n");
+ if (NoStreaming(method)) {
+ printer->Print(
+ *vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, const $Request$* request, "
+ "$Response$* response) GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ } else if (ClientOnlyStreaming(method)) {
+ printer->Print(
+ *vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerReader< $Request$>* reader, "
+ "$Response$* response) GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ } else if (ServerOnlyStreaming(method)) {
+ printer->Print(
+ *vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, const $Request$* request, "
+ "::grpc::ServerWriter< $Response$>* writer) GRPC_FINAL GRPC_OVERRIDE "
+ "{\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
+ } else if (BidiStreaming(method)) {
+ printer->Print(
+ *vars,
+ "// disable synchronous version of this method\n"
+ "::grpc::Status $Method$("
+ "::grpc::ServerContext* context, "
+ "::grpc::ServerReaderWriter< $Response$, $Request$>* stream) "
+ "GRPC_FINAL GRPC_OVERRIDE {\n"
+ " abort();\n"
+ " return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, \"\");\n"
+ "}\n");
}
+ printer->Outdent();
+ printer->Print(*vars, "};\n");
}
void PrintHeaderService(grpc::protobuf::io::Printer *printer,
@@ -557,14 +703,14 @@ void PrintHeaderService(grpc::protobuf::io::Printer *printer,
"class Stub GRPC_FINAL : public StubInterface"
" {\n public:\n");
printer->Indent();
- printer->Print("Stub(const std::shared_ptr< ::grpc::Channel>& channel);\n");
+ printer->Print("Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel);\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethod(printer, service->method(i), vars, true);
}
printer->Outdent();
printer->Print("\n private:\n");
printer->Indent();
- printer->Print("std::shared_ptr< ::grpc::Channel> channel_;\n");
+ printer->Print("std::shared_ptr< ::grpc::ChannelInterface> channel_;\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderClientMethod(printer, service->method(i), vars, false);
}
@@ -575,14 +721,14 @@ void PrintHeaderService(grpc::protobuf::io::Printer *printer,
printer->Print("};\n");
printer->Print(
"static std::unique_ptr<Stub> NewStub(const std::shared_ptr< "
- "::grpc::Channel>& channel, "
+ "::grpc::ChannelInterface>& channel, "
"const ::grpc::StubOptions& options = ::grpc::StubOptions());\n");
printer->Print("\n");
- // Server side - Synchronous
+ // Server side - base
printer->Print(
- "class Service : public ::grpc::SynchronousService {\n"
+ "class Service : public ::grpc::Service {\n"
" public:\n");
printer->Indent();
printer->Print("Service();\n");
@@ -590,26 +736,32 @@ void PrintHeaderService(grpc::protobuf::io::Printer *printer,
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderServerMethodSync(printer, service->method(i), vars);
}
- printer->Print("::grpc::RpcService* service() GRPC_OVERRIDE GRPC_FINAL;\n");
printer->Outdent();
- printer->Print(
- " private:\n"
- " std::unique_ptr< ::grpc::RpcService> service_;\n");
printer->Print("};\n");
// Server side - Asynchronous
- printer->Print(
- "class AsyncService GRPC_FINAL : public ::grpc::AsynchronousService {\n"
- " public:\n");
- printer->Indent();
- (*vars)["MethodCount"] = as_string(service->method_count());
- printer->Print("explicit AsyncService();\n");
- printer->Print("~AsyncService() {};\n");
for (int i = 0; i < service->method_count(); ++i) {
+ (*vars)["Idx"] = as_string(i);
PrintHeaderServerMethodAsync(printer, service->method(i), vars);
}
- printer->Outdent();
- printer->Print("};\n");
+
+ printer->Print("typedef ");
+
+ for (int i = 0; i < service->method_count(); ++i) {
+ (*vars)["method_name"] = service->method(i)->name();
+ printer->Print(*vars, "WithAsyncMethod_$method_name$<");
+ }
+ printer->Print("Service");
+ for (int i = 0; i < service->method_count(); ++i) {
+ printer->Print(" >");
+ }
+ printer->Print(" AsyncService;\n");
+
+ // Server side - Generic
+ for (int i = 0; i < service->method_count(); ++i) {
+ (*vars)["Idx"] = as_string(i);
+ PrintHeaderServerMethodGeneric(printer, service->method(i), vars);
+ }
printer->Outdent();
printer->Print("};\n");
@@ -623,6 +775,12 @@ grpc::string GetHeaderServices(const grpc::protobuf::FileDescriptor *file,
grpc::protobuf::io::StringOutputStream output_stream(&output);
grpc::protobuf::io::Printer printer(&output_stream, '$');
std::map<grpc::string, grpc::string> vars;
+ // Package string is empty or ends with a dot. It is used to fully qualify
+ // method names.
+ vars["Package"] = file->package();
+ if (!file->package().empty()) {
+ vars["Package"].append(".");
+ }
if (!params.services_namespace.empty()) {
vars["services_namespace"] = params.services_namespace;
@@ -702,13 +860,16 @@ grpc::string GetSourceIncludes(const grpc::protobuf::FileDescriptor *file,
grpc::protobuf::io::Printer printer(&output_stream, '$');
std::map<grpc::string, grpc::string> vars;
- printer.Print(vars, "#include <grpc++/channel.h>\n");
- printer.Print(vars, "#include <grpc++/impl/client_unary_call.h>\n");
- printer.Print(vars, "#include <grpc++/impl/rpc_service_method.h>\n");
- printer.Print(vars, "#include <grpc++/impl/service_type.h>\n");
- printer.Print(vars, "#include <grpc++/support/async_unary_call.h>\n");
- printer.Print(vars, "#include <grpc++/support/async_stream.h>\n");
- printer.Print(vars, "#include <grpc++/support/sync_stream.h>\n");
+ printer.Print(vars, "#include <grpc++/impl/codegen/async_stream.h>\n");
+ printer.Print(vars, "#include <grpc++/impl/codegen/async_unary_call.h>\n");
+ printer.Print(vars, "#include <grpc++/impl/codegen/channel_interface.h>\n");
+ printer.Print(vars, "#include <grpc++/impl/codegen/client_unary_call.h>\n");
+ printer.Print(vars,
+ "#include <grpc++/impl/codegen/method_handler_impl.h>\n");
+ printer.Print(vars,
+ "#include <grpc++/impl/codegen/rpc_service_method.h>\n");
+ printer.Print(vars, "#include <grpc++/impl/codegen/service_type.h>\n");
+ printer.Print(vars, "#include <grpc++/impl/codegen/sync_stream.h>\n");
if (!file->package().empty()) {
std::vector<grpc::string> parts =
@@ -889,69 +1050,6 @@ void PrintSourceServerMethod(grpc::protobuf::io::Printer *printer,
}
}
-void PrintSourceServerAsyncMethod(
- grpc::protobuf::io::Printer *printer,
- const grpc::protobuf::MethodDescriptor *method,
- std::map<grpc::string, grpc::string> *vars) {
- (*vars)["Method"] = method->name();
- (*vars)["Request"] =
- grpc_cpp_generator::ClassName(method->input_type(), true);
- (*vars)["Response"] =
- grpc_cpp_generator::ClassName(method->output_type(), true);
- if (NoStreaming(method)) {
- printer->Print(
- *vars,
- "void $ns$$Service$::AsyncService::Request$Method$("
- "::grpc::ServerContext* context, "
- "$Request$* request, "
- "::grpc::ServerAsyncResponseWriter< $Response$>* response, "
- "::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
- printer->Print(*vars,
- " AsynchronousService::RequestAsyncUnary($Idx$, context, "
- "request, response, new_call_cq, notification_cq, tag);\n");
- printer->Print("}\n\n");
- } else if (ClientOnlyStreaming(method)) {
- printer->Print(
- *vars,
- "void $ns$$Service$::AsyncService::Request$Method$("
- "::grpc::ServerContext* context, "
- "::grpc::ServerAsyncReader< $Response$, $Request$>* reader, "
- "::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
- printer->Print(*vars,
- " AsynchronousService::RequestClientStreaming($Idx$, "
- "context, reader, new_call_cq, notification_cq, tag);\n");
- printer->Print("}\n\n");
- } else if (ServerOnlyStreaming(method)) {
- printer->Print(
- *vars,
- "void $ns$$Service$::AsyncService::Request$Method$("
- "::grpc::ServerContext* context, "
- "$Request$* request, "
- "::grpc::ServerAsyncWriter< $Response$>* writer, "
- "::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
- printer->Print(
- *vars,
- " AsynchronousService::RequestServerStreaming($Idx$, "
- "context, request, writer, new_call_cq, notification_cq, tag);\n");
- printer->Print("}\n\n");
- } else if (BidiStreaming(method)) {
- printer->Print(
- *vars,
- "void $ns$$Service$::AsyncService::Request$Method$("
- "::grpc::ServerContext* context, "
- "::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
- "::grpc::CompletionQueue* new_call_cq, "
- "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
- printer->Print(*vars,
- " AsynchronousService::RequestBidiStreaming($Idx$, "
- "context, stream, new_call_cq, notification_cq, tag);\n");
- printer->Print("}\n\n");
- }
-}
-
void PrintSourceService(grpc::protobuf::io::Printer *printer,
const grpc::protobuf::ServiceDescriptor *service,
std::map<grpc::string, grpc::string> *vars) {
@@ -967,7 +1065,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
printer->Print(*vars,
"std::unique_ptr< $ns$$Service$::Stub> $ns$$Service$::NewStub("
- "const std::shared_ptr< ::grpc::Channel>& channel, "
+ "const std::shared_ptr< ::grpc::ChannelInterface>& channel, "
"const ::grpc::StubOptions& options) {\n"
" std::unique_ptr< $ns$$Service$::Stub> stub(new "
"$ns$$Service$::Stub(channel));\n"
@@ -975,7 +1073,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
"}\n\n");
printer->Print(*vars,
"$ns$$Service$::Stub::Stub(const std::shared_ptr< "
- "::grpc::Channel>& channel)\n");
+ "::grpc::ChannelInterface>& channel)\n");
printer->Indent();
printer->Print(": channel_(channel)");
for (int i = 0; i < service->method_count(); ++i) {
@@ -1006,32 +1104,8 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
PrintSourceClientMethod(printer, service->method(i), vars);
}
- (*vars)["MethodCount"] = as_string(service->method_count());
- printer->Print(*vars,
- "$ns$$Service$::AsyncService::AsyncService() : "
- "::grpc::AsynchronousService("
- "$prefix$$Service$_method_names, $MethodCount$) "
- "{}\n\n");
-
- printer->Print(*vars,
- "$ns$$Service$::Service::Service() {\n"
- "}\n\n");
- printer->Print(*vars,
- "$ns$$Service$::Service::~Service() {\n"
- "}\n\n");
- for (int i = 0; i < service->method_count(); ++i) {
- (*vars)["Idx"] = as_string(i);
- PrintSourceServerMethod(printer, service->method(i), vars);
- PrintSourceServerAsyncMethod(printer, service->method(i), vars);
- }
- printer->Print(*vars,
- "::grpc::RpcService* $ns$$Service$::Service::service() {\n");
+ printer->Print(*vars, "$ns$$Service$::Service::Service() {\n");
printer->Indent();
- printer->Print(
- "if (service_) {\n"
- " return service_.get();\n"
- "}\n");
- printer->Print("service_ = std::unique_ptr< ::grpc::RpcService>(new ::grpc::RpcService());\n");
for (int i = 0; i < service->method_count(); ++i) {
const grpc::protobuf::MethodDescriptor *method = service->method(i);
(*vars)["Idx"] = as_string(i);
@@ -1043,7 +1117,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
if (NoStreaming(method)) {
printer->Print(
*vars,
- "service_->AddMethod(new ::grpc::RpcServiceMethod(\n"
+ "AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::RpcMethod::NORMAL_RPC,\n"
" new ::grpc::RpcMethodHandler< $ns$$Service$::Service, "
@@ -1053,7 +1127,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
} else if (ClientOnlyStreaming(method)) {
printer->Print(
*vars,
- "service_->AddMethod(new ::grpc::RpcServiceMethod(\n"
+ "AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::RpcMethod::CLIENT_STREAMING,\n"
" new ::grpc::ClientStreamingHandler< "
@@ -1062,7 +1136,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
} else if (ServerOnlyStreaming(method)) {
printer->Print(
*vars,
- "service_->AddMethod(new ::grpc::RpcServiceMethod(\n"
+ "AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::RpcMethod::SERVER_STREAMING,\n"
" new ::grpc::ServerStreamingHandler< "
@@ -1071,7 +1145,7 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
} else if (BidiStreaming(method)) {
printer->Print(
*vars,
- "service_->AddMethod(new ::grpc::RpcServiceMethod(\n"
+ "AddMethod(new ::grpc::RpcServiceMethod(\n"
" $prefix$$Service$_method_names[$Idx$],\n"
" ::grpc::RpcMethod::BIDI_STREAMING,\n"
" new ::grpc::BidiStreamingHandler< "
@@ -1079,9 +1153,15 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
" std::mem_fn(&$ns$$Service$::Service::$Method$), this)));\n");
}
}
- printer->Print("return service_.get();\n");
printer->Outdent();
- printer->Print("}\n\n");
+ printer->Print(*vars, "}\n\n");
+ printer->Print(*vars,
+ "$ns$$Service$::Service::~Service() {\n"
+ "}\n\n");
+ for (int i = 0; i < service->method_count(); ++i) {
+ (*vars)["Idx"] = as_string(i);
+ PrintSourceServerMethod(printer, service->method(i), vars);
+ }
}
grpc::string GetSourceServices(const grpc::protobuf::FileDescriptor *file,
diff --git a/src/compiler/python_generator.cc b/src/compiler/python_generator.cc
index c754ae299b..4fd5dfbecb 100644
--- a/src/compiler/python_generator.cc
+++ b/src/compiler/python_generator.cc
@@ -42,7 +42,6 @@
#include <tuple>
#include <vector>
-#include <grpc++/support/config.h>
#include "src/compiler/config.h"
#include "src/compiler/generator_helpers.h"
#include "src/compiler/python_generator.h"
diff --git a/src/core/census/grpc_filter.c b/src/core/census/grpc_filter.c
index ee1f62f8c2..a8db32b9d5 100644
--- a/src/core/census/grpc_filter.c
+++ b/src/core/census/grpc_filter.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -91,7 +91,7 @@ static void client_start_transport_op(grpc_exec_ctx *exec_ctx,
}
static void server_on_done_recv(grpc_exec_ctx *exec_ctx, void *ptr,
- int success) {
+ bool success) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
diff --git a/src/core/channel/channel_args.c b/src/core/channel/channel_args.c
index 487db1119a..0427ce0b8d 100644
--- a/src/core/channel/channel_args.c
+++ b/src/core/channel/channel_args.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -35,6 +35,7 @@
#include "src/core/channel/channel_args.h"
#include "src/core/support/string.h"
+#include <grpc/census.h>
#include <grpc/support/alloc.h>
#include <grpc/support/string_util.h>
#include <grpc/support/useful.h>
@@ -119,10 +120,10 @@ int grpc_channel_args_is_census_enabled(const grpc_channel_args *a) {
if (a == NULL) return 0;
for (i = 0; i < a->num_args; i++) {
if (0 == strcmp(a->args[i].key, GRPC_ARG_ENABLE_CENSUS)) {
- return a->args[i].value.integer != 0;
+ return a->args[i].value.integer != 0 && census_enabled();
}
}
- return 0;
+ return census_enabled();
}
grpc_compression_algorithm grpc_channel_args_get_compression_algorithm(
diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c
index a92a6ecaf2..7176c01b05 100644
--- a/src/core/channel/client_channel.c
+++ b/src/core/channel/client_channel.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -135,7 +135,7 @@ static void on_lb_policy_state_changed_locked(
}
static void on_lb_policy_state_changed(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
lb_policy_connectivity_watcher *w = arg;
gpr_mu_lock(&w->chand->mu_config);
@@ -161,7 +161,7 @@ static void watch_lb_policy(grpc_exec_ctx *exec_ctx, channel_data *chand,
}
static void cc_on_config_changed(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
channel_data *chand = arg;
grpc_lb_policy *lb_policy = NULL;
grpc_lb_policy *old_lb_policy;
@@ -191,7 +191,8 @@ static void cc_on_config_changed(grpc_exec_ctx *exec_ctx, void *arg,
old_lb_policy = chand->lb_policy;
chand->lb_policy = lb_policy;
if (lb_policy != NULL || chand->resolver == NULL /* disconnected */) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &chand->waiting_for_config_closures,
+ NULL);
}
if (lb_policy != NULL && chand->exit_idle_when_lb_policy_arrives) {
GRPC_LB_POLICY_REF(lb_policy, "exit_idle");
@@ -249,7 +250,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
channel_data *chand = elem->channel_data;
grpc_resolver *destroy_resolver = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
GPR_ASSERT(op->set_accept_stream == NULL);
if (op->bind_pollset != NULL) {
@@ -268,7 +269,7 @@ static void cc_start_transport_op(grpc_exec_ctx *exec_ctx,
if (op->send_ping != NULL) {
if (chand->lb_policy == NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, op->send_ping, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, op->send_ping, false, NULL);
} else {
grpc_lb_policy_ping_one(exec_ctx, chand->lb_policy, op->send_ping);
op->bind_pollset = NULL;
@@ -310,15 +311,15 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
grpc_connected_subchannel **connected_subchannel,
grpc_closure *on_ready);
-static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void continue_picking(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
continue_picking_args *cpa = arg;
if (!success) {
- grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, false, NULL);
} else if (cpa->connected_subchannel == NULL) {
/* cancelled, do nothing */
} else if (cc_pick_subchannel(exec_ctx, cpa->elem, cpa->initial_metadata,
cpa->connected_subchannel, cpa->on_ready)) {
- grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, true, NULL);
}
gpr_free(cpa);
}
@@ -346,7 +347,7 @@ static int cc_pick_subchannel(grpc_exec_ctx *exec_ctx, void *elemp,
cpa = closure->cb_arg;
if (cpa->connected_subchannel == connected_subchannel) {
cpa->connected_subchannel = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cpa->on_ready, false, NULL);
}
}
gpr_mu_unlock(&chand->mu_config);
@@ -497,7 +498,7 @@ typedef struct {
} external_connectivity_watcher;
static void on_external_watch_complete(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
external_connectivity_watcher *w = arg;
grpc_closure *follow_up = w->on_complete;
grpc_pollset_set_del_pollset(exec_ctx, &w->chand->interested_parties,
diff --git a/src/core/channel/client_uchannel.c b/src/core/channel/client_uchannel.c
index 2c0b07d8bf..b1e7155773 100644
--- a/src/core/channel/client_uchannel.c
+++ b/src/core/channel/client_uchannel.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -79,7 +79,7 @@ typedef struct client_uchannel_channel_data {
typedef grpc_subchannel_call_holder call_data;
static void monitor_subchannel(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
channel_data *chand = arg;
grpc_connectivity_state_set(exec_ctx, &chand->state_tracker,
chand->subchannel_connectivity,
@@ -105,7 +105,7 @@ static void cuc_start_transport_op(grpc_exec_ctx *exec_ctx,
grpc_transport_op *op) {
channel_data *chand = elem->channel_data;
- grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
GPR_ASSERT(op->set_accept_stream == NULL);
GPR_ASSERT(op->bind_pollset == NULL);
diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c
index 3de1ef8a35..3e7ca08fd2 100644
--- a/src/core/channel/compress_filter.c
+++ b/src/core/channel/compress_filter.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -153,7 +153,7 @@ static void process_send_initial_metadata(
static void continue_send_message(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem);
-static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, int success) {
+static void send_done(grpc_exec_ctx *exec_ctx, void *elemp, bool success) {
grpc_call_element *elem = elemp;
call_data *calld = elem->call_data;
gpr_slice_buffer_reset_and_unref(&calld->slices);
@@ -183,7 +183,7 @@ static void finish_send_message(grpc_exec_ctx *exec_ctx,
grpc_call_next_op(exec_ctx, elem, &calld->send_op);
}
-static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, int success) {
+static void got_slice(grpc_exec_ctx *exec_ctx, void *elemp, bool success) {
grpc_call_element *elem = elemp;
call_data *calld = elem->call_data;
gpr_slice_buffer_add(&calld->slices, calld->incoming_slice);
diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c
index 65cfb778bb..43eee046b8 100644
--- a/src/core/channel/http_client_filter.c
+++ b/src/core/channel/http_client_filter.c
@@ -1,5 +1,5 @@
/*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -80,7 +80,7 @@ static grpc_mdelem *client_recv_filter(void *user_data, grpc_mdelem *md) {
return md;
}
-static void hc_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, int success) {
+static void hc_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, bool success) {
grpc_call_element *elem = user_data;
call_data *calld = elem->call_data;
client_recv_filter_args a;
diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c
index a10b105494..bb75323933 100644
--- a/src/core/channel/http_server_filter.c
+++ b/src/core/channel/http_server_filter.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -131,7 +131,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
}
}
-static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, int success) {
+static void hs_on_recv(grpc_exec_ctx *exec_ctx, void *user_data, bool success) {
grpc_call_element *elem = user_data;
call_data *calld = elem->call_data;
if (success) {
diff --git a/src/core/channel/subchannel_call_holder.c b/src/core/channel/subchannel_call_holder.c
index 1e8f4ba1b6..3ad9fd9efb 100644
--- a/src/core/channel/subchannel_call_holder.c
+++ b/src/core/channel/subchannel_call_holder.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -43,9 +43,9 @@
#define CANCELLED_CALL ((grpc_subchannel_call *)1)
static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *holder,
- int success);
+ bool success);
static void retry_ops(grpc_exec_ctx *exec_ctx, void *retry_ops_args,
- int success);
+ bool success);
static void add_waiting_locked(grpc_subchannel_call_holder *holder,
grpc_transport_stream_op *op);
@@ -166,7 +166,7 @@ retry:
GPR_TIMER_END("grpc_subchannel_call_holder_perform_op", 0);
}
-static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void subchannel_ready(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
grpc_subchannel_call_holder *holder = arg;
grpc_subchannel_call *call;
gpr_mu_lock(&holder->mu);
@@ -209,10 +209,11 @@ static void retry_waiting_locked(grpc_exec_ctx *exec_ctx,
holder->waiting_ops_count = 0;
holder->waiting_ops_capacity = 0;
GRPC_SUBCHANNEL_CALL_REF(a->call, "retry_ops");
- grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(retry_ops, a), 1);
+ grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(retry_ops, a), true,
+ NULL);
}
-static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, int success) {
+static void retry_ops(grpc_exec_ctx *exec_ctx, void *args, bool success) {
retry_ops_args *a = args;
size_t i;
for (i = 0; i < a->nops; i++) {
@@ -240,9 +241,10 @@ static void fail_locked(grpc_exec_ctx *exec_ctx,
grpc_subchannel_call_holder *holder) {
size_t i;
for (i = 0; i < holder->waiting_ops_count; i++) {
- grpc_exec_ctx_enqueue(exec_ctx, holder->waiting_ops[i].on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, holder->waiting_ops[i].on_complete, false,
+ NULL);
grpc_exec_ctx_enqueue(exec_ctx, holder->waiting_ops[i].recv_message_ready,
- 0);
+ false, NULL);
}
holder->waiting_ops_count = 0;
}
diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c
index 5b10600ab5..459bbebb68 100644
--- a/src/core/client_config/lb_policies/pick_first.c
+++ b/src/core/client_config/lb_policies/pick_first.c
@@ -121,7 +121,7 @@ void pf_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
*pp->target = NULL;
grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
pp->pollset);
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, true, NULL);
gpr_free(pp);
pp = next;
}
@@ -140,7 +140,7 @@ static void pf_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
pp->pollset);
*target = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, false, NULL);
gpr_free(pp);
} else {
pp->next = p->pending_picks;
@@ -209,7 +209,7 @@ int pf_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
}
static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
pick_first_lb_policy *p = arg;
size_t i;
size_t num_subchannels = p->num_subchannels;
@@ -230,7 +230,7 @@ static void destroy_subchannels(grpc_exec_ctx *exec_ctx, void *arg,
}
static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
pick_first_lb_policy *p = arg;
grpc_subchannel *selected_subchannel;
pending_pick *pp;
@@ -272,15 +272,15 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
/* drop the pick list: we are connected now */
GRPC_LB_POLICY_WEAK_REF(&p->base, "destroy_subchannels");
gpr_atm_rel_store(&p->selected, (gpr_atm)selected);
- grpc_exec_ctx_enqueue(exec_ctx,
- grpc_closure_create(destroy_subchannels, p), 1);
+ grpc_exec_ctx_enqueue(
+ exec_ctx, grpc_closure_create(destroy_subchannels, p), true, NULL);
/* update any calls that were waiting for a pick */
while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
*pp->target = selected;
grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
pp->pollset);
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, true, NULL);
gpr_free(pp);
}
grpc_connected_subchannel_notify_on_state_change(
@@ -327,7 +327,7 @@ static void pf_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
*pp->target = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, true, NULL);
gpr_free(pp);
}
GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &p->base,
@@ -374,7 +374,7 @@ void pf_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
if (selected) {
grpc_connected_subchannel_ping(exec_ctx, selected, closure);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
}
}
diff --git a/src/core/client_config/lb_policies/round_robin.c b/src/core/client_config/lb_policies/round_robin.c
index d487456363..b1171c45b0 100644
--- a/src/core/client_config/lb_policies/round_robin.c
+++ b/src/core/client_config/lb_policies/round_robin.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -237,7 +237,7 @@ void rr_shutdown(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
*pp->target = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, false, NULL);
gpr_free(pp);
}
grpc_connectivity_state_set(exec_ctx, &p->state_tracker,
@@ -263,7 +263,7 @@ static void rr_cancel_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
pp->pollset);
*target = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, false, NULL);
gpr_free(pp);
} else {
pp->next = p->pending_picks;
@@ -336,7 +336,7 @@ int rr_pick(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol, grpc_pollset *pollset,
}
static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
subchannel_data *sd = arg;
round_robin_lb_policy *p = sd->policy;
pending_pick *pp;
@@ -376,7 +376,7 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
}
grpc_pollset_set_del_pollset(exec_ctx, &p->base.interested_parties,
pp->pollset);
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, true, NULL);
gpr_free(pp);
}
grpc_subchannel_notify_on_state_change(
@@ -428,7 +428,7 @@ static void rr_connectivity_changed(grpc_exec_ctx *exec_ctx, void *arg,
while ((pp = p->pending_picks)) {
p->pending_picks = pp->next;
*pp->target = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pp->on_complete, true, NULL);
gpr_free(pp);
}
} else {
@@ -479,7 +479,7 @@ static void rr_ping_one(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
grpc_connected_subchannel_ping(exec_ctx, target, closure);
} else {
gpr_mu_unlock(&p->mu);
- grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
}
}
diff --git a/src/core/client_config/resolvers/dns_resolver.c b/src/core/client_config/resolvers/dns_resolver.c
index 28ca30b946..376b6b3d76 100644
--- a/src/core/client_config/resolvers/dns_resolver.c
+++ b/src/core/client_config/resolvers/dns_resolver.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -93,7 +93,7 @@ static void dns_shutdown(grpc_exec_ctx *exec_ctx, grpc_resolver *resolver) {
gpr_mu_lock(&r->mu);
if (r->next_completion != NULL) {
*r->target_config = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, true, NULL);
r->next_completion = NULL;
}
gpr_mu_unlock(&r->mu);
@@ -182,7 +182,7 @@ static void dns_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
if (r->resolved_config) {
grpc_client_config_ref(r->resolved_config);
}
- grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, true, NULL);
r->next_completion = NULL;
r->published_version = r->resolved_version;
}
diff --git a/src/core/client_config/resolvers/sockaddr_resolver.c b/src/core/client_config/resolvers/sockaddr_resolver.c
index 6343259246..68910ad975 100644
--- a/src/core/client_config/resolvers/sockaddr_resolver.c
+++ b/src/core/client_config/resolvers/sockaddr_resolver.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -98,7 +98,7 @@ static void sockaddr_shutdown(grpc_exec_ctx *exec_ctx,
gpr_mu_lock(&r->mu);
if (r->next_completion != NULL) {
*r->target_config = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, true, NULL);
r->next_completion = NULL;
}
gpr_mu_unlock(&r->mu);
@@ -153,7 +153,7 @@ static void sockaddr_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
GRPC_LB_POLICY_UNREF(exec_ctx, lb_policy, "sockaddr");
r->published = 1;
*r->target_config = cfg;
- grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, true, NULL);
r->next_completion = NULL;
}
}
diff --git a/src/core/client_config/resolvers/zookeeper_resolver.c b/src/core/client_config/resolvers/zookeeper_resolver.c
index 4924ca77d6..166738e768 100644
--- a/src/core/client_config/resolvers/zookeeper_resolver.c
+++ b/src/core/client_config/resolvers/zookeeper_resolver.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -420,7 +420,7 @@ static void zookeeper_maybe_finish_next_locked(grpc_exec_ctx *exec_ctx,
if (r->resolved_config != NULL) {
grpc_client_config_ref(r->resolved_config);
}
- grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, r->next_completion, true, NULL);
r->next_completion = NULL;
r->published_version = r->resolved_version;
}
diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c
index fccc1dda54..0a996a1e8b 100644
--- a/src/core/client_config/subchannel.c
+++ b/src/core/client_config/subchannel.c
@@ -145,7 +145,7 @@ struct grpc_subchannel_call {
static gpr_timespec compute_connect_deadline(grpc_subchannel *c);
static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel,
- int iomgr_success);
+ bool iomgr_success);
#ifdef GRPC_STREAM_REFCOUNT_DEBUG
#define REF_REASON reason
@@ -175,7 +175,7 @@ static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *subchannel,
*/
static void connection_destroy(grpc_exec_ctx *exec_ctx, void *arg,
- int success) {
+ bool success) {
grpc_connected_subchannel *c = arg;
grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CONNECTION(c));
gpr_free(c);
@@ -198,7 +198,7 @@ void grpc_connected_subchannel_unref(grpc_exec_ctx *exec_ctx,
*/
static void subchannel_destroy(grpc_exec_ctx *exec_ctx, void *arg,
- int success) {
+ bool success) {
grpc_subchannel *c = arg;
gpr_free((void *)c->filters);
grpc_channel_args_destroy(c->args);
@@ -268,7 +268,7 @@ void grpc_subchannel_weak_unref(grpc_exec_ctx *exec_ctx,
old_refs = ref_mutate(c, -(gpr_atm)1, 1 REF_MUTATE_PURPOSE("WEAK_UNREF"));
if (old_refs == 1) {
grpc_exec_ctx_enqueue(exec_ctx, grpc_closure_create(subchannel_destroy, c),
- 1);
+ true, NULL);
}
}
@@ -341,7 +341,7 @@ grpc_connectivity_state grpc_subchannel_check_connectivity(grpc_subchannel *c) {
}
static void on_external_state_watcher_done(grpc_exec_ctx *exec_ctx, void *arg,
- int success) {
+ bool success) {
external_state_watcher *w = arg;
grpc_closure *follow_up = w->notify;
if (w->pollset_set != NULL) {
@@ -413,7 +413,7 @@ void grpc_connected_subchannel_process_transport_op(
}
static void subchannel_on_child_state_changed(grpc_exec_ctx *exec_ctx, void *p,
- int iomgr_success) {
+ bool iomgr_success) {
state_watcher *sw = p;
grpc_subchannel *c = sw->subchannel;
gpr_mu *mu = &c->mu;
@@ -594,7 +594,7 @@ static void update_reconnect_parameters(grpc_subchannel *c) {
gpr_time_add(c->next_attempt, gpr_time_from_millis(jitter, GPR_TIMESPAN));
}
-static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, int iomgr_success) {
+static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, bool iomgr_success) {
grpc_subchannel *c = arg;
gpr_mu_lock(&c->mu);
c->have_alarm = 0;
@@ -611,7 +611,7 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *arg, int iomgr_success) {
}
static void subchannel_connected(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
grpc_subchannel *c = arg;
if (c->connecting_result.transport != NULL) {
@@ -647,7 +647,7 @@ static gpr_timespec compute_connect_deadline(grpc_subchannel *c) {
*/
static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call,
- int success) {
+ bool success) {
grpc_subchannel_call *c = call;
GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0);
grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c));
diff --git a/src/core/httpcli/httpcli.c b/src/core/httpcli/httpcli.c
index b5cd8d8d2a..71237bb614 100644
--- a/src/core/httpcli/httpcli.c
+++ b/src/core/httpcli/httpcli.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -114,13 +114,13 @@ static void finish(grpc_exec_ctx *exec_ctx, internal_request *req,
gpr_free(req);
}
-static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, int success);
+static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, bool success);
static void do_read(grpc_exec_ctx *exec_ctx, internal_request *req) {
grpc_endpoint_read(exec_ctx, req->ep, &req->incoming, &req->on_read);
}
-static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, int success) {
+static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, bool success) {
internal_request *req = user_data;
size_t i;
@@ -147,7 +147,7 @@ static void on_written(grpc_exec_ctx *exec_ctx, internal_request *req) {
do_read(exec_ctx, req);
}
-static void done_write(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void done_write(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
internal_request *req = arg;
if (success) {
on_written(exec_ctx, req);
@@ -175,7 +175,7 @@ static void on_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
start_write(exec_ctx, req);
}
-static void on_connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void on_connected(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
internal_request *req = arg;
if (!req->ep) {
diff --git a/src/core/iomgr/closure.c b/src/core/iomgr/closure.c
index 7646a88ac5..3a96f7385f 100644
--- a/src/core/iomgr/closure.c
+++ b/src/core/iomgr/closure.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -43,7 +43,7 @@ void grpc_closure_init(grpc_closure *closure, grpc_iomgr_cb_func cb,
}
void grpc_closure_list_add(grpc_closure_list *closure_list,
- grpc_closure *closure, int success) {
+ grpc_closure *closure, bool success) {
if (closure == NULL) return;
closure->final_data = (success != 0);
if (closure_list->head == NULL) {
@@ -54,7 +54,7 @@ void grpc_closure_list_add(grpc_closure_list *closure_list,
closure_list->tail = closure;
}
-int grpc_closure_list_empty(grpc_closure_list closure_list) {
+bool grpc_closure_list_empty(grpc_closure_list closure_list) {
return closure_list.head == NULL;
}
@@ -77,7 +77,7 @@ typedef struct {
grpc_closure wrapper;
} wrapped_closure;
-static void closure_wrapper(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void closure_wrapper(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
wrapped_closure *wc = arg;
grpc_iomgr_cb_func cb = wc->cb;
void *cb_arg = wc->cb_arg;
diff --git a/src/core/iomgr/closure.h b/src/core/iomgr/closure.h
index 98ef91e1db..ea96c19c71 100644
--- a/src/core/iomgr/closure.h
+++ b/src/core/iomgr/closure.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -35,6 +35,7 @@
#define GRPC_INTERNAL_CORE_IOMGR_CLOSURE_H
#include <grpc/support/port_platform.h>
+#include <stdbool.h>
struct grpc_closure;
typedef struct grpc_closure grpc_closure;
@@ -54,7 +55,7 @@ typedef struct grpc_closure_list {
* \param success An indication on the state of the iomgr. On false, cleanup
* actions should be taken (eg, shutdown). */
typedef void (*grpc_iomgr_cb_func)(grpc_exec_ctx *exec_ctx, void *arg,
- int success);
+ bool success);
/** A closure over a grpc_iomgr_cb_func. */
struct grpc_closure {
@@ -83,13 +84,13 @@ grpc_closure *grpc_closure_create(grpc_iomgr_cb_func cb, void *cb_arg);
/** add \a closure to the end of \a list and set \a closure's success to \a
* success */
void grpc_closure_list_add(grpc_closure_list *list, grpc_closure *closure,
- int success);
+ bool success);
/** append all closures from \a src to \a dst and empty \a src. */
void grpc_closure_list_move(grpc_closure_list *src, grpc_closure_list *dst);
/** return whether \a list is empty. */
-int grpc_closure_list_empty(grpc_closure_list list);
+bool grpc_closure_list_empty(grpc_closure_list list);
/** return the next pointer for a queued closure list */
grpc_closure *grpc_closure_next(grpc_closure *closure);
diff --git a/src/core/iomgr/exec_ctx.c b/src/core/iomgr/exec_ctx.c
index 6059a6031c..1fd79f6eba 100644
--- a/src/core/iomgr/exec_ctx.c
+++ b/src/core/iomgr/exec_ctx.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -37,16 +37,16 @@
#include "src/core/profiling/timers.h"
-int grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
- int did_something = 0;
+bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
+ bool did_something = 0;
GPR_TIMER_BEGIN("grpc_exec_ctx_flush", 0);
while (!grpc_closure_list_empty(exec_ctx->closure_list)) {
grpc_closure *c = exec_ctx->closure_list.head;
exec_ctx->closure_list.head = exec_ctx->closure_list.tail = NULL;
while (c != NULL) {
- int success = (int)(c->final_data & 1);
+ bool success = (bool)(c->final_data & 1);
grpc_closure *next = (grpc_closure *)(c->final_data & ~(uintptr_t)1);
- did_something++;
+ did_something = true;
GPR_TIMER_BEGIN("grpc_exec_ctx_flush.cb", 0);
c->cb(exec_ctx, c->cb_arg, success);
GPR_TIMER_END("grpc_exec_ctx_flush.cb", 0);
@@ -62,11 +62,15 @@ void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx) {
}
void grpc_exec_ctx_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
- int success) {
+ bool success,
+ grpc_workqueue *offload_target_or_null) {
+ GPR_ASSERT(offload_target_or_null == NULL);
grpc_closure_list_add(&exec_ctx->closure_list, closure, success);
}
void grpc_exec_ctx_enqueue_list(grpc_exec_ctx *exec_ctx,
- grpc_closure_list *list) {
+ grpc_closure_list *list,
+ grpc_workqueue *offload_target_or_null) {
+ GPR_ASSERT(offload_target_or_null == NULL);
grpc_closure_list_move(list, &exec_ctx->closure_list);
}
diff --git a/src/core/iomgr/exec_ctx.h b/src/core/iomgr/exec_ctx.h
index 43df488094..9a9b2e55fa 100644
--- a/src/core/iomgr/exec_ctx.h
+++ b/src/core/iomgr/exec_ctx.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -57,22 +57,29 @@ struct grpc_exec_ctx {
grpc_closure_list closure_list;
};
+/** A workqueue represents a list of work to be executed asynchronously.
+ Forward declared here to avoid a circular dependency with workqueue.h. */
+struct grpc_workqueue;
+typedef struct grpc_workqueue grpc_workqueue;
+
#define GRPC_EXEC_CTX_INIT \
{ GRPC_CLOSURE_LIST_INIT }
/** Flush any work that has been enqueued onto this grpc_exec_ctx.
* Caller must guarantee that no interfering locks are held.
- * Returns 1 if work was performed, 0 otherwise. */
-int grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx);
+ * Returns true if work was performed, false otherwise. */
+bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx);
/** Finish any pending work for a grpc_exec_ctx. Must be called before
* the instance is destroyed, or work may be lost. */
void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx);
/** Add a closure to be executed at the next flush/finish point */
void grpc_exec_ctx_enqueue(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
- int success);
+ bool success,
+ grpc_workqueue *offload_target_or_null);
/** Add a list of closures to be executed at the next flush/finish point.
* Leaves \a list empty. */
void grpc_exec_ctx_enqueue_list(grpc_exec_ctx *exec_ctx,
- grpc_closure_list *list);
+ grpc_closure_list *list,
+ grpc_workqueue *offload_target_or_null);
#endif
diff --git a/src/core/iomgr/executor.c b/src/core/iomgr/executor.c
index 00c68f7828..f22d8f30ac 100644
--- a/src/core/iomgr/executor.c
+++ b/src/core/iomgr/executor.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -77,7 +77,7 @@ static void closure_exec_thread_func(void *ignored) {
gpr_mu_unlock(&g_executor.mu);
break;
} else {
- grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures);
+ grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures, NULL);
}
gpr_mu_unlock(&g_executor.mu);
grpc_exec_ctx_flush(&exec_ctx);
@@ -112,7 +112,7 @@ static void maybe_spawn_locked() {
g_executor.pending_join = 1;
}
-void grpc_executor_enqueue(grpc_closure *closure, int success) {
+void grpc_executor_enqueue(grpc_closure *closure, bool success) {
gpr_mu_lock(&g_executor.mu);
if (g_executor.shutting_down == 0) {
grpc_closure_list_add(&g_executor.closures, closure, success);
@@ -133,7 +133,7 @@ void grpc_executor_shutdown() {
* list below because we aren't accepting new work */
/* Execute pending callbacks, some may be performing cleanups */
- grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures);
+ grpc_exec_ctx_enqueue_list(&exec_ctx, &g_executor.closures, NULL);
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(grpc_closure_list_empty(g_executor.closures));
if (pending_join) {
diff --git a/src/core/iomgr/executor.h b/src/core/iomgr/executor.h
index 6da446ae9c..aac057ddf5 100644
--- a/src/core/iomgr/executor.h
+++ b/src/core/iomgr/executor.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -45,7 +45,7 @@ void grpc_executor_init();
/** Enqueue \a closure for its eventual execution of \a f(arg) on a separate
* thread */
-void grpc_executor_enqueue(grpc_closure *closure, int success);
+void grpc_executor_enqueue(grpc_closure *closure, bool success);
/** Shutdown the executor, running all pending work as part of the call */
void grpc_executor_shutdown();
diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c
index 89c938bc04..85eadd754b 100644
--- a/src/core/iomgr/fd_posix.c
+++ b/src/core/iomgr/fd_posix.c
@@ -218,7 +218,7 @@ static void close_fd_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
} else {
grpc_remove_fd_from_all_epoll_sets(fd->fd);
}
- grpc_exec_ctx_enqueue(exec_ctx, fd->on_done_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, fd->on_done_closure, true, NULL);
}
int grpc_fd_wrapped_fd(grpc_fd *fd) {
@@ -273,7 +273,7 @@ static void notify_on_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
} else if (*st == CLOSURE_READY) {
/* already ready ==> queue the closure to run immediately */
*st = CLOSURE_NOT_READY;
- grpc_exec_ctx_enqueue(exec_ctx, closure, !fd->shutdown);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, !fd->shutdown, NULL);
maybe_wake_one_watcher_locked(fd);
} else {
/* upcallptr was set to a different closure. This is an error! */
@@ -296,7 +296,7 @@ static int set_ready_locked(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
return 0;
} else {
/* waiting ==> queue closure */
- grpc_exec_ctx_enqueue(exec_ctx, *st, !fd->shutdown);
+ grpc_exec_ctx_enqueue(exec_ctx, *st, !fd->shutdown, NULL);
*st = CLOSURE_NOT_READY;
return 1;
}
diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c
index 65e371842f..96b6f81024 100644
--- a/src/core/iomgr/iocp_windows.c
+++ b/src/core/iomgr/iocp_windows.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -120,7 +120,7 @@ void grpc_iocp_work(grpc_exec_ctx *exec_ctx, gpr_timespec deadline) {
info->has_pending_iocp = 1;
}
gpr_mu_unlock(&socket->state_mu);
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
}
void grpc_iocp_init(void) {
@@ -183,7 +183,7 @@ static void socket_notify_on_iocp(grpc_exec_ctx *exec_ctx,
gpr_mu_lock(&socket->state_mu);
if (info->has_pending_iocp) {
info->has_pending_iocp = 0;
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
} else {
info->closure = closure;
}
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index d117485327..4acae2bb71 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -141,7 +141,7 @@ static void finally_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
}
static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_status) {
+ bool iomgr_status) {
delayed_add *da = arg;
if (!grpc_fd_is_orphaned(da->fd)) {
@@ -154,7 +154,7 @@ static void perform_delayed_add(grpc_exec_ctx *exec_ctx, void *arg,
/* We don't care about this pollset anymore. */
if (da->pollset->in_flight_cbs == 0 && !da->pollset->called_shutdown) {
da->pollset->called_shutdown = 1;
- grpc_exec_ctx_enqueue(exec_ctx, da->pollset->shutdown_done, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, da->pollset->shutdown_done, true, NULL);
}
}
gpr_mu_unlock(&da->pollset->mu);
@@ -178,7 +178,7 @@ static void multipoll_with_epoll_pollset_add_fd(grpc_exec_ctx *exec_ctx,
GRPC_FD_REF(fd, "delayed_add");
grpc_closure_init(&da->closure, perform_delayed_add, da);
pollset->in_flight_cbs++;
- grpc_exec_ctx_enqueue(exec_ctx, &da->closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &da->closure, true, NULL);
}
}
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index c325b634ae..a8e2e22977 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -243,7 +243,7 @@ void grpc_pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
GPR_ASSERT(grpc_closure_list_empty(pollset->idle_jobs));
pollset->vtable->finish_shutdown(pollset);
- grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pollset->shutdown_done, true, NULL);
}
void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
@@ -271,7 +271,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
if (!grpc_pollset_has_workers(pollset) &&
!grpc_closure_list_empty(pollset->idle_jobs)) {
GPR_TIMER_MARK("grpc_pollset_work.idle_jobs", 0);
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
goto done;
}
/* Check alarms - these are a global resource so we just ping
@@ -365,7 +365,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
* TODO(dklempner): Can we refactor the shutdown logic to avoid this? */
gpr_mu_lock(&pollset->mu);
} else if (!grpc_closure_list_empty(pollset->idle_jobs)) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
gpr_mu_unlock(&pollset->mu);
grpc_exec_ctx_flush(exec_ctx);
gpr_mu_lock(&pollset->mu);
@@ -381,7 +381,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
pollset->shutdown_done = closure;
grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
if (!grpc_pollset_has_workers(pollset)) {
- grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs);
+ grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs, NULL);
}
if (!pollset->called_shutdown && pollset->in_flight_cbs == 0 &&
!grpc_pollset_has_workers(pollset)) {
@@ -419,7 +419,8 @@ typedef struct grpc_unary_promote_args {
grpc_closure promotion_closure;
} grpc_unary_promote_args;
-static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args, int success) {
+static void basic_do_promote(grpc_exec_ctx *exec_ctx, void *args,
+ bool success) {
grpc_unary_promote_args *up_args = args;
const grpc_pollset_vtable *original_vtable = up_args->original_vtable;
grpc_pollset *pollset = up_args->pollset;
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index bfd9e69a16..2e650ca939 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -107,7 +107,7 @@ void grpc_pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
pollset->shutting_down = 1;
grpc_pollset_kick(pollset, GRPC_POLLSET_KICK_BROADCAST);
if (!pollset->is_iocp_worker) {
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
} else {
pollset->on_shutdown = closure;
}
@@ -165,7 +165,7 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
}
if (pollset->shutting_down && pollset->on_shutdown != NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, pollset->on_shutdown, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, pollset->on_shutdown, true, NULL);
pollset->on_shutdown = NULL;
}
goto done;
diff --git a/src/core/iomgr/resolve_address_posix.c b/src/core/iomgr/resolve_address_posix.c
index 555c74ce7e..c51745b918 100644
--- a/src/core/iomgr/resolve_address_posix.c
+++ b/src/core/iomgr/resolve_address_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -152,7 +152,7 @@ done:
/* Callback to be passed to grpc_executor to asynch-ify
* grpc_blocking_resolve_address */
-static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, int success) {
+static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, bool success) {
request *r = rp;
grpc_resolved_addresses *resolved =
grpc_blocking_resolve_address(r->name, r->default_port);
diff --git a/src/core/iomgr/resolve_address_windows.c b/src/core/iomgr/resolve_address_windows.c
index 007c855d10..28c8661e73 100644
--- a/src/core/iomgr/resolve_address_windows.c
+++ b/src/core/iomgr/resolve_address_windows.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -135,7 +135,7 @@ done:
/* Callback to be passed to grpc_executor to asynch-ify
* grpc_blocking_resolve_address */
-static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, int success) {
+static void do_request_thread(grpc_exec_ctx *exec_ctx, void *rp, bool success) {
request *r = rp;
grpc_resolved_addresses *resolved =
grpc_blocking_resolve_address(r->name, r->default_port);
diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c
index d9d24ee9a3..c76c2e3b0f 100644
--- a/src/core/iomgr/tcp_client_posix.c
+++ b/src/core/iomgr/tcp_client_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -91,7 +91,7 @@ error:
return 0;
}
-static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int success) {
+static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, bool success) {
int done;
async_connect *ac = acp;
if (grpc_tcp_trace) {
@@ -111,7 +111,7 @@ static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int success) {
}
}
-static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, int success) {
+static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, bool success) {
async_connect *ac = acp;
int so_error = 0;
socklen_t so_error_size;
@@ -206,7 +206,7 @@ finish:
gpr_free(ac->addr_str);
gpr_free(ac);
}
- grpc_exec_ctx_enqueue(exec_ctx, closure, *ep != NULL);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, *ep != NULL, NULL);
}
void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
@@ -243,7 +243,7 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
addr_len = sizeof(addr4_copy);
}
if (!prepare_socket(addr, fd)) {
- grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
return;
}
@@ -259,14 +259,14 @@ void grpc_tcp_client_connect(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
if (err >= 0) {
*ep = grpc_tcp_create(fdobj, GRPC_TCP_DEFAULT_READ_SLICE_SIZE, addr_str);
- grpc_exec_ctx_enqueue(exec_ctx, closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, true, NULL);
goto done;
}
if (errno != EWOULDBLOCK && errno != EINPROGRESS) {
gpr_log(GPR_ERROR, "connect error to '%s': %s", addr_str, strerror(errno));
grpc_fd_orphan(exec_ctx, fdobj, NULL, NULL, "tcp_client_connect_error");
- grpc_exec_ctx_enqueue(exec_ctx, closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, false, NULL);
goto done;
}
diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c
index e5691b7e12..689c6f7b10 100644
--- a/src/core/iomgr/tcp_client_windows.c
+++ b/src/core/iomgr/tcp_client_windows.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -74,7 +74,7 @@ static void async_connect_unlock_and_cleanup(async_connect *ac) {
}
}
-static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int occured) {
+static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, bool occured) {
async_connect *ac = acp;
gpr_mu_lock(&ac->mu);
/* If the alarm didn't occur, it got cancelled. */
@@ -84,7 +84,7 @@ static void on_alarm(grpc_exec_ctx *exec_ctx, void *acp, int occured) {
async_connect_unlock_and_cleanup(ac);
}
-static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, int from_iocp) {
+static void on_connect(grpc_exec_ctx *exec_ctx, void *acp, bool from_iocp) {
async_connect *ac = acp;
SOCKET sock = ac->socket->socket;
grpc_endpoint **ep = ac->endpoint;
@@ -215,7 +215,7 @@ failure:
} else if (sock != INVALID_SOCKET) {
closesocket(sock);
}
- grpc_exec_ctx_enqueue(exec_ctx, on_done, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, on_done, false, NULL);
}
#endif /* GPR_WINSOCK_SOCKET */
diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c
index 4fa8ca8c71..048e907441 100644
--- a/src/core/iomgr/tcp_posix.c
+++ b/src/core/iomgr/tcp_posix.c
@@ -100,9 +100,9 @@ typedef struct {
} grpc_tcp;
static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success);
+ bool success);
static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success);
+ bool success);
static void tcp_shutdown(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *)ep;
@@ -247,7 +247,7 @@ static void tcp_continue_read(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp) {
}
static void tcp_handle_read(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success) {
+ bool success) {
grpc_tcp *tcp = (grpc_tcp *)arg;
GPR_ASSERT(!tcp->finished_edge);
@@ -273,7 +273,7 @@ static void tcp_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
tcp->finished_edge = 0;
grpc_fd_notify_on_read(exec_ctx, tcp->em_fd, &tcp->read_closure);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, &tcp->read_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &tcp->read_closure, true, NULL);
}
}
@@ -360,7 +360,7 @@ static flush_result tcp_flush(grpc_tcp *tcp) {
}
static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */,
- int success) {
+ bool success) {
grpc_tcp *tcp = (grpc_tcp *)arg;
flush_result status;
grpc_closure *cb;
@@ -407,7 +407,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
if (buf->length == 0) {
GPR_TIMER_END("tcp_write", 0);
- grpc_exec_ctx_enqueue(exec_ctx, cb, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, true, NULL);
return;
}
tcp->outgoing_buffer = buf;
@@ -420,7 +420,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
tcp->write_cb = cb;
grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, cb, status == FLUSH_DONE);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, status == FLUSH_DONE, NULL);
}
GPR_TIMER_END("tcp_write", 0);
diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c
index adf14aeb59..5e07f8261c 100644
--- a/src/core/iomgr/tcp_server_posix.c
+++ b/src/core/iomgr/tcp_server_posix.c
@@ -160,7 +160,7 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) {
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
if (s->shutdown_complete != NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, true, NULL);
}
gpr_mu_destroy(&s->mu);
@@ -174,7 +174,8 @@ static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
gpr_free(s);
}
-static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server, int success) {
+static void destroyed_port(grpc_exec_ctx *exec_ctx, void *server,
+ bool success) {
grpc_tcp_server *s = server;
gpr_mu_lock(&s->mu);
s->destroyed_ports++;
@@ -317,7 +318,7 @@ error:
}
/* event manager callback when reads are ready */
-static void on_read(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void on_read(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
grpc_tcp_listener *sp = arg;
grpc_tcp_server_acceptor acceptor = {sp->server, sp->port_index,
sp->fd_index};
@@ -602,7 +603,7 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
/* Complete shutdown_starting work before destroying. */
grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT;
gpr_mu_lock(&s->mu);
- grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting);
+ grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting, NULL);
gpr_mu_unlock(&s->mu);
if (exec_ctx == NULL) {
grpc_exec_ctx_flush(&local_exec_ctx);
diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c
index 00d381f264..ce930b8f41 100644
--- a/src/core/iomgr/tcp_server_windows.c
+++ b/src/core/iomgr/tcp_server_windows.c
@@ -119,7 +119,7 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) {
static void finish_shutdown(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
if (s->shutdown_complete != NULL) {
- grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, s->shutdown_complete, true, NULL);
}
/* Now that the accepts have been aborted, we can destroy the sockets.
@@ -173,7 +173,7 @@ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) {
/* Complete shutdown_starting work before destroying. */
grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT;
gpr_mu_lock(&s->mu);
- grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting);
+ grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting, NULL);
gpr_mu_unlock(&s->mu);
if (exec_ctx == NULL) {
grpc_exec_ctx_flush(&local_exec_ctx);
@@ -311,7 +311,7 @@ failure:
}
/* Event manager callback when reads are ready. */
-static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, int from_iocp) {
+static void on_accept(grpc_exec_ctx *exec_ctx, void *arg, bool from_iocp) {
grpc_tcp_listener *sp = arg;
grpc_tcp_server_acceptor acceptor = {sp->server, sp->port_index, 0};
SOCKET sock = sp->new_socket;
diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index f09588cc21..038e4158c8 100644
--- a/src/core/iomgr/tcp_windows.c
+++ b/src/core/iomgr/tcp_windows.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -138,7 +138,7 @@ static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); }
#endif
/* Asynchronous callback from the IOCP, or the background thread. */
-static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, int success) {
+static void on_read(grpc_exec_ctx *exec_ctx, void *tcpp, bool success) {
grpc_tcp *tcp = tcpp;
grpc_closure *cb = tcp->read_cb;
grpc_winsocket *socket = tcp->socket;
@@ -184,7 +184,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
WSABUF buffer;
if (tcp->shutting_down) {
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
@@ -208,7 +208,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
/* Did we get data immediately ? Yay. */
if (info->wsa_error != WSAEWOULDBLOCK) {
info->bytes_transfered = bytes_read;
- grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, true, NULL);
return;
}
@@ -221,7 +221,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
info->wsa_error = wsa_error;
- grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, &tcp->on_read, false, NULL);
return;
}
}
@@ -230,7 +230,7 @@ static void win_read(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
}
/* Asynchronous callback from the IOCP, or the background thread. */
-static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, int success) {
+static void on_write(grpc_exec_ctx *exec_ctx, void *tcpp, bool success) {
grpc_tcp *tcp = (grpc_tcp *)tcpp;
grpc_winsocket *handle = tcp->socket;
grpc_winsocket_callback_info *info = &handle->write_info;
@@ -273,7 +273,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
size_t len;
if (tcp->shutting_down) {
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
@@ -301,9 +301,9 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
connection that has its send queue filled up. But if we don't, then we can
avoid doing an async write operation at all. */
if (info->wsa_error != WSAEWOULDBLOCK) {
- int ok = 0;
+ bool ok = false;
if (status == 0) {
- ok = 1;
+ ok = true;
GPR_ASSERT(bytes_sent == tcp->write_slices->length);
} else {
if (socket->read_info.wsa_error != WSAECONNRESET) {
@@ -313,7 +313,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
}
}
if (allocated) gpr_free(allocated);
- grpc_exec_ctx_enqueue(exec_ctx, cb, ok);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, ok, NULL);
return;
}
@@ -330,7 +330,7 @@ static void win_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep,
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
TCP_UNREF(tcp, "write");
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
}
diff --git a/src/core/iomgr/timer.c b/src/core/iomgr/timer.c
index 24d6204e07..a33d8f63a0 100644
--- a/src/core/iomgr/timer.c
+++ b/src/core/iomgr/timer.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -224,7 +224,7 @@ void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) {
shard_type *shard = &g_shards[shard_idx(timer)];
gpr_mu_lock(&shard->mu);
if (!timer->triggered) {
- grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, false, NULL);
timer->triggered = 1;
if (timer->heap_index == INVALID_HEAP_INDEX) {
list_remove(timer);
@@ -290,7 +290,7 @@ static size_t pop_timers(grpc_exec_ctx *exec_ctx, shard_type *shard,
grpc_timer *timer;
gpr_mu_lock(&shard->mu);
while ((timer = pop_one(shard, now))) {
- grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, success);
+ grpc_exec_ctx_enqueue(exec_ctx, &timer->closure, success, NULL);
n++;
}
*new_min_deadline = compute_min_deadline(shard);
diff --git a/src/core/iomgr/udp_server.c b/src/core/iomgr/udp_server.c
index a1a6b04cad..fe006c603c 100644
--- a/src/core/iomgr/udp_server.c
+++ b/src/core/iomgr/udp_server.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -425,15 +425,5 @@ void grpc_udp_server_start(grpc_exec_ctx *exec_ctx, grpc_udp_server *s,
gpr_mu_unlock(&s->mu);
}
-/* TODO(rjshade): Add a test for this method. */
-void grpc_udp_server_write(server_port *sp, const char *buffer, size_t buf_len,
- const struct sockaddr *peer_address) {
- ssize_t rc;
- rc = sendto(sp->fd, buffer, buf_len, 0, peer_address, sizeof(peer_address));
- if (rc < 0) {
- gpr_log(GPR_ERROR, "Unable to send data: %s", strerror(errno));
- }
-}
-
#endif
#endif
diff --git a/src/core/iomgr/udp_server.h b/src/core/iomgr/udp_server.h
index de5736c426..73a21c80ab 100644
--- a/src/core/iomgr/udp_server.h
+++ b/src/core/iomgr/udp_server.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -72,12 +72,4 @@ int grpc_udp_server_add_port(grpc_udp_server *s, const void *addr,
void grpc_udp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_udp_server *server,
grpc_closure *on_done);
-/* Write the contents of buffer to the underlying UDP socket. */
-/*
-void grpc_udp_server_write(grpc_udp_server *s,
- const char *buffer,
- int buf_len,
- const struct sockaddr* to);
- */
-
#endif /* GRPC_INTERNAL_CORE_IOMGR_UDP_SERVER_H */
diff --git a/src/core/iomgr/workqueue.h b/src/core/iomgr/workqueue.h
index 714536233c..36dd133468 100644
--- a/src/core/iomgr/workqueue.h
+++ b/src/core/iomgr/workqueue.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -47,9 +47,7 @@
#include "src/core/iomgr/workqueue_windows.h"
#endif
-/** A workqueue represents a list of work to be executed asynchronously. */
-struct grpc_workqueue;
-typedef struct grpc_workqueue grpc_workqueue;
+/* grpc_workqueue is forward declared in exec_ctx.h */
/** Create a work queue */
grpc_workqueue *grpc_workqueue_create(grpc_exec_ctx *exec_ctx);
diff --git a/src/core/iomgr/workqueue_posix.c b/src/core/iomgr/workqueue_posix.c
index d2a1c34612..da11df67ef 100644
--- a/src/core/iomgr/workqueue_posix.c
+++ b/src/core/iomgr/workqueue_posix.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -45,7 +45,7 @@
#include "src/core/iomgr/fd_posix.h"
-static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, int success);
+static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, bool success);
grpc_workqueue *grpc_workqueue_create(grpc_exec_ctx *exec_ctx) {
char name[32];
@@ -110,7 +110,7 @@ void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {
gpr_mu_unlock(&workqueue->mu);
}
-static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void on_readable(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
grpc_workqueue *workqueue = arg;
if (!success) {
diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c
index 8b56c57645..afba0079f5 100644
--- a/src/core/security/credentials.c
+++ b/src/core/security/credentials.c
@@ -793,7 +793,7 @@ static void md_only_test_destruct(grpc_call_credentials *creds) {
}
static void on_simulated_token_fetch_done(grpc_exec_ctx *exec_ctx,
- void *user_data, int success) {
+ void *user_data, bool success) {
grpc_credentials_metadata_request *r =
(grpc_credentials_metadata_request *)user_data;
grpc_md_only_test_credentials *c = (grpc_md_only_test_credentials *)r->creds;
@@ -812,7 +812,7 @@ static void md_only_test_get_request_metadata(
grpc_credentials_metadata_request *cb_arg =
grpc_credentials_metadata_request_create(creds, cb, user_data);
grpc_executor_enqueue(
- grpc_closure_create(on_simulated_token_fetch_done, cb_arg), 1);
+ grpc_closure_create(on_simulated_token_fetch_done, cb_arg), true);
} else {
cb(exec_ctx, user_data, c->md_store->entries, 1, GRPC_CREDENTIALS_OK);
}
diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c
index 5385e41130..ae71107bef 100644
--- a/src/core/security/google_default_credentials.c
+++ b/src/core/security/google_default_credentials.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -86,7 +86,7 @@ static void on_compute_engine_detection_http_response(
gpr_mu_unlock(GRPC_POLLSET_MU(&detector->pollset));
}
-static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, int s) {
+static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, bool s) {
grpc_pollset_destroy(p);
}
diff --git a/src/core/security/handshake.c b/src/core/security/handshake.c
index 364b765396..a8b2fef629 100644
--- a/src/core/security/handshake.c
+++ b/src/core/security/handshake.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -61,10 +61,10 @@ typedef struct {
} grpc_security_handshake;
static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
- void *setup, int success);
+ void *setup, bool success);
static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx, void *setup,
- int success);
+ bool success);
static void security_connector_remove_handshake(grpc_security_handshake *h) {
grpc_security_connector_handshake_list *node;
@@ -198,7 +198,8 @@ static void send_handshake_bytes_to_peer(grpc_exec_ctx *exec_ctx,
}
static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
- void *handshake, int success) {
+ void *handshake,
+ bool success) {
grpc_security_handshake *h = handshake;
size_t consumed_slice_size = 0;
tsi_result result = TSI_OK;
@@ -265,7 +266,7 @@ static void on_handshake_data_received_from_peer(grpc_exec_ctx *exec_ctx,
/* If handshake is NULL, the handshake is done. */
static void on_handshake_data_sent_to_peer(grpc_exec_ctx *exec_ctx,
- void *handshake, int success) {
+ void *handshake, bool success) {
grpc_security_handshake *h = handshake;
/* Make sure that write is OK. */
diff --git a/src/core/security/secure_endpoint.c b/src/core/security/secure_endpoint.c
index 9b87e268c6..d11c43be20 100644
--- a/src/core/security/secure_endpoint.c
+++ b/src/core/security/secure_endpoint.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -126,7 +126,7 @@ static void flush_read_staging_buffer(secure_endpoint *ep, uint8_t **cur,
}
static void call_read_cb(grpc_exec_ctx *exec_ctx, secure_endpoint *ep,
- int success) {
+ bool success) {
if (grpc_trace_secure_endpoint) {
size_t i;
for (i = 0; i < ep->read_buffer->count; i++) {
@@ -137,11 +137,11 @@ static void call_read_cb(grpc_exec_ctx *exec_ctx, secure_endpoint *ep,
}
}
ep->read_buffer = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, ep->read_cb, success);
+ grpc_exec_ctx_enqueue(exec_ctx, ep->read_cb, success, NULL);
SECURE_ENDPOINT_UNREF(exec_ctx, ep, "read");
}
-static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, int success) {
+static void on_read(grpc_exec_ctx *exec_ctx, void *user_data, bool success) {
unsigned i;
uint8_t keep_looping = 0;
tsi_result result = TSI_OK;
@@ -315,7 +315,7 @@ static void endpoint_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *secure_ep,
if (result != TSI_OK) {
/* TODO(yangg) do different things according to the error type? */
gpr_slice_buffer_reset_and_unref(&ep->output_buffer);
- grpc_exec_ctx_enqueue(exec_ctx, cb, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, cb, false, NULL);
return;
}
diff --git a/src/core/security/server_auth_filter.c b/src/core/security/server_auth_filter.c
index d5c8c54369..4c78711387 100644
--- a/src/core/security/server_auth_filter.c
+++ b/src/core/security/server_auth_filter.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -153,7 +153,7 @@ static void on_md_processing_done(
}
static void auth_on_recv(grpc_exec_ctx *exec_ctx, void *user_data,
- int success) {
+ bool success) {
grpc_call_element *elem = user_data;
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c
index 08713fceaf..ee3443c828 100644
--- a/src/core/security/server_secure_chttp2.c
+++ b/src/core/security/server_secure_chttp2.c
@@ -142,7 +142,7 @@ static void start(grpc_exec_ctx *exec_ctx, grpc_server *server, void *statep,
on_accept, state);
}
-static void destroy_done(grpc_exec_ctx *exec_ctx, void *statep, int success) {
+static void destroy_done(grpc_exec_ctx *exec_ctx, void *statep, bool success) {
grpc_server_secure_state *state = statep;
if (state->destroy_callback != NULL) {
state->destroy_callback->cb(exec_ctx, state->destroy_callback->cb_arg,
diff --git a/src/core/surface/call.c b/src/core/surface/call.c
index 880666bb38..9495e748b5 100644
--- a/src/core/surface/call.c
+++ b/src/core/surface/call.c
@@ -229,9 +229,9 @@ static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
grpc_status_code status,
const char *description);
static void destroy_call(grpc_exec_ctx *exec_ctx, void *call_stack,
- int success);
+ bool success);
static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
- int success);
+ bool success);
grpc_call *grpc_call_create(grpc_channel *channel, grpc_call *parent_call,
uint32_t propagation_mask,
@@ -351,7 +351,7 @@ void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *c REF_ARG) {
GRPC_CALL_STACK_UNREF(exec_ctx, CALL_STACK_FROM_CALL(c), REF_REASON);
}
-static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, int success) {
+static void destroy_call(grpc_exec_ctx *exec_ctx, void *call, bool success) {
size_t i;
int ii;
grpc_call *c = call;
@@ -688,13 +688,13 @@ typedef struct cancel_closure {
grpc_status_code status;
} cancel_closure;
-static void done_cancel(grpc_exec_ctx *exec_ctx, void *ccp, int success) {
+static void done_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
cancel_closure *cc = ccp;
GRPC_CALL_INTERNAL_UNREF(exec_ctx, cc->call, "cancel");
gpr_free(cc);
}
-static void send_cancel(grpc_exec_ctx *exec_ctx, void *ccp, int success) {
+static void send_cancel(grpc_exec_ctx *exec_ctx, void *ccp, bool success) {
grpc_transport_stream_op op;
cancel_closure *cc = ccp;
memset(&op, 0, sizeof(op));
@@ -721,7 +721,7 @@ static grpc_call_error cancel_with_status(grpc_exec_ctx *exec_ctx, grpc_call *c,
cc->call = c;
cc->status = status;
GRPC_CALL_INTERNAL_REF(c, "cancel");
- grpc_exec_ctx_enqueue(exec_ctx, &cc->closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &cc->closure, true, NULL);
return GRPC_CALL_OK;
}
@@ -757,7 +757,7 @@ grpc_call *grpc_call_from_top_element(grpc_call_element *elem) {
return CALL_FROM_TOP_ELEM(elem);
}
-static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void call_alarm(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
grpc_call *call = arg;
gpr_mu_lock(&call->mu);
call->have_alarm = 0;
@@ -934,7 +934,7 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
batch_control *bctl) {
grpc_call *call = bctl->call;
if (bctl->is_notify_tag_closure) {
- grpc_exec_ctx_enqueue(exec_ctx, bctl->notify_tag, bctl->success);
+ grpc_exec_ctx_enqueue(exec_ctx, bctl->notify_tag, bctl->success, NULL);
gpr_mu_lock(&call->mu);
bctl->call->used_batches =
(uint8_t)(bctl->call->used_batches &
@@ -974,7 +974,7 @@ static void continue_receiving_slices(grpc_exec_ctx *exec_ctx,
}
static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
- int success) {
+ bool success) {
batch_control *bctl = bctlp;
grpc_call *call = bctl->call;
@@ -993,7 +993,7 @@ static void receiving_slice_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
}
}
-static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, int success) {
+static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, bool success) {
batch_control *bctl = bctlp;
grpc_call *call = bctl->call;
grpc_call *child_call;
@@ -1066,7 +1066,7 @@ static void finish_batch(grpc_exec_ctx *exec_ctx, void *bctlp, int success) {
}
static void receiving_stream_ready(grpc_exec_ctx *exec_ctx, void *bctlp,
- int success) {
+ bool success) {
batch_control *bctl = bctlp;
grpc_call *call = bctl->call;
diff --git a/src/core/surface/channel.c b/src/core/surface/channel.c
index 001692376f..12d8ebceb9 100644
--- a/src/core/surface/channel.c
+++ b/src/core/surface/channel.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -80,7 +80,7 @@ struct grpc_channel {
/* the protobuf library will (by default) start warning at 100megs */
#define DEFAULT_MAX_MESSAGE_LENGTH (100 * 1024 * 1024)
-static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, int success);
+static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg, bool success);
grpc_channel *grpc_channel_create_from_filters(
grpc_exec_ctx *exec_ctx, const char *target,
@@ -268,7 +268,7 @@ void grpc_channel_internal_unref(grpc_exec_ctx *exec_ctx,
}
static void destroy_channel(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_success) {
+ bool iomgr_success) {
grpc_channel *channel = arg;
grpc_channel_stack_destroy(exec_ctx, CHANNEL_STACK_FROM_CHANNEL(channel));
while (channel->registered_calls) {
diff --git a/src/core/surface/channel_connectivity.c b/src/core/surface/channel_connectivity.c
index 10f5c4da4d..2dd4fce26b 100644
--- a/src/core/surface/channel_connectivity.c
+++ b/src/core/surface/channel_connectivity.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -165,11 +165,11 @@ static void partly_done(grpc_exec_ctx *exec_ctx, state_watcher *w,
}
}
-static void watch_complete(grpc_exec_ctx *exec_ctx, void *pw, int success) {
+static void watch_complete(grpc_exec_ctx *exec_ctx, void *pw, bool success) {
partly_done(exec_ctx, pw, 1);
}
-static void timeout_complete(grpc_exec_ctx *exec_ctx, void *pw, int success) {
+static void timeout_complete(grpc_exec_ctx *exec_ctx, void *pw, bool success) {
partly_done(exec_ctx, pw, 0);
}
diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c
index 49083f0870..4d4337d288 100644
--- a/src/core/surface/channel_create.c
+++ b/src/core/surface/channel_create.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -80,11 +80,11 @@ static void connector_unref(grpc_exec_ctx *exec_ctx, grpc_connector *con) {
}
static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
- int success) {
+ bool success) {
connector_unref(exec_ctx, arg);
}
-static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void connected(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
connector *c = arg;
grpc_closure *notify;
grpc_endpoint *tcp = c->tcp;
diff --git a/src/core/surface/channel_ping.c b/src/core/surface/channel_ping.c
index b4ce282787..983f1c8a66 100644
--- a/src/core/surface/channel_ping.c
+++ b/src/core/surface/channel_ping.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -53,7 +53,7 @@ static void ping_destroy(grpc_exec_ctx *exec_ctx, void *arg,
gpr_free(arg);
}
-static void ping_done(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void ping_done(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
ping_result *pr = arg;
grpc_cq_end_op(exec_ctx, pr->cq, pr->tag, success, ping_destroy, pr,
&pr->completion_storage);
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index c0db9c508a..75298eb795 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -86,7 +86,7 @@ static gpr_mu g_freelist_mu;
grpc_completion_queue *g_freelist;
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
- int success);
+ bool success);
void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); }
@@ -169,7 +169,7 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) {
}
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg,
- int success) {
+ bool success) {
grpc_completion_queue *cc = arg;
GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
}
diff --git a/src/core/surface/lame_client.c b/src/core/surface/lame_client.c
index a60e9d20da..705996cad3 100644
--- a/src/core/surface/lame_client.c
+++ b/src/core/surface/lame_client.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -78,8 +78,8 @@ static void lame_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
} else if (op->recv_trailing_metadata != NULL) {
fill_metadata(elem, op->recv_trailing_metadata);
}
- grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, 0);
- grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, false, NULL);
+ grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, false, NULL);
}
static char *lame_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c
index 552a570713..dd1441ad67 100644
--- a/src/core/surface/secure_channel_create.c
+++ b/src/core/surface/secure_channel_create.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -128,14 +128,14 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg,
}
static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg,
- int success) {
+ bool success) {
connector *c = arg;
grpc_security_connector_do_handshake(exec_ctx, &c->security_connector->base,
c->connecting_endpoint,
on_secure_handshake_done, c);
}
-static void connected(grpc_exec_ctx *exec_ctx, void *arg, int success) {
+static void connected(grpc_exec_ctx *exec_ctx, void *arg, bool success) {
connector *c = arg;
grpc_closure *notify;
grpc_endpoint *tcp = c->newly_connecting_endpoint;
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 79db13810a..42cffccb4c 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -260,7 +260,7 @@ struct shutdown_cleanup_args {
};
static void shutdown_cleanup(grpc_exec_ctx *exec_ctx, void *arg,
- int iomgr_status_ignored) {
+ bool iomgr_status_ignored) {
struct shutdown_cleanup_args *a = arg;
gpr_slice_unref(a->slice);
gpr_free(a);
@@ -313,7 +313,7 @@ static void request_matcher_destroy(request_matcher *rm) {
gpr_stack_lockfree_destroy(rm->requests);
}
-static void kill_zombie(grpc_exec_ctx *exec_ctx, void *elem, int success) {
+static void kill_zombie(grpc_exec_ctx *exec_ctx, void *elem, bool success) {
grpc_call_destroy(grpc_call_from_top_element(elem));
}
@@ -328,7 +328,7 @@ static void request_matcher_zombify_all_pending_calls(grpc_exec_ctx *exec_ctx,
grpc_closure_init(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
- grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, true, NULL);
}
}
@@ -392,7 +392,7 @@ static void orphan_channel(channel_data *chand) {
}
static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
- int success) {
+ bool success) {
channel_data *chand = cd;
grpc_server *server = chand->server;
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "server");
@@ -407,7 +407,8 @@ static void destroy_channel(grpc_exec_ctx *exec_ctx, channel_data *chand) {
maybe_finish_shutdown(exec_ctx, chand->server);
chand->finish_destroy_channel_closure.cb = finish_destroy_channel;
chand->finish_destroy_channel_closure.cb_arg = chand;
- grpc_exec_ctx_enqueue(exec_ctx, &chand->finish_destroy_channel_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &chand->finish_destroy_channel_closure, true,
+ NULL);
}
static void finish_start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_server *server,
@@ -420,7 +421,7 @@ static void finish_start_new_rpc(grpc_exec_ctx *exec_ctx, grpc_server *server,
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
- grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, true, NULL);
return;
}
@@ -569,7 +570,7 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
}
static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
- int success) {
+ bool success) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
gpr_timespec op_deadline;
@@ -609,7 +610,7 @@ static void server_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
}
static void got_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
- int success) {
+ bool success) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
if (success) {
@@ -620,7 +621,7 @@ static void got_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
- grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, true, NULL);
} else if (calld->state == PENDING) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
@@ -653,7 +654,7 @@ static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd,
}
static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
- int iomgr_status_ignored) {
+ bool iomgr_status_ignored) {
channel_data *chand = cd;
grpc_server *server = chand->server;
if (chand->connectivity_state != GRPC_CHANNEL_FATAL_FAILURE) {
@@ -779,9 +780,7 @@ grpc_server *grpc_server_create_from_filters(
const grpc_channel_filter **filters, size_t filter_count,
const grpc_channel_args *args) {
size_t i;
- /* TODO(census): restore this once we finalize census filter etc.
- int census_enabled = grpc_channel_args_is_census_enabled(args); */
- int census_enabled = 0;
+ int census_enabled = grpc_channel_args_is_census_enabled(args);
grpc_server *server = gpr_malloc(sizeof(grpc_server));
@@ -987,7 +986,7 @@ void done_published_shutdown(grpc_exec_ctx *exec_ctx, void *done_arg,
}
static void listener_destroy_done(grpc_exec_ctx *exec_ctx, void *s,
- int success) {
+ bool success) {
grpc_server *server = s;
gpr_mu_lock(&server->mu_global);
server->listeners_destroyed++;
@@ -1142,7 +1141,8 @@ static grpc_call_error queue_call_request(grpc_exec_ctx *exec_ctx,
grpc_closure_init(
&calld->kill_zombie_closure, kill_zombie,
grpc_call_stack_element(grpc_call_get_call_stack(calld->call), 0));
- grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, true,
+ NULL);
} else {
GPR_ASSERT(calld->state == PENDING);
calld->state = ACTIVATED;
@@ -1231,7 +1231,7 @@ done:
}
static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx,
- void *user_data, int success);
+ void *user_data, bool success);
static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
gpr_slice slice = value->slice;
@@ -1317,7 +1317,7 @@ static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
}
static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx, void *prc,
- int success) {
+ bool success) {
requested_call *rc = prc;
grpc_call *call = *rc->call;
grpc_call_element *elem =
diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c
index 6e21d2dcd7..ce970dfe73 100644
--- a/src/core/surface/server_chttp2.c
+++ b/src/core/surface/server_chttp2.c
@@ -82,7 +82,7 @@ static void destroy(grpc_exec_ctx *exec_ctx, grpc_server *server, void *tcpp,
grpc_closure *destroy_done) {
grpc_tcp_server *tcp = tcpp;
grpc_tcp_server_unref(exec_ctx, tcp);
- grpc_exec_ctx_enqueue(exec_ctx, destroy_done, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, destroy_done, true, NULL);
}
int grpc_server_add_insecure_http2_port(grpc_server *server, const char *addr) {
diff --git a/src/core/surface/server_create.c b/src/core/surface/server_create.c
index f30093e06b..5e37e80948 100644
--- a/src/core/surface/server_create.c
+++ b/src/core/surface/server_create.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -43,9 +43,6 @@ grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
const grpc_channel_filter *filters[3];
size_t num_filters = 0;
filters[num_filters++] = &grpc_compress_filter;
- if (grpc_channel_args_is_census_enabled(args)) {
- filters[num_filters++] = &grpc_server_census_filter;
- }
GRPC_API_TRACE("grpc_server_create(%p, %p)", 2, (args, reserved));
return grpc_server_create_from_filters(filters, num_filters, args);
}
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index a8262b7af2..6823f8f551 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -488,7 +488,7 @@ void grpc_chttp2_perform_writes(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing,
grpc_endpoint *endpoint);
void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
- void *transport_writing, int success);
+ void *transport_writing, bool success);
void grpc_chttp2_cleanup_writing(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *global,
grpc_chttp2_transport_writing *writing);
diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c
index fdad05b5fb..f78adb9e26 100644
--- a/src/core/transport/chttp2/writing.c
+++ b/src/core/transport/chttp2/writing.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -188,7 +188,7 @@ void grpc_chttp2_perform_writes(
grpc_endpoint_write(exec_ctx, endpoint, &transport_writing->outbuf,
&transport_writing->done_cb);
} else {
- grpc_exec_ctx_enqueue(exec_ctx, &transport_writing->done_cb, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &transport_writing->done_cb, true, NULL);
}
}
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 05b25fd8b0..9298573c7f 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -86,14 +86,14 @@ static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
/* forward declarations of various callbacks that we'll build closures around */
static void writing_action(grpc_exec_ctx *exec_ctx, void *t,
- int iomgr_success_ignored);
+ bool iomgr_success_ignored);
/** Set a transport level setting, and push it to our peer */
static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
uint32_t value);
/** Endpoint callback to process incoming data */
-static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success);
+static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, bool success);
/** Start disconnection chain */
static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
@@ -183,7 +183,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
and maybe they hold resources that need to be freed */
while (t->global.pings.next != &t->global.pings) {
grpc_chttp2_outstanding_ping *ping = t->global.pings.next;
- grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, false, NULL);
ping->next->prev = ping->prev;
ping->prev->next = ping->next;
gpr_free(ping);
@@ -602,7 +602,7 @@ static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
t->parsing_active)) {
t->writing_active = 1;
REF_TRANSPORT(t, "writing");
- grpc_exec_ctx_enqueue(exec_ctx, &t->writing_action, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &t->writing_action, true, NULL);
prevent_endpoint_shutdown(t);
}
check_read_ops(exec_ctx, &t->global);
@@ -631,7 +631,7 @@ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
}
void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
- void *transport_writing_ptr, int success) {
+ void *transport_writing_ptr, bool success) {
grpc_chttp2_transport_writing *transport_writing = transport_writing_ptr;
grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing);
grpc_chttp2_stream_global *stream_global;
@@ -669,7 +669,7 @@ void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
}
static void writing_action(grpc_exec_ctx *exec_ctx, void *gt,
- int iomgr_success_ignored) {
+ bool iomgr_success_ignored) {
grpc_chttp2_transport *t = gt;
GPR_TIMER_BEGIN("writing_action", 0);
grpc_chttp2_perform_writes(exec_ctx, &t->writing, t->ep);
@@ -759,7 +759,7 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
closure->final_data |= 1;
}
if (closure->final_data < 2) {
- grpc_exec_ctx_enqueue(exec_ctx, closure, closure->final_data == 0);
+ grpc_exec_ctx_enqueue(exec_ctx, closure, closure->final_data == 0, NULL);
}
*pclosure = NULL;
}
@@ -777,7 +777,7 @@ static int contains_non_ok_status(
return 0;
}
-static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, int success) {}
+static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, bool success) {}
static void perform_stream_op_locked(
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
@@ -934,7 +934,7 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
for (ping = transport_global->pings.next; ping != &transport_global->pings;
ping = ping->next) {
if (0 == memcmp(opaque_8bytes, ping->id, 8)) {
- grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, ping->on_recv, true, NULL);
ping->next->prev = ping->prev;
ping->prev->next = ping->next;
gpr_free(ping);
@@ -951,7 +951,7 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
lock(t);
- grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, true, NULL);
if (op->on_connectivity_state_change != NULL) {
grpc_connectivity_state_notify_on_state_change(
@@ -1022,11 +1022,13 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
*stream_global->recv_message = grpc_chttp2_incoming_frame_queue_pop(
&stream_global->incoming_frames);
GPR_ASSERT(*stream_global->recv_message != NULL);
- grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, true,
+ NULL);
stream_global->recv_message_ready = NULL;
} else if (stream_global->published_trailing_metadata) {
*stream_global->recv_message = NULL;
- grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, true,
+ NULL);
stream_global->recv_message_ready = NULL;
}
}
@@ -1336,7 +1338,7 @@ static void read_error_locked(grpc_exec_ctx *exec_ctx,
}
/* tcp read callback */
-static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) {
+static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, bool success) {
size_t i;
int keep_reading = 0;
grpc_chttp2_transport *t = tp;
@@ -1523,7 +1525,7 @@ static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx,
unlock(exec_ctx, bs->transport);
return 1;
} else if (bs->failed) {
- grpc_exec_ctx_enqueue(exec_ctx, on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, on_complete, false, NULL);
unlock(exec_ctx, bs->transport);
return 0;
} else {
@@ -1552,7 +1554,7 @@ void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,
gpr_mu_lock(&bs->transport->mu);
if (bs->on_next != NULL) {
*bs->next = slice;
- grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, true, NULL);
bs->on_next = NULL;
} else {
gpr_slice_buffer_add(&bs->slices, slice);
@@ -1567,7 +1569,7 @@ void grpc_chttp2_incoming_byte_stream_finished(
if (from_parsing_thread) {
gpr_mu_lock(&bs->transport->mu);
}
- grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, false, NULL);
bs->on_next = NULL;
bs->failed = 1;
if (from_parsing_thread) {
diff --git a/src/core/transport/connectivity_state.c b/src/core/transport/connectivity_state.c
index 3c3fd4671d..87765b9799 100644
--- a/src/core/transport/connectivity_state.c
+++ b/src/core/transport/connectivity_state.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -78,7 +78,7 @@ void grpc_connectivity_state_destroy(grpc_exec_ctx *exec_ctx,
} else {
success = 0;
}
- grpc_exec_ctx_enqueue(exec_ctx, w->notify, success);
+ grpc_exec_ctx_enqueue(exec_ctx, w->notify, success, NULL);
gpr_free(w);
}
gpr_free(tracker->name);
@@ -109,7 +109,7 @@ int grpc_connectivity_state_notify_on_state_change(
if (current == NULL) {
grpc_connectivity_state_watcher *w = tracker->watchers;
if (w != NULL && w->notify == notify) {
- grpc_exec_ctx_enqueue(exec_ctx, notify, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, notify, false, NULL);
tracker->watchers = w->next;
gpr_free(w);
return 0;
@@ -117,7 +117,7 @@ int grpc_connectivity_state_notify_on_state_change(
while (w != NULL) {
grpc_connectivity_state_watcher *rm_candidate = w->next;
if (rm_candidate != NULL && rm_candidate->notify == notify) {
- grpc_exec_ctx_enqueue(exec_ctx, notify, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, notify, false, NULL);
w->next = w->next->next;
gpr_free(rm_candidate);
return 0;
@@ -128,7 +128,7 @@ int grpc_connectivity_state_notify_on_state_change(
} else {
if (tracker->current_state != *current) {
*current = tracker->current_state;
- grpc_exec_ctx_enqueue(exec_ctx, notify, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, notify, true, NULL);
} else {
grpc_connectivity_state_watcher *w = gpr_malloc(sizeof(*w));
w->current = current;
@@ -158,7 +158,7 @@ void grpc_connectivity_state_set(grpc_exec_ctx *exec_ctx,
while ((w = tracker->watchers) != NULL) {
*w->current = tracker->current_state;
tracker->watchers = w->next;
- grpc_exec_ctx_enqueue(exec_ctx, w->notify, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, w->notify, true, NULL);
gpr_free(w);
}
}
diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c
index 2ab978be46..08d685668c 100644
--- a/src/core/transport/transport.c
+++ b/src/core/transport/transport.c
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -59,7 +59,7 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx,
grpc_stream_refcount *refcount) {
#endif
if (gpr_unref(&refcount->refs)) {
- grpc_exec_ctx_enqueue(exec_ctx, &refcount->destroy, 1);
+ grpc_exec_ctx_enqueue(exec_ctx, &refcount->destroy, true, NULL);
}
}
@@ -125,8 +125,8 @@ char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx,
void grpc_transport_stream_op_finish_with_failure(
grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) {
- grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, 0);
- grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, 0);
+ grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, false, NULL);
+ grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, false, NULL);
}
void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op,
@@ -150,7 +150,7 @@ typedef struct {
grpc_closure closure;
} close_message_data;
-static void free_message(grpc_exec_ctx *exec_ctx, void *p, int iomgr_success) {
+static void free_message(grpc_exec_ctx *exec_ctx, void *p, bool iomgr_success) {
close_message_data *cmd = p;
gpr_slice_unref(cmd->message);
if (cmd->then_call != NULL) {
diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h
index f94f0ae76e..f5cac77adc 100644
--- a/src/core/transport/transport.h
+++ b/src/core/transport/transport.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -81,16 +81,29 @@ void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount);
/* Transport stream op: a set of operations to perform on a transport
against a single stream */
typedef struct grpc_transport_stream_op {
+ /** Send initial metadata to the peer, from the provided metadata batch. */
grpc_metadata_batch *send_initial_metadata;
+
+ /** Send trailing metadata to the peer, from the provided metadata batch. */
grpc_metadata_batch *send_trailing_metadata;
+ /** Send message data to the peer, from the provided byte stream. */
grpc_byte_stream *send_message;
+ /** Receive initial metadata from the stream, into provided metadata batch. */
grpc_metadata_batch *recv_initial_metadata;
+
+ /** Receive message data from the stream, into provided byte stream. */
grpc_byte_stream **recv_message;
+ /** Should be enqueued when one message is ready to be processed. */
grpc_closure *recv_message_ready;
+
+ /** Receive trailing metadata from the stream, into provided metadata batch.
+ */
grpc_metadata_batch *recv_trailing_metadata;
+ /** Should be enqueued when all requested operations (excluding recv_message
+ which has its own closure) in a given batch have been completed. */
grpc_closure *on_complete;
/** If != GRPC_STATUS_OK, cancel this stream */
diff --git a/src/cpp/client/channel.cc b/src/cpp/client/channel.cc
index c7974d655b..ae20392d11 100644
--- a/src/cpp/client/channel.cc
+++ b/src/cpp/client/channel.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -35,24 +35,29 @@
#include <memory>
-#include <grpc/grpc.h>
-#include <grpc/support/log.h>
-#include <grpc/support/slice.h>
#include <grpc++/client_context.h>
#include <grpc++/completion_queue.h>
-#include <grpc++/security/credentials.h>
#include <grpc++/impl/call.h>
+#include <grpc++/impl/codegen/completion_queue_tag.h>
+#include <grpc++/impl/grpc_library.h>
#include <grpc++/impl/rpc_method.h>
+#include <grpc++/security/credentials.h>
#include <grpc++/support/channel_arguments.h>
#include <grpc++/support/config.h>
#include <grpc++/support/status.h>
#include <grpc++/support/time.h>
+#include <grpc/grpc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/slice.h>
#include "src/core/profiling/timers.h"
namespace grpc {
+static internal::GrpcLibraryInitializer g_gli_initializer;
Channel::Channel(const grpc::string& host, grpc_channel* channel)
- : host_(host), c_channel_(channel) {}
+ : host_(host), c_channel_(channel) {
+ g_gli_initializer.summon();
+}
Channel::~Channel() { grpc_channel_destroy(c_channel_); }
diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index 710d7cb5c2..73147fd7bb 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -33,6 +33,7 @@
#include <grpc++/client_context.h>
+#include <grpc/compression.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/string_util.h>
diff --git a/src/cpp/client/create_channel.cc b/src/cpp/client/create_channel.cc
index 3bbca807d3..fdaa28ffef 100644
--- a/src/cpp/client/create_channel.cc
+++ b/src/cpp/client/create_channel.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -36,6 +36,7 @@
#include <grpc++/channel.h>
#include <grpc++/create_channel.h>
+#include <grpc++/impl/grpc_library.h>
#include <grpc++/support/channel_arguments.h>
#include "src/cpp/client/create_channel_internal.h"
@@ -53,7 +54,8 @@ std::shared_ptr<Channel> CreateCustomChannel(
const grpc::string& target,
const std::shared_ptr<ChannelCredentials>& creds,
const ChannelArguments& args) {
- GrpcLibrary init_lib; // We need to call init in case of a bad creds.
+ internal::GrpcLibrary
+ init_lib; // We need to call init in case of a bad creds.
ChannelArguments cp_args = args;
std::ostringstream user_agent_prefix;
user_agent_prefix << "grpc-c++/" << grpc_version_string();
diff --git a/src/cpp/client/credentials.cc b/src/cpp/client/credentials.cc
index 0c08db11a9..6fb620b0ea 100644
--- a/src/cpp/client/credentials.cc
+++ b/src/cpp/client/credentials.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -31,12 +31,18 @@
*
*/
+#include <grpc++/impl/grpc_library.h>
#include <grpc++/security/credentials.h>
namespace grpc {
+static internal::GrpcLibraryInitializer g_gli_initializer;
+ChannelCredentials::ChannelCredentials() { g_gli_initializer.summon(); }
+
ChannelCredentials::~ChannelCredentials() {}
+CallCredentials::CallCredentials() { g_gli_initializer.summon(); }
+
CallCredentials::~CallCredentials() {}
} // namespace grpc
diff --git a/src/cpp/client/secure_credentials.cc b/src/cpp/client/secure_credentials.cc
index 96ae25b761..074dae7ca7 100644
--- a/src/cpp/client/secure_credentials.cc
+++ b/src/cpp/client/secure_credentials.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -41,6 +41,13 @@
namespace grpc {
+static internal::GrpcLibraryInitializer g_gli_initializer;
+SecureChannelCredentials::SecureChannelCredentials(
+ grpc_channel_credentials* c_creds)
+ : c_creds_(c_creds) {
+ g_gli_initializer.summon();
+}
+
std::shared_ptr<grpc::Channel> SecureChannelCredentials::CreateChannel(
const string& target, const grpc::ChannelArguments& args) {
grpc_channel_args channel_args;
@@ -51,6 +58,12 @@ std::shared_ptr<grpc::Channel> SecureChannelCredentials::CreateChannel(
nullptr));
}
+SecureCallCredentials::SecureCallCredentials(grpc_call_credentials* c_creds)
+ : c_creds_(c_creds) {
+ internal::GrpcLibraryInitializer gli_initializer;
+ gli_initializer.summon();
+}
+
bool SecureCallCredentials::ApplyToCall(grpc_call* call) {
return grpc_call_set_credentials(call, c_creds_) == GRPC_CALL_OK;
}
diff --git a/src/cpp/client/secure_credentials.h b/src/cpp/client/secure_credentials.h
index cef59292dd..9e84102154 100644
--- a/src/cpp/client/secure_credentials.h
+++ b/src/cpp/client/secure_credentials.h
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -45,11 +45,8 @@ namespace grpc {
class SecureChannelCredentials GRPC_FINAL : public ChannelCredentials {
public:
- explicit SecureChannelCredentials(grpc_channel_credentials* c_creds)
- : c_creds_(c_creds) {}
- ~SecureChannelCredentials() GRPC_OVERRIDE {
- grpc_channel_credentials_release(c_creds_);
- }
+ explicit SecureChannelCredentials(grpc_channel_credentials* c_creds);
+ ~SecureChannelCredentials() { grpc_channel_credentials_release(c_creds_); }
grpc_channel_credentials* GetRawCreds() { return c_creds_; }
std::shared_ptr<grpc::Channel> CreateChannel(
@@ -62,11 +59,8 @@ class SecureChannelCredentials GRPC_FINAL : public ChannelCredentials {
class SecureCallCredentials GRPC_FINAL : public CallCredentials {
public:
- explicit SecureCallCredentials(grpc_call_credentials* c_creds)
- : c_creds_(c_creds) {}
- ~SecureCallCredentials() GRPC_OVERRIDE {
- grpc_call_credentials_release(c_creds_);
- }
+ explicit SecureCallCredentials(grpc_call_credentials* c_creds);
+ ~SecureCallCredentials() { grpc_call_credentials_release(c_creds_); }
grpc_call_credentials* GetRawCreds() { return c_creds_; }
bool ApplyToCall(grpc_call* call) GRPC_OVERRIDE;
diff --git a/src/cpp/codegen/grpc_library.cc b/src/cpp/codegen/grpc_library.cc
new file mode 100644
index 0000000000..48acec3f3d
--- /dev/null
+++ b/src/cpp/codegen/grpc_library.cc
@@ -0,0 +1,40 @@
+/*
+ *
+ * Copyright 2016, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+#include <grpc++/impl/codegen/grpc_library.h>
+
+namespace grpc {
+
+GrpcLibraryInterface *g_glip = nullptr;
+
+} // namespace grpc
diff --git a/src/cpp/common/alarm.cc b/src/cpp/common/alarm.cc
index 1f0f04175e..807a67df24 100644
--- a/src/cpp/common/alarm.cc
+++ b/src/cpp/common/alarm.cc
@@ -1,5 +1,5 @@
/*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -30,13 +30,18 @@
*
*/
-#include <grpc/grpc.h>
#include <grpc++/alarm.h>
+#include <grpc++/completion_queue.h>
+#include <grpc++/impl/grpc_library.h>
+#include <grpc/grpc.h>
namespace grpc {
+static internal::GrpcLibraryInitializer g_gli_initializer;
Alarm::Alarm(CompletionQueue* cq, gpr_timespec deadline, void* tag)
- : alarm_(grpc_alarm_create(cq->cq(), deadline, tag)) {}
+ : alarm_(grpc_alarm_create(cq->cq(), deadline, tag)) {
+ g_gli_initializer.summon();
+}
Alarm::~Alarm() { grpc_alarm_destroy(alarm_); }
diff --git a/src/cpp/common/completion_queue.cc b/src/cpp/common/completion_queue.cc
index a175beb452..4f76dfff1d 100644
--- a/src/cpp/common/completion_queue.cc
+++ b/src/cpp/common/completion_queue.cc
@@ -1,5 +1,5 @@
/*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -34,13 +34,17 @@
#include <memory>
+#include <grpc++/impl/codegen/completion_queue_tag.h>
+#include <grpc++/impl/grpc_library.h>
+#include <grpc++/support/time.h>
#include <grpc/grpc.h>
#include <grpc/support/log.h>
-#include <grpc++/support/time.h>
namespace grpc {
+static internal::GrpcLibraryInitializer g_gli_initializer;
CompletionQueue::CompletionQueue() {
+ g_gli_initializer.summon();
cq_ = grpc_completion_queue_create(nullptr);
}
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index 3bf9f3fa0f..0d31140924 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -35,16 +35,19 @@
#include <utility>
-#include <grpc/grpc.h>
-#include <grpc/support/alloc.h>
-#include <grpc/support/log.h>
#include <grpc++/completion_queue.h>
#include <grpc++/generic/async_generic_service.h>
+#include <grpc++/impl/codegen/completion_queue_tag.h>
+#include <grpc++/impl/grpc_library.h>
+#include <grpc++/impl/method_handler_impl.h>
#include <grpc++/impl/rpc_service_method.h>
#include <grpc++/impl/service_type.h>
-#include <grpc++/server_context.h>
#include <grpc++/security/server_credentials.h>
+#include <grpc++/server_context.h>
#include <grpc++/support/time.h>
+#include <grpc/grpc.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
#include "src/core/profiling/timers.h"
#include "src/cpp/server/thread_pool_interface.h"
@@ -275,6 +278,7 @@ static grpc_server* CreateServer(const ChannelArguments& args) {
return grpc_server_create(&channel_args, nullptr);
}
+static internal::GrpcLibraryInitializer g_gli_initializer;
Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
int max_message_size, const ChannelArguments& args)
: max_message_size_(max_message_size),
@@ -286,6 +290,7 @@ Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
server_(CreateServer(args)),
thread_pool_(thread_pool),
thread_pool_owned_(thread_pool_owned) {
+ g_gli_initializer.summon();
gpr_once_init(&g_once_init_callbacks, InitGlobalCallbacks);
global_callbacks_ = g_callbacks;
grpc_server_register_completion_queue(server_, cq_.cq(), nullptr);
@@ -297,6 +302,8 @@ Server::~Server() {
if (started_ && !shutdown_) {
lock.unlock();
Shutdown();
+ } else if (!started_) {
+ cq_.Shutdown();
}
}
void* got_tag;
@@ -315,36 +322,31 @@ void Server::SetGlobalCallbacks(GlobalCallbacks* callbacks) {
g_callbacks.reset(callbacks);
}
-bool Server::RegisterService(const grpc::string* host, RpcService* service) {
- for (int i = 0; i < service->GetMethodCount(); ++i) {
- RpcServiceMethod* method = service->GetMethod(i);
+bool Server::RegisterService(const grpc::string* host, Service* service) {
+ bool has_async_methods = service->has_async_methods();
+ if (has_async_methods) {
+ GPR_ASSERT(service->server_ == nullptr &&
+ "Can only register an asynchronous service against one server.");
+ service->server_ = this;
+ }
+ for (auto it = service->methods_.begin(); it != service->methods_.end();
+ ++it) {
+ if (it->get() == nullptr) { // Handled by generic service if any.
+ continue;
+ }
+ RpcServiceMethod* method = it->get();
void* tag = grpc_server_register_method(server_, method->name(),
host ? host->c_str() : nullptr);
- if (!tag) {
+ if (tag == nullptr) {
gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
method->name());
return false;
}
- sync_methods_->emplace_back(method, tag);
- }
- return true;
-}
-
-bool Server::RegisterAsyncService(const grpc::string* host,
- AsynchronousService* service) {
- GPR_ASSERT(service->server_ == nullptr &&
- "Can only register an asynchronous service against one server.");
- service->server_ = this;
- service->request_args_ = new void* [service->method_count_];
- for (size_t i = 0; i < service->method_count_; ++i) {
- void* tag = grpc_server_register_method(server_, service->method_names_[i],
- host ? host->c_str() : nullptr);
- if (!tag) {
- gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
- service->method_names_[i]);
- return false;
+ if (method->handler() == nullptr) {
+ method->set_server_tag(tag);
+ } else {
+ sync_methods_->emplace_back(method, tag);
}
- service->request_args_[i] = tag;
}
return true;
}
@@ -440,8 +442,8 @@ void Server::PerformOpsOnCall(CallOpSetInterface* ops, Call* call) {
GPR_ASSERT(GRPC_CALL_OK == result);
}
-Server::BaseAsyncRequest::BaseAsyncRequest(
- Server* server, ServerContext* context,
+ServerInterface::BaseAsyncRequest::BaseAsyncRequest(
+ ServerInterface* server, ServerContext* context,
ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag,
bool delete_on_finalize)
: server_(server),
@@ -454,9 +456,8 @@ Server::BaseAsyncRequest::BaseAsyncRequest(
memset(&initial_metadata_array_, 0, sizeof(initial_metadata_array_));
}
-Server::BaseAsyncRequest::~BaseAsyncRequest() {}
-
-bool Server::BaseAsyncRequest::FinalizeResult(void** tag, bool* status) {
+bool ServerInterface::BaseAsyncRequest::FinalizeResult(void** tag,
+ bool* status) {
if (*status) {
for (size_t i = 0; i < initial_metadata_array_.count; i++) {
context_->client_metadata_.insert(
@@ -470,7 +471,7 @@ bool Server::BaseAsyncRequest::FinalizeResult(void** tag, bool* status) {
grpc_metadata_array_destroy(&initial_metadata_array_);
context_->set_call(call_);
context_->cq_ = call_cq_;
- Call call(call_, server_, call_cq_, server_->max_message_size_);
+ Call call(call_, server_, call_cq_, server_->max_message_size());
if (*status && call_) {
context_->BeginCompletionOp(&call);
}
@@ -483,22 +484,22 @@ bool Server::BaseAsyncRequest::FinalizeResult(void** tag, bool* status) {
return true;
}
-Server::RegisteredAsyncRequest::RegisteredAsyncRequest(
- Server* server, ServerContext* context,
+ServerInterface::RegisteredAsyncRequest::RegisteredAsyncRequest(
+ ServerInterface* server, ServerContext* context,
ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq, void* tag)
: BaseAsyncRequest(server, context, stream, call_cq, tag, true) {}
-void Server::RegisteredAsyncRequest::IssueRequest(
+void ServerInterface::RegisteredAsyncRequest::IssueRequest(
void* registered_method, grpc_byte_buffer** payload,
ServerCompletionQueue* notification_cq) {
grpc_server_request_registered_call(
- server_->server_, registered_method, &call_, &context_->deadline_,
+ server_->server(), registered_method, &call_, &context_->deadline_,
&initial_metadata_array_, payload, call_cq_->cq(), notification_cq->cq(),
this);
}
-Server::GenericAsyncRequest::GenericAsyncRequest(
- Server* server, GenericServerContext* context,
+ServerInterface::GenericAsyncRequest::GenericAsyncRequest(
+ ServerInterface* server, GenericServerContext* context,
ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
ServerCompletionQueue* notification_cq, void* tag, bool delete_on_finalize)
: BaseAsyncRequest(server, context, stream, call_cq, tag,
@@ -506,12 +507,13 @@ Server::GenericAsyncRequest::GenericAsyncRequest(
grpc_call_details_init(&call_details_);
GPR_ASSERT(notification_cq);
GPR_ASSERT(call_cq);
- grpc_server_request_call(server->server_, &call_, &call_details_,
+ grpc_server_request_call(server->server(), &call_, &call_details_,
&initial_metadata_array_, call_cq->cq(),
notification_cq->cq(), this);
}
-bool Server::GenericAsyncRequest::FinalizeResult(void** tag, bool* status) {
+bool ServerInterface::GenericAsyncRequest::FinalizeResult(void** tag,
+ bool* status) {
// TODO(yangg) remove the copy here.
if (*status) {
static_cast<GenericServerContext*>(context_)->method_ =
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index 26c0724a30..a8c188e5a5 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -43,7 +43,7 @@
namespace grpc {
ServerBuilder::ServerBuilder()
- : max_message_size_(-1), generic_service_(nullptr), thread_pool_(nullptr) {
+ : max_message_size_(-1), generic_service_(nullptr) {
grpc_compression_options_init(&compression_options_);
}
@@ -53,24 +53,13 @@ std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue() {
return std::unique_ptr<ServerCompletionQueue>(cq);
}
-void ServerBuilder::RegisterService(SynchronousService* service) {
- services_.emplace_back(new NamedService<RpcService>(service->service()));
-}
-
-void ServerBuilder::RegisterAsyncService(AsynchronousService* service) {
- async_services_.emplace_back(new NamedService<AsynchronousService>(service));
+void ServerBuilder::RegisterService(Service* service) {
+ services_.emplace_back(new NamedService(service));
}
void ServerBuilder::RegisterService(const grpc::string& addr,
- SynchronousService* service) {
- services_.emplace_back(
- new NamedService<RpcService>(addr, service->service()));
-}
-
-void ServerBuilder::RegisterAsyncService(const grpc::string& addr,
- AsynchronousService* service) {
- async_services_.emplace_back(
- new NamedService<AsynchronousService>(addr, service));
+ Service* service) {
+ services_.emplace_back(new NamedService(addr, service));
}
void ServerBuilder::RegisterAsyncGenericService(AsyncGenericService* service) {
@@ -96,14 +85,14 @@ void ServerBuilder::AddListeningPort(const grpc::string& addr,
}
std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
- bool thread_pool_owned = false;
- if (!async_services_.empty() && !services_.empty()) {
- gpr_log(GPR_ERROR, "Mixing async and sync services is unsupported for now");
- return nullptr;
- }
- if (!thread_pool_ && !services_.empty()) {
- thread_pool_ = CreateDefaultThreadPool();
- thread_pool_owned = true;
+ std::unique_ptr<ThreadPoolInterface> thread_pool;
+ for (auto it = services_.begin(); it != services_.end(); ++it) {
+ if ((*it)->service->has_synchronous_methods()) {
+ if (thread_pool == nullptr) {
+ thread_pool.reset(CreateDefaultThreadPool());
+ break;
+ }
+ }
}
ChannelArguments args;
for (auto option = options_.begin(); option != options_.end(); ++option) {
@@ -115,7 +104,7 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
args.SetInt(GRPC_COMPRESSION_ALGORITHM_STATE_ARG,
compression_options_.enabled_algorithms_bitset);
std::unique_ptr<Server> server(
- new Server(thread_pool_, thread_pool_owned, max_message_size_, args));
+ new Server(thread_pool.release(), true, max_message_size_, args));
for (auto cq = cqs_.begin(); cq != cqs_.end(); ++cq) {
grpc_server_register_completion_queue(server->server_, (*cq)->cq(),
nullptr);
@@ -126,15 +115,17 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
return nullptr;
}
}
- for (auto service = async_services_.begin(); service != async_services_.end();
- service++) {
- if (!server->RegisterAsyncService((*service)->host.get(),
- (*service)->service)) {
- return nullptr;
- }
- }
if (generic_service_) {
server->RegisterAsyncGenericService(generic_service_);
+ } else {
+ for (auto it = services_.begin(); it != services_.end(); ++it) {
+ if ((*it)->service->has_generic_methods()) {
+ gpr_log(GPR_ERROR,
+ "Some methods were marked generic but there is no "
+ "generic service registered.");
+ return nullptr;
+ }
+ }
}
for (auto port = ports_.begin(); port != ports_.end(); port++) {
int r = server->AddListeningPort(port->addr, port->creds.get());
diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc
index b3a74c7fce..3732c1f090 100644
--- a/src/cpp/server/server_context.cc
+++ b/src/cpp/server/server_context.cc
@@ -33,9 +33,11 @@
#include <grpc++/server_context.h>
+#include <grpc/compression.h>
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
#include <grpc/support/log.h>
+#include <grpc++/completion_queue.h>
#include <grpc++/impl/call.h>
#include <grpc++/impl/sync.h>
#include <grpc++/support/time.h>
diff --git a/src/node/performance/benchmark_server.js b/src/node/performance/benchmark_server.js
index ba61e52ba1..e48acd48f5 100644
--- a/src/node/performance/benchmark_server.js
+++ b/src/node/performance/benchmark_server.js
@@ -76,7 +76,7 @@ function unaryCall(call, callback) {
*/
function streamingCall(call) {
call.on('data', function(value) {
- var payload = {body: zeroBuffer(value.repsonse_size)};
+ var payload = {body: zeroBuffer(value.response_size)};
call.write({payload: payload});
});
call.on('end', function() {
diff --git a/src/node/performance/worker_service_impl.js b/src/node/performance/worker_service_impl.js
index 8841ae13c3..99ae32127e 100644
--- a/src/node/performance/worker_service_impl.js
+++ b/src/node/performance/worker_service_impl.js
@@ -1,6 +1,6 @@
/*
*
- * Copyright 2015, Google Inc.
+ * Copyright 2015-2016, Google Inc.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@@ -99,7 +99,7 @@ exports.runServer = function runServer(call) {
var stats;
switch (request.argtype) {
case 'setup':
- server = new BenchmarkServer(request.setup.host, request.setup.port,
+ server = new BenchmarkServer('[::]', request.setup.port,
request.setup.security_params);
server.start();
stats = server.mark();
diff --git a/src/node/src/grpc_extension.js b/src/node/src/grpc_extension.js
index d4eca65fcb..6a8fe2c03c 100644
--- a/src/node/src/grpc_extension.js
+++ b/src/node/src/grpc_extension.js
@@ -33,8 +33,8 @@
var binary = require('node-pre-gyp');
var path = require('path');
-var binding_path = binary.find(path.resolve(
- path.join(__dirname,'../../../package.json')));
+var binding_path =
+ binary.find(path.resolve(path.join(__dirname, '../../../package.json')));
var binding = require(binding_path);
module.exports = binding;
diff --git a/src/node/test/surface_test.js b/src/node/test/surface_test.js
index c65d95f767..530f1f7749 100644
--- a/src/node/test/surface_test.js
+++ b/src/node/test/surface_test.js
@@ -970,9 +970,7 @@ describe('Call propagation', function() {
proxy.start();
var proxy_client = new Client('localhost:' + proxy_port,
grpc.credentials.createInsecure());
- call = proxy_client.unary({}, function(err, value) {
- done();
- });
+ call = proxy_client.unary({}, function(err, value) { done(); });
});
it('With a client stream call', function(done) {
done = multiDone(done, 2);
@@ -994,9 +992,7 @@ describe('Call propagation', function() {
proxy.start();
var proxy_client = new Client('localhost:' + proxy_port,
grpc.credentials.createInsecure());
- call = proxy_client.clientStream(function(err, value) {
- done();
- });
+ call = proxy_client.clientStream(function(err, value) { done(); });
});
it('With a server stream call', function(done) {
done = multiDone(done, 2);
diff --git a/src/objective-c/BoringSSL.podspec b/src/objective-c/BoringSSL.podspec
index f5b738203d..4a6df910a7 100644
--- a/src/objective-c/BoringSSL.podspec
+++ b/src/objective-c/BoringSSL.podspec
@@ -31,7 +31,7 @@
Pod::Spec.new do |s|
s.name = 'BoringSSL'
- s.version = '1.0'
+ s.version = '2.0'
s.summary = 'BoringSSL is a fork of OpenSSL that is designed to meet Google’s needs.'
# Adapted from the homepage:
s.description = <<-DESC
@@ -67,7 +67,7 @@ Pod::Spec.new do |s|
s.authors = 'Adam Langley', 'David Benjamin', 'Matt Braithwaite'
s.source = { :git => 'https://boringssl.googlesource.com/boringssl',
- :tag => 'version_for_cocoapods_1.0' }
+ :tag => 'version_for_cocoapods_2.0' }
s.source_files = 'ssl/*.{h,c}',
'ssl/**/*.{h,c}',
diff --git a/src/proto/grpc/testing/control.proto b/src/proto/grpc/testing/control.proto
index 7ba6f9856f..2f352e652f 100644
--- a/src/proto/grpc/testing/control.proto
+++ b/src/proto/grpc/testing/control.proto
@@ -127,8 +127,6 @@ message ClientArgs {
message ServerConfig {
ServerType server_type = 1;
SecurityParams security_params = 2;
- // Host on which to listen.
- string host = 3;
// Port on which to listen. Zero means pick unused port.
int32 port = 4;
// Only for async server. Number of threads used to serve the requests.
diff --git a/src/python/README.md b/src/python/README.md
deleted file mode 100644
index 8b167659ff..0000000000
--- a/src/python/README.md
+++ /dev/null
@@ -1,100 +0,0 @@
-gRPC Python
-=========
-The Python facility of gRPC.
-
-Status
--------
-Beta : Core behavior well-used and proven; bugs lurk off the beaten path.
-
-PREREQUISITES
--------------
-- python-virtualenv, python-pip, python3-pip, python-dev, python3-dev
-- [homebrew][] on Mac OS X. These simplify the installation of the gRPC C core.
-
-INSTALLATION
--------------
-
-**Linux (Debian):**
-
-Add [Debian jessie-backports][] to your `sources.list` file. Example:
-
-```sh
-echo "deb http://http.debian.net/debian jessie-backports main" | \
-sudo tee -a /etc/apt/sources.list
-```
-
-Install the gRPC Debian package
-
-```sh
-sudo apt-get update
-sudo apt-get install libgrpc-dev
-```
-
-Install the gRPC Python module
-
-```sh
-sudo pip install grpcio
-```
-
-**Mac OS X**
-
-Install [homebrew][]. Run the following command to install gRPC Python.
-```sh
-$ curl -fsSL https://goo.gl/getgrpc | bash -s python
-```
-This will download and run the [gRPC install script][], then install the latest version of the gRPC Python package. It also installs the Protocol Buffers compiler (_protoc_) and the gRPC _protoc_ plugin for python.
-
-EXAMPLES
---------
-Please read our online documentation for a [Quick Start][] and a [detailed example][]
-
-BUILDING FROM SOURCE
----------------------
-- Clone this repository
-
-- Install tox
-```
-$ sudo pip install tox
-```
-
-- Initialize the git submodules
-```
-$ git submodule update --init
-```
-
-- Make the libraries
-```
-$ make
-```
-
-- Use build_python.sh to build the Python code and install it into a virtual environment
-```
-$ CONFIG=opt tools/run_tests/build_python.sh
-```
-
-TESTING
--------
-
-- Use run_python.sh to run gRPC as it was installed into the virtual environment
-```
-$ CONFIG=opt PYVER=2.7 tools/run_tests/run_python.sh
-```
-
-PACKAGING
----------
-
-- Install packaging dependencies
-```
-$ pip install setuptools twine
-```
-
-- Push to PyPI
-```
-$ ../../tools/distrib/python/submit.py
-```
-
-[homebrew]:http://brew.sh
-[gRPC install script]:https://raw.githubusercontent.com/grpc/homebrew-grpc/master/scripts/install
-[Quick Start]:http://www.grpc.io/docs/tutorials/basic/python.html
-[detailed example]:http://www.grpc.io/docs/installation/python.html
-[Debian jessie-backports]:http://backports.debian.org/Instructions/
diff --git a/src/ruby/ext/grpc/rb_call.c b/src/ruby/ext/grpc/rb_call.c
index 8f33d72417..07864ebf59 100644
--- a/src/ruby/ext/grpc/rb_call.c
+++ b/src/ruby/ext/grpc/rb_call.c
@@ -530,7 +530,8 @@ typedef struct run_batch_stack {
/* grpc_run_batch_stack_init ensures the run_batch_stack is properly
* initialized */
-static void grpc_run_batch_stack_init(run_batch_stack *st, unsigned write_flag) {
+static void grpc_run_batch_stack_init(run_batch_stack *st,
+ unsigned write_flag) {
MEMZERO(st, run_batch_stack, 1);
grpc_metadata_array_init(&st->send_metadata);
grpc_metadata_array_init(&st->send_trailing_metadata);