stream_op cleanup: surface layer

reviewable/pr3993/r1
Craig Tiller 9 years ago
parent 9d35a1f9ce
commit c7e1a2a38d
  1. 97
      src/core/surface/byte_buffer_queue.c
  2. 62
      src/core/surface/byte_buffer_queue.h
  3. 1884
      src/core/surface/call.c
  4. 67
      src/core/surface/call.h
  5. 3
      src/core/surface/call_log_batch.c
  6. 1
      src/core/surface/call_test_only.h
  7. 82
      src/core/surface/completion_queue.c
  8. 3
      src/core/surface/completion_queue.h
  9. 5
      src/core/surface/init.c
  10. 74
      src/core/surface/lame_client.c
  11. 230
      src/core/surface/server.c

@ -1,97 +0,0 @@
/*
*
* 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/core/surface/byte_buffer_queue.h"
#include <grpc/support/alloc.h>
#include <grpc/support/useful.h>
static void bba_destroy(grpc_bbq_array *array, size_t start_pos) {
size_t i;
for (i = start_pos; i < array->count; i++) {
grpc_byte_buffer_destroy(array->data[i]);
}
gpr_free(array->data);
}
/* Append an operation to an array, expanding as needed */
static void bba_push(grpc_bbq_array *a, grpc_byte_buffer *buffer) {
if (a->count == a->capacity) {
a->capacity = GPR_MAX(a->capacity * 2, 8);
a->data = gpr_realloc(a->data, sizeof(grpc_byte_buffer *) * a->capacity);
}
a->data[a->count++] = buffer;
}
void grpc_bbq_destroy(grpc_byte_buffer_queue *q) {
bba_destroy(&q->filling, 0);
bba_destroy(&q->draining, q->drain_pos);
}
int grpc_bbq_empty(grpc_byte_buffer_queue *q) {
return (q->drain_pos == q->draining.count && q->filling.count == 0);
}
void grpc_bbq_push(grpc_byte_buffer_queue *q, grpc_byte_buffer *buffer) {
q->bytes += grpc_byte_buffer_length(buffer);
bba_push(&q->filling, buffer);
}
void grpc_bbq_flush(grpc_byte_buffer_queue *q) {
grpc_byte_buffer *bb;
while ((bb = grpc_bbq_pop(q))) {
grpc_byte_buffer_destroy(bb);
}
}
size_t grpc_bbq_bytes(grpc_byte_buffer_queue *q) { return q->bytes; }
grpc_byte_buffer *grpc_bbq_pop(grpc_byte_buffer_queue *q) {
grpc_bbq_array temp_array;
grpc_byte_buffer *out;
if (q->drain_pos == q->draining.count) {
if (q->filling.count == 0) {
return NULL;
}
q->draining.count = 0;
q->drain_pos = 0;
/* swap arrays */
temp_array = q->filling;
q->filling = q->draining;
q->draining = temp_array;
}
out = q->draining.data[q->drain_pos++];
q->bytes -= grpc_byte_buffer_length(out);
return out;
}

@ -1,62 +0,0 @@
/*
*
* 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_CORE_SURFACE_BYTE_BUFFER_QUEUE_H
#define GRPC_INTERNAL_CORE_SURFACE_BYTE_BUFFER_QUEUE_H
#include <grpc/byte_buffer.h>
/* TODO(ctiller): inline an element or two into this struct to avoid per-call
allocations */
typedef struct {
grpc_byte_buffer **data;
size_t count;
size_t capacity;
} grpc_bbq_array;
/* should be initialized by zeroing memory */
typedef struct {
size_t drain_pos;
grpc_bbq_array filling;
grpc_bbq_array draining;
size_t bytes;
} grpc_byte_buffer_queue;
void grpc_bbq_destroy(grpc_byte_buffer_queue *q);
grpc_byte_buffer *grpc_bbq_pop(grpc_byte_buffer_queue *q);
void grpc_bbq_flush(grpc_byte_buffer_queue *q);
int grpc_bbq_empty(grpc_byte_buffer_queue *q);
void grpc_bbq_push(grpc_byte_buffer_queue *q, grpc_byte_buffer *bb);
size_t grpc_bbq_bytes(grpc_byte_buffer_queue *q);
#endif /* GRPC_INTERNAL_CORE_SURFACE_BYTE_BUFFER_QUEUE_H */

