diff --git a/include/grpc++/async_generic_service.h b/include/grpc++/async_generic_service.h
index 911d31cb1f4dbc352321f907b8ee402cee2d4383..b435c6e73d23d92adbe2bde64e085712451fefdd 100644
--- a/include/grpc++/async_generic_service.h
+++ b/include/grpc++/async_generic_service.h
@@ -65,10 +65,8 @@ class AsyncGenericService GRPC_FINAL {
void RequestCall(GenericServerContext* ctx,
GenericServerAsyncReaderWriter* reader_writer,
- CompletionQueue* cq, void* tag);
-
- // The new rpc event should be obtained from this completion queue.
- CompletionQueue* completion_queue();
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq, void* tag);
private:
friend class Server;
diff --git a/include/grpc++/completion_queue.h b/include/grpc++/completion_queue.h
index 5c2b1cce93d1557f9c139598be6d930834d268df..e8429c8f417c28379df40a4eeafeccebacce41ee 100644
--- a/include/grpc++/completion_queue.h
+++ b/include/grpc++/completion_queue.h
@@ -58,6 +58,7 @@ class ServerReaderWriter;
class CompletionQueue;
class Server;
+class ServerBuilder;
class ServerContext;
class CompletionQueueTag {
@@ -137,6 +138,12 @@ class CompletionQueue : public GrpcLibrary {
grpc_completion_queue* cq_; // owned
};
+class ServerCompletionQueue : public CompletionQueue {
+ private:
+ friend class ServerBuilder;
+ ServerCompletionQueue() {}
+};
+
} // namespace grpc
#endif // GRPCXX_COMPLETION_QUEUE_H
diff --git a/include/grpc++/impl/service_type.h b/include/grpc++/impl/service_type.h
index 7cd3ddad6b7a9f594805869938c2ec5160b6c4fa..bc39bb82ac32683db8956cc9f283a592b2909fcd 100644
--- a/include/grpc++/impl/service_type.h
+++ b/include/grpc++/impl/service_type.h
@@ -39,8 +39,10 @@
namespace grpc {
class Call;
+class CompletionQueue;
class RpcService;
class Server;
+class ServerCompletionQueue;
class ServerContext;
class Status;
@@ -70,52 +72,55 @@ class AsynchronousService {
ServerContext* context,
::grpc::protobuf::Message* request,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) = 0;
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag) = 0;
};
- AsynchronousService(CompletionQueue* cq, const char** method_names,
- size_t method_count)
- : cq_(cq),
- dispatch_impl_(nullptr),
+ AsynchronousService(const char** method_names, size_t method_count)
+ : dispatch_impl_(nullptr),
method_names_(method_names),
method_count_(method_count),
request_args_(nullptr) {}
~AsynchronousService() { delete[] request_args_; }
- CompletionQueue* completion_queue() const { return cq_; }
-
protected:
void RequestAsyncUnary(int index, ServerContext* context,
grpc::protobuf::Message* request,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) {
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq, void* tag) {
dispatch_impl_->RequestAsyncCall(request_args_[index], context, request,
- stream, cq, tag);
+ stream, call_cq, notification_cq, tag);
}
void RequestClientStreaming(int index, ServerContext* context,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) {
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag) {
dispatch_impl_->RequestAsyncCall(request_args_[index], context, nullptr,
- stream, cq, tag);
+ stream, call_cq, notification_cq, tag);
}
void RequestServerStreaming(int index, ServerContext* context,
grpc::protobuf::Message* request,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) {
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag) {
dispatch_impl_->RequestAsyncCall(request_args_[index], context, request,
- stream, cq, tag);
+ stream, call_cq, notification_cq, tag);
}
void RequestBidiStreaming(int index, ServerContext* context,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) {
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq, void* tag) {
dispatch_impl_->RequestAsyncCall(request_args_[index], context, nullptr,
- stream, cq, tag);
+ stream, call_cq, notification_cq, tag);
}
private:
friend class Server;
- CompletionQueue* const cq_;
DispatchImpl* dispatch_impl_;
const char** const method_names_;
size_t method_count_;
diff --git a/include/grpc++/server.h b/include/grpc++/server.h
index b2b9044dcab877504d624ade2c9a366f707d74f7..50a24163219bee7cc5f6de163da49eca48b031a4 100644
--- a/include/grpc++/server.h
+++ b/include/grpc++/server.h
@@ -101,11 +101,15 @@ class Server GRPC_FINAL : public GrpcLibrary,
void RequestAsyncCall(void* registered_method, ServerContext* context,
grpc::protobuf::Message* request,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) GRPC_OVERRIDE;
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag) GRPC_OVERRIDE;
void RequestAsyncGenericCall(GenericServerContext* context,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag);
+ CompletionQueue* cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag);
const int max_message_size_;
diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h
index 7155c7fd46237d535c7824d85c92c62dc734c011..ecee475e3e43f5d95d65d0feb016de071a8cf8a9 100644
--- a/include/grpc++/server_builder.h
+++ b/include/grpc++/server_builder.h
@@ -46,6 +46,7 @@ class AsynchronousService;
class CompletionQueue;
class RpcService;
class Server;
+class ServerCompletionQueue;
class ServerCredentials;
class SynchronousService;
class ThreadPoolInterface;
@@ -82,6 +83,11 @@ class ServerBuilder {
// Does not take ownership.
void SetThreadPool(ThreadPoolInterface* thread_pool);
+ // Add a completion queue for handling asynchronous services
+ // Caller is required to keep this completion queue live until calling
+ // BuildAndStart()
+ std::unique_ptr<ServerCompletionQueue> AddCompletionQueue();
+
// Return a running server which is ready for processing rpcs.
std::unique_ptr<Server> BuildAndStart();
@@ -96,6 +102,7 @@ class ServerBuilder {
std::vector<RpcService*> services_;
std::vector<AsynchronousService*> async_services_;
std::vector<Port> ports_;
+ std::vector<ServerCompletionQueue*> cqs_;
std::shared_ptr<ServerCredentials> creds_;
AsyncGenericService* generic_service_;
ThreadPoolInterface* thread_pool_;
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index 33486539567c8afda2429fc7bf8cced6d462874d..0212b42188c821f454f66018a7ebdbd01b2de42c 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -462,7 +462,8 @@ void grpc_call_destroy(grpc_call *call);
grpc_call_error grpc_server_request_call(
grpc_server *server, grpc_call **call, grpc_call_details *details,
grpc_metadata_array *request_metadata,
- grpc_completion_queue *cq_bound_to_call, void *tag_new);
+ grpc_completion_queue *cq_bound_to_call,
+ grpc_completion_queue *cq_for_notification, void *tag_new);
/* Registers a method in the server.
Methods to this (host, method) pair will not be reported by
@@ -472,21 +473,26 @@ grpc_call_error grpc_server_request_call(
Must be called before grpc_server_start.
Returns NULL on failure. */
void *grpc_server_register_method(grpc_server *server, const char *method,
- const char *host,
- grpc_completion_queue *new_call_cq);
+ const char *host);
/* Request notification of a new pre-registered call */
grpc_call_error grpc_server_request_registered_call(
grpc_server *server, void *registered_method, grpc_call **call,
gpr_timespec *deadline, grpc_metadata_array *request_metadata,
grpc_byte_buffer **optional_payload,
- grpc_completion_queue *cq_bound_to_call, void *tag_new);
+ grpc_completion_queue *cq_bound_to_call,
+ grpc_completion_queue *cq_for_notification, void *tag_new);
/* Create a server. Additional configuration for each incoming channel can
be specified with args. If no additional configuration is needed, args can
be NULL. See grpc_channel_args for more. */
-grpc_server *grpc_server_create(grpc_completion_queue *cq,
- const grpc_channel_args *args);
+grpc_server *grpc_server_create(const grpc_channel_args *args);
+
+/* Register a completion queue with the server. Must be done for any completion
+ queue that is passed to grpc_server_request_* call. Must be performed prior
+ to grpc_server_start. */
+void grpc_server_register_completion_queue(grpc_server *server,
+ grpc_completion_queue *cq);
/* Add a HTTP2 over plaintext over tcp listener.
Returns bound port number on success, 0 on failure.
diff --git a/src/compiler/cpp_generator.cc b/src/compiler/cpp_generator.cc
index 084db17a30ed1aa8509b21cf20961fdc37335dba..acac1475f7c174b8c439cd69bf2cf8c72b0fa4ff 100644
--- a/src/compiler/cpp_generator.cc
+++ b/src/compiler/cpp_generator.cc
@@ -120,6 +120,7 @@ grpc::string GetHeaderIncludes(const grpc::protobuf::FileDescriptor *file,
"class CompletionQueue;\n"
"class ChannelInterface;\n"
"class RpcService;\n"
+ "class ServerCompletionQueue;\n"
"class ServerContext;\n"
"} // namespace grpc\n\n";
@@ -499,30 +500,37 @@ void PrintHeaderServerMethodAsync(
(*vars)["Response"] =
grpc_cpp_generator::ClassName(method->output_type(), true);
if (NoStreaming(method)) {
- printer->Print(*vars,
- "void Request$Method$("
- "::grpc::ServerContext* context, $Request$* request, "
- "::grpc::ServerAsyncResponseWriter< $Response$>* response, "
- "::grpc::CompletionQueue* cq, void *tag);\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");
} else if (ClientOnlyStreaming(method)) {
- printer->Print(*vars,
- "void Request$Method$("
- "::grpc::ServerContext* context, "
- "::grpc::ServerAsyncReader< $Response$, $Request$>* reader, "
- "::grpc::CompletionQueue* cq, void *tag);\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");
} else if (ServerOnlyStreaming(method)) {
- printer->Print(*vars,
- "void Request$Method$("
- "::grpc::ServerContext* context, $Request$* request, "
- "::grpc::ServerAsyncWriter< $Response$>* writer, "
- "::grpc::CompletionQueue* cq, void *tag);\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");
} else if (BidiStreaming(method)) {
printer->Print(
*vars,
"void Request$Method$("
"::grpc::ServerContext* context, "
"::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
- "::grpc::CompletionQueue* cq, void *tag);\n");
+ "::grpc::CompletionQueue* new_call_cq, "
+ "::grpc::ServerCompletionQueue* notification_cq, void *tag);\n");
}
}
@@ -603,7 +611,7 @@ void PrintHeaderService(grpc::protobuf::io::Printer *printer,
" public:\n");
printer->Indent();
(*vars)["MethodCount"] = as_string(service->method_count());
- printer->Print("explicit AsyncService(::grpc::CompletionQueue* cq);\n");
+ printer->Print("explicit AsyncService();\n");
printer->Print("~AsyncService() {};\n");
for (int i = 0; i < service->method_count(); ++i) {
PrintHeaderServerMethodAsync(printer, service->method(i), vars);
@@ -878,36 +886,43 @@ void PrintSourceServerAsyncMethod(
(*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* cq, void* tag) {\n");
+ 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, cq, tag);\n");
+ "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* cq, void* tag) {\n");
+ 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, cq, tag);\n");
+ "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* cq, void* tag) {\n");
- printer->Print(*vars,
- " AsynchronousService::RequestServerStreaming($Idx$, "
- "context, request, writer, cq, tag);\n");
+ 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(
@@ -915,10 +930,11 @@ void PrintSourceServerAsyncMethod(
"void $ns$$Service$::AsyncService::Request$Method$("
"::grpc::ServerContext* context, "
"::grpc::ServerAsyncReaderWriter< $Response$, $Request$>* stream, "
- "::grpc::CompletionQueue* cq, void *tag) {\n");
+ "::grpc::CompletionQueue* new_call_cq, "
+ "::grpc::ServerCompletionQueue* notification_cq, void *tag) {\n");
printer->Print(*vars,
" AsynchronousService::RequestBidiStreaming($Idx$, "
- "context, stream, cq, tag);\n");
+ "context, stream, new_call_cq, notification_cq, tag);\n");
printer->Print("}\n\n");
}
}
@@ -980,9 +996,8 @@ void PrintSourceService(grpc::protobuf::io::Printer *printer,
(*vars)["MethodCount"] = as_string(service->method_count());
printer->Print(*vars,
- "$ns$$Service$::AsyncService::AsyncService(::grpc::"
- "CompletionQueue* cq) : "
- "::grpc::AsynchronousService(cq, "
+ "$ns$$Service$::AsyncService::AsyncService() : "
+ "::grpc::AsynchronousService("
"$prefix$$Service$_method_names, $MethodCount$) "
"{}\n\n");
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 01644b447186f7a5b5f616d013c507b92a864a81..96c1b7c3eb2c87292678969789ce8f6b5492118d 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -74,16 +74,15 @@ typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type;
typedef struct {
requested_call_type type;
void *tag;
+ grpc_completion_queue *cq_bound_to_call;
+ grpc_completion_queue *cq_for_notification;
+ grpc_call **call;
union {
struct {
- grpc_completion_queue *cq_bind;
- grpc_call **call;
grpc_call_details *details;
grpc_metadata_array *initial_metadata;
} batch;
struct {
- grpc_completion_queue *cq_bind;
- grpc_call **call;
registered_method *registered_method;
gpr_timespec *deadline;
grpc_metadata_array *initial_metadata;
@@ -103,7 +102,6 @@ struct registered_method {
char *host;
call_data *pending;
requested_call_array requested;
- grpc_completion_queue *cq;
registered_method *next;
};
@@ -130,7 +128,6 @@ struct grpc_server {
size_t channel_filter_count;
const grpc_channel_filter **channel_filters;
grpc_channel_args *channel_args;
- grpc_completion_queue *unregistered_cq;
grpc_completion_queue **cqs;
grpc_pollset **pollsets;
@@ -602,7 +599,8 @@ static const grpc_channel_filter server_surface_filter = {
destroy_channel_elem, "server",
};
-static void addcq(grpc_server *server, grpc_completion_queue *cq) {
+void grpc_server_register_completion_queue(grpc_server *server,
+ grpc_completion_queue *cq) {
size_t i, n;
for (i = 0; i < server->cq_count; i++) {
if (server->cqs[i] == cq) return;
@@ -614,8 +612,7 @@ static void addcq(grpc_server *server, grpc_completion_queue *cq) {
server->cqs[n] = cq;
}
-grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
- grpc_channel_filter **filters,
+grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters,
size_t filter_count,
const grpc_channel_args *args) {
size_t i;
@@ -626,12 +623,10 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
GPR_ASSERT(grpc_is_initialized() && "call grpc_init()");
memset(server, 0, sizeof(grpc_server));
- if (cq) addcq(server, cq);
gpr_mu_init(&server->mu);
gpr_cv_init(&server->cv);
- server->unregistered_cq = cq;
/* decremented by grpc_server_destroy */
gpr_ref_init(&server->internal_refcount, 1);
server->root_channel_data.next = server->root_channel_data.prev =
@@ -667,8 +662,7 @@ static int streq(const char *a, const char *b) {
}
void *grpc_server_register_method(grpc_server *server, const char *method,
- const char *host,
- grpc_completion_queue *cq_new_rpc) {
+ const char *host) {
registered_method *m;
if (!method) {
gpr_log(GPR_ERROR, "%s method string cannot be NULL", __FUNCTION__);
@@ -681,13 +675,11 @@ void *grpc_server_register_method(grpc_server *server, const char *method,
return NULL;
}
}
- addcq(server, cq_new_rpc);
m = gpr_malloc(sizeof(registered_method));
memset(m, 0, sizeof(*m));
m->method = gpr_strdup(method);
m->host = gpr_strdup(host);
m->next = server->registered_methods;
- m->cq = cq_new_rpc;
server->registered_methods = m;
return m;
}
@@ -1012,17 +1004,18 @@ static grpc_call_error queue_call_request(grpc_server *server,
}
}
-grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
- grpc_call_details *details,
- grpc_metadata_array *initial_metadata,
- grpc_completion_queue *cq_bind,
- void *tag) {
+grpc_call_error grpc_server_request_call(
+ grpc_server *server, grpc_call **call, grpc_call_details *details,
+ grpc_metadata_array *initial_metadata,
+ grpc_completion_queue *cq_bound_to_call,
+ grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc;
- grpc_cq_begin_op(server->unregistered_cq, NULL, GRPC_OP_COMPLETE);
+ grpc_cq_begin_op(cq_for_notification, NULL, GRPC_OP_COMPLETE);
rc.type = BATCH_CALL;
rc.tag = tag;
- rc.data.batch.cq_bind = cq_bind;
- rc.data.batch.call = call;
+ rc.cq_bound_to_call = cq_bound_to_call;
+ rc.cq_for_notification = cq_for_notification;
+ rc.call = call;
rc.data.batch.details = details;
rc.data.batch.initial_metadata = initial_metadata;
return queue_call_request(server, &rc);
@@ -1031,14 +1024,16 @@ grpc_call_error grpc_server_request_call(grpc_server *server, grpc_call **call,
grpc_call_error grpc_server_request_registered_call(
grpc_server *server, void *rm, grpc_call **call, gpr_timespec *deadline,
grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload,
- grpc_completion_queue *cq_bind, void *tag) {
+ grpc_completion_queue *cq_bound_to_call,
+ grpc_completion_queue *cq_for_notification, void *tag) {
requested_call rc;
registered_method *registered_method = rm;
- grpc_cq_begin_op(registered_method->cq, NULL, GRPC_OP_COMPLETE);
+ grpc_cq_begin_op(cq_for_notification, NULL, GRPC_OP_COMPLETE);
rc.type = REGISTERED_CALL;
rc.tag = tag;
- rc.data.registered.cq_bind = cq_bind;
- rc.data.registered.call = call;
+ rc.cq_bound_to_call = cq_bound_to_call;
+ rc.cq_for_notification = cq_for_notification;
+ rc.call = call;
rc.data.registered.registered_method = registered_method;
rc.data.registered.deadline = deadline;
rc.data.registered.initial_metadata = initial_metadata;
@@ -1076,6 +1071,9 @@ static void begin_call(grpc_server *server, call_data *calld,
fill in the metadata array passed by the client, we need to perform
an ioreq op, that should complete immediately. */
+ grpc_call_set_completion_queue(calld->call, rc->cq_bound_to_call);
+ *rc->call = calld->call;
+ calld->cq_new = rc->cq_for_notification;
switch (rc->type) {
case BATCH_CALL:
cpstr(&rc->data.batch.details->host,
@@ -1083,18 +1081,13 @@ static void begin_call(grpc_server *server, call_data *calld,
cpstr(&rc->data.batch.details->method,
&rc->data.batch.details->method_capacity, calld->path);
rc->data.batch.details->deadline = calld->deadline;
- grpc_call_set_completion_queue(calld->call, rc->data.batch.cq_bind);
- *rc->data.batch.call = calld->call;
r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
r->data.recv_metadata = rc->data.batch.initial_metadata;
r++;
- calld->cq_new = server->unregistered_cq;
publish = publish_registered_or_batch;
break;
case REGISTERED_CALL:
*rc->data.registered.deadline = calld->deadline;
- grpc_call_set_completion_queue(calld->call, rc->data.registered.cq_bind);
- *rc->data.registered.call = calld->call;
r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
r->data.recv_metadata = rc->data.registered.initial_metadata;
r++;
@@ -1103,7 +1096,6 @@ static void begin_call(grpc_server *server, call_data *calld,
r->data.recv_message = rc->data.registered.optional_payload;
r++;
}
- calld->cq_new = rc->data.registered.registered_method->cq;
publish = publish_registered_or_batch;
break;
}
@@ -1114,20 +1106,17 @@ static void begin_call(grpc_server *server, call_data *calld,
}
static void fail_call(grpc_server *server, requested_call *rc) {
+ *rc->call = NULL;
switch (rc->type) {
case BATCH_CALL:
- *rc->data.batch.call = NULL;
rc->data.batch.initial_metadata->count = 0;
- grpc_cq_end_op(server->unregistered_cq, rc->tag, NULL, do_nothing, NULL,
- GRPC_OP_ERROR);
break;
case REGISTERED_CALL:
- *rc->data.registered.call = NULL;
rc->data.registered.initial_metadata->count = 0;
- grpc_cq_end_op(rc->data.registered.registered_method->cq, rc->tag, NULL,
- do_nothing, NULL, GRPC_OP_ERROR);
break;
}
+ grpc_cq_end_op(rc->cq_for_notification, rc->tag, NULL, do_nothing, NULL,
+ GRPC_OP_ERROR);
}
static void publish_registered_or_batch(grpc_call *call, grpc_op_error status,
diff --git a/src/core/surface/server.h b/src/core/surface/server.h
index 2cfa38fa43661df759bb22d4ba7a127a3fa062ae..c6331033e010b1948065b7748b321bea451a92a9 100644
--- a/src/core/surface/server.h
+++ b/src/core/surface/server.h
@@ -39,8 +39,7 @@
#include "src/core/transport/transport.h"
/* Create a server */
-grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq,
- grpc_channel_filter **filters,
+grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters,
size_t filter_count,
const grpc_channel_args *args);
diff --git a/src/core/surface/server_create.c b/src/core/surface/server_create.c
index f629c7c72de87ea3d48fd22d59877241633a79df..b7390675adb401b06d1d3b3a39ddce45561a895a 100644
--- a/src/core/surface/server_create.c
+++ b/src/core/surface/server_create.c
@@ -35,7 +35,6 @@
#include "src/core/surface/completion_queue.h"
#include "src/core/surface/server.h"
-grpc_server *grpc_server_create(grpc_completion_queue *cq,
- const grpc_channel_args *args) {
- return grpc_server_create_from_filters(cq, NULL, 0, args);
+grpc_server *grpc_server_create(const grpc_channel_args *args) {
+ return grpc_server_create_from_filters(NULL, 0, args);
}
diff --git a/src/cpp/server/async_generic_service.cc b/src/cpp/server/async_generic_service.cc
index 07cb933715289d0061e2424cb7bd159e47353e8e..2e99afcb5f136b379a43d2973a6b9eb5e8235389 100644
--- a/src/cpp/server/async_generic_service.cc
+++ b/src/cpp/server/async_generic_service.cc
@@ -39,12 +39,10 @@ namespace grpc {
void AsyncGenericService::RequestCall(
GenericServerContext* ctx, GenericServerAsyncReaderWriter* reader_writer,
- CompletionQueue* cq, void* tag) {
- server_->RequestAsyncGenericCall(ctx, reader_writer, cq, tag);
-}
-
-CompletionQueue* AsyncGenericService::completion_queue() {
- return &server_->cq_;
+ CompletionQueue* call_cq, ServerCompletionQueue* notification_cq,
+ void* tag) {
+ server_->RequestAsyncGenericCall(ctx, reader_writer, call_cq, notification_cq,
+ tag);
}
} // namespace grpc
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index 1ff9ff4b32b695e24ac90e578afcf92f24c2c359..62f4020d7e934d8ae383fbe5324ec12e0535c598 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -78,7 +78,7 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
return mrd;
}
- void Request(grpc_server* server) {
+ void Request(grpc_server* server, grpc_completion_queue* notify_cq) {
GPR_ASSERT(!in_flight_);
in_flight_ = true;
cq_ = grpc_completion_queue_create();
@@ -86,7 +86,7 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
grpc_server_request_registered_call(
server, tag_, &call_, &deadline_, &request_metadata_,
has_request_payload_ ? &request_payload_ : nullptr, cq_,
- this));
+ notify_cq, this));
}
bool FinalizeResult(void** tag, bool* status) GRPC_OVERRIDE {
@@ -179,16 +179,16 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
grpc_completion_queue* cq_;
};
-grpc_server* CreateServer(grpc_completion_queue* cq, int max_message_size) {
+static grpc_server* CreateServer(int max_message_size) {
if (max_message_size > 0) {
grpc_arg arg;
arg.type = GRPC_ARG_INTEGER;
arg.key = const_cast<char*>(GRPC_ARG_MAX_MESSAGE_LENGTH);
arg.value.integer = max_message_size;
grpc_channel_args args = {1, &arg};
- return grpc_server_create(cq, &args);
+ return grpc_server_create(&args);
} else {
- return grpc_server_create(cq, nullptr);
+ return grpc_server_create(nullptr);
}
}
@@ -199,9 +199,11 @@ Server::Server(ThreadPoolInterface* thread_pool, bool thread_pool_owned,
shutdown_(false),
num_running_cb_(0),
sync_methods_(new std::list<SyncRequest>),
- server_(CreateServer(cq_.cq(), max_message_size)),
+ server_(CreateServer(max_message_size)),
thread_pool_(thread_pool),
- thread_pool_owned_(thread_pool_owned) {}
+ thread_pool_owned_(thread_pool_owned) {
+ grpc_server_register_completion_queue(server_, cq_.cq());
+}
Server::~Server() {
{
@@ -221,8 +223,7 @@ Server::~Server() {
bool Server::RegisterService(RpcService* service) {
for (int i = 0; i < service->GetMethodCount(); ++i) {
RpcServiceMethod* method = service->GetMethod(i);
- void* tag =
- grpc_server_register_method(server_, method->name(), nullptr, cq_.cq());
+ void* tag = grpc_server_register_method(server_, method->name(), nullptr);
if (!tag) {
gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
method->name());
@@ -240,9 +241,8 @@ bool Server::RegisterAsyncService(AsynchronousService* service) {
service->dispatch_impl_ = 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], nullptr,
- service->completion_queue()->cq());
+ void* tag = grpc_server_register_method(server_, service->method_names_[i],
+ nullptr);
if (!tag) {
gpr_log(GPR_DEBUG, "Attempt to register %s multiple times",
service->method_names_[i]);
@@ -273,7 +273,7 @@ bool Server::Start() {
// Start processing rpcs.
if (!sync_methods_->empty()) {
for (auto m = sync_methods_->begin(); m != sync_methods_->end(); m++) {
- m->Request(server_);
+ m->Request(server_, cq_.cq());
}
ScheduleCallback();
@@ -316,12 +316,12 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
public:
AsyncRequest(Server* server, void* registered_method, ServerContext* ctx,
grpc::protobuf::Message* request,
- ServerAsyncStreamingInterface* stream, CompletionQueue* cq,
- void* tag)
+ ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq, void* tag)
: tag_(tag),
request_(request),
stream_(stream),
- cq_(cq),
+ call_cq_(call_cq),
ctx_(ctx),
generic_ctx_(nullptr),
server_(server),
@@ -329,18 +329,21 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
payload_(nullptr) {
memset(&array_, 0, sizeof(array_));
grpc_call_details_init(&call_details_);
+ GPR_ASSERT(notification_cq);
+ GPR_ASSERT(call_cq);
grpc_server_request_registered_call(
server->server_, registered_method, &call_, &call_details_.deadline,
- &array_, request ? &payload_ : nullptr, cq->cq(), this);
+ &array_, request ? &payload_ : nullptr, call_cq->cq(),
+ notification_cq->cq(), this);
}
AsyncRequest(Server* server, GenericServerContext* ctx,
- ServerAsyncStreamingInterface* stream, CompletionQueue* cq,
- void* tag)
+ ServerAsyncStreamingInterface* stream, CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq, void* tag)
: tag_(tag),
request_(nullptr),
stream_(stream),
- cq_(cq),
+ call_cq_(call_cq),
ctx_(nullptr),
generic_ctx_(ctx),
server_(server),
@@ -348,8 +351,10 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
payload_(nullptr) {
memset(&array_, 0, sizeof(array_));
grpc_call_details_init(&call_details_);
+ GPR_ASSERT(notification_cq);
+ GPR_ASSERT(call_cq);
grpc_server_request_call(server->server_, &call_, &call_details_, &array_,
- cq->cq(), this);
+ call_cq->cq(), notification_cq->cq(), this);
}
~AsyncRequest() {
@@ -392,8 +397,8 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
}
}
ctx->call_ = call_;
- ctx->cq_ = cq_;
- Call call(call_, server_, cq_, server_->max_message_size_);
+ ctx->cq_ = call_cq_;
+ Call call(call_, server_, call_cq_, server_->max_message_size_);
if (orig_status && call_) {
ctx->BeginCompletionOp(&call);
}
@@ -407,7 +412,7 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
void* const tag_;
grpc::protobuf::Message* const request_;
ServerAsyncStreamingInterface* const stream_;
- CompletionQueue* const cq_;
+ CompletionQueue* const call_cq_;
ServerContext* const ctx_;
GenericServerContext* const generic_ctx_;
Server* const server_;
@@ -420,14 +425,19 @@ class Server::AsyncRequest GRPC_FINAL : public CompletionQueueTag {
void Server::RequestAsyncCall(void* registered_method, ServerContext* context,
grpc::protobuf::Message* request,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) {
- new AsyncRequest(this, registered_method, context, request, stream, cq, tag);
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag) {
+ new AsyncRequest(this, registered_method, context, request, stream, call_cq,
+ notification_cq, tag);
}
void Server::RequestAsyncGenericCall(GenericServerContext* context,
ServerAsyncStreamingInterface* stream,
- CompletionQueue* cq, void* tag) {
- new AsyncRequest(this, context, stream, cq, tag);
+ CompletionQueue* call_cq,
+ ServerCompletionQueue* notification_cq,
+ void* tag) {
+ new AsyncRequest(this, context, stream, call_cq, notification_cq, tag);
}
void Server::ScheduleCallback() {
@@ -449,7 +459,7 @@ void Server::RunRpc() {
{
grpc::unique_lock<grpc::mutex> lock(mu_);
if (!shutdown_) {
- mrd->Request(server_);
+ mrd->Request(server_, cq_.cq());
}
}
cd.Run();
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index e48d1eeb42695da02cc8bf995842eadc26356c7c..4bcbd82952193c77d9c00efbea0609b9b1dd181f 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -44,6 +44,12 @@ namespace grpc {
ServerBuilder::ServerBuilder()
: max_message_size_(-1), generic_service_(nullptr), thread_pool_(nullptr) {}
+std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue() {
+ ServerCompletionQueue* cq = new ServerCompletionQueue();
+ cqs_.push_back(cq);
+ return std::unique_ptr<ServerCompletionQueue>(cq);
+}
+
void ServerBuilder::RegisterService(SynchronousService* service) {
services_.push_back(service->service());
}
@@ -88,6 +94,9 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
}
std::unique_ptr<Server> server(
new Server(thread_pool_, thread_pool_owned, max_message_size_));
+ for (auto cq = cqs_.begin(); cq != cqs_.end(); ++cq) {
+ grpc_server_register_completion_queue(server->server_, (*cq)->cq());
+ }
for (auto service = services_.begin(); service != services_.end();
service++) {
if (!server->RegisterService(*service)) {
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index a8cc1b29a4586925246d47fc6b50c7e46c829353..6cefacece24c0cb249e56d89b46ae2f6f598b8cb 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -673,7 +673,9 @@ grpcsharp_call_start_serverside(grpc_call *call, callback_funcptr callback) {
GPR_EXPORT grpc_server *GPR_CALLTYPE
grpcsharp_server_create(grpc_completion_queue *cq,
const grpc_channel_args *args) {
- return grpc_server_create(cq, args);
+ grpc_server *server = grpc_server_create(args);
+ grpc_server_register_completion_queue(server, cq);
+ return server;
}
GPR_EXPORT gpr_int32 GPR_CALLTYPE
@@ -706,7 +708,7 @@ grpcsharp_server_request_call(grpc_server *server, grpc_completion_queue *cq,
return grpc_server_request_call(
server, &(ctx->server_rpc_new.call), &(ctx->server_rpc_new.call_details),
- &(ctx->server_rpc_new.request_metadata), cq, ctx);
+ &(ctx->server_rpc_new.request_metadata), cq, cq, ctx);
}
/* Security */
diff --git a/src/node/ext/server.cc b/src/node/ext/server.cc
index 3c2396b81012788a6dc954f43276018bd389519c..eb97f7348b4b5a6fb37bd11d22cbf98f306c72fb 100644
--- a/src/node/ext/server.cc
+++ b/src/node/ext/server.cc
@@ -161,7 +161,7 @@ NAN_METHOD(Server::New) {
grpc_server *wrapped_server;
grpc_completion_queue *queue = CompletionQueueAsyncWorker::GetQueue();
if (args[0]->IsUndefined()) {
- wrapped_server = grpc_server_create(queue, NULL);
+ wrapped_server = grpc_server_create(NULL);
} else if (args[0]->IsObject()) {
Handle<Object> args_hash(args[0]->ToObject());
Handle<Array> keys(args_hash->GetOwnPropertyNames());
@@ -190,11 +190,12 @@ NAN_METHOD(Server::New) {
return NanThrowTypeError("Arg values must be strings");
}
}
- wrapped_server = grpc_server_create(queue, &channel_args);
+ wrapped_server = grpc_server_create(&channel_args);
free(channel_args.args);
} else {
return NanThrowTypeError("Server expects an object");
}
+ grpc_server_register_completion_queue(wrapped_server, queue);
Server *server = new Server(wrapped_server);
server->Wrap(args.This());
NanReturnValue(args.This());
@@ -212,6 +213,7 @@ NAN_METHOD(Server::RequestCall) {
grpc_call_error error = grpc_server_request_call(
server->wrapped_server, &op->call, &op->details, &op->request_metadata,
CompletionQueueAsyncWorker::GetQueue(),
+ CompletionQueueAsyncWorker::GetQueue(),
new struct tag(new NanCallback(args[0].As<Function>()), ops.release(),
shared_ptr<Resources>(nullptr)));
if (error != GRPC_CALL_OK) {
diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c
index dbb9425619f43b99f5528aa5a1b85f4481547dca..18d34ab14b54b4b25db837b1fa6720d4d531265b 100644
--- a/src/php/ext/grpc/server.c
+++ b/src/php/ext/grpc/server.c
@@ -119,12 +119,13 @@ PHP_METHOD(Server, __construct) {
}
server->queue = grpc_completion_queue_create();
if (args_array == NULL) {
- server->wrapped = grpc_server_create(server->queue, NULL);
+ server->wrapped = grpc_server_create(NULL);
} else {
php_grpc_read_args_array(args_array, &args);
- server->wrapped = grpc_server_create(server->queue, &args);
+ server->wrapped = grpc_server_create(&args);
efree(args.args);
}
+ grpc_server_register_completion_queue(server->wrapped, server->queue);
}
/**
@@ -146,8 +147,9 @@ PHP_METHOD(Server, requestCall) {
object_init(result);
grpc_call_details_init(&details);
grpc_metadata_array_init(&metadata);
- error_code = grpc_server_request_call(server->wrapped, &call, &details,
- &metadata, server->queue, NULL);
+ error_code =
+ grpc_server_request_call(server->wrapped, &call, &details, &metadata,
+ server->queue, server->queue, NULL);
if (error_code != GRPC_CALL_OK) {
zend_throw_exception(spl_ce_LogicException, "request_call failed",
(long)error_code TSRMLS_CC);
diff --git a/src/python/src/grpc/_adapter/_server.c b/src/python/src/grpc/_adapter/_server.c
index e7c5917724a3c0ee7c7ada52c210916bb67ae885..a6c20bf13237cb56c43cbab643b41e59c2b65ec4 100644
--- a/src/python/src/grpc/_adapter/_server.c
+++ b/src/python/src/grpc/_adapter/_server.c
@@ -51,8 +51,9 @@ static int pygrpc_server_init(Server *self, PyObject *args, PyObject *kwds) {
&completion_queue)) {
return -1;
}
- self->c_server = grpc_server_create(
- completion_queue->c_completion_queue, NULL);
+ self->c_server = grpc_server_create(NULL);
+ grpc_server_register_completion_queue(self->c_server,
+ completion_queue->c_completion_queue);
self->completion_queue = completion_queue;
Py_INCREF(completion_queue);
return 0;
@@ -122,7 +123,7 @@ static const PyObject *pygrpc_server_service(Server *self, PyObject *tag) {
call_error = grpc_server_request_call(
self->c_server, &c_tag->call->c_call, &c_tag->call->call_details,
&c_tag->call->recv_metadata, self->completion_queue->c_completion_queue,
- c_tag);
+ self->completion_queue->c_completion_queue, c_tag);
result = pygrpc_translate_call_error(call_error);
if (result != NULL) {
diff --git a/src/ruby/ext/grpc/rb_server.c b/src/ruby/ext/grpc/rb_server.c
index bc0878af05cb83f4f0f7cc9f1657aa1c804b5240..85ada1958bb458b350a4ccec328bd5cb1fbd12c0 100644
--- a/src/ruby/ext/grpc/rb_server.c
+++ b/src/ruby/ext/grpc/rb_server.c
@@ -123,7 +123,7 @@ static VALUE grpc_rb_server_init(VALUE self, VALUE cqueue, VALUE channel_args) {
TypedData_Get_Struct(self, grpc_rb_server, &grpc_rb_server_data_type,
wrapper);
grpc_rb_hash_convert_to_channel_args(channel_args, &args);
- srv = grpc_server_create(cq, &args);
+ srv = grpc_server_create(&args);
if (args.args != NULL) {
xfree(args.args); /* Allocated by grpc_rb_hash_convert_to_channel_args */
@@ -131,6 +131,7 @@ static VALUE grpc_rb_server_init(VALUE self, VALUE cqueue, VALUE channel_args) {
if (srv == NULL) {
rb_raise(rb_eRuntimeError, "could not create a gRPC server, not sure why");
}
+ grpc_server_register_completion_queue(srv, cq);
wrapper->wrapped = srv;
/* Add the cq as the server's mark object. This ensures the ruby cq can't be
@@ -218,6 +219,7 @@ static VALUE grpc_rb_server_request_call(VALUE self, VALUE cqueue,
err = grpc_server_request_call(
s->wrapped, &call, &st.details, &st.md_ary,
grpc_rb_get_wrapped_completion_queue(cqueue),
+ grpc_rb_get_wrapped_completion_queue(cqueue),
ROBJECT(tag_new));
if (err != GRPC_CALL_OK) {
grpc_request_call_stack_cleanup(&st);
diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c
index 5e278ca66c93bf7bfefe47c959ab48e727d68dd6..ad970845024ac48e93346e63111ede7e1b4c6aba 100644
--- a/test/core/end2end/dualstack_socket_test.c
+++ b/test/core/end2end/dualstack_socket_test.c
@@ -99,7 +99,8 @@ void test_connect(const char *server_host, const char *client_host, int port,
/* Create server. */
server_cq = grpc_completion_queue_create();
- server = grpc_server_create(server_cq, NULL);
+ server = grpc_server_create(NULL);
+ grpc_server_register_completion_queue(server, server_cq);
GPR_ASSERT((got_port = grpc_server_add_http2_port(server, server_hostport)) >
0);
if (port == 0) {
@@ -155,10 +156,10 @@ void test_connect(const char *server_host, const char *client_host, int port,
if (expect_ok) {
/* Check for a successful request. */
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(server, &s,
- &call_details,
- &request_metadata_recv,
- server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(server, &s, &call_details,
+ &request_metadata_recv, server_cq,
+ server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/fixtures/chttp2_fake_security.c b/test/core/end2end/fixtures/chttp2_fake_security.c
index c1ac9163edc6a51618dd4891b66566d1b1f9a90f..5323e29e82532e630abe10765f82de3b26b64b90 100644
--- a/test/core/end2end/fixtures/chttp2_fake_security.c
+++ b/test/core/end2end/fixtures/chttp2_fake_security.c
@@ -82,8 +82,8 @@ static void chttp2_init_server_secure_fullstack(
if (f->server) {
grpc_server_destroy(f->server);
}
- f->server =
- grpc_server_create(f->server_cq, server_args);
+ f->server = grpc_server_create(server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds));
grpc_server_credentials_release(server_creds);
grpc_server_start(f->server);
diff --git a/test/core/end2end/fixtures/chttp2_fullstack.c b/test/core/end2end/fixtures/chttp2_fullstack.c
index d7de5e543481688e1cb3d929ff0289dd9186288a..f92b40efebf6e86745fdc5e838f41558a36025cd 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack.c
@@ -83,7 +83,8 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f,
if (f->server) {
grpc_server_destroy(f->server);
}
- f->server = grpc_server_create(f->server_cq, server_args);
+ f->server = grpc_server_create(server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
index 53803b0f1d8a38d5acca205c253b7de4f67fc01b..876782df846e06e48a530c2ba41104c9496c8887 100644
--- a/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
+++ b/test/core/end2end/fixtures/chttp2_fullstack_uds_posix.c
@@ -88,7 +88,8 @@ void chttp2_init_server_fullstack(grpc_end2end_test_fixture *f,
if (f->server) {
grpc_server_destroy(f->server);
}
- f->server = grpc_server_create(f->server_cq, server_args);
+ f->server = grpc_server_create(server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
GPR_ASSERT(grpc_server_add_http2_port(f->server, ffd->localaddr));
grpc_server_start(f->server);
}
diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c
index 3d6c0cf3f09ec6f62f0e4bdcd37e69afeef0d156..6d1b7b5ff050a360d503c12cbb3cafcea1599ec0 100644
--- a/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_simple_ssl_fullstack.c
@@ -85,8 +85,8 @@ static void chttp2_init_server_secure_fullstack(
if (f->server) {
grpc_server_destroy(f->server);
}
- f->server =
- grpc_server_create(f->server_cq, server_args);
+ f->server = grpc_server_create(server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds));
grpc_server_credentials_release(server_creds);
grpc_server_start(f->server);
diff --git a/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c b/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c
index b57872f4f0069e642912042492aa0e465eacab58..4a15d502a51efe054ba8343fd0ebae6e481ef737 100644
--- a/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c
+++ b/test/core/end2end/fixtures/chttp2_simple_ssl_with_oauth2_fullstack.c
@@ -83,8 +83,8 @@ static void chttp2_init_server_secure_fullstack(
if (f->server) {
grpc_server_destroy(f->server);
}
- f->server =
- grpc_server_create(f->server_cq, server_args);
+ f->server = grpc_server_create(server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
GPR_ASSERT(grpc_server_add_secure_http2_port(f->server, ffd->localaddr, server_creds));
grpc_server_credentials_release(server_creds);
grpc_server_start(f->server);
diff --git a/test/core/end2end/fixtures/chttp2_socket_pair.c b/test/core/end2end/fixtures/chttp2_socket_pair.c
index d19ceb178bc42cc58b872a6963dfcc2a2e910fdb..43ebf7eed587863304aa0bfc5b7ab80d420d5f2e 100644
--- a/test/core/end2end/fixtures/chttp2_socket_pair.c
+++ b/test/core/end2end/fixtures/chttp2_socket_pair.c
@@ -117,8 +117,8 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
grpc_channel_args *server_args) {
grpc_endpoint_pair *sfd = f->fixture_data;
GPR_ASSERT(!f->server);
- f->server =
- grpc_server_create_from_filters(f->server_cq, NULL, 0, server_args);
+ f->server = grpc_server_create_from_filters(NULL, 0, server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
grpc_server_start(f->server);
grpc_create_chttp2_transport(server_setup_transport, f, server_args,
sfd->server, NULL, 0, grpc_mdctx_create(), 0);
diff --git a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
index ddde585b8391cc70bbbfd18380ac314ca93fb642..385d5a4e811d6515ada7b2a939780c5b489372a1 100644
--- a/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
+++ b/test/core/end2end/fixtures/chttp2_socket_pair_one_byte_at_a_time.c
@@ -117,8 +117,8 @@ static void chttp2_init_server_socketpair(grpc_end2end_test_fixture *f,
grpc_channel_args *server_args) {
grpc_endpoint_pair *sfd = f->fixture_data;
GPR_ASSERT(!f->server);
- f->server =
- grpc_server_create_from_filters(f->server_cq, NULL, 0, server_args);
+ f->server = grpc_server_create_from_filters(NULL, 0, server_args);
+ grpc_server_register_completion_queue(f->server, f->server_cq);
grpc_server_start(f->server);
grpc_create_chttp2_transport(server_setup_transport, f, server_args,
sfd->server, NULL, 0, grpc_mdctx_create(), 0);
diff --git a/test/core/end2end/tests/cancel_after_accept.c b/test/core/end2end/tests/cancel_after_accept.c
index 184434cc97f60115f0fe7354e0868a06e085e40f..9222310aa3ef3f762c1e016cad66593b34d1ff71 100644
--- a/test/core/end2end/tests/cancel_after_accept.c
+++ b/test/core/end2end/tests/cancel_after_accept.c
@@ -160,9 +160,10 @@ static void test_cancel_after_accept(grpc_end2end_test_config config,
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(
- f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq, tag(2)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(2)));
cq_expect_completion(v_server, tag(2), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
index f84c3343be6dbf8b66a1a9fd7297a15159c138fb..eb236e19c401242d456318db70c3ba2412091b6f 100644
--- a/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
+++ b/test/core/end2end/tests/cancel_after_accept_and_writes_closed.c
@@ -162,9 +162,10 @@ static void test_cancel_after_accept_and_writes_closed(
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(
- f.server, &s, &call_details,
- &request_metadata_recv, f.server_cq, tag(2)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(2)));
cq_expect_completion(v_server, tag(2), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/census_simple_request.c b/test/core/end2end/tests/census_simple_request.c
index 45406975db629b49606ca11285384ce5e362e855..f3f051018ef37b6c8153b3798e9090db79d2e760 100644
--- a/test/core/end2end/tests/census_simple_request.c
+++ b/test/core/end2end/tests/census_simple_request.c
@@ -141,10 +141,10 @@ static void test_body(grpc_end2end_test_fixture f) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/disappearing_server.c b/test/core/end2end/tests/disappearing_server.c
index 3e3ea3052ba039cc07d60920413f7f9f57428f1d..f280d971f71143ce898d5b57b48c1edfdfd394de 100644
--- a/test/core/end2end/tests/disappearing_server.c
+++ b/test/core/end2end/tests/disappearing_server.c
@@ -132,10 +132,10 @@ static void do_request_and_shutdown_server(grpc_end2end_test_fixture *f,
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f->server, &s,
- &call_details,
- &request_metadata_recv,
- f->server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f->server, &s, &call_details,
+ &request_metadata_recv, f->server_cq,
+ f->server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
index 28ace256ef6350aba2e30c8548b054c7bad530f8..2dc3f2330fc486072b4e70e375e7fd2b7b565d04 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_inflight_calls.c
@@ -147,10 +147,10 @@ static void test_early_server_shutdown_finishes_inflight_calls(
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
index c909e94ac2334854520105275f756f9cb8c71f30..7a6e274264682ba69d72907e2964a02829f27d3b 100644
--- a/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
+++ b/test/core/end2end/tests/early_server_shutdown_finishes_tags.c
@@ -114,10 +114,10 @@ static void test_early_server_shutdown_finishes_tags(
/* upon shutdown, the server should finish all requested calls indicating
no new call */
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
grpc_server_shutdown(f.server);
cq_expect_completion(v_server, tag(101), GRPC_OP_ERROR);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/graceful_server_shutdown.c b/test/core/end2end/tests/graceful_server_shutdown.c
index 17057c4ca731c0d8d04b7a7c640ac46a380f2d96..309276aedd6846e92ca70d1735ec37596daaf1ed 100644
--- a/test/core/end2end/tests/graceful_server_shutdown.c
+++ b/test/core/end2end/tests/graceful_server_shutdown.c
@@ -146,10 +146,10 @@ static void test_early_server_shutdown_finishes_inflight_calls(
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/invoke_large_request.c b/test/core/end2end/tests/invoke_large_request.c
index 5aa378dd38c3d24d36f859e63ab2cc614a71254d..2e54f407bb6d1d4e8982768d562c78ad991f8912 100644
--- a/test/core/end2end/tests/invoke_large_request.c
+++ b/test/core/end2end/tests/invoke_large_request.c
@@ -164,10 +164,10 @@ static void test_invoke_large_request(grpc_end2end_test_config config) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/max_concurrent_streams.c b/test/core/end2end/tests/max_concurrent_streams.c
index 4640f3287cd530b35da555de184ea12c17b8c78e..1f074c703abf196cb1ed25dbff568d4dabfc46e7 100644
--- a/test/core/end2end/tests/max_concurrent_streams.c
+++ b/test/core/end2end/tests/max_concurrent_streams.c
@@ -144,10 +144,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
@@ -253,10 +253,10 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) {
"foo.test.google.fr:1234", deadline);
GPR_ASSERT(c2);
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s1,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s1, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
op = ops;
op->op = GRPC_OP_SEND_INITIAL_METADATA;
@@ -341,10 +341,10 @@ static void test_max_concurrent_streams(grpc_end2end_test_config config) {
cq_expect_completion(v_client, tag(live_call + 1), GRPC_OP_OK);
cq_verify(v_client);
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s2,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(201)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s2, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(201)));
cq_expect_completion(v_server, tag(201), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/max_message_length.c b/test/core/end2end/tests/max_message_length.c
index 40fcbfba8b3364ce5dfc2f17627f0479f6d68a9d..cd86b08349592f2a82a79223ef800072430bfaec 100644
--- a/test/core/end2end/tests/max_message_length.c
+++ b/test/core/end2end/tests/max_message_length.c
@@ -163,10 +163,10 @@ static void test_max_message_length(grpc_end2end_test_config config) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/ping_pong_streaming.c b/test/core/end2end/tests/ping_pong_streaming.c
index caea950172a02c0e12e5d8bab758c911b13de4ba..473b4f94ac098feae20781ae27e276334eae5f24 100644
--- a/test/core/end2end/tests/ping_pong_streaming.c
+++ b/test/core/end2end/tests/ping_pong_streaming.c
@@ -152,10 +152,10 @@ static void test_pingpong_streaming(grpc_end2end_test_config config,
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(100)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(100)));
cq_expect_completion(v_server, tag(100), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/registered_call.c b/test/core/end2end/tests/registered_call.c
index 0fe21a86c9e7c3ab60f69c31860af73898dc3de6..dc9c407deb486152742be08e23aa3e631cfa23c8 100644
--- a/test/core/end2end/tests/registered_call.c
+++ b/test/core/end2end/tests/registered_call.c
@@ -145,10 +145,10 @@ static void simple_request_body(grpc_end2end_test_fixture f, void *rc) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
index 6edc641a54aff6068e392fbdcc4cb0db95b5bb1d..91d580341e40028ba909200df98b466afa5bd100 100644
--- a/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_binary_metadata_and_payload.c
@@ -180,10 +180,10 @@ static void test_request_response_with_metadata_and_payload(
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
index 9f3c1242a376f1b97a2378ea33dc8aee60ffb307..5f44704fa8750b0cce914a3051012ea8d65a9465 100644
--- a/test/core/end2end/tests/request_response_with_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_metadata_and_payload.c
@@ -166,10 +166,10 @@ static void test_request_response_with_metadata_and_payload(
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_payload.c b/test/core/end2end/tests/request_response_with_payload.c
index e40d6c723b6d8a0dee0f8ef2dead04764269283b..a483ef535aa4a3a33a48d2d94877600375e25ee4 100644
--- a/test/core/end2end/tests/request_response_with_payload.c
+++ b/test/core/end2end/tests/request_response_with_payload.c
@@ -158,10 +158,10 @@ static void request_response_with_payload(grpc_end2end_test_fixture f) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
index 0339d5ab99732bf721939885f6c2f341d01c0f0d..fce53950db51e67d1f0bfc19447dc77154210455 100644
--- a/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
+++ b/test/core/end2end/tests/request_response_with_payload_and_call_creds.c
@@ -216,7 +216,8 @@ static void request_response_with_payload_and_call_creds(
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
index 04306df074ff1ddd6772c1216a381be2a9d81591..9c0e1d214b84721506e7915884def088dff271df 100644
--- a/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
+++ b/test/core/end2end/tests/request_response_with_trailing_metadata_and_payload.c
@@ -168,7 +168,8 @@ static void test_request_response_with_metadata_and_payload(
GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
&call_details,
&request_metadata_recv,
- f.server_cq, tag(101)));
+ f.server_cq, f.server_cq,
+ tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_with_large_metadata.c b/test/core/end2end/tests/request_with_large_metadata.c
index 506766a73fef0ac29f079d2816daab60cc01448e..18a6da853cf8f128acec61beec8fd0734f0f9032 100644
--- a/test/core/end2end/tests/request_with_large_metadata.c
+++ b/test/core/end2end/tests/request_with_large_metadata.c
@@ -162,10 +162,10 @@ static void test_request_with_large_metadata(grpc_end2end_test_config config) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/request_with_payload.c b/test/core/end2end/tests/request_with_payload.c
index fea71b41931eb7ada90ef29f24ec926315376478..0533b71da9278fb73a1937b12ac1d8a878986aaa 100644
--- a/test/core/end2end/tests/request_with_payload.c
+++ b/test/core/end2end/tests/request_with_payload.c
@@ -153,10 +153,10 @@ static void test_invoke_request_with_payload(grpc_end2end_test_config config) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/simple_delayed_request.c b/test/core/end2end/tests/simple_delayed_request.c
index 231e2602275134a646f43c796554ab55b6e5d660..184cda3f959ec16ec7fcfc2489f359423479f6c4 100644
--- a/test/core/end2end/tests/simple_delayed_request.c
+++ b/test/core/end2end/tests/simple_delayed_request.c
@@ -140,10 +140,10 @@ static void simple_delayed_request_body(grpc_end2end_test_config config,
config.init_server(f, server_args);
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f->server, &s,
- &call_details,
- &request_metadata_recv,
- f->server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f->server, &s, &call_details,
+ &request_metadata_recv, f->server_cq,
+ f->server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/simple_request.c b/test/core/end2end/tests/simple_request.c
index e6fe2b3435f65378f72e7b6981c1a911e5dbe7dd..f1b670be5a166f2057fadb67675abbda3a2be226 100644
--- a/test/core/end2end/tests/simple_request.c
+++ b/test/core/end2end/tests/simple_request.c
@@ -146,10 +146,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
index 0105d00c5d3c95f7faf1493796e72133605c4a88..5ef3133ebebfed640684a6abc48740d95d40abd7 100644
--- a/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
+++ b/test/core/end2end/tests/simple_request_with_high_initial_sequence_number.c
@@ -146,10 +146,10 @@ static void simple_request_body(grpc_end2end_test_fixture f) {
op++;
GPR_ASSERT(GRPC_CALL_OK == grpc_call_start_batch(c, ops, op - ops, tag(1)));
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(f.server, &s,
- &call_details,
- &request_metadata_recv,
- f.server_cq, tag(101)));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(f.server, &s, &call_details,
+ &request_metadata_recv, f.server_cq,
+ f.server_cq, tag(101)));
cq_expect_completion(v_server, tag(101), GRPC_OP_OK);
cq_verify(v_server);
diff --git a/test/core/fling/server.c b/test/core/fling/server.c
index a35afa107714e6933ec9bc9ee94141478ad1d860..1f1f857bd9f323495bfe7807d42070cbeff145c9 100644
--- a/test/core/fling/server.c
+++ b/test/core/fling/server.c
@@ -92,7 +92,7 @@ typedef struct {
static void request_call(void) {
grpc_metadata_array_init(&request_metadata_recv);
grpc_server_request_call(server, &call, &call_details, &request_metadata_recv,
- cq, tag(FLING_SERVER_NEW_REQUEST));
+ cq, cq, tag(FLING_SERVER_NEW_REQUEST));
}
static void handle_unary_method(void) {
@@ -211,13 +211,14 @@ int main(int argc, char **argv) {
test_server1_cert};
grpc_server_credentials *ssl_creds =
grpc_ssl_server_credentials_create(NULL, &pem_key_cert_pair, 1);
- server = grpc_server_create(cq, NULL);
+ server = grpc_server_create(NULL);
GPR_ASSERT(grpc_server_add_secure_http2_port(server, addr, ssl_creds));
grpc_server_credentials_release(ssl_creds);
} else {
- server = grpc_server_create(cq, NULL);
+ server = grpc_server_create(NULL);
GPR_ASSERT(grpc_server_add_http2_port(server, addr));
}
+ grpc_server_register_completion_queue(server, cq);
grpc_server_start(server);
gpr_free(addr_buf);
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index bf5540f70670042919a97de3bf6f062228b39cb6..24595a820f62582681299fe768977623863ee668 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -91,7 +91,7 @@ void verify_timed_ok(
class AsyncEnd2endTest : public ::testing::Test {
protected:
- AsyncEnd2endTest() : service_(&srv_cq_) {}
+ AsyncEnd2endTest() {}
void SetUp() GRPC_OVERRIDE {
int port = grpc_pick_unused_port_or_die();
@@ -100,6 +100,7 @@ class AsyncEnd2endTest : public ::testing::Test {
ServerBuilder builder;
builder.AddListeningPort(server_address_.str(), grpc::InsecureServerCredentials());
builder.RegisterAsyncService(&service_);
+ srv_cq_ = builder.AddCompletionQueue();
server_ = builder.BuildAndStart();
}
@@ -108,10 +109,10 @@ class AsyncEnd2endTest : public ::testing::Test {
void* ignored_tag;
bool ignored_ok;
cli_cq_.Shutdown();
- srv_cq_.Shutdown();
+ srv_cq_->Shutdown();
while (cli_cq_.Next(&ignored_tag, &ignored_ok))
;
- while (srv_cq_.Next(&ignored_tag, &ignored_ok))
+ while (srv_cq_->Next(&ignored_tag, &ignored_ok))
;
}
@@ -121,9 +122,9 @@ class AsyncEnd2endTest : public ::testing::Test {
stub_ = std::move(grpc::cpp::test::util::TestService::NewStub(channel));
}
- void server_ok(int i) { verify_ok(&srv_cq_, i, true); }
+ void server_ok(int i) { verify_ok(srv_cq_.get(), i, true); }
void client_ok(int i) { verify_ok(&cli_cq_, i, true); }
- void server_fail(int i) { verify_ok(&srv_cq_, i, false); }
+ void server_fail(int i) { verify_ok(srv_cq_.get(), i, false); }
void client_fail(int i) { verify_ok(&cli_cq_, i, false); }
void SendRpc(int num_rpcs) {
@@ -142,8 +143,8 @@ class AsyncEnd2endTest : public ::testing::Test {
std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_));
- service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, &srv_cq_,
- tag(2));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer,
+ srv_cq_.get(), srv_cq_.get(), tag(2));
server_ok(2);
EXPECT_EQ(send_request.message(), recv_request.message());
@@ -161,7 +162,7 @@ class AsyncEnd2endTest : public ::testing::Test {
}
CompletionQueue cli_cq_;
- CompletionQueue srv_cq_;
+ std::unique_ptr<ServerCompletionQueue> srv_cq_;
std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_;
std::unique_ptr<Server> server_;
grpc::cpp::test::util::TestService::AsyncService service_;
@@ -200,18 +201,18 @@ TEST_F(AsyncEnd2endTest, AsyncNextRpc) {
std::chrono::system_clock::now());
std::chrono::system_clock::time_point time_limit(
std::chrono::system_clock::now() + std::chrono::seconds(10));
- verify_timed_ok(&srv_cq_, -1, true, time_now, CompletionQueue::TIMEOUT);
+ verify_timed_ok(srv_cq_.get(), -1, true, time_now, CompletionQueue::TIMEOUT);
verify_timed_ok(&cli_cq_, -1, true, time_now, CompletionQueue::TIMEOUT);
- service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, &srv_cq_,
- tag(2));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
- verify_timed_ok(&srv_cq_, 2, true, time_limit);
+ verify_timed_ok(srv_cq_.get(), 2, true, time_limit);
EXPECT_EQ(send_request.message(), recv_request.message());
send_response.set_message(recv_request.message());
response_writer.Finish(send_response, Status::OK, tag(3));
- verify_timed_ok(&srv_cq_, 3, true);
+ verify_timed_ok(srv_cq_.get(), 3, true);
response_reader->Finish(&recv_response, &recv_status, tag(4));
verify_timed_ok(&cli_cq_, 4, true);
@@ -237,7 +238,8 @@ TEST_F(AsyncEnd2endTest, SimpleClientStreaming) {
std::unique_ptr<ClientAsyncWriter<EchoRequest> > cli_stream(
stub_->AsyncRequestStream(&cli_ctx, &recv_response, &cli_cq_, tag(1)));
- service_.RequestRequestStream(&srv_ctx, &srv_stream, &srv_cq_, tag(2));
+ service_.RequestRequestStream(&srv_ctx, &srv_stream, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
server_ok(2);
client_ok(1);
@@ -290,8 +292,8 @@ TEST_F(AsyncEnd2endTest, SimpleServerStreaming) {
std::unique_ptr<ClientAsyncReader<EchoResponse> > cli_stream(
stub_->AsyncResponseStream(&cli_ctx, send_request, &cli_cq_, tag(1)));
- service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream, &srv_cq_,
- tag(2));
+ service_.RequestResponseStream(&srv_ctx, &recv_request, &srv_stream,
+ srv_cq_.get(), srv_cq_.get(), tag(2));
server_ok(2);
client_ok(1);
@@ -341,7 +343,8 @@ TEST_F(AsyncEnd2endTest, SimpleBidiStreaming) {
std::unique_ptr<ClientAsyncReaderWriter<EchoRequest, EchoResponse> >
cli_stream(stub_->AsyncBidiStream(&cli_ctx, &cli_cq_, tag(1)));
- service_.RequestBidiStream(&srv_ctx, &srv_stream, &srv_cq_, tag(2));
+ service_.RequestBidiStream(&srv_ctx, &srv_stream, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
server_ok(2);
client_ok(1);
@@ -399,8 +402,8 @@ TEST_F(AsyncEnd2endTest, ClientInitialMetadataRpc) {
std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_));
- service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, &srv_cq_,
- tag(2));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
server_ok(2);
EXPECT_EQ(send_request.message(), recv_request.message());
auto client_initial_metadata = srv_ctx.client_metadata();
@@ -440,8 +443,8 @@ TEST_F(AsyncEnd2endTest, ServerInitialMetadataRpc) {
std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_));
- service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, &srv_cq_,
- tag(2));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
server_ok(2);
EXPECT_EQ(send_request.message(), recv_request.message());
srv_ctx.AddInitialMetadata(meta1.first, meta1.second);
@@ -487,8 +490,8 @@ TEST_F(AsyncEnd2endTest, ServerTrailingMetadataRpc) {
std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_));
- service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, &srv_cq_,
- tag(2));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
server_ok(2);
EXPECT_EQ(send_request.message(), recv_request.message());
response_writer.SendInitialMetadata(tag(3));
@@ -547,8 +550,8 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) {
std::unique_ptr<ClientAsyncResponseReader<EchoResponse> > response_reader(
stub_->AsyncEcho(&cli_ctx, send_request, &cli_cq_));
- service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, &srv_cq_,
- tag(2));
+ service_.RequestEcho(&srv_ctx, &recv_request, &response_writer, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
server_ok(2);
EXPECT_EQ(send_request.message(), recv_request.message());
auto client_initial_metadata = srv_ctx.client_metadata();
diff --git a/test/cpp/end2end/generic_end2end_test.cc b/test/cpp/end2end/generic_end2end_test.cc
index 103f613f70efa23b1b7edab8fe733a812cc9d071..80e43fd8544688ec1e74f1e3b78c18c3bea09dda 100644
--- a/test/cpp/end2end/generic_end2end_test.cc
+++ b/test/cpp/end2end/generic_end2end_test.cc
@@ -109,6 +109,7 @@ class GenericEnd2endTest : public ::testing::Test {
ServerBuilder builder;
builder.AddListeningPort(server_address_.str(), InsecureServerCredentials());
builder.RegisterAsyncGenericService(&generic_service_);
+ srv_cq_ = builder.AddCompletionQueue();
server_ = builder.BuildAndStart();
}
@@ -117,10 +118,10 @@ class GenericEnd2endTest : public ::testing::Test {
void* ignored_tag;
bool ignored_ok;
cli_cq_.Shutdown();
- srv_cq_.Shutdown();
+ srv_cq_->Shutdown();
while (cli_cq_.Next(&ignored_tag, &ignored_ok))
;
- while (srv_cq_.Next(&ignored_tag, &ignored_ok))
+ while (srv_cq_->Next(&ignored_tag, &ignored_ok))
;
}
@@ -130,9 +131,9 @@ class GenericEnd2endTest : public ::testing::Test {
generic_stub_.reset(new GenericStub(channel));
}
- void server_ok(int i) { verify_ok(&srv_cq_, i, true); }
+ void server_ok(int i) { verify_ok(srv_cq_.get(), i, true); }
void client_ok(int i) { verify_ok(&cli_cq_, i, true); }
- void server_fail(int i) { verify_ok(&srv_cq_, i, false); }
+ void server_fail(int i) { verify_ok(srv_cq_.get(), i, false); }
void client_fail(int i) { verify_ok(&cli_cq_, i, false); }
void SendRpc(int num_rpcs) {
@@ -160,9 +161,10 @@ class GenericEnd2endTest : public ::testing::Test {
call->WritesDone(tag(3));
client_ok(3);
- generic_service_.RequestCall(&srv_ctx, &stream, &srv_cq_, tag(4));
+ generic_service_.RequestCall(&srv_ctx, &stream, srv_cq_.get(),
+ srv_cq_.get(), tag(4));
- verify_ok(generic_service_.completion_queue(), 4, true);
+ verify_ok(srv_cq_.get(), 4, true);
EXPECT_EQ(server_address_.str(), srv_ctx.host());
EXPECT_EQ(kMethodName, srv_ctx.method());
ByteBuffer recv_buffer;
@@ -193,7 +195,7 @@ class GenericEnd2endTest : public ::testing::Test {
}
CompletionQueue cli_cq_;
- CompletionQueue srv_cq_;
+ std::unique_ptr<ServerCompletionQueue> srv_cq_;
std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub_;
std::unique_ptr<grpc::GenericStub> generic_stub_;
std::unique_ptr<Server> server_;
@@ -230,9 +232,10 @@ TEST_F(GenericEnd2endTest, SimpleBidiStreaming) {
generic_stub_->Call(&cli_ctx, kMethodName, &cli_cq_, tag(1));
client_ok(1);
- generic_service_.RequestCall(&srv_ctx, &srv_stream, &srv_cq_, tag(2));
+ generic_service_.RequestCall(&srv_ctx, &srv_stream, srv_cq_.get(),
+ srv_cq_.get(), tag(2));
- verify_ok(generic_service_.completion_queue(), 2, true);
+ verify_ok(srv_cq_.get(), 2, true);
EXPECT_EQ(server_address_.str(), srv_ctx.host());
EXPECT_EQ(kMethodName, srv_ctx.method());
diff --git a/test/cpp/qps/server_async.cc b/test/cpp/qps/server_async.cc
index b19c443c82333647f0602e70231821c5ffd1c819..6cb3192908df7c2ddd38be8a6524a3fb9e4b888b 100644
--- a/test/cpp/qps/server_async.cc
+++ b/test/cpp/qps/server_async.cc
@@ -63,9 +63,7 @@ namespace testing {
class AsyncQpsServerTest : public Server {
public:
- AsyncQpsServerTest(const ServerConfig& config, int port)
- : srv_cq_(), async_service_(&srv_cq_), server_(nullptr),
- shutdown_(false) {
+ AsyncQpsServerTest(const ServerConfig &config, int port) : shutdown_(false) {
char* server_address = NULL;
gpr_join_host_port(&server_address, "::", port);
@@ -74,15 +72,17 @@ class AsyncQpsServerTest : public Server {
gpr_free(server_address);
builder.RegisterAsyncService(&async_service_);
+ srv_cq_ = builder.AddCompletionQueue();
server_ = builder.BuildAndStart();
using namespace std::placeholders;
- request_unary_ = std::bind(&TestService::AsyncService::RequestUnaryCall,
- &async_service_, _1, _2, _3, &srv_cq_, _4);
+ request_unary_ =
+ std::bind(&TestService::AsyncService::RequestUnaryCall, &async_service_,
+ _1, _2, _3, srv_cq_.get(), srv_cq_.get(), _4);
request_streaming_ =
- std::bind(&TestService::AsyncService::RequestStreamingCall,
- &async_service_, _1, _2, &srv_cq_, _3);
+ std::bind(&TestService::AsyncService::RequestStreamingCall,
+ &async_service_, _1, _2, srv_cq_.get(), srv_cq_.get(), _3);
for (int i = 0; i < 100; i++) {
contexts_.push_front(
new ServerRpcContextUnaryImpl<SimpleRequest, SimpleResponse>(
@@ -96,7 +96,7 @@ class AsyncQpsServerTest : public Server {
// Wait until work is available or we are shutting down
bool ok;
void* got_tag;
- while (srv_cq_.Next(&got_tag, &ok)) {
+ while (srv_cq_->Next(&got_tag, &ok)) {
ServerRpcContext* ctx = detag(got_tag);
// The tag is a pointer to an RPC context to invoke
if (ctx->RunNextState(ok) == false) {
@@ -116,7 +116,7 @@ class AsyncQpsServerTest : public Server {
{
std::lock_guard<std::mutex> g(shutdown_mutex_);
shutdown_ = true;
- srv_cq_.Shutdown();
+ srv_cq_->Shutdown();
}
for (auto thr = threads_.begin(); thr != threads_.end(); thr++) {
thr->join();
@@ -290,10 +290,10 @@ class AsyncQpsServerTest : public Server {
}
return Status::OK;
}
- CompletionQueue srv_cq_;
- TestService::AsyncService async_service_;
std::vector<std::thread> threads_;
std::unique_ptr<grpc::Server> server_;
+ std::unique_ptr<grpc::ServerCompletionQueue> srv_cq_;
+ TestService::AsyncService async_service_;
std::function<void(ServerContext*, SimpleRequest*,
grpc::ServerAsyncResponseWriter<SimpleResponse>*, void*)>
request_unary_;