diff options
Diffstat (limited to 'src')
103 files changed, 4179 insertions, 565 deletions
diff --git a/src/core/channel/census_filter.c b/src/core/channel/census_filter.c index 43ef5fb3ff..d996c3475e 100644 --- a/src/core/channel/census_filter.c +++ b/src/core/channel/census_filter.c @@ -151,7 +151,7 @@ static void client_init_call_elem(grpc_call_element* elem, call_data* d = elem->call_data; GPR_ASSERT(d != NULL); init_rpc_stats(&d->stats); - d->start_ts = gpr_now(); + d->start_ts = gpr_now(GPR_CLOCK_REALTIME); d->op_id = census_tracing_start_op(); if (initial_op) client_mutate_op(elem, initial_op); } @@ -169,7 +169,7 @@ static void server_init_call_elem(grpc_call_element* elem, call_data* d = elem->call_data; GPR_ASSERT(d != NULL); init_rpc_stats(&d->stats); - d->start_ts = gpr_now(); + d->start_ts = gpr_now(GPR_CLOCK_REALTIME); d->op_id = census_tracing_start_op(); if (initial_op) server_mutate_op(elem, initial_op); } @@ -177,8 +177,8 @@ static void server_init_call_elem(grpc_call_element* elem, static void server_destroy_call_elem(grpc_call_element* elem) { call_data* d = elem->call_data; GPR_ASSERT(d != NULL); - d->stats.elapsed_time_ms = - gpr_timespec_to_micros(gpr_time_sub(gpr_now(), d->start_ts)); + d->stats.elapsed_time_ms = gpr_timespec_to_micros( + gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME), d->start_ts)); census_record_rpc_server_stats(d->op_id, &d->stats); census_tracing_end_op(d->op_id); } diff --git a/src/core/client_config/subchannel.c b/src/core/client_config/subchannel.c index 6cf9062ab0..8cdad1015f 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_timespec now = gpr_now(GPR_CLOCK_REALTIME); c->next_attempt = now; c->backoff_delta = gpr_time_from_seconds(1); @@ -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()); + grpc_alarm_init(&c->alarm, c->next_attempt, on_alarm, c, gpr_now(GPR_CLOCK_REALTIME)); gpr_mu_unlock(&c->mu); } } diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index c507e7c26a..cca92d3b32 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -59,7 +59,7 @@ static void background_callback_executor(void *ignored) { while (!g_shutdown) { gpr_timespec deadline = gpr_inf_future; gpr_timespec short_deadline = - gpr_time_add(gpr_now(), gpr_time_from_millis(100)); + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100)); if (g_cbs_head) { grpc_iomgr_closure *closure = g_cbs_head; g_cbs_head = closure->next; @@ -67,7 +67,8 @@ 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(), &deadline)) { + } else if (grpc_alarm_check(&g_mu, gpr_now(GPR_CLOCK_REALTIME), + &deadline)) { } else { gpr_mu_unlock(&g_mu); gpr_sleep_until(gpr_time_min(short_deadline, deadline)); @@ -89,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()); + grpc_alarm_list_init(gpr_now(GPR_CLOCK_REALTIME)); g_root_object.next = g_root_object.prev = &g_root_object; g_root_object.name = "root"; grpc_iomgr_platform_init(); @@ -110,21 +111,27 @@ void grpc_iomgr_shutdown(void) { grpc_iomgr_object *obj; grpc_iomgr_closure *closure; gpr_timespec shutdown_deadline = - gpr_time_add(gpr_now(), gpr_time_from_seconds(10)); + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_seconds(10)); + gpr_timespec last_warning_time = gpr_now(GPR_CLOCK_REALTIME); gpr_mu_lock(&g_mu); g_shutdown = 1; while (g_cbs_head != NULL || g_root_object.next != &g_root_object) { - if (g_cbs_head != NULL && g_root_object.next != &g_root_object) { - gpr_log(GPR_DEBUG, - "Waiting for %d iomgr objects to be destroyed and executing " - "final callbacks", - count_objects()); - } else if (g_cbs_head != NULL) { - gpr_log(GPR_DEBUG, "Executing final iomgr callbacks"); - } else { - gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed", - count_objects()); + if (gpr_time_cmp( + gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME), last_warning_time), + gpr_time_from_seconds(1)) >= 0) { + if (g_cbs_head != NULL && g_root_object.next != &g_root_object) { + gpr_log(GPR_DEBUG, + "Waiting for %d iomgr objects to be destroyed and executing " + "final callbacks", + count_objects()); + } else if (g_cbs_head != NULL) { + gpr_log(GPR_DEBUG, "Executing final iomgr callbacks"); + } else { + gpr_log(GPR_DEBUG, "Waiting for %d iomgr objects to be destroyed", + count_objects()); + } + last_warning_time = gpr_now(GPR_CLOCK_REALTIME); } if (g_cbs_head) { do { @@ -145,9 +152,9 @@ void grpc_iomgr_shutdown(void) { if (g_root_object.next != &g_root_object) { int timeout = 0; gpr_timespec short_deadline = - gpr_time_add(gpr_now(), gpr_time_from_millis(100)); + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(100)); while (gpr_cv_wait(&g_rcv, &g_mu, short_deadline) && g_cbs_head == NULL) { - if (gpr_time_cmp(gpr_now(), shutdown_deadline) > 0) { + if (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), shutdown_deadline) > 0) { timeout = 1; break; } diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index 1900bbf9e1..3746c8edaf 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -105,10 +105,11 @@ static void multipoll_with_epoll_pollset_maybe_work( * here. */ - timeout_ms = grpc_poll_deadline_to_millis_timeout(deadline, now); pollset->counter += 1; gpr_mu_unlock(&pollset->mu); + timeout_ms = grpc_poll_deadline_to_millis_timeout(deadline, now); + do { ep_rv = epoll_wait(h->epoll_fd, ep_ev, GRPC_EPOLL_MAX_EVENTS, timeout_ms); if (ep_rv < 0) { diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index 12496440de..85101764d2 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -122,7 +122,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_timespec now = gpr_now(GPR_CLOCK_REALTIME); if (gpr_time_cmp(now, deadline) > 0) { return 0; } @@ -187,15 +187,16 @@ void grpc_pollset_destroy(grpc_pollset *pollset) { gpr_mu_destroy(&pollset->mu); } -int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline, gpr_timespec now) { +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) == 0) { return -1; } if (gpr_time_cmp( - deadline, - gpr_time_add(now, gpr_time_from_micros(max_spin_polling_us))) <= 0) { + deadline, + gpr_time_add(now, gpr_time_from_micros(max_spin_polling_us))) <= 0) { return 0; } timeout = gpr_time_sub(deadline, now); diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c index 8d6bc79c96..24226cc980 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(); + now = gpr_now(GPR_CLOCK_REALTIME); if (gpr_time_cmp(now, deadline) > 0) { return 0 /* GPR_FALSE */; } @@ -86,8 +86,6 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) { return 1 /* GPR_TRUE */; } -void grpc_pollset_kick(grpc_pollset *p) { - gpr_cv_signal(&p->cv); -} +void grpc_pollset_kick(grpc_pollset *p) { gpr_cv_signal(&p->cv); } #endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c index fbf3fdc949..897408ded2 100644 --- a/src/core/iomgr/socket_windows.c +++ b/src/core/iomgr/socket_windows.c @@ -45,11 +45,14 @@ #include "src/core/iomgr/socket_windows.h" grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name) { + char *final_name; grpc_winsocket *r = gpr_malloc(sizeof(grpc_winsocket)); memset(r, 0, sizeof(grpc_winsocket)); r->socket = socket; gpr_mu_init(&r->state_mu); - grpc_iomgr_register_object(&r->iomgr_object, name); + gpr_asprintf(&final_name, "%s:socket=0x%p", name, r); + grpc_iomgr_register_object(&r->iomgr_object, final_name); + gpr_free(final_name); grpc_iocp_add_socket(r); return r; } diff --git a/src/core/iomgr/tcp_client_posix.c b/src/core/iomgr/tcp_client_posix.c index d981aaf028..dc0489e64f 100644 --- a/src/core/iomgr/tcp_client_posix.c +++ b/src/core/iomgr/tcp_client_posix.c @@ -114,6 +114,7 @@ static void on_writable(void *acp, int success) { void (*cb)(void *arg, grpc_endpoint *tcp) = ac->cb; void *cb_arg = ac->cb_arg; + gpr_mu_lock(&ac->mu); if (success) { do { so_error_size = sizeof(so_error); @@ -139,6 +140,7 @@ static void on_writable(void *acp, int success) { opened too many network connections. The "easy" fix: don't do that! */ gpr_log(GPR_ERROR, "kernel out of buffers"); + gpr_mu_unlock(&ac->mu); grpc_fd_notify_on_write(ac->fd, &ac->write_closure); return; } else { @@ -165,10 +167,11 @@ static void on_writable(void *acp, int success) { abort(); finish: - gpr_mu_lock(&ac->mu); - if (!ep) { + if (ep == NULL) { grpc_pollset_set_del_fd(ac->interested_parties, ac->fd); grpc_fd_orphan(ac->fd, NULL, "tcp_client_orphan"); + } else { + ac->fd = NULL; } done = (--ac->refs == 0); gpr_mu_unlock(&ac->mu); @@ -250,7 +253,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()); + grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, + gpr_now(GPR_CLOCK_REALTIME)); 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 b1a169b519..16741452b9 100644 --- a/src/core/iomgr/tcp_client_windows.c +++ b/src/core/iomgr/tcp_client_windows.c @@ -215,7 +215,8 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), ac->refs = 2; ac->aborted = 0; - grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now()); + grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, + gpr_now(GPR_CLOCK_REALTIME)); socket->write_info.outstanding = 1; grpc_socket_notify_on_write(socket, on_connect, ac); return; diff --git a/src/core/profiling/timers_preciseclock.h b/src/core/profiling/timers_preciseclock.h index 163d52b797..5c251b47e6 100644 --- a/src/core/profiling/timers_preciseclock.h +++ b/src/core/profiling/timers_preciseclock.h @@ -82,7 +82,7 @@ struct grpc_precise_clock { gpr_timespec clock; }; static void grpc_precise_clock_now(grpc_precise_clock* clk) { - clk->clock = gpr_now(); + clk->clock = gpr_now(GPR_CLOCK_REALTIME); } #define GRPC_PRECISE_CLOCK_FORMAT "%ld.%09d" #define GRPC_PRECISE_CLOCK_PRINTF_ARGS(clk) \ diff --git a/src/core/security/client_auth_filter.c b/src/core/security/client_auth_filter.c index 16f2abe04b..9e49a807f1 100644 --- a/src/core/security/client_auth_filter.c +++ b/src/core/security/client_auth_filter.c @@ -61,6 +61,7 @@ typedef struct { grpc_transport_stream_op op; size_t op_md_idx; int sent_initial_metadata; + gpr_uint8 security_context_set; grpc_linked_mdelem md_links[MAX_CREDENTIALS_METADATA_COUNT]; } call_data; @@ -199,8 +200,22 @@ static void auth_start_transport_op(grpc_call_element *elem, channel_data *chand = elem->channel_data; grpc_linked_mdelem *l; size_t i; + grpc_client_security_context* sec_ctx = NULL; - /* TODO(jboeuf): write the call auth context. */ + if (calld->security_context_set == 0) { + calld->security_context_set = 1; + GPR_ASSERT(op->context); + if (op->context[GRPC_CONTEXT_SECURITY].value == NULL) { + op->context[GRPC_CONTEXT_SECURITY].value = + grpc_client_security_context_create(); + op->context[GRPC_CONTEXT_SECURITY].destroy = + grpc_client_security_context_destroy; + } + sec_ctx = op->context[GRPC_CONTEXT_SECURITY].value; + GRPC_AUTH_CONTEXT_UNREF(sec_ctx->auth_context, "client auth filter"); + sec_ctx->auth_context = GRPC_AUTH_CONTEXT_REF( + chand->security_connector->base.auth_context, "client_auth_filter"); + } if (op->bind_pollset) { calld->pollset = op->bind_pollset; @@ -263,6 +278,7 @@ static void init_call_elem(grpc_call_element *elem, calld->method = NULL; calld->pollset = NULL; calld->sent_initial_metadata = 0; + calld->security_context_set = 0; GPR_ASSERT(!initial_op || !initial_op->send_ops); } diff --git a/src/core/security/credentials.c b/src/core/security/credentials.c index 8d694c2f79..230f0dfb85 100644 --- a/src/core/security/credentials.c +++ b/src/core/security/credentials.c @@ -41,7 +41,6 @@ #include "src/core/json/json.h" #include "src/core/httpcli/httpcli.h" #include "src/core/iomgr/iomgr.h" -#include "src/core/security/json_token.h" #include "src/core/support/string.h" #include <grpc/support/alloc.h> @@ -52,12 +51,12 @@ /* -- Common. -- */ -typedef struct { +struct grpc_credentials_metadata_request { grpc_credentials *creds; grpc_credentials_metadata_cb cb; grpc_iomgr_closure *on_simulated_token_fetch_done_closure; void *user_data; -} grpc_credentials_metadata_request; +}; static grpc_credentials_metadata_request * grpc_credentials_metadata_request_create(grpc_credentials *creds, @@ -152,16 +151,6 @@ grpc_security_status grpc_server_credentials_create_security_connector( /* -- Ssl credentials. -- */ -typedef struct { - grpc_credentials base; - grpc_ssl_config config; -} grpc_ssl_credentials; - -typedef struct { - grpc_server_credentials base; - grpc_ssl_server_config config; -} grpc_ssl_server_credentials; - static void ssl_destroy(grpc_credentials *creds) { grpc_ssl_credentials *c = (grpc_ssl_credentials *)creds; if (c->config.pem_root_certs != NULL) gpr_free(c->config.pem_root_certs); @@ -326,22 +315,6 @@ grpc_server_credentials *grpc_ssl_server_credentials_create( /* -- Jwt credentials -- */ -typedef struct { - grpc_credentials base; - - /* Have a simple cache for now with just 1 entry. We could have a map based on - the service_url for a more sophisticated one. */ - gpr_mu cache_mu; - struct { - grpc_credentials_md_store *jwt_md; - char *service_url; - gpr_timespec jwt_expiration; - } cached; - - grpc_auth_json_key key; - gpr_timespec jwt_lifetime; -} grpc_jwt_credentials; - static void jwt_reset_cache(grpc_jwt_credentials *c) { if (c->cached.jwt_md != NULL) { grpc_credentials_md_store_unref(c->cached.jwt_md); @@ -384,7 +357,8 @@ static void jwt_get_request_metadata(grpc_credentials *creds, if (c->cached.service_url != NULL && strcmp(c->cached.service_url, service_url) == 0 && c->cached.jwt_md != NULL && - (gpr_time_cmp(gpr_time_sub(c->cached.jwt_expiration, gpr_now()), + (gpr_time_cmp(gpr_time_sub(c->cached.jwt_expiration, + gpr_now(GPR_CLOCK_REALTIME)), refresh_threshold) > 0)) { jwt_md = grpc_credentials_md_store_ref(c->cached.jwt_md); } @@ -401,7 +375,8 @@ static void jwt_get_request_metadata(grpc_credentials *creds, char *md_value; gpr_asprintf(&md_value, "Bearer %s", jwt); gpr_free(jwt); - c->cached.jwt_expiration = gpr_time_add(gpr_now(), c->jwt_lifetime); + c->cached.jwt_expiration = + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), c->jwt_lifetime); c->cached.service_url = gpr_strdup(service_url); c->cached.jwt_md = grpc_credentials_md_store_create(1); grpc_credentials_md_store_add_cstrings( @@ -424,10 +399,9 @@ static grpc_credentials_vtable jwt_vtable = { jwt_destroy, jwt_has_request_metadata, jwt_has_request_metadata_only, jwt_get_request_metadata, NULL}; -grpc_credentials *grpc_jwt_credentials_create(const char *json_key, - gpr_timespec token_lifetime) { +grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( + grpc_auth_json_key key, gpr_timespec token_lifetime) { grpc_jwt_credentials *c; - grpc_auth_json_key key = grpc_auth_json_key_create_from_string(json_key); if (!grpc_auth_json_key_is_valid(&key)) { gpr_log(GPR_ERROR, "Invalid input for jwt credentials creation"); return NULL; @@ -444,25 +418,13 @@ grpc_credentials *grpc_jwt_credentials_create(const char *json_key, return &c->base; } -/* -- Oauth2TokenFetcher credentials -- */ - -/* This object is a base for credentials that need to acquire an oauth2 token - from an http service. */ - -typedef void (*grpc_fetch_oauth2_func)(grpc_credentials_metadata_request *req, - grpc_httpcli_context *http_context, - grpc_pollset *pollset, - grpc_httpcli_response_cb response_cb, - gpr_timespec deadline); +grpc_credentials *grpc_jwt_credentials_create(const char *json_key, + gpr_timespec token_lifetime) { + return grpc_jwt_credentials_create_from_auth_json_key( + grpc_auth_json_key_create_from_string(json_key), token_lifetime); +} -typedef struct { - grpc_credentials base; - gpr_mu mu; - grpc_credentials_md_store *access_token_md; - gpr_timespec token_expiration; - grpc_httpcli_context httpcli_context; - grpc_fetch_oauth2_func fetch_func; -} grpc_oauth2_token_fetcher_credentials; +/* -- Oauth2TokenFetcher credentials -- */ static void oauth2_token_fetcher_destroy(grpc_credentials *creds) { grpc_oauth2_token_fetcher_credentials *c = @@ -585,7 +547,8 @@ static void on_oauth2_token_fetcher_http_response( status = grpc_oauth2_token_fetcher_credentials_parse_server_response( response, &c->access_token_md, &token_lifetime); if (status == GRPC_CREDENTIALS_OK) { - c->token_expiration = gpr_time_add(gpr_now(), token_lifetime); + c->token_expiration = + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), token_lifetime); r->cb(r->user_data, c->access_token_md->entries, c->access_token_md->num_entries, status); } else { @@ -607,8 +570,9 @@ static void oauth2_token_fetcher_get_request_metadata( { gpr_mu_lock(&c->mu); if (c->access_token_md != NULL && - (gpr_time_cmp(gpr_time_sub(c->token_expiration, gpr_now()), - refresh_threshold) > 0)) { + (gpr_time_cmp( + gpr_time_sub(c->token_expiration, gpr_now(GPR_CLOCK_REALTIME)), + refresh_threshold) > 0)) { cached_access_token_md = grpc_credentials_md_store_ref(c->access_token_md); } @@ -622,7 +586,7 @@ static void oauth2_token_fetcher_get_request_metadata( c->fetch_func( grpc_credentials_metadata_request_create(creds, cb, user_data), &c->httpcli_context, pollset, on_oauth2_token_fetcher_http_response, - gpr_time_add(gpr_now(), refresh_threshold)); + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), refresh_threshold)); } } @@ -669,13 +633,6 @@ grpc_credentials *grpc_compute_engine_credentials_create(void) { /* -- ServiceAccount credentials. -- */ -typedef struct { - grpc_oauth2_token_fetcher_credentials base; - grpc_auth_json_key key; - char *scope; - gpr_timespec token_lifetime; -} grpc_service_account_credentials; - static void service_account_destroy(grpc_credentials *creds) { grpc_service_account_credentials *c = (grpc_service_account_credentials *)creds; @@ -746,11 +703,6 @@ grpc_credentials *grpc_service_account_credentials_create( /* -- RefreshToken credentials. -- */ -typedef struct { - grpc_oauth2_token_fetcher_credentials base; - grpc_auth_refresh_token refresh_token; -} grpc_refresh_token_credentials; - static void refresh_token_destroy(grpc_credentials *creds) { grpc_refresh_token_credentials *c = (grpc_refresh_token_credentials *)creds; grpc_auth_refresh_token_destruct(&c->refresh_token); @@ -786,12 +738,9 @@ static void refresh_token_fetch_oauth2( gpr_free(body); } -grpc_credentials *grpc_refresh_token_credentials_create( - const char *json_refresh_token) { +grpc_credentials *grpc_refresh_token_credentials_create_from_auth_refresh_token( + grpc_auth_refresh_token refresh_token) { grpc_refresh_token_credentials *c; - grpc_auth_refresh_token refresh_token = - grpc_auth_refresh_token_create_from_string(json_refresh_token); - if (!grpc_auth_refresh_token_is_valid(&refresh_token)) { gpr_log(GPR_ERROR, "Invalid input for refresh token credentials creation"); return NULL; @@ -804,13 +753,13 @@ grpc_credentials *grpc_refresh_token_credentials_create( return &c->base.base; } -/* -- Fake Oauth2 credentials. -- */ +grpc_credentials *grpc_refresh_token_credentials_create( + const char *json_refresh_token) { + return grpc_refresh_token_credentials_create_from_auth_refresh_token( + grpc_auth_refresh_token_create_from_string(json_refresh_token)); +} -typedef struct { - grpc_credentials base; - grpc_credentials_md_store *access_token_md; - int is_async; -} grpc_fake_oauth2_credentials; +/* -- Fake Oauth2 credentials. -- */ static void fake_oauth2_destroy(grpc_credentials *creds) { grpc_fake_oauth2_credentials *c = (grpc_fake_oauth2_credentials *)creds; @@ -877,11 +826,6 @@ grpc_credentials *grpc_fake_oauth2_credentials_create( /* -- Oauth2 Access Token credentials. -- */ -typedef struct { - grpc_credentials base; - grpc_credentials_md_store *access_token_md; -} grpc_access_token_credentials; - static void access_token_destroy(grpc_credentials *creds) { grpc_access_token_credentials *c = (grpc_access_token_credentials *)creds; grpc_credentials_md_store_unref(c->access_token_md); @@ -997,12 +941,6 @@ grpc_server_credentials *grpc_fake_transport_security_server_credentials_create( /* -- Composite credentials. -- */ typedef struct { - grpc_credentials base; - grpc_credentials_array inner; - grpc_credentials *connector_creds; -} grpc_composite_credentials; - -typedef struct { grpc_composite_credentials *composite_creds; size_t creds_index; grpc_credentials_md_store *md_elems; @@ -1232,11 +1170,6 @@ grpc_credentials *grpc_credentials_contains_type( /* -- IAM credentials. -- */ -typedef struct { - grpc_credentials base; - grpc_credentials_md_store *iam_md; -} grpc_iam_credentials; - static void iam_destroy(grpc_credentials *creds) { grpc_iam_credentials *c = (grpc_iam_credentials *)creds; grpc_credentials_md_store_unref(c->iam_md); diff --git a/src/core/security/credentials.h b/src/core/security/credentials.h index 75af73a0c6..d988901cf7 100644 --- a/src/core/security/credentials.h +++ b/src/core/security/credentials.h @@ -39,6 +39,8 @@ #include <grpc/grpc_security.h> #include <grpc/support/sync.h> +#include "src/core/httpcli/httpcli.h" +#include "src/core/security/json_token.h" #include "src/core/security/security_connector.h" struct grpc_httpcli_response; @@ -178,11 +180,22 @@ grpc_credentials_status grpc_oauth2_token_fetcher_credentials_parse_server_response( const struct grpc_httpcli_response *response, grpc_credentials_md_store **token_md, gpr_timespec *token_lifetime); +void grpc_flush_cached_google_default_credentials(void); /* Simulates an oauth2 token fetch with the specified value for testing. */ grpc_credentials *grpc_fake_oauth2_credentials_create( const char *token_md_value, int is_async); +/* Private constructor for jwt credentials from an already parsed json key. + Takes ownership of the key. */ +grpc_credentials *grpc_jwt_credentials_create_from_auth_json_key( + grpc_auth_json_key key, gpr_timespec token_lifetime); + +/* Private constructor for refresh token credentials from an already parsed + refresh token. Takes ownership of the refresh token. */ +grpc_credentials *grpc_refresh_token_credentials_create_from_auth_refresh_token( + grpc_auth_refresh_token token); + /* --- grpc_server_credentials. --- */ typedef struct { @@ -199,4 +212,103 @@ struct grpc_server_credentials { grpc_security_status grpc_server_credentials_create_security_connector( grpc_server_credentials *creds, grpc_security_connector **sc); +/* -- Ssl credentials. -- */ + +typedef struct { + grpc_credentials base; + grpc_ssl_config config; +} grpc_ssl_credentials; + +typedef struct { + grpc_server_credentials base; + grpc_ssl_server_config config; +} grpc_ssl_server_credentials; + +/* -- Jwt credentials -- */ + +typedef struct { + grpc_credentials base; + + /* Have a simple cache for now with just 1 entry. We could have a map based on + the service_url for a more sophisticated one. */ + gpr_mu cache_mu; + struct { + grpc_credentials_md_store *jwt_md; + char *service_url; + gpr_timespec jwt_expiration; + } cached; + + grpc_auth_json_key key; + gpr_timespec jwt_lifetime; +} grpc_jwt_credentials; + +/* -- Oauth2TokenFetcher credentials -- + + This object is a base for credentials that need to acquire an oauth2 token + from an http service. */ + +typedef struct grpc_credentials_metadata_request + grpc_credentials_metadata_request; + +typedef void (*grpc_fetch_oauth2_func)(grpc_credentials_metadata_request *req, + grpc_httpcli_context *http_context, + grpc_pollset *pollset, + grpc_httpcli_response_cb response_cb, + gpr_timespec deadline); + +typedef struct { + grpc_credentials base; + gpr_mu mu; + grpc_credentials_md_store *access_token_md; + gpr_timespec token_expiration; + grpc_httpcli_context httpcli_context; + grpc_fetch_oauth2_func fetch_func; +} grpc_oauth2_token_fetcher_credentials; + +/* -- ServiceAccount credentials. -- */ + +typedef struct { + grpc_oauth2_token_fetcher_credentials base; + grpc_auth_json_key key; + char *scope; + gpr_timespec token_lifetime; +} grpc_service_account_credentials; + +/* -- RefreshToken credentials. -- */ + +typedef struct { + grpc_oauth2_token_fetcher_credentials base; + grpc_auth_refresh_token refresh_token; +} grpc_refresh_token_credentials; + +/* -- Oauth2 Access Token credentials. -- */ + +typedef struct { + grpc_credentials base; + grpc_credentials_md_store *access_token_md; +} grpc_access_token_credentials; + +/* -- Fake Oauth2 credentials. -- */ + +typedef struct { + grpc_credentials base; + grpc_credentials_md_store *access_token_md; + int is_async; +} grpc_fake_oauth2_credentials; + +/* -- IAM credentials. -- */ + +typedef struct { + grpc_credentials base; + grpc_credentials_md_store *iam_md; +} grpc_iam_credentials; + +/* -- Composite credentials. -- */ + +typedef struct { + grpc_credentials base; + grpc_credentials_array inner; + grpc_credentials *connector_creds; +} grpc_composite_credentials; + #endif /* GRPC_INTERNAL_CORE_SECURITY_CREDENTIALS_H */ diff --git a/src/core/security/google_default_credentials.c b/src/core/security/google_default_credentials.c index 5822ce6337..f622deff42 100644 --- a/src/core/security/google_default_credentials.c +++ b/src/core/security/google_default_credentials.c @@ -46,7 +46,6 @@ /* -- Constants. -- */ #define GRPC_COMPUTE_ENGINE_DETECTION_HOST "metadata.google.internal" -#define GRPC_GOOGLE_CREDENTIALS_ENV_VAR "GOOGLE_APPLICATION_CREDENTIALS" /* -- Default credentials. -- */ @@ -104,9 +103,10 @@ static int is_stack_running_on_compute_engine(void) { grpc_httpcli_context_init(&context); - grpc_httpcli_get(&context, &detector.pollset, &request, - gpr_time_add(gpr_now(), max_detection_delay), - on_compute_engine_detection_http_response, &detector); + grpc_httpcli_get( + &context, &detector.pollset, &request, + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), max_detection_delay), + on_compute_engine_detection_http_response, &detector); /* Block until we get the response. This is not ideal but this should only be called once for the lifetime of the process by the default credentials. */ @@ -123,36 +123,40 @@ static int is_stack_running_on_compute_engine(void) { } /* Takes ownership of creds_path if not NULL. */ -static grpc_credentials *create_jwt_creds_from_path(char *creds_path) { +static grpc_credentials *create_default_creds_from_path(char *creds_path) { + grpc_json *json = NULL; + grpc_auth_json_key key; + grpc_auth_refresh_token token; grpc_credentials *result = NULL; - gpr_slice creds_data; + gpr_slice creds_data = gpr_empty_slice(); int file_ok = 0; - if (creds_path == NULL) return NULL; - creds_data = gpr_load_file(creds_path, 1, &file_ok); - gpr_free(creds_path); - if (file_ok) { - result = grpc_jwt_credentials_create( - (const char *)GPR_SLICE_START_PTR(creds_data), - grpc_max_auth_token_lifetime); - gpr_slice_unref(creds_data); + if (creds_path == NULL) goto end; + creds_data = gpr_load_file(creds_path, 0, &file_ok); + if (!file_ok) goto end; + json = grpc_json_parse_string_with_len( + (char *)GPR_SLICE_START_PTR(creds_data), GPR_SLICE_LENGTH(creds_data)); + if (json == NULL) goto end; + + /* First, try an auth json key. */ + key = grpc_auth_json_key_create_from_json(json); + if (grpc_auth_json_key_is_valid(&key)) { + result = grpc_jwt_credentials_create_from_auth_json_key( + key, grpc_max_auth_token_lifetime); + goto end; } - return result; -} -/* Takes ownership of creds_path if not NULL. */ -static grpc_credentials *create_refresh_token_creds_from_path( - char *creds_path) { - grpc_credentials *result = NULL; - gpr_slice creds_data; - int file_ok = 0; - if (creds_path == NULL) return NULL; - creds_data = gpr_load_file(creds_path, 1, &file_ok); - gpr_free(creds_path); - if (file_ok) { - result = grpc_refresh_token_credentials_create( - (const char *)GPR_SLICE_START_PTR(creds_data)); - gpr_slice_unref(creds_data); + /* Then try a refresh token if the auth json key was invalid. */ + token = grpc_auth_refresh_token_create_from_json(json); + if (grpc_auth_refresh_token_is_valid(&token)) { + result = + grpc_refresh_token_credentials_create_from_auth_refresh_token(token); + goto end; } + +end: + if (creds_path != NULL) gpr_free(creds_path); + gpr_slice_unref(creds_data); + if (json != NULL) grpc_json_destroy(json); return result; } @@ -170,12 +174,12 @@ grpc_credentials *grpc_google_default_credentials_create(void) { } /* First, try the environment variable. */ - result = - create_jwt_creds_from_path(gpr_getenv(GRPC_GOOGLE_CREDENTIALS_ENV_VAR)); + result = create_default_creds_from_path( + gpr_getenv(GRPC_GOOGLE_CREDENTIALS_ENV_VAR)); if (result != NULL) goto end; /* Then the well-known file. */ - result = create_refresh_token_creds_from_path( + result = create_default_creds_from_path( grpc_get_well_known_google_credentials_file_path()); if (result != NULL) goto end; @@ -193,11 +197,24 @@ end: if (!serving_cached_credentials && result != NULL) { /* Blend with default ssl credentials and add a global reference so that it can be cached and re-served. */ - result = grpc_composite_credentials_create( - grpc_ssl_credentials_create(NULL, NULL), result); - GPR_ASSERT(result != NULL); - default_credentials = grpc_credentials_ref(result); + grpc_credentials *ssl_creds = grpc_ssl_credentials_create(NULL, NULL); + default_credentials = grpc_credentials_ref(grpc_composite_credentials_create( + ssl_creds, result)); + GPR_ASSERT(default_credentials != NULL); + grpc_credentials_unref(ssl_creds); + grpc_credentials_unref(result); + result = default_credentials; } gpr_mu_unlock(&g_mu); return result; } + +void grpc_flush_cached_google_default_credentials(void) { + gpr_once_init(&g_once, init_default_credentials); + gpr_mu_lock(&g_mu); + if (default_credentials != NULL) { + grpc_credentials_unref(default_credentials); + default_credentials = NULL; + } + gpr_mu_unlock(&g_mu); +} diff --git a/src/core/security/json_token.c b/src/core/security/json_token.c index 6116f1d767..9b1ea255ae 100644 --- a/src/core/security/json_token.c +++ b/src/core/security/json_token.c @@ -46,17 +46,11 @@ #include <openssl/evp.h> #include <openssl/pem.h> -#include "src/core/json/json.h" - /* --- Constants. --- */ /* 1 hour max. */ const gpr_timespec grpc_max_auth_token_lifetime = {3600, 0}; -#define GRPC_AUTH_JSON_TYPE_INVALID "invalid" -#define GRPC_AUTH_JSON_TYPE_SERVICE_ACCOUNT "service_account" -#define GRPC_AUTH_JSON_TYPE_AUTHORIZED_USER "authorized_user" - #define GRPC_JWT_RSA_SHA256_ALGORITHM "RS256" #define GRPC_JWT_TYPE "JWT" @@ -66,7 +60,7 @@ static grpc_jwt_encode_and_sign_override g_jwt_encode_and_sign_override = NULL; /* --- grpc_auth_json_key. --- */ -static const char *json_get_string_property(grpc_json *json, +static const char *json_get_string_property(const grpc_json *json, const char *prop_name) { grpc_json *child; for (child = json->child; child != NULL; child = child->next) { @@ -79,7 +73,8 @@ static const char *json_get_string_property(grpc_json *json, return child->value; } -static int set_json_key_string_property(grpc_json *json, const char *prop_name, +static int set_json_key_string_property(const grpc_json *json, + const char *prop_name, char **json_key_field) { const char *prop_value = json_get_string_property(json, prop_name); if (prop_value == NULL) return 0; @@ -92,11 +87,8 @@ int grpc_auth_json_key_is_valid(const grpc_auth_json_key *json_key) { strcmp(json_key->type, GRPC_AUTH_JSON_TYPE_INVALID); } -grpc_auth_json_key grpc_auth_json_key_create_from_string( - const char *json_string) { +grpc_auth_json_key grpc_auth_json_key_create_from_json(const grpc_json *json) { grpc_auth_json_key result; - char *scratchpad = gpr_strdup(json_string); - grpc_json *json = grpc_json_parse_string(scratchpad); BIO *bio = NULL; const char *prop_value; int success = 0; @@ -104,7 +96,7 @@ grpc_auth_json_key grpc_auth_json_key_create_from_string( memset(&result, 0, sizeof(grpc_auth_json_key)); result.type = GRPC_AUTH_JSON_TYPE_INVALID; if (json == NULL) { - gpr_log(GPR_ERROR, "Invalid json string %s", json_string); + gpr_log(GPR_ERROR, "Invalid json."); goto end; } @@ -142,8 +134,16 @@ grpc_auth_json_key grpc_auth_json_key_create_from_string( end: if (bio != NULL) BIO_free(bio); - if (json != NULL) grpc_json_destroy(json); if (!success) grpc_auth_json_key_destruct(&result); + return result; +} + +grpc_auth_json_key grpc_auth_json_key_create_from_string( + const char *json_string) { + char *scratchpad = gpr_strdup(json_string); + grpc_json *json = grpc_json_parse_string(scratchpad); + grpc_auth_json_key result = grpc_auth_json_key_create_from_json(json); + if (json != NULL) grpc_json_destroy(json); gpr_free(scratchpad); return result; } @@ -207,7 +207,7 @@ static char *encoded_jwt_claim(const grpc_auth_json_key *json_key, grpc_json *child = NULL; char *json_str = NULL; char *result = NULL; - gpr_timespec now = gpr_now(); + gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); gpr_timespec expiration = gpr_time_add(now, token_lifetime); char now_str[GPR_LTOA_MIN_BUFSIZE]; char expiration_str[GPR_LTOA_MIN_BUFSIZE]; @@ -218,8 +218,8 @@ static char *encoded_jwt_claim(const grpc_auth_json_key *json_key, gpr_ltoa(now.tv_sec, now_str); gpr_ltoa(expiration.tv_sec, expiration_str); - child = create_child(NULL, json, "iss", json_key->client_email, - GRPC_JSON_STRING); + child = + create_child(NULL, json, "iss", json_key->client_email, GRPC_JSON_STRING); if (scope != NULL) { child = create_child(child, json, "scope", scope, GRPC_JSON_STRING); } else { @@ -342,18 +342,16 @@ int grpc_auth_refresh_token_is_valid( strcmp(refresh_token->type, GRPC_AUTH_JSON_TYPE_INVALID); } -grpc_auth_refresh_token grpc_auth_refresh_token_create_from_string( - const char *json_string) { +grpc_auth_refresh_token grpc_auth_refresh_token_create_from_json( + const grpc_json *json) { grpc_auth_refresh_token result; - char *scratchpad = gpr_strdup(json_string); - grpc_json *json = grpc_json_parse_string(scratchpad); const char *prop_value; int success = 0; memset(&result, 0, sizeof(grpc_auth_refresh_token)); result.type = GRPC_AUTH_JSON_TYPE_INVALID; if (json == NULL) { - gpr_log(GPR_ERROR, "Invalid json string %s", json_string); + gpr_log(GPR_ERROR, "Invalid json."); goto end; } @@ -374,8 +372,17 @@ grpc_auth_refresh_token grpc_auth_refresh_token_create_from_string( success = 1; end: - if (json != NULL) grpc_json_destroy(json); if (!success) grpc_auth_refresh_token_destruct(&result); + return result; +} + +grpc_auth_refresh_token grpc_auth_refresh_token_create_from_string( + const char *json_string) { + char *scratchpad = gpr_strdup(json_string); + grpc_json *json = grpc_json_parse_string(scratchpad); + grpc_auth_refresh_token result = + grpc_auth_refresh_token_create_from_json(json); + if (json != NULL) grpc_json_destroy(json); gpr_free(scratchpad); return result; } @@ -396,4 +403,3 @@ void grpc_auth_refresh_token_destruct(grpc_auth_refresh_token *refresh_token) { refresh_token->refresh_token = NULL; } } - diff --git a/src/core/security/json_token.h b/src/core/security/json_token.h index 197796ab4c..091dfefb6e 100644 --- a/src/core/security/json_token.h +++ b/src/core/security/json_token.h @@ -37,10 +37,16 @@ #include <grpc/support/slice.h> #include <openssl/rsa.h> +#include "src/core/json/json.h" + /* --- Constants. --- */ #define GRPC_JWT_OAUTH2_AUDIENCE "https://www.googleapis.com/oauth2/v3/token" +#define GRPC_AUTH_JSON_TYPE_INVALID "invalid" +#define GRPC_AUTH_JSON_TYPE_SERVICE_ACCOUNT "service_account" +#define GRPC_AUTH_JSON_TYPE_AUTHORIZED_USER "authorized_user" + /* --- auth_json_key parsing. --- */ typedef struct { @@ -59,6 +65,10 @@ int grpc_auth_json_key_is_valid(const grpc_auth_json_key *json_key); grpc_auth_json_key grpc_auth_json_key_create_from_string( const char *json_string); +/* Creates a json_key object from parsed json. Returns an invalid object if a + parsing error has been encountered. */ +grpc_auth_json_key grpc_auth_json_key_create_from_json(const grpc_json *json); + /* Destructs the object. */ void grpc_auth_json_key_destruct(grpc_auth_json_key *json_key); @@ -97,6 +107,11 @@ int grpc_auth_refresh_token_is_valid( grpc_auth_refresh_token grpc_auth_refresh_token_create_from_string( const char *json_string); +/* Creates a refresh token object from parsed json. Returns an invalid object if + a parsing error has been encountered. */ +grpc_auth_refresh_token grpc_auth_refresh_token_create_from_json( + const grpc_json *json); + /* Destructs the object. */ void grpc_auth_refresh_token_destruct(grpc_auth_refresh_token *refresh_token); diff --git a/src/core/security/jwt_verifier.c b/src/core/security/jwt_verifier.c index 01007a1a84..9140eb2ef7 100644 --- a/src/core/security/jwt_verifier.c +++ b/src/core/security/jwt_verifier.c @@ -189,7 +189,6 @@ struct grpc_jwt_claims { gpr_slice buffer; }; - void grpc_jwt_claims_destroy(grpc_jwt_claims *claims) { grpc_json_destroy(claims->json); gpr_slice_unref(claims->buffer); @@ -286,12 +285,14 @@ grpc_jwt_verifier_status grpc_jwt_claims_check(const grpc_jwt_claims *claims, GPR_ASSERT(claims != NULL); - skewed_now = gpr_time_add(gpr_now(), grpc_jwt_verifier_clock_skew); + skewed_now = + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_clock_skew); if (gpr_time_cmp(skewed_now, claims->nbf) < 0) { gpr_log(GPR_ERROR, "JWT is not valid yet."); return GRPC_JWT_VERIFIER_TIME_CONSTRAINT_FAILURE; } - skewed_now = gpr_time_sub(gpr_now(), grpc_jwt_verifier_clock_skew); + skewed_now = + gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_clock_skew); if (gpr_time_cmp(skewed_now, claims->exp) > 0) { gpr_log(GPR_ERROR, "JWT is expired."); return GRPC_JWT_VERIFIER_TIME_CONSTRAINT_FAILURE; @@ -327,10 +328,10 @@ typedef struct { /* Takes ownership of the header, claims and signature. */ static verifier_cb_ctx *verifier_cb_ctx_create( - grpc_jwt_verifier *verifier, grpc_pollset *pollset, - jose_header * header, grpc_jwt_claims *claims, const char *audience, - gpr_slice signature, const char *signed_jwt, size_t signed_jwt_len, - void *user_data, grpc_jwt_verification_done_cb cb) { + grpc_jwt_verifier *verifier, grpc_pollset *pollset, jose_header *header, + grpc_jwt_claims *claims, const char *audience, gpr_slice signature, + const char *signed_jwt, size_t signed_jwt_len, void *user_data, + grpc_jwt_verification_done_cb cb) { verifier_cb_ctx *ctx = gpr_malloc(sizeof(verifier_cb_ctx)); memset(ctx, 0, sizeof(verifier_cb_ctx)); ctx->verifier = verifier; @@ -604,7 +605,7 @@ end: static void on_openid_config_retrieved(void *user_data, const grpc_httpcli_response *response) { - const grpc_json* cur; + const grpc_json *cur; grpc_json *json = json_from_http(response); verifier_cb_ctx *ctx = (verifier_cb_ctx *)user_data; grpc_httpcli_request req; @@ -632,9 +633,10 @@ static void on_openid_config_retrieved(void *user_data, } else { *(req.host + (req.path - jwks_uri)) = '\0'; } - grpc_httpcli_get(&ctx->verifier->http_ctx, ctx->pollset, &req, - gpr_time_add(gpr_now(), grpc_jwt_verifier_max_delay), - on_keys_retrieved, ctx); + grpc_httpcli_get( + &ctx->verifier->http_ctx, ctx->pollset, &req, + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_max_delay), + on_keys_retrieved, ctx); grpc_json_destroy(json); gpr_free(req.host); return; @@ -645,8 +647,8 @@ error: verifier_cb_ctx_destroy(ctx); } -static email_key_mapping *verifier_get_mapping( - grpc_jwt_verifier *v, const char *email_domain) { +static email_key_mapping *verifier_get_mapping(grpc_jwt_verifier *v, + const char *email_domain) { size_t i; if (v->mappings == NULL) return NULL; for (i = 0; i < v->num_mappings; i++) { @@ -733,9 +735,10 @@ static void retrieve_key_and_verify(verifier_cb_ctx *ctx) { http_cb = on_openid_config_retrieved; } - grpc_httpcli_get(&ctx->verifier->http_ctx, ctx->pollset, &req, - gpr_time_add(gpr_now(), grpc_jwt_verifier_max_delay), - http_cb, ctx); + grpc_httpcli_get( + &ctx->verifier->http_ctx, ctx->pollset, &req, + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), grpc_jwt_verifier_max_delay), + http_cb, ctx); gpr_free(req.host); gpr_free(req.path); return; @@ -764,7 +767,7 @@ void grpc_jwt_verifier_verify(grpc_jwt_verifier *verifier, dot = strchr(cur, '.'); if (dot == NULL) goto error; json = parse_json_part_from_jwt(cur, dot - cur, &header_buffer); - if (json == NULL) goto error; + if (json == NULL) goto error; header = jose_header_from_json(json, header_buffer); if (header == NULL) goto error; @@ -772,7 +775,7 @@ void grpc_jwt_verifier_verify(grpc_jwt_verifier *verifier, dot = strchr(cur, '.'); if (dot == NULL) goto error; json = parse_json_part_from_jwt(cur, dot - cur, &claims_buffer); - if (json == NULL) goto error; + if (json == NULL) goto error; claims = grpc_jwt_claims_from_json(json, claims_buffer); if (claims == NULL) goto error; @@ -827,4 +830,3 @@ void grpc_jwt_verifier_destroy(grpc_jwt_verifier *v) { } gpr_free(v); } - diff --git a/src/core/security/security_context.c b/src/core/security/security_context.c index 4d56549f9b..8ce7876bd8 100644 --- a/src/core/security/security_context.c +++ b/src/core/security/security_context.c @@ -69,12 +69,20 @@ grpc_call_error grpc_call_set_credentials(grpc_call *call, return GRPC_CALL_OK; } -const grpc_auth_context *grpc_call_auth_context(grpc_call *call) { +grpc_auth_context *grpc_call_auth_context(grpc_call *call) { void *sec_ctx = grpc_call_context_get(call, GRPC_CONTEXT_SECURITY); if (sec_ctx == NULL) return NULL; return grpc_call_is_client(call) - ? ((grpc_client_security_context *)sec_ctx)->auth_context - : ((grpc_server_security_context *)sec_ctx)->auth_context; + ? GRPC_AUTH_CONTEXT_REF( + ((grpc_client_security_context *)sec_ctx)->auth_context, + "grpc_call_auth_context client") + : GRPC_AUTH_CONTEXT_REF( + ((grpc_server_security_context *)sec_ctx)->auth_context, + "grpc_call_auth_context server"); +} + +void grpc_auth_context_release(grpc_auth_context *context) { + GRPC_AUTH_CONTEXT_UNREF(context, "grpc_auth_context_unref"); } /* --- grpc_client_security_context --- */ diff --git a/src/core/security/security_context.h b/src/core/security/security_context.h index 20c4390898..76a45910bb 100644 --- a/src/core/security/security_context.h +++ b/src/core/security/security_context.h @@ -36,6 +36,10 @@ #include "src/core/security/credentials.h" +#ifdef __cplusplus +extern "C" { +#endif + /* --- grpc_auth_context --- High level authentication context object. Can optionally be chained. */ @@ -103,5 +107,9 @@ typedef struct { grpc_server_security_context *grpc_server_security_context_create(void); void grpc_server_security_context_destroy(void *ctx); +#ifdef __cplusplus +} +#endif + #endif /* GRPC_INTERNAL_CORE_SECURITY_SECURITY_CONTEXT_H */ diff --git a/src/core/statistics/census_rpc_stats.c b/src/core/statistics/census_rpc_stats.c index 0491c91947..3e571b1143 100644 --- a/src/core/statistics/census_rpc_stats.c +++ b/src/core/statistics/census_rpc_stats.c @@ -157,7 +157,7 @@ static void record_stats(census_ht* store, census_op_id op_id, key.ptr = gpr_strdup(key.ptr); census_ht_insert(store, key, (void*)window_stats); } - census_window_stats_add(window_stats, gpr_now(), stats); + census_window_stats_add(window_stats, gpr_now(GPR_CLOCK_REALTIME), stats); } else { census_internal_unlock_trace_store(); } @@ -185,7 +185,7 @@ static void get_stats(census_ht* store, census_aggregated_rpc_stats* data) { if (store != NULL) { size_t n; unsigned i, j; - gpr_timespec now = gpr_now(); + gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); census_ht_kv* kv = census_ht_get_all_elements(store, &n); if (kv != NULL) { data->num_entries = n; diff --git a/src/core/statistics/census_tracing.c b/src/core/statistics/census_tracing.c index 05e72b99c0..3036ba5407 100644 --- a/src/core/statistics/census_tracing.c +++ b/src/core/statistics/census_tracing.c @@ -94,7 +94,7 @@ census_op_id census_tracing_start_op(void) { g_id++; memcpy(&ret->id, &g_id, sizeof(census_op_id)); ret->rpc_stats.cnt = 1; - ret->ts = gpr_now(); + ret->ts = gpr_now(GPR_CLOCK_REALTIME); census_ht_insert(g_trace_store, op_id_as_key(&ret->id), (void*)ret); gpr_log(GPR_DEBUG, "Start tracing for id %lu", g_id); gpr_mu_unlock(&g_mu); @@ -122,7 +122,7 @@ void census_tracing_print(census_op_id op_id, const char* anno_txt) { trace = census_ht_find(g_trace_store, op_id_as_key(&op_id)); if (trace != NULL) { census_trace_annotation* anno = gpr_malloc(sizeof(census_trace_annotation)); - anno->ts = gpr_now(); + anno->ts = gpr_now(GPR_CLOCK_REALTIME); { char* d = anno->txt; const char* s = anno_txt; @@ -143,8 +143,8 @@ void census_tracing_end_op(census_op_id op_id) { gpr_mu_lock(&g_mu); trace = census_ht_find(g_trace_store, op_id_as_key(&op_id)); if (trace != NULL) { - trace->rpc_stats.elapsed_time_ms = - gpr_timespec_to_micros(gpr_time_sub(gpr_now(), trace->ts)); + trace->rpc_stats.elapsed_time_ms = gpr_timespec_to_micros( + gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME), trace->ts)); gpr_log(GPR_DEBUG, "End tracing for id %lu, method %s, latency %f us", op_id_2_uint64(&op_id), trace->method, trace->rpc_stats.elapsed_time_ms); @@ -194,8 +194,8 @@ const char* census_get_trace_method_name(const census_trace_obj* trace) { static census_trace_annotation* dup_annotation_chain( census_trace_annotation* from) { - census_trace_annotation *ret = NULL; - census_trace_annotation **to = &ret; + census_trace_annotation* ret = NULL; + census_trace_annotation** to = &ret; for (; from != NULL; from = from->next) { *to = gpr_malloc(sizeof(census_trace_annotation)); memcpy(*to, from, sizeof(census_trace_annotation)); @@ -223,9 +223,9 @@ census_trace_obj** census_get_active_ops(int* num_active_ops) { size_t n = 0; census_ht_kv* all_kvs = census_ht_get_all_elements(g_trace_store, &n); *num_active_ops = (int)n; - if (n != 0 ) { + if (n != 0) { size_t i = 0; - ret = gpr_malloc(sizeof(census_trace_obj *) * n); + ret = gpr_malloc(sizeof(census_trace_obj*) * n); for (i = 0; i < n; i++) { ret[i] = trace_obj_dup((census_trace_obj*)all_kvs[i].v); } diff --git a/src/core/statistics/window_stats.h b/src/core/statistics/window_stats.h index d733d8d247..0020f6b44c 100644 --- a/src/core/statistics/window_stats.h +++ b/src/core/statistics/window_stats.h @@ -90,11 +90,11 @@ // Record a new event, taking 15.3ms, transferring 1784 bytes. stat.latency = 0.153; stat.bytes = 1784; - census_window_stats_add(stats, gpr_now(), &stat); + census_window_stats_add(stats, gpr_now(GPR_CLOCK_REALTIME), &stat); // Get sums and print them out result[kMinInterval].statistic = &sums[kMinInterval]; result[kHourInterval].statistic = &sums[kHourInterval]; - census_window_stats_get_sums(stats, gpr_now(), result); + census_window_stats_get_sums(stats, gpr_now(GPR_CLOCK_REALTIME), result); printf("%d events/min, average time %gs, average bytes %g\n", result[kMinInterval].count, (my_stat*)result[kMinInterval].statistic->latency / @@ -170,4 +170,4 @@ void census_window_stats_get_sums(const struct census_window_stats* wstats, assertion failure). This function is thread-compatible. */ void census_window_stats_destroy(struct census_window_stats* wstats); -#endif /* GRPC_INTERNAL_CORE_STATISTICS_WINDOW_STATS_H */ +#endif /* GRPC_INTERNAL_CORE_STATISTICS_WINDOW_STATS_H */ diff --git a/src/core/support/cancellable.c b/src/core/support/cancellable.c index 5a4d488dd3..3cbb318ab6 100644 --- a/src/core/support/cancellable.c +++ b/src/core/support/cancellable.c @@ -121,8 +121,8 @@ void gpr_cancellable_cancel(gpr_cancellable *c) { } else { gpr_event ev; gpr_event_init(&ev); - gpr_event_wait(&ev, - gpr_time_add(gpr_now(), gpr_time_from_micros(1000))); + gpr_event_wait(&ev, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_micros(1000))); } } } while (failures != 0); diff --git a/src/core/support/log_linux.c b/src/core/support/log_linux.c index 7937466b79..5ac36e7b95 100644 --- a/src/core/support/log_linux.c +++ b/src/core/support/log_linux.c @@ -76,7 +76,7 @@ void gpr_default_log(gpr_log_func_args *args) { char *prefix; const char *display_file; char time_buffer[64]; - gpr_timespec now = gpr_now(); + gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); struct tm tm; final_slash = strrchr(args->file, '/'); diff --git a/src/core/support/log_posix.c b/src/core/support/log_posix.c index afca792c40..940ee20f15 100644 --- a/src/core/support/log_posix.c +++ b/src/core/support/log_posix.c @@ -75,7 +75,7 @@ void gpr_default_log(gpr_log_func_args *args) { char *final_slash; const char *display_file; char time_buffer[64]; - gpr_timespec now = gpr_now(); + gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); struct tm tm; final_slash = strrchr(args->file, '/'); diff --git a/src/core/support/log_win32.c b/src/core/support/log_win32.c index d249be7d2e..629781da8a 100644 --- a/src/core/support/log_win32.c +++ b/src/core/support/log_win32.c @@ -82,7 +82,7 @@ void gpr_log(const char *file, int line, gpr_log_severity severity, /* Simple starter implementation */ void gpr_default_log(gpr_log_func_args *args) { char time_buffer[64]; - gpr_timespec now = gpr_now(); + gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); struct tm tm; if (localtime_s(&tm, &now.tv_sec)) { diff --git a/src/core/support/sync_win32.c b/src/core/support/sync_win32.c index 72f39f8d46..29b77fc4c2 100644 --- a/src/core/support/sync_win32.c +++ b/src/core/support/sync_win32.c @@ -86,7 +86,7 @@ int gpr_cv_wait(gpr_cv *cv, gpr_mu *mu, gpr_timespec abs_deadline) { if (gpr_time_cmp(abs_deadline, gpr_inf_future) == 0) { SleepConditionVariableCS(cv, &mu->cs, INFINITE); } else { - gpr_timespec now = gpr_now(); + gpr_timespec now = gpr_now(GPR_CLOCK_REALTIME); 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_posix.c b/src/core/support/time_posix.c index afb58ef231..f9b7958783 100644 --- a/src/core/support/time_posix.c +++ b/src/core/support/time_posix.c @@ -55,22 +55,52 @@ static gpr_timespec gpr_from_timespec(struct timespec ts) { return rv; } -gpr_timespec gpr_now(void) { +/** maps gpr_clock_type --> clockid_t for clock_gettime */ +static clockid_t clockid_for_gpr_clock[] = {CLOCK_MONOTONIC, CLOCK_REALTIME}; + +void gpr_time_init(void) {} + +gpr_timespec gpr_now(gpr_clock_type clock) { struct timespec now; - clock_gettime(CLOCK_REALTIME, &now); + clock_gettime(clockid_for_gpr_clock[clock], &now); return gpr_from_timespec(now); } #else /* For some reason Apple's OSes haven't implemented clock_gettime. */ #include <sys/time.h> +#include <mach/mach.h> +#include <mach/mach_time.h> + +static double g_time_scale; +static uint64_t g_time_start; + +void gpr_time_init(void) { + mach_timebase_info_data_t tb = {0, 1}; + mach_timebase_info(&tb); + g_time_scale = tb.numer; + g_time_scale /= tb.denom; + g_time_start = mach_absolute_time(); +} -gpr_timespec gpr_now(void) { +gpr_timespec gpr_now(gpr_clock_type clock) { gpr_timespec now; struct timeval now_tv; - gettimeofday(&now_tv, NULL); - now.tv_sec = now_tv.tv_sec; - now.tv_nsec = now_tv.tv_usec * 1000; + double now_dbl; + + switch (clock) { + case GPR_CLOCK_REALTIME: + gettimeofday(&now_tv, NULL); + now.tv_sec = now_tv.tv_sec; + now.tv_nsec = now_tv.tv_usec * 1000; + break; + case GPR_CLOCK_MONOTONIC: + now_dbl = (mach_absolute_time() - g_time_start) * g_time_scale; + now.tv_sec = now_dbl * 1e-9; + now.tv_nsec = now_dbl - now.tv_sec * 1e9; + break; + } + return now; } #endif @@ -83,7 +113,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(); + now = gpr_now(GPR_CLOCK_REALTIME); 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 9db267c01b..fa77c74eeb 100644 --- a/src/core/support/time_win32.c +++ b/src/core/support/time_win32.c @@ -40,12 +40,34 @@ #include <grpc/support/time.h> #include <sys/timeb.h> -gpr_timespec gpr_now(void) { +static LARGE_INTEGER g_start_time; +static double g_time_scale; + +void gpr_time_init(void) { + LARGE_INTEGER frequency; + QueryPerformanceFrequency(&frequency); + QueryPerformanceCounter(&g_start_time); + g_time_scale = 1.0 / frequency.QuadPart; +} + +gpr_timespec gpr_now(gpr_clock_type clock) { gpr_timespec now_tv; struct _timeb now_tb; - _ftime_s(&now_tb); - now_tv.tv_sec = now_tb.time; - now_tv.tv_nsec = now_tb.millitm * 1000000; + LARGE_INTEGER timestamp; + double now_dbl; + switch (clock) { + case GPR_CLOCK_REALTIME: + _ftime_s(&now_tb); + now_tv.tv_sec = now_tb.time; + now_tv.tv_nsec = now_tb.millitm * 1000000; + break; + case GPR_CLOCK_MONOTONIC: + QueryPerformanceCounter(×tamp); + now_dbl = (timestamp.QuadPart - g_start_time.QuadPart) * g_time_scale; + now_tv.tv_sec = (time_t)now_dbl; + now_tv.tv_nsec = (int)((now_dbl - (double)now_tv.tv_sec) * 1e9); + break; + } return now_tv; } @@ -57,13 +79,14 @@ 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(); + now = gpr_now(GPR_CLOCK_REALTIME); if (gpr_time_cmp(until, now) <= 0) { return; } delta = gpr_time_sub(until, now); - sleep_millis = (DWORD)delta.tv_sec * GPR_MS_PER_SEC + delta.tv_nsec / GPR_NS_PER_MS; + sleep_millis = + (DWORD)delta.tv_sec * GPR_MS_PER_SEC + delta.tv_nsec / GPR_NS_PER_MS; Sleep(sleep_millis); } } diff --git a/src/core/surface/call.c b/src/core/surface/call.c index fb3b0b1918..0a551ac47f 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -49,6 +49,17 @@ #include <stdlib.h> #include <string.h> +/** The maximum number of completions possible. + Based upon the maximum number of individually queueable ops in the batch + api: + - initial metadata send + - message send + - status/close send (depending on client/server) + - initial metadata recv + - message recv + - status/close recv (depending on client/server) */ +#define MAX_CONCURRENT_COMPLETIONS 6 + typedef enum { REQ_INITIAL = 0, REQ_READY, REQ_DONE } req_state; typedef enum { @@ -135,6 +146,7 @@ struct grpc_call { grpc_mdctx *metadata_context; /* TODO(ctiller): share with cq if possible? */ gpr_mu mu; + gpr_mu completion_mu; /* how far through the stream have we read? */ read_state read_state; @@ -162,6 +174,8 @@ struct grpc_call { gpr_uint8 error_status_set; /** should the alarm be cancelled */ gpr_uint8 cancel_alarm; + /** bitmask of allocated completion events in completions */ + gpr_uint8 allocated_completions; /* flags with bits corresponding to write states allowing us to determine what was sent */ @@ -250,6 +264,9 @@ struct grpc_call { grpc_iomgr_closure on_done_recv; grpc_iomgr_closure on_done_send; grpc_iomgr_closure on_done_bind; + + /** completion events - for completion queue use */ + grpc_cq_completion completions[MAX_CONCURRENT_COMPLETIONS]; }; #define CALL_STACK_FROM_CALL(call) ((grpc_call_stack *)((call) + 1)) @@ -286,6 +303,7 @@ grpc_call *grpc_call_create(grpc_channel *channel, grpc_completion_queue *cq, gpr_malloc(sizeof(grpc_call) + channel_stack->call_stack_size); memset(call, 0, sizeof(grpc_call)); gpr_mu_init(&call->mu); + gpr_mu_init(&call->completion_mu); call->channel = channel; call->cq = cq; if (cq) { @@ -349,6 +367,29 @@ grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call) { return call->cq; } +static grpc_cq_completion *allocate_completion(grpc_call *call) { + gpr_uint8 i; + gpr_mu_lock(&call->completion_mu); + for (i = 0; i < GPR_ARRAY_SIZE(call->completions); i++) { + if (call->allocated_completions & (1u << i)) { + continue; + } + call->allocated_completions |= 1u << i; + gpr_mu_unlock(&call->completion_mu); + return &call->completions[i]; + } + gpr_log(GPR_ERROR, "should never reach here"); + abort(); +} + +static void done_completion(void *call, grpc_cq_completion *completion) { + grpc_call *c = call; + gpr_mu_lock(&c->completion_mu); + c->allocated_completions &= ~(1u << (completion - c->completions)); + gpr_mu_unlock(&c->completion_mu); + GRPC_CALL_INTERNAL_UNREF(c, "completion", 1); +} + #ifdef GRPC_CALL_REF_COUNT_DEBUG void grpc_call_internal_ref(grpc_call *c, const char *reason) { gpr_log(GPR_DEBUG, "CALL: ref %p %d -> %d [%s]", c, @@ -365,6 +406,7 @@ static void destroy_call(void *call, int ignored_success) { grpc_call_stack_destroy(CALL_STACK_FROM_CALL(c)); GRPC_CHANNEL_INTERNAL_UNREF(c->channel, "call"); gpr_mu_destroy(&c->mu); + gpr_mu_destroy(&c->completion_mu); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { if (c->status[i].details) { GRPC_MDSTR_UNREF(c->status[i].details); @@ -1188,7 +1230,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()); + grpc_alarm_init(&call->alarm, deadline, call_alarm, call, + gpr_now(GPR_CLOCK_REALTIME)); } /* we offset status by a small amount when storing it into transport metadata @@ -1316,11 +1359,13 @@ static void set_cancelled_value(grpc_status_code status, void *dest) { } static void finish_batch(grpc_call *call, int success, void *tag) { - grpc_cq_end_op(call->cq, tag, call, success); + grpc_cq_end_op(call->cq, tag, success, done_completion, call, + allocate_completion(call)); } static void finish_batch_with_close(grpc_call *call, int success, void *tag) { - grpc_cq_end_op(call->cq, tag, call, 1); + grpc_cq_end_op(call->cq, tag, 1, done_completion, call, + allocate_completion(call)); } static int are_write_flags_valid(gpr_uint32 flags) { @@ -1343,8 +1388,10 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag); if (nops == 0) { - grpc_cq_begin_op(call->cq, call); - grpc_cq_end_op(call->cq, tag, call, 1); + grpc_cq_begin_op(call->cq); + GRPC_CALL_INTERNAL_REF(call, "completion"); + grpc_cq_end_op(call->cq, tag, 1, done_completion, call, + allocate_completion(call)); return GRPC_CALL_OK; } @@ -1466,7 +1513,8 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, } } - grpc_cq_begin_op(call->cq, call); + GRPC_CALL_INTERNAL_REF(call, "completion"); + grpc_cq_begin_op(call->cq); return grpc_call_start_ioreq_and_call_back(call, reqs, out, finish_func, tag); } diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 030a8b4e6f..c67f75fc5c 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -45,34 +45,20 @@ #include <grpc/support/atm.h> #include <grpc/support/log.h> -#define NUM_TAG_BUCKETS 31 - -/* A single event: extends grpc_event to form a linked list with a destruction - function (on_finish) that is hidden from outside this module */ -typedef struct event { - grpc_event base; - struct event *queue_next; - struct event *queue_prev; - struct event *bucket_next; - struct event *bucket_prev; -} event; - /* Completion queue structure */ struct grpc_completion_queue { - /* When refs drops to zero, we are in shutdown mode, and will be destroyable - once all queued events are drained */ - gpr_refcount refs; - /* Once owning_refs drops to zero, we will destroy the cq */ + /** completed events */ + grpc_cq_completion completed_head; + grpc_cq_completion *completed_tail; + /** Number of pending events (+1 if we're not shutdown) */ + gpr_refcount pending_events; + /** Once owning_refs drops to zero, we will destroy the cq */ gpr_refcount owning_refs; - /* the set of low level i/o things that concern this cq */ + /** the set of low level i/o things that concern this cq */ grpc_pollset pollset; - /* 0 initially, 1 once we've begun shutting down */ + /** 0 initially, 1 once we've begun shutting down */ int shutdown; int shutdown_called; - /* Head of a linked list of queued events (prev points to the last element) */ - event *queue; - /* Fixed size chained hash table of events for pluck() */ - event *buckets[NUM_TAG_BUCKETS]; int is_server_cq; }; @@ -80,19 +66,20 @@ grpc_completion_queue *grpc_completion_queue_create(void) { grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue)); memset(cc, 0, sizeof(*cc)); /* Initial ref is dropped by grpc_completion_queue_shutdown */ - gpr_ref_init(&cc->refs, 1); + gpr_ref_init(&cc->pending_events, 1); /* One for destroy(), one for pollset_shutdown */ gpr_ref_init(&cc->owning_refs, 2); grpc_pollset_init(&cc->pollset); + cc->completed_tail = &cc->completed_head; + cc->completed_head.next = (gpr_uintptr)cc->completed_tail; return cc; } #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, const char *file, int line) { - gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p ref %d -> %d %s", - cc, (int)cc->owning_refs.count, (int)cc->owning_refs.count + 1, - reason); + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p ref %d -> %d %s", cc, + (int)cc->owning_refs.count, (int)cc->owning_refs.count + 1, reason); #else void grpc_cq_internal_ref(grpc_completion_queue *cc) { #endif @@ -107,186 +94,135 @@ static void on_pollset_destroy_done(void *arg) { #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason, const char *file, int line) { - gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p unref %d -> %d %s", - cc, (int)cc->owning_refs.count, (int)cc->owning_refs.count - 1, - reason); + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p unref %d -> %d %s", cc, + (int)cc->owning_refs.count, (int)cc->owning_refs.count - 1, reason); #else void grpc_cq_internal_unref(grpc_completion_queue *cc) { #endif if (gpr_unref(&cc->owning_refs)) { - GPR_ASSERT(cc->queue == NULL); + GPR_ASSERT(cc->completed_head.next == (gpr_uintptr)&cc->completed_head); grpc_pollset_destroy(&cc->pollset); gpr_free(cc); } } -/* Create and append an event to the queue. Returns the event so that its data - members can be filled in. - Requires GRPC_POLLSET_MU(&cc->pollset) locked. */ -static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type, - void *tag, grpc_call *call) { - event *ev = gpr_malloc(sizeof(event)); - gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS; - ev->base.type = type; - ev->base.tag = tag; - if (cc->queue == NULL) { - cc->queue = ev->queue_next = ev->queue_prev = ev; - } else { - ev->queue_next = cc->queue; - ev->queue_prev = cc->queue->queue_prev; - ev->queue_next->queue_prev = ev->queue_prev->queue_next = ev; - } - if (cc->buckets[bucket] == NULL) { - cc->buckets[bucket] = ev->bucket_next = ev->bucket_prev = ev; - } else { - ev->bucket_next = cc->buckets[bucket]; - ev->bucket_prev = cc->buckets[bucket]->bucket_prev; - ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev; - } - grpc_pollset_kick(&cc->pollset); - return ev; -} - -void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call) { - gpr_ref(&cc->refs); - if (call) GRPC_CALL_INTERNAL_REF(call, "cq"); +void grpc_cq_begin_op(grpc_completion_queue *cc) { + gpr_ref(&cc->pending_events); } /* Signal the end of an operation - if this is the last waiting-to-be-queued event, then enter shutdown mode */ -void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call, - int success) { - event *ev; - int shutdown = 0; - gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); - ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call); - ev->base.success = success; - if (gpr_unref(&cc->refs)) { +/* Queue a GRPC_OP_COMPLETED operation */ +void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, int success, + void (*done)(void *done_arg, grpc_cq_completion *storage), + void *done_arg, grpc_cq_completion *storage) { + int shutdown = gpr_unref(&cc->pending_events); + + storage->tag = tag; + storage->done = done; + storage->done_arg = done_arg; + storage->next = + ((gpr_uintptr)&cc->completed_head) | ((gpr_uintptr)(success != 0)); + + if (!shutdown) { + gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); + cc->completed_tail->next = + ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next); + cc->completed_tail = storage; + grpc_pollset_kick(&cc->pollset); + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + } else { + gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); + cc->completed_tail->next = + ((gpr_uintptr)storage) | (1u & (gpr_uintptr)cc->completed_tail->next); + cc->completed_tail = storage; GPR_ASSERT(!cc->shutdown); GPR_ASSERT(cc->shutdown_called); cc->shutdown = 1; - shutdown = 1; - } - gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - if (call) GRPC_CALL_INTERNAL_UNREF(call, "cq", 0); - if (shutdown) { + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc); } } -/* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */ -static event *create_shutdown_event(void) { - event *ev = gpr_malloc(sizeof(event)); - ev->base.type = GRPC_QUEUE_SHUTDOWN; - ev->base.tag = NULL; - return ev; -} - grpc_event grpc_completion_queue_next(grpc_completion_queue *cc, gpr_timespec deadline) { - event *ev = NULL; grpc_event ret; GRPC_CQ_INTERNAL_REF(cc, "next"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { - if (cc->queue != NULL) { - gpr_uintptr bucket; - ev = cc->queue; - bucket = ((gpr_uintptr)ev->base.tag) % NUM_TAG_BUCKETS; - cc->queue = ev->queue_next; - ev->queue_next->queue_prev = ev->queue_prev; - ev->queue_prev->queue_next = ev->queue_next; - ev->bucket_next->bucket_prev = ev->bucket_prev; - ev->bucket_prev->bucket_next = ev->bucket_next; - if (ev == cc->buckets[bucket]) { - cc->buckets[bucket] = ev->bucket_next; - if (ev == cc->buckets[bucket]) { - cc->buckets[bucket] = NULL; - } - } - if (cc->queue == ev) { - cc->queue = NULL; + if (cc->completed_tail != &cc->completed_head) { + grpc_cq_completion *c = (grpc_cq_completion *)cc->completed_head.next; + cc->completed_head.next = c->next & ~(gpr_uintptr)1; + if (c == cc->completed_tail) { + cc->completed_tail = &cc->completed_head; } + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + ret.type = GRPC_OP_COMPLETE; + ret.success = c->next & 1u; + ret.tag = c->tag; + c->done(c->done_arg, c); break; } if (cc->shutdown) { - ev = create_shutdown_event(); + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + memset(&ret, 0, sizeof(ret)); + ret.type = GRPC_QUEUE_SHUTDOWN; break; } if (!grpc_pollset_work(&cc->pollset, deadline)) { gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); memset(&ret, 0, sizeof(ret)); ret.type = GRPC_QUEUE_TIMEOUT; - GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); - GRPC_CQ_INTERNAL_UNREF(cc, "next"); - return ret; + break; } } - gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - ret = ev->base; - gpr_free(ev); GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); GRPC_CQ_INTERNAL_UNREF(cc, "next"); return ret; } -static event *pluck_event(grpc_completion_queue *cc, void *tag) { - gpr_uintptr bucket = ((gpr_uintptr)tag) % NUM_TAG_BUCKETS; - event *ev = cc->buckets[bucket]; - if (ev == NULL) return NULL; - do { - if (ev->base.tag == tag) { - ev->queue_next->queue_prev = ev->queue_prev; - ev->queue_prev->queue_next = ev->queue_next; - ev->bucket_next->bucket_prev = ev->bucket_prev; - ev->bucket_prev->bucket_next = ev->bucket_next; - if (ev == cc->buckets[bucket]) { - cc->buckets[bucket] = ev->bucket_next; - if (ev == cc->buckets[bucket]) { - cc->buckets[bucket] = NULL; - } - } - if (cc->queue == ev) { - cc->queue = ev->queue_next; - if (cc->queue == ev) { - cc->queue = NULL; - } - } - return ev; - } - ev = ev->bucket_next; - } while (ev != cc->buckets[bucket]); - return NULL; -} - grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag, gpr_timespec deadline) { - event *ev = NULL; grpc_event ret; + grpc_cq_completion *c; + grpc_cq_completion *prev; GRPC_CQ_INTERNAL_REF(cc, "pluck"); gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); for (;;) { - if ((ev = pluck_event(cc, tag))) { - break; + prev = &cc->completed_head; + while ((c = (grpc_cq_completion *)(prev->next & ~(gpr_uintptr)1)) != + &cc->completed_head) { + if (c->tag == tag) { + prev->next = + (prev->next & (gpr_uintptr)1) | (c->next & ~(gpr_uintptr)1); + if (c == cc->completed_tail) { + cc->completed_tail = prev; + } + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + ret.type = GRPC_OP_COMPLETE; + ret.success = c->next & 1u; + ret.tag = c->tag; + c->done(c->done_arg, c); + goto done; + } + prev = c; } if (cc->shutdown) { - ev = create_shutdown_event(); + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); + memset(&ret, 0, sizeof(ret)); + ret.type = GRPC_QUEUE_SHUTDOWN; break; } if (!grpc_pollset_work(&cc->pollset, deadline)) { gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); memset(&ret, 0, sizeof(ret)); ret.type = GRPC_QUEUE_TIMEOUT; - GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); - GRPC_CQ_INTERNAL_UNREF(cc, "pluck"); - return ret; + break; } } - gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - ret = ev->base; - gpr_free(ev); +done: GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret); GRPC_CQ_INTERNAL_UNREF(cc, "pluck"); return ret; @@ -303,7 +239,7 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) { cc->shutdown_called = 1; gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); - if (gpr_unref(&cc->refs)) { + if (gpr_unref(&cc->pending_events)) { gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); GPR_ASSERT(!cc->shutdown); cc->shutdown = 1; @@ -324,8 +260,8 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) { void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) { gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); grpc_pollset_kick(&cc->pollset); - grpc_pollset_work(&cc->pollset, - gpr_time_add(gpr_now(), gpr_time_from_millis(100))); + grpc_pollset_work(&cc->pollset, gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), + gpr_time_from_millis(100))); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); } diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h index 1b9010f462..f944f48d8e 100644 --- a/src/core/surface/completion_queue.h +++ b/src/core/surface/completion_queue.h @@ -39,6 +39,17 @@ #include "src/core/iomgr/pollset.h" #include <grpc/grpc.h> +typedef struct grpc_cq_completion { + /** user supplied tag */ + void *tag; + /** done callback - called when this queue element is no longer + needed by the completion queue */ + void (*done)(void *done_arg, struct grpc_cq_completion *c); + void *done_arg; + /** next pointer; low bit is used to indicate success or not */ + gpr_uintptr next; +} grpc_cq_completion; + #ifdef GRPC_CQ_REF_COUNT_DEBUG void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason, const char *file, int line); @@ -57,11 +68,12 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc); /* Flag that an operation is beginning: the completion channel will not finish shutdown until a corrensponding grpc_cq_end_* call is made */ -void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call); +void grpc_cq_begin_op(grpc_completion_queue *cc); /* Queue a GRPC_OP_COMPLETED operation */ -void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call, - int success); +void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, int success, + void (*done)(void *done_arg, grpc_cq_completion *storage), + void *done_arg, grpc_cq_completion *storage); grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); diff --git a/src/core/surface/init.c b/src/core/surface/init.c index 3847ded28c..04e27d30ac 100644 --- a/src/core/surface/init.c +++ b/src/core/surface/init.c @@ -35,6 +35,7 @@ #include <grpc/census.h> #include <grpc/grpc.h> +#include <grpc/support/time.h> #include "src/core/channel/channel_stack.h" #include "src/core/client_config/resolver_registry.h" #include "src/core/client_config/resolvers/dns_resolver.h" @@ -64,6 +65,7 @@ void grpc_init(void) { gpr_mu_lock(&g_init_mu); if (++g_initializations == 1) { + gpr_time_init(); grpc_resolver_registry_init("dns:///"); grpc_register_resolver_type("dns", grpc_dns_resolver_factory_create()); #ifdef GPR_POSIX_SOCKET diff --git a/src/core/surface/server.c b/src/core/surface/server.c index a9d8940631..4dc51bf031 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -72,12 +72,14 @@ typedef struct { typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type; -typedef struct { +typedef struct requested_call { requested_call_type type; + struct requested_call *next; void *tag; grpc_completion_queue *cq_bound_to_call; grpc_completion_queue *cq_for_notification; grpc_call **call; + grpc_cq_completion completion; union { struct { grpc_call_details *details; @@ -92,17 +94,11 @@ typedef struct { } data; } requested_call; -typedef struct { - requested_call *calls; - size_t count; - size_t capacity; -} requested_call_array; - struct registered_method { char *method; char *host; call_data *pending; - requested_call_array requested; + requested_call *requests; registered_method *next; }; @@ -131,6 +127,7 @@ struct channel_data { typedef struct shutdown_tag { void *tag; grpc_completion_queue *cq; + grpc_cq_completion completion; } shutdown_tag; struct grpc_server { @@ -153,7 +150,7 @@ struct grpc_server { gpr_mu mu_call; /* mutex for call-specific state */ registered_method *registered_methods; - requested_call_array requested_calls; + requested_call *requests; gpr_uint8 shutdown; gpr_uint8 shutdown_published; @@ -166,6 +163,9 @@ struct grpc_server { listener *listeners; int listeners_destroyed; gpr_refcount internal_refcount; + + /** when did we print the last shutdown progress message */ + gpr_timespec last_shutdown_message_time; }; typedef enum { @@ -270,7 +270,8 @@ static void send_shutdown(grpc_channel *channel, int send_goaway, } static void channel_broadcaster_shutdown(channel_broadcaster *cb, - int send_goaway, int force_disconnect) { + int send_goaway, + int force_disconnect) { size_t i; for (i = 0; i < cb->num_channels; i++) { @@ -329,22 +330,6 @@ static int call_list_remove(call_data *call, call_list list) { return 1; } -static void requested_call_array_destroy(requested_call_array *array) { - gpr_free(array->calls); -} - -static requested_call *requested_call_array_add(requested_call_array *array) { - requested_call *rc; - if (array->count == array->capacity) { - array->capacity = GPR_MAX(array->capacity + 8, array->capacity * 2); - array->calls = - gpr_realloc(array->calls, sizeof(requested_call) * array->capacity); - } - rc = &array->calls[array->count++]; - memset(rc, 0, sizeof(*rc)); - return rc; -} - static void server_ref(grpc_server *server) { gpr_ref(&server->internal_refcount); } @@ -356,12 +341,10 @@ static void server_delete(grpc_server *server) { gpr_mu_destroy(&server->mu_global); gpr_mu_destroy(&server->mu_call); gpr_free(server->channel_filters); - requested_call_array_destroy(&server->requested_calls); while ((rm = server->registered_methods) != NULL) { server->registered_methods = rm->next; gpr_free(rm->method); gpr_free(rm->host); - requested_call_array_destroy(&rm->requested); gpr_free(rm); } for (i = 0; i < server->cq_count; i++) { @@ -409,23 +392,24 @@ static void destroy_channel(channel_data *chand) { static void finish_start_new_rpc(grpc_server *server, grpc_call_element *elem, call_data **pending_root, - requested_call_array *array) { - requested_call rc; + requested_call **requests) { + requested_call *rc; call_data *calld = elem->call_data; gpr_mu_lock(&server->mu_call); - if (array->count == 0) { + rc = *requests; + if (rc == NULL) { gpr_mu_lock(&calld->mu_state); calld->state = PENDING; gpr_mu_unlock(&calld->mu_state); call_list_join(pending_root, calld, PENDING_START); gpr_mu_unlock(&server->mu_call); } else { - rc = array->calls[--array->count]; + *requests = rc->next; gpr_mu_lock(&calld->mu_state); calld->state = ACTIVATED; gpr_mu_unlock(&calld->mu_state); gpr_mu_unlock(&server->mu_call); - begin_call(server, calld, &rc); + begin_call(server, calld, rc); } } @@ -441,14 +425,14 @@ static void start_new_rpc(grpc_call_element *elem) { /* TODO(ctiller): unify these two searches */ /* check for an exact match with host */ hash = GRPC_MDSTR_KV_HASH(calld->host->hash, calld->path->hash); - for (i = 0; i < chand->registered_method_max_probes; i++) { + for (i = 0; i <= chand->registered_method_max_probes; i++) { rm = &chand->registered_methods[(hash + i) % chand->registered_method_slots]; if (!rm) break; if (rm->host != calld->host) continue; if (rm->method != calld->path) continue; finish_start_new_rpc(server, elem, &rm->server_registered_method->pending, - &rm->server_registered_method->requested); + &rm->server_registered_method->requests); return; } /* check for a wildcard method definition (no host set) */ @@ -460,12 +444,12 @@ static void start_new_rpc(grpc_call_element *elem) { if (rm->host != NULL) continue; if (rm->method != calld->path) continue; finish_start_new_rpc(server, elem, &rm->server_registered_method->pending, - &rm->server_registered_method->requested); + &rm->server_registered_method->requests); return; } } finish_start_new_rpc(server, elem, &server->lists[PENDING_START], - &server->requested_calls); + &server->requests); } static void kill_zombie(void *elem, int success) { @@ -481,26 +465,47 @@ static int num_listeners(grpc_server *server) { return n; } +static void done_shutdown_event(void *server, grpc_cq_completion *completion) { + server_unref(server); +} + +static int num_channels(grpc_server *server) { + channel_data *chand; + int n = 0; + for (chand = server->root_channel_data.next; + chand != &server->root_channel_data; chand = chand->next) { + n++; + } + return n; +} + static void maybe_finish_shutdown(grpc_server *server) { size_t i; if (!server->shutdown || server->shutdown_published) { return; } - if (server->root_channel_data.next != &server->root_channel_data) { - gpr_log(GPR_DEBUG, - "Waiting for all channels to close before destroying server"); - return; - } - if (server->listeners_destroyed < num_listeners(server)) { - gpr_log(GPR_DEBUG, "Waiting for all listeners to be destroyed (@ %d/%d)", - server->listeners_destroyed, num_listeners(server)); + if (server->root_channel_data.next != &server->root_channel_data || + server->listeners_destroyed < num_listeners(server)) { + if (gpr_time_cmp( + gpr_time_sub(gpr_now(GPR_CLOCK_REALTIME), server->last_shutdown_message_time), + gpr_time_from_seconds(1)) >= 0) { + server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME); + gpr_log(GPR_DEBUG, + "Waiting for %d channels and %d/%d listeners to be destroyed" + " before shutting down server", + num_channels(server), + num_listeners(server) - server->listeners_destroyed, + num_listeners(server)); + } return; } server->shutdown_published = 1; for (i = 0; i < server->num_shutdown_tags; i++) { - grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, - NULL, 1); + server_ref(server); + grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag, 1, + done_shutdown_event, server, + &server->shutdown_tags[i].completion); } } @@ -924,15 +929,14 @@ void grpc_server_setup_transport(grpc_server *s, grpc_transport *transport, void grpc_server_shutdown_and_notify(grpc_server *server, grpc_completion_queue *cq, void *tag) { listener *l; - requested_call_array requested_calls; - size_t i; + requested_call *requests = NULL; registered_method *rm; shutdown_tag *sdt; channel_broadcaster broadcaster; /* lock, and gather up some stuff to do */ gpr_mu_lock(&server->mu_global); - grpc_cq_begin_op(cq, NULL); + grpc_cq_begin_op(cq); server->shutdown_tags = gpr_realloc(server->shutdown_tags, sizeof(shutdown_tag) * (server->num_shutdown_tags + 1)); @@ -944,27 +948,21 @@ void grpc_server_shutdown_and_notify(grpc_server *server, return; } + server->last_shutdown_message_time = gpr_now(GPR_CLOCK_REALTIME); + channel_broadcaster_init(server, &broadcaster); /* collect all unregistered then registered calls */ gpr_mu_lock(&server->mu_call); - requested_calls = server->requested_calls; - memset(&server->requested_calls, 0, sizeof(server->requested_calls)); + requests = server->requests; + server->requests = NULL; for (rm = server->registered_methods; rm; rm = rm->next) { - if (requested_calls.count + rm->requested.count > - requested_calls.capacity) { - requested_calls.capacity = - GPR_MAX(requested_calls.count + rm->requested.count, - 2 * requested_calls.capacity); - requested_calls.calls = - gpr_realloc(requested_calls.calls, sizeof(*requested_calls.calls) * - requested_calls.capacity); + while (rm->requests != NULL) { + requested_call *c = rm->requests; + rm->requests = c->next; + c->next = requests; + requests = c; } - memcpy(requested_calls.calls + requested_calls.count, rm->requested.calls, - sizeof(*requested_calls.calls) * rm->requested.count); - requested_calls.count += rm->requested.count; - gpr_free(rm->requested.calls); - memset(&rm->requested, 0, sizeof(rm->requested)); } gpr_mu_unlock(&server->mu_call); @@ -973,10 +971,11 @@ void grpc_server_shutdown_and_notify(grpc_server *server, gpr_mu_unlock(&server->mu_global); /* terminate all the requested calls */ - for (i = 0; i < requested_calls.count; i++) { - fail_call(server, &requested_calls.calls[i]); + while (requests != NULL) { + requested_call *next = requests->next; + fail_call(server, requests); + requests = next; } - gpr_free(requested_calls.calls); /* Shutdown listeners */ for (l = server->listeners; l; l = l->next) { @@ -1038,7 +1037,7 @@ void grpc_server_add_listener(grpc_server *server, void *arg, static grpc_call_error queue_call_request(grpc_server *server, requested_call *rc) { call_data *calld = NULL; - requested_call_array *requested_calls = NULL; + requested_call **requests = NULL; gpr_mu_lock(&server->mu_call); if (server->shutdown) { gpr_mu_unlock(&server->mu_call); @@ -1049,12 +1048,12 @@ static grpc_call_error queue_call_request(grpc_server *server, case BATCH_CALL: calld = call_list_remove_head(&server->lists[PENDING_START], PENDING_START); - requested_calls = &server->requested_calls; + requests = &server->requests; break; case REGISTERED_CALL: calld = call_list_remove_head( &rc->data.registered.registered_method->pending, PENDING_START); - requested_calls = &rc->data.registered.registered_method->requested; + requests = &rc->data.registered.registered_method->requests; break; } if (calld != NULL) { @@ -1066,7 +1065,8 @@ static grpc_call_error queue_call_request(grpc_server *server, begin_call(server, calld, rc); return GRPC_CALL_OK; } else { - *requested_call_array_add(requested_calls) = *rc; + rc->next = *requests; + *requests = rc; gpr_mu_unlock(&server->mu_call); return GRPC_CALL_OK; } @@ -1077,22 +1077,23 @@ grpc_call_error grpc_server_request_call( grpc_metadata_array *initial_metadata, grpc_completion_queue *cq_bound_to_call, grpc_completion_queue *cq_for_notification, void *tag) { - requested_call rc; + requested_call *rc = gpr_malloc(sizeof(*rc)); GRPC_SERVER_LOG_REQUEST_CALL(GPR_INFO, server, call, details, initial_metadata, cq_bound_to_call, cq_for_notification, tag); if (!grpc_cq_is_server_cq(cq_for_notification)) { + gpr_free(rc); return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; } - grpc_cq_begin_op(cq_for_notification, NULL); - rc.type = BATCH_CALL; - rc.tag = tag; - rc.cq_bound_to_call = cq_bound_to_call; - rc.cq_for_notification = cq_for_notification; - rc.call = call; - rc.data.batch.details = details; - rc.data.batch.initial_metadata = initial_metadata; - return queue_call_request(server, &rc); + grpc_cq_begin_op(cq_for_notification); + rc->type = BATCH_CALL; + rc->tag = tag; + rc->cq_bound_to_call = cq_bound_to_call; + rc->cq_for_notification = cq_for_notification; + rc->call = call; + rc->data.batch.details = details; + rc->data.batch.initial_metadata = initial_metadata; + return queue_call_request(server, rc); } grpc_call_error grpc_server_request_registered_call( @@ -1100,22 +1101,23 @@ grpc_call_error grpc_server_request_registered_call( grpc_metadata_array *initial_metadata, grpc_byte_buffer **optional_payload, grpc_completion_queue *cq_bound_to_call, grpc_completion_queue *cq_for_notification, void *tag) { - requested_call rc; + requested_call *rc = gpr_malloc(sizeof(*rc)); registered_method *registered_method = rm; if (!grpc_cq_is_server_cq(cq_for_notification)) { + gpr_free(rc); return GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE; } - grpc_cq_begin_op(cq_for_notification, NULL); - rc.type = REGISTERED_CALL; - rc.tag = tag; - rc.cq_bound_to_call = cq_bound_to_call; - rc.cq_for_notification = cq_for_notification; - rc.call = call; - rc.data.registered.registered_method = registered_method; - rc.data.registered.deadline = deadline; - rc.data.registered.initial_metadata = initial_metadata; - rc.data.registered.optional_payload = optional_payload; - return queue_call_request(server, &rc); + grpc_cq_begin_op(cq_for_notification); + rc->type = REGISTERED_CALL; + rc->tag = tag; + rc->cq_bound_to_call = cq_bound_to_call; + rc->cq_for_notification = cq_for_notification; + rc->call = call; + rc->data.registered.registered_method = registered_method; + rc->data.registered.deadline = deadline; + rc->data.registered.initial_metadata = initial_metadata; + rc->data.registered.optional_payload = optional_payload; + return queue_call_request(server, rc); } static void publish_registered_or_batch(grpc_call *call, int success, @@ -1182,8 +1184,11 @@ static void begin_call(grpc_server *server, call_data *calld, } GRPC_CALL_INTERNAL_REF(calld->call, "server"); - grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish, - rc->tag); + grpc_call_start_ioreq_and_call_back(calld->call, req, r - req, publish, rc); +} + +static void done_request_event(void *req, grpc_cq_completion *c) { + gpr_free(req); } static void fail_call(grpc_server *server, requested_call *rc) { @@ -1196,15 +1201,19 @@ static void fail_call(grpc_server *server, requested_call *rc) { rc->data.registered.initial_metadata->count = 0; break; } - grpc_cq_end_op(rc->cq_for_notification, rc->tag, NULL, 0); + grpc_cq_end_op(rc->cq_for_notification, rc->tag, 0, done_request_event, rc, + &rc->completion); } static void publish_registered_or_batch(grpc_call *call, int success, - void *tag) { + void *prc) { grpc_call_element *elem = grpc_call_stack_element(grpc_call_get_call_stack(call), 0); + requested_call *rc = prc; call_data *calld = elem->call_data; - grpc_cq_end_op(calld->cq_new, tag, call, success); + grpc_cq_end_op(calld->cq_new, rc->tag, success, done_request_event, rc, + &rc->completion); + GRPC_CALL_INTERNAL_UNREF(call, "server", 0); } const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server) { diff --git a/src/core/surface/version.c b/src/core/surface/version.c new file mode 100644 index 0000000000..4f5d648371 --- /dev/null +++ b/src/core/surface/version.c @@ -0,0 +1,41 @@ +/* + * + * 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. + * + */ + +/* This file is autogenerated from: + templates/src/core/surface/version.c.template */ + +#include <grpc/grpc.h> + +const char *grpc_version_string(void) { + return "0.10.0.0"; +} diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index 130167f830..9597395aab 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -205,7 +205,8 @@ void grpc_chttp2_publish_reads( } if (stream_parsing->saw_rst_stream) { stream_global->cancelled = 1; - stream_global->cancelled_status = grpc_chttp2_http2_error_to_grpc_status(stream_parsing->rst_stream_reason); + stream_global->cancelled_status = grpc_chttp2_http2_error_to_grpc_status( + stream_parsing->rst_stream_reason); if (stream_parsing->rst_stream_reason == GRPC_CHTTP2_NO_ERROR) { stream_global->published_cancelled = 1; } @@ -599,7 +600,7 @@ static void on_header(void *tp, grpc_mdelem *md) { } grpc_chttp2_incoming_metadata_buffer_set_deadline( &stream_parsing->incoming_metadata, - gpr_time_add(gpr_now(), *cached_timeout)); + gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), *cached_timeout)); GRPC_MDELEM_UNREF(md); } else { grpc_chttp2_incoming_metadata_buffer_add(&stream_parsing->incoming_metadata, diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index 56ab82006a..d553d80085 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -437,7 +437,8 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline, framer_state *st) { char timeout_str[GRPC_CHTTP2_TIMEOUT_ENCODE_MIN_BUFSIZE]; grpc_mdelem *mdelem; - grpc_chttp2_encode_timeout(gpr_time_sub(deadline, gpr_now()), timeout_str); + grpc_chttp2_encode_timeout( + gpr_time_sub(deadline, gpr_now(GPR_CLOCK_REALTIME)), 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)); diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 322ff39820..ac399e4a1d 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -876,11 +876,19 @@ static void update_global_window(void *args, gpr_uint32 id, void *stream) { grpc_chttp2_stream *s = stream; grpc_chttp2_transport_global *transport_global = &t->global; grpc_chttp2_stream_global *stream_global = &s->global; + int was_zero; + int is_zero; GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("settings", transport_global, stream_global, outgoing_window, t->parsing.initial_window_update); + was_zero = stream_global->outgoing_window <= 0; stream_global->outgoing_window += t->parsing.initial_window_update; + is_zero = stream_global->outgoing_window <= 0; + + if (was_zero && !is_zero) { + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + } } static void read_error_locked(grpc_chttp2_transport *t) { diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c index 9cbb0952d0..e95b7a21f9 100644 --- a/src/core/transport/metadata.c +++ b/src/core/transport/metadata.c @@ -87,6 +87,7 @@ typedef struct internal_metadata { gpr_atm refcnt; /* private only data */ + gpr_mu mu_user_data; void *user_data; void (*destroy_user_data)(void *user_data); @@ -183,7 +184,7 @@ grpc_mdctx *grpc_mdctx_create(void) { /* This seed is used to prevent remote connections from controlling hash table * collisions. It needs to be somewhat unpredictable to a remote connection. */ - return grpc_mdctx_create_with_seed(gpr_now().tv_nsec); + return grpc_mdctx_create_with_seed(gpr_now(GPR_CLOCK_REALTIME).tv_nsec); } static void discard_metadata(grpc_mdctx *ctx) { @@ -200,6 +201,7 @@ static void discard_metadata(grpc_mdctx *ctx) { if (cur->user_data) { cur->destroy_user_data(cur->user_data); } + gpr_mu_destroy(&cur->mu_user_data); gpr_free(cur); cur = next; ctx->mdtab_free--; @@ -467,6 +469,7 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdctx *ctx, md->user_data = NULL; md->destroy_user_data = NULL; md->bucket_next = ctx->mdtab[hash % ctx->mdtab_capacity]; + gpr_mu_init(&md->mu_user_data); #ifdef GRPC_METADATA_REFCOUNT_DEBUG gpr_log(GPR_DEBUG, "ELM NEW:%p:%d: '%s' = '%s'", md, gpr_atm_no_barrier_load(&md->refcnt), @@ -581,18 +584,29 @@ size_t grpc_mdctx_get_mdtab_free_test_only(grpc_mdctx *ctx) { void *grpc_mdelem_get_user_data(grpc_mdelem *md, void (*if_destroy_func)(void *)) { internal_metadata *im = (internal_metadata *)md; - return im->destroy_user_data == if_destroy_func ? im->user_data : NULL; + void *result; + gpr_mu_lock(&im->mu_user_data); + result = im->destroy_user_data == if_destroy_func ? im->user_data : NULL; + gpr_mu_unlock(&im->mu_user_data); + return result; } void grpc_mdelem_set_user_data(grpc_mdelem *md, void (*destroy_func)(void *), void *user_data) { internal_metadata *im = (internal_metadata *)md; GPR_ASSERT((user_data == NULL) == (destroy_func == NULL)); + gpr_mu_lock(&im->mu_user_data); if (im->destroy_user_data) { - im->destroy_user_data(im->user_data); + /* user data can only be set once */ + gpr_mu_unlock(&im->mu_user_data); + if (destroy_func != NULL) { + destroy_func(user_data); + } + return; } im->destroy_user_data = destroy_func; im->user_data = user_data; + gpr_mu_unlock(&im->mu_user_data); } gpr_slice grpc_mdstr_as_base64_encoded_and_huffman_compressed(grpc_mdstr *gs) { diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc index 72cdd49d19..c68f6dd9f8 100644 --- a/src/cpp/client/client_context.cc +++ b/src/cpp/client/client_context.cc @@ -36,6 +36,7 @@ #include <grpc/grpc.h> #include <grpc++/credentials.h> #include <grpc++/time.h> +#include "src/cpp/common/create_auth_context.h" namespace grpc { @@ -75,6 +76,13 @@ void ClientContext::set_call(grpc_call* call, } } +std::shared_ptr<const AuthContext> ClientContext::auth_context() const { + if (auth_context_.get() == nullptr) { + auth_context_ = CreateAuthContext(call_); + } + return auth_context_; +} + void ClientContext::TryCancel() { if (call_) { grpc_call_cancel(call_); diff --git a/src/cpp/common/create_auth_context.h b/src/cpp/common/create_auth_context.h new file mode 100644 index 0000000000..9082a90c6d --- /dev/null +++ b/src/cpp/common/create_auth_context.h @@ -0,0 +1,42 @@ +/* + * + * 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. + * + */ +#include <memory> + +#include <grpc/grpc.h> +#include <grpc++/auth_context.h> + +namespace grpc { + +std::shared_ptr<const AuthContext> CreateAuthContext(grpc_call* call); + +} // namespace grpc diff --git a/src/cpp/common/insecure_create_auth_context.cc b/src/cpp/common/insecure_create_auth_context.cc new file mode 100644 index 0000000000..07fc0bd549 --- /dev/null +++ b/src/cpp/common/insecure_create_auth_context.cc @@ -0,0 +1,45 @@ +/* + * + * 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. + * + */ +#include <memory> + +#include <grpc/grpc.h> +#include <grpc++/auth_context.h> + +namespace grpc { + +std::shared_ptr<const AuthContext> CreateAuthContext(grpc_call* call) { + (void)call; + return std::shared_ptr<const AuthContext>(); +} + +} // namespace grpc diff --git a/src/cpp/common/secure_auth_context.cc b/src/cpp/common/secure_auth_context.cc new file mode 100644 index 0000000000..4513723653 --- /dev/null +++ b/src/cpp/common/secure_auth_context.cc @@ -0,0 +1,80 @@ +/* + * + * 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. + * + */ + +#include "src/cpp/common/secure_auth_context.h" + +#include <grpc/grpc_security.h> + +namespace grpc { + +SecureAuthContext::SecureAuthContext(grpc_auth_context* ctx) : ctx_(ctx) {} + +SecureAuthContext::~SecureAuthContext() { grpc_auth_context_release(ctx_); } + +std::vector<grpc::string> SecureAuthContext::GetPeerIdentity() const { + if (!ctx_) { + return std::vector<grpc::string>(); + } + grpc_auth_property_iterator iter = grpc_auth_context_peer_identity(ctx_); + std::vector<grpc::string> identity; + const grpc_auth_property* property = nullptr; + while ((property = grpc_auth_property_iterator_next(&iter))) { + identity.push_back(grpc::string(property->value, property->value_length)); + } + return identity; +} + +grpc::string SecureAuthContext::GetPeerIdentityPropertyName() const { + if (!ctx_) { + return ""; + } + const char* name = grpc_auth_context_peer_identity_property_name(ctx_); + return name == nullptr ? "" : name; +} + +std::vector<grpc::string> SecureAuthContext::FindPropertyValues( + const grpc::string& name) const { + if (!ctx_) { + return std::vector<grpc::string>(); + } + grpc_auth_property_iterator iter = + grpc_auth_context_find_properties_by_name(ctx_, name.c_str()); + const grpc_auth_property* property = nullptr; + std::vector<grpc::string> values; + while ((property = grpc_auth_property_iterator_next(&iter))) { + values.push_back(grpc::string(property->value, property->value_length)); + } + return values; +} + +} // namespace grpc diff --git a/src/cpp/common/secure_auth_context.h b/src/cpp/common/secure_auth_context.h new file mode 100644 index 0000000000..bba46803cd --- /dev/null +++ b/src/cpp/common/secure_auth_context.h @@ -0,0 +1,62 @@ +/* + * + * 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. + * + */ + +#ifndef GRPC_INTERNAL_CPP_COMMON_SECURE_AUTH_CONTEXT_H +#define GRPC_INTERNAL_CPP_COMMON_SECURE_AUTH_CONTEXT_H + +#include <grpc++/auth_context.h> + +struct grpc_auth_context; + +namespace grpc { + +class SecureAuthContext GRPC_FINAL : public AuthContext { + public: + SecureAuthContext(grpc_auth_context* ctx); + + ~SecureAuthContext() GRPC_OVERRIDE; + + std::vector<grpc::string> GetPeerIdentity() const GRPC_OVERRIDE; + + grpc::string GetPeerIdentityPropertyName() const GRPC_OVERRIDE; + + std::vector<grpc::string> FindPropertyValues(const grpc::string& name) const + GRPC_OVERRIDE; + + private: + grpc_auth_context* ctx_; +}; + +} // namespace grpc + +#endif // GRPC_INTERNAL_CPP_COMMON_SECURE_AUTH_CONTEXT_H diff --git a/src/cpp/common/secure_create_auth_context.cc b/src/cpp/common/secure_create_auth_context.cc new file mode 100644 index 0000000000..d81f4bbc4a --- /dev/null +++ b/src/cpp/common/secure_create_auth_context.cc @@ -0,0 +1,50 @@ +/* + * + * 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. + * + */ +#include <memory> + +#include <grpc/grpc.h> +#include <grpc/grpc_security.h> +#include <grpc++/auth_context.h> +#include "src/cpp/common/secure_auth_context.h" + +namespace grpc { + +std::shared_ptr<const AuthContext> CreateAuthContext(grpc_call* call) { + if (call == nullptr) { + return std::shared_ptr<const AuthContext>(); + } + return std::shared_ptr<const AuthContext>( + new SecureAuthContext(grpc_call_auth_context(call))); +} + +} // namespace grpc diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc index f9d20ff579..e6761d6244 100644 --- a/src/cpp/server/server.cc +++ b/src/cpp/server/server.cc @@ -118,7 +118,7 @@ class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag { has_request_payload_(mrd->has_request_payload_), request_payload_(mrd->request_payload_), method_(mrd->method_) { - ctx_.call_ = mrd->call_; + ctx_.set_call(mrd->call_); ctx_.cq_ = &cq_; GPR_ASSERT(mrd->in_flight_); mrd->in_flight_ = false; @@ -326,7 +326,7 @@ bool Server::BaseAsyncRequest::FinalizeResult(void** tag, bool* status) { } } grpc_metadata_array_destroy(&initial_metadata_array_); - context_->call_ = call_; + context_->set_call(call_); context_->cq_ = call_cq_; Call call(call_, server_, call_cq_, server_->max_message_size_); if (*status && call_) { diff --git a/src/cpp/server/server_context.cc b/src/cpp/server/server_context.cc index 699895a3cf..1bb3a8bcc4 100644 --- a/src/cpp/server/server_context.cc +++ b/src/cpp/server/server_context.cc @@ -39,6 +39,8 @@ #include <grpc++/impl/sync.h> #include <grpc++/time.h> +#include "src/cpp/common/create_auth_context.h" + namespace grpc { // CompletionOp @@ -146,4 +148,9 @@ bool ServerContext::IsCancelled() { return completion_op_ && completion_op_->CheckCancelled(cq_); } +void ServerContext::set_call(grpc_call* call) { + call_ = call; + auth_context_ = CreateAuthContext(call); +} + } // namespace grpc diff --git a/src/csharp/Grpc.Auth/Grpc.Auth.nuspec b/src/csharp/Grpc.Auth/Grpc.Auth.nuspec index e7a538b21b..978b04d70b 100644 --- a/src/csharp/Grpc.Auth/Grpc.Auth.nuspec +++ b/src/csharp/Grpc.Auth/Grpc.Auth.nuspec @@ -5,19 +5,19 @@ <title>gRPC C# Auth</title> <summary>Auth library for C# implementation of gRPC - an RPC library and framework</summary> <description>Auth library for C# implementation of gRPC - an RPC library and framework. See project site for more info.</description> - <version>0.5.1</version> + <version>0.6.0</version> <authors>Google Inc.</authors> <owners>grpc-packages</owners> <licenseUrl>https://github.com/grpc/grpc/blob/master/LICENSE</licenseUrl> <projectUrl>https://github.com/grpc/grpc</projectUrl> <requireLicenseAcceptance>false</requireLicenseAcceptance> - <releaseNotes>Release 0.5.1 of gRPC C#</releaseNotes> + <releaseNotes>Release 0.6.0 of gRPC C#</releaseNotes> <copyright>Copyright 2015, Google Inc.</copyright> <tags>gRPC RPC Protocol HTTP/2 Auth OAuth2</tags> <dependencies> <dependency id="BouncyCastle" version="1.7.0" /> <dependency id="Google.Apis.Auth" version="1.9.1" /> - <dependency id="Grpc.Core" version="0.5.1" /> + <dependency id="Grpc.Core" version="0.6.0" /> </dependencies> </metadata> <files> diff --git a/src/csharp/Grpc.Auth/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Auth/Properties/AssemblyInfo.cs index c442ccc977..2cdf643597 100644 --- a/src/csharp/Grpc.Auth/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Auth/Properties/AssemblyInfo.cs @@ -9,6 +9,6 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] [assembly: InternalsVisibleTo("Grpc.Auth.Tests")] diff --git a/src/csharp/Grpc.Core.Tests/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Core.Tests/Properties/AssemblyInfo.cs index 7f6133a992..d5fffb8b18 100644 --- a/src/csharp/Grpc.Core.Tests/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Core.Tests/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.Core/Grpc.Core.csproj b/src/csharp/Grpc.Core/Grpc.Core.csproj index a36a6a5acc..0e67da3245 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.csproj +++ b/src/csharp/Grpc.Core/Grpc.Core.csproj @@ -33,8 +33,9 @@ </PropertyGroup> <ItemGroup> <Reference Include="System" /> - <Reference Include="System.Collections.Immutable"> - <HintPath>..\packages\Microsoft.Bcl.Immutable.1.0.34\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath> + <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> @@ -130,4 +131,4 @@ </Target> <Import Project="..\packages\grpc.dependencies.openssl.redist.1.0.2.2\build\portable-net45\grpc.dependencies.openssl.redist.targets" Condition="Exists('..\packages\grpc.dependencies.openssl.redist.1.0.2.2\build\portable-net45\grpc.dependencies.openssl.redist.targets')" /> <Import Project="..\packages\grpc.dependencies.zlib.redist.1.2.8.9\build\portable-net45\grpc.dependencies.zlib.redist.targets" Condition="Exists('..\packages\grpc.dependencies.zlib.redist.1.2.8.9\build\portable-net45\grpc.dependencies.zlib.redist.targets')" /> -</Project> +</Project>
\ No newline at end of file diff --git a/src/csharp/Grpc.Core/Grpc.Core.nuspec b/src/csharp/Grpc.Core/Grpc.Core.nuspec index 629b978fdf..457983532a 100644 --- a/src/csharp/Grpc.Core/Grpc.Core.nuspec +++ b/src/csharp/Grpc.Core/Grpc.Core.nuspec @@ -5,19 +5,19 @@ <title>gRPC C# Core</title> <summary>Core C# implementation of gRPC - an RPC library and framework</summary> <description>Core C# implementation of gRPC - an RPC library and framework. See project site for more info.</description> - <version>0.5.1</version> + <version>0.6.0</version> <authors>Google Inc.</authors> <owners>grpc-packages</owners> <licenseUrl>https://github.com/grpc/grpc/blob/master/LICENSE</licenseUrl> <projectUrl>https://github.com/grpc/grpc</projectUrl> <requireLicenseAcceptance>false</requireLicenseAcceptance> - <releaseNotes>Release 0.5.1 of gRPC C#</releaseNotes> + <releaseNotes>Release 0.6.0 of gRPC C#</releaseNotes> <copyright>Copyright 2015, Google Inc.</copyright> <tags>gRPC RPC Protocol HTTP/2</tags> <dependencies> - <dependency id="Microsoft.Bcl.Immutable" version="1.0.34" /> + <dependency id="System.Collections.Immutable" version="1.1.36" /> <dependency id="Ix-Async" version="1.2.3" /> - <dependency id="grpc.native.csharp_ext" version="0.9.1" /> + <dependency id="grpc.native.csharp_ext" version="0.10.0" /> </dependencies> </metadata> <files> diff --git a/src/csharp/Grpc.Core/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Core/Properties/AssemblyInfo.cs index 03b682181a..c57eef65aa 100644 --- a/src/csharp/Grpc.Core/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Core/Properties/AssemblyInfo.cs @@ -9,6 +9,6 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] [assembly: InternalsVisibleTo("Grpc.Core.Tests")] diff --git a/src/csharp/Grpc.Core/packages.config b/src/csharp/Grpc.Core/packages.config index fb7eaaeeda..6cdcdf2656 100644 --- a/src/csharp/Grpc.Core/packages.config +++ b/src/csharp/Grpc.Core/packages.config @@ -3,5 +3,5 @@ <package id="grpc.dependencies.openssl.redist" version="1.0.2.2" targetFramework="net45" /> <package id="grpc.dependencies.zlib.redist" version="1.2.8.9" targetFramework="net45" /> <package id="Ix-Async" version="1.2.3" targetFramework="net45" /> - <package id="Microsoft.Bcl.Immutable" version="1.0.34" targetFramework="net45" /> + <package id="System.Collections.Immutable" version="1.1.36" targetFramework="net45" /> </packages>
\ No newline at end of file diff --git a/src/csharp/Grpc.Examples.MathClient/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Examples.MathClient/Properties/AssemblyInfo.cs index 4b77472fbd..a57c540215 100644 --- a/src/csharp/Grpc.Examples.MathClient/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Examples.MathClient/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.Examples.MathServer/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Examples.MathServer/Properties/AssemblyInfo.cs index c18fc251d4..6c772cb45c 100644 --- a/src/csharp/Grpc.Examples.MathServer/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Examples.MathServer/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.Examples.Tests/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Examples.Tests/Properties/AssemblyInfo.cs index c1ba396a95..4acaeaa438 100644 --- a/src/csharp/Grpc.Examples.Tests/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Examples.Tests/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.Examples/Properties/AssemblyInfo.cs b/src/csharp/Grpc.Examples/Properties/AssemblyInfo.cs index a63e05d7f6..60a7aaea13 100644 --- a/src/csharp/Grpc.Examples/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.Examples/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.IntegrationTesting.Client/Properties/AssemblyInfo.cs b/src/csharp/Grpc.IntegrationTesting.Client/Properties/AssemblyInfo.cs index 26c0dcc3b9..9a389a1e30 100644 --- a/src/csharp/Grpc.IntegrationTesting.Client/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.IntegrationTesting.Client/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.IntegrationTesting.Server/Properties/AssemblyInfo.cs b/src/csharp/Grpc.IntegrationTesting.Server/Properties/AssemblyInfo.cs index 2d518d7b72..ff31035d53 100644 --- a/src/csharp/Grpc.IntegrationTesting.Server/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.IntegrationTesting.Server/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj index b3a0a2917b..a6d847ca65 100644 --- a/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj +++ b/src/csharp/Grpc.IntegrationTesting/Grpc.IntegrationTesting.csproj @@ -54,6 +54,10 @@ <Reference Include="Google.ProtocolBuffers"> <HintPath>..\packages\Google.ProtocolBuffers.2.4.1.521\lib\net40\Google.ProtocolBuffers.dll</HintPath> </Reference> + <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> @@ -69,9 +73,6 @@ <Reference Include="Newtonsoft.Json"> <HintPath>..\packages\Newtonsoft.Json.6.0.6\lib\net45\Newtonsoft.Json.dll</HintPath> </Reference> - <Reference Include="System.Collections.Immutable"> - <HintPath>..\packages\Microsoft.Bcl.Immutable.1.0.34\lib\portable-net45+win8+wp8+wpa81\System.Collections.Immutable.dll</HintPath> - </Reference> </ItemGroup> <ItemGroup> <Compile Include="Properties\AssemblyInfo.cs" /> diff --git a/src/csharp/Grpc.IntegrationTesting/Properties/AssemblyInfo.cs b/src/csharp/Grpc.IntegrationTesting/Properties/AssemblyInfo.cs index f73575e8bd..7134b04892 100644 --- a/src/csharp/Grpc.IntegrationTesting/Properties/AssemblyInfo.cs +++ b/src/csharp/Grpc.IntegrationTesting/Properties/AssemblyInfo.cs @@ -9,4 +9,4 @@ using System.Runtime.CompilerServices; [assembly: AssemblyCopyright("Google Inc. All rights reserved.")] [assembly: AssemblyTrademark("")] [assembly: AssemblyCulture("")] -[assembly: AssemblyVersion("0.5.*")] +[assembly: AssemblyVersion("0.6.*")] diff --git a/src/csharp/Grpc.IntegrationTesting/packages.config b/src/csharp/Grpc.IntegrationTesting/packages.config index 291b7b8599..c74ac75d79 100644 --- a/src/csharp/Grpc.IntegrationTesting/packages.config +++ b/src/csharp/Grpc.IntegrationTesting/packages.config @@ -7,8 +7,8 @@ <package id="Microsoft.Bcl" version="1.1.9" targetFramework="net45" /> <package id="Microsoft.Bcl.Async" version="1.0.168" targetFramework="net45" /> <package id="Microsoft.Bcl.Build" version="1.0.14" targetFramework="net45" /> - <package id="Microsoft.Bcl.Immutable" version="1.0.34" targetFramework="net45" /> <package id="Microsoft.Net.Http" version="2.2.28" targetFramework="net45" /> <package id="Newtonsoft.Json" version="6.0.6" targetFramework="net45" /> <package id="NUnit" version="2.6.4" targetFramework="net45" /> + <package id="System.Collections.Immutable" version="1.1.36" targetFramework="net45" /> </packages>
\ No newline at end of file diff --git a/src/csharp/Grpc.Tools.nuspec b/src/csharp/Grpc.Tools.nuspec index 913d4c8f4b..0f4fa79277 100644 --- a/src/csharp/Grpc.Tools.nuspec +++ b/src/csharp/Grpc.Tools.nuspec @@ -5,13 +5,13 @@ <title>gRPC C# Tools</title> <summary>Tools for C# implementation of gRPC - an RPC library and framework</summary> <description>Precompiled Windows binaries for generating protocol buffer messages and gRPC client/server code</description> - <version>0.5.1</version> + <version>0.6.0</version> <authors>Google Inc.</authors> <owners>grpc-packages</owners> <licenseUrl>https://github.com/grpc/grpc/blob/master/LICENSE</licenseUrl> <projectUrl>https://github.com/grpc/grpc</projectUrl> <requireLicenseAcceptance>false</requireLicenseAcceptance> - <releaseNotes>protoc.exe - protocol buffer compiler v3.0.0-alpha-3; grpc_csharp_plugin.exe - gRPC C# protoc plugin version 0.5.1</releaseNotes> + <releaseNotes>protoc.exe - protocol buffer compiler v3.0.0-alpha-3; grpc_csharp_plugin.exe - gRPC C# protoc plugin version 0.6.0</releaseNotes> <copyright>Copyright 2015, Google Inc.</copyright> <tags>gRPC RPC Protocol HTTP/2</tags> </metadata> diff --git a/src/csharp/Grpc.nuspec b/src/csharp/Grpc.nuspec index cf4c74fa2d..70203a6203 100644 --- a/src/csharp/Grpc.nuspec +++ b/src/csharp/Grpc.nuspec @@ -5,17 +5,17 @@ <title>gRPC C#</title> <summary>C# implementation of gRPC - an RPC library and framework</summary> <description>C# implementation of gRPC - an RPC library and framework. See project site for more info.</description> - <version>0.5.1</version> + <version>0.6.0</version> <authors>Google Inc.</authors> <owners>grpc-packages</owners> <licenseUrl>https://github.com/grpc/grpc/blob/master/LICENSE</licenseUrl> <projectUrl>https://github.com/grpc/grpc</projectUrl> <requireLicenseAcceptance>false</requireLicenseAcceptance> - <releaseNotes>Release 0.5.1 of gRPC C#</releaseNotes> + <releaseNotes>Release 0.6.0 of gRPC C#</releaseNotes> <copyright>Copyright 2015, Google Inc.</copyright> <tags>gRPC RPC Protocol HTTP/2</tags> <dependencies> - <dependency id="Grpc.Core" version="0.5.1" /> + <dependency id="Grpc.Core" version="0.6.0" /> </dependencies> </metadata> <files/> diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c index ec125db78b..a55cc9e247 100644 --- a/src/csharp/ext/grpc_csharp_ext.c +++ b/src/csharp/ext/grpc_csharp_ext.c @@ -379,7 +379,7 @@ grpcsharp_channel_args_destroy(grpc_channel_args *args) { /* Timespec */ -GPR_EXPORT gpr_timespec GPR_CALLTYPE gprsharp_now(void) { return gpr_now(); } +GPR_EXPORT gpr_timespec GPR_CALLTYPE gprsharp_now(void) { return gpr_now(GPR_CLOCK_REALTIME); } GPR_EXPORT gpr_timespec GPR_CALLTYPE gprsharp_inf_future(void) { return gpr_inf_future; diff --git a/src/node/health_check/health.js b/src/node/health_check/health.js new file mode 100644 index 0000000000..87e00197fe --- /dev/null +++ b/src/node/health_check/health.js @@ -0,0 +1,70 @@ +/* + * + * 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. + * + */ + +'use strict'; + +var grpc = require('../'); + +var _ = require('lodash'); + +var health_proto = grpc.load(__dirname + '/health.proto'); + +var HealthClient = health_proto.grpc.health.v1alpha.Health; + +function HealthImplementation(statusMap) { + this.statusMap = _.clone(statusMap); +} + +HealthImplementation.prototype.setStatus = function(host, service, status) { + if (!this.statusMap[host]) { + this.statusMap[host] = {}; + } + this.statusMap[host][service] = status; +}; + +HealthImplementation.prototype.check = function(call, callback){ + var host = call.request.host; + var service = call.request.service; + var status = _.get(this.statusMap, [host, service], null); + if (status === null) { + callback({code:grpc.status.NOT_FOUND}); + } else { + callback(null, {status: status}); + } +}; + +module.exports = { + Client: HealthClient, + service: HealthClient.service, + Implementation: HealthImplementation +}; diff --git a/src/node/health_check/health.proto b/src/node/health_check/health.proto new file mode 100644 index 0000000000..d31df1e0a7 --- /dev/null +++ b/src/node/health_check/health.proto @@ -0,0 +1,50 @@ +// 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. + +syntax = "proto3"; + +package grpc.health.v1alpha; + +message HealthCheckRequest { + string host = 1; + string service = 2; +} + +message HealthCheckResponse { + enum ServingStatus { + UNKNOWN = 0; + SERVING = 1; + NOT_SERVING = 2; + } + ServingStatus status = 1; +} + +service Health { + rpc Check(HealthCheckRequest) returns (HealthCheckResponse); +}
\ No newline at end of file diff --git a/src/node/package.json b/src/node/package.json index 6b545705e1..1caf158792 100644 --- a/src/node/package.json +++ b/src/node/package.json @@ -1,6 +1,6 @@ { "name": "grpc", - "version": "0.9.1", + "version": "0.10.0", "author": "Google Inc.", "description": "gRPC Library for Node", "homepage": "http://www.grpc.io/", diff --git a/src/node/src/server.js b/src/node/src/server.js index c6cf9e7eb8..9ac428f3ee 100644 --- a/src/node/src/server.js +++ b/src/node/src/server.js @@ -634,7 +634,8 @@ function makeServerConstructor(service_attr_map) { } var serialize = attrs.responseSerialize; var deserialize = attrs.requestDeserialize; - server.register(attrs.path, service_handlers[service_name][name], + server.register(attrs.path, _.bind(service_handlers[service_name][name], + service_handlers[service_name]), serialize, deserialize, method_type); }); }, this); diff --git a/src/node/test/health_test.js b/src/node/test/health_test.js new file mode 100644 index 0000000000..4d1a5082e0 --- /dev/null +++ b/src/node/test/health_test.js @@ -0,0 +1,103 @@ +/* + * + * 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. + * + */ + +'use strict'; + +var assert = require('assert'); + +var health = require('../health_check/health.js'); + +var grpc = require('../'); + +describe('Health Checking', function() { + var statusMap = { + '': { + '': 'SERVING', + 'grpc.test.TestService': 'NOT_SERVING', + }, + virtual_host: { + 'grpc.test.TestService': 'SERVING' + } + }; + var HealthServer = grpc.buildServer([health.service]); + var healthServer = new HealthServer({ + 'grpc.health.v1alpha.Health': new health.Implementation(statusMap) + }); + var healthClient; + before(function() { + var port_num = healthServer.bind('0.0.0.0:0'); + healthServer.listen(); + healthClient = new health.Client('localhost:' + port_num); + }); + after(function() { + healthServer.shutdown(); + }); + it('should say an enabled service is SERVING', function(done) { + healthClient.check({service: ''}, function(err, response) { + assert.ifError(err); + assert.strictEqual(response.status, 'SERVING'); + done(); + }); + }); + it('should say that a disabled service is NOT_SERVING', function(done) { + healthClient.check({service: 'grpc.test.TestService'}, + function(err, response) { + assert.ifError(err); + assert.strictEqual(response.status, 'NOT_SERVING'); + done(); + }); + }); + it('should say that a service on another host is SERVING', function(done) { + healthClient.check({host: 'virtual_host', service: 'grpc.test.TestService'}, + function(err, response) { + assert.ifError(err); + assert.strictEqual(response.status, 'SERVING'); + done(); + }); + }); + it('should get NOT_FOUND if the service is not registered', function(done) { + healthClient.check({service: 'not_registered'}, function(err, response) { + assert(err); + assert.strictEqual(err.code, grpc.status.NOT_FOUND); + done(); + }); + }); + it('should get NOT_FOUND if the host is not registered', function(done) { + healthClient.check({host: 'wrong_host', service: 'grpc.test.TestService'}, + function(err, response) { + assert(err); + assert.strictEqual(err.code, grpc.status.NOT_FOUND); + done(); + }); + }); +}); diff --git a/src/php/ext/grpc/README.md b/src/php/ext/grpc/README.md index 0ac09e1835..88d2c481ec 100644 --- a/src/php/ext/grpc/README.md +++ b/src/php/ext/grpc/README.md @@ -4,7 +4,7 @@ gRPC PHP Extension # Requirements * PHP 5.5+ - * [gRPC core library](https://github.com/grpc/grpc) 0.9.1 + * [gRPC core library](https://github.com/grpc/grpc) 0.10.0 # Installation @@ -55,7 +55,7 @@ $ sudo pecl install grpc Note: before a stable release, you may need to do ```sh -$ sudo pecl install grpc-0.5.0 +$ sudo pecl install grpc-0.5.1 ``` OR diff --git a/src/php/ext/grpc/package.xml b/src/php/ext/grpc/package.xml index 2c89829d51..bc2a05923b 100644 --- a/src/php/ext/grpc/package.xml +++ b/src/php/ext/grpc/package.xml @@ -10,11 +10,11 @@ <email>grpc-packages@google.com</email> <active>yes</active> </lead> - <date>2015-06-16</date> - <time>20:12:55</time> + <date>2015-07-09</date> + <time>21:47:27</time> <version> - <release>0.5.0</release> - <api>0.5.0</api> + <release>0.5.1</release> + <api>0.5.1</api> </version> <stability> <release>alpha</release> @@ -22,7 +22,7 @@ </stability> <license>BSD</license> <notes> -First alpha release +Update to wrap gRPC C Core version 0.10.0 </notes> <contents> <dir baseinstalldir="/" name="/"> @@ -34,15 +34,15 @@ First alpha release <file baseinstalldir="/" md5sum="f1b66029daeced20b47cf00cc6523fc8" name="channel.h" role="src" /> <file baseinstalldir="/" md5sum="81a1193e93d8b6602add8ac360de565b" name="completion_queue.c" role="src" /> <file baseinstalldir="/" md5sum="f10b5bb232d74a6878e829e2e76cdaa2" name="completion_queue.h" role="src" /> - <file baseinstalldir="/" md5sum="a9181ed994a072ac5f41e7c9705c170f" name="config.m4" role="src" /> + <file baseinstalldir="/" md5sum="a22f8eac0164761058cc4d9eb2ceb069" name="config.m4" role="src" /> <file baseinstalldir="/" md5sum="8c3f1e11dac623001378bfd53b554f08" name="credentials.c" role="src" /> <file baseinstalldir="/" md5sum="6988d6e97c19c8f8e3eb92371cf8246b" name="credentials.h" role="src" /> <file baseinstalldir="/" md5sum="38a1bc979d810c36ebc2a52d4b7b5319" name="CREDITS" role="doc" /> <file baseinstalldir="/" md5sum="3f35b472bbdef5a788cd90617d7d0847" name="LICENSE" role="doc" /> <file baseinstalldir="/" md5sum="6aaa7a290122d230f2d8c4e4e05da4a9" name="php_grpc.c" role="src" /> <file baseinstalldir="/" md5sum="673b07859d9f69232f8a754c56780686" name="php_grpc.h" role="src" /> - <file baseinstalldir="/" md5sum="4d4d3382f8d10cae2e4378468e5516b9" name="README.md" role="doc" /> - <file baseinstalldir="/" md5sum="53fda0ee6937f6ddc8e271886018d441" name="server.c" role="src" /> + <file baseinstalldir="/" md5sum="c1d0b42fd77b7d6740bf7744bee90af5" name="README.md" role="doc" /> + <file baseinstalldir="/" md5sum="30997dd423403e1f8ad09dcee598e5c4" name="server.c" role="src" /> <file baseinstalldir="/" md5sum="4b730f06d14cbbb0642bdbd194749595" name="server.h" role="src" /> <file baseinstalldir="/" md5sum="f6930beafb6c0e061899262f2f077e98" name="server_credentials.c" role="src" /> <file baseinstalldir="/" md5sum="9c4b4cc06356a8a39a16a085a9b85996" name="server_credentials.h" role="src" /> @@ -78,5 +78,20 @@ First alpha release First alpha release </notes> </release> + <release> + <version> + <release>0.5.1</release> + <api>0.5.1</api> + </version> + <stability> + <release>alpha</release> + <api>alpha</api> + </stability> + <date>2015-07-09</date> + <license>BSD</license> + <notes> +Update to wrap gRPC C Core version 0.10.0 + </notes> + </release> </changelog> </package> diff --git a/src/php/ext/grpc/timeval.c b/src/php/ext/grpc/timeval.c index 8a278d6760..ccf7f0f81a 100644 --- a/src/php/ext/grpc/timeval.c +++ b/src/php/ext/grpc/timeval.c @@ -208,7 +208,7 @@ PHP_METHOD(Timeval, similar) { * @return Timeval The current time */ PHP_METHOD(Timeval, now) { - zval *now = grpc_php_wrap_timeval(gpr_now()); + zval *now = grpc_php_wrap_timeval(gpr_now(GPR_CLOCK_REALTIME)); RETURN_DESTROY_ZVAL(now); } diff --git a/src/python/interop/setup.py b/src/python/interop/setup.py index 502fcbedd8..75012b0d8f 100644 --- a/src/python/interop/setup.py +++ b/src/python/interop/setup.py @@ -45,7 +45,7 @@ _PACKAGE_DATA = { 'credentials/server1.pem',] } -_INSTALL_REQUIRES = ['oauth2client>=1.4.7', 'grpcio>=0.4.0a4'] +_INSTALL_REQUIRES = ['oauth2client>=1.4.7', 'grpcio>=0.10.0a0'] setuptools.setup( name='interop', diff --git a/src/python/requirements.txt b/src/python/requirements.txt index 43395df03b..41d633a2dd 100644 --- a/src/python/requirements.txt +++ b/src/python/requirements.txt @@ -1,3 +1,4 @@ enum34==1.0.4 futures==2.2.0 protobuf==3.0.0a3 +cython>=0.22 diff --git a/src/python/src/.gitignore b/src/python/src/.gitignore index bc15a52cf1..144e501237 100644 --- a/src/python/src/.gitignore +++ b/src/python/src/.gitignore @@ -1,3 +1,4 @@ MANIFEST grpcio.egg-info/ +build/ dist/ diff --git a/src/python/src/grpc/_adapter/.gitignore b/src/python/src/grpc/_adapter/.gitignore new file mode 100644 index 0000000000..a6f96cd6db --- /dev/null +++ b/src/python/src/grpc/_adapter/.gitignore @@ -0,0 +1,5 @@ +*.a +*.so +*.dll +*.pyc +*.pyd diff --git a/src/python/src/grpc/_cython/.gitignore b/src/python/src/grpc/_cython/.gitignore new file mode 100644 index 0000000000..c315029288 --- /dev/null +++ b/src/python/src/grpc/_cython/.gitignore @@ -0,0 +1,7 @@ +*.h +*.c +*.a +*.so +*.dll +*.pyc +*.pyd diff --git a/src/python/src/grpc/_cython/README.rst b/src/python/src/grpc/_cython/README.rst new file mode 100644 index 0000000000..c0e66734e8 --- /dev/null +++ b/src/python/src/grpc/_cython/README.rst @@ -0,0 +1,52 @@ +GRPC Python Cython layer +======================== + +Package for the GRPC Python Cython layer. + +What is Cython? +--------------- + +Cython is both a superset of the Python language with extensions for dealing +with C types and a tool that transpiles this superset into C code. It provides +convenient means of statically typing expressions and of converting Python +strings to pointers (among other niceties), thus dramatically smoothing the +Python/C interop by allowing fluid use of APIs in both from the same source. +See the wonderful `Cython website`_. + +Why Cython? +----------- + +- **Python 2 and 3 support** + Cython generated C code has precompiler macros to target both Python 2 and + Python 3 C APIs, even while acting as a superset of just the Python 2 + language (e.g. using ``basestring``). +- **Significantly less semantic noise** + A lot of CPython code is just glue, especially human-error-prone + ``Py_INCREF``-ing and ``Py_DECREF``-ing around error handlers and such. + Cython takes care of that automagically. +- **Possible PyPy support** + One of the major developments in Cython over the past few years was the + addition of support for PyPy. We might soon be able to provide such support + ourselves through our use of Cython. +- **Less Python glue code** + There existed several adapter layers in and around the original CPython code + to smooth the surface exposed to Python due to how much trouble it was to + make such a smooth surface via the CPython API alone. Cython makes writing + such a surface incredibly easy, so these adapter layers may be removed. + +Implications for Users +---------------------- + +Nothing additional will be required for users. PyPI packages will contain +Cython generated C code and thus not necessitate a Cython installation. + +Implications for GRPC Developers +-------------------------------- + +A typical edit-compile-debug cycle now requires Cython. We install Cython in +the ``virtualenv`` generated for the Python tests in this repository, so +initial test runs may take an extra 2+ minutes to complete. Subsequent test +runs won't reinstall ``Cython`` (unless required versions change and the +``virtualenv`` doesn't have installed versions that satisfy the change). + +.. _`Cython website`: http://cython.org/ diff --git a/src/python/src/grpc/_cython/__init__.py b/src/python/src/grpc/_cython/__init__.py new file mode 100644 index 0000000000..b89398809f --- /dev/null +++ b/src/python/src/grpc/_cython/__init__.py @@ -0,0 +1,28 @@ +# 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. diff --git a/src/python/src/grpc/_cython/_cygrpc/__init__.py b/src/python/src/grpc/_cython/_cygrpc/__init__.py new file mode 100644 index 0000000000..b89398809f --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/__init__.py @@ -0,0 +1,28 @@ +# 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. diff --git a/src/python/src/grpc/_cython/_cygrpc/call.pxd b/src/python/src/grpc/_cython/_cygrpc/call.pxd new file mode 100644 index 0000000000..fe9b81e3d3 --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/call.pxd @@ -0,0 +1,37 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc + + +cdef class Call: + + cdef grpc.grpc_call *c_call + cdef list references + diff --git a/src/python/src/grpc/_cython/_cygrpc/call.pyx b/src/python/src/grpc/_cython/_cygrpc/call.pyx new file mode 100644 index 0000000000..4349786b3a --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/call.pyx @@ -0,0 +1,82 @@ +# 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. + +cimport cpython + +from grpc._cython._cygrpc cimport records + + +cdef class Call: + + def __cinit__(self): + # Create an *empty* call + self.c_call = NULL + self.references = [] + + def start_batch(self, operations, tag): + if not self.is_valid: + raise ValueError("invalid call object cannot be used from Python") + cdef records.Operations cy_operations = records.Operations(operations) + cdef records.OperationTag operation_tag = records.OperationTag(tag) + operation_tag.operation_call = self + operation_tag.batch_operations = cy_operations + cpython.Py_INCREF(operation_tag) + return grpc.grpc_call_start_batch( + self.c_call, cy_operations.c_ops, cy_operations.c_nops, + <cpython.PyObject *>operation_tag) + + def cancel(self, + grpc.grpc_status_code error_code=grpc.GRPC_STATUS__DO_NOT_USE, + details=None): + if not self.is_valid: + raise ValueError("invalid call object cannot be used from Python") + if (details is None) != (error_code == grpc.GRPC_STATUS__DO_NOT_USE): + raise ValueError("if error_code is specified, so must details " + "(and vice-versa)") + if isinstance(details, bytes): + pass + elif isinstance(details, basestring): + details = details.encode() + else: + raise TypeError("expected details to be str or bytes") + if error_code != grpc.GRPC_STATUS__DO_NOT_USE: + self.references.append(details) + return grpc.grpc_call_cancel_with_status(self.c_call, error_code, details) + else: + return grpc.grpc_call_cancel(self.c_call) + + def __dealloc__(self): + if self.c_call != NULL: + grpc.grpc_call_destroy(self.c_call) + + # The object *should* always be valid from Python. Used for debugging. + @property + def is_valid(self): + return self.c_call != NULL + diff --git a/src/python/src/grpc/_cython/_cygrpc/channel.pxd b/src/python/src/grpc/_cython/_cygrpc/channel.pxd new file mode 100644 index 0000000000..3e341bf222 --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/channel.pxd @@ -0,0 +1,36 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc + + +cdef class Channel: + + cdef grpc.grpc_channel *c_channel + cdef list references diff --git a/src/python/src/grpc/_cython/_cygrpc/channel.pyx b/src/python/src/grpc/_cython/_cygrpc/channel.pyx new file mode 100644 index 0000000000..b20313818d --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/channel.pyx @@ -0,0 +1,84 @@ +# 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. + +from grpc._cython._cygrpc cimport call +from grpc._cython._cygrpc cimport completion_queue +from grpc._cython._cygrpc cimport credentials +from grpc._cython._cygrpc cimport records + + +cdef class Channel: + + def __cinit__(self, target, records.ChannelArgs arguments=None, + credentials.ClientCredentials client_credentials=None): + cdef grpc.grpc_channel_args *c_arguments = NULL + self.c_channel = NULL + self.references = [] + if arguments is not None: + c_arguments = &arguments.c_args + if isinstance(target, bytes): + pass + elif isinstance(target, basestring): + target = target.encode() + else: + raise TypeError("expected target to be str or bytes") + if client_credentials is None: + self.c_channel = grpc.grpc_channel_create(target, c_arguments) + else: + self.c_channel = grpc.grpc_secure_channel_create( + client_credentials.c_credentials, target, c_arguments) + self.references.append(client_credentials) + self.references.append(target) + self.references.append(arguments) + + def create_call(self, completion_queue.CompletionQueue queue not None, + method, host, records.Timespec deadline not None): + if queue.is_shutting_down: + raise ValueError("queue must not be shutting down or shutdown") + if isinstance(method, bytes): + pass + elif isinstance(method, basestring): + method = method.encode() + else: + raise TypeError("expected method to be str or bytes") + if isinstance(host, bytes): + pass + elif isinstance(host, basestring): + host = host.encode() + else: + raise TypeError("expected host to be str or bytes") + cdef call.Call operation_call = call.Call() + operation_call.references = [self, method, host, queue] + operation_call.c_call = grpc.grpc_channel_create_call( + self.c_channel, queue.c_completion_queue, method, host, deadline.c_time) + return operation_call + + def __dealloc__(self): + if self.c_channel != NULL: + grpc.grpc_channel_destroy(self.c_channel) diff --git a/src/python/src/grpc/_cython/_cygrpc/completion_queue.pxd b/src/python/src/grpc/_cython/_cygrpc/completion_queue.pxd new file mode 100644 index 0000000000..fd562ad75b --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/completion_queue.pxd @@ -0,0 +1,39 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc + + +cdef class CompletionQueue: + + cdef grpc.grpc_completion_queue *c_completion_queue + cdef object poll_condition + cdef bint is_polling + cdef bint is_shutting_down + cdef bint is_shutdown diff --git a/src/python/src/grpc/_cython/_cygrpc/completion_queue.pyx b/src/python/src/grpc/_cython/_cygrpc/completion_queue.pyx new file mode 100644 index 0000000000..886d85360a --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/completion_queue.pyx @@ -0,0 +1,117 @@ +# 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. + +cimport cpython + +from grpc._cython._cygrpc cimport call +from grpc._cython._cygrpc cimport records + +import threading +import time + + +cdef class CompletionQueue: + + def __cinit__(self): + self.c_completion_queue = grpc.grpc_completion_queue_create() + self.is_shutting_down = False + self.is_shutdown = False + self.poll_condition = threading.Condition() + self.is_polling = False + + def poll(self, records.Timespec deadline=None): + # We name this 'poll' to avoid problems with CPython's expectations for + # 'special' methods (like next and __next__). + cdef grpc.gpr_timespec c_deadline = grpc.gpr_inf_future + cdef records.OperationTag tag = None + cdef object user_tag = None + cdef call.Call operation_call = None + cdef records.CallDetails request_call_details = None + cdef records.Metadata request_metadata = None + cdef records.Operations batch_operations = None + if deadline is not None: + c_deadline = deadline.c_time + cdef grpc.grpc_event event + + # Poll within a critical section + with self.poll_condition: + while self.is_polling: + self.poll_condition.wait(float(deadline) - time.time()) + self.is_polling = True + with nogil: + event = grpc.grpc_completion_queue_next( + self.c_completion_queue, c_deadline) + with self.poll_condition: + self.is_polling = False + self.poll_condition.notify() + + if event.type == grpc.GRPC_QUEUE_TIMEOUT: + return records.Event(event.type, False, None, None, None, None, None) + elif event.type == grpc.GRPC_QUEUE_SHUTDOWN: + self.is_shutdown = True + return records.Event(event.type, True, None, None, None, None, None) + else: + if event.tag != NULL: + tag = <records.OperationTag>event.tag + # We receive event tags only after they've been inc-ref'd elsewhere in + # the code. + cpython.Py_DECREF(tag) + if tag.shutting_down_server is not None: + tag.shutting_down_server.notify_shutdown_complete() + user_tag = tag.user_tag + operation_call = tag.operation_call + request_call_details = tag.request_call_details + request_metadata = tag.request_metadata + batch_operations = tag.batch_operations + if tag.is_new_request: + # Stuff in the tag not explicitly handled by us needs to live through + # the life of the call + operation_call.references.extend(tag.references) + return records.Event( + event.type, event.success, user_tag, operation_call, + request_call_details, request_metadata, batch_operations) + + def shutdown(self): + grpc.grpc_completion_queue_shutdown(self.c_completion_queue) + self.is_shutting_down = True + + def clear(self): + if not self.is_shutting_down: + raise ValueError('queue must be shutting down to be cleared') + while self.poll().type != grpc.GRPC_QUEUE_SHUTDOWN: + pass + + def __dealloc__(self): + if self.c_completion_queue != NULL: + # Ensure shutdown, pump the queue + if not self.is_shutting_down: + self.shutdown() + while not self.is_shutdown: + self.poll() + grpc.grpc_completion_queue_destroy(self.c_completion_queue) diff --git a/src/python/src/grpc/_cython/_cygrpc/credentials.pxd b/src/python/src/grpc/_cython/_cygrpc/credentials.pxd new file mode 100644 index 0000000000..6b74a267e0 --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/credentials.pxd @@ -0,0 +1,45 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc + + +cdef class ClientCredentials: + + cdef grpc.grpc_credentials *c_credentials + cdef grpc.grpc_ssl_pem_key_cert_pair c_ssl_pem_key_cert_pair + cdef list references + + +cdef class ServerCredentials: + + cdef grpc.grpc_server_credentials *c_credentials + cdef grpc.grpc_ssl_pem_key_cert_pair *c_ssl_pem_key_cert_pairs + cdef size_t c_ssl_pem_key_cert_pairs_count + cdef list references diff --git a/src/python/src/grpc/_cython/_cygrpc/credentials.pyx b/src/python/src/grpc/_cython/_cygrpc/credentials.pyx new file mode 100644 index 0000000000..c14d8844dd --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/credentials.pyx @@ -0,0 +1,217 @@ +# 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. + +from grpc._cython._cygrpc cimport records + + +cdef class ClientCredentials: + + def __cinit__(self): + self.c_credentials = NULL + self.c_ssl_pem_key_cert_pair.private_key = NULL + self.c_ssl_pem_key_cert_pair.certificate_chain = NULL + self.references = [] + + # The object *can* be invalid in Python if we fail to make the credentials + # (and the core thus returns NULL credentials). Used primarily for debugging. + @property + def is_valid(self): + return self.c_credentials != NULL + + def __dealloc__(self): + if self.c_credentials != NULL: + grpc.grpc_credentials_release(self.c_credentials) + + +cdef class ServerCredentials: + + def __cinit__(self): + self.c_credentials = NULL + + def __dealloc__(self): + if self.c_credentials != NULL: + grpc.grpc_server_credentials_release(self.c_credentials) + + +def client_credentials_google_default(): + cdef ClientCredentials credentials = ClientCredentials(); + credentials.c_credentials = grpc.grpc_google_default_credentials_create() + return credentials + +def client_credentials_ssl(pem_root_certificates, + records.SslPemKeyCertPair ssl_pem_key_cert_pair): + if pem_root_certificates is None: + pass + elif isinstance(pem_root_certificates, bytes): + pass + elif isinstance(pem_root_certificates, basestring): + pem_root_certificates = pem_root_certificates.encode() + else: + raise TypeError("expected str or bytes for pem_root_certificates") + cdef ClientCredentials credentials = ClientCredentials() + cdef const char *c_pem_root_certificates = NULL + if pem_root_certificates is not None: + c_pem_root_certificates = pem_root_certificates + credentials.references.append(pem_root_certificates) + if ssl_pem_key_cert_pair is not None: + credentials.c_credentials = grpc.grpc_ssl_credentials_create( + c_pem_root_certificates, &ssl_pem_key_cert_pair.c_pair + ) + credentials.references.append(ssl_pem_key_cert_pair) + else: + credentials.c_credentials = grpc.grpc_ssl_credentials_create( + c_pem_root_certificates, NULL + ) + +def client_credentials_composite_credentials( + ClientCredentials credentials_1 not None, + ClientCredentials credentials_2 not None): + if not credentials_1.is_valid or not credentials_2.is_valid: + raise ValueError("passed credentials must both be valid") + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = grpc.grpc_composite_credentials_create( + credentials_1.c_credentials, credentials_2.c_credentials) + credentials.references.append(credentials_1) + credentials.references.append(credentials_2) + return credentials + +def client_credentials_compute_engine(): + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = grpc.grpc_compute_engine_credentials_create() + return credentials + +def client_credentials_service_account( + json_key, scope, records.Timespec token_lifetime not None): + if isinstance(json_key, bytes): + pass + elif isinstance(json_key, basestring): + json_key = json_key.encode() + else: + raise TypeError("expected json_key to be str or bytes") + if isinstance(scope, bytes): + pass + elif isinstance(scope, basestring): + scope = scope.encode() + else: + raise TypeError("expected scope to be str or bytes") + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = grpc.grpc_service_account_credentials_create( + json_key, scope, token_lifetime.c_time) + credentials.references.extend([json_key, scope]) + return credentials + +def client_credentials_jwt(json_key, records.Timespec token_lifetime not None): + if isinstance(json_key, bytes): + pass + elif isinstance(json_key, basestring): + json_key = json_key.encode() + else: + raise TypeError("expected json_key to be str or bytes") + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = grpc.grpc_jwt_credentials_create( + json_key, token_lifetime.c_time) + credentials.references.append(json_key) + return credentials + +def client_credentials_refresh_token(json_refresh_token): + if isinstance(json_refresh_token, bytes): + pass + elif isinstance(json_refresh_token, basestring): + json_refresh_token = json_refresh_token.encode() + else: + raise TypeError("expected json_refresh_token to be str or bytes") + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = grpc.grpc_refresh_token_credentials_create( + json_refresh_token) + credentials.references.append(json_refresh_token) + return credentials + +def client_credentials_fake_transport_security(): + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = ( + grpc.grpc_fake_transport_security_credentials_create()) + return credentials + +def client_credentials_iam(authorization_token, authority_selector): + if isinstance(authorization_token, bytes): + pass + elif isinstance(authorization_token, basestring): + authorization_token = authorization_token.encode() + else: + raise TypeError("expected authorization_token to be str or bytes") + if isinstance(authority_selector, bytes): + pass + elif isinstance(authority_selector, basestring): + authority_selector = authority_selector.encode() + else: + raise TypeError("expected authority_selector to be str or bytes") + cdef ClientCredentials credentials = ClientCredentials() + credentials.c_credentials = grpc.grpc_iam_credentials_create( + authorization_token, authority_selector) + credentials.references.append(authorization_token) + credentials.references.append(authority_selector) + return credentials + +def server_credentials_ssl(pem_root_certs, pem_key_cert_pairs): + if pem_root_certs is None: + pass + elif isinstance(pem_root_certs, bytes): + pass + elif isinstance(pem_root_certs, basestring): + pem_root_certs = pem_root_certs.encode() + else: + raise TypeError("expected pem_root_certs to be str or bytes") + pem_key_cert_pairs = list(pem_key_cert_pairs) + for pair in pem_key_cert_pairs: + if not isinstance(pair, records.SslPemKeyCertPair): + raise TypeError("expected pem_key_cert_pairs to be sequence of " + "records.SslPemKeyCertPair") + cdef ServerCredentials credentials = ServerCredentials() + credentials.references.append(pem_key_cert_pairs) + credentials.references.append(pem_root_certs) + credentials.c_ssl_pem_key_cert_pairs_count = len(pem_key_cert_pairs) + credentials.c_ssl_pem_key_cert_pairs = ( + <grpc.grpc_ssl_pem_key_cert_pair *>grpc.gpr_malloc( + sizeof(grpc.grpc_ssl_pem_key_cert_pair) * + credentials.c_ssl_pem_key_cert_pairs_count + )) + for i in range(credentials.c_ssl_pem_key_cert_pairs_count): + credentials.c_ssl_pem_key_cert_pairs[i] = ( + (<records.SslPemKeyCertPair>pem_key_cert_pairs[i]).c_pair) + credentials.c_credentials = grpc.grpc_ssl_server_credentials_create( + pem_root_certs, credentials.c_ssl_pem_key_cert_pairs, + credentials.c_ssl_pem_key_cert_pairs_count + ) + return credentials + +def server_credentials_fake_transport_security(): + cdef ServerCredentials credentials = ServerCredentials() + credentials.c_credentials = ( + grpc.grpc_fake_transport_security_server_credentials_create()) + return credentials diff --git a/src/python/src/grpc/_cython/_cygrpc/grpc.pxd b/src/python/src/grpc/_cython/_cygrpc/grpc.pxd new file mode 100644 index 0000000000..7db8fbe31c --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/grpc.pxd @@ -0,0 +1,344 @@ +# 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. + +cimport libc.time + + +cdef extern from "grpc/support/alloc.h": + void *gpr_malloc(size_t size) + void gpr_free(void *ptr) + void *gpr_realloc(void *p, size_t size) + +cdef extern from "grpc/support/slice.h": + ctypedef struct gpr_slice: + # don't worry about writing out the members of gpr_slice; we never access + # them directly. + pass + + gpr_slice gpr_slice_ref(gpr_slice s) + void gpr_slice_unref(gpr_slice s) + gpr_slice gpr_slice_new(void *p, size_t len, void (*destroy)(void *)) + gpr_slice gpr_slice_new_with_len( + void *p, size_t len, void (*destroy)(void *, size_t)) + gpr_slice gpr_slice_malloc(size_t length) + gpr_slice gpr_slice_from_copied_string(const char *source) + gpr_slice gpr_slice_from_copied_buffer(const char *source, size_t len) + + # Declare functions for function-like macros (because Cython)... + void *gpr_slice_start_ptr "GPR_SLICE_START_PTR" (gpr_slice s) + size_t gpr_slice_length "GPR_SLICE_LENGTH" (gpr_slice s) + + +cdef extern from "grpc/support/port_platform.h": + # As long as the header file gets this type right, we don't need to get this + # type exactly; just close enough that the operations will be supported in the + # underlying C layers. + ctypedef unsigned int gpr_uint32 + + +cdef extern from "grpc/support/time.h": + + ctypedef struct gpr_timespec: + libc.time.time_t seconds "tv_sec" + int nanoseconds "tv_nsec" + + cdef gpr_timespec gpr_time_0 + cdef gpr_timespec gpr_inf_future + cdef gpr_timespec gpr_inf_past + + gpr_timespec gpr_now() + + +cdef extern from "grpc/status.h": + ctypedef enum grpc_status_code: + GRPC_STATUS_OK + GRPC_STATUS_CANCELLED + GRPC_STATUS_UNKNOWN + GRPC_STATUS_INVALID_ARGUMENT + GRPC_STATUS_DEADLINE_EXCEEDED + GRPC_STATUS_NOT_FOUND + GRPC_STATUS_ALREADY_EXISTS + GRPC_STATUS_PERMISSION_DENIED + GRPC_STATUS_UNAUTHENTICATED + GRPC_STATUS_RESOURCE_EXHAUSTED + GRPC_STATUS_FAILED_PRECONDITION + GRPC_STATUS_ABORTED + GRPC_STATUS_OUT_OF_RANGE + GRPC_STATUS_UNIMPLEMENTED + GRPC_STATUS_INTERNAL + GRPC_STATUS_UNAVAILABLE + GRPC_STATUS_DATA_LOSS + GRPC_STATUS__DO_NOT_USE + + +cdef extern from "grpc/byte_buffer_reader.h": + struct grpc_byte_buffer_reader: + # We don't care about the internals + pass + + +cdef extern from "grpc/byte_buffer.h": + ctypedef struct grpc_byte_buffer: + # We don't care about the internals. + pass + + grpc_byte_buffer *grpc_raw_byte_buffer_create(gpr_slice *slices, + size_t nslices) + size_t grpc_byte_buffer_length(grpc_byte_buffer *bb) + void grpc_byte_buffer_destroy(grpc_byte_buffer *byte_buffer) + + void grpc_byte_buffer_reader_init(grpc_byte_buffer_reader *reader, + grpc_byte_buffer *buffer) + int grpc_byte_buffer_reader_next(grpc_byte_buffer_reader *reader, + gpr_slice *slice) + void grpc_byte_buffer_reader_destroy(grpc_byte_buffer_reader *reader) + + +cdef extern from "grpc/grpc.h": + + ctypedef struct grpc_completion_queue: + # We don't care about the internals (and in fact don't know them) + pass + + ctypedef struct grpc_channel: + # We don't care about the internals (and in fact don't know them) + pass + + ctypedef struct grpc_server: + # We don't care about the internals (and in fact don't know them) + pass + + ctypedef struct grpc_call: + # We don't care about the internals (and in fact don't know them) + pass + + ctypedef enum grpc_arg_type: + grpc_arg_string "GRPC_ARG_STRING" + grpc_arg_integer "GRPC_ARG_INTEGER" + grpc_arg_pointer "GRPC_ARG_POINTER" + + ctypedef struct grpc_arg_value_pointer: + void *address "p" + void *(*copy)(void *) + void (*destroy)(void *) + + union grpc_arg_value: + char *string + int integer + grpc_arg_value_pointer pointer + + ctypedef struct grpc_arg: + grpc_arg_type type + char *key + grpc_arg_value value + + ctypedef struct grpc_channel_args: + size_t arguments_length "num_args" + grpc_arg *arguments "args" + + ctypedef enum grpc_call_error: + GRPC_CALL_OK + GRPC_CALL_ERROR + GRPC_CALL_ERROR_NOT_ON_SERVER + GRPC_CALL_ERROR_NOT_ON_CLIENT + GRPC_CALL_ERROR_ALREADY_ACCEPTED + GRPC_CALL_ERROR_ALREADY_INVOKED + GRPC_CALL_ERROR_NOT_INVOKED + GRPC_CALL_ERROR_ALREADY_FINISHED + GRPC_CALL_ERROR_TOO_MANY_OPERATIONS + GRPC_CALL_ERROR_INVALID_FLAGS + GRPC_CALL_ERROR_INVALID_METADATA + + ctypedef struct grpc_metadata: + const char *key + const char *value + size_t value_length + # ignore the 'internal_data.obfuscated' fields. + + ctypedef enum grpc_completion_type: + GRPC_QUEUE_SHUTDOWN + GRPC_QUEUE_TIMEOUT + GRPC_OP_COMPLETE + + ctypedef struct grpc_event: + grpc_completion_type type + int success + void *tag + + ctypedef struct grpc_metadata_array: + size_t count + size_t capacity + grpc_metadata *metadata + + void grpc_metadata_array_init(grpc_metadata_array *array) + void grpc_metadata_array_destroy(grpc_metadata_array *array) + + ctypedef struct grpc_call_details: + char *method + size_t method_capacity + char *host + size_t host_capacity + gpr_timespec deadline + + void grpc_call_details_init(grpc_call_details *details) + void grpc_call_details_destroy(grpc_call_details *details) + + ctypedef enum grpc_op_type: + GRPC_OP_SEND_INITIAL_METADATA + GRPC_OP_SEND_MESSAGE + GRPC_OP_SEND_CLOSE_FROM_CLIENT + GRPC_OP_SEND_STATUS_FROM_SERVER + GRPC_OP_RECV_INITIAL_METADATA + GRPC_OP_RECV_MESSAGE + GRPC_OP_RECV_STATUS_ON_CLIENT + GRPC_OP_RECV_CLOSE_ON_SERVER + + ctypedef struct grpc_op_data_send_initial_metadata: + size_t count + grpc_metadata *metadata + + ctypedef struct grpc_op_data_send_status_from_server: + size_t trailing_metadata_count + grpc_metadata *trailing_metadata + grpc_status_code status + const char *status_details + + ctypedef struct grpc_op_data_recv_status_on_client: + grpc_metadata_array *trailing_metadata + grpc_status_code *status + char **status_details + size_t *status_details_capacity + + ctypedef struct grpc_op_data_recv_close_on_server: + int *cancelled + + union grpc_op_data: + grpc_op_data_send_initial_metadata send_initial_metadata + grpc_byte_buffer *send_message + grpc_op_data_send_status_from_server send_status_from_server + grpc_metadata_array *receive_initial_metadata "recv_initial_metadata" + grpc_byte_buffer **receive_message "recv_message" + grpc_op_data_recv_status_on_client receive_status_on_client "recv_status_on_client" + grpc_op_data_recv_close_on_server receive_close_on_server "recv_close_on_server" + + ctypedef struct grpc_op: + grpc_op_type type "op" + gpr_uint32 flags + grpc_op_data data + + void grpc_init() + void grpc_shutdown() + + grpc_completion_queue *grpc_completion_queue_create() + grpc_event grpc_completion_queue_next(grpc_completion_queue *cq, + gpr_timespec deadline) nogil + void grpc_completion_queue_shutdown(grpc_completion_queue *cq) + void grpc_completion_queue_destroy(grpc_completion_queue *cq) + + grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, + size_t nops, void *tag) + grpc_call_error grpc_call_cancel(grpc_call *call) + grpc_call_error grpc_call_cancel_with_status(grpc_call *call, + grpc_status_code status, + const char *description) + void grpc_call_destroy(grpc_call *call) + + + grpc_channel *grpc_channel_create(const char *target, + const grpc_channel_args *args) + grpc_call *grpc_channel_create_call(grpc_channel *channel, + grpc_completion_queue *completion_queue, + const char *method, const char *host, + gpr_timespec deadline) + void grpc_channel_destroy(grpc_channel *channel) + + grpc_server *grpc_server_create(const grpc_channel_args *args) + grpc_call_error grpc_server_request_call( + grpc_server *server, grpc_call **call, grpc_call_details *details, + grpc_metadata_array *request_metadata, grpc_completion_queue + *cq_bound_to_call, grpc_completion_queue *cq_for_notification, void + *tag_new) + void grpc_server_register_completion_queue(grpc_server *server, + grpc_completion_queue *cq) + int grpc_server_add_http2_port(grpc_server *server, const char *addr) + void grpc_server_start(grpc_server *server) + void grpc_server_shutdown_and_notify( + grpc_server *server, grpc_completion_queue *cq, void *tag) + void grpc_server_cancel_all_calls(grpc_server *server) + void grpc_server_destroy(grpc_server *server) + + +cdef extern from "grpc/grpc_security.h": + + ctypedef struct grpc_ssl_pem_key_cert_pair: + const char *private_key + const char *certificate_chain "cert_chain" + + ctypedef struct grpc_credentials: + # We don't care about the internals (and in fact don't know them) + pass + + grpc_credentials *grpc_google_default_credentials_create() + grpc_credentials *grpc_ssl_credentials_create( + const char *pem_root_certs, grpc_ssl_pem_key_cert_pair *pem_key_cert_pair) + + grpc_credentials *grpc_composite_credentials_create(grpc_credentials *creds1, + grpc_credentials *creds2) + grpc_credentials *grpc_compute_engine_credentials_create() + grpc_credentials *grpc_service_account_credentials_create( + const char *json_key, const char *scope, gpr_timespec token_lifetime) + grpc_credentials *grpc_jwt_credentials_create(const char *json_key, + gpr_timespec token_lifetime) + grpc_credentials *grpc_refresh_token_credentials_create( + const char *json_refresh_token) + grpc_credentials *grpc_fake_transport_security_credentials_create() + grpc_credentials *grpc_iam_credentials_create(const char *authorization_token, + const char *authority_selector) + void grpc_credentials_release(grpc_credentials *creds) + + grpc_channel *grpc_secure_channel_create( + grpc_credentials *creds, const char *target, + const grpc_channel_args *args) + + ctypedef struct grpc_server_credentials: + # We don't care about the internals (and in fact don't know them) + pass + + grpc_server_credentials *grpc_ssl_server_credentials_create( + const char *pem_root_certs, + grpc_ssl_pem_key_cert_pair *pem_key_cert_pairs, + size_t num_key_cert_pairs); + grpc_server_credentials *grpc_fake_transport_security_server_credentials_create() + void grpc_server_credentials_release(grpc_server_credentials *creds) + + int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, + grpc_server_credentials *creds) + + grpc_call_error grpc_call_set_credentials(grpc_call *call, + grpc_credentials *creds) diff --git a/src/python/src/grpc/_cython/_cygrpc/records.pxd b/src/python/src/grpc/_cython/_cygrpc/records.pxd new file mode 100644 index 0000000000..9ee487882a --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/records.pxd @@ -0,0 +1,129 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc +from grpc._cython._cygrpc cimport call +from grpc._cython._cygrpc cimport server + + +cdef class Timespec: + + cdef grpc.gpr_timespec c_time + + +cdef class CallDetails: + + cdef grpc.grpc_call_details c_details + + +cdef class OperationTag: + + cdef object user_tag + cdef list references + # This allows CompletionQueue to notify the Python Server object that the + # underlying GRPC core server has shutdown + cdef server.Server shutting_down_server + cdef call.Call operation_call + cdef CallDetails request_call_details + cdef Metadata request_metadata + cdef Operations batch_operations + cdef bint is_new_request + + +cdef class Event: + + cdef readonly grpc.grpc_completion_type type + cdef readonly bint success + cdef readonly object tag + + # For operations with calls + cdef readonly call.Call operation_call + + # For Server.request_call + cdef readonly CallDetails request_call_details + cdef readonly Metadata request_metadata + + # For Call.start_batch + cdef readonly Operations batch_operations + + +cdef class ByteBuffer: + + cdef grpc.grpc_byte_buffer *c_byte_buffer + + +cdef class SslPemKeyCertPair: + + cdef grpc.grpc_ssl_pem_key_cert_pair c_pair + cdef readonly object private_key, certificate_chain + + +cdef class ChannelArg: + + cdef grpc.grpc_arg c_arg + cdef readonly object key, value + + +cdef class ChannelArgs: + + cdef grpc.grpc_channel_args c_args + cdef list args + + +cdef class Metadatum: + + cdef grpc.grpc_metadata c_metadata + cdef object _key, _value + + +cdef class Metadata: + + cdef grpc.grpc_metadata_array c_metadata_array + cdef object metadata + + +cdef class Operation: + + cdef grpc.grpc_op c_op + cdef ByteBuffer _received_message + cdef Metadata _received_metadata + cdef grpc.grpc_status_code _received_status_code + cdef char *_received_status_details + cdef size_t _received_status_details_capacity + cdef int _received_cancelled + cdef readonly bint is_valid + cdef object references + + +cdef class Operations: + + cdef grpc.grpc_op *c_ops + cdef size_t c_nops + cdef list operations + diff --git a/src/python/src/grpc/_cython/_cygrpc/records.pyx b/src/python/src/grpc/_cython/_cygrpc/records.pyx new file mode 100644 index 0000000000..4814769fd2 --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/records.pyx @@ -0,0 +1,575 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc +from grpc._cython._cygrpc cimport call +from grpc._cython._cygrpc cimport server + + +class StatusCode: + ok = grpc.GRPC_STATUS_OK + cancelled = grpc.GRPC_STATUS_CANCELLED + unknown = grpc.GRPC_STATUS_UNKNOWN + invalid_argument = grpc.GRPC_STATUS_INVALID_ARGUMENT + deadline_exceeded = grpc.GRPC_STATUS_DEADLINE_EXCEEDED + not_found = grpc.GRPC_STATUS_NOT_FOUND + already_exists = grpc.GRPC_STATUS_ALREADY_EXISTS + permission_denied = grpc.GRPC_STATUS_PERMISSION_DENIED + unauthenticated = grpc.GRPC_STATUS_UNAUTHENTICATED + resource_exhausted = grpc.GRPC_STATUS_RESOURCE_EXHAUSTED + failed_precondition = grpc.GRPC_STATUS_FAILED_PRECONDITION + aborted = grpc.GRPC_STATUS_ABORTED + out_of_range = grpc.GRPC_STATUS_OUT_OF_RANGE + unimplemented = grpc.GRPC_STATUS_UNIMPLEMENTED + internal = grpc.GRPC_STATUS_INTERNAL + unavailable = grpc.GRPC_STATUS_UNAVAILABLE + data_loss = grpc.GRPC_STATUS_DATA_LOSS + + +class CallError: + ok = grpc.GRPC_CALL_OK + error = grpc.GRPC_CALL_ERROR + not_on_server = grpc.GRPC_CALL_ERROR_NOT_ON_SERVER + not_on_client = grpc.GRPC_CALL_ERROR_NOT_ON_CLIENT + already_accepted = grpc.GRPC_CALL_ERROR_ALREADY_ACCEPTED + already_invoked = grpc.GRPC_CALL_ERROR_ALREADY_INVOKED + not_invoked = grpc.GRPC_CALL_ERROR_NOT_INVOKED + already_finished = grpc.GRPC_CALL_ERROR_ALREADY_FINISHED + too_many_operations = grpc.GRPC_CALL_ERROR_TOO_MANY_OPERATIONS + invalid_flags = grpc.GRPC_CALL_ERROR_INVALID_FLAGS + invalid_metadata = grpc.GRPC_CALL_ERROR_INVALID_METADATA + + +class CompletionType: + queue_shutdown = grpc.GRPC_QUEUE_SHUTDOWN + queue_timeout = grpc.GRPC_QUEUE_TIMEOUT + operation_complete = grpc.GRPC_OP_COMPLETE + + +class OperationType: + send_initial_metadata = grpc.GRPC_OP_SEND_INITIAL_METADATA + send_message = grpc.GRPC_OP_SEND_MESSAGE + send_close_from_client = grpc.GRPC_OP_SEND_CLOSE_FROM_CLIENT + send_status_from_server = grpc.GRPC_OP_SEND_STATUS_FROM_SERVER + receive_initial_metadata = grpc.GRPC_OP_RECV_INITIAL_METADATA + receive_message = grpc.GRPC_OP_RECV_MESSAGE + receive_status_on_client = grpc.GRPC_OP_RECV_STATUS_ON_CLIENT + receive_close_on_server = grpc.GRPC_OP_RECV_CLOSE_ON_SERVER + + +cdef class Timespec: + + def __cinit__(self, time): + if time is None: + self.c_time = grpc.gpr_now() + elif isinstance(time, float): + if time == float("+inf"): + self.c_time = grpc.gpr_inf_future + elif time == float("-inf"): + self.c_time = grpc.gpr_inf_past + else: + self.c_time.seconds = time + self.c_time.nanoseconds = (time - float(self.c_time.seconds)) * 1e9 + else: + raise TypeError("expected time to be float") + + @property + def seconds(self): + return self.c_time.seconds + + @property + def nanoseconds(self): + return self.c_time.nanoseconds + + def __float__(self): + return <double>self.c_time.seconds + <double>self.c_time.nanoseconds / 1e9 + + infinite_future = Timespec(float("+inf")) + infinite_past = Timespec(float("-inf")) + + +cdef class CallDetails: + + def __cinit__(self): + grpc.grpc_call_details_init(&self.c_details) + + def __dealloc__(self): + grpc.grpc_call_details_destroy(&self.c_details) + + @property + def method(self): + if self.c_details.method != NULL: + return <bytes>self.c_details.method + else: + return None + + @property + def host(self): + if self.c_details.host != NULL: + return <bytes>self.c_details.host + else: + return None + + @property + def deadline(self): + timespec = Timespec(float("-inf")) + timespec.c_time = self.c_details.deadline + return timespec + + +cdef class OperationTag: + + def __cinit__(self, user_tag): + self.user_tag = user_tag + self.references = [] + + +cdef class Event: + + def __cinit__(self, grpc.grpc_completion_type type, bint success, + object tag, call.Call operation_call, + CallDetails request_call_details, + Metadata request_metadata, + Operations batch_operations): + self.type = type + self.success = success + self.tag = tag + self.operation_call = operation_call + self.request_call_details = request_call_details + self.request_metadata = request_metadata + self.batch_operations = batch_operations + + +cdef class ByteBuffer: + + def __cinit__(self, data): + if data is None: + self.c_byte_buffer = NULL + return + if isinstance(data, bytes): + pass + elif isinstance(data, basestring): + data = data.encode() + else: + raise TypeError("expected value to be of type str or bytes") + + cdef char *c_data = data + data_slice = grpc.gpr_slice_from_copied_buffer(c_data, len(data)) + self.c_byte_buffer = grpc.grpc_raw_byte_buffer_create( + &data_slice, 1) + grpc.gpr_slice_unref(data_slice) + + def bytes(self): + cdef grpc.grpc_byte_buffer_reader reader + cdef grpc.gpr_slice data_slice + cdef size_t data_slice_length + cdef void *data_slice_pointer + if self.c_byte_buffer != NULL: + grpc.grpc_byte_buffer_reader_init(&reader, self.c_byte_buffer) + result = b"" + while grpc.grpc_byte_buffer_reader_next(&reader, &data_slice): + data_slice_pointer = grpc.gpr_slice_start_ptr(data_slice) + data_slice_length = grpc.gpr_slice_length(data_slice) + result += (<char *>data_slice_pointer)[:data_slice_length] + grpc.grpc_byte_buffer_reader_destroy(&reader) + return result + else: + return None + + def __len__(self): + if self.c_byte_buffer != NULL: + return grpc.grpc_byte_buffer_length(self.c_byte_buffer) + else: + return 0 + + def __str__(self): + return self.bytes() + + def __dealloc__(self): + if self.c_byte_buffer != NULL: + grpc.grpc_byte_buffer_destroy(self.c_byte_buffer) + + +cdef class SslPemKeyCertPair: + + def __cinit__(self, private_key, certificate_chain): + if isinstance(private_key, bytes): + self.private_key = private_key + elif isinstance(private_key, basestring): + self.private_key = private_key.encode() + else: + raise TypeError("expected private_key to be of type str or bytes") + if isinstance(certificate_chain, bytes): + self.certificate_chain = certificate_chain + elif isinstance(certificate_chain, basestring): + self.certificate_chain = certificate_chain.encode() + else: + raise TypeError("expected certificate_chain to be of type str or bytes " + "or int") + self.c_pair.private_key = self.private_key + self.c_pair.certificate_chain = self.certificate_chain + + +cdef class ChannelArg: + + def __cinit__(self, key, value): + if isinstance(key, bytes): + self.key = key + elif isinstance(key, basestring): + self.key = key.encode() + else: + raise TypeError("expected key to be of type str or bytes") + if isinstance(value, bytes): + self.value = value + self.c_arg.type = grpc.GRPC_ARG_STRING + self.c_arg.value.string = self.value + elif isinstance(value, basestring): + self.value = value.encode() + self.c_arg.type = grpc.GRPC_ARG_STRING + self.c_arg.value.string = self.value + elif isinstance(value, int): + self.value = int(value) + self.c_arg.type = grpc.GRPC_ARG_INTEGER + self.c_arg.value.integer = self.value + else: + raise TypeError("expected value to be of type str or bytes or int") + self.c_arg.key = self.key + + +cdef class ChannelArgs: + + def __cinit__(self, args): + self.args = list(args) + for arg in self.args: + if not isinstance(arg, ChannelArg): + raise TypeError("expected list of ChannelArg") + self.c_args.arguments_length = len(self.args) + self.c_args.arguments = <grpc.grpc_arg *>grpc.gpr_malloc( + self.c_args.arguments_length*sizeof(grpc.grpc_arg) + ) + for i in range(self.c_args.arguments_length): + self.c_args.arguments[i] = (<ChannelArg>self.args[i]).c_arg + + def __dealloc__(self): + grpc.gpr_free(self.c_args.arguments) + + def __len__(self): + # self.args is never stale; it's only updated from this file + return len(self.args) + + def __getitem__(self, size_t i): + # self.args is never stale; it's only updated from this file + return self.args[i] + + +cdef class Metadatum: + + def __cinit__(self, key, value): + if isinstance(key, bytes): + self._key = key + elif isinstance(key, basestring): + self._key = key.encode() + else: + raise TypeError("expected key to be of type str or bytes") + if isinstance(value, bytes): + self._value = value + elif isinstance(value, basestring): + self._value = value.encode() + else: + raise TypeError("expected value to be of type str or bytes") + self.c_metadata.key = self._key + self.c_metadata.value = self._value + self.c_metadata.value_length = len(self._value) + + @property + def key(self): + return <bytes>self.c_metadata.key + + @property + def value(self): + return <bytes>self.c_metadata.value[:self.c_metadata.value_length] + + def __len__(self): + return 2 + + def __getitem__(self, size_t i): + if i == 0: + return self.key + elif i == 1: + return self.value + else: + raise IndexError("index must be 0 (key) or 1 (value)") + + def __iter__(self): + return iter((self.key, self.value)) + + +cdef class _MetadataIterator: + + cdef size_t i + cdef Metadata metadata + + def __cinit__(self, Metadata metadata not None): + self.i = 0 + self.metadata = metadata + + def __next__(self): + if self.i < len(self.metadata): + result = self.metadata[self.i] + self.i = self.i + 1 + return result + else: + raise StopIteration() + + +cdef class Metadata: + + def __cinit__(self, metadata): + self.metadata = list(metadata) + for metadatum in metadata: + if not isinstance(metadatum, Metadatum): + raise TypeError("expected list of Metadatum") + grpc.grpc_metadata_array_init(&self.c_metadata_array) + self.c_metadata_array.count = len(self.metadata) + self.c_metadata_array.capacity = len(self.metadata) + self.c_metadata_array.metadata = <grpc.grpc_metadata *>grpc.gpr_malloc( + self.c_metadata_array.count*sizeof(grpc.grpc_metadata) + ) + for i in range(self.c_metadata_array.count): + self.c_metadata_array.metadata[i] = ( + (<Metadatum>self.metadata[i]).c_metadata) + + def __dealloc__(self): + # this frees the allocated memory for the grpc_metadata_array (although + # it'd be nice if that were documented somewhere...) TODO(atash): document + # this in the C core + grpc.grpc_metadata_array_destroy(&self.c_metadata_array) + + def __len__(self): + return self.c_metadata_array.count + + def __getitem__(self, size_t i): + return Metadatum( + key=<bytes>self.c_metadata_array.metadata[i].key, + value=<bytes>self.c_metadata_array.metadata[i].value[ + :self.c_metadata_array.metadata[i].value_length]) + + def __iter__(self): + return _MetadataIterator(self) + + +cdef class Operation: + + def __cinit__(self): + self.references = [] + self._received_status_details = NULL + self._received_status_details_capacity = 0 + self.is_valid = False + + @property + def type(self): + return self.c_op.type + + @property + def received_message(self): + if self.c_op.type != grpc.GRPC_OP_RECV_MESSAGE: + raise TypeError("self must be an operation receiving a message") + return self._received_message + + @property + def received_metadata(self): + if (self.c_op.type != grpc.GRPC_OP_RECV_INITIAL_METADATA and + self.c_op.type != grpc.GRPC_OP_RECV_STATUS_ON_CLIENT): + raise TypeError("self must be an operation receiving metadata") + return self._received_metadata + + @property + def received_status_code(self): + if self.c_op.type != grpc.GRPC_OP_RECV_STATUS_ON_CLIENT: + raise TypeError("self must be an operation receiving a status code") + return self._received_status_code + + @property + def received_status_details(self): + if self.c_op.type != grpc.GRPC_OP_RECV_STATUS_ON_CLIENT: + raise TypeError("self must be an operation receiving status details") + if self._received_status_details: + return self._received_status_details + else: + return None + + @property + def received_cancelled(self): + if self.c_op.type != grpc.GRPC_OP_RECV_CLOSE_ON_SERVER: + raise TypeError("self must be an operation receiving cancellation " + "information") + return False if self._received_cancelled == 0 else True + + def __dealloc__(self): + # We *almost* don't need to do anything; most of the objects are handled by + # Python. The remaining one(s) are primitive fields filled in by GRPC core. + # This means that we need to clean up after receive_status_on_client. + if self.c_op.type == grpc.GRPC_OP_RECV_STATUS_ON_CLIENT: + grpc.gpr_free(self._received_status_details) + +def operation_send_initial_metadata(Metadata metadata): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_SEND_INITIAL_METADATA + op.c_op.data.send_initial_metadata.count = metadata.c_metadata_array.count + op.c_op.data.send_initial_metadata.metadata = ( + metadata.c_metadata_array.metadata) + op.references.append(metadata) + op.is_valid = True + return op + +def operation_send_message(data): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_SEND_MESSAGE + byte_buffer = ByteBuffer(data) + op.c_op.data.send_message = byte_buffer.c_byte_buffer + op.references.append(byte_buffer) + op.is_valid = True + return op + +def operation_send_close_from_client(): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_SEND_CLOSE_FROM_CLIENT + op.is_valid = True + return op + +def operation_send_status_from_server( + Metadata metadata, grpc.grpc_status_code code, details): + if isinstance(details, bytes): + pass + elif isinstance(details, basestring): + details = details.encode() + else: + raise TypeError("expected a str or bytes object for details") + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_SEND_STATUS_FROM_SERVER + op.c_op.data.send_status_from_server.trailing_metadata_count = ( + metadata.c_metadata_array.count) + op.c_op.data.send_status_from_server.trailing_metadata = ( + metadata.c_metadata_array.metadata) + op.c_op.data.send_status_from_server.status = code + op.c_op.data.send_status_from_server.status_details = details + op.references.append(metadata) + op.references.append(details) + op.is_valid = True + return op + +def operation_receive_initial_metadata(): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_RECV_INITIAL_METADATA + op._received_metadata = Metadata([]) + op.c_op.data.receive_initial_metadata = ( + &op._received_metadata.c_metadata_array) + op.is_valid = True + return op + +def operation_receive_message(): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_RECV_MESSAGE + op._received_message = ByteBuffer(None) + # n.b. the c_op.data.receive_message field needs to be deleted by us, + # anyway, so we just let that be handled by the ByteBuffer() we allocated + # the line before. + op.c_op.data.receive_message = &op._received_message.c_byte_buffer + op.is_valid = True + return op + +def operation_receive_status_on_client(): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_RECV_STATUS_ON_CLIENT + op._received_metadata = Metadata([]) + op.c_op.data.receive_status_on_client.trailing_metadata = ( + &op._received_metadata.c_metadata_array) + op.c_op.data.receive_status_on_client.status = ( + &op._received_status_code) + op.c_op.data.receive_status_on_client.status_details = ( + &op._received_status_details) + op.c_op.data.receive_status_on_client.status_details_capacity = ( + &op._received_status_details_capacity) + op.is_valid = True + return op + +def operation_receive_close_on_server(): + cdef Operation op = Operation() + op.c_op.type = grpc.GRPC_OP_RECV_CLOSE_ON_SERVER + op.c_op.data.receive_close_on_server.cancelled = &op._received_cancelled + op.is_valid = True + return op + + +cdef class _OperationsIterator: + + cdef size_t i + cdef Operations operations + + def __cinit__(self, Operations operations not None): + self.i = 0 + self.operations = operations + + def __next__(self): + if self.i < len(self.operations): + result = self.operations[self.i] + self.i = self.i + 1 + return result + else: + raise StopIteration() + + +cdef class Operations: + + def __cinit__(self, operations): + self.operations = list(operations) # normalize iterable + self.c_ops = NULL + self.c_nops = 0 + for operation in self.operations: + if not isinstance(operation, Operation): + raise TypeError("expected operations to be iterable of Operation") + self.c_nops = len(self.operations) + self.c_ops = <grpc.grpc_op *>grpc.gpr_malloc( + sizeof(grpc.grpc_op)*self.c_nops) + for i in range(self.c_nops): + self.c_ops[i] = (<Operation>(self.operations[i])).c_op + + def __len__(self): + return self.c_nops + + def __getitem__(self, size_t i): + # self.operations is never stale; it's only updated from this file + return self.operations[i] + + def __dealloc__(self): + grpc.gpr_free(self.c_ops) + + def __iter__(self): + return _OperationsIterator(self) + diff --git a/src/python/src/grpc/_cython/_cygrpc/server.pxd b/src/python/src/grpc/_cython/_cygrpc/server.pxd new file mode 100644 index 0000000000..0257542a03 --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/server.pxd @@ -0,0 +1,45 @@ +# 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. + +from grpc._cython._cygrpc cimport grpc +from grpc._cython._cygrpc cimport completion_queue + + +cdef class Server: + + cdef grpc.grpc_server *c_server + cdef bint is_started # start has been called + cdef bint is_shutting_down # shutdown has been called + cdef bint is_shutdown # notification of complete shutdown received + # used at dealloc when user forgets to shutdown + cdef completion_queue.CompletionQueue backup_shutdown_queue + cdef list references + cdef list registered_completion_queues + + cdef notify_shutdown_complete(self) diff --git a/src/python/src/grpc/_cython/_cygrpc/server.pyx b/src/python/src/grpc/_cython/_cygrpc/server.pyx new file mode 100644 index 0000000000..dcf9d38337 --- /dev/null +++ b/src/python/src/grpc/_cython/_cygrpc/server.pyx @@ -0,0 +1,158 @@ +# 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. + +cimport cpython + +from grpc._cython._cygrpc cimport call +from grpc._cython._cygrpc cimport completion_queue +from grpc._cython._cygrpc cimport credentials +from grpc._cython._cygrpc cimport records + +import time + + +cdef class Server: + + def __cinit__(self, records.ChannelArgs arguments=None): + cdef grpc.grpc_channel_args *c_arguments = NULL + self.references = [] + self.registered_completion_queues = [] + if arguments is not None: + c_arguments = &arguments.c_args + self.references.append(arguments) + self.c_server = grpc.grpc_server_create(c_arguments) + self.is_started = False + self.is_shutting_down = False + self.is_shutdown = False + + def request_call( + self, completion_queue.CompletionQueue call_queue not None, + completion_queue.CompletionQueue server_queue not None, tag): + if not self.is_started or self.is_shutting_down: + raise ValueError("server must be started and not shutting down") + if server_queue not in self.registered_completion_queues: + raise ValueError("server_queue must be a registered completion queue") + cdef records.OperationTag operation_tag = records.OperationTag(tag) + operation_tag.operation_call = call.Call() + operation_tag.request_call_details = records.CallDetails() + operation_tag.request_metadata = records.Metadata([]) + operation_tag.references.extend([self, call_queue, server_queue]) + operation_tag.is_new_request = True + operation_tag.batch_operations = records.Operations([]) + cpython.Py_INCREF(operation_tag) + return grpc.grpc_server_request_call( + self.c_server, &operation_tag.operation_call.c_call, + &operation_tag.request_call_details.c_details, + &operation_tag.request_metadata.c_metadata_array, + call_queue.c_completion_queue, server_queue.c_completion_queue, + <cpython.PyObject *>operation_tag) + + def register_completion_queue( + self, completion_queue.CompletionQueue queue not None): + if self.is_started: + raise ValueError("cannot register completion queues after start") + grpc.grpc_server_register_completion_queue( + self.c_server, queue.c_completion_queue) + self.registered_completion_queues.append(queue) + + def start(self): + if self.is_started: + raise ValueError("the server has already started") + self.backup_shutdown_queue = completion_queue.CompletionQueue() + self.register_completion_queue(self.backup_shutdown_queue) + self.is_started = True + grpc.grpc_server_start(self.c_server) + + def add_http2_port(self, address, + credentials.ServerCredentials server_credentials=None): + if isinstance(address, bytes): + pass + elif isinstance(address, basestring): + address = address.encode() + else: + raise TypeError("expected address to be a str or bytes") + self.references.append(address) + if server_credentials is not None: + self.references.append(server_credentials) + return grpc.grpc_server_add_secure_http2_port( + self.c_server, address, server_credentials.c_credentials) + else: + return grpc.grpc_server_add_http2_port(self.c_server, address) + + def shutdown(self, completion_queue.CompletionQueue queue not None, tag): + cdef records.OperationTag operation_tag + if queue.is_shutting_down: + raise ValueError("queue must be live") + elif not self.is_started: + raise ValueError("the server hasn't started yet") + elif self.is_shutting_down: + return + elif queue not in self.registered_completion_queues: + raise ValueError("expected registered completion queue") + else: + self.is_shutting_down = True + operation_tag = records.OperationTag(tag) + operation_tag.shutting_down_server = self + operation_tag.references.extend([self, queue]) + cpython.Py_INCREF(operation_tag) + grpc.grpc_server_shutdown_and_notify( + self.c_server, queue.c_completion_queue, + <cpython.PyObject *>operation_tag) + + cdef notify_shutdown_complete(self): + # called only by a completion queue on receiving our shutdown operation tag + self.is_shutdown = True + + def cancel_all_calls(self): + if not self.is_shutting_down: + raise ValueError("the server must be shutting down to cancel all calls") + elif self.is_shutdown: + return + else: + grpc.grpc_server_cancel_all_calls(self.c_server) + + def __dealloc__(self): + if self.c_server != NULL: + if not self.is_started: + pass + elif self.is_shutdown: + pass + elif not self.is_shutting_down: + # the user didn't call shutdown - use our backup queue + self.shutdown(self.backup_shutdown_queue, None) + # and now we wait + while not self.is_shutdown: + self.backup_shutdown_queue.poll() + else: + # We're in the process of shutting down, but have not shutdown; can't do + # much but repeatedly release the GIL and wait + while not self.is_shutdown: + time.sleep(0) + grpc.grpc_server_destroy(self.c_server) + diff --git a/src/python/src/grpc/_cython/adapter_low.py b/src/python/src/grpc/_cython/adapter_low.py new file mode 100644 index 0000000000..7546dd1599 --- /dev/null +++ b/src/python/src/grpc/_cython/adapter_low.py @@ -0,0 +1,114 @@ +# 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. + + +# Adapter from grpc._cython.types to the surface expected by +# grpc._adapter._intermediary_low. +# +# TODO(atash): Once this is plugged into grpc._adapter._intermediary_low, remove +# both grpc._adapter._intermediary_low and this file. The fore and rear links in +# grpc._adapter should be able to use grpc._cython.types directly. + +from grpc._adapter import _types as type_interfaces +from grpc._cython import cygrpc + + +class ClientCredentials(object): + def __init__(self): + raise NotImplementedError() + + @staticmethod + def google_default(): + raise NotImplementedError() + + @staticmethod + def ssl(): + raise NotImplementedError() + + @staticmethod + def composite(): + raise NotImplementedError() + + @staticmethod + def compute_engine(): + raise NotImplementedError() + + @staticmethod + def service_account(): + raise NotImplementedError() + + @staticmethod + def jwt(): + raise NotImplementedError() + + @staticmethod + def refresh_token(): + raise NotImplementedError() + + @staticmethod + def fake_transport_security(): + raise NotImplementedError() + + @staticmethod + def iam(): + raise NotImplementedError() + + +class ServerCredentials(object): + def __init__(self): + raise NotImplementedError() + + @staticmethod + def ssl(): + raise NotImplementedError() + + @staticmethod + def fake_transport_security(): + raise NotImplementedError() + + +class CompletionQueue(type_interfaces.CompletionQueue): + def __init__(self): + raise NotImplementedError() + + +class Call(type_interfaces.Call): + def __init__(self): + raise NotImplementedError() + + +class Channel(type_interfaces.Channel): + def __init__(self): + raise NotImplementedError() + + +class Server(type_interfaces.Server): + def __init__(self): + raise NotImplementedError() + diff --git a/src/python/src/grpc/_cython/adapter_low_test.py b/src/python/src/grpc/_cython/adapter_low_test.py new file mode 100644 index 0000000000..9bab930e56 --- /dev/null +++ b/src/python/src/grpc/_cython/adapter_low_test.py @@ -0,0 +1,187 @@ +# 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. + +# Fork of grpc._adapter._low_test; the grpc._cython.types adapter in +# grpc._cython.low should transparently support the semantics expected of +# grpc._adapter._low. + +import time +import unittest + +from grpc._adapter import _types +from grpc._cython import adapter_low as _low + + +class InsecureServerInsecureClient(unittest.TestCase): + + def setUp(self): + self.server_completion_queue = _low.CompletionQueue() + self.server = _low.Server(self.server_completion_queue, []) + self.port = self.server.add_http2_port('[::]:0') + self.client_completion_queue = _low.CompletionQueue() + self.client_channel = _low.Channel('localhost:%d'%self.port, []) + + self.server.start() + + def tearDown(self): + self.server.shutdown() + del self.client_channel + + self.client_completion_queue.shutdown() + while (self.client_completion_queue.next().type != + _types.EventType.QUEUE_SHUTDOWN): + pass + self.server_completion_queue.shutdown() + while (self.server_completion_queue.next().type != + _types.EventType.QUEUE_SHUTDOWN): + pass + + del self.client_completion_queue + del self.server_completion_queue + del self.server + + @unittest.skip('TODO(atash): implement grpc._cython.adapter_low') + def testEcho(self): + DEADLINE = time.time()+5 + DEADLINE_TOLERANCE = 0.25 + CLIENT_METADATA_ASCII_KEY = 'key' + CLIENT_METADATA_ASCII_VALUE = 'val' + CLIENT_METADATA_BIN_KEY = 'key-bin' + CLIENT_METADATA_BIN_VALUE = b'\0'*1000 + SERVER_INITIAL_METADATA_KEY = 'init_me_me_me' + SERVER_INITIAL_METADATA_VALUE = 'whodawha?' + SERVER_TRAILING_METADATA_KEY = 'California_is_in_a_drought' + SERVER_TRAILING_METADATA_VALUE = 'zomg it is' + SERVER_STATUS_CODE = _types.StatusCode.OK + SERVER_STATUS_DETAILS = 'our work is never over' + REQUEST = 'in death a member of project mayhem has a name' + RESPONSE = 'his name is robert paulson' + METHOD = 'twinkies' + HOST = 'hostess' + server_request_tag = object() + request_call_result = self.server.request_call(self.server_completion_queue, + server_request_tag) + + self.assertEqual(_types.CallError.OK, request_call_result) + + client_call_tag = object() + client_call = self.client_channel.create_call(self.client_completion_queue, + METHOD, HOST, DEADLINE) + client_initial_metadata = [ + (CLIENT_METADATA_ASCII_KEY, CLIENT_METADATA_ASCII_VALUE), + (CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)] + client_start_batch_result = client_call.start_batch([ + _types.OpArgs.send_initial_metadata(client_initial_metadata), + _types.OpArgs.send_message(REQUEST), + _types.OpArgs.send_close_from_client(), + _types.OpArgs.recv_initial_metadata(), + _types.OpArgs.recv_message(), + _types.OpArgs.recv_status_on_client() + ], client_call_tag) + self.assertEqual(_types.CallError.OK, client_start_batch_result) + + request_event = self.server_completion_queue.next(DEADLINE) + self.assertEqual(_types.EventType.OP_COMPLETE, request_event.type) + self.assertIsInstance(request_event.call, _low.Call) + self.assertIs(server_request_tag, request_event.tag) + self.assertEqual(1, len(request_event.results)) + self.assertEqual(dict(client_initial_metadata), + dict(request_event.results[0].initial_metadata)) + self.assertEqual(METHOD, request_event.call_details.method) + self.assertEqual(HOST, request_event.call_details.host) + self.assertLess(abs(DEADLINE - request_event.call_details.deadline), + DEADLINE_TOLERANCE) + + server_call_tag = object() + server_call = request_event.call + server_initial_metadata = [ + (SERVER_INITIAL_METADATA_KEY, SERVER_INITIAL_METADATA_VALUE)] + server_trailing_metadata = [ + (SERVER_TRAILING_METADATA_KEY, SERVER_TRAILING_METADATA_VALUE)] + server_start_batch_result = server_call.start_batch([ + _types.OpArgs.send_initial_metadata(server_initial_metadata), + _types.OpArgs.recv_message(), + _types.OpArgs.send_message(RESPONSE), + _types.OpArgs.recv_close_on_server(), + _types.OpArgs.send_status_from_server( + server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS) + ], server_call_tag) + self.assertEqual(_types.CallError.OK, server_start_batch_result) + + client_event = self.client_completion_queue.next(DEADLINE) + server_event = self.server_completion_queue.next(DEADLINE) + + self.assertEqual(6, len(client_event.results)) + found_client_op_types = set() + for client_result in client_event.results: + # we expect each op type to be unique + self.assertNotIn(client_result.type, found_client_op_types) + found_client_op_types.add(client_result.type) + if client_result.type == _types.OpType.RECV_INITIAL_METADATA: + self.assertEqual(dict(server_initial_metadata), + dict(client_result.initial_metadata)) + elif client_result.type == _types.OpType.RECV_MESSAGE: + self.assertEqual(RESPONSE, client_result.message) + elif client_result.type == _types.OpType.RECV_STATUS_ON_CLIENT: + self.assertEqual(dict(server_trailing_metadata), + dict(client_result.trailing_metadata)) + self.assertEqual(SERVER_STATUS_DETAILS, client_result.status.details) + self.assertEqual(SERVER_STATUS_CODE, client_result.status.code) + self.assertEqual(set([ + _types.OpType.SEND_INITIAL_METADATA, + _types.OpType.SEND_MESSAGE, + _types.OpType.SEND_CLOSE_FROM_CLIENT, + _types.OpType.RECV_INITIAL_METADATA, + _types.OpType.RECV_MESSAGE, + _types.OpType.RECV_STATUS_ON_CLIENT + ]), found_client_op_types) + + self.assertEqual(5, len(server_event.results)) + found_server_op_types = set() + for server_result in server_event.results: + self.assertNotIn(client_result.type, found_server_op_types) + found_server_op_types.add(server_result.type) + if server_result.type == _types.OpType.RECV_MESSAGE: + self.assertEqual(REQUEST, server_result.message) + elif server_result.type == _types.OpType.RECV_CLOSE_ON_SERVER: + self.assertFalse(server_result.cancelled) + self.assertEqual(set([ + _types.OpType.SEND_INITIAL_METADATA, + _types.OpType.RECV_MESSAGE, + _types.OpType.SEND_MESSAGE, + _types.OpType.RECV_CLOSE_ON_SERVER, + _types.OpType.SEND_STATUS_FROM_SERVER + ]), found_server_op_types) + + del client_call + del server_call + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/src/grpc/_cython/cygrpc.pyx b/src/python/src/grpc/_cython/cygrpc.pyx new file mode 100644 index 0000000000..dcb06f345c --- /dev/null +++ b/src/python/src/grpc/_cython/cygrpc.pyx @@ -0,0 +1,111 @@ +# 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. + +cimport cpython + +from grpc._cython._cygrpc cimport grpc +from grpc._cython._cygrpc cimport call +from grpc._cython._cygrpc cimport channel +from grpc._cython._cygrpc cimport credentials +from grpc._cython._cygrpc cimport completion_queue +from grpc._cython._cygrpc cimport records +from grpc._cython._cygrpc cimport server + +from grpc._cython._cygrpc import call +from grpc._cython._cygrpc import channel +from grpc._cython._cygrpc import credentials +from grpc._cython._cygrpc import completion_queue +from grpc._cython._cygrpc import records +from grpc._cython._cygrpc import server + +StatusCode = records.StatusCode +CallError = records.CallError +CompletionType = records.CompletionType +OperationType = records.OperationType +Timespec = records.Timespec +CallDetails = records.CallDetails +Event = records.Event +ByteBuffer = records.ByteBuffer +SslPemKeyCertPair = records.SslPemKeyCertPair +ChannelArg = records.ChannelArg +ChannelArgs = records.ChannelArgs +Metadatum = records.Metadatum +Metadata = records.Metadata +Operation = records.Operation + +operation_send_initial_metadata = records.operation_send_initial_metadata +operation_send_message = records.operation_send_message +operation_send_close_from_client = records.operation_send_close_from_client +operation_send_status_from_server = records.operation_send_status_from_server +operation_receive_initial_metadata = records.operation_receive_initial_metadata +operation_receive_message = records.operation_receive_message +operation_receive_status_on_client = records.operation_receive_status_on_client +operation_receive_close_on_server = records.operation_receive_close_on_server + +Operations = records.Operations + +ClientCredentials = credentials.ClientCredentials +ServerCredentials = credentials.ServerCredentials + +client_credentials_google_default = ( + credentials.client_credentials_google_default) +client_credentials_ssl = credentials.client_credentials_ssl +client_credentials_composite_credentials = ( + credentials.client_credentials_composite_credentials) +client_credentials_compute_engine = ( + credentials.client_credentials_compute_engine) +client_credentials_jwt = credentials.client_credentials_jwt +client_credentials_refresh_token = credentials.client_credentials_refresh_token +client_credentials_fake_transport_security = ( + credentials.client_credentials_fake_transport_security) +client_credentials_iam = credentials.client_credentials_iam +server_credentials_ssl = credentials.server_credentials_ssl +server_credentials_fake_transport_security = ( + credentials.server_credentials_fake_transport_security) + +CompletionQueue = completion_queue.CompletionQueue +Channel = channel.Channel +Server = server.Server +Call = call.Call + + +# +# Global state +# + +cdef class _ModuleState: + + def __cinit__(self): + grpc.grpc_init() + + def __dealloc__(self): + grpc.grpc_shutdown() + +_module_state = _ModuleState() + diff --git a/src/python/src/grpc/_cython/cygrpc_test.py b/src/python/src/grpc/_cython/cygrpc_test.py new file mode 100644 index 0000000000..838e1e2254 --- /dev/null +++ b/src/python/src/grpc/_cython/cygrpc_test.py @@ -0,0 +1,276 @@ +# 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. + +import time +import unittest + +from grpc._cython import cygrpc +from grpc._cython import test_utilities + + +class TypeSmokeTest(unittest.TestCase): + + def testStringsInUtilitiesUpDown(self): + self.assertEqual(0, cygrpc.StatusCode.ok) + metadatum = cygrpc.Metadatum('a', 'b') + self.assertEqual('a'.encode(), metadatum.key) + self.assertEqual('b'.encode(), metadatum.value) + metadata = cygrpc.Metadata([metadatum]) + self.assertEqual(1, len(metadata)) + self.assertEqual(metadatum.key, metadata[0].key) + + def testMetadataIteration(self): + metadata = cygrpc.Metadata([ + cygrpc.Metadatum('a', 'b'), cygrpc.Metadatum('c', 'd')]) + iterator = iter(metadata) + metadatum = next(iterator) + self.assertIsInstance(metadatum, cygrpc.Metadatum) + self.assertEqual(metadatum.key, 'a'.encode()) + self.assertEqual(metadatum.value, 'b'.encode()) + metadatum = next(iterator) + self.assertIsInstance(metadatum, cygrpc.Metadatum) + self.assertEqual(metadatum.key, 'c'.encode()) + self.assertEqual(metadatum.value, 'd'.encode()) + with self.assertRaises(StopIteration): + next(iterator) + + def testOperationsIteration(self): + operations = cygrpc.Operations([ + cygrpc.operation_send_message('asdf')]) + iterator = iter(operations) + operation = next(iterator) + self.assertIsInstance(operation, cygrpc.Operation) + # `Operation`s are write-only structures; can't directly debug anything out + # of them. Just check that we stop iterating. + with self.assertRaises(StopIteration): + next(iterator) + + def testTimespec(self): + now = time.time() + timespec = cygrpc.Timespec(now) + self.assertAlmostEqual(now, float(timespec), places=8) + + def testClientCredentialsUpDown(self): + credentials = cygrpc.client_credentials_fake_transport_security() + del credentials + + def testServerCredentialsUpDown(self): + credentials = cygrpc.server_credentials_fake_transport_security() + del credentials + + def testCompletionQueueUpDown(self): + completion_queue = cygrpc.CompletionQueue() + del completion_queue + + def testServerUpDown(self): + server = cygrpc.Server(cygrpc.ChannelArgs([])) + del server + + def testChannelUpDown(self): + channel = cygrpc.Channel('[::]:0', cygrpc.ChannelArgs([])) + del channel + + def testSecureChannelUpDown(self): + channel = cygrpc.Channel( + '[::]:0', cygrpc.ChannelArgs([]), + cygrpc.client_credentials_fake_transport_security()) + del channel + + @unittest.skip('TODO(atash): undo skip after #2229 is merged') + def testServerStartNoExplicitShutdown(self): + server = cygrpc.Server() + completion_queue = cygrpc.CompletionQueue() + server.register_completion_queue(completion_queue) + port = server.add_http2_port('[::]:0') + self.assertIsInstance(port, int) + server.start() + del server + + @unittest.skip('TODO(atash): undo skip after #2229 is merged') + def testServerStartShutdown(self): + completion_queue = cygrpc.CompletionQueue() + server = cygrpc.Server() + server.add_http2_port('[::]:0') + server.register_completion_queue(completion_queue) + server.start() + shutdown_tag = object() + server.shutdown(completion_queue, shutdown_tag) + event = completion_queue.poll() + self.assertEqual(cygrpc.CompletionType.operation_complete, event.type) + self.assertIs(shutdown_tag, event.tag) + del server + del completion_queue + + +class InsecureServerInsecureClient(unittest.TestCase): + + def setUp(self): + self.server_completion_queue = cygrpc.CompletionQueue() + self.server = cygrpc.Server() + self.server.register_completion_queue(self.server_completion_queue) + self.port = self.server.add_http2_port('[::]:0') + self.server.start() + self.client_completion_queue = cygrpc.CompletionQueue() + self.client_channel = cygrpc.Channel('localhost:{}'.format(self.port)) + + def tearDown(self): + del self.server + del self.client_completion_queue + del self.server_completion_queue + + def testEcho(self): + DEADLINE = time.time()+5 + DEADLINE_TOLERANCE = 0.25 + CLIENT_METADATA_ASCII_KEY = b'key' + CLIENT_METADATA_ASCII_VALUE = b'val' + CLIENT_METADATA_BIN_KEY = b'key-bin' + CLIENT_METADATA_BIN_VALUE = b'\0'*1000 + SERVER_INITIAL_METADATA_KEY = b'init_me_me_me' + SERVER_INITIAL_METADATA_VALUE = b'whodawha?' + SERVER_TRAILING_METADATA_KEY = b'California_is_in_a_drought' + SERVER_TRAILING_METADATA_VALUE = b'zomg it is' + SERVER_STATUS_CODE = cygrpc.StatusCode.ok + SERVER_STATUS_DETAILS = b'our work is never over' + REQUEST = b'in death a member of project mayhem has a name' + RESPONSE = b'his name is robert paulson' + METHOD = b'twinkies' + HOST = b'hostess' + + cygrpc_deadline = cygrpc.Timespec(DEADLINE) + + server_request_tag = object() + request_call_result = self.server.request_call( + self.server_completion_queue, self.server_completion_queue, + server_request_tag) + + self.assertEqual(cygrpc.CallError.ok, request_call_result) + + client_call_tag = object() + client_call = self.client_channel.create_call(self.client_completion_queue, + METHOD, HOST, cygrpc_deadline) + client_initial_metadata = cygrpc.Metadata([ + cygrpc.Metadatum(CLIENT_METADATA_ASCII_KEY, + CLIENT_METADATA_ASCII_VALUE), + cygrpc.Metadatum(CLIENT_METADATA_BIN_KEY, CLIENT_METADATA_BIN_VALUE)]) + client_start_batch_result = client_call.start_batch(cygrpc.Operations([ + cygrpc.operation_send_initial_metadata(client_initial_metadata), + cygrpc.operation_send_message(REQUEST), + cygrpc.operation_send_close_from_client(), + cygrpc.operation_receive_initial_metadata(), + cygrpc.operation_receive_message(), + cygrpc.operation_receive_status_on_client() + ]), client_call_tag) + self.assertEqual(cygrpc.CallError.ok, client_start_batch_result) + client_event_future = test_utilities.CompletionQueuePollFuture( + self.client_completion_queue, cygrpc_deadline) + + request_event = self.server_completion_queue.poll(cygrpc_deadline) + self.assertEqual(cygrpc.CompletionType.operation_complete, + request_event.type) + self.assertIsInstance(request_event.operation_call, cygrpc.Call) + self.assertIs(server_request_tag, request_event.tag) + self.assertEqual(0, len(request_event.batch_operations)) + self.assertEqual(dict(client_initial_metadata), + dict(request_event.request_metadata)) + self.assertEqual(METHOD, request_event.request_call_details.method) + self.assertEqual(HOST, request_event.request_call_details.host) + self.assertLess( + abs(DEADLINE - float(request_event.request_call_details.deadline)), + DEADLINE_TOLERANCE) + + server_call_tag = object() + server_call = request_event.operation_call + server_initial_metadata = cygrpc.Metadata([ + cygrpc.Metadatum(SERVER_INITIAL_METADATA_KEY, + SERVER_INITIAL_METADATA_VALUE)]) + server_trailing_metadata = cygrpc.Metadata([ + cygrpc.Metadatum(SERVER_TRAILING_METADATA_KEY, + SERVER_TRAILING_METADATA_VALUE)]) + server_start_batch_result = server_call.start_batch([ + cygrpc.operation_send_initial_metadata(server_initial_metadata), + cygrpc.operation_receive_message(), + cygrpc.operation_send_message(RESPONSE), + cygrpc.operation_receive_close_on_server(), + cygrpc.operation_send_status_from_server( + server_trailing_metadata, SERVER_STATUS_CODE, SERVER_STATUS_DETAILS) + ], server_call_tag) + self.assertEqual(cygrpc.CallError.ok, server_start_batch_result) + + client_event = client_event_future.result() + server_event = self.server_completion_queue.poll(cygrpc_deadline) + + self.assertEqual(6, len(client_event.batch_operations)) + found_client_op_types = set() + for client_result in client_event.batch_operations: + # we expect each op type to be unique + self.assertNotIn(client_result.type, found_client_op_types) + found_client_op_types.add(client_result.type) + if client_result.type == cygrpc.OperationType.receive_initial_metadata: + self.assertEqual(dict(server_initial_metadata), + dict(client_result.received_metadata)) + elif client_result.type == cygrpc.OperationType.receive_message: + self.assertEqual(RESPONSE, client_result.received_message.bytes()) + elif client_result.type == cygrpc.OperationType.receive_status_on_client: + self.assertEqual(dict(server_trailing_metadata), + dict(client_result.received_metadata)) + self.assertEqual(SERVER_STATUS_DETAILS, + client_result.received_status_details) + self.assertEqual(SERVER_STATUS_CODE, client_result.received_status_code) + self.assertEqual(set([ + cygrpc.OperationType.send_initial_metadata, + cygrpc.OperationType.send_message, + cygrpc.OperationType.send_close_from_client, + cygrpc.OperationType.receive_initial_metadata, + cygrpc.OperationType.receive_message, + cygrpc.OperationType.receive_status_on_client + ]), found_client_op_types) + + self.assertEqual(5, len(server_event.batch_operations)) + found_server_op_types = set() + for server_result in server_event.batch_operations: + self.assertNotIn(client_result.type, found_server_op_types) + found_server_op_types.add(server_result.type) + if server_result.type == cygrpc.OperationType.receive_message: + self.assertEqual(REQUEST, server_result.received_message.bytes()) + elif server_result.type == cygrpc.OperationType.receive_close_on_server: + self.assertFalse(server_result.received_cancelled) + self.assertEqual(set([ + cygrpc.OperationType.send_initial_metadata, + cygrpc.OperationType.receive_message, + cygrpc.OperationType.send_message, + cygrpc.OperationType.receive_close_on_server, + cygrpc.OperationType.send_status_from_server + ]), found_server_op_types) + + del client_call + del server_call + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/python/src/grpc/_cython/test_utilities.py b/src/python/src/grpc/_cython/test_utilities.py new file mode 100644 index 0000000000..21ea3075b4 --- /dev/null +++ b/src/python/src/grpc/_cython/test_utilities.py @@ -0,0 +1,46 @@ +# 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. + +import threading + +from grpc._cython._cygrpc import completion_queue + + +class CompletionQueuePollFuture: + + def __init__(self, completion_queue, deadline): + def poller_function(): + self._event_result = completion_queue.poll(deadline) + self._event_result = None + self._thread = threading.Thread(target=poller_function) + self._thread.start() + + def result(self): + self._thread.join() + return self._event_result diff --git a/src/python/src/setup.py b/src/python/src/setup.py index 5398b09936..2053e9a5ad 100644 --- a/src/python/src/setup.py +++ b/src/python/src/setup.py @@ -29,11 +29,20 @@ """A setup module for the GRPC Python package.""" +import os +import sys + from distutils import core as _core +from distutils import extension as _extension import setuptools -import sys -_EXTENSION_SOURCES = ( + +# Use environment variables to determine whether or not the Cython extension +# should *use* Cython or use the generated C files. Note that this requires the +# C files to have been generated by building first *with* Cython support. +_BUILD_WITH_CYTHON = os.environ.get('GRPC_PYTHON_BUILD_WITH_CYTHON', False) + +_C_EXTENSION_SOURCES = ( 'grpc/_adapter/_c/module.c', 'grpc/_adapter/_c/types.c', 'grpc/_adapter/_c/utility.c', @@ -45,6 +54,19 @@ _EXTENSION_SOURCES = ( 'grpc/_adapter/_c/types/server.c', ) +_CYTHON_EXTENSION_PACKAGE_NAMES = ( +) + +_CYTHON_EXTENSION_MODULE_NAMES = ( + 'grpc._cython.cygrpc', + 'grpc._cython._cygrpc.call', + 'grpc._cython._cygrpc.channel', + 'grpc._cython._cygrpc.completion_queue', + 'grpc._cython._cygrpc.credentials', + 'grpc._cython._cygrpc.records', + 'grpc._cython._cygrpc.server', +) + _EXTENSION_INCLUDE_DIRECTORIES = ( '.', ) @@ -56,15 +78,50 @@ _EXTENSION_LIBRARIES = ( if not "darwin" in sys.platform: _EXTENSION_LIBRARIES += ('rt',) -_EXTENSION_MODULE = _core.Extension( - 'grpc._adapter._c', sources=list(_EXTENSION_SOURCES), + +_C_EXTENSION_MODULE = _core.Extension( + 'grpc._adapter._c', sources=list(_C_EXTENSION_SOURCES), include_dirs=list(_EXTENSION_INCLUDE_DIRECTORIES), libraries=list(_EXTENSION_LIBRARIES), - ) +) +_C_EXTENSION_MODULES = [_C_EXTENSION_MODULE] + + +def cython_extensions(package_names, module_names, include_dirs, libraries, + build_with_cython=False): + file_extension = 'pyx' if build_with_cython else 'c' + module_files = [name.replace('.', '/') + '.' + file_extension + for name in module_names] + extensions = [ + _extension.Extension( + name=module_name, sources=[module_file], + include_dirs=include_dirs, libraries=libraries + ) for (module_name, module_file) in zip(module_names, module_files) + ] + if build_with_cython: + import Cython.Build + return Cython.Build.cythonize(extensions) + else: + return extensions + +_CYTHON_EXTENSION_MODULES = cython_extensions( + list(_CYTHON_EXTENSION_PACKAGE_NAMES), list(_CYTHON_EXTENSION_MODULE_NAMES), + list(_EXTENSION_INCLUDE_DIRECTORIES), list(_EXTENSION_LIBRARIES), + bool(_BUILD_WITH_CYTHON)) + +# TODO(atash): We shouldn't need to gate any C code based on the python version +# from the distutils build system. Remove this hackery once we're on Cython and +# 3.x C API compliant. +_EXTENSION_MODULES = list(_CYTHON_EXTENSION_MODULES) +if sys.version_info[0:2] <= (2, 7): + _EXTENSION_MODULES += _C_EXTENSION_MODULES + _PACKAGES = ( 'grpc', 'grpc._adapter', + 'grpc._cython', + 'grpc._cython._cygrpc', 'grpc._junkdrawer', 'grpc.early_adopter', 'grpc.framework', @@ -79,6 +136,7 @@ _PACKAGES = ( _PACKAGE_DIRECTORIES = { 'grpc': 'grpc', 'grpc._adapter': 'grpc/_adapter', + 'grpc._cython': 'grpc/_cython', 'grpc._junkdrawer': 'grpc/_junkdrawer', 'grpc.early_adopter': 'grpc/early_adopter', 'grpc.framework': 'grpc/framework', @@ -86,8 +144,8 @@ _PACKAGE_DIRECTORIES = { setuptools.setup( name='grpcio', - version='0.9.0a1', - ext_modules=[_EXTENSION_MODULE], + version='0.10.0a0', + ext_modules=_EXTENSION_MODULES, packages=list(_PACKAGES), package_dir=_PACKAGE_DIRECTORIES, install_requires=[ diff --git a/src/ruby/ext/grpc/rb_completion_queue.c b/src/ruby/ext/grpc/rb_completion_queue.c index 8fb3949b3d..2d52d96dc8 100644 --- a/src/ruby/ext/grpc/rb_completion_queue.c +++ b/src/ruby/ext/grpc/rb_completion_queue.c @@ -91,7 +91,7 @@ static void grpc_rb_completion_queue_shutdown_drain(grpc_completion_queue *cq) { * - investigate further, this is probably another example of C-level cleanup * not working consistently in all cases. */ - next_call.timeout = gpr_time_add(gpr_now(), gpr_time_from_micros(5e3)); + next_call.timeout = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_micros(5e3)); do { rb_thread_call_without_gvl(grpc_rb_completion_queue_next_no_gil, (void *)&next_call, NULL, NULL); diff --git a/src/ruby/lib/grpc/version.rb b/src/ruby/lib/grpc/version.rb index 028fea5a4c..431e8774b5 100644 --- a/src/ruby/lib/grpc/version.rb +++ b/src/ruby/lib/grpc/version.rb @@ -29,5 +29,5 @@ # GRPC contains the General RPC module. module GRPC - VERSION = '0.9.4' + VERSION = '0.10.0' end |