File diff suppressed because it is too large Load Diff

@ -44,51 +44,6 @@
extern "C" {
#endif
/* Primitive operation types - grpc_op's get rewritten into these */
typedef enum {
GRPC_IOREQ_RECV_INITIAL_METADATA,
GRPC_IOREQ_RECV_MESSAGE,
GRPC_IOREQ_RECV_TRAILING_METADATA,
GRPC_IOREQ_RECV_STATUS,
GRPC_IOREQ_RECV_STATUS_DETAILS,
GRPC_IOREQ_RECV_CLOSE,
GRPC_IOREQ_SEND_INITIAL_METADATA,
GRPC_IOREQ_SEND_MESSAGE,
GRPC_IOREQ_SEND_TRAILING_METADATA,
GRPC_IOREQ_SEND_STATUS,
GRPC_IOREQ_SEND_CLOSE,
GRPC_IOREQ_OP_COUNT
} grpc_ioreq_op;
typedef union {
grpc_metadata_array *recv_metadata;
grpc_byte_buffer **recv_message;
struct {
void (*set_value)(grpc_status_code status, void *user_data);
void *user_data;
} recv_status;
struct {
char **details;
size_t *details_capacity;
} recv_status_details;
struct {
size_t count;
grpc_metadata *metadata;
} send_metadata;
grpc_byte_buffer *send_message;
struct {
grpc_status_code code;
grpc_mdstr *details;
} send_status;
} grpc_ioreq_data;
typedef struct {
grpc_ioreq_op op;
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_exec_ctx *exec_ctx,
grpc_call *call, int success,
void *user_data);
@ -105,7 +60,7 @@ void grpc_call_set_completion_queue(grpc_exec_ctx *exec_ctx, grpc_call *call,
grpc_completion_queue *cq);
grpc_completion_queue *grpc_call_get_completion_queue(grpc_call *call);
#ifdef GRPC_CALL_REF_COUNT_DEBUG
#ifdef GRPC_STREAM_REFCOUNT_DEBUG
void grpc_call_internal_ref(grpc_call *call, const char *reason);
void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *call,
const char *reason);
@ -121,12 +76,14 @@ void grpc_call_internal_unref(grpc_exec_ctx *exec_ctx, grpc_call *call);
grpc_call_internal_unref(exec_ctx, call)
#endif
grpc_call_error grpc_call_start_ioreq_and_call_back(
grpc_exec_ctx *exec_ctx, grpc_call *call, const grpc_ioreq *reqs,
size_t nreqs, grpc_ioreq_completion_func on_complete, void *user_data);
grpc_call_stack *grpc_call_get_call_stack(grpc_call *call);
grpc_call_error grpc_call_start_batch_and_execute(grpc_exec_ctx *exec_ctx,
grpc_call *call,
const grpc_op *ops,
size_t nops,
grpc_closure *closure);
/* Given the top call_element, get the call object. */
grpc_call *grpc_call_from_top_element(grpc_call_element *surface_element);
@ -157,16 +114,6 @@ void *grpc_call_context_get(grpc_call *call, grpc_context_index elem);
#define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \
if (grpc_api_trace) grpc_call_log_batch(sev, call, ops, nops, tag)
#define GRPC_SERVER_LOG_REQUEST_CALL(sev, server, call, details, \
initial_metadata, cq_bound_to_call, \
cq_for_notifications, tag) \
if (grpc_api_trace) \
grpc_server_log_request_call(sev, server, call, details, initial_metadata, \
cq_bound_to_call, cq_for_notifications, tag)
#define GRPC_SERVER_LOG_SHUTDOWN(sev, server, cq, tag) \
if (grpc_api_trace) grpc_server_log_shutdown(sev, server, cq, tag)
gpr_uint8 grpc_call_is_client(grpc_call *call);
#ifdef __cplusplus

