Line data Source code
1 : /*
2 : *
3 : * Copyright 2015, Google Inc.
4 : * All rights reserved.
5 : *
6 : * Redistribution and use in source and binary forms, with or without
7 : * modification, are permitted provided that the following conditions are
8 : * met:
9 : *
10 : * * Redistributions of source code must retain the above copyright
11 : * notice, this list of conditions and the following disclaimer.
12 : * * Redistributions in binary form must reproduce the above
13 : * copyright notice, this list of conditions and the following disclaimer
14 : * in the documentation and/or other materials provided with the
15 : * distribution.
16 : * * Neither the name of Google Inc. nor the names of its
17 : * contributors may be used to endorse or promote products derived from
18 : * this software without specific prior written permission.
19 : *
20 : * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 : * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 : * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23 : * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24 : * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25 : * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26 : * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27 : * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28 : * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29 : * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30 : * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31 : *
32 : */
33 :
34 : #include "src/core/channel/connected_channel.h"
35 :
36 : #include <stdarg.h>
37 : #include <stdio.h>
38 : #include <string.h>
39 :
40 : #include "src/core/support/string.h"
41 : #include "src/core/transport/transport.h"
42 : #include "src/core/profiling/timers.h"
43 : #include <grpc/byte_buffer.h>
44 : #include <grpc/support/alloc.h>
45 : #include <grpc/support/log.h>
46 : #include <grpc/support/slice_buffer.h>
47 :
48 : #define MAX_BUFFER_LENGTH 8192
49 :
50 : typedef struct connected_channel_channel_data {
51 : grpc_transport *transport;
52 : } channel_data;
53 :
54 : typedef struct connected_channel_call_data { void *unused; } call_data;
55 :
56 : /* We perform a small hack to locate transport data alongside the connected
57 : channel data in call allocations, to allow everything to be pulled in minimal
58 : cache line requests */
59 : #define TRANSPORT_STREAM_FROM_CALL_DATA(calld) ((grpc_stream *)((calld) + 1))
60 : #define CALL_DATA_FROM_TRANSPORT_STREAM(transport_stream) \
61 : (((call_data *)(transport_stream)) - 1)
62 :
63 : /* Intercept a call operation and either push it directly up or translate it
64 : into transport stream operations */
65 12327294 : static void con_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
66 : grpc_call_element *elem,
67 : grpc_transport_stream_op *op) {
68 12327294 : call_data *calld = elem->call_data;
69 12327294 : channel_data *chand = elem->channel_data;
70 12327294 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
71 12327294 : GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
72 :
73 12327294 : grpc_transport_perform_stream_op(exec_ctx, chand->transport,
74 : TRANSPORT_STREAM_FROM_CALL_DATA(calld), op);
75 12381772 : }
76 :
77 10662 : static void con_start_transport_op(grpc_exec_ctx *exec_ctx,
78 : grpc_channel_element *elem,
79 : grpc_transport_op *op) {
80 10662 : channel_data *chand = elem->channel_data;
81 10662 : grpc_transport_perform_op(exec_ctx, chand->transport, op);
82 10663 : }
83 :
84 : /* Constructor for call_data */
85 4440499 : static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
86 : grpc_call_element_args *args) {
87 4440499 : call_data *calld = elem->call_data;
88 4440499 : channel_data *chand = elem->channel_data;
89 : int r;
90 :
91 4440499 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
92 4440499 : r = grpc_transport_init_stream(exec_ctx, chand->transport,
93 : TRANSPORT_STREAM_FROM_CALL_DATA(calld),
94 : args->refcount, args->server_transport_data);
95 4443353 : GPR_ASSERT(r == 0);
96 4443353 : }
97 :
98 4439743 : static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
99 : grpc_pollset *pollset) {
100 4439743 : call_data *calld = elem->call_data;
101 4439743 : channel_data *chand = elem->channel_data;
102 4439743 : grpc_transport_set_pollset(exec_ctx, chand->transport,
103 : TRANSPORT_STREAM_FROM_CALL_DATA(calld), pollset);
104 4443190 : }
105 :
106 : /* Destructor for call_data */
107 4438787 : static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
108 : grpc_call_element *elem) {
109 4438787 : call_data *calld = elem->call_data;
110 4438787 : channel_data *chand = elem->channel_data;
111 4438787 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
112 4438787 : grpc_transport_destroy_stream(exec_ctx, chand->transport,
113 : TRANSPORT_STREAM_FROM_CALL_DATA(calld));
114 4443170 : }
115 :
116 : /* Constructor for channel_data */
117 6038 : static void init_channel_elem(grpc_exec_ctx *exec_ctx,
118 : grpc_channel_element *elem,
119 : grpc_channel_element_args *args) {
120 6038 : channel_data *cd = (channel_data *)elem->channel_data;
121 6038 : GPR_ASSERT(args->is_last);
122 6038 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
123 6038 : cd->transport = NULL;
124 6038 : }
125 :
126 : /* Destructor for channel_data */
127 5964 : static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
128 : grpc_channel_element *elem) {
129 5964 : channel_data *cd = (channel_data *)elem->channel_data;
130 5964 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
131 5964 : grpc_transport_destroy(exec_ctx, cd->transport);
132 5964 : }
133 :
134 1245 : static char *con_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
135 1245 : channel_data *chand = elem->channel_data;
136 1245 : return grpc_transport_get_peer(exec_ctx, chand->transport);
137 : }
138 :
139 : const grpc_channel_filter grpc_connected_channel_filter = {
140 : con_start_transport_stream_op, con_start_transport_op, sizeof(call_data),
141 : init_call_elem, set_pollset, destroy_call_elem, sizeof(channel_data),
142 : init_channel_elem, destroy_channel_elem, con_get_peer, "connected",
143 : };
144 :
145 6038 : void grpc_connected_channel_bind_transport(grpc_channel_stack *channel_stack,
146 : grpc_transport *transport) {
147 : /* Assumes that the connected channel filter is always the last filter
148 : in a channel stack */
149 6038 : grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack);
150 6037 : channel_data *cd = (channel_data *)elem->channel_data;
151 6037 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
152 6037 : GPR_ASSERT(cd->transport == NULL);
153 6037 : cd->transport = transport;
154 :
155 : /* HACK(ctiller): increase call stack size for the channel to make space
156 : for channel data. We need a cleaner (but performant) way to do this,
157 : and I'm not sure what that is yet.
158 : This is only "safe" because call stacks place no additional data after
159 : the last call element, and the last call element MUST be the connected
160 : channel. */
161 6037 : channel_stack->call_stack_size += grpc_transport_stream_size(transport);
162 6038 : }
163 :
164 0 : grpc_stream *grpc_connected_channel_get_stream(grpc_call_element *elem) {
165 0 : call_data *calld = elem->call_data;
166 0 : return TRANSPORT_STREAM_FROM_CALL_DATA(calld);
167 : }
|
