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