@ -110,9 +110,6 @@ void grpc_call_log_batch(char *file, int line, gpr_log_severity severity,
void *tag) {
char *tmp;
size_t i;
gpr_log(file, line, severity,
"grpc_call_start_batch(call=%p, ops=%p, nops=%d, tag=%p)", call, ops,
nops, tag);
for (i = 0; i < nops; i++) {
tmp = grpc_op_string(&ops[i]);
gpr_log(file, line, severity, "ops[%d]: %s", i, tmp);

@ -57,7 +57,6 @@ gpr_uint32 grpc_call_test_only_get_message_flags(grpc_call *call);
* To be indexed by grpc_compression_algorithm enum values. */
gpr_uint32 grpc_call_test_only_get_encodings_accepted_by_peer(grpc_call *call);
#ifdef __cplusplus
}
#endif

@ -71,9 +71,29 @@ struct grpc_completion_queue {
int is_server_cq;
int num_pluckers;
plucker pluckers[GRPC_MAX_COMPLETION_QUEUE_PLUCKERS];
grpc_closure pollset_destroy_done;
grpc_closure pollset_shutdown_done;
grpc_completion_queue *next_free;
};
static gpr_mu g_freelist_mu;
grpc_completion_queue *g_freelist;
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
int success);
void grpc_cq_global_init(void) { gpr_mu_init(&g_freelist_mu); }
void grpc_cq_global_shutdown(void) {
gpr_mu_destroy(&g_freelist_mu);
while (g_freelist) {
grpc_completion_queue *next = g_freelist->next_free;
grpc_pollset_destroy(&g_freelist->pollset);
gpr_free(g_freelist);
g_freelist = next;
}
}
struct grpc_cq_alarm {
grpc_timer alarm;
grpc_cq_completion completion;
@ -83,22 +103,41 @@ struct grpc_cq_alarm {
void *tag;
};
static void on_pollset_destroy_done(grpc_exec_ctx *exec_ctx, void *cc,
int success);
grpc_completion_queue *grpc_completion_queue_create(void *reserved) {
grpc_completion_queue *cc = gpr_malloc(sizeof(grpc_completion_queue));
GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved));
grpc_completion_queue *cc;
GPR_ASSERT(!reserved);
memset(cc, 0, sizeof(*cc));
GPR_TIMER_BEGIN("grpc_completion_queue_create", 0);
GRPC_API_TRACE("grpc_completion_queue_create(reserved=%p)", 1, (reserved));
gpr_mu_lock(&g_freelist_mu);
if (g_freelist == NULL) {
gpr_mu_unlock(&g_freelist_mu);
cc = gpr_malloc(sizeof(grpc_completion_queue));
grpc_pollset_init(&cc->pollset);
} else {
cc = g_freelist;
g_freelist = g_freelist->next_free;
gpr_mu_unlock(&g_freelist_mu);
/* pollset already initialized */
}
/* Initial ref is dropped by grpc_completion_queue_shutdown */
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;
grpc_closure_init(&cc->pollset_destroy_done, on_pollset_destroy_done, cc);
cc->shutdown = 0;
cc->shutdown_called = 0;
cc->is_server_cq = 0;
cc->num_pluckers = 0;
grpc_closure_init(&cc->pollset_shutdown_done, on_pollset_shutdown_done, cc);
GPR_TIMER_END("grpc_completion_queue_create", 0);
return cc;
}
@ -113,8 +152,8 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) {
gpr_ref(&cc->owning_refs);
}
static void on_pollset_destroy_done(grpc_exec_ctx *exec_ctx, void *arg,
int success) {
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg,
int success) {
grpc_completion_queue *cc = arg;
GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
}
@ -129,8 +168,11 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) {
#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->completed_head.next == (gpr_uintptr)&cc->completed_head);
grpc_pollset_destroy(&cc->pollset);
gpr_free(cc);
grpc_pollset_reset(&cc->pollset);
gpr_mu_lock(&g_freelist_mu);
cc->next_free = g_freelist;
g_freelist = cc;
gpr_mu_unlock(&g_freelist_mu);
}
}
@ -185,8 +227,8 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
GPR_ASSERT(!cc->shutdown);
GPR_ASSERT(cc->shutdown_called);
cc->shutdown = 1;
grpc_pollset_shutdown(exec_ctx, &cc->pollset, &cc->pollset_shutdown_done);
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
grpc_pollset_shutdown(exec_ctx, &cc->pollset, &cc->pollset_destroy_done);
}
GPR_TIMER_END("grpc_cq_end_op", 0);
@ -365,29 +407,31 @@ done:
to zero here, then enter shutdown mode and wake up any waiters */
void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
GPR_TIMER_BEGIN("grpc_completion_queue_shutdown", 0);
GRPC_API_TRACE("grpc_completion_queue_shutdown(cc=%p)", 1, (cc));
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
if (cc->shutdown_called) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
return;
}
cc->shutdown_called = 1;
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
if (gpr_unref(&cc->pending_events)) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
GPR_ASSERT(!cc->shutdown);
cc->shutdown = 1;
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
grpc_pollset_shutdown(&exec_ctx, &cc->pollset, &cc->pollset_destroy_done);
grpc_pollset_shutdown(&exec_ctx, &cc->pollset, &cc->pollset_shutdown_done);
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
grpc_exec_ctx_finish(&exec_ctx);
GPR_TIMER_END("grpc_completion_queue_shutdown", 0);
}
void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
GRPC_API_TRACE("grpc_completion_queue_destroy(cc=%p)", 1, (cc));
GPR_TIMER_BEGIN("grpc_completion_queue_destroy", 0);
grpc_completion_queue_shutdown(cc);
GRPC_CQ_INTERNAL_UNREF(cc, "destroy");
GPR_TIMER_END("grpc_completion_queue_destroy", 0);
}
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {

@ -83,4 +83,7 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
int grpc_cq_is_server_cq(grpc_completion_queue *cc);
void grpc_cq_global_init(void);
void grpc_cq_global_shutdown(void);
#endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */

@ -52,6 +52,7 @@
#include "src/core/profiling/timers.h"
#include "src/core/surface/api_trace.h"
#include "src/core/surface/call.h"
#include "src/core/surface/completion_queue.h"
#include "src/core/surface/init.h"
#include "src/core/surface/surface_trace.h"
#include "src/core/transport/chttp2_transport.h"
@ -118,6 +119,7 @@ void grpc_init(void) {
}
}
gpr_timers_global_init();
grpc_cq_global_init();
for (i = 0; i < g_number_of_plugins; i++) {
if (g_all_of_the_plugins[i].init != NULL) {
g_all_of_the_plugins[i].init();
@ -133,8 +135,9 @@ void grpc_shutdown(void) {
GRPC_API_TRACE("grpc_shutdown(void)", 0, ());
gpr_mu_lock(&g_init_mu);
if (--g_initializations == 0) {
grpc_iomgr_shutdown();
grpc_executor_shutdown();
grpc_cq_global_shutdown();
grpc_iomgr_shutdown();
census_shutdown();
gpr_timers_global_destroy();
grpc_tracer_shutdown();

@ -55,38 +55,33 @@ typedef struct {
const char *error_message;
} channel_data;
static void fill_metadata(grpc_call_element *elem, grpc_metadata_batch *mdb) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
char tmp[GPR_LTOA_MIN_BUFSIZE];
gpr_ltoa(chand->error_code, tmp);
calld->status.md = grpc_mdelem_from_strings(chand->mdctx, "grpc-status", tmp);
calld->details.md = grpc_mdelem_from_strings(chand->mdctx, "grpc-message",
chand->error_message);
calld->status.prev = calld->details.next = NULL;
calld->status.next = &calld->details;
calld->details.prev = &calld->status;
mdb->list.head = &calld->status;
mdb->list.tail = &calld->details;
mdb->deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
}
static void lame_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem,
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
if (op->send_ops != NULL) {
grpc_stream_ops_unref_owned_objects(op->send_ops->ops, op->send_ops->nops);
op->on_done_send->cb(exec_ctx, op->on_done_send->cb_arg, 0);
}
if (op->recv_ops != NULL) {
char tmp[GPR_LTOA_MIN_BUFSIZE];
grpc_metadata_batch mdb;
gpr_ltoa(chand->error_code, tmp);
calld->status.md =
grpc_mdelem_from_strings(chand->mdctx, "grpc-status", tmp);
calld->details.md = grpc_mdelem_from_strings(chand->mdctx, "grpc-message",
chand->error_message);
calld->status.prev = calld->details.next = NULL;
calld->status.next = &calld->details;
calld->details.prev = &calld->status;
mdb.list.head = &calld->status;
mdb.list.tail = &calld->details;
mdb.garbage.head = mdb.garbage.tail = NULL;
mdb.deadline = gpr_inf_future(GPR_CLOCK_REALTIME);
grpc_sopb_add_metadata(op->recv_ops, mdb);
*op->recv_state = GRPC_STREAM_CLOSED;
op->on_done_recv->cb(exec_ctx, op->on_done_recv->cb_arg, 1);
}
if (op->on_consumed != NULL) {
op->on_consumed->cb(exec_ctx, op->on_consumed->cb_arg, 0);
if (op->recv_initial_metadata != NULL) {
fill_metadata(elem, op->recv_initial_metadata);
} else if (op->recv_trailing_metadata != NULL) {
fill_metadata(elem, op->recv_trailing_metadata);
}
grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, 0);
grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, 0);
}
static char *lame_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
@ -109,25 +104,19 @@ static void lame_start_transport_op(grpc_exec_ctx *exec_ctx,
}
static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
const void *transport_server_data,
grpc_transport_stream_op *initial_op) {
if (initial_op) {
grpc_transport_stream_op_finish_with_failure(exec_ctx, initial_op);
}
}
grpc_call_element_args *args) {}
static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
grpc_call_element *elem) {}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem, grpc_channel *master,
const grpc_channel_args *args, grpc_mdctx *mdctx,
int is_first, int is_last) {
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
GPR_ASSERT(is_first);
GPR_ASSERT(is_last);
chand->mdctx = mdctx;
chand->master = master;
GPR_ASSERT(args->is_first);
GPR_ASSERT(args->is_last);
chand->mdctx = args->metadata_context;
chand->master = args->master;
}
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
@ -135,8 +124,9 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
static const grpc_channel_filter lame_filter = {
lame_start_transport_stream_op, lame_start_transport_op, sizeof(call_data),
init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem,
destroy_channel_elem, lame_get_peer, "lame-client",
init_call_elem, grpc_call_stack_ignore_set_pollset, destroy_call_elem,
sizeof(channel_data), init_channel_elem, destroy_channel_elem,
lame_get_peer, "lame-client",
};
#define CHANNEL_STACK_FROM_CHANNEL(c) ((grpc_channel_stack *)((c) + 1))

