3 * Copyright 2015 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
19 #ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_SUBCHANNEL_H
20 #define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_SUBCHANNEL_H
22 #include <grpc/support/port_platform.h>
24 #include "src/core/ext/filters/client_channel/client_channel_channelz.h"
25 #include "src/core/ext/filters/client_channel/connector.h"
26 #include "src/core/ext/filters/client_channel/subchannel_pool_interface.h"
27 #include "src/core/lib/backoff/backoff.h"
28 #include "src/core/lib/channel/channel_stack.h"
29 #include "src/core/lib/gpr/arena.h"
30 #include "src/core/lib/gprpp/ref_counted.h"
31 #include "src/core/lib/gprpp/ref_counted_ptr.h"
32 #include "src/core/lib/iomgr/polling_entity.h"
33 #include "src/core/lib/iomgr/timer.h"
34 #include "src/core/lib/transport/connectivity_state.h"
35 #include "src/core/lib/transport/metadata.h"
37 // Channel arg containing a grpc_resolved_address to connect to.
38 #define GRPC_ARG_SUBCHANNEL_ADDRESS "grpc.subchannel_address"
40 // For debugging refcounting.
42 #define GRPC_SUBCHANNEL_REF(p, r) (p)->Ref(__FILE__, __LINE__, (r))
43 #define GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(p, r) \
44 (p)->RefFromWeakRef(__FILE__, __LINE__, (r))
45 #define GRPC_SUBCHANNEL_UNREF(p, r) (p)->Unref(__FILE__, __LINE__, (r))
46 #define GRPC_SUBCHANNEL_WEAK_REF(p, r) (p)->WeakRef(__FILE__, __LINE__, (r))
47 #define GRPC_SUBCHANNEL_WEAK_UNREF(p, r) (p)->WeakUnref(__FILE__, __LINE__, (r))
48 #define GRPC_SUBCHANNEL_REF_EXTRA_ARGS \
49 const char *file, int line, const char *reason
50 #define GRPC_SUBCHANNEL_REF_REASON reason
51 #define GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS \
52 , GRPC_SUBCHANNEL_REF_EXTRA_ARGS, const char* purpose
53 #define GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE(x) , file, line, reason, x
55 #define GRPC_SUBCHANNEL_REF(p, r) (p)->Ref()
56 #define GRPC_SUBCHANNEL_REF_FROM_WEAK_REF(p, r) (p)->RefFromWeakRef()
57 #define GRPC_SUBCHANNEL_UNREF(p, r) (p)->Unref()
58 #define GRPC_SUBCHANNEL_WEAK_REF(p, r) (p)->WeakRef()
59 #define GRPC_SUBCHANNEL_WEAK_UNREF(p, r) (p)->WeakUnref()
60 #define GRPC_SUBCHANNEL_REF_EXTRA_ARGS
61 #define GRPC_SUBCHANNEL_REF_REASON ""
62 #define GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS
63 #define GRPC_SUBCHANNEL_REF_MUTATE_PURPOSE(x)
70 class ConnectedSubchannel : public RefCounted<ConnectedSubchannel> {
73 grpc_polling_entity* pollent;
75 gpr_timespec start_time;
78 grpc_call_context_element* context;
79 grpc_call_combiner* call_combiner;
80 size_t parent_data_size;
84 grpc_channel_stack* channel_stack, const grpc_channel_args* args,
85 RefCountedPtr<channelz::SubchannelNode> channelz_subchannel,
86 intptr_t socket_uuid);
87 ~ConnectedSubchannel();
89 void NotifyOnStateChange(grpc_pollset_set* interested_parties,
90 grpc_connectivity_state* state,
91 grpc_closure* closure);
92 void Ping(grpc_closure* on_initiate, grpc_closure* on_ack);
93 RefCountedPtr<SubchannelCall> CreateCall(const CallArgs& args,
96 grpc_channel_stack* channel_stack() const { return channel_stack_; }
97 const grpc_channel_args* args() const { return args_; }
98 channelz::SubchannelNode* channelz_subchannel() const {
99 return channelz_subchannel_.get();
101 intptr_t socket_uuid() const { return socket_uuid_; }
103 size_t GetInitialCallSizeEstimate(size_t parent_data_size) const;
106 grpc_channel_stack* channel_stack_;
107 grpc_channel_args* args_;
108 // ref counted pointer to the channelz node in this connected subchannel's
109 // owning subchannel.
110 RefCountedPtr<channelz::SubchannelNode> channelz_subchannel_;
111 // uuid of this subchannel's socket. 0 if this subchannel is not connected.
112 const intptr_t socket_uuid_;
115 // Implements the interface of RefCounted<>.
116 class SubchannelCall {
118 SubchannelCall(RefCountedPtr<ConnectedSubchannel> connected_subchannel,
119 const ConnectedSubchannel::CallArgs& args)
120 : connected_subchannel_(std::move(connected_subchannel)),
121 deadline_(args.deadline) {}
123 // Continues processing a transport stream op batch.
124 void StartTransportStreamOpBatch(grpc_transport_stream_op_batch* batch);
126 // Returns a pointer to the parent data associated with the subchannel call.
127 // The data will be of the size specified in \a parent_data_size field of
128 // the args passed to \a ConnectedSubchannel::CreateCall().
129 void* GetParentData();
131 // Returns the call stack of the subchannel call.
132 grpc_call_stack* GetCallStack();
134 // Sets the 'then_schedule_closure' argument for call stack destruction.
135 // Must be called once per call.
136 void SetAfterCallStackDestroy(grpc_closure* closure);
138 // Interface of RefCounted<>.
139 RefCountedPtr<SubchannelCall> Ref() GRPC_MUST_USE_RESULT;
140 RefCountedPtr<SubchannelCall> Ref(const DebugLocation& location,
141 const char* reason) GRPC_MUST_USE_RESULT;
142 // When refcount drops to 0, destroys itself and the associated call stack,
143 // but does NOT free the memory because it's in the call arena.
145 void Unref(const DebugLocation& location, const char* reason);
147 static void Destroy(void* arg, grpc_error* error);
150 // Allow RefCountedPtr<> to access IncrementRefCount().
151 template <typename T>
152 friend class RefCountedPtr;
154 // If channelz is enabled, intercepts recv_trailing so that we may check the
155 // status and associate it to a subchannel.
156 void MaybeInterceptRecvTrailingMetadata(
157 grpc_transport_stream_op_batch* batch);
159 static void RecvTrailingMetadataReady(void* arg, grpc_error* error);
161 // Interface of RefCounted<>.
162 void IncrementRefCount();
163 void IncrementRefCount(const DebugLocation& location, const char* reason);
165 RefCountedPtr<ConnectedSubchannel> connected_subchannel_;
166 grpc_closure* after_call_stack_destroy_ = nullptr;
167 // State needed to support channelz interception of recv trailing metadata.
168 grpc_closure recv_trailing_metadata_ready_;
169 grpc_closure* original_recv_trailing_metadata_ = nullptr;
170 grpc_metadata_batch* recv_trailing_metadata_ = nullptr;
171 grpc_millis deadline_;
174 // A subchannel that knows how to connect to exactly one target address. It
175 // provides a target for load balancing.
178 // The ctor and dtor are not intended to use directly.
179 Subchannel(SubchannelKey* key, grpc_connector* connector,
180 const grpc_channel_args* args);
183 // Creates a subchannel given \a connector and \a args.
184 static Subchannel* Create(grpc_connector* connector,
185 const grpc_channel_args* args);
187 // Strong and weak refcounting.
188 Subchannel* Ref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
189 void Unref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
190 Subchannel* WeakRef(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
191 void WeakUnref(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
192 // Attempts to return a strong ref when only the weak refcount is guaranteed
193 // non-zero. If the strong refcount is zero, does not alter the refcount and
195 Subchannel* RefFromWeakRef(GRPC_SUBCHANNEL_REF_EXTRA_ARGS);
197 intptr_t GetChildSocketUuid();
199 // Gets the string representing the subchannel address.
200 // Caller doesn't take ownership.
201 const char* GetTargetAddress();
203 // Gets the connected subchannel - or nullptr if not connected (which may
204 // happen before it initially connects or during transient failures).
205 RefCountedPtr<ConnectedSubchannel> connected_subchannel();
207 channelz::SubchannelNode* channelz_node();
209 // Polls the current connectivity state of the subchannel.
210 grpc_connectivity_state CheckConnectivity(grpc_error** error,
211 bool inhibit_health_checking);
213 // When the connectivity state of the subchannel changes from \a *state,
214 // invokes \a notify and updates \a *state with the new state.
215 void NotifyOnStateChange(grpc_pollset_set* interested_parties,
216 grpc_connectivity_state* state, grpc_closure* notify,
217 bool inhibit_health_checking);
219 // Resets the connection backoff of the subchannel.
220 // TODO(roth): Move connection backoff out of subchannels and up into LB
221 // policy code (probably by adding a SubchannelGroup between
222 // SubchannelList and SubchannelData), at which point this method can
226 // Returns a new channel arg encoding the subchannel address as a URI
227 // string. Caller is responsible for freeing the string.
228 static grpc_arg CreateSubchannelAddressArg(const grpc_resolved_address* addr);
230 // Returns the URI string from the subchannel address arg in \a args.
231 static const char* GetUriFromSubchannelAddressArg(
232 const grpc_channel_args* args);
234 // Sets \a addr from the subchannel address arg in \a args.
235 static void GetAddressFromSubchannelAddressArg(const grpc_channel_args* args,
236 grpc_resolved_address* addr);
239 struct ExternalStateWatcher;
240 class ConnectedSubchannelStateWatcher;
242 // Sets the subchannel's connectivity state to \a state.
243 void SetConnectivityStateLocked(grpc_connectivity_state state,
244 grpc_error* error, const char* reason);
246 // Methods for connection.
247 void MaybeStartConnectingLocked();
248 static void OnRetryAlarm(void* arg, grpc_error* error);
249 void ContinueConnectingLocked();
250 static void OnConnectingFinished(void* arg, grpc_error* error);
251 bool PublishTransportLocked();
254 gpr_atm RefMutate(gpr_atm delta,
255 int barrier GRPC_SUBCHANNEL_REF_MUTATE_EXTRA_ARGS);
257 // The subchannel pool this subchannel is in.
258 RefCountedPtr<SubchannelPoolInterface> subchannel_pool_;
259 // TODO(juanlishen): Consider using args_ as key_ directly.
260 // Subchannel key that identifies this subchannel in the subchannel pool.
263 grpc_channel_args* args_;
264 // pollset_set tracking who's interested in a connection being setup.
265 grpc_pollset_set* pollset_set_;
266 // Protects the other members.
269 // - lower INTERNAL_REF_BITS bits are for internal references:
270 // these do not keep the subchannel open.
271 // - upper remaining bits are for public references: these do
272 // keep the subchannel open
275 // Connection states.
276 grpc_connector* connector_ = nullptr;
277 // Set during connection.
278 grpc_connect_out_args connecting_result_;
279 grpc_closure on_connecting_finished_;
280 // Active connection, or null.
281 RefCountedPtr<ConnectedSubchannel> connected_subchannel_;
282 OrphanablePtr<ConnectedSubchannelStateWatcher> connected_subchannel_watcher_;
283 bool connecting_ = false;
284 bool disconnected_ = false;
286 // Connectivity state tracking.
287 grpc_connectivity_state_tracker state_tracker_;
288 grpc_connectivity_state_tracker state_and_health_tracker_;
289 UniquePtr<char> health_check_service_name_;
290 ExternalStateWatcher* external_state_watcher_list_ = nullptr;
294 grpc_millis next_attempt_deadline_;
295 grpc_millis min_connect_timeout_ms_;
296 bool backoff_begun_ = false;
299 grpc_timer retry_alarm_;
300 grpc_closure on_retry_alarm_;
301 bool have_retry_alarm_ = false;
302 // reset_backoff() was called while alarm was pending.
303 bool retry_immediately_ = false;
305 // Channelz tracking.
306 RefCountedPtr<channelz::SubchannelNode> channelz_node_;
309 } // namespace grpc_core
311 #endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_SUBCHANNEL_H */