diff options
106 files changed, 1674 insertions, 374 deletions
@@ -117,7 +117,7 @@ most Mac installations. Do the "git submodule" command listed above. Then execute the following for all the needed build dependencies $ sudo /opt/local/bin/port install autoconf automake libtool gflags cmake - $ mkdir ~/gtest + $ mkdir ~/gtest-svn $ svn checkout http://googletest.googlecode.com/svn/trunk/ gtest-svn $ mkdir mybuild $ cd mybuild @@ -145,7 +145,7 @@ CC_tsan = clang CXX_tsan = clang++ LD_tsan = clang LDXX_tsan = clang++ -CPPFLAGS_tsan = -O0 -fsanitize=thread -fno-omit-frame-pointer +CPPFLAGS_tsan = -O0 -fsanitize=thread -fno-omit-frame-pointer -Wno-error=unused-command-line-argument LDFLAGS_tsan = -fsanitize=thread DEFINES_tsan = NDEBUG GRPC_TEST_SLOWDOWN_BUILD_FACTOR=10 @@ -155,7 +155,7 @@ CC_asan = clang CXX_asan = clang++ LD_asan = clang LDXX_asan = clang++ -CPPFLAGS_asan = -O0 -fsanitize=address -fno-omit-frame-pointer +CPPFLAGS_asan = -O0 -fsanitize=address -fno-omit-frame-pointer -Wno-error=unused-command-line-argument LDFLAGS_asan = -fsanitize=address DEFINES_asan = GRPC_TEST_SLOWDOWN_BUILD_FACTOR=3 @@ -165,7 +165,7 @@ CC_msan = clang CXX_msan = clang++-libc++ LD_msan = clang LDXX_msan = clang++-libc++ -CPPFLAGS_msan = -O0 -fsanitize=memory -fsanitize-memory-track-origins -fno-omit-frame-pointer -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1 +CPPFLAGS_msan = -O0 -fsanitize=memory -fsanitize-memory-track-origins -fno-omit-frame-pointer -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1 -Wno-error=unused-command-line-argument OPENSSL_CFLAGS_msan = -DPURIFY LDFLAGS_msan = -fsanitize=memory -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1 DEFINES_msan = NDEBUG GRPC_TEST_SLOWDOWN_BUILD_FACTOR=4 @@ -176,7 +176,7 @@ CC_ubsan = clang CXX_ubsan = clang++ LD_ubsan = clang LDXX_ubsan = clang++ -CPPFLAGS_ubsan = -O1 -fsanitize=undefined -fno-omit-frame-pointer +CPPFLAGS_ubsan = -O1 -fsanitize=undefined -fno-omit-frame-pointer -Wno-error=unused-command-line-argument OPENSSL_CFLAGS_ubsan = -DPURIFY LDFLAGS_ubsan = -fsanitize=undefined DEFINES_ubsan = NDEBUG GRPC_TEST_SLOWDOWN_BUILD_FACTOR=3 @@ -241,10 +241,6 @@ HOST_CXX = $(CXX) HOST_LD = $(LD) HOST_LDXX = $(LDXX) -CPPFLAGS += $(CPPFLAGS_$(CONFIG)) -DEFINES += $(DEFINES_$(CONFIG)) INSTALL_PREFIX=\"$(prefix)\" -LDFLAGS += $(LDFLAGS_$(CONFIG)) - ifdef EXTRA_DEFINES DEFINES += $(EXTRA_DEFINES) endif @@ -258,6 +254,10 @@ endif CPPFLAGS += -g -Wall -Wextra -Werror -Wno-long-long -Wno-unused-parameter LDFLAGS += -g +CPPFLAGS += $(CPPFLAGS_$(CONFIG)) +DEFINES += $(DEFINES_$(CONFIG)) INSTALL_PREFIX=\"$(prefix)\" +LDFLAGS += $(LDFLAGS_$(CONFIG)) + ifneq ($(SYSTEM),MINGW32) PIC_CPPFLAGS = -fPIC CPPFLAGS += -fPIC @@ -1443,7 +1443,7 @@ run_dep_checks: $(LIBDIR)/$(CONFIG)/zlib/libz.a: $(E) "[MAKE] Building zlib" - $(Q)(cd third_party/zlib ; CC="$(CC)" CFLAGS="$(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG))" ./configure --static) + $(Q)(cd third_party/zlib ; CC="$(CC)" CFLAGS="$(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(ZLIB_CFLAGS_EXTRA)" ./configure --static) $(Q)$(MAKE) -C third_party/zlib clean $(Q)$(MAKE) -C third_party/zlib $(Q)mkdir -p $(LIBDIR)/$(CONFIG)/zlib @@ -1452,7 +1452,7 @@ $(LIBDIR)/$(CONFIG)/zlib/libz.a: $(LIBDIR)/$(CONFIG)/openssl/libssl.a: $(E) "[MAKE] Building openssl for $(SYSTEM)" ifeq ($(SYSTEM),Darwin) - $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG))" ./Configure darwin64-x86_64-cc) + $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_EXTRA)" ./Configure darwin64-x86_64-cc) else ifeq ($(SYSTEM),MINGW32) @echo "We currently don't have a good way to compile OpenSSL in-place under msys." @@ -1473,7 +1473,7 @@ ifeq ($(SYSTEM),MINGW32) @echo " CPPFLAGS=-I/c/OpenSSL-Win64/include LDFLAGS=-L/c/OpenSSL-Win64 make" @false else - $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG))" ./config no-asm $(OPENSSL_CONFIG_$(CONFIG))) + $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_EXTRA)" ./config no-asm $(OPENSSL_CONFIG_$(CONFIG))) endif endif $(Q)$(MAKE) -C third_party/openssl clean @@ -1487,7 +1487,7 @@ third_party/protobuf/configure: $(LIBDIR)/$(CONFIG)/protobuf/libprotobuf.a: third_party/protobuf/configure $(E) "[MAKE] Building protobuf" - $(Q)(cd third_party/protobuf ; CC="$(CC)" CXX="$(CXX)" LDFLAGS="$(LDFLAGS_$(CONFIG)) -g" CPPFLAGS="$(PIC_CPPFLAGS) $(CPPFLAGS_$(CONFIG)) -g" ./configure --disable-shared --enable-static) + $(Q)(cd third_party/protobuf ; CC="$(CC)" CXX="$(CXX)" LDFLAGS="$(LDFLAGS_$(CONFIG)) -g $(PROTOBUF_LDFLAGS_EXTRA)" CPPFLAGS="$(PIC_CPPFLAGS) $(CPPFLAGS_$(CONFIG)) -g $(PROTOBUF_CPPFLAGS_EXTRA)" ./configure --disable-shared --enable-static) $(Q)$(MAKE) -C third_party/protobuf clean $(Q)$(MAKE) -C third_party/protobuf $(Q)mkdir -p $(LIBDIR)/$(CONFIG)/protobuf diff --git a/include/grpc++/channel_arguments.h b/include/grpc++/channel_arguments.h index d726b9ffea..4d926377ec 100644 --- a/include/grpc++/channel_arguments.h +++ b/include/grpc++/channel_arguments.h @@ -54,6 +54,14 @@ class ChannelArguments { ChannelArguments() {} ~ChannelArguments() {} + ChannelArguments(const ChannelArguments& other); + ChannelArguments& operator=(ChannelArguments other) { + Swap(other); + return *this; + } + + void Swap(ChannelArguments& other); + // grpc specific channel argument setters // Set target name override for SSL host name checking. void SetSslTargetNameOverride(const grpc::string& name); @@ -73,10 +81,6 @@ class ChannelArguments { friend class SecureCredentials; friend class testing::ChannelArgumentsTest; - // TODO(yangg) implement copy and assign - ChannelArguments(const ChannelArguments&); - ChannelArguments& operator=(const ChannelArguments&); - // Returns empty string when it is not set. grpc::string GetSslTargetNameOverride() const; diff --git a/include/grpc/grpc.h b/include/grpc/grpc.h index f8218bec4c..b05e4d65de 100644 --- a/include/grpc/grpc.h +++ b/include/grpc/grpc.h @@ -126,6 +126,12 @@ typedef struct { /** Initial sequence number for http2 transports */ #define GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER \ "grpc.http2.initial_sequence_number" +/** Primary user agent: goes at the start of the user-agent metadata + sent on each request */ +#define GRPC_ARG_PRIMARY_USER_AGENT_STRING "grpc.primary_user_agent" +/** Secondary user agent: goes at the end of the user-agent metadata + sent on each request */ +#define GRPC_ARG_SECONDARY_USER_AGENT_STRING "grpc.secondary_user_agent" /** Connectivity state of a channel. */ typedef enum { diff --git a/include/grpc/support/port_platform.h b/include/grpc/support/port_platform.h index a5d1b62702..57fed18cf6 100644 --- a/include/grpc/support/port_platform.h +++ b/include/grpc/support/port_platform.h @@ -71,6 +71,7 @@ #if !defined(GPR_NO_AUTODETECT_PLATFORM) #if defined(_WIN64) || defined(WIN64) +#define GPR_PLATFORM_STRING "windows" #define GPR_WIN32 1 #define GPR_ARCH_64 1 #define GPR_GETPID_IN_PROCESS_H 1 @@ -84,6 +85,7 @@ #endif #define GPR_WINDOWS_CRASH_HANDLER 1 #elif defined(_WIN32) || defined(WIN32) +#define GPR_PLATFORM_STRING "windows" #define GPR_ARCH_32 1 #define GPR_WIN32 1 #define GPR_GETPID_IN_PROCESS_H 1 @@ -97,6 +99,7 @@ #endif #define GPR_WINDOWS_CRASH_HANDLER 1 #elif defined(ANDROID) || defined(__ANDROID__) +#define GPR_PLATFORM_STRING "android" #define GPR_ANDROID 1 #define GPR_ARCH_32 1 #define GPR_CPU_LINUX 1 @@ -117,6 +120,7 @@ #define GPR_GETPID_IN_UNISTD_H 1 #define GPR_HAVE_MSG_NOSIGNAL 1 #elif defined(__linux__) +#define GPR_PLATFORM_STRING "linux" #ifndef _BSD_SOURCE #define _BSD_SOURCE #endif @@ -173,9 +177,11 @@ #define _BSD_SOURCE #endif #if TARGET_OS_IPHONE +#define GPR_PLATFORM_STRING "ios" #define GPR_CPU_IPHONE 1 #define GPR_PTHREAD_TLS 1 #else /* TARGET_OS_IPHONE */ +#define GPR_PLATFORM_STRING "osx" #define GPR_CPU_POSIX 1 #define GPR_GCC_TLS 1 #endif @@ -201,6 +207,7 @@ #define GPR_ARCH_32 1 #endif /* _LP64 */ #elif defined(__FreeBSD__) +#define GPR_PLATFORM_STRING "freebsd" #ifndef _BSD_SOURCE #define _BSD_SOURCE #endif @@ -232,6 +239,11 @@ #endif #endif /* GPR_NO_AUTODETECT_PLATFORM */ +#ifndef GPR_PLATFORM_STRING +#warning "GPR_PLATFORM_STRING not auto-detected" +#define GPR_PLATFORM_STRING "unknown" +#endif + /* For a common case, assume that the platform has a C99-like stdint.h */ #include <stdint.h> diff --git a/include/grpc/support/time.h b/include/grpc/support/time.h index 3f375f6ecd..be59c37956 100644 --- a/include/grpc/support/time.h +++ b/include/grpc/support/time.h @@ -83,6 +83,9 @@ void gpr_time_init(void); /* Return the current time measured from the given clocks epoch. */ gpr_timespec gpr_now(gpr_clock_type clock); +/* Convert a timespec from one clock to another */ +gpr_timespec gpr_convert_clock_type(gpr_timespec t, gpr_clock_type target_clock); + /* Return -ve, 0, or +ve according to whether a < b, a == b, or a > b respectively. */ int gpr_time_cmp(gpr_timespec a, gpr_timespec b); diff --git a/src/compiler/csharp_generator.cc b/src/compiler/csharp_generator.cc index 1910e9bd2d..64371047e0 100644 --- a/src/compiler/csharp_generator.cc +++ b/src/compiler/csharp_generator.cc @@ -149,7 +149,7 @@ std::string GetMethodRequestParamMaybe(const MethodDescriptor *method) { std::string GetMethodReturnTypeClient(const MethodDescriptor *method) { switch (GetMethodType(method)) { case METHODTYPE_NO_STREAMING: - return "Task<" + GetClassName(method->output_type()) + ">"; + return "AsyncUnaryCall<" + GetClassName(method->output_type()) + ">"; case METHODTYPE_CLIENT_STREAMING: return "AsyncClientStreamingCall<" + GetClassName(method->input_type()) + ", " + GetClassName(method->output_type()) + ">"; @@ -298,7 +298,7 @@ void GenerateServerInterface(Printer* out, const ServiceDescriptor *service) { out->Indent(); for (int i = 0; i < service->method_count(); i++) { const MethodDescriptor *method = service->method(i); - out->Print("$returntype$ $methodname$(ServerCallContext context, $request$$response_stream_maybe$);\n", + out->Print("$returntype$ $methodname$($request$$response_stream_maybe$, ServerCallContext context);\n", "methodname", method->name(), "returntype", GetMethodReturnTypeServer(method), "request", GetMethodRequestParamServer(method), "response_stream_maybe", diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 8eb95ca822..108a6dfdf1 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -236,21 +236,6 @@ static void picked_target(void *arg, int iomgr_success) { } } -static void pick_target(grpc_lb_policy *lb_policy, call_data *calld) { - grpc_metadata_batch *initial_metadata; - grpc_transport_stream_op *op = &calld->waiting_op; - - GPR_ASSERT(op->bind_pollset); - GPR_ASSERT(op->send_ops); - GPR_ASSERT(op->send_ops->nops >= 1); - GPR_ASSERT(op->send_ops->ops[0].type == GRPC_OP_METADATA); - initial_metadata = &op->send_ops->ops[0].data.metadata; - - grpc_iomgr_closure_init(&calld->async_setup_task, picked_target, calld); - grpc_lb_policy_pick(lb_policy, op->bind_pollset, initial_metadata, - &calld->picked_channel, &calld->async_setup_task); -} - static grpc_iomgr_closure *merge_into_waiting_op( grpc_call_element *elem, grpc_transport_stream_op *new_op) { call_data *calld = elem->call_data; @@ -378,12 +363,23 @@ static void perform_transport_stream_op(grpc_call_element *elem, gpr_mu_lock(&chand->mu_config); lb_policy = chand->lb_policy; if (lb_policy) { + grpc_transport_stream_op *op = &calld->waiting_op; + grpc_pollset *bind_pollset = op->bind_pollset; + grpc_metadata_batch *initial_metadata = &op->send_ops->ops[0].data.metadata; GRPC_LB_POLICY_REF(lb_policy, "pick"); gpr_mu_unlock(&chand->mu_config); calld->state = CALL_WAITING_FOR_PICK; + + GPR_ASSERT(op->bind_pollset); + GPR_ASSERT(op->send_ops); + GPR_ASSERT(op->send_ops->nops >= 1); + GPR_ASSERT( + op->send_ops->ops[0].type == GRPC_OP_METADATA); gpr_mu_unlock(&calld->mu_state); - pick_target(lb_policy, calld); + grpc_iomgr_closure_init(&calld->async_setup_task, picked_target, calld); + grpc_lb_policy_pick(lb_policy, bind_pollset, initial_metadata, + &calld->picked_channel, &calld->async_setup_task); GRPC_LB_POLICY_UNREF(lb_policy, "pick"); } else if (chand->resolver != NULL) { diff --git a/src/core/channel/compress_filter.c b/src/core/channel/compress_filter.c index d07d96f3f9..14e8ca7325 100644 --- a/src/core/channel/compress_filter.c +++ b/src/core/channel/compress_filter.c @@ -200,7 +200,7 @@ static void process_send_ops(grpc_call_element *elem, channeld->default_compression_algorithm; calld->has_compression_algorithm = 1; /* GPR_TRUE */ } - grpc_metadata_batch_add_head( + grpc_metadata_batch_add_tail( &(sop->data.metadata), &calld->compression_algorithm_storage, grpc_mdelem_ref(channeld->mdelem_compression_algorithms [calld->compression_algorithm])); diff --git a/src/core/channel/http_client_filter.c b/src/core/channel/http_client_filter.c index 6e8c287e3d..6ae8488070 100644 --- a/src/core/channel/http_client_filter.c +++ b/src/core/channel/http_client_filter.c @@ -32,13 +32,17 @@ #include "src/core/channel/http_client_filter.h" #include <string.h> +#include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include "src/core/support/string.h" typedef struct call_data { grpc_linked_mdelem method; grpc_linked_mdelem scheme; grpc_linked_mdelem te_trailers; grpc_linked_mdelem content_type; + grpc_linked_mdelem user_agent; int sent_initial_metadata; int got_initial_metadata; @@ -58,6 +62,8 @@ typedef struct channel_data { grpc_mdelem *scheme; grpc_mdelem *content_type; grpc_mdelem *status; + /** complete user agent mdelem */ + grpc_mdelem *user_agent; } channel_data; /* used to silence 'variable not used' warnings */ @@ -92,6 +98,18 @@ static void hc_on_recv(void *user_data, int success) { calld->on_done_recv->cb(calld->on_done_recv->cb_arg, success); } +static grpc_mdelem *client_strip_filter(void *user_data, grpc_mdelem *md) { + grpc_call_element *elem = user_data; + channel_data *channeld = elem->channel_data; + /* eat the things we'd like to set ourselves */ + if (md->key == channeld->method->key) return NULL; + if (md->key == channeld->scheme->key) return NULL; + if (md->key == channeld->te_trailers->key) return NULL; + if (md->key == channeld->content_type->key) return NULL; + if (md->key == channeld->user_agent->key) return NULL; + return md; +} + static void hc_mutate_op(grpc_call_element *elem, grpc_transport_stream_op *op) { /* grab pointers to our data from the call element */ @@ -105,6 +123,7 @@ static void hc_mutate_op(grpc_call_element *elem, grpc_stream_op *op = &ops[i]; if (op->type != GRPC_OP_METADATA) continue; calld->sent_initial_metadata = 1; + grpc_metadata_batch_filter(&op->data.metadata, client_strip_filter, elem); /* Send : prefixed headers, which have to be before any application layer headers. */ grpc_metadata_batch_add_head(&op->data.metadata, &calld->method, @@ -115,6 +134,8 @@ static void hc_mutate_op(grpc_call_element *elem, GRPC_MDELEM_REF(channeld->te_trailers)); grpc_metadata_batch_add_tail(&op->data.metadata, &calld->content_type, GRPC_MDELEM_REF(channeld->content_type)); + grpc_metadata_batch_add_tail(&op->data.metadata, &calld->user_agent, + GRPC_MDELEM_REF(channeld->user_agent)); break; } } @@ -169,6 +190,55 @@ static const char *scheme_from_args(const grpc_channel_args *args) { return "http"; } +static grpc_mdstr *user_agent_from_args(grpc_mdctx *mdctx, + const grpc_channel_args *args) { + gpr_strvec v; + size_t i; + int is_first = 1; + char *tmp; + grpc_mdstr *result; + + gpr_strvec_init(&v); + + for (i = 0; args && i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_PRIMARY_USER_AGENT_STRING)) { + if (args->args[i].type != GRPC_ARG_STRING) { + gpr_log(GPR_ERROR, "Channel argument '%s' should be a string", + GRPC_ARG_PRIMARY_USER_AGENT_STRING); + } else { + if (!is_first) gpr_strvec_add(&v, gpr_strdup(" ")); + is_first = 0; + gpr_strvec_add(&v, gpr_strdup(args->args[i].value.string)); + } + } + } + + gpr_asprintf(&tmp, "%sgrpc-c/%s (%s)", is_first ? "" : " ", + grpc_version_string(), GPR_PLATFORM_STRING); + is_first = 0; + gpr_strvec_add(&v, tmp); + + for (i = 0; args && i < args->num_args; i++) { + if (0 == strcmp(args->args[i].key, GRPC_ARG_SECONDARY_USER_AGENT_STRING)) { + if (args->args[i].type != GRPC_ARG_STRING) { + gpr_log(GPR_ERROR, "Channel argument '%s' should be a string", + GRPC_ARG_SECONDARY_USER_AGENT_STRING); + } else { + if (!is_first) gpr_strvec_add(&v, gpr_strdup(" ")); + is_first = 0; + gpr_strvec_add(&v, gpr_strdup(args->args[i].value.string)); + } + } + } + + tmp = gpr_strvec_flatten(&v, NULL); + gpr_strvec_destroy(&v); + result = grpc_mdstr_from_string(mdctx, tmp); + gpr_free(tmp); + + return result; +} + /* Constructor for channel_data */ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, const grpc_channel_args *args, grpc_mdctx *mdctx, @@ -189,6 +259,9 @@ static void init_channel_elem(grpc_channel_element *elem, grpc_channel *master, channeld->content_type = grpc_mdelem_from_strings(mdctx, "content-type", "application/grpc"); channeld->status = grpc_mdelem_from_strings(mdctx, ":status", "200"); + channeld->user_agent = grpc_mdelem_from_metadata_strings( + mdctx, grpc_mdstr_from_string(mdctx, "user-agent"), + user_agent_from_args(mdctx, args)); } /* Destructor for channel data */ @@ -201,6 +274,7 @@ static void destroy_channel_elem(grpc_channel_element *elem) { GRPC_MDELEM_UNREF(channeld->scheme); GRPC_MDELEM_UNREF(channeld->content_type); GRPC_MDELEM_UNREF(channeld->status); + GRPC_MDELEM_UNREF(channeld->user_agent); } const grpc_channel_filter grpc_http_client_filter = { diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c index 2a12fbc86d..358b907e59 100644 --- a/src/core/client_config/subchannel.c +++ b/src/core/client_config/subchannel.c @@ -300,7 +300,7 @@ static void continue_connect(grpc_subchannel *c) { } static void start_connect(grpc_subchannel *c) { - gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); c->next_attempt = now; c->backoff_delta = gpr_time_from_seconds(1, GPR_TIMESPAN); @@ -585,7 +585,7 @@ static void subchannel_connected(void *arg, int iomgr_success) { c->have_alarm = 1; c->next_attempt = gpr_time_add(c->next_attempt, c->backoff_delta); c->backoff_delta = gpr_time_add(c->backoff_delta, c->backoff_delta); - grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_MONOTONIC)); gpr_mu_unlock(&c->mu); } } diff --git a/src/core/iomgr/alarm.c b/src/core/iomgr/alarm.c index 5b9a37e5dd..931f746f75 100644 --- a/src/core/iomgr/alarm.c +++ b/src/core/iomgr/alarm.c @@ -36,6 +36,7 @@ #include "src/core/iomgr/alarm_heap.h" #include "src/core/iomgr/alarm_internal.h" #include "src/core/iomgr/time_averaged_stats.h" +#include <grpc/support/log.h> #include <grpc/support/sync.h> #include <grpc/support/useful.h> @@ -67,6 +68,7 @@ typedef struct { static gpr_mu g_mu; /* Allow only one run_some_expired_alarms at once */ static gpr_mu g_checker_mu; +static gpr_clock_type g_clock_type; static shard_type g_shards[NUM_SHARDS]; /* Protected by g_mu */ static shard_type *g_shard_queue[NUM_SHARDS]; @@ -85,6 +87,7 @@ void grpc_alarm_list_init(gpr_timespec now) { gpr_mu_init(&g_mu); gpr_mu_init(&g_checker_mu); + g_clock_type = now.clock_type; for (i = 0; i < NUM_SHARDS; i++) { shard_type *shard = &g_shards[i]; @@ -102,7 +105,7 @@ void grpc_alarm_list_init(gpr_timespec now) { void grpc_alarm_list_shutdown(void) { int i; - while (run_some_expired_alarms(NULL, gpr_inf_future(GPR_CLOCK_REALTIME), NULL, + while (run_some_expired_alarms(NULL, gpr_inf_future(g_clock_type), NULL, 0)) ; for (i = 0; i < NUM_SHARDS; i++) { @@ -175,6 +178,8 @@ void grpc_alarm_init(grpc_alarm *alarm, gpr_timespec deadline, gpr_timespec now) { int is_first_alarm = 0; shard_type *shard = &g_shards[shard_idx(alarm)]; + GPR_ASSERT(deadline.clock_type == g_clock_type); + GPR_ASSERT(now.clock_type == g_clock_type); alarm->cb = alarm_cb; alarm->cb_arg = alarm_cb_arg; alarm->deadline = deadline; @@ -355,6 +360,7 @@ static int run_some_expired_alarms(gpr_mu *drop_mu, gpr_timespec now, } int grpc_alarm_check(gpr_mu *drop_mu, gpr_timespec now, gpr_timespec *next) { + GPR_ASSERT(now.clock_type == g_clock_type); return run_some_expired_alarms(drop_mu, now, next, 1); } diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index 0244f689b1..a18c176b30 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -57,9 +57,9 @@ static grpc_iomgr_object g_root_object; static void background_callback_executor(void *ignored) { gpr_mu_lock(&g_mu); while (!g_shutdown) { - gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_REALTIME); + gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); gpr_timespec short_deadline = gpr_time_add( - gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100, GPR_TIMESPAN)); + gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_millis(100, GPR_TIMESPAN)); if (g_cbs_head) { grpc_iomgr_closure *closure = g_cbs_head; g_cbs_head = closure->next; @@ -67,7 +67,7 @@ static void background_callback_executor(void *ignored) { gpr_mu_unlock(&g_mu); closure->cb(closure->cb_arg, closure->success); gpr_mu_lock(&g_mu); - } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_REALTIME), + } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_MONOTONIC), &deadline)) { } else { gpr_mu_unlock(&g_mu); @@ -90,7 +90,7 @@ void grpc_iomgr_init(void) { gpr_thd_id id; gpr_mu_init(&g_mu); gpr_cv_init(&g_rcv); - grpc_alarm_list_init(gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_list_init(gpr_now(GPR_CLOCK_MONOTONIC)); g_root_object.next = g_root_object.prev = &g_root_object; g_root_object.name = "root"; grpc_iomgr_platform_init(); @@ -145,7 +145,7 @@ void grpc_iomgr_shutdown(void) { } while (g_cbs_head); continue; } - if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_REALTIME), NULL)) { + if (grpc_alarm_check(&g_mu, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL)) { gpr_log(GPR_DEBUG, "got late alarm"); continue; } diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index efb301d81c..c8646af615 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -136,7 +136,7 @@ static void finish_shutdown(grpc_pollset *pollset) { int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { /* pollset->mu already held */ - gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); if (gpr_time_cmp(now, deadline) > 0) { return 0; } @@ -205,7 +205,7 @@ int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, gpr_timespec now) { gpr_timespec timeout; static const int max_spin_polling_us = 10; - if (gpr_time_cmp(deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) { return -1; } if (gpr_time_cmp(deadline, gpr_time_add(now, gpr_time_from_micros( diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c index 24226cc980..a9c4739c7c 100644 --- a/src/core/iomgr/pollset_windows.c +++ b/src/core/iomgr/pollset_windows.c @@ -70,7 +70,7 @@ void grpc_pollset_destroy(grpc_pollset *pollset) { int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { gpr_timespec now; - now = gpr_now(GPR_CLOCK_REALTIME); + now = gpr_now(GPR_CLOCK_MONOTONIC); if (gpr_time_cmp(now, deadline) > 0) { return 0 /* GPR_FALSE */; } diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c index 427cd86c4e..392eda999e 100644 --- a/src/core/iomgr/tcp_client_posix.c +++ b/src/core/iomgr/tcp_client_posix.c @@ -116,6 +116,8 @@ static void on_writable(void *acp, int success) { void (*cb)(void *arg, grpc_endpoint *tcp) = ac->cb; void *cb_arg = ac->cb_arg; + grpc_alarm_cancel(&ac->alarm); + gpr_mu_lock(&ac->mu); if (success) { do { @@ -182,8 +184,6 @@ finish: gpr_mu_destroy(&ac->mu); gpr_free(ac->addr_str); gpr_free(ac); - } else { - grpc_alarm_cancel(&ac->alarm); } cb(cb_arg, ep); } @@ -259,8 +259,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *ep), ac->write_closure.cb_arg = ac; gpr_mu_lock(&ac->mu); - grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, - gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_init(&ac->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), + on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC)); grpc_fd_notify_on_write(ac->fd, &ac->write_closure); gpr_mu_unlock(&ac->mu); diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c index 32dd1ec11d..79a58fe2af 100644 --- a/src/core/iomgr/tcp_client_windows.c +++ b/src/core/iomgr/tcp_client_windows.c @@ -219,7 +219,7 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), ac->aborted = 0; grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, - gpr_now(GPR_CLOCK_REALTIME)); + gpr_now(GPR_CLOCK_MONOTONIC)); socket->write_info.outstanding = 1; grpc_socket_notify_on_write(socket, on_connect, ac); return; diff --git a/src/core/support/sync_posix.c b/src/core/support/sync_posix.c index 41af8ceb0a..61572b9a8e 100644 --- a/src/core/support/sync_posix.c +++ b/src/core/support/sync_posix.c @@ -63,10 +63,11 @@ void gpr_cv_destroy(gpr_cv *cv) { GPR_ASSERT(pthread_cond_destroy(cv) == 0); } int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) { int err = 0; - if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) { err = pthread_cond_wait(cv, mu); } else { struct timespec abs_deadline_ts; + abs_deadline = gpr_convert_clock_type(abs_deadline, GPR_CLOCK_REALTIME); abs_deadline_ts.tv_sec = abs_deadline.tv_sec; abs_deadline_ts.tv_nsec = abs_deadline.tv_nsec; err = pthread_cond_timedwait(cv, mu, &abs_deadline_ts); diff --git a/src/core/support/sync_win32.c b/src/core/support/sync_win32.c index 63196d10d3..54f84a46ac 100644 --- a/src/core/support/sync_win32.c +++ b/src/core/support/sync_win32.c @@ -83,10 +83,10 @@ int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) { int timeout = 0; DWORD timeout_max_ms; mu->locked = 0; - if (gpr_time_cmp(abs_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) == 0) { + if (gpr_time_cmp(abs_deadline, gpr_inf_future(abs_deadline.clock_type)) == 0) { SleepConditionVariableCS(cv, &mu->cs, INFINITE); } else { - gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); + gpr_timespec now = gpr_now(abs_deadline.clock_type); gpr_int64 now_ms = now.tv_sec * 1000 + now.tv_nsec / 1000000; gpr_int64 deadline_ms = abs_deadline.tv_sec * 1000 + abs_deadline.tv_nsec / 1000000; diff --git a/src/core/support/time.c b/src/core/support/time.c index 570f195bd1..b523ae01cc 100644 --- a/src/core/support/time.c +++ b/src/core/support/time.c @@ -290,3 +290,30 @@ gpr_int32 gpr_time_to_millis(gpr_timespec t) { double gpr_timespec_to_micros(gpr_timespec t) { return (double)t.tv_sec * GPR_US_PER_SEC + t.tv_nsec * 1e-3; } + +gpr_timespec gpr_convert_clock_type(gpr_timespec t, gpr_clock_type clock_type) { + if (t.clock_type == clock_type) { + return t; + } + + if (t.tv_nsec == 0) { + if (t.tv_sec == TYPE_MAX(time_t)) { + t.clock_type = clock_type; + return t; + } + if (t.tv_sec == TYPE_MIN(time_t)) { + t.clock_type = clock_type; + return t; + } + } + + if (clock_type == GPR_TIMESPAN) { + return gpr_time_sub(t, gpr_now(t.clock_type)); + } + + if (t.clock_type == GPR_TIMESPAN) { + return gpr_time_add(gpr_now(clock_type), t); + } + + return gpr_time_add(gpr_now(clock_type), gpr_time_sub(t, gpr_now(t.clock_type))); +} diff --git a/src/core/support/time_posix.c b/src/core/support/time_posix.c index 258b2e640e..841485c4b4 100644 --- a/src/core/support/time_posix.c +++ b/src/core/support/time_posix.c @@ -120,7 +120,7 @@ void gpr_sleep_until(gpr_timespec until) { for (;;) { /* We could simplify by using clock_nanosleep instead, but it might be * slightly less portable. */ - now = gpr_now(GPR_CLOCK_REALTIME); + now = gpr_now(until.clock_type); if (gpr_time_cmp(until, now) <= 0) { return; } diff --git a/src/core/support/time_win32.c b/src/core/support/time_win32.c index 238cd07ebc..7f64c80e27 100644 --- a/src/core/support/time_win32.c +++ b/src/core/support/time_win32.c @@ -80,7 +80,7 @@ void gpr_sleep_until(gpr_timespec until) { for (;;) { /* We could simplify by using clock_nanosleep instead, but it might be * slightly less portable. */ - now = gpr_now(GPR_CLOCK_REALTIME); + now = gpr_now(until.clock_type); if (gpr_time_cmp(until, now) <= 0) { return; } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index 2ba851da3d..aefcbad244 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -348,7 +348,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, } grpc_call_stack_init(channel_stack, server_transport_data, initial_op_ptr, CALL_STACK_FROM_CALL(call)); - if (gpr_time_cmp(send_deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { + if (gpr_time_cmp(send_deadline, gpr_inf_future(send_deadline.clock_type)) != 0) { set_deadline_alarm(call, send_deadline); } return call; @@ -1283,8 +1283,8 @@ static void set_deadline_alarm(grpc_call *call, gpr_timespec deadline) { } GRPC_CALL_INTERNAL_REF(call, "alarm"); call->have_alarm = 1; - grpc_alarm_init(&call->alarm, deadline, call_alarm, call, - gpr_now(GPR_CLOCK_REALTIME)); + grpc_alarm_init(&call->alarm, gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC), call_alarm, call, + gpr_now(GPR_CLOCK_MONOTONIC)); } /* we offset status by a small amount when storing it into transport metadata diff --git a/src/core/surface/call.h b/src/core/surface/call.h index 3b6f9c942e..265638d519 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -134,6 +134,10 @@ void grpc_server_log_request_call(char *file, int line, grpc_completion_queue *cq_for_notification, void *tag); +void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity, + grpc_server *server, grpc_completion_queue *cq, + void *tag); + /* Set a context pointer. No thread safety guarantees are made wrt this value. */ void grpc_call_context_set(grpc_call *call, grpc_context_index elem, @@ -151,6 +155,9 @@ void *grpc_call_context_get(grpc_call *call, grpc_context_index elem); grpc_server_log_request_call(sev, server, call, details, initial_metadata, \ cq_bound_to_call, cq_for_notifications, tag) +#define GRPC_SERVER_LOG_SHUTDOWN(sev, server, cq, tag) \ + if (grpc_trace_batch) grpc_server_log_shutdown(sev, server, cq, tag) + gpr_uint8 grpc_call_is_client(grpc_call *call); #endif /* GRPC_INTERNAL_CORE_SURFACE_CALL_H */ diff --git a/src/core/surface/call_log_batch.c b/src/core/surface/call_log_batch.c index 997046d954..7bf8cafc24 100644 --- a/src/core/surface/call_log_batch.c +++ b/src/core/surface/call_log_batch.c @@ -136,3 +136,11 @@ void grpc_server_log_request_call(char *file, int line, "tag=%p)", server, call, details, initial_metadata, cq_bound_to_call, cq_for_notification, tag); } + +void grpc_server_log_shutdown(char *file, int line, gpr_log_severity severity, + grpc_server *server, grpc_completion_queue *cq, + void *tag) { + gpr_log(file, line, severity, + "grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", server, + cq, tag); +} diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 8484418247..3f60b0b0ba 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -148,6 +148,8 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, gpr_timespec deadline) { grpc_event ret; + deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + GRPC_CQ_INTERNAL_REF(cc, "next"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { @@ -188,6 +190,8 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, grpc_cq_completion *c; grpc_cq_completion *prev; + deadline = gpr_convert_clock_type(deadline, GPR_CLOCK_MONOTONIC); + GRPC_CQ_INTERNAL_REF(cc, "pluck"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 5ba6f513c0..24f35298ef 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -981,6 +981,8 @@ void grpc_server_shutdown_and_notify(grpc_server *server, channel_broadcaster broadcaster; request_killer reqkill; + GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag); + /* lock, and gather up some stuff to do */ gpr_mu_lock(&server->mu_global); grpc_cq_begin_op(cq); diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index aa32f2e44a..904b9afce7 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -588,7 +588,7 @@ static void on_header(void *tp, grpc_mdelem *md) { GPR_ASSERT(stream_parsing); GRPC_CHTTP2_IF_TRACING(gpr_log( - GPR_INFO, "HTTP:%d:HDR: %s: %s", stream_parsing->id, + GPR_INFO, "HTTP:%d:HDR:%s: %s: %s", stream_parsing->id, transport_parsing->is_client ? "CLI" : "SVR", grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index d30059abf8..65b31a5afd 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -438,7 +438,7 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline, char timeout_str[GRPC_CHTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE]; grpc_mdelem *mdelem; grpc_chttp2_encode_timeout( - gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)), timeout_str); + gpr_time_sub(deadline, gpr_now(deadline.clock_type)), timeout_str); mdelem = grpc_mdelem_from_metadata_strings( c->mdctx, GRPC_MDSTR_REF(c->timeout_key_str), grpc_mdstr_from_string(c->mdctx, timeout_str)); @@ -560,6 +560,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, grpc_mdctx *mdctx = compressor->mdctx; grpc_linked_mdelem *l; int need_unref = 0; + gpr_timespec deadline; GPR_ASSERT(stream_id != 0); @@ -589,9 +590,9 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, l->md = hpack_enc(compressor, l->md, &st); need_unref |= l->md != NULL; } - if (gpr_time_cmp(op->data.metadata.deadline, - gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { - deadline_enc(compressor, op->data.metadata.deadline, &st); + deadline = op->data.metadata.deadline; + if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) { + deadline_enc(compressor, deadline, &st); } curop++; break; diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index 9d127c5472..10d796fc15 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -61,7 +61,7 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) { if (m != md.list.head) gpr_strvec_add(b, gpr_strdup(", ")); put_metadata(b, m->md); } - if (gpr_time_cmp(md.deadline, gpr_inf_future(GPR_CLOCK_REALTIME)) != 0) { + if (gpr_time_cmp(md.deadline, gpr_inf_future(md.deadline.clock_type)) != 0) { char *tmp; gpr_asprintf(&tmp, " deadline=%d.%09d", md.deadline.tv_sec, md.deadline.tv_nsec); diff --git a/src/cpp/client/channel_arguments.cc b/src/cpp/client/channel_arguments.cc index 4263e377a8..da6602e7af 100644 --- a/src/cpp/client/channel_arguments.cc +++ b/src/cpp/client/channel_arguments.cc @@ -33,10 +33,48 @@ #include <grpc++/channel_arguments.h> +#include <grpc/support/log.h> + #include "src/core/channel/channel_args.h" namespace grpc { +ChannelArguments::ChannelArguments(const ChannelArguments& other) + : strings_(other.strings_) { + args_.reserve(other.args_.size()); + auto list_it_dst = strings_.begin(); + auto list_it_src = other.strings_.begin(); + for (auto a = other.args_.begin(); a != other.args_.end(); ++a) { + grpc_arg ap; + ap.type = a->type; + GPR_ASSERT(list_it_src->c_str() == a->key); + ap.key = const_cast<char*>(list_it_dst->c_str()); + ++list_it_src; + ++list_it_dst; + switch (a->type) { + case GRPC_ARG_INTEGER: + ap.value.integer = a->value.integer; + break; + case GRPC_ARG_STRING: + GPR_ASSERT(list_it_src->c_str() == a->value.string); + ap.value.string = const_cast<char*>(list_it_dst->c_str()); + ++list_it_src; + ++list_it_dst; + break; + case GRPC_ARG_POINTER: + ap.value.pointer = a->value.pointer; + ap.value.pointer.p = a->value.pointer.copy(ap.value.pointer.p); + break; + } + args_.push_back(ap); + } +} + +void ChannelArguments::Swap(ChannelArguments& other) { + args_.swap(other.args_); + strings_.swap(other.strings_); +} + void ChannelArguments::SetCompressionAlgorithm( grpc_compression_algorithm algorithm) { SetInt(GRPC_COMPRESSION_ALGORITHM_ARG, algorithm); diff --git a/src/cpp/client/create_channel.cc b/src/cpp/client/create_channel.cc index d85daabd20..62f179d361 100644 --- a/src/cpp/client/create_channel.cc +++ b/src/cpp/client/create_channel.cc @@ -32,9 +32,11 @@ */ #include <memory> +#include <sstream> #include "src/cpp/client/channel.h" #include <grpc++/channel_interface.h> +#include <grpc++/channel_arguments.h> #include <grpc++/create_channel.h> namespace grpc { @@ -43,7 +45,12 @@ class ChannelArguments; std::shared_ptr<ChannelInterface> CreateChannel( const grpc::string& target, const std::shared_ptr<Credentials>& creds, const ChannelArguments& args) { - return creds ? creds->CreateChannel(target, args) + ChannelArguments cp_args = args; + std::ostringstream user_agent_prefix; + user_agent_prefix << "grpc-c++/" << grpc_version_string(); + cp_args.SetString(GRPC_ARG_PRIMARY_USER_AGENT_STRING, + user_agent_prefix.str()); + return creds ? creds->CreateChannel(target, cp_args) : std::shared_ptr<ChannelInterface>( new Channel(target, grpc_lame_client_channel_create(NULL))); } diff --git a/src/csharp/Grpc.Auth/GoogleCredential.cs b/src/csharp/Grpc.Auth/GoogleCredential.cs index 8d5e543a21..7edf19ed67 100644 --- a/src/csharp/Grpc.Auth/GoogleCredential.cs +++ b/src/csharp/Grpc.Auth/GoogleCredential.cs @@ -35,8 +35,11 @@ using System; using System.Collections.Generic; using System.IO; using System.Security.Cryptography; +using System.Threading; +using System.Threading.Tasks; using Google.Apis.Auth.OAuth2; +using Google.Apis.Auth.OAuth2.Responses; using Newtonsoft.Json.Linq; using Org.BouncyCastle.Crypto.Parameters; using Org.BouncyCastle.Security; @@ -100,6 +103,19 @@ namespace Grpc.Auth return new GoogleCredential(serviceCredential); } + public Task<bool> RequestAccessTokenAsync(CancellationToken taskCancellationToken) + { + return credential.RequestAccessTokenAsync(taskCancellationToken); + } + + public TokenResponse Token + { + get + { + return credential.Token; + } + } + internal ServiceCredential InternalCredential { get diff --git a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs index e797dd82f2..8ba2c8a9a2 100644 --- a/src/csharp/Grpc.Core.Tests/ClientServerTest.cs +++ b/src/csharp/Grpc.Core.Tests/ClientServerTest.cs @@ -33,6 +33,7 @@ using System; using System.Diagnostics; +using System.Linq; using System.Threading; using System.Threading.Tasks; using Grpc.Core; @@ -99,17 +100,17 @@ namespace Grpc.Core.Tests [Test] public void UnaryCall() { - var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); - Assert.AreEqual("ABC", Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None)); + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + Assert.AreEqual("ABC", Calls.BlockingUnaryCall(internalCall, "ABC", CancellationToken.None)); } [Test] public void UnaryCall_ServerHandlerThrows() { - var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); try { - Calls.BlockingUnaryCall(call, "THROW", CancellationToken.None); + Calls.BlockingUnaryCall(internalCall, "THROW", CancellationToken.None); Assert.Fail(); } catch (RpcException e) @@ -119,10 +120,40 @@ namespace Grpc.Core.Tests } [Test] + public void UnaryCall_ServerHandlerThrowsRpcException() + { + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + try + { + Calls.BlockingUnaryCall(internalCall, "THROW_UNAUTHENTICATED", CancellationToken.None); + Assert.Fail(); + } + catch (RpcException e) + { + Assert.AreEqual(StatusCode.Unauthenticated, e.Status.StatusCode); + } + } + + [Test] + public void UnaryCall_ServerHandlerSetsStatus() + { + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + try + { + Calls.BlockingUnaryCall(internalCall, "SET_UNAUTHENTICATED", CancellationToken.None); + Assert.Fail(); + } + catch (RpcException e) + { + Assert.AreEqual(StatusCode.Unauthenticated, e.Status.StatusCode); + } + } + + [Test] public void AsyncUnaryCall() { - var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); - var result = Calls.AsyncUnaryCall(call, "ABC", CancellationToken.None).Result; + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + var result = Calls.AsyncUnaryCall(internalCall, "ABC", CancellationToken.None).ResponseAsync.Result; Assert.AreEqual("ABC", result); } @@ -131,10 +162,10 @@ namespace Grpc.Core.Tests { Task.Run(async () => { - var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); try { - await Calls.AsyncUnaryCall(call, "THROW", CancellationToken.None); + await Calls.AsyncUnaryCall(internalCall, "THROW", CancellationToken.None); Assert.Fail(); } catch (RpcException e) @@ -149,11 +180,11 @@ namespace Grpc.Core.Tests { Task.Run(async () => { - var call = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); - var callResult = Calls.AsyncClientStreamingCall(call, CancellationToken.None); + var internalCall = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); + var call = Calls.AsyncClientStreamingCall(internalCall, CancellationToken.None); - await callResult.RequestStream.WriteAll(new string[] { "A", "B", "C" }); - Assert.AreEqual("ABC", await callResult.Result); + await call.RequestStream.WriteAll(new string[] { "A", "B", "C" }); + Assert.AreEqual("ABC", await call.ResponseAsync); }).Wait(); } @@ -162,10 +193,10 @@ namespace Grpc.Core.Tests { Task.Run(async () => { - var call = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); + var internalCall = new Call<string, string>(ServiceName, ConcatAndEchoMethod, channel, Metadata.Empty); var cts = new CancellationTokenSource(); - var callResult = Calls.AsyncClientStreamingCall(call, cts.Token); + var call = Calls.AsyncClientStreamingCall(internalCall, cts.Token); // TODO(jtattermusch): we need this to ensure call has been initiated once we cancel it. await Task.Delay(1000); @@ -173,7 +204,7 @@ namespace Grpc.Core.Tests try { - await callResult.Result; + await call.ResponseAsync; } catch (RpcException e) { @@ -183,29 +214,53 @@ namespace Grpc.Core.Tests } [Test] + public void AsyncUnaryCall_EchoMetadata() + { + var headers = new Metadata + { + new Metadata.Entry("asciiHeader", "abcdefg"), + new Metadata.Entry("binaryHeader-bin", new byte[] { 1, 2, 3, 0, 0xff }), + }; + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, headers); + var call = Calls.AsyncUnaryCall(internalCall, "ABC", CancellationToken.None); + + Assert.AreEqual("ABC", call.ResponseAsync.Result); + + Assert.AreEqual(StatusCode.OK, call.GetStatus().StatusCode); + + var trailers = call.GetTrailers(); + Assert.AreEqual(2, trailers.Count); + Assert.AreEqual(headers[0].Key, trailers[0].Key); + Assert.AreEqual(headers[0].Value, trailers[0].Value); + + Assert.AreEqual(headers[1].Key, trailers[1].Key); + CollectionAssert.AreEqual(headers[1].ValueBytes, trailers[1].ValueBytes); + } + + [Test] public void UnaryCall_DisposedChannel() { channel.Dispose(); - var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); - Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(call, "ABC", CancellationToken.None)); + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + Assert.Throws(typeof(ObjectDisposedException), () => Calls.BlockingUnaryCall(internalCall, "ABC", CancellationToken.None)); } [Test] public void UnaryCallPerformance() { - var call = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); BenchmarkUtil.RunBenchmark(100, 100, - () => { Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); }); + () => { Calls.BlockingUnaryCall(internalCall, "ABC", default(CancellationToken)); }); } [Test] public void UnknownMethodHandler() { - var call = new Call<string, string>(ServiceName, NonexistentMethod, channel, Metadata.Empty); + var internalCall = new Call<string, string>(ServiceName, NonexistentMethod, channel, Metadata.Empty); try { - Calls.BlockingUnaryCall(call, "ABC", default(CancellationToken)); + Calls.BlockingUnaryCall(internalCall, "ABC", default(CancellationToken)); Assert.Fail(); } catch (RpcException e) @@ -214,16 +269,48 @@ namespace Grpc.Core.Tests } } - private static async Task<string> EchoHandler(ServerCallContext context, string request) + [Test] + public void UserAgentStringPresent() { + var internalCall = new Call<string, string>(ServiceName, EchoMethod, channel, Metadata.Empty); + string userAgent = Calls.BlockingUnaryCall(internalCall, "RETURN-USER-AGENT", CancellationToken.None); + Assert.IsTrue(userAgent.StartsWith("grpc-csharp/")); + } + + private static async Task<string> EchoHandler(string request, ServerCallContext context) + { + foreach (Metadata.Entry metadataEntry in context.RequestHeaders) + { + if (metadataEntry.Key != "user-agent") + { + context.ResponseTrailers.Add(metadataEntry); + } + } + + if (request == "RETURN-USER-AGENT") + { + return context.RequestHeaders.Where(entry => entry.Key == "user-agent").Single().Value; + } + if (request == "THROW") { throw new Exception("This was thrown on purpose by a test"); } + + if (request == "THROW_UNAUTHENTICATED") + { + throw new RpcException(new Status(StatusCode.Unauthenticated, "")); + } + + if (request == "SET_UNAUTHENTICATED") + { + context.Status = new Status(StatusCode.Unauthenticated, ""); + } + return request; } - private static async Task<string> ConcatAndEchoHandler(ServerCallContext context, IAsyncStreamReader<string> requestStream) + private static async Task<string> ConcatAndEchoHandler(IAsyncStreamReader<string> requestStream, ServerCallContext context) { string result = ""; await requestStream.ForEach(async (request) => diff --git a/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs b/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs index 320423b245..46469113c5 100644 --- a/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs +++ b/src/csharp/Grpc.Core.Tests/Internal/MetadataArraySafeHandleTest.cs @@ -51,12 +51,34 @@ namespace Grpc.Core.Internal.Tests [Test] public void CreateAndDestroy() { - var metadata = new Metadata { + var metadata = new Metadata + { new Metadata.Entry("host", "somehost"), new Metadata.Entry("header2", "header value"), }; var nativeMetadata = MetadataArraySafeHandle.Create(metadata); nativeMetadata.Dispose(); } + + [Test] + public void ReadMetadataFromPtrUnsafe() + { + var metadata = new Metadata + { + new Metadata.Entry("host", "somehost"), + new Metadata.Entry("header2", "header value"), + }; + var nativeMetadata = MetadataArraySafeHandle.Create(metadata); + + var copy = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(nativeMetadata.Handle); + Assert.AreEqual(2, copy.Count); + + Assert.AreEqual("host", copy[0].Key); + Assert.AreEqual("somehost", copy[0].Value); + Assert.AreEqual("header2", copy[1].Key); + Assert.AreEqual("header value", copy[1].Value); + + nativeMetadata.Dispose(); + } } } diff --git a/src/csharp/Grpc.Core.Tests/TimespecTest.cs b/src/csharp/Grpc.Core.Tests/TimespecTest.cs index 5831121add..a34b407a01 100644 --- a/src/csharp/Grpc.Core.Tests/TimespecTest.cs +++ b/src/csharp/Grpc.Core.Tests/TimespecTest.cs @@ -59,6 +59,19 @@ namespace Grpc.Core.Internal.Tests } [Test] + public void ToDateTime() + { + Assert.AreEqual(new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc), + new Timespec(IntPtr.Zero, 0).ToDateTime()); + + Assert.AreEqual(new DateTime(1970, 1, 1, 0, 0, 10, DateTimeKind.Utc).AddTicks(50), + new Timespec(new IntPtr(10), 5000).ToDateTime()); + + Assert.AreEqual(new DateTime(2015, 7, 21, 4, 21, 48, DateTimeKind.Utc), + new Timespec(new IntPtr(1437452508), 0).ToDateTime()); + } + + [Test] public void Add() { var t = new Timespec { tv_sec = new IntPtr(12345), tv_nsec = 123456789 }; diff --git a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs index d66b0d4974..bf020cd627 100644 --- a/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs +++ b/src/csharp/Grpc.Core/AsyncClientStreamingCall.cs @@ -43,24 +43,28 @@ namespace Grpc.Core public sealed class AsyncClientStreamingCall<TRequest, TResponse> : IDisposable { readonly IClientStreamWriter<TRequest> requestStream; - readonly Task<TResponse> result; + readonly Task<TResponse> responseAsync; + readonly Func<Status> getStatusFunc; + readonly Func<Metadata> getTrailersFunc; readonly Action disposeAction; - public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> result, Action disposeAction) + public AsyncClientStreamingCall(IClientStreamWriter<TRequest> requestStream, Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction) { this.requestStream = requestStream; - this.result = result; + this.responseAsync = responseAsync; + this.getStatusFunc = getStatusFunc; + this.getTrailersFunc = getTrailersFunc; this.disposeAction = disposeAction; } /// <summary> /// Asynchronous call result. /// </summary> - public Task<TResponse> Result + public Task<TResponse> ResponseAsync { get { - return this.result; + return this.responseAsync; } } @@ -81,11 +85,11 @@ namespace Grpc.Core /// <returns></returns> public TaskAwaiter<TResponse> GetAwaiter() { - return result.GetAwaiter(); + return responseAsync.GetAwaiter(); } /// <summary> - /// Provides means to provide after the call. + /// Provides means to cleanup after the call. /// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything. /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call. /// As a result, all resources being used by the call should be released eventually. diff --git a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs index 4c0d5936ac..0979de606f 100644 --- a/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs +++ b/src/csharp/Grpc.Core/AsyncDuplexStreamingCall.cs @@ -44,12 +44,16 @@ namespace Grpc.Core { readonly IClientStreamWriter<TRequest> requestStream; readonly IAsyncStreamReader<TResponse> responseStream; + readonly Func<Status> getStatusFunc; + readonly Func<Metadata> getTrailersFunc; readonly Action disposeAction; - public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Action disposeAction) + public AsyncDuplexStreamingCall(IClientStreamWriter<TRequest> requestStream, IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction) { this.requestStream = requestStream; this.responseStream = responseStream; + this.getStatusFunc = getStatusFunc; + this.getTrailersFunc = getTrailersFunc; this.disposeAction = disposeAction; } @@ -76,6 +80,24 @@ namespace Grpc.Core } /// <summary> + /// Gets the call status if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Status GetStatus() + { + return getStatusFunc(); + } + + /// <summary> + /// Gets the call trailing metadata if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Metadata GetTrailers() + { + return getTrailersFunc(); + } + + /// <summary> /// Provides means to cleanup after the call. /// If the call has already finished normally (request stream has been completed and response stream has been fully read), doesn't do anything. /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call. diff --git a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs index 7a479b9a23..380efcdb0e 100644 --- a/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs +++ b/src/csharp/Grpc.Core/AsyncServerStreamingCall.cs @@ -43,11 +43,15 @@ namespace Grpc.Core public sealed class AsyncServerStreamingCall<TResponse> : IDisposable { readonly IAsyncStreamReader<TResponse> responseStream; + readonly Func<Status> getStatusFunc; + readonly Func<Metadata> getTrailersFunc; readonly Action disposeAction; - public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Action disposeAction) + public AsyncServerStreamingCall(IAsyncStreamReader<TResponse> responseStream, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction) { this.responseStream = responseStream; + this.getStatusFunc = getStatusFunc; + this.getTrailersFunc = getTrailersFunc; this.disposeAction = disposeAction; } @@ -63,6 +67,24 @@ namespace Grpc.Core } /// <summary> + /// Gets the call status if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Status GetStatus() + { + return getStatusFunc(); + } + + /// <summary> + /// Gets the call trailing metadata if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Metadata GetTrailers() + { + return getTrailersFunc(); + } + + /// <summary> /// Provides means to cleanup after the call. /// If the call has already finished normally (response stream has been fully read), doesn't do anything. /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call. diff --git a/src/csharp/Grpc.Core/AsyncUnaryCall.cs b/src/csharp/Grpc.Core/AsyncUnaryCall.cs new file mode 100644 index 0000000000..224e343916 --- /dev/null +++ b/src/csharp/Grpc.Core/AsyncUnaryCall.cs @@ -0,0 +1,106 @@ +#region Copyright notice and license + +// Copyright 2015, 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. + +#endregion + +using System; +using System.Runtime.CompilerServices; +using System.Threading.Tasks; + +namespace Grpc.Core +{ + /// <summary> + /// Return type for single request - single response call. + /// </summary> + public sealed class AsyncUnaryCall<TResponse> : IDisposable + { + readonly Task<TResponse> responseAsync; + readonly Func<Status> getStatusFunc; + readonly Func<Metadata> getTrailersFunc; + readonly Action disposeAction; + + public AsyncUnaryCall(Task<TResponse> responseAsync, Func<Status> getStatusFunc, Func<Metadata> getTrailersFunc, Action disposeAction) + { + this.responseAsync = responseAsync; + this.getStatusFunc = getStatusFunc; + this.getTrailersFunc = getTrailersFunc; + this.disposeAction = disposeAction; + } + + /// <summary> + /// Asynchronous call result. + /// </summary> + public Task<TResponse> ResponseAsync + { + get + { + return this.responseAsync; + } + } + + /// <summary> + /// Allows awaiting this object directly. + /// </summary> + public TaskAwaiter<TResponse> GetAwaiter() + { + return responseAsync.GetAwaiter(); + } + + /// <summary> + /// Gets the call status if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Status GetStatus() + { + return getStatusFunc(); + } + + /// <summary> + /// Gets the call trailing metadata if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Metadata GetTrailers() + { + return getTrailersFunc(); + } + + /// <summary> + /// Provides means to cleanup after the call. + /// If the call has already finished normally (request stream has been completed and call result has been received), doesn't do anything. + /// Otherwise, requests cancellation of the call which should terminate all pending async operations associated with the call. + /// As a result, all resources being used by the call should be released eventually. + /// </summary> + public void Dispose() + { + disposeAction.Invoke(); + } + } +} diff --git a/src/csharp/Grpc.Core/Calls.cs b/src/csharp/Grpc.Core/Calls.cs index 9e95182c72..359fe53741 100644 --- a/src/csharp/Grpc.Core/Calls.cs +++ b/src/csharp/Grpc.Core/Calls.cs @@ -53,7 +53,7 @@ namespace Grpc.Core return asyncCall.UnaryCall(call.Channel, call.Name, req, call.Headers); } - public static async Task<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token) + public static AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token) where TRequest : class where TResponse : class { @@ -61,7 +61,7 @@ namespace Grpc.Core asyncCall.Initialize(call.Channel, call.Channel.CompletionQueue, call.Name); var asyncResult = asyncCall.UnaryCallAsync(req, call.Headers); RegisterCancellationCallback(asyncCall, token); - return await asyncResult; + return new AsyncUnaryCall<TResponse>(asyncResult, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); } public static AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, TRequest req, CancellationToken token) @@ -73,7 +73,7 @@ namespace Grpc.Core asyncCall.StartServerStreamingCall(req, call.Headers); RegisterCancellationCallback(asyncCall, token); var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall); - return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.Cancel); + return new AsyncServerStreamingCall<TResponse>(responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); } public static AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token) @@ -85,7 +85,7 @@ namespace Grpc.Core var resultTask = asyncCall.ClientStreamingCallAsync(call.Headers); RegisterCancellationCallback(asyncCall, token); var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall); - return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.Cancel); + return new AsyncClientStreamingCall<TRequest, TResponse>(requestStream, resultTask, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); } public static AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Call<TRequest, TResponse> call, CancellationToken token) @@ -98,7 +98,7 @@ namespace Grpc.Core RegisterCancellationCallback(asyncCall, token); var requestStream = new ClientRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ClientResponseStream<TRequest, TResponse>(asyncCall); - return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.Cancel); + return new AsyncDuplexStreamingCall<TRequest, TResponse>(requestStream, responseStream, asyncCall.GetStatus, asyncCall.GetTrailers, asyncCall.Cancel); } private static void RegisterCancellationCallback<TRequest, TResponse>(AsyncCall<TRequest, TResponse> asyncCall, CancellationToken token) diff --git a/src/csharp/Grpc.Core/Channel.cs b/src/csharp/Grpc.Core/Channel.cs index 5baf260003..e5c6abd2cb 100644 --- a/src/csharp/Grpc.Core/Channel.cs +++ b/src/csharp/Grpc.Core/Channel.cs @@ -28,11 +28,14 @@ // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #endregion + using System; using System.Collections.Generic; +using System.Linq; using System.Runtime.InteropServices; using System.Threading; using System.Threading.Tasks; + using Grpc.Core.Internal; namespace Grpc.Core @@ -44,6 +47,7 @@ namespace Grpc.Core { readonly GrpcEnvironment environment; readonly ChannelSafeHandle handle; + readonly List<ChannelOption> options; readonly string target; bool disposed; @@ -57,7 +61,10 @@ namespace Grpc.Core public Channel(string host, Credentials credentials = null, IEnumerable<ChannelOption> options = null) { this.environment = GrpcEnvironment.GetInstance(); - using (ChannelArgsSafeHandle nativeChannelArgs = ChannelOptions.CreateChannelArgs(options)) + this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>(); + + EnsureUserAgentChannelOption(this.options); + using (ChannelArgsSafeHandle nativeChannelArgs = ChannelOptions.CreateChannelArgs(this.options)) { if (credentials != null) { @@ -71,7 +78,7 @@ namespace Grpc.Core this.handle = ChannelSafeHandle.Create(host, nativeChannelArgs); } } - this.target = GetOverridenTarget(host, options); + this.target = GetOverridenTarget(host, this.options); } /// <summary> @@ -141,6 +148,20 @@ namespace Grpc.Core } } + private static void EnsureUserAgentChannelOption(List<ChannelOption> options) + { + if (!options.Any((option) => option.Name == ChannelOptions.PrimaryUserAgentString)) + { + options.Add(new ChannelOption(ChannelOptions.PrimaryUserAgentString, GetUserAgentString())); + } + } + + private static string GetUserAgentString() + { + // TODO(jtattermusch): it would be useful to also provide .NET/mono version. + return string.Format("grpc-csharp/{0}", VersionInfo.CurrentVersion); + } + /// <summary> /// Look for SslTargetNameOverride option and return its value instead of originalTarget /// if found. diff --git a/src/csharp/Grpc.Core/ChannelOptions.cs b/src/csharp/Grpc.Core/ChannelOptions.cs index bc23bb59b1..9fe03d2805 100644 --- a/src/csharp/Grpc.Core/ChannelOptions.cs +++ b/src/csharp/Grpc.Core/ChannelOptions.cs @@ -115,41 +115,49 @@ namespace Grpc.Core } } + /// <summary> + /// Defines names of supported channel options. + /// </summary> public static class ChannelOptions { - // Override SSL target check. Only to be used for testing. + /// <summary>Override SSL target check. Only to be used for testing.</summary> public const string SslTargetNameOverride = "grpc.ssl_target_name_override"; - // Enable census for tracing and stats collection + /// <summary>Enable census for tracing and stats collection</summary> public const string Census = "grpc.census"; - // Maximum number of concurrent incoming streams to allow on a http2 connection + /// <summary>Maximum number of concurrent incoming streams to allow on a http2 connection</summary> public const string MaxConcurrentStreams = "grpc.max_concurrent_streams"; - // Maximum message length that the channel can receive + /// <summary>Maximum message length that the channel can receive</summary> public const string MaxMessageLength = "grpc.max_message_length"; - // Initial sequence number for http2 transports + /// <summary>Initial sequence number for http2 transports</summary> public const string Http2InitialSequenceNumber = "grpc.http2.initial_sequence_number"; + /// <summary>Primary user agent: goes at the start of the user-agent metadata</summary> + public const string PrimaryUserAgentString = "grpc.primary_user_agent"; + + /// <summary> Secondary user agent: goes at the end of the user-agent metadata</summary> + public const string SecondaryUserAgentString = "grpc.secondary_user_agent"; + /// <summary> /// Creates native object for a collection of channel options. /// </summary> /// <returns>The native channel arguments.</returns> - internal static ChannelArgsSafeHandle CreateChannelArgs(IEnumerable<ChannelOption> options) + internal static ChannelArgsSafeHandle CreateChannelArgs(List<ChannelOption> options) { - if (options == null) + if (options == null || options.Count == 0) { return ChannelArgsSafeHandle.CreateNull(); } - var optionList = new List<ChannelOption>(options); // It's better to do defensive copy ChannelArgsSafeHandle nativeArgs = null; try { - nativeArgs = ChannelArgsSafeHandle.Create(optionList.Count); - for (int i = 0; i < optionList.Count; i++) + nativeArgs = ChannelArgsSafeHandle.Create(options.Count); + for (int i = 0; i < options.Count; i++) { - var option = optionList[i]; + var option = options[i]; if (option.Type == ChannelOption.OptionType.Integer) { nativeArgs.SetInteger(i, option.Name, option.IntValue); diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index a227fe5477..fd68b91851 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -33,13 +33,12 @@ </PropertyGroup> <ItemGroup> <Reference Include="System" /> - <Reference Include="System.Collections.Immutable, Version=1.1.36.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL"> - <SpecificVersion>False</SpecificVersion> - <HintPath>..\packages\System.Collections.Immutable.1.1.36\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath> - </Reference> <Reference Include="System.Interactive.Async"> <HintPath>..\packages\Ix-Async.1.2.3\lib\net45\System.Interactive.Async.dll</HintPath> </Reference> + <Reference Include="System.Collections.Immutable"> + <HintPath>..\packages\System.Collections.Immutable.1.1.36\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath> + </Reference> </ItemGroup> <ItemGroup> <Compile Include="AsyncDuplexStreamingCall.cs" /> @@ -102,6 +101,8 @@ <Compile Include="Internal\CompletionRegistry.cs" /> <Compile Include="Internal\BatchContextSafeHandle.cs" /> <Compile Include="ChannelOptions.cs" /> + <Compile Include="AsyncUnaryCall.cs" /> + <Compile Include="VersionInfo.cs" /> </ItemGroup> <ItemGroup> <None Include="Grpc.Core.nuspec" /> diff --git a/src/csharp/Grpc.Core/Internal/AsyncCall.cs b/src/csharp/Grpc.Core/Internal/AsyncCall.cs index 24b75d1668..f983dbb759 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCall.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCall.cs @@ -52,8 +52,8 @@ namespace Grpc.Core.Internal // Completion of a pending unary response if not null. TaskCompletionSource<TResponse> unaryResponseTcs; - // Set after status is received. Only used for streaming response calls. - Status? finishedStatus; + // Set after status is received. Used for both unary and streaming response calls. + ClientSideStatus? finishedStatus; bool readObserverCompleted; // True if readObserver has already been completed. @@ -249,6 +249,32 @@ namespace Grpc.Core.Internal } /// <summary> + /// Gets the resulting status if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Status GetStatus() + { + lock (myLock) + { + Preconditions.CheckState(finishedStatus.HasValue, "Status can only be accessed once the call has finished."); + return finishedStatus.Value.Status; + } + } + + /// <summary> + /// Gets the trailing metadata if the call has already finished. + /// Throws InvalidOperationException otherwise. + /// </summary> + public Metadata GetTrailers() + { + lock (myLock) + { + Preconditions.CheckState(finishedStatus.HasValue, "Trailers can only be accessed once the call has finished."); + return finishedStatus.Value.Trailers; + } + } + + /// <summary> /// On client-side, we only fire readCompletionDelegate once all messages have been read /// and status has been received. /// </summary> @@ -265,7 +291,7 @@ namespace Grpc.Core.Internal if (shouldComplete) { - var status = finishedStatus.Value; + var status = finishedStatus.Value.Status; if (status.StatusCode != StatusCode.OK) { FireCompletion(completionDelegate, default(TResponse), new RpcException(status)); @@ -288,9 +314,13 @@ namespace Grpc.Core.Internal /// </summary> private void HandleUnaryResponse(bool success, BatchContextSafeHandle ctx) { + var fullStatus = ctx.GetReceivedStatusOnClient(); + lock (myLock) { finished = true; + finishedStatus = fullStatus; + halfclosed = true; ReleaseResourcesIfPossible(); @@ -302,7 +332,8 @@ namespace Grpc.Core.Internal return; } - var status = ctx.GetReceivedStatus(); + var status = fullStatus.Status; + if (status.StatusCode != StatusCode.OK) { unaryResponseTcs.SetException(new RpcException(status)); @@ -321,13 +352,13 @@ namespace Grpc.Core.Internal /// </summary> private void HandleFinished(bool success, BatchContextSafeHandle ctx) { - var status = ctx.GetReceivedStatus(); + var fullStatus = ctx.GetReceivedStatusOnClient(); AsyncCompletionDelegate<TResponse> origReadCompletionDelegate = null; lock (myLock) { finished = true; - finishedStatus = status; + finishedStatus = fullStatus; origReadCompletionDelegate = readCompletionDelegate; diff --git a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs index 309067ea9d..f809f4a84c 100644 --- a/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs +++ b/src/csharp/Grpc.Core/Internal/AsyncCallServer.cs @@ -101,14 +101,17 @@ namespace Grpc.Core.Internal /// Only one pending send action is allowed at any given time. /// completionDelegate is called when the operation finishes. /// </summary> - public void StartSendStatusFromServer(Status status, AsyncCompletionDelegate<object> completionDelegate) + public void StartSendStatusFromServer(Status status, Metadata trailers, AsyncCompletionDelegate<object> completionDelegate) { lock (myLock) { Preconditions.CheckNotNull(completionDelegate, "Completion delegate cannot be null"); CheckSendingAllowed(); - call.StartSendStatusFromServer(status, HandleHalfclosed); + using (var metadataArray = MetadataArraySafeHandle.Create(trailers)) + { + call.StartSendStatusFromServer(status, HandleHalfclosed, metadataArray); + } halfcloseRequested = true; readingDone = true; sendCompletionDelegate = completionDelegate; diff --git a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs index 861cbbe4c6..6a2add54db 100644 --- a/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/BatchContextSafeHandle.cs @@ -38,7 +38,6 @@ using Grpc.Core; namespace Grpc.Core.Internal { /// <summary> - /// Not owned version of /// grpcsharp_batch_context /// </summary> internal class BatchContextSafeHandle : SafeHandleZeroIsInvalid @@ -47,6 +46,9 @@ namespace Grpc.Core.Internal static extern BatchContextSafeHandle grpcsharp_batch_context_create(); [DllImport("grpc_csharp_ext.dll")] + static extern IntPtr grpcsharp_batch_context_recv_initial_metadata(BatchContextSafeHandle ctx); + + [DllImport("grpc_csharp_ext.dll")] static extern IntPtr grpcsharp_batch_context_recv_message_length(BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] @@ -59,12 +61,24 @@ namespace Grpc.Core.Internal static extern IntPtr grpcsharp_batch_context_recv_status_on_client_details(BatchContextSafeHandle ctx); // returns const char* [DllImport("grpc_csharp_ext.dll")] + static extern IntPtr grpcsharp_batch_context_recv_status_on_client_trailing_metadata(BatchContextSafeHandle ctx); + + [DllImport("grpc_csharp_ext.dll")] static extern CallSafeHandle grpcsharp_batch_context_server_rpc_new_call(BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] static extern IntPtr grpcsharp_batch_context_server_rpc_new_method(BatchContextSafeHandle ctx); // returns const char* [DllImport("grpc_csharp_ext.dll")] + static extern IntPtr grpcsharp_batch_context_server_rpc_new_host(BatchContextSafeHandle ctx); // returns const char* + + [DllImport("grpc_csharp_ext.dll")] + static extern Timespec grpcsharp_batch_context_server_rpc_new_deadline(BatchContextSafeHandle ctx); + + [DllImport("grpc_csharp_ext.dll")] + static extern IntPtr grpcsharp_batch_context_server_rpc_new_request_metadata(BatchContextSafeHandle ctx); + + [DllImport("grpc_csharp_ext.dll")] static extern int grpcsharp_batch_context_recv_close_on_server_cancelled(BatchContextSafeHandle ctx); [DllImport("grpc_csharp_ext.dll")] @@ -87,13 +101,26 @@ namespace Grpc.Core.Internal } } - public Status GetReceivedStatus() + // Gets data of recv_initial_metadata completion. + public Metadata GetReceivedInitialMetadata() + { + IntPtr metadataArrayPtr = grpcsharp_batch_context_recv_initial_metadata(this); + return MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); + } + + // Gets data of recv_status_on_client completion. + public ClientSideStatus GetReceivedStatusOnClient() { - // TODO: can the native method return string directly? string details = Marshal.PtrToStringAnsi(grpcsharp_batch_context_recv_status_on_client_details(this)); - return new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details); + var status = new Status(grpcsharp_batch_context_recv_status_on_client_status(this), details); + + IntPtr metadataArrayPtr = grpcsharp_batch_context_recv_status_on_client_trailing_metadata(this); + var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); + + return new ClientSideStatus(status, metadata); } + // Gets data of recv_message completion. public byte[] GetReceivedMessage() { IntPtr len = grpcsharp_batch_context_recv_message_length(this); @@ -106,16 +133,22 @@ namespace Grpc.Core.Internal return data; } - public CallSafeHandle GetServerRpcNewCall() + // Gets data of server_rpc_new completion. + public ServerRpcNew GetServerRpcNew() { - return grpcsharp_batch_context_server_rpc_new_call(this); - } + var call = grpcsharp_batch_context_server_rpc_new_call(this); - public string GetServerRpcNewMethod() - { - return Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this)); + var method = Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_method(this)); + var host = Marshal.PtrToStringAnsi(grpcsharp_batch_context_server_rpc_new_host(this)); + var deadline = grpcsharp_batch_context_server_rpc_new_deadline(this); + + IntPtr metadataArrayPtr = grpcsharp_batch_context_server_rpc_new_request_metadata(this); + var metadata = MetadataArraySafeHandle.ReadMetadataFromPtrUnsafe(metadataArrayPtr); + + return new ServerRpcNew(call, method, host, deadline, metadata); } + // Gets data of receive_close_on_server completion. public bool GetReceivedCloseOnServerCancelled() { return grpcsharp_batch_context_recv_close_on_server_cancelled(this) != 0; @@ -127,4 +160,97 @@ namespace Grpc.Core.Internal return true; } } + + /// <summary> + /// Status + metadata received on client side when call finishes. + /// (when receive_status_on_client operation finishes). + /// </summary> + internal struct ClientSideStatus + { + readonly Status status; + readonly Metadata trailers; + + public ClientSideStatus(Status status, Metadata trailers) + { + this.status = status; + this.trailers = trailers; + } + + public Status Status + { + get + { + return this.status; + } + } + + public Metadata Trailers + { + get + { + return this.trailers; + } + } + } + + /// <summary> + /// Details of a newly received RPC. + /// </summary> + internal struct ServerRpcNew + { + readonly CallSafeHandle call; + readonly string method; + readonly string host; + readonly Timespec deadline; + readonly Metadata requestMetadata; + + public ServerRpcNew(CallSafeHandle call, string method, string host, Timespec deadline, Metadata requestMetadata) + { + this.call = call; + this.method = method; + this.host = host; + this.deadline = deadline; + this.requestMetadata = requestMetadata; + } + + public CallSafeHandle Call + { + get + { + return this.call; + } + } + + public string Method + { + get + { + return this.method; + } + } + + public string Host + { + get + { + return this.host; + } + } + + public Timespec Deadline + { + get + { + return this.deadline; + } + } + + public Metadata RequestMetadata + { + get + { + return this.requestMetadata; + } + } + } }
\ No newline at end of file diff --git a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs index 3b246ac01b..19dbb83f24 100644 --- a/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/CallSafeHandle.cs @@ -81,7 +81,7 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_send_status_from_server(CallSafeHandle call, - BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage); + BatchContextSafeHandle ctx, StatusCode statusCode, string statusMessage, MetadataArraySafeHandle metadataArray); [DllImport("grpc_csharp_ext.dll")] static extern GRPCCallError grpcsharp_call_recv_message(CallSafeHandle call, @@ -159,11 +159,11 @@ namespace Grpc.Core.Internal grpcsharp_call_send_close_from_client(this, ctx).CheckOk(); } - public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback) + public void StartSendStatusFromServer(Status status, BatchCompletionDelegate callback, MetadataArraySafeHandle metadataArray) { var ctx = BatchContextSafeHandle.Create(); completionRegistry.RegisterBatchCompletion(ctx, callback); - grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail).CheckOk(); + grpcsharp_call_send_status_from_server(this, ctx, status.StatusCode, status.Detail, metadataArray).CheckOk(); } public void StartReceiveMessage(BatchCompletionDelegate callback) diff --git a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs index 80aa7f5603..427c16fac6 100644 --- a/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs +++ b/src/csharp/Grpc.Core/Internal/MetadataArraySafeHandle.cs @@ -46,12 +46,24 @@ namespace Grpc.Core.Internal static extern void grpcsharp_metadata_array_add(MetadataArraySafeHandle array, string key, byte[] value, UIntPtr valueLength); [DllImport("grpc_csharp_ext.dll")] + static extern UIntPtr grpcsharp_metadata_array_count(IntPtr metadataArray); + + [DllImport("grpc_csharp_ext.dll")] + static extern IntPtr grpcsharp_metadata_array_get_key(IntPtr metadataArray, UIntPtr index); + + [DllImport("grpc_csharp_ext.dll")] + static extern IntPtr grpcsharp_metadata_array_get_value(IntPtr metadataArray, UIntPtr index); + + [DllImport("grpc_csharp_ext.dll")] + static extern UIntPtr grpcsharp_metadata_array_get_value_length(IntPtr metadataArray, UIntPtr index); + + [DllImport("grpc_csharp_ext.dll")] static extern void grpcsharp_metadata_array_destroy_full(IntPtr array); private MetadataArraySafeHandle() { } - + public static MetadataArraySafeHandle Create(Metadata metadata) { // TODO(jtattermusch): we might wanna check that the metadata is readonly @@ -63,6 +75,38 @@ namespace Grpc.Core.Internal return metadataArray; } + /// <summary> + /// Reads metadata from pointer to grpc_metadata_array + /// </summary> + public static Metadata ReadMetadataFromPtrUnsafe(IntPtr metadataArray) + { + if (metadataArray == IntPtr.Zero) + { + return null; + } + + ulong count = grpcsharp_metadata_array_count(metadataArray).ToUInt64(); + + var metadata = new Metadata(); + for (ulong i = 0; i < count; i++) + { + var index = new UIntPtr(i); + string key = Marshal.PtrToStringAnsi(grpcsharp_metadata_array_get_key(metadataArray, index)); + var bytes = new byte[grpcsharp_metadata_array_get_value_length(metadataArray, index).ToUInt64()]; + Marshal.Copy(grpcsharp_metadata_array_get_value(metadataArray, index), bytes, 0, bytes.Length); + metadata.Add(new Metadata.Entry(key, bytes)); + } + return metadata; + } + + internal IntPtr Handle + { + get + { + return handle; + } + } + protected override bool ReleaseHandle() { grpcsharp_metadata_array_destroy_full(handle); diff --git a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs index 594e46b159..3680b1e791 100644 --- a/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs +++ b/src/csharp/Grpc.Core/Internal/ServerCallHandler.cs @@ -34,6 +34,7 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Grpc.Core.Internal; using Grpc.Core.Utils; @@ -42,7 +43,7 @@ namespace Grpc.Core.Internal { internal interface IServerCallHandler { - Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment); + Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment); } internal class UnaryServerCallHandler<TRequest, TResponse> : IServerCallHandler @@ -58,27 +59,28 @@ namespace Grpc.Core.Internal this.handler = handler; } - public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment) + public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment) { var asyncCall = new AsyncCallServer<TRequest, TResponse>( method.ResponseMarshaller.Serializer, method.RequestMarshaller.Deserializer, environment); - asyncCall.Initialize(call); + asyncCall.Initialize(newRpc.Call); var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); - Status status = Status.DefaultSuccess; + Status status; + var context = HandlerUtils.NewContext(newRpc); try { Preconditions.CheckArgument(await requestStream.MoveNext()); var request = requestStream.Current; // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. Preconditions.CheckArgument(!await requestStream.MoveNext()); - var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context - var result = await handler(context, request); + var result = await handler(request, context); + status = context.Status; await responseStream.WriteAsync(result); } catch (Exception e) @@ -88,7 +90,7 @@ namespace Grpc.Core.Internal } try { - await responseStream.WriteStatusAsync(status); + await responseStream.WriteStatusAsync(status, context.ResponseTrailers); } catch (OperationCanceledException) { @@ -111,28 +113,28 @@ namespace Grpc.Core.Internal this.handler = handler; } - public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment) + public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment) { var asyncCall = new AsyncCallServer<TRequest, TResponse>( method.ResponseMarshaller.Serializer, method.RequestMarshaller.Deserializer, environment); - asyncCall.Initialize(call); + asyncCall.Initialize(newRpc.Call); var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); - Status status = Status.DefaultSuccess; + Status status; + var context = HandlerUtils.NewContext(newRpc); try { Preconditions.CheckArgument(await requestStream.MoveNext()); var request = requestStream.Current; // TODO(jtattermusch): we need to read the full stream so that native callhandle gets deallocated. Preconditions.CheckArgument(!await requestStream.MoveNext()); - - var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context - await handler(context, request, responseStream); + await handler(request, responseStream, context); + status = context.Status; } catch (Exception e) { @@ -142,7 +144,7 @@ namespace Grpc.Core.Internal try { - await responseStream.WriteStatusAsync(status); + await responseStream.WriteStatusAsync(status, context.ResponseTrailers); } catch (OperationCanceledException) { @@ -165,23 +167,24 @@ namespace Grpc.Core.Internal this.handler = handler; } - public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment) + public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment) { var asyncCall = new AsyncCallServer<TRequest, TResponse>( method.ResponseMarshaller.Serializer, method.RequestMarshaller.Deserializer, environment); - asyncCall.Initialize(call); + asyncCall.Initialize(newRpc.Call); var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); - var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context - Status status = Status.DefaultSuccess; + Status status; + var context = HandlerUtils.NewContext(newRpc); try { - var result = await handler(context, requestStream); + var result = await handler(requestStream, context); + status = context.Status; try { await responseStream.WriteAsync(result); @@ -199,7 +202,7 @@ namespace Grpc.Core.Internal try { - await responseStream.WriteStatusAsync(status); + await responseStream.WriteStatusAsync(status, context.ResponseTrailers); } catch (OperationCanceledException) { @@ -222,23 +225,24 @@ namespace Grpc.Core.Internal this.handler = handler; } - public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment) + public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment) { var asyncCall = new AsyncCallServer<TRequest, TResponse>( method.ResponseMarshaller.Serializer, method.RequestMarshaller.Deserializer, environment); - asyncCall.Initialize(call); + asyncCall.Initialize(newRpc.Call); var finishedTask = asyncCall.ServerSideCallAsync(); var requestStream = new ServerRequestStream<TRequest, TResponse>(asyncCall); var responseStream = new ServerResponseStream<TRequest, TResponse>(asyncCall); - var context = new ServerCallContext(); // TODO(jtattermusch): initialize the context - Status status = Status.DefaultSuccess; + Status status; + var context = HandlerUtils.NewContext(newRpc); try { - await handler(context, requestStream, responseStream); + await handler(requestStream, responseStream, context); + status = context.Status; } catch (Exception e) { @@ -247,7 +251,7 @@ namespace Grpc.Core.Internal } try { - await responseStream.WriteStatusAsync(status); + await responseStream.WriteStatusAsync(status, context.ResponseTrailers); } catch (OperationCanceledException) { @@ -259,18 +263,19 @@ namespace Grpc.Core.Internal internal class NoSuchMethodCallHandler : IServerCallHandler { - public async Task HandleCall(string methodName, CallSafeHandle call, GrpcEnvironment environment) + public static readonly NoSuchMethodCallHandler Instance = new NoSuchMethodCallHandler(); + + public async Task HandleCall(ServerRpcNew newRpc, GrpcEnvironment environment) { // We don't care about the payload type here. var asyncCall = new AsyncCallServer<byte[], byte[]>( (payload) => payload, (payload) => payload, environment); - asyncCall.Initialize(call); + asyncCall.Initialize(newRpc.Call); var finishedTask = asyncCall.ServerSideCallAsync(); - var requestStream = new ServerRequestStream<byte[], byte[]>(asyncCall); var responseStream = new ServerResponseStream<byte[], byte[]>(asyncCall); - await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method.")); + await responseStream.WriteStatusAsync(new Status(StatusCode.Unimplemented, "No such method."), Metadata.Empty); await finishedTask; } } @@ -279,8 +284,22 @@ namespace Grpc.Core.Internal { public static Status StatusFromException(Exception e) { + var rpcException = e as RpcException; + if (rpcException != null) + { + // use the status thrown by handler. + return rpcException.Status; + } + // TODO(jtattermusch): what is the right status code here? return new Status(StatusCode.Unknown, "Exception was thrown by handler."); } + + public static ServerCallContext NewContext(ServerRpcNew newRpc) + { + return new ServerCallContext( + newRpc.Method, newRpc.Host, newRpc.Deadline.ToDateTime(), + newRpc.RequestMetadata, CancellationToken.None); + } } } diff --git a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs index a2d77dd5b7..756dcee87f 100644 --- a/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs +++ b/src/csharp/Grpc.Core/Internal/ServerResponseStream.cs @@ -56,10 +56,10 @@ namespace Grpc.Core.Internal return taskSource.Task; } - public Task WriteStatusAsync(Status status) + public Task WriteStatusAsync(Status status, Metadata trailers) { var taskSource = new AsyncCompletionTaskSource<object>(); - call.StartSendStatusFromServer(status, taskSource.CompletionDelegate); + call.StartSendStatusFromServer(status, trailers, taskSource.CompletionDelegate); return taskSource.Task; } } diff --git a/src/csharp/Grpc.Core/Internal/Timespec.cs b/src/csharp/Grpc.Core/Internal/Timespec.cs index de783f5a4b..da2819f14d 100644 --- a/src/csharp/Grpc.Core/Internal/Timespec.cs +++ b/src/csharp/Grpc.Core/Internal/Timespec.cs @@ -43,6 +43,8 @@ namespace Grpc.Core.Internal const int NanosPerSecond = 1000 * 1000 * 1000; const int NanosPerTick = 100; + static readonly DateTime UnixEpoch = new DateTime(1970, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc); + [DllImport("grpc_csharp_ext.dll")] static extern Timespec gprsharp_now(); @@ -52,6 +54,13 @@ namespace Grpc.Core.Internal [DllImport("grpc_csharp_ext.dll")] static extern int gprsharp_sizeof_timespec(); + public Timespec(IntPtr tv_sec, int tv_nsec) + { + this.tv_sec = tv_sec; + this.tv_nsec = tv_nsec; + this.clock_type = GPRClockType.Realtime; + } + // NOTE: on linux 64bit sizeof(gpr_timespec) = 16, on windows 32bit sizeof(gpr_timespec) = 8 // so IntPtr seems to have the right size to work on both. public System.IntPtr tv_sec; @@ -76,6 +85,11 @@ namespace Grpc.Core.Internal return gprsharp_now(); } } + + public DateTime ToDateTime() + { + return UnixEpoch.AddTicks(tv_sec.ToInt64() * (NanosPerSecond / NanosPerTick) + tv_nsec / NanosPerTick); + } internal static int NativeSize { diff --git a/src/csharp/Grpc.Core/Metadata.cs b/src/csharp/Grpc.Core/Metadata.cs index 4552d39d88..2f308cbb11 100644 --- a/src/csharp/Grpc.Core/Metadata.cs +++ b/src/csharp/Grpc.Core/Metadata.cs @@ -220,6 +220,11 @@ namespace Grpc.Core return value; } } + + public override string ToString() + { + return string.Format("[Entry: key={0}, value={1}]", Key, Value); + } } } } diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs index cbf77196cf..fd30735359 100644 --- a/src/csharp/Grpc.Core/Server.cs +++ b/src/csharp/Grpc.Core/Server.cs @@ -53,6 +53,7 @@ namespace Grpc.Core public const int PickUnusedPort = 0; readonly GrpcEnvironment environment; + readonly List<ChannelOption> options; readonly ServerSafeHandle handle; readonly object myLock = new object(); @@ -69,7 +70,8 @@ namespace Grpc.Core public Server(IEnumerable<ChannelOption> options = null) { this.environment = GrpcEnvironment.GetInstance(); - using (var channelArgs = ChannelOptions.CreateChannelArgs(options)) + this.options = options != null ? new List<ChannelOption>(options) : new List<ChannelOption>(); + using (var channelArgs = ChannelOptions.CreateChannelArgs(this.options)) { this.handle = ServerSafeHandle.NewServer(environment.CompletionQueue, channelArgs); } @@ -218,16 +220,16 @@ namespace Grpc.Core /// <summary> /// Selects corresponding handler for given call and handles the call. /// </summary> - private async Task InvokeCallHandler(CallSafeHandle call, string method) + private async Task HandleCallAsync(ServerRpcNew newRpc) { try { IServerCallHandler callHandler; - if (!callHandlers.TryGetValue(method, out callHandler)) + if (!callHandlers.TryGetValue(newRpc.Method, out callHandler)) { - callHandler = new NoSuchMethodCallHandler(); + callHandler = NoSuchMethodCallHandler.Instance; } - await callHandler.HandleCall(method, call, environment); + await callHandler.HandleCall(newRpc, environment); } catch (Exception e) { @@ -240,15 +242,15 @@ namespace Grpc.Core /// </summary> private void HandleNewServerRpc(bool success, BatchContextSafeHandle ctx) { - // TODO: handle error - - CallSafeHandle call = ctx.GetServerRpcNewCall(); - string method = ctx.GetServerRpcNewMethod(); - - // after server shutdown, the callback returns with null call - if (!call.IsInvalid) + if (success) { - Task.Run(async () => await InvokeCallHandler(call, method)); + ServerRpcNew newRpc = ctx.GetServerRpcNew(); + + // after server shutdown, the callback returns with null call + if (!newRpc.Call.IsInvalid) + { + Task.Run(async () => await HandleCallAsync(newRpc)); + } } AllowOneRpc(); diff --git a/src/csharp/Grpc.Core/ServerCallContext.cs b/src/csharp/Grpc.Core/ServerCallContext.cs index bc9a499c51..17a2eefd07 100644 --- a/src/csharp/Grpc.Core/ServerCallContext.cs +++ b/src/csharp/Grpc.Core/ServerCallContext.cs @@ -33,6 +33,7 @@ using System; using System.Runtime.CompilerServices; +using System.Threading; using System.Threading.Tasks; namespace Grpc.Core @@ -42,14 +43,93 @@ namespace Grpc.Core /// </summary> public sealed class ServerCallContext { - // TODO(jtattermusch): add cancellationToken + // TODO(jtattermusch): expose method to send initial metadata back to client - // TODO(jtattermusch): add deadline info + private readonly string method; + private readonly string host; + private readonly DateTime deadline; + private readonly Metadata requestHeaders; + private readonly CancellationToken cancellationToken; + private readonly Metadata responseTrailers = new Metadata(); - // TODO(jtattermusch): expose initial metadata sent by client for reading + private Status status = Status.DefaultSuccess; - // TODO(jtattermusch): expose method to send initial metadata back to client + public ServerCallContext(string method, string host, DateTime deadline, Metadata requestHeaders, CancellationToken cancellationToken) + { + this.method = method; + this.host = host; + this.deadline = deadline; + this.requestHeaders = requestHeaders; + this.cancellationToken = cancellationToken; + } + + /// <summary> Name of method called in this RPC. </summary> + public string Method + { + get + { + return this.method; + } + } + + /// <summary> Name of host called in this RPC. </summary> + public string Host + { + get + { + return this.host; + } + } + + /// <summary> Deadline for this RPC. </summary> + public DateTime Deadline + { + get + { + return this.deadline; + } + } + + /// <summary> Initial metadata sent by client. </summary> + public Metadata RequestHeaders + { + get + { + return this.requestHeaders; + } + } + + // TODO(jtattermusch): support signalling cancellation. + /// <summary> Cancellation token signals when call is cancelled. </summary> + public CancellationToken CancellationToken + { + get + { + return this.cancellationToken; + } + } + + /// <summary> Trailers to send back to client after RPC finishes.</summary> + public Metadata ResponseTrailers + { + get + { + return this.responseTrailers; + } + } + + /// <summary> Status to send back to client after RPC finishes.</summary> + public Status Status + { + get + { + return this.status; + } - // TODO(jtattermusch): allow setting status and trailing metadata to send after handler completes. + set + { + status = value; + } + } } } diff --git a/src/csharp/Grpc.Core/ServerMethods.cs b/src/csharp/Grpc.Core/ServerMethods.cs index 377b78eb30..d457770203 100644 --- a/src/csharp/Grpc.Core/ServerMethods.cs +++ b/src/csharp/Grpc.Core/ServerMethods.cs @@ -42,28 +42,28 @@ namespace Grpc.Core /// <summary> /// Server-side handler for unary call. /// </summary> - public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request) + public delegate Task<TResponse> UnaryServerMethod<TRequest, TResponse>(TRequest request, ServerCallContext context) where TRequest : class where TResponse : class; /// <summary> /// Server-side handler for client streaming call. /// </summary> - public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream) + public delegate Task<TResponse> ClientStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, ServerCallContext context) where TRequest : class where TResponse : class; /// <summary> /// Server-side handler for server streaming call. /// </summary> - public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, TRequest request, IServerStreamWriter<TResponse> responseStream) + public delegate Task ServerStreamingServerMethod<TRequest, TResponse>(TRequest request, IServerStreamWriter<TResponse> responseStream, ServerCallContext context) where TRequest : class where TResponse : class; /// <summary> /// Server-side handler for bidi streaming call. /// </summary> - public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(ServerCallContext context, IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream) + public delegate Task DuplexStreamingServerMethod<TRequest, TResponse>(IAsyncStreamReader<TRequest> requestStream, IServerStreamWriter<TResponse> responseStream, ServerCallContext context) where TRequest : class where TResponse : class; } diff --git a/src/csharp/Grpc.Core/Version.cs b/src/csharp/Grpc.Core/Version.cs index f1db1f6157..b5cb652945 100644 --- a/src/csharp/Grpc.Core/Version.cs +++ b/src/csharp/Grpc.Core/Version.cs @@ -2,4 +2,4 @@ using System.Reflection; using System.Runtime.CompilerServices; // The current version of gRPC C#. -[assembly: AssemblyVersion("0.6.0.*")] +[assembly: AssemblyVersion(Grpc.Core.VersionInfo.CurrentVersion + ".*")] diff --git a/src/csharp/Grpc.Core/VersionInfo.cs b/src/csharp/Grpc.Core/VersionInfo.cs new file mode 100644 index 0000000000..656a3d47bb --- /dev/null +++ b/src/csharp/Grpc.Core/VersionInfo.cs @@ -0,0 +1,13 @@ +using System.Reflection; +using System.Runtime.CompilerServices; + +namespace Grpc.Core +{ + public static class VersionInfo + { + /// <summary> + /// Current version of gRPC + /// </summary> + public const string CurrentVersion = "0.6.0"; + } +} diff --git a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs index e7c4b33120..7a957c5b6f 100644 --- a/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs +++ b/src/csharp/Grpc.Examples.Tests/MathClientServerTests.cs @@ -144,7 +144,7 @@ namespace math.Tests n => Num.CreateBuilder().SetNum_(n).Build()); await call.RequestStream.WriteAll(numbers); - var result = await call.Result; + var result = await call.ResponseAsync; Assert.AreEqual(60, result.Num_); } }).Wait(); diff --git a/src/csharp/Grpc.Examples/MathExamples.cs b/src/csharp/Grpc.Examples/MathExamples.cs index 7deb651689..06d81a4d83 100644 --- a/src/csharp/Grpc.Examples/MathExamples.cs +++ b/src/csharp/Grpc.Examples/MathExamples.cs @@ -46,8 +46,7 @@ namespace math public static async Task DivAsyncExample(Math.IMathClient client) { - Task<DivReply> resultTask = client.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build()); - DivReply result = await resultTask; + DivReply result = await client.DivAsync(new DivArgs.Builder { Dividend = 4, Divisor = 5 }.Build()); Console.WriteLine("DivAsync Result: " + result); } @@ -72,7 +71,7 @@ namespace math using (var call = client.Sum()) { await call.RequestStream.WriteAll(numbers); - Console.WriteLine("Sum Result: " + await call.Result); + Console.WriteLine("Sum Result: " + await call.ResponseAsync); } } @@ -104,7 +103,7 @@ namespace math using (var sumCall = client.Sum()) { await sumCall.RequestStream.WriteAll(numbers); - sum = await sumCall.Result; + sum = await sumCall.ResponseAsync; } DivReply result = await client.DivAsync(new DivArgs.Builder { Dividend = sum.Num_, Divisor = numbers.Count }.Build()); diff --git a/src/csharp/Grpc.Examples/MathGrpc.cs b/src/csharp/Grpc.Examples/MathGrpc.cs index 1805972ce3..ef787cf1d8 100644 --- a/src/csharp/Grpc.Examples/MathGrpc.cs +++ b/src/csharp/Grpc.Examples/MathGrpc.cs @@ -45,7 +45,7 @@ namespace math { public interface IMathClient { global::math.DivReply Div(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); - Task<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); + AsyncUnaryCall<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); AsyncDuplexStreamingCall<global::math.DivArgs, global::math.DivReply> DivMany(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); AsyncServerStreamingCall<global::math.Num> Fib(global::math.FibArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); AsyncClientStreamingCall<global::math.Num, global::math.Num> Sum(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); @@ -54,10 +54,10 @@ namespace math { // server-side interface public interface IMath { - Task<global::math.DivReply> Div(ServerCallContext context, global::math.DivArgs request); - Task DivMany(ServerCallContext context, IAsyncStreamReader<global::math.DivArgs> requestStream, IServerStreamWriter<global::math.DivReply> responseStream); - Task Fib(ServerCallContext context, global::math.FibArgs request, IServerStreamWriter<global::math.Num> responseStream); - Task<global::math.Num> Sum(ServerCallContext context, IAsyncStreamReader<global::math.Num> requestStream); + Task<global::math.DivReply> Div(global::math.DivArgs request, ServerCallContext context); + Task DivMany(IAsyncStreamReader<global::math.DivArgs> requestStream, IServerStreamWriter<global::math.DivReply> responseStream, ServerCallContext context); + Task Fib(global::math.FibArgs request, IServerStreamWriter<global::math.Num> responseStream, ServerCallContext context); + Task<global::math.Num> Sum(IAsyncStreamReader<global::math.Num> requestStream, ServerCallContext context); } // client stub @@ -71,7 +71,7 @@ namespace math { var call = CreateCall(__ServiceName, __Method_Div, headers); return Calls.BlockingUnaryCall(call, request, cancellationToken); } - public Task<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) + public AsyncUnaryCall<global::math.DivReply> DivAsync(global::math.DivArgs request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) { var call = CreateCall(__ServiceName, __Method_Div, headers); return Calls.AsyncUnaryCall(call, request, cancellationToken); diff --git a/src/csharp/Grpc.Examples/MathServiceImpl.cs b/src/csharp/Grpc.Examples/MathServiceImpl.cs index e247ac9d73..3dd0f53a0d 100644 --- a/src/csharp/Grpc.Examples/MathServiceImpl.cs +++ b/src/csharp/Grpc.Examples/MathServiceImpl.cs @@ -45,12 +45,12 @@ namespace math /// </summary> public class MathServiceImpl : Math.IMath { - public Task<DivReply> Div(ServerCallContext context, DivArgs request) + public Task<DivReply> Div(DivArgs request, ServerCallContext context) { return Task.FromResult(DivInternal(request)); } - public async Task Fib(ServerCallContext context, FibArgs request, IServerStreamWriter<Num> responseStream) + public async Task Fib(FibArgs request, IServerStreamWriter<Num> responseStream, ServerCallContext context) { if (request.Limit <= 0) { @@ -67,7 +67,7 @@ namespace math } } - public async Task<Num> Sum(ServerCallContext context, IAsyncStreamReader<Num> requestStream) + public async Task<Num> Sum(IAsyncStreamReader<Num> requestStream, ServerCallContext context) { long sum = 0; await requestStream.ForEach(async num => @@ -77,7 +77,7 @@ namespace math return Num.CreateBuilder().SetNum_(sum).Build(); } - public async Task DivMany(ServerCallContext context, IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream) + public async Task DivMany(IAsyncStreamReader<DivArgs> requestStream, IServerStreamWriter<DivReply> responseStream, ServerCallContext context) { await requestStream.ForEach(async divArgs => { diff --git a/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs b/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs index 73ff0e74b5..bc14a0a62f 100644 --- a/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs +++ b/src/csharp/Grpc.HealthCheck.Tests/HealthClientServerTest.cs @@ -87,9 +87,7 @@ namespace Grpc.HealthCheck.Tests [Test] public void ServiceDoesntExist() { - // TODO(jtattermusch): currently, this returns wrong status code, because we don't enable sending arbitrary status code from - // server handlers yet. - Assert.Throws(typeof(RpcException), () => client.Check(HealthCheckRequest.CreateBuilder().SetHost("").SetService("nonexistent.service").Build())); + Assert.Throws(Is.TypeOf(typeof(RpcException)).And.Property("Status").Property("StatusCode").EqualTo(StatusCode.NotFound), () => client.Check(HealthCheckRequest.CreateBuilder().SetHost("").SetService("nonexistent.service").Build())); } // TODO(jtattermusch): add test with timeout once timeouts are supported diff --git a/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs b/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs index 9b7c4f2140..7184415655 100644 --- a/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs +++ b/src/csharp/Grpc.HealthCheck.Tests/HealthServiceImplTest.cs @@ -101,7 +101,7 @@ namespace Grpc.HealthCheck.Tests private static HealthCheckResponse.Types.ServingStatus GetStatusHelper(HealthServiceImpl impl, string host, string service) { - return impl.Check(null, HealthCheckRequest.CreateBuilder().SetHost(host).SetService(service).Build()).Result.Status; + return impl.Check(HealthCheckRequest.CreateBuilder().SetHost(host).SetService(service).Build(), null).Result.Status; } } } diff --git a/src/csharp/Grpc.HealthCheck/HealthGrpc.cs b/src/csharp/Grpc.HealthCheck/HealthGrpc.cs index 3aebdcb557..217127eca7 100644 --- a/src/csharp/Grpc.HealthCheck/HealthGrpc.cs +++ b/src/csharp/Grpc.HealthCheck/HealthGrpc.cs @@ -25,13 +25,13 @@ namespace Grpc.Health.V1Alpha { public interface IHealthClient { global::Grpc.Health.V1Alpha.HealthCheckResponse Check(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); - Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); + AsyncUnaryCall<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); } // server-side interface public interface IHealth { - Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> Check(ServerCallContext context, global::Grpc.Health.V1Alpha.HealthCheckRequest request); + Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> Check(global::Grpc.Health.V1Alpha.HealthCheckRequest request, ServerCallContext context); } // client stub @@ -45,7 +45,7 @@ namespace Grpc.Health.V1Alpha { var call = CreateCall(__ServiceName, __Method_Check, headers); return Calls.BlockingUnaryCall(call, request, cancellationToken); } - public Task<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) + public AsyncUnaryCall<global::Grpc.Health.V1Alpha.HealthCheckResponse> CheckAsync(global::Grpc.Health.V1Alpha.HealthCheckRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) { var call = CreateCall(__ServiceName, __Method_Check, headers); return Calls.AsyncUnaryCall(call, request, cancellationToken); diff --git a/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs b/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs index db3a2a0942..3c3b9c35f1 100644 --- a/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs +++ b/src/csharp/Grpc.HealthCheck/HealthServiceImpl.cs @@ -95,7 +95,7 @@ namespace Grpc.HealthCheck } } - public Task<HealthCheckResponse> Check(ServerCallContext context, HealthCheckRequest request) + public Task<HealthCheckResponse> Check(HealthCheckRequest request, ServerCallContext context) { lock (myLock) { diff --git a/src/csharp/Grpc.HealthCheck/Settings.StyleCop b/src/csharp/Grpc.HealthCheck/Settings.StyleCop new file mode 100644 index 0000000000..2942add962 --- /dev/null +++ b/src/csharp/Grpc.HealthCheck/Settings.StyleCop @@ -0,0 +1,10 @@ +<StyleCopSettings Version="105"> + <SourceFileList> + <SourceFile>Health.cs</SourceFile> + <Settings> + <GlobalSettings> + <BooleanProperty Name="RulesEnabledByDefault">False</BooleanProperty> + </GlobalSettings> + </Settings> + </SourceFileList> +</StyleCopSettings> diff --git a/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj b/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj index 328acb5b47..dc1d0a44c0 100644 --- a/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj +++ b/src/csharp/Grpc.IntegrationTesting.Client/Grpc.IntegrationTesting.Client.csproj @@ -3,8 +3,6 @@ <PropertyGroup> <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration> <Platform Condition=" '$(Platform)' == '' ">x86</Platform> - <ProductVersion>10.0.0</ProductVersion> - <SchemaVersion>2.0</SchemaVersion> <ProjectGuid>{3D166931-BA2D-416E-95A3-D36E8F6E90B9}</ProjectGuid> <OutputType>Exe</OutputType> <RootNamespace>Grpc.IntegrationTesting.Client</RootNamespace> @@ -48,6 +46,10 @@ <Project>{C61154BA-DD4A-4838-8420-0162A28925E0}</Project> <Name>Grpc.IntegrationTesting</Name> </ProjectReference> + <ProjectReference Include="..\Grpc.Core\Grpc.Core.csproj"> + <Project>{CCC4440E-49F7-4790-B0AF-FEABB0837AE7}</Project> + <Name>Grpc.Core</Name> + </ProjectReference> </ItemGroup> <ItemGroup> <None Include="app.config" /> diff --git a/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj b/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj index ae184c1dc7..f03c8f3ce3 100644 --- a/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj +++ b/src/csharp/Grpc.IntegrationTesting.Server/Grpc.IntegrationTesting.Server.csproj @@ -3,8 +3,6 @@ <PropertyGroup> <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration> <Platform Condition=" '$(Platform)' == '' ">x86</Platform> - <ProductVersion>10.0.0</ProductVersion> - <SchemaVersion>2.0</SchemaVersion> <ProjectGuid>{A654F3B8-E859-4E6A-B30D-227527DBEF0D}</ProjectGuid> <OutputType>Exe</OutputType> <RootNamespace>Grpc.IntegrationTesting.Server</RootNamespace> @@ -48,6 +46,10 @@ <Project>{C61154BA-DD4A-4838-8420-0162A28925E0}</Project> <Name>Grpc.IntegrationTesting</Name> </ProjectReference> + <ProjectReference Include="..\Grpc.Core\Grpc.Core.csproj"> + <Project>{CCC4440E-49F7-4790-B0AF-FEABB0837AE7}</Project> + <Name>Grpc.Core</Name> + </ProjectReference> </ItemGroup> <ItemGroup> <None Include="app.config" /> diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj index af4a75a034..d3c69ab9eb 100644 --- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj +++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj @@ -32,21 +32,34 @@ <PlatformTarget>x86</PlatformTarget> </PropertyGroup> <ItemGroup> - <Reference Include="Google.Apis.Auth.PlatformServices"> + <Reference Include="Google.Apis.Auth, Version=1.9.1.12395, Culture=neutral, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> + <HintPath>..\packages\Google.Apis.Auth.1.9.1\lib\net40\Google.Apis.Auth.dll</HintPath> + </Reference> + <Reference Include="Google.Apis.Auth.PlatformServices, Version=1.9.1.12399, Culture=neutral, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Google.Apis.Auth.1.9.1\lib\net40\Google.Apis.Auth.PlatformServices.dll</HintPath> </Reference> - <Reference Include="Google.Apis.Core"> + <Reference Include="Google.Apis.Core, Version=1.9.1.12394, Culture=neutral, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Google.Apis.Core.1.9.1\lib\portable-net40+sl50+win+wpa81+wp80\Google.Apis.Core.dll</HintPath> </Reference> - <Reference Include="Microsoft.Threading.Tasks"> + <Reference Include="Microsoft.Threading.Tasks, Version=1.0.12.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Microsoft.Bcl.Async.1.0.168\lib\net40\Microsoft.Threading.Tasks.dll</HintPath> </Reference> - <Reference Include="Microsoft.Threading.Tasks.Extensions"> + <Reference Include="Microsoft.Threading.Tasks.Extensions, Version=1.0.12.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Microsoft.Bcl.Async.1.0.168\lib\net40\Microsoft.Threading.Tasks.Extensions.dll</HintPath> </Reference> - <Reference Include="Microsoft.Threading.Tasks.Extensions.Desktop"> + <Reference Include="Microsoft.Threading.Tasks.Extensions.Desktop, Version=1.0.168.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Microsoft.Bcl.Async.1.0.168\lib\net40\Microsoft.Threading.Tasks.Extensions.Desktop.dll</HintPath> </Reference> + <Reference Include="Newtonsoft.Json, Version=6.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> + <HintPath>..\packages\Newtonsoft.Json.6.0.6\lib\net45\Newtonsoft.Json.dll</HintPath> + </Reference> <Reference Include="nunit.framework"> <HintPath>..\packages\NUnit.2.6.4\lib\nunit.framework.dll</HintPath> </Reference> @@ -63,16 +76,15 @@ </Reference> <Reference Include="System.Net" /> <Reference Include="System.Net.Http" /> - <Reference Include="System.Net.Http.Extensions"> + <Reference Include="System.Net.Http.Extensions, Version=2.2.28.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Microsoft.Net.Http.2.2.28\lib\net45\System.Net.Http.Extensions.dll</HintPath> </Reference> - <Reference Include="System.Net.Http.Primitives"> + <Reference Include="System.Net.Http.Primitives, Version=4.2.28.0, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a, processorArchitecture=MSIL"> + <SpecificVersion>False</SpecificVersion> <HintPath>..\packages\Microsoft.Net.Http.2.2.28\lib\net45\System.Net.Http.Primitives.dll</HintPath> </Reference> <Reference Include="System.Net.Http.WebRequest" /> - <Reference Include="Newtonsoft.Json"> - <HintPath>..\packages\Newtonsoft.Json.6.0.6\lib\net45\Newtonsoft.Json.dll</HintPath> - </Reference> </ItemGroup> <ItemGroup> <Compile Include="..\Grpc.Core\Version.cs"> diff --git a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs index 05e732dbd4..ce255f9423 100644 --- a/src/csharp/Grpc.IntegrationTesting/InteropClient.cs +++ b/src/csharp/Grpc.IntegrationTesting/InteropClient.cs @@ -135,7 +135,7 @@ namespace Grpc.IntegrationTesting GrpcEnvironment.Shutdown(); } - private void RunTestCase(string testCase, TestService.ITestServiceClient client) + private void RunTestCase(string testCase, TestService.TestServiceClient client) { switch (testCase) { @@ -163,6 +163,12 @@ namespace Grpc.IntegrationTesting case "compute_engine_creds": RunComputeEngineCreds(client); break; + case "oauth2_auth_token": + RunOAuth2AuthToken(client); + break; + case "per_rpc_creds": + RunPerRpcCreds(client); + break; case "cancel_after_begin": RunCancelAfterBegin(client); break; @@ -213,7 +219,7 @@ namespace Grpc.IntegrationTesting { await call.RequestStream.WriteAll(bodySizes); - var response = await call.Result; + var response = await call.ResponseAsync; Assert.AreEqual(74922, response.AggregatedPayloadSize); } Console.WriteLine("Passed!"); @@ -355,6 +361,51 @@ namespace Grpc.IntegrationTesting Console.WriteLine("Passed!"); } + public static void RunOAuth2AuthToken(TestService.TestServiceClient client) + { + Console.WriteLine("running oauth2_auth_token"); + var credential = GoogleCredential.GetApplicationDefault().CreateScoped(new[] { AuthScope }); + Assert.IsTrue(credential.RequestAccessTokenAsync(CancellationToken.None).Result); + string oauth2Token = credential.Token.AccessToken; + + // Intercept calls with an OAuth2 token obtained out-of-band. + client.HeaderInterceptor = new MetadataInterceptorDelegate((metadata) => + { + metadata.Add(new Metadata.Entry("Authorization", "Bearer " + oauth2Token)); + }); + + var request = SimpleRequest.CreateBuilder() + .SetFillUsername(true) + .SetFillOauthScope(true) + .Build(); + + var response = client.UnaryCall(request); + + Assert.AreEqual(AuthScopeResponse, response.OauthScope); + Assert.AreEqual(ServiceAccountUser, response.Username); + Console.WriteLine("Passed!"); + } + + public static void RunPerRpcCreds(TestService.TestServiceClient client) + { + Console.WriteLine("running per_rpc_creds"); + + var credential = GoogleCredential.GetApplicationDefault().CreateScoped(new[] { AuthScope }); + Assert.IsTrue(credential.RequestAccessTokenAsync(CancellationToken.None).Result); + string oauth2Token = credential.Token.AccessToken; + + var request = SimpleRequest.CreateBuilder() + .SetFillUsername(true) + .SetFillOauthScope(true) + .Build(); + + var response = client.UnaryCall(request, headers: new Metadata { new Metadata.Entry("Authorization", "Bearer " + oauth2Token) }); + + Assert.AreEqual(AuthScopeResponse, response.OauthScope); + Assert.AreEqual(ServiceAccountUser, response.Username); + Console.WriteLine("Passed!"); + } + public static void RunCancelAfterBegin(TestService.ITestServiceClient client) { Task.Run(async () => @@ -370,7 +421,7 @@ namespace Grpc.IntegrationTesting try { - var response = await call.Result; + var response = await call.ResponseAsync; Assert.Fail(); } catch (RpcException e) diff --git a/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs b/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs index 96d9b23717..de2fa07441 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestGrpc.cs @@ -60,9 +60,9 @@ namespace grpc.testing { public interface ITestServiceClient { global::grpc.testing.Empty EmptyCall(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); - Task<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); + AsyncUnaryCall<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); global::grpc.testing.SimpleResponse UnaryCall(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); - Task<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); + AsyncUnaryCall<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); AsyncServerStreamingCall<global::grpc.testing.StreamingOutputCallResponse> StreamingOutputCall(global::grpc.testing.StreamingOutputCallRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); AsyncClientStreamingCall<global::grpc.testing.StreamingInputCallRequest, global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); AsyncDuplexStreamingCall<global::grpc.testing.StreamingOutputCallRequest, global::grpc.testing.StreamingOutputCallResponse> FullDuplexCall(Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)); @@ -72,12 +72,12 @@ namespace grpc.testing { // server-side interface public interface ITestService { - Task<global::grpc.testing.Empty> EmptyCall(ServerCallContext context, global::grpc.testing.Empty request); - Task<global::grpc.testing.SimpleResponse> UnaryCall(ServerCallContext context, global::grpc.testing.SimpleRequest request); - Task StreamingOutputCall(ServerCallContext context, global::grpc.testing.StreamingOutputCallRequest request, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream); - Task<global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingInputCallRequest> requestStream); - Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream); - Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream); + Task<global::grpc.testing.Empty> EmptyCall(global::grpc.testing.Empty request, ServerCallContext context); + Task<global::grpc.testing.SimpleResponse> UnaryCall(global::grpc.testing.SimpleRequest request, ServerCallContext context); + Task StreamingOutputCall(global::grpc.testing.StreamingOutputCallRequest request, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context); + Task<global::grpc.testing.StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<global::grpc.testing.StreamingInputCallRequest> requestStream, ServerCallContext context); + Task FullDuplexCall(IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context); + Task HalfDuplexCall(IAsyncStreamReader<global::grpc.testing.StreamingOutputCallRequest> requestStream, IServerStreamWriter<global::grpc.testing.StreamingOutputCallResponse> responseStream, ServerCallContext context); } // client stub @@ -91,7 +91,7 @@ namespace grpc.testing { var call = CreateCall(__ServiceName, __Method_EmptyCall, headers); return Calls.BlockingUnaryCall(call, request, cancellationToken); } - public Task<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) + public AsyncUnaryCall<global::grpc.testing.Empty> EmptyCallAsync(global::grpc.testing.Empty request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) { var call = CreateCall(__ServiceName, __Method_EmptyCall, headers); return Calls.AsyncUnaryCall(call, request, cancellationToken); @@ -101,7 +101,7 @@ namespace grpc.testing { var call = CreateCall(__ServiceName, __Method_UnaryCall, headers); return Calls.BlockingUnaryCall(call, request, cancellationToken); } - public Task<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) + public AsyncUnaryCall<global::grpc.testing.SimpleResponse> UnaryCallAsync(global::grpc.testing.SimpleRequest request, Metadata headers = null, CancellationToken cancellationToken = default(CancellationToken)) { var call = CreateCall(__ServiceName, __Method_UnaryCall, headers); return Calls.AsyncUnaryCall(call, request, cancellationToken); diff --git a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs index 6bd997d1f4..ccf9fe6ced 100644 --- a/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs +++ b/src/csharp/Grpc.IntegrationTesting/TestServiceImpl.cs @@ -46,19 +46,19 @@ namespace grpc.testing /// </summary> public class TestServiceImpl : TestService.ITestService { - public Task<Empty> EmptyCall(ServerCallContext context, Empty request) + public Task<Empty> EmptyCall(Empty request, ServerCallContext context) { return Task.FromResult(Empty.DefaultInstance); } - public Task<SimpleResponse> UnaryCall(ServerCallContext context, SimpleRequest request) + public Task<SimpleResponse> UnaryCall(SimpleRequest request, ServerCallContext context) { var response = SimpleResponse.CreateBuilder() .SetPayload(CreateZerosPayload(request.ResponseSize)).Build(); return Task.FromResult(response); } - public async Task StreamingOutputCall(ServerCallContext context, StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream) + public async Task StreamingOutputCall(StreamingOutputCallRequest request, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context) { foreach (var responseParam in request.ResponseParametersList) { @@ -68,7 +68,7 @@ namespace grpc.testing } } - public async Task<StreamingInputCallResponse> StreamingInputCall(ServerCallContext context, IAsyncStreamReader<StreamingInputCallRequest> requestStream) + public async Task<StreamingInputCallResponse> StreamingInputCall(IAsyncStreamReader<StreamingInputCallRequest> requestStream, ServerCallContext context) { int sum = 0; await requestStream.ForEach(async request => @@ -78,7 +78,7 @@ namespace grpc.testing return StreamingInputCallResponse.CreateBuilder().SetAggregatedPayloadSize(sum).Build(); } - public async Task FullDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream) + public async Task FullDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context) { await requestStream.ForEach(async request => { @@ -91,7 +91,7 @@ namespace grpc.testing }); } - public async Task HalfDuplexCall(ServerCallContext context, IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream) + public async Task HalfDuplexCall(IAsyncStreamReader<StreamingOutputCallRequest> requestStream, IServerStreamWriter<StreamingOutputCallResponse> responseStream, ServerCallContext context) { throw new NotImplementedException(); } diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c index 7dd1959a5f..682521446f 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -167,6 +167,29 @@ grpcsharp_metadata_array_add(grpc_metadata_array *array, const char *key, array->count++; } +GPR_EXPORT gpr_intptr GPR_CALLTYPE +grpcsharp_metadata_array_count(grpc_metadata_array *array) { + return (gpr_intptr) array->count; +} + +GPR_EXPORT const char *GPR_CALLTYPE +grpcsharp_metadata_array_get_key(grpc_metadata_array *array, size_t index) { + GPR_ASSERT(index < array->count); + return array->metadata[index].key; +} + +GPR_EXPORT const char *GPR_CALLTYPE +grpcsharp_metadata_array_get_value(grpc_metadata_array *array, size_t index) { + GPR_ASSERT(index < array->count); + return array->metadata[index].value; +} + +GPR_EXPORT gpr_intptr GPR_CALLTYPE +grpcsharp_metadata_array_get_value_length(grpc_metadata_array *array, size_t index) { + GPR_ASSERT(index < array->count); + return (gpr_intptr) array->metadata[index].value_length; +} + /* Move contents of metadata array */ void grpcsharp_metadata_array_move(grpc_metadata_array *dest, grpc_metadata_array *src) { @@ -218,6 +241,12 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_batch_context_destroy(grpcsharp_batch_con gpr_free(ctx); } +GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE +grpcsharp_batch_context_recv_initial_metadata( + const grpcsharp_batch_context *ctx) { + return &(ctx->recv_initial_metadata); +} + GPR_EXPORT gpr_intptr GPR_CALLTYPE grpcsharp_batch_context_recv_message_length( const grpcsharp_batch_context *ctx) { if (!ctx->recv_message) { @@ -260,6 +289,12 @@ grpcsharp_batch_context_recv_status_on_client_details( return ctx->recv_status_on_client.status_details; } +GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE +grpcsharp_batch_context_recv_status_on_client_trailing_metadata( + const grpcsharp_batch_context *ctx) { + return &(ctx->recv_status_on_client.trailing_metadata); +} + GPR_EXPORT grpc_call *GPR_CALLTYPE grpcsharp_batch_context_server_rpc_new_call( const grpcsharp_batch_context *ctx) { return ctx->server_rpc_new.call; @@ -271,6 +306,24 @@ grpcsharp_batch_context_server_rpc_new_method( return ctx->server_rpc_new.call_details.method; } +GPR_EXPORT const char *GPR_CALLTYPE +grpcsharp_batch_context_server_rpc_new_host( + const grpcsharp_batch_context *ctx) { + return ctx->server_rpc_new.call_details.host; +} + +GPR_EXPORT gpr_timespec GPR_CALLTYPE +grpcsharp_batch_context_server_rpc_new_deadline( + const grpcsharp_batch_context *ctx) { + return ctx->server_rpc_new.call_details.deadline; +} + +GPR_EXPORT const grpc_metadata_array *GPR_CALLTYPE +grpcsharp_batch_context_server_rpc_new_request_metadata( + const grpcsharp_batch_context *ctx) { + return &(ctx->server_rpc_new.request_metadata); +} + GPR_EXPORT gpr_int32 GPR_CALLTYPE grpcsharp_batch_context_recv_close_on_server_cancelled( const grpcsharp_batch_context *ctx) { @@ -589,15 +642,20 @@ GPR_EXPORT grpc_call_error GPR_CALLTYPE grpcsharp_call_send_status_from_server(grpc_call *call, grpcsharp_batch_context *ctx, grpc_status_code status_code, - const char *status_details) { + const char *status_details, + grpc_metadata_array *trailing_metadata) { /* TODO: don't use magic number */ grpc_op ops[1]; ops[0].op = GRPC_OP_SEND_STATUS_FROM_SERVER; ops[0].data.send_status_from_server.status = status_code; ops[0].data.send_status_from_server.status_details = gpr_strdup(status_details); - ops[0].data.send_status_from_server.trailing_metadata = NULL; - ops[0].data.send_status_from_server.trailing_metadata_count = 0; + grpcsharp_metadata_array_move(&(ctx->send_status_from_server.trailing_metadata), + trailing_metadata); + ops[0].data.send_status_from_server.trailing_metadata_count = + ctx->send_status_from_server.trailing_metadata.count; + ops[0].data.send_status_from_server.trailing_metadata = + ctx->send_status_from_server.trailing_metadata.metadata; ops[0].flags = 0; return grpc_call_start_batch(call, ops, sizeof(ops) / sizeof(ops[0]), ctx); diff --git a/src/node/interop/interop_client.js b/src/node/interop/interop_client.js index b61b0b63c0..e810e68e45 100644 --- a/src/node/interop/interop_client.js +++ b/src/node/interop/interop_client.js @@ -318,6 +318,51 @@ function authTest(expected_user, scope, client, done) { }); } +function oauth2Test(expected_user, scope, per_rpc, client, done) { + (new GoogleAuth()).getApplicationDefault(function(err, credential) { + assert.ifError(err); + var arg = { + fill_username: true, + fill_oauth_scope: true + }; + credential = credential.createScoped(scope); + credential.getAccessToken(function(err, token) { + assert.ifError(err); + var updateMetadata = function(authURI, metadata, callback) { + metadata = _.clone(metadata); + if (metadata.Authorization) { + metadata.Authorization = _.clone(metadata.Authorization); + } else { + metadata.Authorization = []; + } + metadata.Authorization.push('Bearer ' + token); + callback(null, metadata); + }; + var makeTestCall = function(error, client_metadata) { + assert.ifError(error); + var call = client.unaryCall(arg, function(err, resp) { + assert.ifError(err); + assert.strictEqual(resp.username, expected_user); + assert.strictEqual(resp.oauth_scope, AUTH_SCOPE_RESPONSE); + }); + call.on('status', function(status) { + assert.strictEqual(status.code, grpc.status.OK); + if (done) { + done(); + } + }); + }; + if (per_rpc) { + updateMetadata('', {}, makeTestCall); + } else { + client.updateMetadata = updateMetadata; + makeTestCall(null, {}); + } + + }); + }); +} + /** * Map from test case names to test functions */ @@ -333,7 +378,9 @@ var test_cases = { timeout_on_sleeping_server: timeoutOnSleepingServer, compute_engine_creds: _.partial(authTest, COMPUTE_ENGINE_USER, null), service_account_creds: _.partial(authTest, AUTH_USER, AUTH_SCOPE), - jwt_token_creds: _.partial(authTest, AUTH_USER, null) + jwt_token_creds: _.partial(authTest, AUTH_USER, null), + oauth2_auth_token: _.partial(oauth2Test, AUTH_USER, AUTH_SCOPE, false), + per_rpc_creds: _.partial(oauth2Test, AUTH_USER, AUTH_SCOPE, true) }; /** diff --git a/src/node/src/client.js b/src/node/src/client.js index b7bad949d4..da6327b432 100644 --- a/src/node/src/client.js +++ b/src/node/src/client.js @@ -47,6 +47,7 @@ var Readable = stream.Readable; var Writable = stream.Writable; var Duplex = stream.Duplex; var util = require('util'); +var version = require('../package.json').version; util.inherits(ClientWritableStream, Writable); @@ -517,9 +518,12 @@ function makeClientConstructor(methods, serviceName) { callback(null, metadata); }; } - - this.server_address = address.replace(/\/$/, ''); + if (!options) { + options = {}; + } + options['grpc.primary_user_agent'] = 'grpc-node/' + version; this.channel = new grpc.Channel(address, options); + this.server_address = address.replace(/\/$/, ''); this.auth_uri = this.server_address + '/' + serviceName; this.updateMetadata = updateMetadata; } diff --git a/src/node/test/surface_test.js b/src/node/test/surface_test.js index 18178e49e4..3cb68f8cd8 100644 --- a/src/node/test/surface_test.js +++ b/src/node/test/surface_test.js @@ -258,6 +258,16 @@ describe('Echo metadata', function() { }); call.end(); }); + it('shows the correct user-agent string', function(done) { + var version = require('../package.json').version; + var call = client.unary({}, function(err, data) { + assert.ifError(err); + }, {key: ['value']}); + call.on('metadata', function(metadata) { + assert(_.startsWith(metadata['user-agent'], 'grpc-node/' + version)); + done(); + }); + }); }); describe('Other conditions', function() { var client; diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c index c319526e42..8b8d5b2f47 100644 --- a/src/php/ext/grpc/server.c +++ b/src/php/ext/grpc/server.c @@ -64,6 +64,7 @@ void free_wrapped_grpc_server(void *object TSRMLS_DC) { wrapped_grpc_server *server = (wrapped_grpc_server *)object; if (server->wrapped != NULL) { grpc_server_shutdown_and_notify(server->wrapped, completion_queue, NULL); + grpc_server_cancel_all_calls(server->wrapped); grpc_completion_queue_pluck(completion_queue, NULL, gpr_inf_future(GPR_CLOCK_REALTIME)); grpc_server_destroy(server->wrapped); diff --git a/src/php/tests/unit_tests/EndToEndTest.php b/src/php/tests/unit_tests/EndToEndTest.php index 296873fa8f..2980dca4a7 100755 --- a/src/php/tests/unit_tests/EndToEndTest.php +++ b/src/php/tests/unit_tests/EndToEndTest.php @@ -61,7 +61,6 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{ $event = $this->server->requestCall(); $this->assertSame('dummy_method', $event->method); - $this->assertSame([], $event->metadata); $server_call = $event->call; $event = $server_call->startBatch([ @@ -83,7 +82,6 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{ Grpc\OP_RECV_STATUS_ON_CLIENT => true ]); - $this->assertSame([], $event->metadata); $status = $event->status; $this->assertSame([], $status->metadata); $this->assertSame(Grpc\STATUS_OK, $status->code); diff --git a/src/php/tests/unit_tests/SecureEndToEndTest.php b/src/php/tests/unit_tests/SecureEndToEndTest.php index 0c18cd3e91..f91c006da5 100755 --- a/src/php/tests/unit_tests/SecureEndToEndTest.php +++ b/src/php/tests/unit_tests/SecureEndToEndTest.php @@ -73,7 +73,6 @@ class SecureEndToEndTest extends PHPUnit_Framework_TestCase{ $event = $this->server->requestCall(); $this->assertSame('dummy_method', $event->method); - $this->assertSame([], $event->metadata); $server_call = $event->call; $event = $server_call->startBatch([ diff --git a/src/python/src/grpc/_adapter/_c/utility.c b/src/python/src/grpc/_adapter/_c/utility.c index 000c8d0c38..d9f911a41a 100644 --- a/src/python/src/grpc/_adapter/_c/utility.c +++ b/src/python/src/grpc/_adapter/_c/utility.c @@ -489,10 +489,10 @@ PyObject *pygrpc_cast_metadata_array_to_pyseq(grpc_metadata_array metadata) { void pygrpc_byte_buffer_to_bytes( grpc_byte_buffer *buffer, char **result, size_t *result_size) { grpc_byte_buffer_reader reader; - grpc_byte_buffer_reader_init(&reader, buffer); gpr_slice slice; char *read_result = NULL; size_t size = 0; + grpc_byte_buffer_reader_init(&reader, buffer); while (grpc_byte_buffer_reader_next(&reader, &slice)) { read_result = gpr_realloc(read_result, size + GPR_SLICE_LENGTH(slice)); memcpy(read_result + size, GPR_SLICE_START_PTR(slice), diff --git a/src/python/src/grpc/_adapter/_low_test.py b/src/python/src/grpc/_adapter/_low_test.py index 268e5fe765..a49cd007bf 100644 --- a/src/python/src/grpc/_adapter/_low_test.py +++ b/src/python/src/grpc/_adapter/_low_test.py @@ -129,7 +129,10 @@ class InsecureServerInsecureClient(unittest.TestCase): self.assertIsInstance(request_event.call, _low.Call) self.assertIs(server_request_tag, request_event.tag) self.assertEquals(1, len(request_event.results)) - self.assertEquals(dict(client_initial_metadata), dict(request_event.results[0].initial_metadata)) + got_initial_metadata = dict(request_event.results[0].initial_metadata) + self.assertEquals( + dict(client_initial_metadata), + dict((x, got_initial_metadata[x]) for x in zip(*client_initial_metadata)[0])) self.assertEquals(METHOD, request_event.call_details.method) self.assertEquals(HOST, request_event.call_details.host) self.assertLess(abs(DEADLINE - request_event.call_details.deadline), DEADLINE_TOLERANCE) diff --git a/src/python/src/grpc/_links/_transmission_test.py b/src/python/src/grpc/_links/_transmission_test.py index c5ef1edb25..3eeec03f46 100644 --- a/src/python/src/grpc/_links/_transmission_test.py +++ b/src/python/src/grpc/_links/_transmission_test.py @@ -93,8 +93,13 @@ class TransmissionTest(test_cases.TransmissionTest, unittest.TestCase): def create_service_completion(self): return _intermediary_low.Code.OK, 'An exuberant test "details" message!' - def assertMetadataEqual(self, original_metadata, transmitted_metadata): - self.assertSequenceEqual(original_metadata, transmitted_metadata) + def assertMetadataTransmitted(self, original_metadata, transmitted_metadata): + # we need to filter out any additional metadata added in transmitted_metadata + # since implementations are allowed to add to what is sent (in any position) + keys, _ = zip(*original_metadata) + self.assertSequenceEqual( + original_metadata, + [x for x in transmitted_metadata if x[0] in keys]) class RoundTripTest(unittest.TestCase): diff --git a/src/python/src/grpc/framework/interfaces/links/test_cases.py b/src/python/src/grpc/framework/interfaces/links/test_cases.py index 3ac212ebdf..bf1f09d99d 100644 --- a/src/python/src/grpc/framework/interfaces/links/test_cases.py +++ b/src/python/src/grpc/framework/interfaces/links/test_cases.py @@ -161,8 +161,8 @@ class TransmissionTest(object): raise NotImplementedError() @abc.abstractmethod - def assertMetadataEqual(self, original_metadata, transmitted_metadata): - """Asserts that two metadata objects are equal. + def assertMetadataTransmitted(self, original_metadata, transmitted_metadata): + """Asserts that transmitted_metadata contains original_metadata. Args: original_metadata: A metadata object used in this test. @@ -170,7 +170,8 @@ class TransmissionTest(object): through the system under test. Raises: - AssertionError: if the two metadata objects are not equal. + AssertionError: if the transmitted_metadata object does not contain + original_metadata. """ raise NotImplementedError() @@ -239,7 +240,7 @@ class TransmissionTest(object): self.assertFalse(initial_metadata_seen) self.assertFalse(seen_payloads) self.assertFalse(terminal_metadata_seen) - self.assertMetadataEqual(initial_metadata, ticket.initial_metadata) + self.assertMetadataTransmitted(initial_metadata, ticket.initial_metadata) initial_metadata_seen = True if ticket.payload is not None: @@ -248,7 +249,7 @@ class TransmissionTest(object): if ticket.terminal_metadata is not None: self.assertFalse(terminal_metadata_seen) - self.assertMetadataEqual(terminal_metadata, ticket.terminal_metadata) + self.assertMetadataTransmitted(terminal_metadata, ticket.terminal_metadata) terminal_metadata_seen = True self.assertSequenceEqual(payloads, seen_payloads) diff --git a/src/ruby/spec/generic/rpc_server_spec.rb b/src/ruby/spec/generic/rpc_server_spec.rb index f2403de77c..0326f6e894 100644 --- a/src/ruby/spec/generic/rpc_server_spec.rb +++ b/src/ruby/spec/generic/rpc_server_spec.rb @@ -35,6 +35,14 @@ def load_test_certs files.map { |f| File.open(File.join(test_root, f)).read } end +def check_md(wanted_md, received_md) + wanted_md.zip(received_md).each do |w, r| + w.each do |key, value| + expect(r[key]).to eq(value) + end + end +end + # A test message class EchoMsg def self.marshal(_o) @@ -376,7 +384,7 @@ describe GRPC::RpcServer do stub = EchoStub.new(@host, **client_opts) expect(stub.an_rpc(req, k1: 'v1', k2: 'v2')).to be_a(EchoMsg) wanted_md = [{ 'k1' => 'v1', 'k2' => 'v2' }] - expect(service.received_md).to eq(wanted_md) + check_md(wanted_md, service.received_md) @srv.stop t.join end @@ -391,7 +399,7 @@ describe GRPC::RpcServer do deadline = service.delay + 1.0 # wait for long enough expect(stub.an_rpc(req, deadline, k1: 'v1', k2: 'v2')).to be_a(EchoMsg) wanted_md = [{ 'k1' => 'v1', 'k2' => 'v2' }] - expect(service.received_md).to eq(wanted_md) + check_md(wanted_md, service.received_md) @srv.stop t.join end @@ -443,7 +451,7 @@ describe GRPC::RpcServer do expect(stub.an_rpc(req, k1: 'v1', k2: 'v2')).to be_a(EchoMsg) wanted_md = [{ 'k1' => 'updated-v1', 'k2' => 'v2', 'jwt_aud_uri' => "https://#{@host}/EchoService" }] - expect(service.received_md).to eq(wanted_md) + check_md(wanted_md, service.received_md) @srv.stop t.join end @@ -535,7 +543,9 @@ describe GRPC::RpcServer do 'method' => '/EchoService/an_rpc', 'connect_k1' => 'connect_v1' } - expect(op.metadata).to eq(wanted_md) + wanted_md.each do |key, value| + expect(op.metadata[key]).to eq(value) + end @srv.stop t.join end diff --git a/templates/Makefile.template b/templates/Makefile.template index 044db4dbfe..1e46db11dc 100644 --- a/templates/Makefile.template +++ b/templates/Makefile.template @@ -159,7 +159,7 @@ CC_tsan = clang CXX_tsan = clang++ LD_tsan = clang LDXX_tsan = clang++ -CPPFLAGS_tsan = -O0 -fsanitize=thread -fno-omit-frame-pointer +CPPFLAGS_tsan = -O0 -fsanitize=thread -fno-omit-frame-pointer -Wno-error=unused-command-line-argument LDFLAGS_tsan = -fsanitize=thread DEFINES_tsan = NDEBUG GRPC_TEST_SLOWDOWN_BUILD_FACTOR=10 @@ -169,7 +169,7 @@ CC_asan = clang CXX_asan = clang++ LD_asan = clang LDXX_asan = clang++ -CPPFLAGS_asan = -O0 -fsanitize=address -fno-omit-frame-pointer +CPPFLAGS_asan = -O0 -fsanitize=address -fno-omit-frame-pointer -Wno-error=unused-command-line-argument LDFLAGS_asan = -fsanitize=address DEFINES_asan = GRPC_TEST_SLOWDOWN_BUILD_FACTOR=3 @@ -179,7 +179,7 @@ CC_msan = clang CXX_msan = clang++-libc++ LD_msan = clang LDXX_msan = clang++-libc++ -CPPFLAGS_msan = -O0 -fsanitize=memory -fsanitize-memory-track-origins -fno-omit-frame-pointer -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1 +CPPFLAGS_msan = -O0 -fsanitize=memory -fsanitize-memory-track-origins -fno-omit-frame-pointer -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1 -Wno-error=unused-command-line-argument OPENSSL_CFLAGS_msan = -DPURIFY LDFLAGS_msan = -fsanitize=memory -DGTEST_HAS_TR1_TUPLE=0 -DGTEST_USE_OWN_TR1_TUPLE=1 DEFINES_msan = NDEBUG GRPC_TEST_SLOWDOWN_BUILD_FACTOR=4 @@ -190,7 +190,7 @@ CC_ubsan = clang CXX_ubsan = clang++ LD_ubsan = clang LDXX_ubsan = clang++ -CPPFLAGS_ubsan = -O1 -fsanitize=undefined -fno-omit-frame-pointer +CPPFLAGS_ubsan = -O1 -fsanitize=undefined -fno-omit-frame-pointer -Wno-error=unused-command-line-argument OPENSSL_CFLAGS_ubsan = -DPURIFY LDFLAGS_ubsan = -fsanitize=undefined DEFINES_ubsan = NDEBUG GRPC_TEST_SLOWDOWN_BUILD_FACTOR=3 @@ -255,10 +255,6 @@ HOST_CXX = $(CXX) HOST_LD = $(LD) HOST_LDXX = $(LDXX) -CPPFLAGS += $(CPPFLAGS_$(CONFIG)) -DEFINES += $(DEFINES_$(CONFIG)) INSTALL_PREFIX=\"$(prefix)\" -LDFLAGS += $(LDFLAGS_$(CONFIG)) - ifdef EXTRA_DEFINES DEFINES += $(EXTRA_DEFINES) endif @@ -272,6 +268,10 @@ endif CPPFLAGS += -g -Wall -Wextra -Werror -Wno-long-long -Wno-unused-parameter LDFLAGS += -g +CPPFLAGS += $(CPPFLAGS_$(CONFIG)) +DEFINES += $(DEFINES_$(CONFIG)) INSTALL_PREFIX=\"$(prefix)\" +LDFLAGS += $(LDFLAGS_$(CONFIG)) + ifneq ($(SYSTEM),MINGW32) PIC_CPPFLAGS = -fPIC CPPFLAGS += -fPIC @@ -816,7 +816,7 @@ run_dep_checks: $(LIBDIR)/$(CONFIG)/zlib/libz.a: $(E) "[MAKE] Building zlib" - $(Q)(cd third_party/zlib ; CC="$(CC)" CFLAGS="$(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG))" ./configure --static) + $(Q)(cd third_party/zlib ; CC="$(CC)" CFLAGS="$(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(ZLIB_CFLAGS_EXTRA)" ./configure --static) $(Q)$(MAKE) -C third_party/zlib clean $(Q)$(MAKE) -C third_party/zlib $(Q)mkdir -p $(LIBDIR)/$(CONFIG)/zlib @@ -825,7 +825,7 @@ $(LIBDIR)/$(CONFIG)/zlib/libz.a: $(LIBDIR)/$(CONFIG)/openssl/libssl.a: $(E) "[MAKE] Building openssl for $(SYSTEM)" ifeq ($(SYSTEM),Darwin) - $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG))" ./Configure darwin64-x86_64-cc) + $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_EXTRA)" ./Configure darwin64-x86_64-cc) else ifeq ($(SYSTEM),MINGW32) @echo "We currently don't have a good way to compile OpenSSL in-place under msys." @@ -846,7 +846,7 @@ ifeq ($(SYSTEM),MINGW32) @echo " CPPFLAGS=-I/c/OpenSSL-Win64/include LDFLAGS=-L/c/OpenSSL-Win64 make" @false else - $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG))" ./config no-asm $(OPENSSL_CONFIG_$(CONFIG))) + $(Q)(cd third_party/openssl ; CC="$(CC) $(PIC_CPPFLAGS) -fvisibility=hidden $(CPPFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_$(CONFIG)) $(OPENSSL_CFLAGS_EXTRA)" ./config no-asm $(OPENSSL_CONFIG_$(CONFIG))) endif endif $(Q)$(MAKE) -C third_party/openssl clean @@ -860,7 +860,7 @@ third_party/protobuf/configure: $(LIBDIR)/$(CONFIG)/protobuf/libprotobuf.a: third_party/protobuf/configure $(E) "[MAKE] Building protobuf" - $(Q)(cd third_party/protobuf ; CC="$(CC)" CXX="$(CXX)" LDFLAGS="$(LDFLAGS_$(CONFIG)) -g" CPPFLAGS="$(PIC_CPPFLAGS) $(CPPFLAGS_$(CONFIG)) -g" ./configure --disable-shared --enable-static) + $(Q)(cd third_party/protobuf ; CC="$(CC)" CXX="$(CXX)" LDFLAGS="$(LDFLAGS_$(CONFIG)) -g $(PROTOBUF_LDFLAGS_EXTRA)" CPPFLAGS="$(PIC_CPPFLAGS) $(CPPFLAGS_$(CONFIG)) -g $(PROTOBUF_CPPFLAGS_EXTRA)" ./configure --disable-shared --enable-static) $(Q)$(MAKE) -C third_party/protobuf clean $(Q)$(MAKE) -C third_party/protobuf $(Q)mkdir -p $(LIBDIR)/$(CONFIG)/protobuf diff --git a/test/core/end2end/dualstack_socket_test.c b/test/core/end2end/dualstack_socket_test.c index e7c113b36b..453376caec 100644 --- a/test/core/end2end/dualstack_socket_test.c +++ b/test/core/end2end/dualstack_socket_test.c @@ -222,6 +222,10 @@ void test_connect(const char *server_host, const char *client_host, int port, drain_cq(cq); grpc_completion_queue_destroy(cq); + grpc_metadata_array_destroy(&initial_metadata_recv); + grpc_metadata_array_destroy(&trailing_metadata_recv); + grpc_metadata_array_destroy(&request_metadata_recv); + grpc_call_details_destroy(&call_details); gpr_free(details); } diff --git a/test/core/iomgr/alarm_test.c b/test/core/iomgr/alarm_test.c index 362eb5fe63..55aa517529 100644 --- a/test/core/iomgr/alarm_test.c +++ b/test/core/iomgr/alarm_test.c @@ -41,6 +41,7 @@ #include <stdlib.h> #include <string.h> +#include <grpc/grpc.h> #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/sync.h> @@ -100,7 +101,7 @@ static void test_grpc_alarm(void) { alarm_arg arg2; void *fdone; - grpc_iomgr_init(); + grpc_init(); arg.counter = 0; arg.success = SUCCESS_NOT_SET; @@ -113,7 +114,7 @@ static void test_grpc_alarm(void) { gpr_event_init(&arg.fcb_arg); grpc_alarm_init(&alarm, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(100), alarm_cb, &arg, - gpr_now(GPR_CLOCK_REALTIME)); + gpr_now(GPR_CLOCK_MONOTONIC)); alarm_deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1); gpr_mu_lock(&arg.mu); @@ -165,7 +166,7 @@ static void test_grpc_alarm(void) { gpr_event_init(&arg2.fcb_arg); grpc_alarm_init(&alarm_to_cancel, GRPC_TIMEOUT_MILLIS_TO_DEADLINE(100), - alarm_cb, &arg2, gpr_now(GPR_CLOCK_REALTIME)); + alarm_cb, &arg2, gpr_now(GPR_CLOCK_MONOTONIC)); grpc_alarm_cancel(&alarm_to_cancel); alarm_deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(1); @@ -214,7 +215,7 @@ static void test_grpc_alarm(void) { gpr_mu_destroy(&arg2.mu); gpr_free(arg2.followup_closure); - grpc_iomgr_shutdown(); + grpc_shutdown(); } int main(int argc, char **argv) { diff --git a/test/core/iomgr/endpoint_tests.c b/test/core/iomgr/endpoint_tests.c index 0cfba5fac8..cb6adc58cf 100644 --- a/test/core/iomgr/endpoint_tests.c +++ b/test/core/iomgr/endpoint_tests.c @@ -254,7 +254,7 @@ static void read_and_write_test(grpc_endpoint_test_config config, gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); while (!state.read_done || !state.write_done) { - GPR_ASSERT(gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0); + GPR_ASSERT(gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), deadline) < 0); grpc_pollset_work(g_pollset, deadline); } gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); @@ -350,14 +350,14 @@ static void shutdown_during_write_test(grpc_endpoint_test_config config, deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(10); gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); while (!write_st.done) { - GPR_ASSERT(gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0); + GPR_ASSERT(gpr_time_cmp(gpr_now(deadline.clock_type), deadline) < 0); grpc_pollset_work(g_pollset, deadline); } gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); grpc_endpoint_destroy(write_st.ep); gpr_mu_lock(GRPC_POLLSET_MU(g_pollset)); while (!read_st.done) { - GPR_ASSERT(gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0); + GPR_ASSERT(gpr_time_cmp(gpr_now(deadline.clock_type), deadline) < 0); grpc_pollset_work(g_pollset, deadline); } gpr_mu_unlock(GRPC_POLLSET_MU(g_pollset)); diff --git a/test/core/iomgr/fd_posix_test.c b/test/core/iomgr/fd_posix_test.c index cd268661db..7d00c098cc 100644 --- a/test/core/iomgr/fd_posix_test.c +++ b/test/core/iomgr/fd_posix_test.c @@ -249,7 +249,7 @@ static int server_start(server *sv) { static void server_wait_and_shutdown(server *sv) { gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (!sv->done) { - grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_REALTIME)); + grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_MONOTONIC)); } gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } @@ -356,7 +356,7 @@ static void client_start(client *cl, int port) { static void client_wait_and_shutdown(client *cl) { gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (!cl->done) { - grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_REALTIME)); + grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_MONOTONIC)); } gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); } @@ -445,7 +445,7 @@ static void test_grpc_fd_change(void) { /* And now wait for it to run. */ gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (a.cb_that_ran == NULL) { - grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_REALTIME)); + grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_MONOTONIC)); } GPR_ASSERT(a.cb_that_ran == first_read_callback); gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset)); @@ -463,7 +463,7 @@ static void test_grpc_fd_change(void) { gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (b.cb_that_ran == NULL) { - grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_REALTIME)); + grpc_pollset_work(&g_pollset, gpr_inf_future(GPR_CLOCK_MONOTONIC)); } /* Except now we verify that second_read_callback ran instead */ GPR_ASSERT(b.cb_that_ran == second_read_callback); diff --git a/test/core/iomgr/tcp_client_posix_test.c b/test/core/iomgr/tcp_client_posix_test.c index 637886a738..38b7b5909d 100644 --- a/test/core/iomgr/tcp_client_posix_test.c +++ b/test/core/iomgr/tcp_client_posix_test.c @@ -196,13 +196,13 @@ void test_times_out(void) { gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset)); while (gpr_time_cmp(gpr_time_add(connect_deadline, gpr_time_from_seconds(2, GPR_TIMESPAN)), - gpr_now(GPR_CLOCK_REALTIME)) > 0) { + gpr_now(connect_deadline.clock_type)) > 0) { int is_after_deadline = - gpr_time_cmp(connect_deadline, gpr_now(GPR_CLOCK_REALTIME)) <= 0; + gpr_time_cmp(connect_deadline, gpr_now(GPR_CLOCK_MONOTONIC)) <= 0; if (is_after_deadline && gpr_time_cmp(gpr_time_add(connect_deadline, gpr_time_from_seconds(1, GPR_TIMESPAN)), - gpr_now(GPR_CLOCK_REALTIME)) > 0) { + gpr_now(GPR_CLOCK_MONOTONIC)) > 0) { /* allow some slack before insisting that things be done */ } else { GPR_ASSERT(g_connections_complete == diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c index 83252a889b..f8d0fe8217 100644 --- a/test/core/iomgr/tcp_server_posix_test.c +++ b/test/core/iomgr/tcp_server_posix_test.c @@ -135,7 +135,7 @@ static void test_connect(int n) { gpr_log(GPR_DEBUG, "wait"); while (g_nconnects == nconnects_before && - gpr_time_cmp(deadline, gpr_now(GPR_CLOCK_REALTIME)) > 0) { + gpr_time_cmp(deadline, gpr_now(deadline.clock_type)) > 0) { grpc_pollset_work(&g_pollset, deadline); } gpr_log(GPR_DEBUG, "wait done"); diff --git a/test/core/util/test_config.h b/test/core/util/test_config.h index 063c797ce9..7028ade7b2 100644 --- a/test/core/util/test_config.h +++ b/test/core/util/test_config.h @@ -52,12 +52,12 @@ extern "C" { (GRPC_TEST_SLOWDOWN_BUILD_FACTOR * GRPC_TEST_SLOWDOWN_MACHINE_FACTOR) #define GRPC_TIMEOUT_SECONDS_TO_DEADLINE(x) \ - gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), \ + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), \ gpr_time_from_micros(GRPC_TEST_SLOWDOWN_FACTOR * 1e6 * (x), \ GPR_TIMESPAN)) #define GRPC_TIMEOUT_MILLIS_TO_DEADLINE(x) \ - gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), \ + gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), \ gpr_time_from_micros(GRPC_TEST_SLOWDOWN_FACTOR * 1e3 * (x), \ GPR_TIMESPAN)) diff --git a/test/cpp/client/credentials_test.cc b/test/cpp/client/credentials_test.cc index ee94f455a4..bbf7705f0a 100644 --- a/test/cpp/client/credentials_test.cc +++ b/test/cpp/client/credentials_test.cc @@ -47,7 +47,7 @@ class CredentialsTest : public ::testing::Test { TEST_F(CredentialsTest, InvalidServiceAccountCreds) { std::shared_ptr<Credentials> bad1 = ServiceAccountCredentials("", "", 1); - EXPECT_EQ(nullptr, bad1.get()); + EXPECT_EQ(static_cast<Credentials *>(nullptr), bad1.get()); } } // namespace testing diff --git a/test/cpp/common/secure_auth_context_test.cc b/test/cpp/common/secure_auth_context_test.cc index f18a04178e..d0243a5432 100644 --- a/test/cpp/common/secure_auth_context_test.cc +++ b/test/cpp/common/secure_auth_context_test.cc @@ -92,26 +92,6 @@ TEST_F(SecureAuthContextTest, Iterators) { EXPECT_EQ("bar", p2.second); ++iter; EXPECT_EQ(context.end(), iter); - // Range-based for loop test. - int i = 0; - for (auto p : context) { - switch (i++) { - case 0: - EXPECT_EQ("name", p.first); - EXPECT_EQ("chapi", p.second); - break; - case 1: - EXPECT_EQ("name", p.first); - EXPECT_EQ("chapo", p.second); - break; - case 2: - EXPECT_EQ("foo", p.first); - EXPECT_EQ("bar", p.second); - break; - default: - EXPECT_TRUE(0); - } - } } } // namespace diff --git a/test/cpp/end2end/async_end2end_test.cc b/test/cpp/end2end/async_end2end_test.cc index 117d8bb9fa..b95bdf6b9b 100644 --- a/test/cpp/end2end/async_end2end_test.cc +++ b/test/cpp/end2end/async_end2end_test.cc @@ -415,7 +415,7 @@ TEST_F(AsyncEnd2endTest, ClientInitialMetadataRpc) { auto client_initial_metadata = srv_ctx.client_metadata(); EXPECT_EQ(meta1.second, client_initial_metadata.find(meta1.first)->second); EXPECT_EQ(meta2.second, client_initial_metadata.find(meta2.first)->second); - EXPECT_EQ(static_cast<size_t>(2), client_initial_metadata.size()); + EXPECT_GE(client_initial_metadata.size(), static_cast<size_t>(2)); send_response.set_message(recv_request.message()); response_writer.Finish(send_response, Status::OK, tag(3)); @@ -563,7 +563,7 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { auto client_initial_metadata = srv_ctx.client_metadata(); EXPECT_EQ(meta1.second, client_initial_metadata.find(meta1.first)->second); EXPECT_EQ(meta2.second, client_initial_metadata.find(meta2.first)->second); - EXPECT_EQ(static_cast<size_t>(2), client_initial_metadata.size()); + EXPECT_GE(client_initial_metadata.size(), static_cast<size_t>(2)); srv_ctx.AddInitialMetadata(meta3.first, meta3.second); srv_ctx.AddInitialMetadata(meta4.first, meta4.second); @@ -574,7 +574,7 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { auto server_initial_metadata = cli_ctx.GetServerInitialMetadata(); EXPECT_EQ(meta3.second, server_initial_metadata.find(meta3.first)->second); EXPECT_EQ(meta4.second, server_initial_metadata.find(meta4.first)->second); - EXPECT_EQ(static_cast<size_t>(2), server_initial_metadata.size()); + EXPECT_GE(server_initial_metadata.size(), static_cast<size_t>(2)); send_response.set_message(recv_request.message()); srv_ctx.AddTrailingMetadata(meta5.first, meta5.second); @@ -590,7 +590,7 @@ TEST_F(AsyncEnd2endTest, MetadataRpc) { auto server_trailing_metadata = cli_ctx.GetServerTrailingMetadata(); EXPECT_EQ(meta5.second, server_trailing_metadata.find(meta5.first)->second); EXPECT_EQ(meta6.second, server_trailing_metadata.find(meta6.first)->second); - EXPECT_EQ(static_cast<size_t>(2), server_trailing_metadata.size()); + EXPECT_GE(server_trailing_metadata.size(), static_cast<size_t>(2)); } } // namespace } // namespace testing diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 8b4424c735..20e4c4ed55 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -249,9 +249,10 @@ class End2endTest : public ::testing::Test { void TearDown() GRPC_OVERRIDE { server_->Shutdown(); } void ResetStub() { - std::shared_ptr<ChannelInterface> channel = - CreateChannel(server_address_.str(), FakeTransportSecurityCredentials(), - ChannelArguments()); + ChannelArguments args; + args.SetString(GRPC_ARG_SECONDARY_USER_AGENT_STRING, "end2end_test"); + std::shared_ptr<ChannelInterface> channel = CreateChannel( + server_address_.str(), FakeTransportSecurityCredentials(), args); stub_ = std::move(grpc::cpp::test::util::TestService::NewStub(channel)); } @@ -508,7 +509,7 @@ TEST_F(End2endTest, DiffPackageServices) { // rpc and stream should fail on bad credentials. TEST_F(End2endTest, BadCredentials) { std::shared_ptr<Credentials> bad_creds = ServiceAccountCredentials("", "", 1); - EXPECT_EQ(nullptr, bad_creds.get()); + EXPECT_EQ(static_cast<Credentials *>(nullptr), bad_creds.get()); std::shared_ptr<ChannelInterface> channel = CreateChannel(server_address_.str(), bad_creds, ChannelArguments()); std::unique_ptr<grpc::cpp::test::util::TestService::Stub> stub( diff --git a/test/cpp/interop/client.cc b/test/cpp/interop/client.cc index 1f1e6c1306..d0393fafb2 100644 --- a/test/cpp/interop/client.cc +++ b/test/cpp/interop/client.cc @@ -69,6 +69,7 @@ DEFINE_string(test_case, "large_unary", "compute_engine_creds: large_unary with compute engine auth; " "jwt_token_creds: large_unary with JWT token auth; " "oauth2_auth_token: raw oauth2 access token auth; " + "per_rpc_creds: raw oauth2 access token on a single rpc; " "all : all of above."); DEFINE_string(default_service_account, "", "Email of GCE default service account"); @@ -117,6 +118,9 @@ int main(int argc, char** argv) { } else if (FLAGS_test_case == "oauth2_auth_token") { grpc::string json_key = GetServiceAccountJsonKey(); client.DoOauth2AuthToken(json_key, FLAGS_oauth_scope); + } else if (FLAGS_test_case == "per_rpc_creds") { + grpc::string json_key = GetServiceAccountJsonKey(); + client.DoPerRpcCreds(json_key, FLAGS_oauth_scope); } else if (FLAGS_test_case == "all") { client.DoEmpty(); client.DoLargeUnary(); @@ -133,6 +137,7 @@ int main(int argc, char** argv) { client.DoServiceAccountCreds(json_key, FLAGS_oauth_scope); client.DoJwtTokenCreds(json_key); client.DoOauth2AuthToken(json_key, FLAGS_oauth_scope); + client.DoPerRpcCreds(json_key, FLAGS_oauth_scope); } // compute_engine_creds only runs in GCE. } else { @@ -142,7 +147,7 @@ int main(int argc, char** argv) { "large_unary|client_streaming|server_streaming|half_duplex|ping_pong|" "cancel_after_begin|cancel_after_first_response|" "timeout_on_sleeping_server|service_account_creds|compute_engine_creds|" - "jwt_token_creds|oauth2_auth_token", + "jwt_token_creds|oauth2_auth_token|per_rpc_creds", FLAGS_test_case.c_str()); ret = 1; } diff --git a/test/cpp/interop/interop_client.cc b/test/cpp/interop/interop_client.cc index 4dcd8ad061..e5c0e4631f 100644 --- a/test/cpp/interop/interop_client.cc +++ b/test/cpp/interop/interop_client.cc @@ -41,8 +41,10 @@ #include <grpc/support/log.h> #include <grpc++/channel_interface.h> #include <grpc++/client_context.h> +#include <grpc++/credentials.h> #include <grpc++/status.h> #include <grpc++/stream.h> +#include "test/cpp/interop/client_helper.h" #include "test/proto/test.grpc.pb.h" #include "test/proto/empty.grpc.pb.h" #include "test/proto/messages.grpc.pb.h" @@ -166,6 +168,32 @@ void InteropClient::DoOauth2AuthToken(const grpc::string& username, gpr_log(GPR_INFO, "Unary with oauth2 access token credentials done."); } +void InteropClient::DoPerRpcCreds(const grpc::string& username, + const grpc::string& oauth_scope) { + gpr_log(GPR_INFO, + "Sending a unary rpc with per-rpc raw oauth2 access token ..."); + SimpleRequest request; + SimpleResponse response; + request.set_fill_username(true); + request.set_fill_oauth_scope(true); + std::unique_ptr<TestService::Stub> stub(TestService::NewStub(channel_)); + + ClientContext context; + grpc::string access_token = GetOauth2AccessToken(); + std::shared_ptr<Credentials> creds = AccessTokenCredentials(access_token); + context.set_credentials(creds); + + Status s = stub->UnaryCall(&context, request, &response); + + AssertOkOrPrintErrorStatus(s); + GPR_ASSERT(!response.username().empty()); + GPR_ASSERT(!response.oauth_scope().empty()); + GPR_ASSERT(username.find(response.username()) != grpc::string::npos); + const char* oauth_scope_str = response.oauth_scope().c_str(); + GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos); + gpr_log(GPR_INFO, "Unary with per-rpc oauth2 access token done."); +} + void InteropClient::DoJwtTokenCreds(const grpc::string& username) { gpr_log(GPR_INFO, "Sending a large unary rpc with JWT token credentials ..."); SimpleRequest request; diff --git a/test/cpp/interop/interop_client.h b/test/cpp/interop/interop_client.h index 67eecd9ccc..bf8188325e 100644 --- a/test/cpp/interop/interop_client.h +++ b/test/cpp/interop/interop_client.h @@ -71,6 +71,9 @@ class InteropClient { // username is a string containing the user email void DoOauth2AuthToken(const grpc::string& username, const grpc::string& oauth_scope); + // username is a string containing the user email + void DoPerRpcCreds(const grpc::string& username, + const grpc::string& oauth_scope); private: void PerformLargeUnary(SimpleRequest* request, SimpleResponse* response); diff --git a/test/cpp/qps/qps_test.cc b/test/cpp/qps/qps_test.cc index 07b4834cc0..7b93443f7c 100644 --- a/test/cpp/qps/qps_test.cc +++ b/test/cpp/qps/qps_test.cc @@ -44,8 +44,8 @@ namespace grpc { namespace testing { -static const int WARMUP = 5; -static const int BENCHMARK = 10; +static const int WARMUP = 20; +static const int BENCHMARK = 40; static void RunQPS() { gpr_log(GPR_INFO, "Running QPS test"); @@ -53,8 +53,8 @@ static void RunQPS() { ClientConfig client_config; client_config.set_client_type(ASYNC_CLIENT); client_config.set_enable_ssl(false); - client_config.set_outstanding_rpcs_per_channel(1000); - client_config.set_client_channels(8); + client_config.set_outstanding_rpcs_per_channel(10); + client_config.set_client_channels(800); client_config.set_payload_size(1); client_config.set_async_client_threads(8); client_config.set_rpc_type(UNARY); @@ -62,7 +62,7 @@ static void RunQPS() { ServerConfig server_config; server_config.set_server_type(ASYNC_SERVER); server_config.set_enable_ssl(false); - server_config.set_threads(4); + server_config.set_threads(8); const auto result = RunScenario(client_config, 1, server_config, 1, WARMUP, BENCHMARK, -2); diff --git a/tools/jenkins/docker_run_jenkins.sh b/tools/jenkins/docker_run_jenkins.sh index eb6c9144c6..0a5516c30d 100755 --- a/tools/jenkins/docker_run_jenkins.sh +++ b/tools/jenkins/docker_run_jenkins.sh @@ -41,5 +41,5 @@ git clone --recursive /var/local/jenkins/grpc /var/local/git/grpc cd /var/local/git/grpc nvm use 0.12 rvm use ruby-2.1 -tools/run_tests/prepare_travis.sh -$arch tools/run_tests/run_tests.py -t -c $config -l $language -x report.xml + +setarch $arch tools/run_tests/run_tests.py -t -c $config -l $language -x report.xml diff --git a/tools/jenkins/grpc_jenkins_slave/Dockerfile b/tools/jenkins/grpc_jenkins_slave/Dockerfile index f37c0b9103..9058b0498e 100644 --- a/tools/jenkins/grpc_jenkins_slave/Dockerfile +++ b/tools/jenkins/grpc_jenkins_slave/Dockerfile @@ -38,8 +38,10 @@ RUN apt-get update && apt-get install -y \ autotools-dev \ build-essential \ bzip2 \ + ccache \ curl \ gcc \ + gcc-multilib \ git \ libc6 \ libc6-dbg \ @@ -55,6 +57,14 @@ RUN apt-get update && apt-get install -y \ wget \ zip && apt-get clean +# Prepare ccache +RUN ln -s /usr/bin/ccache /usr/local/bin/gcc +RUN ln -s /usr/bin/ccache /usr/local/bin/g++ +RUN ln -s /usr/bin/ccache /usr/local/bin/cc +RUN ln -s /usr/bin/ccache /usr/local/bin/c++ +RUN ln -s /usr/bin/ccache /usr/local/bin/clang +RUN ln -s /usr/bin/ccache /usr/local/bin/clang++ + ################## # C++ dependencies RUN apt-get update && apt-get -y install libgflags-dev libgtest-dev libc++-dev clang diff --git a/tools/jenkins/grpc_jenkins_slave_32bits/Dockerfile b/tools/jenkins/grpc_jenkins_slave_32bits/Dockerfile new file mode 100644 index 0000000000..2beaf9a820 --- /dev/null +++ b/tools/jenkins/grpc_jenkins_slave_32bits/Dockerfile @@ -0,0 +1,152 @@ +# Copyright 2015, 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. + +# A work-in-progress Dockerfile that allows running gRPC test suites +# inside a docker container. + +FROM 32bit/debian:jessie + +# Install Git. +RUN apt-get update && apt-get install -y \ + autoconf \ + autotools-dev \ + build-essential \ + bzip2 \ + ccache \ + curl \ + gcc \ + gcc-multilib \ + git \ + libc6 \ + libc6-dbg \ + libc6-dev \ + libgtest-dev \ + libtool \ + make \ + strace \ + python-dev \ + python-setuptools \ + telnet \ + unzip \ + wget \ + zip && apt-get clean + +# Prepare ccache +RUN ln -s /usr/bin/ccache /usr/local/bin/gcc +RUN ln -s /usr/bin/ccache /usr/local/bin/g++ +RUN ln -s /usr/bin/ccache /usr/local/bin/cc +RUN ln -s /usr/bin/ccache /usr/local/bin/c++ +RUN ln -s /usr/bin/ccache /usr/local/bin/clang +RUN ln -s /usr/bin/ccache /usr/local/bin/clang++ + +################## +# C++ dependencies +RUN apt-get update && apt-get -y install libgflags-dev libgtest-dev libc++-dev clang + +################# +# C# dependencies + +# Update to a newer version of mono +RUN apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 3FA7E0328081BFF6A14DA29AA6A19B38D3D831EF +RUN echo "deb http://download.mono-project.com/repo/debian wheezy main" | tee /etc/apt/sources.list.d/mono-xamarin.list +RUN echo "deb http://download.mono-project.com/repo/debian wheezy-apache24-compat main" | tee -a /etc/apt/sources.list.d/mono-xamarin.list +RUN echo "deb http://download.mono-project.com/repo/debian wheezy-libjpeg62-compat main" | tee -a /etc/apt/sources.list.d/mono-xamarin.list +RUN echo "deb http://download.mono-project.com/repo/debian wheezy-libtiff-compat main" | tee -a /etc/apt/sources.list.d/mono-xamarin.list + +# Install dependencies +RUN apt-get update && apt-get -y dist-upgrade && apt-get install -y \ + mono-devel \ + nunit \ + nunit-console \ + monodevelop + +# Download NuGet +RUN cd /var/local && wget www.nuget.org/NuGet.exe +ENV NUGET mono /var/local/NuGet.exe + +# TODO(jtattermusch): add dependencies for other languages + +################## +# Node dependencies + +# Install nvm +RUN touch .profile +RUN curl -o- https://raw.githubusercontent.com/creationix/nvm/v0.25.4/install.sh | bash +RUN /bin/bash -l -c "nvm install 0.12" + +################## +# Ruby dependencies + +# Install rvm +RUN gpg --keyserver hkp://keys.gnupg.net --recv-keys 409B6B1796C275462A1703113804BB82D39DC0E3 +RUN \curl -sSL https://get.rvm.io | bash -s stable + +# Install Ruby 2.1 +RUN /bin/bash -l -c "rvm install ruby-2.1" +RUN /bin/bash -l -c "rvm use --default ruby-2.1" +RUN /bin/bash -l -c "echo 'gem: --no-ri --no-rdoc' > ~/.gemrc" +RUN /bin/bash -l -c "echo 'export PATH=/usr/local/rvm/bin:$PATH' >> ~/.bashrc" +RUN /bin/bash -l -c "echo 'rvm --default use ruby-2.1' >> ~/.bashrc" +RUN /bin/bash -l -c "gem install bundler --no-ri --no-rdoc" + +################## +# Python dependencies + +# Install dependencies + +RUN apt-get update && apt-get install -y \ + python-all-dev \ + python3-all-dev \ + python-pip \ + python-virtualenv + +# Install Python packages from PyPI +RUN pip install futures==2.2.0 enum34==1.0.4 protobuf==3.0.0a2 + +# For sanity test +RUN pip install simplejson mako + +################## +# PHP dependencies + +# Install dependencies + +RUN /bin/bash -l -c "echo 'deb http://packages.dotdeb.org wheezy-php55 all' \ + >> /etc/apt/sources.list.d/dotdeb.list" +RUN /bin/bash -l -c "echo 'deb-src http://packages.dotdeb.org wheezy-php55 all' \ + >> /etc/apt/sources.list.d/dotdeb.list" +RUN wget http://www.dotdeb.org/dotdeb.gpg -O- | apt-key add - + +RUN apt-get update && apt-get install -y \ + git php5 php5-dev phpunit unzip + +RUN mkdir /var/local/jenkins + +# Define the default command. +CMD ["bash"] diff --git a/tools/jenkins/run_distribution.sh b/tools/jenkins/run_distribution.sh index fd318692ac..49b7d306d1 100755 --- a/tools/jenkins/run_distribution.sh +++ b/tools/jenkins/run_distribution.sh @@ -32,6 +32,17 @@ # linuxbrew installation of a selected language set -ex +# Our homebrew installation script command, per language +# Can be used in both linux and macos +if [ "$language" == "core" ]; then + command="curl -fsSL https://goo.gl/getgrpc | bash -" +elif [[ "python nodejs ruby php" =~ "$language" ]]; then + command="curl -fsSL https://goo.gl/getgrpc | bash -s $language" +else + echo "unsupported language $language" + exit 1 +fi + if [ "$platform" == "linux" ]; then if [ "$dist_channel" == "homebrew" ]; then @@ -39,19 +50,10 @@ if [ "$platform" == "linux" ]; then sha1=$(sha1sum tools/jenkins/grpc_linuxbrew/Dockerfile | cut -f1 -d\ ) DOCKER_IMAGE_NAME=grpc_linuxbrew_$sha1 + # build docker image, contains all pre-requisites docker build -t $DOCKER_IMAGE_NAME tools/jenkins/grpc_linuxbrew - supported="python nodejs ruby php" - - if [ "$language" == "core" ]; then - command="curl -fsSL https://goo.gl/getgrpc | bash -" - elif [[ "$supported" =~ "$language" ]]; then - command="curl -fsSL https://goo.gl/getgrpc | bash -s $language" - else - echo "unsupported language $language" - exit 1 - fi - + # run per-language homebrew installation script docker run $DOCKER_IMAGE_NAME bash -l \ -c "nvm use 0.12; \ npm set unsafe-perm true; \ @@ -66,26 +68,70 @@ if [ "$platform" == "linux" ]; then elif [ "$platform" == "macos" ]; then if [ "$dist_channel" == "homebrew" ]; then - which brew # TODO: for debug, can be removed later + + echo "Formulas installed by system-wide homebrew (before)" brew list -l - dir=/tmp/homebrew-test-$language - rm -rf $dir - mkdir -p $dir - git clone https://github.com/Homebrew/homebrew.git $dir - cd $dir - # TODO: Uncomment these when the general structure of the script is verified - # PATH=$dir/bin:$PATH brew tap homebrew/dupes - # PATH=$dir/bin:$PATH brew install zlib - # PATH=$dir/bin:$PATH brew install openssl - # PATH=$dir/bin:$PATH brew tap grpc/grpc - # PATH=$dir/bin:$PATH brew install --without-python google-protobuf - # PATH=$dir/bin:$PATH brew install grpc - PATH=$dir/bin:$PATH brew list -l + + # Save the original PATH so that we can run the system `brew` command + # again at the end of the script + export ORIGINAL_PATH=$PATH + + # Set up temp directories for test installation of homebrew + brew_root=/tmp/homebrew-test-$language + rm -rf $brew_root + mkdir -p $brew_root + git clone https://github.com/Homebrew/homebrew.git $brew_root + + # Make sure we are operating at the right copy of temp homebrew + # installation + export PATH=$brew_root/bin:$PATH + + # Set up right environment for each language + case $language in + *python*) + rm -rf jenkins_python_venv + virtualenv jenkins_python_venv + source jenkins_python_venv/bin/activate + ;; + *nodejs*) + export PATH=$HOME/.nvm/versions/node/v0.12.7/bin:$PATH + ;; + *ruby*) + export PATH=/usr/local/rvm/rubies/ruby-2.2.1/bin:$PATH + ;; + *php*) + export CFLAGS="-Wno-parentheses-equality" + ;; + esac + + # Run our homebrew installation script + bash -c "$command" + + # Uninstall / clean up per-language modules/extensions after the test + case $language in + *python*) + deactivate + rm -rf jenkins_python_venv + ;; + *nodejs*) + npm list -g | grep grpc + npm uninstall -g grpc + ;; + *ruby*) + gem list | grep grpc + gem uninstall grpc + ;; + *php*) + rm grpc.so + ;; + esac + + # Clean up + rm -rf $brew_root + + echo "Formulas installed by system-wide homebrew (after, should be unaffected)" + export PATH=$ORIGINAL_PATH brew list -l - cd ~/ - rm -rf $dir - echo $PATH # TODO: for debug, can be removed later - brew list -l # TODO: for debug, can be removed later else echo "Unsupported $platform dist_channel $dist_channel" diff --git a/tools/jenkins/run_jenkins.sh b/tools/jenkins/run_jenkins.sh index 8cb85cb12b..56f9e82ca5 100755 --- a/tools/jenkins/run_jenkins.sh +++ b/tools/jenkins/run_jenkins.sh @@ -46,6 +46,7 @@ case $platform in i386) arch="i386" platform="linux" + docker_suffix=_32bits ;; esac @@ -57,11 +58,13 @@ then git_root=`pwd` cd - + mkdir -p /tmp/ccache + # Use image name based on Dockerfile checksum - DOCKER_IMAGE_NAME=grpc_jenkins_slave_`sha1sum tools/jenkins/grpc_jenkins_slave/Dockerfile | cut -f1 -d\ ` + DOCKER_IMAGE_NAME=grpc_jenkins_slave$docker_suffix_`sha1sum tools/jenkins/grpc_jenkins_slave/Dockerfile | cut -f1 -d\ ` # Make sure docker image has been built. Should be instantaneous if so. - docker build -t $DOCKER_IMAGE_NAME tools/jenkins/grpc_jenkins_slave + docker build -t $DOCKER_IMAGE_NAME tools/jenkins/grpc_jenkins_slave$docker_suffix # Create a local branch so the child Docker script won't complain git branch jenkins-docker @@ -74,8 +77,10 @@ then -e "config=$config" \ -e "language=$language" \ -e "arch=$arch" \ + -e CCACHE_DIR=/tmp/ccache \ -i \ -v "$git_root:/var/local/jenkins/grpc" \ + -v /tmp/ccache:/tmp/ccache \ --cidfile=docker.cid \ $DOCKER_IMAGE_NAME \ bash -l /var/local/jenkins/grpc/tools/jenkins/docker_run_jenkins.sh || DOCKER_FAILED="true" |