@ -84,18 +84,18 @@ typedef struct requested_call {
grpc_completion_queue *cq_for_notification;
grpc_call **call;
grpc_cq_completion completion;
grpc_metadata_array *initial_metadata;
union {
struct {
grpc_call_details *details;
grpc_metadata_array *initial_metadata;
} batch;
struct {
registered_method *registered_method;
gpr_timespec *deadline;
grpc_metadata_array *initial_metadata;
grpc_byte_buffer **optional_payload;
} registered;
} data;
grpc_closure publish;
} requested_call;
typedef struct channel_registered_method {
@ -150,16 +150,16 @@ struct call_data {
grpc_mdstr *path;
grpc_mdstr *host;
gpr_timespec deadline;
int got_initial_metadata;
grpc_completion_queue *cq_new;
grpc_stream_op_buffer *recv_ops;
grpc_stream_state *recv_state;
grpc_closure *on_done_recv;
grpc_metadata_batch *recv_initial_metadata;
grpc_metadata_array initial_metadata;
grpc_closure server_on_recv;
grpc_closure got_initial_metadata;
grpc_closure server_on_recv_initial_metadata;
grpc_closure kill_zombie_closure;
grpc_closure *on_done_recv_initial_metadata;
call_data *pending_next;
};
@ -396,7 +396,6 @@ static void finish_destroy_channel(grpc_exec_ctx *exec_ctx, void *cd,
int success) {
channel_data *chand = cd;
grpc_server *server = chand->server;
gpr_log(GPR_DEBUG, "finish_destroy_channel: %p", chand->channel);
GRPC_CHANNEL_INTERNAL_UNREF(exec_ctx, chand->channel, "server");
server_unref(exec_ctx, server);
}
@ -571,79 +570,35 @@ static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
return md;
}
static void server_on_recv(grpc_exec_ctx *exec_ctx, void *ptr, int success) {
static void server_on_recv_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
int success) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
gpr_timespec op_deadline;
if (success && !calld->got_initial_metadata) {
size_t i;
size_t nops = calld->recv_ops->nops;
grpc_stream_op *ops = calld->recv_ops->ops;
for (i = 0; i < nops; i++) {
grpc_stream_op *op = &ops[i];
if (op->type != GRPC_OP_METADATA) continue;
grpc_metadata_batch_filter(&op->data.metadata, server_filter, elem);
op_deadline = op->data.metadata.deadline;
if (0 !=
gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) {
calld->deadline = op->data.metadata.deadline;
}
if (calld->host && calld->path) {
calld->got_initial_metadata = 1;
start_new_rpc(exec_ctx, elem);
}
break;
}
grpc_metadata_batch_filter(calld->recv_initial_metadata, server_filter, elem);
op_deadline = calld->recv_initial_metadata->deadline;
if (0 != gpr_time_cmp(op_deadline, gpr_inf_future(op_deadline.clock_type))) {
calld->deadline = op_deadline;
}
switch (*calld->recv_state) {
case GRPC_STREAM_OPEN:
break;
case GRPC_STREAM_SEND_CLOSED:
break;
case GRPC_STREAM_RECV_CLOSED:
gpr_mu_lock(&calld->mu_state);
if (calld->state == NOT_STARTED) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
} else {
gpr_mu_unlock(&calld->mu_state);
}
break;
case GRPC_STREAM_CLOSED:
gpr_mu_lock(&calld->mu_state);
if (calld->state == NOT_STARTED) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
} else if (calld->state == PENDING) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
/* zombied call will be destroyed when it's removed from the pending
queue... later */
} else {
gpr_mu_unlock(&calld->mu_state);
}
break;
if (calld->host && calld->path) {
/* do nothing */
} else {
success = 0;
}
calld->on_done_recv->cb(exec_ctx, calld->on_done_recv->cb_arg, success);
calld->on_done_recv_initial_metadata->cb(
exec_ctx, calld->on_done_recv_initial_metadata->cb_arg, success);
}
static void server_mutate_op(grpc_call_element *elem,
grpc_transport_stream_op *op) {
call_data *calld = elem->call_data;
if (op->recv_ops) {
/* substitute our callback for the higher callback */
calld->recv_ops = op->recv_ops;
calld->recv_state = op->recv_state;
calld->on_done_recv = op->on_done_recv;
op->on_done_recv = &calld->server_on_recv;
if (op->recv_initial_metadata != NULL) {
calld->recv_initial_metadata = op->recv_initial_metadata;
calld->on_done_recv_initial_metadata = op->on_complete;
op->on_complete = &calld->server_on_recv_initial_metadata;
}
}
@ -655,12 +610,48 @@ static void server_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
grpc_call_next_op(exec_ctx, elem, op);
}
static void accept_stream(void *cd, grpc_transport *transport,
static void got_initial_metadata(grpc_exec_ctx *exec_ctx, void *ptr,
int success) {
grpc_call_element *elem = ptr;
call_data *calld = elem->call_data;
if (success) {
start_new_rpc(exec_ctx, elem);
} else {
gpr_mu_lock(&calld->mu_state);
if (calld->state == NOT_STARTED) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
grpc_closure_init(&calld->kill_zombie_closure, kill_zombie, elem);
grpc_exec_ctx_enqueue(exec_ctx, &calld->kill_zombie_closure, 1);
} else if (calld->state == PENDING) {
calld->state = ZOMBIED;
gpr_mu_unlock(&calld->mu_state);
/* zombied call will be destroyed when it's removed from the pending
queue... later */
} else {
gpr_mu_unlock(&calld->mu_state);
}
}
}
static void accept_stream(grpc_exec_ctx *exec_ctx, void *cd,
grpc_transport *transport,
const void *transport_server_data) {
channel_data *chand = cd;
/* create a call */
grpc_call_create(chand->channel, NULL, 0, NULL, transport_server_data, NULL,
0, gpr_inf_future(GPR_CLOCK_MONOTONIC));
grpc_call *call =
grpc_call_create(chand->channel, NULL, 0, NULL, transport_server_data,
NULL, 0, gpr_inf_future(GPR_CLOCK_MONOTONIC));
grpc_call_element *elem =
grpc_call_stack_element(grpc_call_get_call_stack(call), 0);
call_data *calld = elem->call_data;
grpc_op op;
memset(&op, 0, sizeof(op));
op.op = GRPC_OP_RECV_INITIAL_METADATA;
op.data.recv_initial_metadata = &calld->initial_metadata;
grpc_closure_init(&calld->got_initial_metadata, got_initial_metadata, elem);
grpc_call_start_batch_and_execute(exec_ctx, call, &op, 1,
&calld->got_initial_metadata);
}
static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
@ -685,8 +676,7 @@ static void channel_connectivity_changed(grpc_exec_ctx *exec_ctx, void *cd,
}
static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
const void *server_transport_data,
grpc_transport_stream_op *initial_op) {
grpc_call_element_args *args) {
call_data *calld = elem->call_data;
channel_data *chand = elem->channel_data;
memset(calld, 0, sizeof(call_data));
@ -694,11 +684,10 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
calld->call = grpc_call_from_top_element(elem);
gpr_mu_init(&calld->mu_state);
grpc_closure_init(&calld->server_on_recv, server_on_recv, elem);
grpc_closure_init(&calld->server_on_recv_initial_metadata,
server_on_recv_initial_metadata, elem);
server_ref(chand->server);
if (initial_op) server_mutate_op(elem, initial_op);
}
static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
@ -714,6 +703,7 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
if (calld->path) {
GRPC_MDSTR_UNREF(calld->path);
}
grpc_metadata_array_destroy(&calld->initial_metadata);
gpr_mu_destroy(&calld->mu_state);
@ -721,17 +711,16 @@ static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
}
static void init_channel_elem(grpc_exec_ctx *exec_ctx,
grpc_channel_element *elem, grpc_channel *master,
const grpc_channel_args *args,
grpc_mdctx *metadata_context, int is_first,
int is_last) {
grpc_channel_element *elem,
grpc_channel_element_args *args) {
channel_data *chand = elem->channel_data;
GPR_ASSERT(is_first);
GPR_ASSERT(!is_last);
GPR_ASSERT(args->is_first);
GPR_ASSERT(!args->is_last);
chand->server = NULL;
chand->channel = NULL;
chand->path_key = grpc_mdstr_from_string(metadata_context, ":path");
chand->authority_key = grpc_mdstr_from_string(metadata_context, ":authority");
chand->path_key = grpc_mdstr_from_string(args->metadata_context, ":path");
chand->authority_key =
grpc_mdstr_from_string(args->metadata_context, ":authority");
chand->next = chand->prev = chand;
chand->registered_methods = NULL;
chand->connectivity_state = GRPC_CHANNEL_IDLE;
@ -769,8 +758,9 @@ static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
static const grpc_channel_filter server_surface_filter = {
server_start_transport_stream_op, grpc_channel_next_op, sizeof(call_data),
init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem,
destroy_channel_elem, grpc_call_next_get_peer, "server",
init_call_elem, grpc_call_stack_ignore_set_pollset, destroy_call_elem,
sizeof(channel_data), init_channel_elem, destroy_channel_elem,
grpc_call_next_get_peer, "server",
};
void grpc_server_register_completion_queue(grpc_server *server,
@ -1022,8 +1012,6 @@ void grpc_server_shutdown_and_notify(grpc_server *server,
GRPC_API_TRACE("grpc_server_shutdown_and_notify(server=%p, cq=%p, tag=%p)", 3,
(server, cq, tag));
GRPC_SERVER_LOG_SHUTDOWN(GPR_INFO, server, cq, tag);
/* lock, and gather up some stuff to do */
gpr_mu_lock(&server->mu_global);
grpc_cq_begin_op(cq);
@ -1187,12 +1175,9 @@ grpc_call_error grpc_server_request_call(
GRPC_API_TRACE(
"grpc_server_request_call("
"server=%p, call=%p, details=%p, initial_metadata=%p, "
"cq_bound_to_call=%p, cq_for_notification=%p, tag%p)",
"cq_bound_to_call=%p, cq_for_notification=%p, tag=%p)",
7, (server, call, details, initial_metadata, cq_bound_to_call,
cq_for_notification, tag));
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);
error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
@ -1207,7 +1192,7 @@ grpc_call_error grpc_server_request_call(
rc->cq_for_notification = cq_for_notification;
rc->call = call;
rc->data.batch.details = details;
rc->data.batch.initial_metadata = initial_metadata;
rc->initial_metadata = initial_metadata;
error = queue_call_request(&exec_ctx, server, rc);
done:
grpc_exec_ctx_finish(&exec_ctx);
@ -1244,7 +1229,7 @@ grpc_call_error grpc_server_request_registered_call(
rc->call = call;
rc->data.registered.registered_method = rm;
rc->data.registered.deadline = deadline;
rc->data.registered.initial_metadata = initial_metadata;
rc->initial_metadata = initial_metadata;
rc->data.registered.optional_payload = optional_payload;
error = queue_call_request(&exec_ctx, server, rc);
done:
@ -1253,12 +1238,7 @@ done:
}
static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx,
grpc_call *call, int success,
void *tag);
static void publish_was_not_set(grpc_exec_ctx *exec_ctx, grpc_call *call,
int success, void *tag) {
abort();
}
void *user_data, int success);
static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
gpr_slice slice = value->slice;
@ -1273,9 +1253,10 @@ static void cpstr(char **dest, size_t *capacity, grpc_mdstr *value) {
static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
call_data *calld, requested_call *rc) {
grpc_ioreq_completion_func publish = publish_was_not_set;
grpc_ioreq req[2];
grpc_ioreq *r = req;
grpc_op ops[1];
grpc_op *op = ops;
memset(ops, 0, sizeof(ops));
/* called once initial metadata has been read by the call, but BEFORE
the ioreq to fetch it out of the call has been executed.
@ -1284,8 +1265,10 @@ static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
an ioreq op, that should complete immediately. */
grpc_call_set_completion_queue(exec_ctx, calld->call, rc->cq_bound_to_call);
grpc_closure_init(&rc->publish, publish_registered_or_batch, rc);
*rc->call = calld->call;
calld->cq_new = rc->cq_for_notification;
GPR_SWAP(grpc_metadata_array, *rc->initial_metadata, calld->initial_metadata);
switch (rc->type) {
case BATCH_CALL:
GPR_ASSERT(calld->host != NULL);
@ -1295,31 +1278,22 @@ static void begin_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
cpstr(&rc->data.batch.details->method,
&rc->data.batch.details->method_capacity, calld->path);
rc->data.batch.details->deadline = calld->deadline;
r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
r->data.recv_metadata = rc->data.batch.initial_metadata;
r->flags = 0;
r++;
publish = publish_registered_or_batch;
break;
case REGISTERED_CALL:
*rc->data.registered.deadline = calld->deadline;
r->op = GRPC_IOREQ_RECV_INITIAL_METADATA;
r->data.recv_metadata = rc->data.registered.initial_metadata;
r->flags = 0;
r++;
if (rc->data.registered.optional_payload) {
r->op = GRPC_IOREQ_RECV_MESSAGE;
r->data.recv_message = rc->data.registered.optional_payload;
r->flags = 0;
r++;
op->op = GRPC_OP_RECV_MESSAGE;
op->data.recv_message = rc->data.registered.optional_payload;
op++;
}
publish = publish_registered_or_batch;
break;
default:
GPR_UNREACHABLE_CODE(return );
}
GRPC_CALL_INTERNAL_REF(calld->call, "server");
grpc_call_start_ioreq_and_call_back(exec_ctx, calld->call, req,
(size_t)(r - req), publish, rc);
grpc_call_start_batch_and_execute(exec_ctx, calld->call, ops,
(size_t)(op - ops), &rc->publish);
}
static void done_request_event(grpc_exec_ctx *exec_ctx, void *req,
@ -1342,25 +1316,19 @@ static void done_request_event(grpc_exec_ctx *exec_ctx, void *req,
static void fail_call(grpc_exec_ctx *exec_ctx, grpc_server *server,
requested_call *rc) {
*rc->call = NULL;
switch (rc->type) {
case BATCH_CALL:
rc->data.batch.initial_metadata->count = 0;
break;
case REGISTERED_CALL:
rc->data.registered.initial_metadata->count = 0;
break;
}
rc->initial_metadata->count = 0;
server_ref(server);
grpc_cq_end_op(exec_ctx, rc->cq_for_notification, rc->tag, 0,
done_request_event, rc, &rc->completion);
}
static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx,
grpc_call *call, int success,
void *prc) {
static void publish_registered_or_batch(grpc_exec_ctx *exec_ctx, void *prc,
int success) {
requested_call *rc = prc;
grpc_call *call = *rc->call;
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;
channel_data *chand = elem->channel_data;
server_ref(chand->server);

Loading…
Cancel
Save