diff options
author | 2015-07-07 13:43:35 -0700 | |
---|---|---|
committer | 2015-07-07 13:43:35 -0700 | |
commit | c5b5290bc6570a7226ed67d6884c374b236b48dc (patch) | |
tree | 0b46ad23f4bb4dd4a94cd1248bea843af18c3f76 /src/core | |
parent | 092d8d1b7f4acd55a7bb89b10deb8e4c33ad6930 (diff) | |
parent | 772187cdf0ff9dfafd2e693474c51eeddfe4c800 (diff) |
Merge github.com:grpc/grpc into not-everyone-can-be-included
Diffstat (limited to 'src/core')
-rw-r--r-- | src/core/client_config/README.md | 16 | ||||
-rw-r--r-- | src/core/client_config/lb_policies/pick_first.c | 4 | ||||
-rw-r--r-- | src/core/iomgr/alarm.h | 2 | ||||
-rw-r--r-- | src/core/iomgr/fd_posix.c | 5 | ||||
-rw-r--r-- | src/core/iomgr/pollset_multipoller_with_epoll.c | 26 | ||||
-rw-r--r-- | src/core/iomgr/pollset_posix.c | 3 | ||||
-rw-r--r-- | src/core/security/server_secure_chttp2.c | 5 | ||||
-rw-r--r-- | src/core/support/log_linux.c | 11 | ||||
-rw-r--r-- | src/core/surface/byte_buffer.c | 14 | ||||
-rw-r--r-- | src/core/surface/call.c | 8 | ||||
-rw-r--r-- | src/core/surface/call.h | 2 | ||||
-rw-r--r-- | src/core/surface/channel_create.c | 3 | ||||
-rw-r--r-- | src/core/surface/secure_channel_create.c | 6 | ||||
-rw-r--r-- | src/core/surface/server_chttp2.c | 3 | ||||
-rw-r--r-- | src/core/transport/chttp2/incoming_metadata.c | 1 | ||||
-rw-r--r-- | src/core/transport/chttp2/parsing.c | 3 | ||||
-rw-r--r-- | src/core/transport/chttp2/writing.c | 13 | ||||
-rw-r--r-- | src/core/transport/chttp2_transport.c | 23 | ||||
-rw-r--r-- | src/core/transport/chttp2_transport.h | 7 | ||||
-rw-r--r-- | src/core/transport/stream_op.h | 2 |
20 files changed, 105 insertions, 52 deletions
diff --git a/src/core/client_config/README.md b/src/core/client_config/README.md index 7cb19cd130..d7aed27223 100644 --- a/src/core/client_config/README.md +++ b/src/core/client_config/README.md @@ -42,3 +42,19 @@ Their behavior is specified by a set of grpc channel filters defined at their construction. To customize this behavior, resolvers build grpc_subchannel_factory objects, which use the decorator pattern to customize construction arguments for concrete grpc_subchannel instances. + + +Naming for GRPC +=============== + +Names in GRPC are represented by a URI. + +The following schemes are currently supported: + +dns:///host:port - dns schemes are currently supported so long as authority is + empty (authority based dns resolution is expected in a future + release) + +unix:path - the unix scheme is used to create and connect to unix domain + sockets - the authority must be empty, and the path represents + the absolute or relative path to the desired socket diff --git a/src/core/client_config/lb_policies/pick_first.c b/src/core/client_config/lb_policies/pick_first.c index 3d57e3136a..73da624aff 100644 --- a/src/core/client_config/lb_policies/pick_first.c +++ b/src/core/client_config/lb_policies/pick_first.c @@ -155,8 +155,6 @@ loop: switch (p->checking_connectivity) { case GRPC_CHANNEL_READY: p->selected = p->subchannels[p->checking_subchannel]; - GPR_ASSERT(grpc_subchannel_check_connectivity(p->selected) == - GRPC_CHANNEL_READY); while ((pp = p->pending_picks)) { p->pending_picks = pp->next; *pp->target = p->selected; @@ -185,6 +183,7 @@ loop: GPR_SWAP(grpc_subchannel *, p->subchannels[p->checking_subchannel], p->subchannels[p->num_subchannels - 1]); p->num_subchannels--; + GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first"); if (p->num_subchannels == 0) { while ((pp = p->pending_picks)) { p->pending_picks = pp->next; @@ -197,7 +196,6 @@ loop: p->checking_subchannel %= p->num_subchannels; p->checking_connectivity = grpc_subchannel_check_connectivity( p->subchannels[p->checking_subchannel]); - GRPC_SUBCHANNEL_UNREF(p->subchannels[p->num_subchannels], "pick_first"); add_interested_parties_locked(p); goto loop; } diff --git a/src/core/iomgr/alarm.h b/src/core/iomgr/alarm.h index e5262e2199..c067a0b8a3 100644 --- a/src/core/iomgr/alarm.h +++ b/src/core/iomgr/alarm.h @@ -41,9 +41,9 @@ typedef struct grpc_alarm { gpr_timespec deadline; gpr_uint32 heap_index; /* INVALID_HEAP_INDEX if not in heap */ + int triggered; struct grpc_alarm *next; struct grpc_alarm *prev; - int triggered; grpc_iomgr_cb_func cb; void *cb_arg; } grpc_alarm; diff --git a/src/core/iomgr/fd_posix.c b/src/core/iomgr/fd_posix.c index e8c24c772a..6ad377ce1c 100644 --- a/src/core/iomgr/fd_posix.c +++ b/src/core/iomgr/fd_posix.c @@ -369,16 +369,17 @@ gpr_uint32 grpc_fd_begin_poll(grpc_fd *fd, grpc_pollset *pollset, watcher->fd = NULL; watcher->pollset = NULL; gpr_mu_unlock(&fd->watcher_mu); + GRPC_FD_UNREF(fd, "poll"); return 0; } /* if there is nobody polling for read, but we need to, then start doing so */ - if (!fd->read_watcher && gpr_atm_acq_load(&fd->readst) > READY) { + if (read_mask && !fd->read_watcher && gpr_atm_acq_load(&fd->readst) > READY) { fd->read_watcher = watcher; mask |= read_mask; } /* if there is nobody polling for write, but we need to, then start doing so */ - if (!fd->write_watcher && gpr_atm_acq_load(&fd->writest) > READY) { + if (write_mask && !fd->write_watcher && gpr_atm_acq_load(&fd->writest) > READY) { fd->write_watcher = watcher; mask |= write_mask; } diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c index dcf08d379c..1900bbf9e1 100644 --- a/src/core/iomgr/pollset_multipoller_with_epoll.c +++ b/src/core/iomgr/pollset_multipoller_with_epoll.c @@ -54,17 +54,25 @@ static void multipoll_with_epoll_pollset_add_fd(grpc_pollset *pollset, pollset_hdr *h = pollset->data.ptr; struct epoll_event ev; int err; - - ev.events = EPOLLIN | EPOLLOUT | EPOLLET; - ev.data.ptr = fd; - err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD, fd->fd, &ev); - if (err < 0) { - /* FDs may be added to a pollset multiple times, so EEXIST is normal. */ - if (errno != EEXIST) { - gpr_log(GPR_ERROR, "epoll_ctl add for %d failed: %s", fd->fd, - strerror(errno)); + grpc_fd_watcher watcher; + + /* We pretend to be polling whilst adding an fd to keep the fd from being + closed during the add. This may result in a spurious wakeup being assigned + to this pollset whilst adding, but that should be benign. */ + GPR_ASSERT(grpc_fd_begin_poll(fd, pollset, 0, 0, &watcher) == 0); + if (watcher.fd != NULL) { + ev.events = EPOLLIN | EPOLLOUT | EPOLLET; + ev.data.ptr = fd; + err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD, fd->fd, &ev); + if (err < 0) { + /* FDs may be added to a pollset multiple times, so EEXIST is normal. */ + if (errno != EEXIST) { + gpr_log(GPR_ERROR, "epoll_ctl add for %d failed: %s", fd->fd, + strerror(errno)); + } } } + grpc_fd_end_poll(&watcher, 0, 0); } static void multipoll_with_epoll_pollset_del_fd(grpc_pollset *pollset, diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index 15ed8e75e6..12496440de 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -249,7 +249,8 @@ static void basic_do_promote(void *args, int success) { pollset->in_flight_cbs--; if (pollset->shutting_down) { /* We don't care about this pollset anymore. */ - if (pollset->in_flight_cbs == 0 && pollset->counter == 0) { + if (pollset->in_flight_cbs == 0 && pollset->counter == 0 && !pollset->called_shutdown) { + pollset->called_shutdown = 1; do_shutdown_cb = 1; } } else if (grpc_fd_is_orphaned(fd)) { diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c index 6a99324da6..8a7ada07af 100644 --- a/src/core/security/server_secure_chttp2.c +++ b/src/core/security/server_secure_chttp2.c @@ -99,9 +99,10 @@ static void on_secure_transport_setup_done(void *statep, if (!state->is_shutdown) { mdctx = grpc_mdctx_create(); transport = grpc_create_chttp2_transport( - grpc_server_get_channel_args(state->server), secure_endpoint, NULL, 0, - mdctx, 0); + grpc_server_get_channel_args(state->server), secure_endpoint, mdctx, + 0); setup_transport(state, transport, mdctx); + grpc_chttp2_transport_start_reading(transport, NULL, 0); } else { /* We need to consume this here, because the server may already have gone * away. */ diff --git a/src/core/support/log_linux.c b/src/core/support/log_linux.c index 48349d2c83..7937466b79 100644 --- a/src/core/support/log_linux.c +++ b/src/core/support/log_linux.c @@ -43,7 +43,9 @@ #ifdef GPR_LINUX +#include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include <grpc/support/string_util.h> #include <grpc/support/time.h> #include <stdio.h> #include <stdarg.h> @@ -71,6 +73,7 @@ void gpr_log(const char *file, int line, gpr_log_severity severity, void gpr_default_log(gpr_log_func_args *args) { char *final_slash; + char *prefix; const char *display_file; char time_buffer[64]; gpr_timespec now = gpr_now(); @@ -89,10 +92,12 @@ void gpr_default_log(gpr_log_func_args *args) { strcpy(time_buffer, "error:strftime"); } - fprintf(stderr, "%s%s.%09d %7ld %s:%d] %s\n", + gpr_asprintf(&prefix, "%s%s.%09d %7tu %s:%d]", gpr_log_severity_string(args->severity), time_buffer, - (int)(now.tv_nsec), gettid(), display_file, args->line, - args->message); + (int)(now.tv_nsec), gettid(), display_file, args->line); + + fprintf(stderr, "%-60s %s\n", prefix, args->message); + gpr_free(prefix); } #endif diff --git a/src/core/surface/byte_buffer.c b/src/core/surface/byte_buffer.c index 4817e00454..a930949f2d 100644 --- a/src/core/surface/byte_buffer.c +++ b/src/core/surface/byte_buffer.c @@ -55,6 +55,20 @@ grpc_byte_buffer *grpc_raw_compressed_byte_buffer_create( return bb; } +grpc_byte_buffer *grpc_raw_byte_buffer_from_reader( + grpc_byte_buffer_reader *reader) { + grpc_byte_buffer *bb = malloc(sizeof(grpc_byte_buffer)); + gpr_slice slice; + bb->type = GRPC_BB_RAW; + bb->data.raw.compression = GRPC_COMPRESS_NONE; + gpr_slice_buffer_init(&bb->data.raw.slice_buffer); + + while (grpc_byte_buffer_reader_next(reader, &slice)) { + gpr_slice_buffer_add(&bb->data.raw.slice_buffer, slice); + } + return bb; +} + grpc_byte_buffer *grpc_byte_buffer_copy(grpc_byte_buffer *bb) { switch (bb->type) { case GRPC_BB_RAW: diff --git a/src/core/surface/call.c b/src/core/surface/call.c index ae1b215767..fc09137b67 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -76,14 +76,14 @@ typedef struct { typedef struct { /* Overall status of the operation: starts OK, may degrade to non-OK */ - int success; - /* Completion function to call at the end of the operation */ - grpc_ioreq_completion_func on_complete; - void *user_data; + gpr_uint8 success; /* a bit mask of which request ops are needed (1u << opid) */ gpr_uint16 need_mask; /* a bit mask of which request ops are now completed */ gpr_uint16 complete_mask; + /* Completion function to call at the end of the operation */ + grpc_ioreq_completion_func on_complete; + void *user_data; } reqinfo_master; /* Status data for a request can come from several sources; this diff --git a/src/core/surface/call.h b/src/core/surface/call.h index fb3662b50d..3b6f9c942e 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -78,8 +78,8 @@ typedef union { typedef struct { grpc_ioreq_op op; - grpc_ioreq_data data; gpr_uint32 flags; /**< A copy of the write flags from grpc_op */ + grpc_ioreq_data data; } grpc_ioreq; typedef void (*grpc_ioreq_completion_func)(grpc_call *call, int success, diff --git a/src/core/surface/channel_create.c b/src/core/surface/channel_create.c index 09b4fb782b..e205f0a9f8 100644 --- a/src/core/surface/channel_create.c +++ b/src/core/surface/channel_create.c @@ -72,7 +72,8 @@ static void connected(void *arg, grpc_endpoint *tcp) { grpc_iomgr_closure *notify; if (tcp != NULL) { c->result->transport = grpc_create_chttp2_transport( - c->args.channel_args, tcp, NULL, 0, c->args.metadata_context, 1); + c->args.channel_args, tcp, c->args.metadata_context, 1); + grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0); GPR_ASSERT(c->result->transport); c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *)); c->result->filters[0] = &grpc_http_client_filter; diff --git a/src/core/surface/secure_channel_create.c b/src/core/surface/secure_channel_create.c index 76fc862621..34ee3f8400 100644 --- a/src/core/surface/secure_channel_create.c +++ b/src/core/surface/secure_channel_create.c @@ -82,9 +82,9 @@ static void on_secure_transport_setup_done(void *arg, gpr_log(GPR_ERROR, "Secure transport setup failed with error %d.", status); memset(c->result, 0, sizeof(*c->result)); } else { - c->result->transport = - grpc_create_chttp2_transport(c->args.channel_args, secure_endpoint, - NULL, 0, c->args.metadata_context, 1); + c->result->transport = grpc_create_chttp2_transport( + c->args.channel_args, secure_endpoint, c->args.metadata_context, 1); + grpc_chttp2_transport_start_reading(c->result->transport, NULL, 0); c->result->filters = gpr_malloc(sizeof(grpc_channel_filter *) * 2); c->result->filters[0] = &grpc_client_auth_filter; c->result->filters[1] = &grpc_http_client_filter; diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c index 9c02c3ef29..78c53466b3 100644 --- a/src/core/surface/server_chttp2.c +++ b/src/core/surface/server_chttp2.c @@ -61,8 +61,9 @@ static void new_transport(void *server, grpc_endpoint *tcp) { */ grpc_mdctx *mdctx = grpc_mdctx_create(); grpc_transport *transport = grpc_create_chttp2_transport( - grpc_server_get_channel_args(server), tcp, NULL, 0, mdctx, 0); + grpc_server_get_channel_args(server), tcp, mdctx, 0); setup_transport(server, transport, mdctx); + grpc_chttp2_transport_start_reading(transport, NULL, 0); } /* Server callback: start listening on our ports */ diff --git a/src/core/transport/chttp2/incoming_metadata.c b/src/core/transport/chttp2/incoming_metadata.c index a4b7174329..68e0912b9c 100644 --- a/src/core/transport/chttp2/incoming_metadata.c +++ b/src/core/transport/chttp2/incoming_metadata.c @@ -124,6 +124,7 @@ void grpc_incoming_metadata_buffer_move_to_referencing_sopb( sopb->ops[i].data.metadata.list.tail = (void *)(delta + (gpr_intptr)sopb->ops[i].data.metadata.list.tail); } + src->count = 0; } void grpc_chttp2_incoming_metadata_buffer_postprocess_sopb_and_begin_live_op( diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index 8f682e9017..4664a0895c 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -109,9 +109,6 @@ void grpc_chttp2_publish_reads( transport_parsing->incoming_stream_id; } - /* TODO(ctiller): re-implement */ - GPR_ASSERT(transport_parsing->initial_window_update == 0); - /* copy parsing qbuf to global qbuf */ gpr_slice_buffer_move_into(&transport_parsing->qbuf, &transport_global->qbuf); diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index fdcc300099..a78654334e 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -97,12 +97,8 @@ int grpc_chttp2_unlocking_check_writes( grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); } - /* we should either exhaust window or have no ops left, but not both */ - if (stream_global->outgoing_sopb->nops == 0) { - stream_global->outgoing_sopb = NULL; - grpc_chttp2_schedule_closure(transport_global, - stream_global->send_done_closure, 1); - } else if (stream_global->outgoing_window > 0) { + if (stream_global->outgoing_window > 0 && + stream_global->outgoing_sopb->nops != 0) { grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } @@ -201,6 +197,11 @@ void grpc_chttp2_cleanup_writing( while (grpc_chttp2_list_pop_written_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { + if (stream_global->outgoing_sopb->nops == 0) { + stream_global->outgoing_sopb = NULL; + grpc_chttp2_schedule_closure(transport_global, + stream_global->send_done_closure, 1); + } if (stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE; if (!transport_global->is_client) { diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 0a7b8f5bf9..3483512ab8 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -201,8 +201,8 @@ static void ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); } static void init_transport(grpc_chttp2_transport *t, const grpc_channel_args *channel_args, - grpc_endpoint *ep, gpr_slice *slices, size_t nslices, - grpc_mdctx *mdctx, int is_client) { + grpc_endpoint *ep, grpc_mdctx *mdctx, + int is_client) { size_t i; int j; @@ -311,9 +311,6 @@ static void init_transport(grpc_chttp2_transport *t, } } } - - REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */ - recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK); } static void destroy_transport(grpc_transport *gt) { @@ -687,7 +684,7 @@ static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) { grpc_chttp2_goaway_append( t->global.last_incoming_stream_id, grpc_chttp2_grpc_status_to_http2_error(op->goaway_status), - *op->goaway_message, &t->global.qbuf); + gpr_slice_ref(*op->goaway_message), &t->global.qbuf); } if (op->set_accept_stream != NULL) { @@ -933,6 +930,7 @@ static void recv_data(void *tp, gpr_slice *slices, size_t nslices, if (t->parsing.initial_window_update != 0) { grpc_chttp2_stream_map_for_each(&t->parsing_stream_map, update_global_window, t); + t->parsing.initial_window_update = 0; } /* handle higher level things */ grpc_chttp2_publish_reads(&t->global, &t->parsing); @@ -1042,9 +1040,16 @@ static const grpc_transport_vtable vtable = { perform_transport_op, destroy_stream, destroy_transport}; grpc_transport *grpc_create_chttp2_transport( - const grpc_channel_args *channel_args, grpc_endpoint *ep, gpr_slice *slices, - size_t nslices, grpc_mdctx *mdctx, int is_client) { + const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx, + int is_client) { grpc_chttp2_transport *t = gpr_malloc(sizeof(grpc_chttp2_transport)); - init_transport(t, channel_args, ep, slices, nslices, mdctx, is_client); + init_transport(t, channel_args, ep, mdctx, is_client); return &t->base; } + +void grpc_chttp2_transport_start_reading(grpc_transport *transport, + gpr_slice *slices, size_t nslices) { + grpc_chttp2_transport *t = (grpc_chttp2_transport *)transport; + REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */ + recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK); +} diff --git a/src/core/transport/chttp2_transport.h b/src/core/transport/chttp2_transport.h index 1747792b95..fa0d6e4151 100644 --- a/src/core/transport/chttp2_transport.h +++ b/src/core/transport/chttp2_transport.h @@ -41,7 +41,10 @@ extern int grpc_http_trace; extern int grpc_flowctl_trace; grpc_transport *grpc_create_chttp2_transport( - const grpc_channel_args *channel_args, grpc_endpoint *ep, gpr_slice *slices, - size_t nslices, grpc_mdctx *metadata_context, int is_client); + const grpc_channel_args *channel_args, grpc_endpoint *ep, + grpc_mdctx *metadata_context, int is_client); + +void grpc_chttp2_transport_start_reading(grpc_transport *transport, + gpr_slice *slices, size_t nslices); #endif /* GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_TRANSPORT_H */ diff --git a/src/core/transport/stream_op.h b/src/core/transport/stream_op.h index 842fc932b9..964d39d14f 100644 --- a/src/core/transport/stream_op.h +++ b/src/core/transport/stream_op.h @@ -41,7 +41,7 @@ #include "src/core/transport/metadata.h" /* this many stream ops are inlined into a sopb before allocating */ -#define GRPC_SOPB_INLINE_ELEMENTS 16 +#define GRPC_SOPB_INLINE_ELEMENTS 4 /* Operations that can be performed on a stream. Used by grpc_stream_op. */ |