Merge github.com:grpc/grpc into hybrid
diff --git a/build.yaml b/build.yaml
index ed7f4cc..a8ba1c8 100644
--- a/build.yaml
+++ b/build.yaml
@@ -3118,7 +3118,7 @@
- linux
- posix
- name: alarm_cpp_test
- gtest: true
+ cpu_cost: 0.1
build: test
language: c++
src:
@@ -3131,7 +3131,7 @@
- gpr_test_util
- gpr
- name: async_end2end_test
- gtest: true
+ cpu_cost: 0.8
build: test
language: c++
src:
@@ -3144,7 +3144,6 @@
- gpr_test_util
- gpr
- name: auth_property_iterator_test
- gtest: true
build: test
language: c++
src:
@@ -3446,6 +3445,7 @@
- linux
- posix
- name: bm_pollset
+ cpu_cost: 0.5
build: test
language: c++
src:
@@ -3467,7 +3467,6 @@
- linux
- posix
- name: channel_arguments_test
- gtest: true
build: test
language: c++
src:
@@ -3477,7 +3476,6 @@
- grpc
- gpr
- name: channel_filter_test
- gtest: true
build: test
language: c++
src:
@@ -3487,7 +3485,6 @@
- grpc
- gpr
- name: cli_call_test
- gtest: true
build: test
language: c++
src:
@@ -3501,7 +3498,6 @@
- gpr_test_util
- gpr
- name: client_crash_test
- gtest: true
cpu_cost: 0.1
build: test
language: c++
@@ -3532,7 +3528,6 @@
- gpr_test_util
- gpr
- name: codegen_test_full
- gtest: true
build: test
language: c++
src:
@@ -3549,7 +3544,6 @@
filegroups:
- grpc++_codegen_base
- name: codegen_test_minimal
- gtest: true
build: test
language: c++
src:
@@ -3566,7 +3560,6 @@
- grpc++_codegen_base
- grpc++_codegen_base_src
- name: credentials_test
- gtest: true
build: test
language: c++
src:
@@ -3576,7 +3569,6 @@
- grpc
- gpr
- name: cxx_byte_buffer_test
- gtest: true
build: test
language: c++
src:
@@ -3588,7 +3580,6 @@
- gpr_test_util
- gpr
- name: cxx_slice_test
- gtest: true
build: test
language: c++
src:
@@ -3600,7 +3591,6 @@
- gpr_test_util
- gpr
- name: cxx_string_ref_test
- gtest: true
build: test
language: c++
src:
@@ -3608,7 +3598,6 @@
deps:
- grpc++
- name: cxx_time_test
- gtest: true
build: test
language: c++
src:
@@ -3620,8 +3609,7 @@
- gpr_test_util
- gpr
- name: end2end_test
- gtest: true
- cpu_cost: 0.5
+ cpu_cost: 1.0
build: test
language: c++
src:
@@ -3634,7 +3622,6 @@
- gpr_test_util
- gpr
- name: error_details_test
- gtest: true
build: test
language: c++
src:
@@ -3644,7 +3631,6 @@
- grpc++_error_details
- grpc++
- name: filter_end2end_test
- gtest: true
build: test
language: c++
src:
@@ -3657,9 +3643,9 @@
- gpr_test_util
- gpr
- name: generic_end2end_test
- gtest: true
build: test
language: c++
+ cpu_cost: 0.1
src:
- test/cpp/end2end/generic_end2end_test.cc
deps:
@@ -3670,7 +3656,6 @@
- gpr_test_util
- gpr
- name: golden_file_test
- gtest: true
build: test
language: c++
src:
@@ -3764,9 +3749,9 @@
vs_config_type: Application
vs_project_guid: '{069E9D05-B78B-4751-9252-D21EBAE7DE8E}'
- name: grpc_tool_test
- gtest: true
build: test
language: c++
+ cpu_cost: 0.2
src:
- src/proto/grpc/testing/echo.proto
- src/proto/grpc/testing/echo_messages.proto
@@ -3784,7 +3769,6 @@
filegroups:
- grpc++_codegen_proto
- name: grpclb_api_test
- gtest: true
build: test
language: c++
src:
@@ -3796,7 +3780,7 @@
- grpc++
- grpc
- name: grpclb_test
- gtest: false
+ cpu_cost: 0.1
build: test
language: c++
src:
@@ -3810,7 +3794,6 @@
- gpr_test_util
- gpr
- name: health_service_end2end_test
- gtest: true
build: test
language: c++
src:
@@ -3939,7 +3922,6 @@
- gpr
- grpc++_test_config
- name: mock_test
- gtest: true
build: test
language: c++
src:
@@ -3960,7 +3942,6 @@
- benchmark
defaults: benchmark
- name: proto_server_reflection_test
- gtest: true
build: test
language: c++
src:
@@ -3975,7 +3956,6 @@
- gpr_test_util
- gpr
- name: proto_utils_test
- gtest: true
build: test
language: c++
src:
@@ -4093,7 +4073,6 @@
- gpr
- grpc++_test_config
- name: round_robin_end2end_test
- gtest: true
build: test
language: c++
src:
@@ -4106,7 +4085,6 @@
- gpr_test_util
- gpr
- name: secure_auth_context_test
- gtest: true
build: test
language: c++
src:
@@ -4136,7 +4114,6 @@
- linux
- posix
- name: server_builder_plugin_test
- gtest: true
build: test
language: c++
src:
@@ -4149,7 +4126,6 @@
- gpr_test_util
- gpr
- name: server_builder_test
- gtest: true
build: test
language: c++
src:
@@ -4164,7 +4140,6 @@
- grpc
- gpr
- name: server_context_test_spouse_test
- gtest: true
build: test
language: c++
src:
@@ -4178,7 +4153,6 @@
uses:
- grpc++_test
- name: server_crash_test
- gtest: true
cpu_cost: 0.1
build: test
language: c++
@@ -4209,7 +4183,6 @@
- gpr_test_util
- gpr
- name: shutdown_test
- gtest: true
build: test
language: c++
src:
@@ -4233,7 +4206,6 @@
- gpr_test_util
- gpr
- name: streaming_throughput_test
- gtest: true
build: test
language: c++
src:
@@ -4287,7 +4259,6 @@
- gpr
- grpc++_test_config
- name: thread_stress_test
- gtest: true
cpu_cost: 100
build: test
language: c++
@@ -4301,8 +4272,7 @@
- gpr_test_util
- gpr
- name: writes_per_rpc_test
- gtest: true
- cpu_cost: 0.5
+ cpu_cost: 0.8
build: test
language: c++
src:
diff --git a/grpc.def b/grpc.def
index ee1dc77..b602172 100644
--- a/grpc.def
+++ b/grpc.def
@@ -89,7 +89,6 @@
grpc_server_request_registered_call
grpc_server_create
grpc_server_register_completion_queue
- grpc_server_register_non_listening_completion_queue
grpc_server_add_insecure_http2_port
grpc_server_start
grpc_server_shutdown_and_notify
diff --git a/include/grpc++/impl/codegen/client_unary_call.h b/include/grpc++/impl/codegen/client_unary_call.h
index a5a4f3d..4bf35ae 100644
--- a/include/grpc++/impl/codegen/client_unary_call.h
+++ b/include/grpc++/impl/codegen/client_unary_call.h
@@ -52,7 +52,9 @@
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result) {
- CompletionQueue cq(true); // Pluckable completion queue
+ CompletionQueue cq(grpc_completion_queue_attributes{
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+ GRPC_CQ_DEFAULT_POLLING}); // Pluckable completion queue
Call call(channel->CreateCall(method, context, &cq));
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,
diff --git a/include/grpc++/impl/codegen/completion_queue.h b/include/grpc++/impl/codegen/completion_queue.h
index 61617f2..c8ab726 100644
--- a/include/grpc++/impl/codegen/completion_queue.h
+++ b/include/grpc++/impl/codegen/completion_queue.h
@@ -102,7 +102,9 @@
public:
/// Default constructor. Implicitly creates a \a grpc_completion_queue
/// instance.
- CompletionQueue() : CompletionQueue(false) {}
+ CompletionQueue()
+ : CompletionQueue(grpc_completion_queue_attributes{
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {}
/// Wrap \a take, taking ownership of the instance.
///
@@ -182,6 +184,16 @@
};
void CompleteAvalanching();
+ protected:
+ /// Private constructor of CompletionQueue only visible to friend classes
+ CompletionQueue(const grpc_completion_queue_attributes& attributes) {
+ cq_ = g_core_codegen_interface->grpc_completion_queue_create(
+ g_core_codegen_interface->grpc_completion_queue_factory_lookup(
+ &attributes),
+ &attributes, NULL);
+ InitialAvalanching(); // reserve this for the future shutdown
+ }
+
private:
// Friend synchronous wrappers so that they can access Pluck(), which is
// a semi-private API geared towards the synchronous implementation.
@@ -215,18 +227,6 @@
const InputMessage& request,
OutputMessage* result);
- /// Private constructor of CompletionQueue only visible to friend classes
- CompletionQueue(bool is_pluck) {
- if (is_pluck) {
- cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_pluck(
- nullptr);
- } else {
- cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_next(
- nullptr);
- }
- InitialAvalanching(); // reserve this for the future shutdown
- }
-
NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
/// Wraps \a grpc_completion_queue_pluck.
@@ -289,17 +289,19 @@
/// by servers. Instantiated by \a ServerBuilder.
class ServerCompletionQueue : public CompletionQueue {
public:
- bool IsFrequentlyPolled() { return is_frequently_polled_; }
+ bool IsFrequentlyPolled() { return polling_type_ != GRPC_CQ_NON_LISTENING; }
private:
- bool is_frequently_polled_;
+ grpc_cq_polling_type polling_type_;
friend class ServerBuilder;
/// \param is_frequently_polled Informs the GRPC library about whether the
/// server completion queue would be actively polled (by calling Next() or
/// AsyncNext()). By default all server completion queues are assumed to be
/// frequently polled.
- ServerCompletionQueue(bool is_frequently_polled = true)
- : is_frequently_polled_(is_frequently_polled) {}
+ ServerCompletionQueue(grpc_cq_polling_type polling_type)
+ : CompletionQueue(grpc_completion_queue_attributes{
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}),
+ polling_type_(polling_type) {}
};
} // namespace grpc
diff --git a/include/grpc++/impl/codegen/core_codegen.h b/include/grpc++/impl/codegen/core_codegen.h
index 8660107..a159372 100644
--- a/include/grpc++/impl/codegen/core_codegen.h
+++ b/include/grpc++/impl/codegen/core_codegen.h
@@ -44,8 +44,15 @@
namespace grpc {
/// Implementation of the core codegen interface.
-class CoreCodegen : public CoreCodegenInterface {
+class CoreCodegen final : public CoreCodegenInterface {
private:
+ virtual const grpc_completion_queue_factory*
+ grpc_completion_queue_factory_lookup(
+ const grpc_completion_queue_attributes* attributes) override;
+ virtual grpc_completion_queue* grpc_completion_queue_create(
+ const grpc_completion_queue_factory* factory,
+ const grpc_completion_queue_attributes* attributes,
+ void* reserved) override;
grpc_completion_queue* grpc_completion_queue_create_for_next(
void* reserved) override;
grpc_completion_queue* grpc_completion_queue_create_for_pluck(
diff --git a/include/grpc++/impl/codegen/core_codegen_interface.h b/include/grpc++/impl/codegen/core_codegen_interface.h
index a0665f4..7cc3a82 100644
--- a/include/grpc++/impl/codegen/core_codegen_interface.h
+++ b/include/grpc++/impl/codegen/core_codegen_interface.h
@@ -59,6 +59,12 @@
virtual void assert_fail(const char* failed_assertion, const char* file,
int line) = 0;
+ virtual const grpc_completion_queue_factory*
+ grpc_completion_queue_factory_lookup(
+ const grpc_completion_queue_attributes* attributes) = 0;
+ virtual grpc_completion_queue* grpc_completion_queue_create(
+ const grpc_completion_queue_factory* factory,
+ const grpc_completion_queue_attributes* attributes, void* reserved) = 0;
virtual grpc_completion_queue* grpc_completion_queue_create_for_next(
void* reserved) = 0;
virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck(
diff --git a/include/grpc++/impl/codegen/sync_stream.h b/include/grpc++/impl/codegen/sync_stream.h
index 328d5cb..a010924 100644
--- a/include/grpc++/impl/codegen/sync_stream.h
+++ b/include/grpc++/impl/codegen/sync_stream.h
@@ -156,7 +156,9 @@
ClientReader(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const W& request)
: context_(context),
- cq_(true), // Pluckable cq
+ cq_(grpc_completion_queue_attributes{
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+ GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
@@ -230,7 +232,9 @@
ClientWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, R* response)
: context_(context),
- cq_(true), // Pluckable cq
+ cq_(grpc_completion_queue_attributes{
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+ GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
@@ -330,7 +334,9 @@
ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context)
: context_(context),
- cq_(true), // Pluckable cq
+ cq_(grpc_completion_queue_attributes{
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
+ GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
if (!context_->initial_metadata_corked_) {
CallOpSet<CallOpSendInitialMetadata> ops;
diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h
index 5ded451..1eeb075 100644
--- a/include/grpc/grpc.h
+++ b/include/grpc/grpc.h
@@ -93,55 +93,6 @@
/** Return a string specifying what the 'g' in gRPC stands for */
GRPCAPI const char *grpc_g_stands_for(void);
-/** Specifies the type of APIs to use to pop events from the completion queue */
-typedef enum {
- /** Events are popped out by calling grpc_completion_queue_next() API ONLY */
- GRPC_CQ_NEXT = 1,
-
- /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
- GRPC_CQ_PLUCK
-} grpc_cq_completion_type;
-
-/** Completion queues internally MAY maintain a set of file descriptors in a
- structure called 'pollset'. This enum specifies if a completion queue has an
- associated pollset and any restrictions on the type of file descriptors that
- can be present in the pollset.
-
- I/O progress can only be made when grpc_completion_queue_next() or
- grpc_completion_queue_pluck() are called on the completion queue (unless the
- grpc_cq_polling_type is GRPC_CQ_NON_POLLING) and hence it is very important
- to actively call these APIs */
-typedef enum {
- /** The completion queue will have an associated pollset and there is no
- restriction on the type of file descriptors the pollset may contain */
- GRPC_CQ_DEFAULT_POLLING,
-
- /** Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will
- not contain any 'listening file descriptors' (i.e file descriptors used to
- listen to incoming channels) */
- GRPC_CQ_NON_LISTENING,
-
- /** The completion queue will not have an associated pollset. Note that
- grpc_completion_queue_next() or grpc_completion_queue_pluck() MUST still
- be called to pop events from the completion queue; it is not required to
- call them actively to make I/O progress */
- GRPC_CQ_NON_POLLING
-} grpc_cq_polling_type;
-
-#define GRPC_CQ_CURRENT_VERSION 1
-typedef struct grpc_completion_queue_attributes {
- /* The version number of this structure. More fields might be added to this
- structure in future. */
- int version; /* Set to GRPC_CQ_CURRENT_VERSION */
-
- grpc_cq_completion_type cq_completion_type;
-
- grpc_cq_polling_type cq_polling_type;
-} grpc_completion_queue_attributes;
-
-/** The completion queue factory structure is opaque to the callers of grpc */
-typedef struct grpc_completion_queue_factory grpc_completion_queue_factory;
-
/** Returns the completion queue factory based on the attributes. MAY return a
NULL if no factory can be found */
GRPCAPI const grpc_completion_queue_factory *
@@ -427,15 +378,6 @@
grpc_completion_queue *cq,
void *reserved);
-/** Register a non-listening completion queue with the server. This API is
- similar to grpc_server_register_completion_queue except that the server will
- not use this completion_queue to listen to any incoming channels.
-
- Registering a non-listening completion queue will have negative performance
- impact and hence this API is not recommended for production use cases. */
-GRPCAPI void grpc_server_register_non_listening_completion_queue(
- grpc_server *server, grpc_completion_queue *q, void *reserved);
-
/** Add a HTTP2 over plaintext over tcp listener.
Returns bound port number on success, 0 on failure.
REQUIRES: server not started */
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index b5d637e..d9f5e07 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -556,6 +556,55 @@
typedef struct grpc_resource_quota grpc_resource_quota;
+/** Completion queues internally MAY maintain a set of file descriptors in a
+ structure called 'pollset'. This enum specifies if a completion queue has an
+ associated pollset and any restrictions on the type of file descriptors that
+ can be present in the pollset.
+
+ I/O progress can only be made when grpc_completion_queue_next() or
+ grpc_completion_queue_pluck() are called on the completion queue (unless the
+ grpc_cq_polling_type is GRPC_CQ_NON_POLLING) and hence it is very important
+ to actively call these APIs */
+typedef enum {
+ /** The completion queue will have an associated pollset and there is no
+ restriction on the type of file descriptors the pollset may contain */
+ GRPC_CQ_DEFAULT_POLLING,
+
+ /** Similar to GRPC_CQ_DEFAULT_POLLING except that the completion queues will
+ not contain any 'listening file descriptors' (i.e file descriptors used to
+ listen to incoming channels) */
+ GRPC_CQ_NON_LISTENING,
+
+ /** The completion queue will not have an associated pollset. Note that
+ grpc_completion_queue_next() or grpc_completion_queue_pluck() MUST still
+ be called to pop events from the completion queue; it is not required to
+ call them actively to make I/O progress */
+ GRPC_CQ_NON_POLLING
+} grpc_cq_polling_type;
+
+/** Specifies the type of APIs to use to pop events from the completion queue */
+typedef enum {
+ /** Events are popped out by calling grpc_completion_queue_next() API ONLY */
+ GRPC_CQ_NEXT = 1,
+
+ /** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
+ GRPC_CQ_PLUCK
+} grpc_cq_completion_type;
+
+#define GRPC_CQ_CURRENT_VERSION 1
+typedef struct grpc_completion_queue_attributes {
+ /* The version number of this structure. More fields might be added to this
+ structure in future. */
+ int version; /* Set to GRPC_CQ_CURRENT_VERSION */
+
+ grpc_cq_completion_type cq_completion_type;
+
+ grpc_cq_polling_type cq_polling_type;
+} grpc_completion_queue_attributes;
+
+/** The completion queue factory structure is opaque to the callers of grpc */
+typedef struct grpc_completion_queue_factory grpc_completion_queue_factory;
+
#ifdef __cplusplus
}
#endif
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index 35e9f7e..eae3f10 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -60,13 +60,154 @@
void *tag;
} plucker;
+typedef struct {
+ bool can_get_pollset;
+ bool can_listen;
+ size_t (*size)(void);
+ void (*init)(grpc_pollset *pollset, gpr_mu **mu);
+ grpc_error *(*kick)(grpc_pollset *pollset,
+ grpc_pollset_worker *specific_worker);
+ grpc_error *(*work)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
+ grpc_pollset_worker **worker, gpr_timespec now,
+ gpr_timespec deadline);
+ void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
+ grpc_closure *closure);
+ void (*destroy)(grpc_pollset *pollset);
+} cq_poller_vtable;
+
+typedef struct non_polling_worker {
+ gpr_cv cv;
+ bool kicked;
+ struct non_polling_worker *next;
+ struct non_polling_worker *prev;
+} non_polling_worker;
+
+typedef struct {
+ gpr_mu mu;
+ non_polling_worker *root;
+ grpc_closure *shutdown;
+} non_polling_poller;
+
+static size_t non_polling_poller_size(void) {
+ return sizeof(non_polling_poller);
+}
+
+static void non_polling_poller_init(grpc_pollset *pollset, gpr_mu **mu) {
+ non_polling_poller *npp = (non_polling_poller *)pollset;
+ gpr_mu_init(&npp->mu);
+ *mu = &npp->mu;
+}
+
+static void non_polling_poller_destroy(grpc_pollset *pollset) {
+ non_polling_poller *npp = (non_polling_poller *)pollset;
+ gpr_mu_destroy(&npp->mu);
+}
+
+static grpc_error *non_polling_poller_work(grpc_exec_ctx *exec_ctx,
+ grpc_pollset *pollset,
+ grpc_pollset_worker **worker,
+ gpr_timespec now,
+ gpr_timespec deadline) {
+ non_polling_poller *npp = (non_polling_poller *)pollset;
+ if (npp->shutdown) return GRPC_ERROR_NONE;
+ non_polling_worker w;
+ gpr_cv_init(&w.cv);
+ if (worker != NULL) *worker = (grpc_pollset_worker *)&w;
+ if (npp->root == NULL) {
+ npp->root = w.next = w.prev = &w;
+ } else {
+ w.next = npp->root;
+ w.prev = w.next->prev;
+ w.next->prev = w.prev->next = &w;
+ }
+ w.kicked = false;
+ while (!npp->shutdown && !w.kicked && !gpr_cv_wait(&w.cv, &npp->mu, deadline))
+ ;
+ if (&w == npp->root) {
+ npp->root = w.next;
+ if (&w == npp->root) {
+ if (npp->shutdown) {
+ grpc_closure_sched(exec_ctx, npp->shutdown, GRPC_ERROR_NONE);
+ }
+ npp->root = NULL;
+ }
+ }
+ w.next->prev = w.prev;
+ w.prev->next = w.next;
+ gpr_cv_destroy(&w.cv);
+ if (worker != NULL) *worker = NULL;
+ return GRPC_ERROR_NONE;
+}
+
+static grpc_error *non_polling_poller_kick(
+ grpc_pollset *pollset, grpc_pollset_worker *specific_worker) {
+ non_polling_poller *p = (non_polling_poller *)pollset;
+ if (specific_worker == NULL) specific_worker = (grpc_pollset_worker *)p->root;
+ if (specific_worker != NULL) {
+ non_polling_worker *w = (non_polling_worker *)specific_worker;
+ if (!w->kicked) {
+ w->kicked = true;
+ gpr_cv_signal(&w->cv);
+ }
+ }
+ return GRPC_ERROR_NONE;
+}
+
+static void non_polling_poller_shutdown(grpc_exec_ctx *exec_ctx,
+ grpc_pollset *pollset,
+ grpc_closure *closure) {
+ non_polling_poller *p = (non_polling_poller *)pollset;
+ GPR_ASSERT(closure != NULL);
+ p->shutdown = closure;
+ if (p->root == NULL) {
+ grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE);
+ } else {
+ non_polling_worker *w = p->root;
+ do {
+ gpr_cv_signal(&w->cv);
+ w = w->next;
+ } while (w != p->root);
+ }
+}
+
+static const cq_poller_vtable g_poller_vtable_by_poller_type[] = {
+ /* GRPC_CQ_DEFAULT_POLLING */
+ {.can_get_pollset = true,
+ .can_listen = true,
+ .size = grpc_pollset_size,
+ .init = grpc_pollset_init,
+ .kick = grpc_pollset_kick,
+ .work = grpc_pollset_work,
+ .shutdown = grpc_pollset_shutdown,
+ .destroy = grpc_pollset_destroy},
+ /* GRPC_CQ_NON_LISTENING */
+ {.can_get_pollset = true,
+ .can_listen = false,
+ .size = grpc_pollset_size,
+ .init = grpc_pollset_init,
+ .kick = grpc_pollset_kick,
+ .work = grpc_pollset_work,
+ .shutdown = grpc_pollset_shutdown,
+ .destroy = grpc_pollset_destroy},
+ /* GRPC_CQ_NON_POLLING */
+ {.can_get_pollset = false,
+ .can_listen = false,
+ .size = non_polling_poller_size,
+ .init = non_polling_poller_init,
+ .kick = non_polling_poller_kick,
+ .work = non_polling_poller_work,
+ .shutdown = non_polling_poller_shutdown,
+ .destroy = non_polling_poller_destroy},
+};
+
/* Completion queue structure */
struct grpc_completion_queue {
/** owned by pollset */
gpr_mu *mu;
grpc_cq_completion_type completion_type;
- grpc_cq_polling_type polling_type;
+
+ const cq_poller_vtable *poller_vtable;
/** completed events */
grpc_cq_completion completed_head;
@@ -127,15 +268,18 @@
"polling_type=%d)",
2, (completion_type, polling_type));
- cc = gpr_zalloc(sizeof(grpc_completion_queue) + grpc_pollset_size());
- grpc_pollset_init(POLLSET_FROM_CQ(cc), &cc->mu);
+ const cq_poller_vtable *poller_vtable =
+ &g_poller_vtable_by_poller_type[polling_type];
+
+ cc = gpr_zalloc(sizeof(grpc_completion_queue) + poller_vtable->size());
+ poller_vtable->init(POLLSET_FROM_CQ(cc), &cc->mu);
#ifndef NDEBUG
cc->outstanding_tags = NULL;
cc->outstanding_tag_capacity = 0;
#endif
cc->completion_type = completion_type;
- cc->polling_type = polling_type;
+ cc->poller_vtable = poller_vtable;
/* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cc->pending_events, 1);
@@ -164,10 +308,6 @@
return cc->completion_type;
}
-grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc) {
- return cc->polling_type;
-}
-
#ifdef GRPC_CQ_REF_COUNT_DEBUG
void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason,
const char *file, int line) {
@@ -195,7 +335,7 @@
#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head);
- grpc_pollset_destroy(POLLSET_FROM_CQ(cc));
+ cc->poller_vtable->destroy(POLLSET_FROM_CQ(cc));
#ifndef NDEBUG
gpr_free(cc->outstanding_tags);
#endif
@@ -280,7 +420,7 @@
}
}
grpc_error *kick_error =
- grpc_pollset_kick(POLLSET_FROM_CQ(cc), pluck_worker);
+ cc->poller_vtable->kick(POLLSET_FROM_CQ(cc), pluck_worker);
gpr_mu_unlock(cc->mu);
if (kick_error != GRPC_ERROR_NONE) {
const char *msg = grpc_error_string(kick_error);
@@ -295,8 +435,8 @@
GPR_ASSERT(!cc->shutdown);
GPR_ASSERT(cc->shutdown_called);
cc->shutdown = 1;
- grpc_pollset_shutdown(exec_ctx, POLLSET_FROM_CQ(cc),
- &cc->pollset_shutdown_done);
+ cc->poller_vtable->shutdown(exec_ctx, POLLSET_FROM_CQ(cc),
+ &cc->pollset_shutdown_done);
gpr_mu_unlock(cc->mu);
}
@@ -452,8 +592,8 @@
gpr_mu_lock(cc->mu);
continue;
} else {
- grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc), NULL,
- now, iteration_deadline);
+ grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc),
+ NULL, now, iteration_deadline);
if (err != GRPC_ERROR_NONE) {
gpr_mu_unlock(cc->mu);
const char *msg = grpc_error_string(err);
@@ -644,8 +784,8 @@
grpc_exec_ctx_flush(&exec_ctx);
gpr_mu_lock(cc->mu);
} else {
- grpc_error *err = grpc_pollset_work(&exec_ctx, POLLSET_FROM_CQ(cc),
- &worker, now, iteration_deadline);
+ grpc_error *err = cc->poller_vtable->work(
+ &exec_ctx, POLLSET_FROM_CQ(cc), &worker, now, iteration_deadline);
if (err != GRPC_ERROR_NONE) {
del_plucker(cc, tag, &worker);
gpr_mu_unlock(cc->mu);
@@ -689,8 +829,8 @@
if (gpr_unref(&cc->pending_events)) {
GPR_ASSERT(!cc->shutdown);
cc->shutdown = 1;
- grpc_pollset_shutdown(&exec_ctx, POLLSET_FROM_CQ(cc),
- &cc->pollset_shutdown_done);
+ cc->poller_vtable->shutdown(&exec_ctx, POLLSET_FROM_CQ(cc),
+ &cc->pollset_shutdown_done);
}
gpr_mu_unlock(cc->mu);
grpc_exec_ctx_finish(&exec_ctx);
@@ -706,7 +846,7 @@
}
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
- return POLLSET_FROM_CQ(cc);
+ return cc->poller_vtable->can_get_pollset ? POLLSET_FROM_CQ(cc) : NULL;
}
grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps) {
@@ -727,4 +867,10 @@
void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; }
-int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; }
+bool grpc_cq_is_server_cq(grpc_completion_queue *cc) {
+ return cc->is_server_cq;
+}
+
+bool grpc_cq_can_listen(grpc_completion_queue *cc) {
+ return cc->poller_vtable->can_listen;
+}
diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h
index 1ff3d64..a932087 100644
--- a/src/core/lib/surface/completion_queue.h
+++ b/src/core/lib/surface/completion_queue.h
@@ -94,13 +94,11 @@
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps);
-void grpc_cq_mark_non_listening_server_cq(grpc_completion_queue *cc);
-bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
-int grpc_cq_is_server_cq(grpc_completion_queue *cc);
+bool grpc_cq_is_server_cq(grpc_completion_queue *cc);
+bool grpc_cq_can_listen(grpc_completion_queue *cc);
grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc);
-grpc_cq_polling_type grpc_get_cq_polling_type(grpc_completion_queue *cc);
grpc_completion_queue *grpc_completion_queue_create_internal(
grpc_cq_completion_type completion_type, grpc_cq_polling_type polling_type);
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index 26c81e9..9e4ab54 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -981,7 +981,7 @@
static void register_completion_queue(grpc_server *server,
grpc_completion_queue *cq,
- bool is_non_listening, void *reserved) {
+ void *reserved) {
size_t i, n;
GPR_ASSERT(!reserved);
for (i = 0; i < server->cq_count; i++) {
@@ -990,10 +990,6 @@
grpc_cq_mark_server_cq(cq);
- if (is_non_listening) {
- grpc_cq_mark_non_listening_server_cq(cq);
- }
-
GRPC_CQ_INTERNAL_REF(cq, "server");
n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs,
@@ -1016,16 +1012,7 @@
calls grpc_completion_queue_pluck() on server completion queues */
}
- register_completion_queue(server, cq, false, reserved);
-}
-
-void grpc_server_register_non_listening_completion_queue(
- grpc_server *server, grpc_completion_queue *cq, void *reserved) {
- GRPC_API_TRACE(
- "grpc_server_register_non_listening_completion_queue(server=%p, cq=%p, "
- "reserved=%p)",
- 3, (server, cq, reserved));
- register_completion_queue(server, cq, true, reserved);
+ register_completion_queue(server, cq, reserved);
}
grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
@@ -1123,7 +1110,7 @@
server->requested_calls_per_cq =
gpr_malloc(sizeof(*server->requested_calls_per_cq) * server->cq_count);
for (i = 0; i < server->cq_count; i++) {
- if (!grpc_cq_is_non_listening_server_cq(server->cqs[i])) {
+ if (grpc_cq_can_listen(server->cqs[i])) {
server->pollsets[server->pollset_count++] =
grpc_cq_pollset(server->cqs[i]);
}
diff --git a/src/cpp/common/core_codegen.cc b/src/cpp/common/core_codegen.cc
index 4f8b826..f679a33 100644
--- a/src/cpp/common/core_codegen.cc
+++ b/src/cpp/common/core_codegen.cc
@@ -54,6 +54,18 @@
namespace grpc {
+const grpc_completion_queue_factory*
+CoreCodegen::grpc_completion_queue_factory_lookup(
+ const grpc_completion_queue_attributes* attributes) {
+ return ::grpc_completion_queue_factory_lookup(attributes);
+}
+
+grpc_completion_queue* CoreCodegen::grpc_completion_queue_create(
+ const grpc_completion_queue_factory* factory,
+ const grpc_completion_queue_attributes* attributes, void* reserved) {
+ return ::grpc_completion_queue_create(factory, attributes, reserved);
+}
+
grpc_completion_queue* CoreCodegen::grpc_completion_queue_create_for_next(
void* reserved) {
return ::grpc_completion_queue_create_for_next(reserved);
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index 3a15afc..2ead048 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -83,7 +83,8 @@
std::unique_ptr<ServerCompletionQueue> ServerBuilder::AddCompletionQueue(
bool is_frequently_polled) {
- ServerCompletionQueue* cq = new ServerCompletionQueue(is_frequently_polled);
+ ServerCompletionQueue* cq = new ServerCompletionQueue(
+ is_frequently_polled ? GRPC_CQ_DEFAULT_POLLING : GRPC_CQ_NON_LISTENING);
cqs_.push_back(cq);
return std::unique_ptr<ServerCompletionQueue>(cq);
}
@@ -242,6 +243,16 @@
sync_server_cqs(std::make_shared<
std::vector<std::unique_ptr<ServerCompletionQueue>>>());
+ int num_frequently_polled_cqs = 0;
+ for (auto it = cqs_.begin(); it != cqs_.end(); ++it) {
+ if ((*it)->IsFrequentlyPolled()) {
+ num_frequently_polled_cqs++;
+ }
+ }
+
+ const bool is_hybrid_server =
+ has_sync_methods && num_frequently_polled_cqs > 0;
+
if (has_sync_methods) {
// This is a Sync server
gpr_log(GPR_INFO,
@@ -251,9 +262,12 @@
sync_server_settings_.max_pollers,
sync_server_settings_.cq_timeout_msec);
+ grpc_cq_polling_type polling_type =
+ is_hybrid_server ? GRPC_CQ_NON_POLLING : GRPC_CQ_DEFAULT_POLLING;
+
// Create completion queues to listen to incoming rpc requests
for (int i = 0; i < sync_server_settings_.num_cqs; i++) {
- sync_server_cqs->emplace_back(new ServerCompletionQueue());
+ sync_server_cqs->emplace_back(new ServerCompletionQueue(polling_type));
}
}
@@ -269,12 +283,10 @@
// server
// 2. cqs_: Completion queues added via AddCompletionQueue() call
- // All sync cqs (if any) are frequently polled by ThreadManager
- int num_frequently_polled_cqs = sync_server_cqs->size();
-
for (auto it = sync_server_cqs->begin(); it != sync_server_cqs->end(); ++it) {
grpc_server_register_completion_queue(server->server_, (*it)->cq(),
nullptr);
+ num_frequently_polled_cqs++;
}
// cqs_ contains the completion queue added by calling the ServerBuilder's
@@ -283,14 +295,8 @@
// listening to incoming channels. Such completion queues must be registered
// as non-listening queues
for (auto it = cqs_.begin(); it != cqs_.end(); ++it) {
- if ((*it)->IsFrequentlyPolled()) {
- grpc_server_register_completion_queue(server->server_, (*it)->cq(),
- nullptr);
- num_frequently_polled_cqs++;
- } else {
- grpc_server_register_non_listening_completion_queue(server->server_,
- (*it)->cq(), nullptr);
- }
+ grpc_server_register_completion_queue(server->server_, (*it)->cq(),
+ nullptr);
}
if (num_frequently_polled_cqs == 0) {
diff --git a/src/node/ext/server_generic.cc b/src/node/ext/server_generic.cc
index 24573bd..088273d 100644
--- a/src/node/ext/server_generic.cc
+++ b/src/node/ext/server_generic.cc
@@ -44,9 +44,11 @@
namespace node {
Server::Server(grpc_server *server) : wrapped_server(server) {
- shutdown_queue = grpc_completion_queue_create_for_pluck(NULL);
- grpc_server_register_non_listening_completion_queue(server, shutdown_queue,
- NULL);
+ grpc_completion_queue_attributes attrs = {
+ GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK, GRPC_CQ_NON_LISTENING};
+ shutdown_queue = grpc_completion_queue_create(
+ grpc_completion_queue_factory_lookup(&attrs), &attrs, NULL);
+ grpc_server_register_completion_queue(server, shutdown_queue, NULL);
}
Server::~Server() {
diff --git a/src/proto/grpc/health/v1/BUILD b/src/proto/grpc/health/v1/BUILD
new file mode 100644
index 0000000..dbb91d9
--- /dev/null
+++ b/src/proto/grpc/health/v1/BUILD
@@ -0,0 +1,39 @@
+# Copyright 2017, 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.
+
+licenses(["notice"]) # 3-clause BSD
+
+package(default_visibility = ["//visibility:public"])
+
+load("//bazel:grpc_build_system.bzl", "grpc_proto_library")
+
+grpc_proto_library(
+ name = "health_proto",
+ srcs = ["health.proto"],
+)
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
index 4f1776e..f66f6e4 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/grpc.pxi
@@ -356,8 +356,6 @@
void grpc_server_register_completion_queue(grpc_server *server,
grpc_completion_queue *cq,
void *reserved) nogil
- void grpc_server_register_non_listening_completion_queue(
- grpc_server *server, grpc_completion_queue *cq, void *reserved) nogil
int grpc_server_add_insecure_http2_port(
grpc_server *server, const char *addr) nogil
void grpc_server_start(grpc_server *server) nogil
@@ -502,4 +500,3 @@
int grpc_compression_options_is_algorithm_enabled(
const grpc_compression_options *opts,
grpc_compression_algorithm algorithm) nogil
-
diff --git a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
index 18db38b..97192ef 100644
--- a/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
+++ b/src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi
@@ -82,20 +82,11 @@
self.c_server, queue.c_completion_queue, NULL)
self.registered_completion_queues.append(queue)
- def register_non_listening_completion_queue(
- self, CompletionQueue queue not None):
- if self.is_started:
- raise ValueError("cannot register completion queues after start")
- with nogil:
- grpc_server_register_non_listening_completion_queue(
- self.c_server, queue.c_completion_queue, NULL)
- self.registered_completion_queues.append(queue)
-
def start(self):
if self.is_started:
raise ValueError("the server has already started")
self.backup_shutdown_queue = CompletionQueue()
- self.register_non_listening_completion_queue(self.backup_shutdown_queue)
+ self.register_completion_queue(self.backup_shutdown_queue)
self.is_started = True
with nogil:
grpc_server_start(self.c_server)
diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.c b/src/ruby/ext/grpc/rb_grpc_imports.generated.c
index 1c68b2d..a412277 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.c
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.c
@@ -127,7 +127,6 @@
grpc_server_request_registered_call_type grpc_server_request_registered_call_import;
grpc_server_create_type grpc_server_create_import;
grpc_server_register_completion_queue_type grpc_server_register_completion_queue_import;
-grpc_server_register_non_listening_completion_queue_type grpc_server_register_non_listening_completion_queue_import;
grpc_server_add_insecure_http2_port_type grpc_server_add_insecure_http2_port_import;
grpc_server_start_type grpc_server_start_import;
grpc_server_shutdown_and_notify_type grpc_server_shutdown_and_notify_import;
@@ -425,7 +424,6 @@
grpc_server_request_registered_call_import = (grpc_server_request_registered_call_type) GetProcAddress(library, "grpc_server_request_registered_call");
grpc_server_create_import = (grpc_server_create_type) GetProcAddress(library, "grpc_server_create");
grpc_server_register_completion_queue_import = (grpc_server_register_completion_queue_type) GetProcAddress(library, "grpc_server_register_completion_queue");
- grpc_server_register_non_listening_completion_queue_import = (grpc_server_register_non_listening_completion_queue_type) GetProcAddress(library, "grpc_server_register_non_listening_completion_queue");
grpc_server_add_insecure_http2_port_import = (grpc_server_add_insecure_http2_port_type) GetProcAddress(library, "grpc_server_add_insecure_http2_port");
grpc_server_start_import = (grpc_server_start_type) GetProcAddress(library, "grpc_server_start");
grpc_server_shutdown_and_notify_import = (grpc_server_shutdown_and_notify_type) GetProcAddress(library, "grpc_server_shutdown_and_notify");
diff --git a/src/ruby/ext/grpc/rb_grpc_imports.generated.h b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
index c33fdd2..7df8dd6 100644
--- a/src/ruby/ext/grpc/rb_grpc_imports.generated.h
+++ b/src/ruby/ext/grpc/rb_grpc_imports.generated.h
@@ -332,9 +332,6 @@
typedef void(*grpc_server_register_completion_queue_type)(grpc_server *server, grpc_completion_queue *cq, void *reserved);
extern grpc_server_register_completion_queue_type grpc_server_register_completion_queue_import;
#define grpc_server_register_completion_queue grpc_server_register_completion_queue_import
-typedef void(*grpc_server_register_non_listening_completion_queue_type)(grpc_server *server, grpc_completion_queue *q, void *reserved);
-extern grpc_server_register_non_listening_completion_queue_type grpc_server_register_non_listening_completion_queue_import;
-#define grpc_server_register_non_listening_completion_queue grpc_server_register_non_listening_completion_queue_import
typedef int(*grpc_server_add_insecure_http2_port_type)(grpc_server *server, const char *addr);
extern grpc_server_add_insecure_http2_port_type grpc_server_add_insecure_http2_port_import;
#define grpc_server_add_insecure_http2_port grpc_server_add_insecure_http2_port_import
diff --git a/test/cpp/end2end/BUILD b/test/cpp/end2end/BUILD
index 0bf7948..f1212e1 100644
--- a/test/cpp/end2end/BUILD
+++ b/test/cpp/end2end/BUILD
@@ -51,6 +51,7 @@
"//src/proto/grpc/testing:echo_messages_proto",
"//src/proto/grpc/testing:echo_proto",
"//src/proto/grpc/testing/duplicate:echo_duplicate_proto",
+ "//src/proto/grpc/health/v1:health_proto",
"//test/core/util:gpr_test_util",
"//test/core/util:grpc_test_util",
"//test/cpp/util:test_util",
diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc
index 0b5215e..cc3958b 100644
--- a/test/cpp/end2end/async_end2end_test.cc
+++ b/test/cpp/end2end/async_end2end_test.cc
@@ -38,6 +38,7 @@
#include <grpc++/channel.h>
#include <grpc++/client_context.h>
#include <grpc++/create_channel.h>
+#include <grpc++/ext/health_check_service_server_builder_option.h>
#include <grpc++/server.h>
#include <grpc++/server_builder.h>
#include <grpc++/server_context.h>
@@ -49,6 +50,7 @@
#include <gtest/gtest.h>
#include "src/core/lib/iomgr/port.h"
+#include "src/proto/grpc/health/v1/health.grpc.pb.h"
#include "src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.h"
#include "src/proto/grpc/testing/echo.grpc.pb.h"
#include "test/core/util/port.h"
@@ -224,13 +226,15 @@
class TestScenario {
public:
- TestScenario(bool non_block, const grpc::string& creds_type,
+ TestScenario(bool non_block, const grpc::string& creds_type, bool hcs,
const grpc::string& content)
: disable_blocking(non_block),
+ health_check_service(hcs),
credentials_type(creds_type),
message_content(content) {}
void Log() const;
bool disable_blocking;
+ bool health_check_service;
// Although the below grpc::string's are logically const, we can't declare
// them const because of a limitation in the way old compilers (e.g., gcc-4.4)
// manage vector insertion using a copy constructor
@@ -243,6 +247,8 @@
return out << "TestScenario{disable_blocking="
<< (scenario.disable_blocking ? "true" : "false")
<< ", credentials='" << scenario.credentials_type
+ << ", health_check_service="
+ << (scenario.health_check_service ? "true" : "false")
<< "', message_size=" << scenario.message_content.size() << "}";
}
@@ -252,6 +258,8 @@
gpr_log(GPR_DEBUG, "%s", out.str().c_str());
}
+class HealthCheck : public health::v1::Health::Service {};
+
class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
protected:
AsyncEnd2endTest() { GetParam().Log(); }
@@ -268,6 +276,9 @@
GetParam().credentials_type);
builder.AddListeningPort(server_address_.str(), server_creds);
builder.RegisterService(&service_);
+ if (GetParam().health_check_service) {
+ builder.RegisterService(&health_check_);
+ }
cq_ = builder.AddCompletionQueue();
// TODO(zyc): make a test option to choose wheather sync plugins should be
@@ -340,6 +351,7 @@
std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
std::unique_ptr<Server> server_;
grpc::testing::EchoTestService::AsyncService service_;
+ HealthCheck health_check_;
std::ostringstream server_address_;
int port_;
@@ -1754,12 +1766,14 @@
messages.push_back(big_msg);
}
- for (auto cred = credentials_types.begin(); cred != credentials_types.end();
- ++cred) {
- for (auto msg = messages.begin(); msg != messages.end(); msg++) {
- scenarios.emplace_back(false, *cred, *msg);
- if (test_disable_blocking) {
- scenarios.emplace_back(true, *cred, *msg);
+ for (auto health_check_service : {false, true}) {
+ for (auto cred = credentials_types.begin(); cred != credentials_types.end();
+ ++cred) {
+ for (auto msg = messages.begin(); msg != messages.end(); msg++) {
+ scenarios.emplace_back(false, *cred, health_check_service, *msg);
+ if (test_disable_blocking) {
+ scenarios.emplace_back(true, *cred, health_check_service, *msg);
+ }
}
}
}
diff --git a/test/cpp/qps/gen_build_yaml.py b/test/cpp/qps/gen_build_yaml.py
index 2f035ab..2edcb86 100755
--- a/test/cpp/qps/gen_build_yaml.py
+++ b/test/cpp/qps/gen_build_yaml.py
@@ -65,7 +65,10 @@
def threads_required(scenario_json, where, is_tsan):
scenario_json = mutate_scenario(scenario_json, is_tsan)
+ if scenario_json['client_config']['outstanding_rpcs_per_channel'] == 1 and scenario_json['client_config']['client_channels'] == 1:
+ return 0.4
if scenario_json['%s_config' % where]['%s_type' % where] == 'ASYNC_%s' % where.upper():
+ if scenario_json['client_config']['client_channels'] == 1: return 1
return scenario_json['%s_config' % where].get('async_%s_threads' % where, 0)
return scenario_json['client_config']['outstanding_rpcs_per_channel'] * scenario_json['client_config']['client_channels']
diff --git a/tools/run_tests/generated/tests.json b/tools/run_tests/generated/tests.json
index 0cc270f..ce4684b 100644
--- a/tools/run_tests/generated/tests.json
+++ b/tools/run_tests/generated/tests.json
@@ -2591,11 +2591,11 @@
"posix",
"windows"
],
- "cpu_cost": 1.0,
+ "cpu_cost": 0.1,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "alarm_cpp_test",
"platforms": [
@@ -2613,11 +2613,11 @@
"posix",
"windows"
],
- "cpu_cost": 1.0,
+ "cpu_cost": 0.8,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "async_end2end_test",
"platforms": [
@@ -2639,7 +2639,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "auth_property_iterator_test",
"platforms": [
@@ -2964,7 +2964,7 @@
"mac",
"posix"
],
- "cpu_cost": 1.0,
+ "cpu_cost": 0.5,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
@@ -2989,7 +2989,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "channel_arguments_test",
"platforms": [
@@ -3011,7 +3011,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "channel_filter_test",
"platforms": [
@@ -3033,7 +3033,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "cli_call_test",
"platforms": [
@@ -3054,7 +3054,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "client_crash_test",
"platforms": [
@@ -3075,7 +3075,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "codegen_test_full",
"platforms": [
@@ -3097,7 +3097,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "codegen_test_minimal",
"platforms": [
@@ -3119,7 +3119,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "credentials_test",
"platforms": [
@@ -3141,7 +3141,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "cxx_byte_buffer_test",
"platforms": [
@@ -3163,7 +3163,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "cxx_slice_test",
"platforms": [
@@ -3185,7 +3185,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "cxx_string_ref_test",
"platforms": [
@@ -3207,7 +3207,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "cxx_time_test",
"platforms": [
@@ -3225,11 +3225,11 @@
"posix",
"windows"
],
- "cpu_cost": 0.5,
+ "cpu_cost": 1.0,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "end2end_test",
"platforms": [
@@ -3251,7 +3251,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "error_details_test",
"platforms": [
@@ -3273,7 +3273,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "filter_end2end_test",
"platforms": [
@@ -3295,7 +3295,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "generic_end2end_test",
"platforms": [
@@ -3319,7 +3319,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "golden_file_test",
"platforms": [
@@ -3341,7 +3341,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "grpc_tool_test",
"platforms": [
@@ -3363,7 +3363,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "grpclb_api_test",
"platforms": [
@@ -3381,7 +3381,7 @@
"posix",
"windows"
],
- "cpu_cost": 1.0,
+ "cpu_cost": 0.1,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
@@ -3407,7 +3407,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "health_service_end2end_test",
"platforms": [
@@ -3471,7 +3471,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "mock_test",
"platforms": [
@@ -3515,7 +3515,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "proto_server_reflection_test",
"platforms": [
@@ -3537,7 +3537,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "proto_utils_test",
"platforms": [
@@ -3579,7 +3579,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "round_robin_end2end_test",
"platforms": [
@@ -3601,7 +3601,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "secure_auth_context_test",
"platforms": [
@@ -3643,7 +3643,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "server_builder_plugin_test",
"platforms": [
@@ -3665,7 +3665,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "server_builder_test",
"platforms": [
@@ -3687,7 +3687,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "server_context_test_spouse_test",
"platforms": [
@@ -3708,7 +3708,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "server_crash_test",
"platforms": [
@@ -3729,7 +3729,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "shutdown_test",
"platforms": [
@@ -3772,7 +3772,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "streaming_throughput_test",
"platforms": [
@@ -3815,7 +3815,7 @@
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "thread_stress_test",
"platforms": [
@@ -3832,11 +3832,11 @@
"mac",
"posix"
],
- "cpu_cost": 0.5,
+ "cpu_cost": 0.8,
"exclude_configs": [],
"exclude_iomgrs": [],
"flaky": false,
- "gtest": true,
+ "gtest": false,
"language": "c++",
"name": "writes_per_rpc_test",
"platforms": [
@@ -41228,7 +41228,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41328,7 +41328,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 100,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41430,7 +41430,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41482,7 +41482,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41507,7 +41507,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41557,7 +41557,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41607,7 +41607,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41707,7 +41707,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41807,7 +41807,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -41907,7 +41907,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 100,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42009,7 +42009,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42061,7 +42061,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42086,7 +42086,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42136,7 +42136,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42186,7 +42186,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42286,7 +42286,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"tsan",
@@ -42386,7 +42386,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -42538,7 +42538,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 100,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -42692,7 +42692,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -42770,7 +42770,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -42808,7 +42808,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -42884,7 +42884,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -42960,7 +42960,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43112,7 +43112,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43264,7 +43264,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43416,7 +43416,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 100,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43570,7 +43570,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43648,7 +43648,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": "capacity",
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43686,7 +43686,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43762,7 +43762,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43838,7 +43838,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
@@ -43990,7 +43990,7 @@
"ci_platforms": [
"linux"
],
- "cpu_cost": 2,
+ "cpu_cost": 0.8,
"defaults": "boringssl",
"exclude_configs": [
"asan-noleaks",
diff --git a/tools/run_tests/run_tests.py b/tools/run_tests/run_tests.py
index 9130bc9..09b700a 100755
--- a/tools/run_tests/run_tests.py
+++ b/tools/run_tests/run_tests.py
@@ -1466,7 +1466,7 @@
# When running on travis, we want out test runs to be as similar as possible
# for reproducibility purposes.
if args.travis:
- massaged_one_run = sorted(one_run, key=lambda x: x.shortname)
+ massaged_one_run = sorted(one_run, key=lambda x: (x.cpu_cost, x.shortname))
else:
# whereas otherwise, we want to shuffle things up to give all tests a
# chance to run.