Line data Source code
1 : #ifndef HEADER_fd_src_waltz_grpc_fd_grpc_client_h
2 : #define HEADER_fd_src_waltz_grpc_fd_grpc_client_h
3 :
4 : /* fd_grpc_client.h provides an API for dispatching unary and server-
5 : streaming gRPC requests over HTTP/2+TLS. */
6 :
7 : #include "fd_grpc_codec.h"
8 : #include "../../ballet/nanopb/pb_firedancer.h" /* pb_msgdesc_t */
9 : #if FD_HAS_OPENSSL
10 : #include <openssl/types.h> /* SSL */
11 : #endif
12 :
13 : struct fd_grpc_client_private;
14 : typedef struct fd_grpc_client_private fd_grpc_client_t;
15 :
16 : struct fd_grpc_h2_stream;
17 : typedef struct fd_grpc_h2_stream fd_grpc_h2_stream_t;
18 :
19 : /* FD_GRPC_CLIENT_MAX_STREAMS specifies the max number of inflight
20 : unary and server-streaming requests. Note that grpc_client does
21 : not scale well to large numbers due to O(n) algorithms. */
22 :
23 510 : #define FD_GRPC_CLIENT_MAX_STREAMS 8
24 :
25 : /* FD_GRPC_DEADLINE_* identify different types of request deadlines. */
26 :
27 30 : #define FD_GRPC_DEADLINE_HEADER 1 /* deadline by which Response-Headers are recevied */
28 18 : #define FD_GRPC_DEADLINE_RX_END 2 /* deadline by which 'end of stream' must have been reached */
29 :
30 : /* fd_grpc_client_metrics_t hold counters that are incremented by a
31 : grpc_client. */
32 :
33 : struct fd_grpc_client_metrics {
34 :
35 : /* wakeup_cnt counts the number of times the gRPC client was polled
36 : for I/O. */
37 : ulong wakeup_cnt;
38 :
39 : /* stream_err_cnt counts the number of survivable stream errors.
40 : These include out-of-memory conditions and decode failures. */
41 : ulong stream_err_cnt;
42 :
43 : /* conn_err_cnt counts the number of connection errors that resulted
44 : in connection termination. These include protocol and I/O errors. */
45 : ulong conn_err_cnt;
46 :
47 : /* stream_chunks_tx_cnt increments whenever a DATA frame containing
48 : request bytes is sent. stream_chunks_tx_bytes counts the number of
49 : stream bytes sent. */
50 : ulong stream_chunks_tx_cnt;
51 : ulong stream_chunks_tx_bytes;
52 :
53 : /* stream_chunks_rx_cnt increments whenever a DATA frame containing
54 : response bytes is received. stream_chunks_rx_bytes counts the
55 : number of stream bytes received. */
56 : ulong stream_chunks_rx_cnt;
57 : ulong stream_chunks_rx_bytes;
58 :
59 : /* requests_sent increments whenever a gRPC request finished sending. */
60 : ulong requests_sent;
61 :
62 : /* streams_active is the number of streams not in 'closed' state. */
63 : long streams_active;
64 :
65 : /* rx_wait_ticks_cum is the cumulative time in ticks that incoming
66 : gRPC messages were in a "waiting" state. The waiting state begins
67 : when the first byte of a HTTP/2 frame is received, and ends when
68 : all gRPC message bytes are received.
69 :
70 : This is a rough measure of server-to-client congestion. On a
71 : healthy connection, this value should be close to zero. */
72 : long rx_wait_ticks_cum;
73 :
74 : /* tx_wait_ticks_cum is the cumulative time in ticks that an outgoing
75 : message was in a "waiting" state. The waiting state begins when
76 : a message is ready to be sent, and ends when all message bytes were
77 : handed to the TCP layer.
78 :
79 : This is a rough measure of client-to-server congestion, which can
80 : be caused by the TCP server receive window, TCP client congestion
81 : control, or HTTP/2 server flow control. On a healthy connection,
82 : this value should be close to zero. */
83 : long tx_wait_ticks_cum;
84 :
85 : };
86 :
87 : typedef struct fd_grpc_client_metrics fd_grpc_client_metrics_t;
88 :
89 : /* fd_grpc_client_callbacks_t is a virtual function table containing
90 : grpc_client->app callbacks. */
91 :
92 : struct fd_grpc_client_callbacks {
93 :
94 : /* conn_established is called when the initial HTTP/2 SETTINGS
95 : exchange concludes. Technically, requests can be sent before this
96 : point, though. */
97 :
98 : void
99 : (* conn_established)( void * app_ctx );
100 :
101 : /* conn_dead is called when the HTTP/2 connection ends. To recover
102 : from this condition, call fd_grpc_client_reset(). */
103 :
104 : void
105 : (* conn_dead)( void * app_ctx,
106 : uint h2_err,
107 : int closed_by );
108 :
109 : /* tx_complete marks the completion of a tx operation. */
110 :
111 : void
112 : (* tx_complete)( void * app_ctx,
113 : ulong request_ctx );
114 :
115 : /* rx_start signals that the server sent back a response header
116 : indicating success. rx_start is always called before the first
117 : call to rx_msg for that request_ctx. */
118 :
119 : void
120 : (* rx_start)( void * app_ctx,
121 : ulong request_ctx );
122 :
123 : /* rx_msg delivers a gRPC message. May be called multiple times for
124 : the same request (server streaming). */
125 :
126 : void
127 : (* rx_msg)( void * app_ctx,
128 : void const * protobuf,
129 : ulong protobuf_sz,
130 : ulong request_ctx );
131 :
132 : /* rx_end indicates that no more rx_msg callbacks will be delivered
133 : for a request. */
134 :
135 : void
136 : (* rx_end)( void * app_ctx,
137 : ulong request_ctx,
138 : fd_grpc_resp_hdrs_t * resp );
139 :
140 : /* rx_timeout indicates that a request deadline was exceeded.
141 : deadline_kind indicates which timer fired. */
142 :
143 : void
144 : (* rx_timeout)( void * app_ctx,
145 : ulong request_ctx,
146 : int deadline_kind );
147 :
148 : /* ping_ack delivers an acknowledgement of a PING that was previously
149 : sent by fd_h2_tx_ping. */
150 :
151 : void
152 : (* ping_ack)( void * app_ctx );
153 :
154 : };
155 :
156 : typedef struct fd_grpc_client_callbacks fd_grpc_client_callbacks_t;
157 :
158 : FD_PROTOTYPES_BEGIN
159 :
160 : /* Constructors */
161 :
162 : ulong
163 : fd_grpc_client_align( void );
164 :
165 : ulong
166 : fd_grpc_client_footprint( ulong buf_max );
167 :
168 : fd_grpc_client_t *
169 : fd_grpc_client_new( void * mem,
170 : fd_grpc_client_callbacks_t const * callbacks,
171 : fd_grpc_client_metrics_t * metrics,
172 : void * app_ctx,
173 : ulong buf_max,
174 : ulong rng_seed );
175 :
176 : void *
177 : fd_grpc_client_delete( fd_grpc_client_t * client );
178 :
179 : /* fd_grpc_client_reset cancels all inflight requests and abandons the
180 : HTTP/2 client connection. Config params are kept intact (e.g. host,
181 : port, version). */
182 :
183 : void
184 : fd_grpc_client_reset( fd_grpc_client_t * client );
185 :
186 : /* fd_grpc_client_set_version sets the gRPC client's version string
187 : (relayed via user-agent header). No reference to the provided string
188 : is kept (the content is copied out to the client object). version
189 : does not have to be null-terminated. version_len must be
190 : FD_GRPC_CLIENT_VERSION_LEN_MAX or less, otherwise a warning is logged
191 : and the client's version string remains unchanged. */
192 :
193 : #define FD_GRPC_CLIENT_VERSION_LEN_MAX (63UL)
194 :
195 : void
196 : fd_grpc_client_set_version( fd_grpc_client_t * client,
197 : char const * version,
198 : ulong version_len );
199 :
200 : /* fd_grpc_client_set_authority sets the authority header to the
201 : specified hostname and port number. host_len should be <= 255,
202 : otherwise host is truncated. */
203 :
204 : void
205 : fd_grpc_client_set_authority( fd_grpc_client_t * client,
206 : char const * host,
207 : ulong host_len,
208 : ushort port );
209 :
210 : #if FD_HAS_OPENSSL
211 :
212 : /* fd_grpc_client_rxtx_ossl drives I/O against the SSL object
213 : (SSL_read_ex and SSL_write_ex).
214 :
215 : This function currently copies back-and-forth between SSL and
216 : fd_h2 rbuf. This could be improved by adding an interface to allow
217 : OpenSSL->h2 or h2->OpenSSL writes to directly place data into the
218 : target buffer.
219 :
220 : Returns 1 on success and 0 if there is an unrecoverable SSL error. */
221 :
222 : int
223 : fd_grpc_client_rxtx_ossl( fd_grpc_client_t * client,
224 : SSL * ssl,
225 : int * charge_busy );
226 :
227 : #endif /* FD_HAS_OPENSSL */
228 :
229 : /* fd_grpc_client_rxtx_socket drives I/O against a TCP socket.
230 : (recvmsg(2) and sendmsg(2)). Uses MSG_NOSIGNAL|MSG_DONTWAIT flags.
231 :
232 : Returns -1 if an error was encountered, and errno will be set.
233 : Otherwise, returns 0. */
234 :
235 : int
236 : fd_grpc_client_rxtx_socket( fd_grpc_client_t * client,
237 : int sock_fd,
238 : int * charge_busy );
239 :
240 : /* fd_grpc_client_request_start queues a gRPC request for send. The
241 : request includes one Protobuf message (unary request). The client
242 : can only write one request payload at a time, but can have multiple
243 : requests pending for responses.
244 :
245 : path is the HTTP request path which usually follows the pattern
246 : '/path.to.package/Service.Function'. If auth_token_sz is greater
247 : than zero, adds a request header 'authorization: Bearer *auth_token'.
248 :
249 : request_ctx is an arbitrary number used to identify the request. It
250 : echoes in callbacks.
251 :
252 : fields points to a generated nanopb descriptor. message points to a
253 : generated nanopb struct that the user filled in with info. Calls
254 : pb_encode() internally.
255 :
256 : auth_token is an optional authorization header. The header value is
257 : prepended with "Bearer ". auth_token_sz==0 omits the auth header.
258 :
259 : is_streaming: If 0, this is a unary request and the stream is closed
260 : after sending the first message (END_STREAM flag set). If non-zero,
261 : this is a client streaming request and the stream remains open for
262 : additional messages via fd_grpc_client_stream_send_msg(). The stream
263 : must be explicitly closed with fd_grpc_client_stream_close().
264 :
265 : Conditions for starting send:
266 : - The connection is not dead and the HTTP/2 handshake is complete.
267 : - Client has quota to open a new stream (MAX_CONCURRENT_STREAMS)
268 : - There is no other request still sending.
269 : - The message serialized size does not exceed buf_max (set in
270 : fd_grpc_client_new())
271 : - rbuf_tx is empty. (HTTP/2 frames all flushed out to sockets) */
272 :
273 : fd_grpc_h2_stream_t *
274 : fd_grpc_client_request_start(
275 : fd_grpc_client_t * client,
276 : char const * path,
277 : ulong path_len, /* in [0,128) */
278 : ulong request_ctx,
279 : pb_msgdesc_t const * fields,
280 : void const * message,
281 : char const * auth_token,
282 : ulong auth_token_sz,
283 : int is_streaming
284 : );
285 :
286 : fd_grpc_h2_stream_t *
287 : fd_grpc_client_request_start1(
288 : fd_grpc_client_t * client,
289 : char const * path,
290 : ulong path_len, /* in [0,128) */
291 : ulong request_ctx,
292 : uchar const * protobuf,
293 : ulong protobuf_sz,
294 : char const * auth_token,
295 : ulong auth_token_sz,
296 : int is_streaming );
297 :
298 : /* fd_grpc_client_stream_send_msg sends an additional message on an
299 : already-open client streaming request. This function can only be
300 : called after fd_grpc_client_request_start() was called with
301 : is_streaming=1.
302 :
303 : Returns 1 on success, 0 if the operation failed (connection dead,
304 : buffers blocked, or encoding failure).
305 :
306 : Conditions for sending:
307 : - The connection is alive
308 : - No other send operation is in progress
309 : - rbuf_tx is empty
310 : - The message serialized size does not exceed buf_max */
311 :
312 : int
313 : fd_grpc_client_stream_send_msg(
314 : fd_grpc_client_t * client,
315 : fd_grpc_h2_stream_t * stream,
316 : pb_msgdesc_t const * fields,
317 : void const * message
318 : );
319 :
320 : int
321 : fd_grpc_client_stream_send_msg1(
322 : fd_grpc_client_t * client,
323 : fd_grpc_h2_stream_t * stream,
324 : uchar const * protobuf,
325 : ulong protobuf_sz );
326 :
327 : /* fd_grpc_client_stream_close explicitly closes a client streaming
328 : request by sending an empty DATA frame with the END_STREAM flag.
329 : This signals to the server that no more messages will be sent.
330 :
331 : This function should be called after all messages have been sent via
332 : fd_grpc_client_stream_send_msg() to complete the client stream.
333 :
334 : Returns 1 on success, 0 if the operation failed (connection dead or
335 : buffers blocked).
336 :
337 : Conditions for closing:
338 : - The connection is alive
339 : - No other send operation is in progress
340 : - rbuf_tx is empty */
341 :
342 : int
343 : fd_grpc_client_stream_close(
344 : fd_grpc_client_t * client,
345 : fd_grpc_h2_stream_t * stream
346 : );
347 :
348 : /* fd_grpc_client_deadline_set sets a request deadline (used to
349 : configure timeouts). deadline_kind is FD_GRPC_DEADLINE_*. Logs
350 : error and aborts app if deadline_kind is unsupported.
351 :
352 : Behavior for different deadline kinds:
353 : - HEADER: Deadline by which gRPC Response-Headers must have been
354 : received
355 : - RX_END: Deadline by which the response stream must have been ended.
356 : For unary responses, this is the point at which the message
357 : has been fully received. For server-streaming responses,
358 : it is the point at which the last message has been
359 : received, and there are no more messages remaining. (Under
360 : the hood, this is indicated by the HTTP/2 END_STREAM flag.) */
361 :
362 : void
363 : fd_grpc_client_deadline_set( fd_grpc_h2_stream_t * stream,
364 : int deadline_kind,
365 : long ts_nanos );
366 :
367 : /* fd_grpc_client_is_connected returns 1 if HTTP/2 SETTINGS were
368 : exchanged, the TLS handshake is complete (if applicable), and the
369 : conn hasn't died. Otherwise, returns 0. */
370 :
371 : int
372 : fd_grpc_client_is_connected( fd_grpc_client_t * client );
373 :
374 : /* fd_grpc_client_request_is_blocked returns 1 if a call to
375 : fd_grpc_client_request_start would certainly fail. Reasons include
376 : SSL / HTTP/2 handshake not complete, or buffers blocked. */
377 :
378 : int
379 : fd_grpc_client_request_is_blocked( fd_grpc_client_t * client );
380 :
381 : int
382 : fd_grpc_client_request_stream_busy( fd_grpc_client_t * client );
383 :
384 : /* Pointers to internals for testing */
385 :
386 : fd_h2_rbuf_t *
387 : fd_grpc_client_rbuf_tx( fd_grpc_client_t * client );
388 :
389 : fd_h2_rbuf_t *
390 : fd_grpc_client_rbuf_rx( fd_grpc_client_t * client );
391 :
392 : fd_h2_conn_t *
393 : fd_grpc_client_h2_conn( fd_grpc_client_t * client );
394 :
395 : extern fd_h2_callbacks_t const fd_grpc_client_h2_callbacks;
396 :
397 : FD_PROTOTYPES_END
398 :
399 : #endif /* HEADER_fd_src_waltz_grpc_fd_grpc_client_h */
|
