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 <grpc/byte_buffer.h>
43 : #include <grpc/support/alloc.h>
44 : #include <grpc/support/log.h>
45 : #include <grpc/support/slice_buffer.h>
46 :
47 : #define MAX_BUFFER_LENGTH 8192
48 :
49 : typedef struct connected_channel_channel_data {
50 : grpc_transport *transport;
51 : } channel_data;
52 :
53 : typedef struct connected_channel_call_data { void *unused; } call_data;
54 :
55 : /* We perform a small hack to locate transport data alongside the connected
56 : channel data in call allocations, to allow everything to be pulled in minimal
57 : cache line requests */
58 : #define TRANSPORT_STREAM_FROM_CALL_DATA(calld) ((grpc_stream *)((calld) + 1))
59 : #define CALL_DATA_FROM_TRANSPORT_STREAM(transport_stream) \
60 : (((call_data *)(transport_stream)) - 1)
61 :
62 : /* Intercept a call operation and either push it directly up or translate it
63 : into transport stream operations */
64 8216263 : static void con_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
65 : grpc_call_element *elem,
66 : grpc_transport_stream_op *op) {
67 8216263 : call_data *calld = elem->call_data;
68 8216263 : channel_data *chand = elem->channel_data;
69 8216263 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
70 8216263 : GRPC_CALL_LOG_OP(GPR_INFO, elem, op);
71 :
72 8216263 : grpc_transport_perform_stream_op(exec_ctx, chand->transport,
73 : TRANSPORT_STREAM_FROM_CALL_DATA(calld), op);
74 8256369 : }
75 :
76 6956 : static void con_start_transport_op(grpc_exec_ctx *exec_ctx,
77 : grpc_channel_element *elem,
78 : grpc_transport_op *op) {
79 6956 : channel_data *chand = elem->channel_data;
80 6956 : grpc_transport_perform_op(exec_ctx, chand->transport, op);
81 6957 : }
82 :
83 : /* Constructor for call_data */
84 2701992 : static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
85 : const void *server_transport_data,
86 : grpc_transport_stream_op *initial_op) {
87 2701992 : call_data *calld = elem->call_data;
88 2701992 : channel_data *chand = elem->channel_data;
89 : int r;
90 :
91 2701992 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
92 2701992 : r = grpc_transport_init_stream(exec_ctx, chand->transport,
93 : TRANSPORT_STREAM_FROM_CALL_DATA(calld),
94 : server_transport_data, initial_op);
95 2704369 : GPR_ASSERT(r == 0);
96 2704369 : }
97 :
98 : /* Destructor for call_data */
99 2700157 : static void destroy_call_elem(grpc_exec_ctx *exec_ctx,
100 : grpc_call_element *elem) {
101 2700157 : call_data *calld = elem->call_data;
102 2700157 : channel_data *chand = elem->channel_data;
103 2700157 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
104 2700157 : grpc_transport_destroy_stream(exec_ctx, chand->transport,
105 : TRANSPORT_STREAM_FROM_CALL_DATA(calld));
106 2704454 : }
107 :
108 : /* Constructor for channel_data */
109 4012 : static void init_channel_elem(grpc_exec_ctx *exec_ctx,
110 : grpc_channel_element *elem, grpc_channel *master,
111 : const grpc_channel_args *args, grpc_mdctx *mdctx,
112 : int is_first, int is_last) {
113 4012 : channel_data *cd = (channel_data *)elem->channel_data;
114 4012 : GPR_ASSERT(is_last);
115 4012 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
116 4012 : cd->transport = NULL;
117 4012 : }
118 :
119 : /* Destructor for channel_data */
120 4014 : static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
121 : grpc_channel_element *elem) {
122 4014 : channel_data *cd = (channel_data *)elem->channel_data;
123 4014 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
124 4014 : grpc_transport_destroy(exec_ctx, cd->transport);
125 4014 : }
126 :
127 1094 : static char *con_get_peer(grpc_exec_ctx *exec_ctx, grpc_call_element *elem) {
128 1094 : channel_data *chand = elem->channel_data;
129 1094 : return grpc_transport_get_peer(exec_ctx, chand->transport);
130 : }
131 :
132 : const grpc_channel_filter grpc_connected_channel_filter = {
133 : con_start_transport_stream_op, con_start_transport_op, sizeof(call_data),
134 : init_call_elem, destroy_call_elem, sizeof(channel_data), init_channel_elem,
135 : destroy_channel_elem, con_get_peer, "connected",
136 : };
137 :
138 4012 : void grpc_connected_channel_bind_transport(grpc_channel_stack *channel_stack,
139 : grpc_transport *transport) {
140 : /* Assumes that the connected channel filter is always the last filter
141 : in a channel stack */
142 4012 : grpc_channel_element *elem = grpc_channel_stack_last_element(channel_stack);
143 4012 : channel_data *cd = (channel_data *)elem->channel_data;
144 4012 : GPR_ASSERT(elem->filter == &grpc_connected_channel_filter);
145 4012 : GPR_ASSERT(cd->transport == NULL);
146 4012 : cd->transport = transport;
147 :
148 : /* HACK(ctiller): increase call stack size for the channel to make space
149 : for channel data. We need a cleaner (but performant) way to do this,
150 : and I'm not sure what that is yet.
151 : This is only "safe" because call stacks place no additional data after
152 : the last call element, and the last call element MUST be the connected
153 : channel. */
154 4012 : channel_stack->call_stack_size += grpc_transport_stream_size(transport);
155 4013 : }
|
