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