Line data Source code
1 : #include "fd_grpc_client.h"
2 : #include "fd_grpc_client_private.h"
3 : #include "../../third_party/nanopb/pb_encode.h" /* pb_msgdesc_t */
4 : #include <sys/socket.h>
5 : #include <poll.h>
6 : #include "../h2/fd_h2_rbuf_sock.h"
7 : #include "../tlsrec/fd_tlsrec.h"
8 : #include "fd_grpc_codec.h"
9 :
10 : static int
11 : fd_grpc_client_request_continue( fd_grpc_client_t * client );
12 :
13 : ulong
14 0 : fd_grpc_client_align( void ) {
15 0 : return fd_ulong_max( alignof(fd_grpc_client_t), fd_grpc_h2_stream_pool_align() );
16 0 : }
17 :
18 : ulong
19 0 : fd_grpc_client_footprint( ulong buf_max ) {
20 0 : ulong l = FD_LAYOUT_INIT;
21 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_grpc_client_t), sizeof(fd_grpc_client_t) );
22 0 : l = FD_LAYOUT_APPEND( l, 1UL, buf_max ); /* nanopb_tx */
23 0 : l = FD_LAYOUT_APPEND( l, 1UL, buf_max ); /* frame_scratch */
24 0 : l = FD_LAYOUT_APPEND( l, 1UL, buf_max ); /* frame_rx_buf */
25 0 : l = FD_LAYOUT_APPEND( l, 1UL, buf_max ); /* frame_tx_buf */
26 0 : l = FD_LAYOUT_APPEND( l, fd_grpc_h2_stream_pool_align(), fd_grpc_h2_stream_pool_footprint( FD_GRPC_CLIENT_MAX_STREAMS ) );
27 0 : l = FD_LAYOUT_APPEND( l, 1UL, buf_max*FD_GRPC_CLIENT_MAX_STREAMS );
28 0 : return FD_LAYOUT_FINI( l, fd_grpc_client_align() );
29 0 : }
30 :
31 : static void
32 0 : fd_grpc_h2_stream_reset( fd_grpc_h2_stream_t * stream ) {
33 0 : memset( &stream->s, 0, sizeof(fd_h2_stream_t) );
34 0 : stream->request_ctx = 0UL;
35 0 : memset( &stream->hdrs, 0, sizeof(fd_grpc_resp_hdrs_t) );
36 0 : stream->hdrs.grpc_status = FD_GRPC_STATUS_UNKNOWN;
37 0 : stream->hdrs_received = 0;
38 0 : stream->msg_buf_used = 0UL;
39 0 : stream->msg_sz = 0UL;
40 0 : stream->has_header_deadline = 0;
41 0 : stream->has_rx_end_deadline = 0;
42 0 : stream->rst_pending = 0;
43 0 : stream->tx_wnd_debt = 0L;
44 0 : }
45 :
46 : fd_grpc_client_t *
47 : fd_grpc_client_new( void * mem,
48 : fd_grpc_client_callbacks_t const * callbacks,
49 : fd_grpc_client_metrics_t * metrics,
50 : void * app_ctx,
51 : ulong buf_max,
52 0 : ulong rng_seed ) {
53 0 : if( FD_UNLIKELY( !mem ) ) {
54 0 : FD_LOG_WARNING(( "NULL mem" ));
55 0 : return NULL;
56 0 : }
57 0 : if( FD_UNLIKELY( buf_max<4096UL ) ) {
58 0 : FD_LOG_WARNING(( "undersz buf_max" ));
59 0 : return NULL;
60 0 : }
61 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)mem, fd_grpc_client_align() ) ) ) {
62 0 : FD_LOG_WARNING(( "unaligned mem" ));
63 0 : return NULL;
64 0 : }
65 :
66 0 : FD_SCRATCH_ALLOC_INIT( l, mem );
67 0 : void * client_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_grpc_client_t), sizeof(fd_grpc_client_t) );
68 0 : void * nanopb_tx = FD_SCRATCH_ALLOC_APPEND( l, 1UL, buf_max ); /* nanopb_tx */
69 0 : void * frame_scratch = FD_SCRATCH_ALLOC_APPEND( l, 1UL, buf_max ); /* frame_scratch */
70 0 : void * frame_rx_buf = FD_SCRATCH_ALLOC_APPEND( l, 1UL, buf_max ); /* frame_rx_buf */
71 0 : void * frame_tx_buf = FD_SCRATCH_ALLOC_APPEND( l, 1UL, buf_max ); /* frame_tx_buf */
72 0 : void * stream_pool_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_grpc_h2_stream_pool_align(), fd_grpc_h2_stream_pool_footprint( FD_GRPC_CLIENT_MAX_STREAMS ) );
73 0 : void * stream_buf_mem = FD_SCRATCH_ALLOC_APPEND( l, 1UL, buf_max*FD_GRPC_CLIENT_MAX_STREAMS );
74 0 : ulong end = FD_SCRATCH_ALLOC_FINI( l, fd_grpc_client_align() );
75 0 : FD_TEST( end-(ulong)mem == fd_grpc_client_footprint( buf_max ) );
76 :
77 0 : fd_grpc_client_t * client = client_mem;
78 :
79 0 : fd_grpc_h2_stream_t * stream_pool =
80 0 : fd_grpc_h2_stream_pool_join( fd_grpc_h2_stream_pool_new( stream_pool_mem, FD_GRPC_CLIENT_MAX_STREAMS ) );
81 0 : if( FD_UNLIKELY( !stream_pool ) ) FD_LOG_CRIT(( "Failed to create stream pool" )); /* unreachable */
82 :
83 0 : *client = (fd_grpc_client_t){
84 0 : .callbacks = callbacks,
85 0 : .ctx = app_ctx,
86 0 : .stream_pool = stream_pool,
87 0 : .stream_bufs = stream_buf_mem,
88 0 : .nanopb_tx = nanopb_tx,
89 0 : .nanopb_tx_max = buf_max,
90 0 : .frame_scratch = frame_scratch,
91 0 : .frame_scratch_max = buf_max,
92 0 : .frame_rx_buf = frame_rx_buf,
93 0 : .frame_rx_buf_max = buf_max,
94 0 : .frame_tx_buf = frame_tx_buf,
95 0 : .frame_tx_buf_max = buf_max,
96 0 : .metrics = metrics
97 0 : };
98 :
99 : /* FIXME for performance, cache this? */
100 0 : fd_h2_hdr_matcher_init( client->matcher, rng_seed );
101 0 : fd_h2_hdr_matcher_insert_literal( client->matcher, FD_GRPC_HDR_STATUS, "grpc-status" );
102 0 : fd_h2_hdr_matcher_insert_literal( client->matcher, FD_GRPC_HDR_MESSAGE, "grpc-message" );
103 :
104 0 : client->version_len = 5;
105 0 : memcpy( client->version, "0.0.0", 5 );
106 :
107 0 : for( ulong i=0UL; i<FD_GRPC_CLIENT_MAX_STREAMS; i++ ) {
108 0 : fd_grpc_h2_stream_t * stream = &client->stream_pool[ i ];
109 0 : stream->msg_buf = (uchar *)stream_buf_mem + (i*buf_max);
110 0 : stream->msg_buf_max = buf_max;
111 0 : FD_TEST( (ulong)( stream->msg_buf + stream->msg_buf_max )<=end );
112 0 : }
113 0 : fd_grpc_client_reset( client );
114 :
115 0 : return client;
116 0 : }
117 :
118 : void *
119 0 : fd_grpc_client_delete( fd_grpc_client_t * client ) {
120 0 : return client;
121 0 : }
122 :
123 : void
124 : fd_grpc_client_set_version( fd_grpc_client_t * client,
125 : char const * version,
126 0 : ulong version_len ) {
127 0 : if( FD_UNLIKELY( version_len > FD_GRPC_CLIENT_VERSION_LEN_MAX ) ) {
128 0 : FD_LOG_WARNING(( "Version string too long (%lu chars), ignoring", version_len ));
129 0 : return;
130 0 : }
131 0 : client->version_len = (uchar)version_len;
132 0 : memcpy( client->version, version, version_len );
133 0 : }
134 :
135 : void
136 : fd_grpc_client_set_authority( fd_grpc_client_t * client,
137 : char const * host,
138 : ulong host_len,
139 0 : ushort port ) {
140 0 : host_len = fd_ulong_min( host_len, sizeof(client->host)-1 );
141 0 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( client->host ), host, host_len ) );
142 0 : client->host_len = (uchar)host_len;
143 0 : client->port = (ushort)port;
144 0 : }
145 :
146 : int
147 0 : fd_grpc_client_stream_acquire_is_safe( fd_grpc_client_t * client ) {
148 : /* Sufficient quota to start a stream? */
149 0 : if( FD_UNLIKELY( client->conn->stream_active_cnt[1]+1 > client->conn->peer_settings.max_concurrent_streams ) ) {
150 0 : return 0;
151 0 : }
152 :
153 : /* Free stream object available? */
154 0 : if( FD_UNLIKELY( !fd_grpc_h2_stream_pool_free( client->stream_pool ) ) ) {
155 0 : return 0;
156 0 : }
157 0 : if( FD_UNLIKELY( client->stream_cnt >= FD_GRPC_CLIENT_MAX_STREAMS ) ) {
158 0 : return 0;
159 0 : }
160 :
161 0 : return 1;
162 0 : }
163 :
164 : fd_grpc_h2_stream_t *
165 : fd_grpc_client_stream_acquire( fd_grpc_client_t * client,
166 0 : ulong request_ctx ) {
167 0 : if( FD_UNLIKELY( client->stream_cnt >= FD_GRPC_CLIENT_MAX_STREAMS ) ) {
168 0 : FD_LOG_CRIT(( "stream pool exhausted" ));
169 0 : }
170 :
171 0 : fd_h2_conn_t * conn = client->conn;
172 0 : uint const stream_id = client->conn->tx_stream_next;
173 0 : conn->tx_stream_next += 2U;
174 :
175 0 : fd_grpc_h2_stream_t * stream = fd_grpc_h2_stream_pool_ele_acquire( client->stream_pool );
176 0 : fd_grpc_h2_stream_reset( stream );
177 0 : stream->request_ctx = request_ctx;
178 :
179 0 : fd_h2_stream_open( fd_h2_stream_init( &stream->s ), conn, stream_id );
180 0 : client->request_stream = stream;
181 0 : client->stream_ids[ client->stream_cnt ] = stream_id;
182 0 : client->streams [ client->stream_cnt ] = stream;
183 0 : client->stream_cnt++;
184 :
185 0 : return stream;
186 0 : }
187 :
188 : void
189 : fd_grpc_client_stream_release( fd_grpc_client_t * client,
190 0 : fd_grpc_h2_stream_t * stream ) {
191 0 : if( FD_UNLIKELY( !client->stream_cnt ) ) FD_LOG_CRIT(( "stream map corrupt" )); /* unreachable */
192 :
193 : /* Deallocate tx_op */
194 0 : if( FD_UNLIKELY( stream == client->request_stream ) ) {
195 0 : client->request_stream = NULL;
196 0 : *client->request_tx_op = (fd_h2_tx_op_t){0};
197 0 : }
198 :
199 : /* Remove stream from map */
200 0 : int map_idx = -1;
201 0 : for( uint i=0UL; i<(client->stream_cnt); i++ ) {
202 0 : if( client->stream_ids[ i ] == stream->s.stream_id ) {
203 0 : map_idx = (int)i;
204 0 : }
205 0 : }
206 0 : if( FD_UNLIKELY( map_idx<0 ) ) FD_LOG_CRIT(( "stream map corrupt" )); /* unreachable */
207 0 : if( (ulong)map_idx+1 < client->stream_cnt ) {
208 0 : client->stream_ids[ map_idx ] = client->stream_ids[ client->stream_cnt-1 ];
209 0 : client->streams [ map_idx ] = client->streams [ client->stream_cnt-1 ];
210 0 : }
211 0 : client->stream_cnt--;
212 :
213 0 : fd_grpc_h2_stream_pool_ele_release( client->stream_pool, stream );
214 0 : }
215 :
216 : void
217 0 : fd_grpc_client_reset( fd_grpc_client_t * client ) {
218 0 : fd_h2_rbuf_init( client->frame_rx, client->frame_rx_buf, client->frame_rx_buf_max );
219 0 : fd_h2_rbuf_init( client->frame_tx, client->frame_tx_buf, client->frame_tx_buf_max );
220 0 : fd_h2_conn_init_client( client->conn );
221 0 : client->conn->ctx = client;
222 0 : client->h2_hs_done = 0;
223 0 : client->window_update_pending = 0;
224 0 : client->request_stream = NULL;
225 0 : fd_tlsrec_sock_init( client->tls_sock );
226 0 : *client->request_tx_op = (fd_h2_tx_op_t){0};
227 :
228 : /* Disable RX flow control */
229 0 : client->conn->self_settings.initial_window_size = (1U<<31)-1U;
230 0 : client->conn->rx_wnd_max = (1U<<31)-1U;
231 0 : client->conn->rx_wnd_wmark = client->conn->rx_wnd_max - (1U<<20);
232 :
233 : /* Free all stream objects */
234 0 : while( client->stream_cnt ) {
235 0 : fd_grpc_h2_stream_t * stream = client->streams[ client->stream_cnt-1 ];
236 0 : fd_grpc_client_stream_release( client, stream );
237 0 : }
238 0 : }
239 :
240 : /* fd_grpc_client_send_stream_quota writes a WINDOW_UPDATE frame, which
241 : eventually allows the peer to send more data bytes. */
242 :
243 : static void
244 : fd_grpc_client_send_stream_quota( fd_h2_rbuf_t * rbuf_tx,
245 : fd_grpc_h2_stream_t * stream,
246 0 : uint bump ) {
247 0 : fd_h2_window_update_t window_update = {
248 0 : .hdr = {
249 0 : .typlen = fd_h2_frame_typlen( FD_H2_FRAME_TYPE_WINDOW_UPDATE, 4UL ),
250 0 : .r_stream_id = fd_uint_bswap( stream->s.stream_id )
251 0 : },
252 0 : .increment = fd_uint_bswap( bump )
253 0 : };
254 0 : fd_h2_rbuf_push( rbuf_tx, &window_update, sizeof(fd_h2_window_update_t) );
255 0 : stream->s.rx_wnd += bump;
256 0 : }
257 :
258 : /* fd_grpc_client_send_timeout is called when a stream timeout triggers.
259 : Writes a RST_STREAM frame unless the stream is closed or frame_tx is
260 : full, then calls back to the user. The caller frees the stream. */
261 :
262 : static void
263 : fd_grpc_client_send_timeout( fd_h2_rbuf_t * rbuf_tx,
264 : fd_grpc_client_t * client,
265 : fd_grpc_h2_stream_t * stream,
266 0 : int deadline_kind ) {
267 0 : if( fd_h2_rbuf_free_sz( rbuf_tx )>=sizeof(fd_h2_rst_stream_t) ) fd_h2_stream_error( &stream->s, client->conn, rbuf_tx, FD_H2_ERR_CANCEL );
268 0 : stream->rst_pending = 1;
269 0 : stream->has_header_deadline = 0;
270 0 : stream->has_rx_end_deadline = 0;
271 0 : if( stream==client->request_stream ) {
272 0 : client->request_stream = NULL;
273 0 : *client->request_tx_op = (fd_h2_tx_op_t){0};
274 0 : }
275 0 : client->callbacks->rx_timeout( client->ctx, stream->request_ctx, deadline_kind );
276 0 : }
277 :
278 : long
279 0 : fd_grpc_client_next_deadline( fd_grpc_client_t const * client ) {
280 0 : long deadline = LONG_MAX;
281 0 : for( ulong i=0UL; i<(client->stream_cnt); i++ ) {
282 0 : fd_grpc_h2_stream_t const * stream = client->streams[ i ];
283 0 : if( stream->rst_pending && fd_h2_rbuf_free_sz( client->frame_tx )>=sizeof(fd_h2_rst_stream_t) &&
284 0 : !( client->conn->flags & (FD_H2_CONN_FLAGS_DEAD|FD_H2_CONN_FLAGS_SEND_GOAWAY) ) ) return 0L; /* deferred RST_STREAM can go out now */
285 0 : if( stream->has_header_deadline ) deadline = fd_long_min( deadline, stream->header_deadline_nanos );
286 0 : if( stream->has_rx_end_deadline ) deadline = fd_long_min( deadline, stream->rx_end_deadline_nanos );
287 0 : }
288 0 : return deadline;
289 0 : }
290 :
291 : int
292 0 : fd_grpc_client_tx_pending( fd_grpc_client_t const * client ) {
293 0 : return !!fd_h2_rbuf_used_sz( client->frame_tx );
294 0 : }
295 :
296 : int
297 0 : fd_grpc_client_tls_rx_pending( fd_grpc_client_t const * client ) {
298 0 : return !!fd_tlsrec_sock_rx_avail( client->tls_sock );
299 0 : }
300 :
301 : int
302 0 : fd_grpc_client_tls_tx_pending( fd_grpc_client_t const * client ) {
303 0 : return fd_tlsrec_sock_tx_pending( client->tls_sock );
304 0 : }
305 :
306 : ulong
307 0 : fd_grpc_client_tx_starved( fd_grpc_client_t const * client ) {
308 0 : fd_grpc_h2_stream_t const * stream = client->request_stream;
309 0 : if( !stream || fd_uint_min( client->conn->tx_wnd, stream->s.tx_wnd ) ) return 0UL;
310 0 : return client->request_tx_op->chunk_sz;
311 0 : }
312 :
313 : void
314 : fd_grpc_client_service_streams( fd_grpc_client_t * client,
315 0 : long ts_nanos ) {
316 0 : ulong const meta_frame_max =
317 0 : fd_ulong_max( sizeof(fd_h2_window_update_t), sizeof(fd_h2_rst_stream_t) );
318 0 : fd_h2_conn_t * conn = client->conn;
319 0 : fd_h2_rbuf_t * rbuf_tx = client->frame_tx;
320 0 : if( FD_UNLIKELY( conn->flags & (FD_H2_CONN_FLAGS_DEAD|FD_H2_CONN_FLAGS_SEND_GOAWAY) ) ) return;
321 0 : uint const wnd_max = conn->self_settings.initial_window_size;
322 0 : uint const wnd_thres = wnd_max / 2;
323 0 : for( ulong i=0UL; i<(client->stream_cnt); i++ ) {
324 0 : fd_grpc_h2_stream_t * stream = client->streams[ i ];
325 :
326 0 : if( FD_UNLIKELY( ( stream->has_header_deadline ) &
327 0 : ( stream->header_deadline_nanos - ts_nanos <= 0L ) ) ) {
328 0 : fd_grpc_client_send_timeout( rbuf_tx, client, stream, FD_GRPC_DEADLINE_HEADER );
329 0 : }
330 :
331 0 : if( FD_UNLIKELY( ( stream->has_rx_end_deadline ) &
332 0 : ( stream->rx_end_deadline_nanos - ts_nanos <= 0L ) ) ) {
333 0 : fd_grpc_client_send_timeout( rbuf_tx, client, stream, FD_GRPC_DEADLINE_RX_END );
334 0 : }
335 :
336 0 : if( FD_UNLIKELY( stream->rst_pending ) ) {
337 0 : if( fd_h2_rbuf_free_sz( rbuf_tx )<meta_frame_max && stream->s.state!=FD_H2_STREAM_STATE_CLOSED ) continue;
338 0 : fd_h2_stream_error( &stream->s, conn, rbuf_tx, FD_H2_ERR_CANCEL );
339 0 : fd_grpc_client_stream_release( client, stream );
340 0 : i--; /* stream removed */
341 0 : continue;
342 0 : }
343 :
344 0 : if( FD_UNLIKELY( stream->s.rx_wnd < wnd_thres && stream->s.state!=FD_H2_STREAM_STATE_CLOSED && fd_h2_rbuf_free_sz( rbuf_tx )>=meta_frame_max ) ) {
345 0 : uint const bump = wnd_max - stream->s.rx_wnd;
346 0 : fd_grpc_client_send_stream_quota( rbuf_tx, stream, bump );
347 0 : }
348 0 : }
349 0 : }
350 :
351 : #if FD_H2_HAS_SOCKETS
352 :
353 : int
354 : fd_grpc_client_tls_flush( fd_grpc_client_t * client,
355 0 : int sock_fd ) {
356 0 : int rc = fd_tlsrec_sock_flush( client->tls_sock, sock_fd );
357 0 : return rc<0 ? -1 : rc;
358 0 : }
359 :
360 : int
361 : fd_grpc_client_rxtx_socket( fd_grpc_client_t * client,
362 : int sock_fd,
363 : long now,
364 0 : int * charge_busy ) {
365 0 : fd_h2_conn_t * conn = client->conn;
366 0 : ulong const frame_rx_lo_0 = client->frame_rx->lo_off;
367 0 : ulong const frame_rx_hi_0 = client->frame_rx->hi_off;
368 0 : ulong const frame_tx_lo_1 = client->frame_tx->lo_off;
369 0 : ulong const frame_tx_hi_1 = client->frame_tx->hi_off;
370 :
371 0 : int rx_err = fd_h2_rbuf_recvmsg( client->frame_rx, sock_fd, MSG_NOSIGNAL|MSG_DONTWAIT );
372 0 : if( FD_UNLIKELY( rx_err ) ) {
373 0 : FD_LOG_INFO(( "Disconnected: recvmsg error (%i-%s)", rx_err, fd_io_strerror( rx_err ) ));
374 0 : errno = rx_err;
375 0 : return -1;
376 0 : }
377 :
378 0 : if( FD_UNLIKELY( conn->flags ) ) fd_h2_tx_control( conn, client->frame_tx, &fd_grpc_client_h2_callbacks );
379 0 : fd_h2_rx( conn, client->frame_rx, client->frame_tx, client->frame_scratch, client->frame_scratch_max, &fd_grpc_client_h2_callbacks );
380 0 : if( FD_UNLIKELY( client->window_update_pending || client->request_stream ) ) {
381 0 : client->window_update_pending = 0;
382 0 : fd_grpc_client_request_continue( client ); /* credit or TX ring space may have freed */
383 0 : }
384 0 : fd_grpc_client_service_streams( client, now );
385 :
386 0 : int tx_err = fd_h2_rbuf_sendmsg( client->frame_tx, sock_fd, MSG_NOSIGNAL|MSG_DONTWAIT );
387 0 : if( FD_UNLIKELY( tx_err && tx_err!=EAGAIN ) ) {
388 0 : FD_LOG_WARNING(( "fd_h2_rbuf_sendmsg failed (%i-%s)", tx_err, fd_io_strerror( tx_err ) ));
389 0 : errno = tx_err;
390 0 : return -1;
391 0 : }
392 :
393 0 : ulong const frame_rx_lo_1 = client->frame_rx->lo_off;
394 0 : ulong const frame_rx_hi_1 = client->frame_rx->hi_off;
395 0 : ulong const frame_tx_lo_0 = client->frame_tx->lo_off;
396 0 : ulong const frame_tx_hi_0 = client->frame_tx->hi_off;
397 :
398 0 : client->metrics->stream_chunks_rx_bytes += frame_rx_hi_1 - frame_rx_hi_0;
399 0 : client->metrics->stream_chunks_tx_bytes += frame_tx_lo_0 - frame_tx_lo_1;
400 :
401 0 : if( frame_rx_lo_0!=frame_rx_lo_1 || frame_rx_hi_0!=frame_rx_hi_1 ||
402 0 : frame_tx_lo_0!=frame_tx_lo_1 || frame_tx_hi_0!=frame_tx_hi_1 ) {
403 0 : *charge_busy = 1;
404 0 : }
405 :
406 0 : return tx_err==EAGAIN ? 1 : 0;
407 0 : }
408 :
409 : int
410 : fd_grpc_client_tx_flush_socket( fd_grpc_client_t * client,
411 0 : int sock_fd ) {
412 0 : ulong const lo_0 = client->frame_tx->lo_off;
413 0 : int tx_err = fd_h2_rbuf_sendmsg( client->frame_tx, sock_fd, MSG_NOSIGNAL|MSG_DONTWAIT );
414 0 : if( FD_UNLIKELY( tx_err && tx_err!=EAGAIN ) ) {
415 0 : FD_LOG_WARNING(( "fd_h2_rbuf_sendmsg failed (%i-%s)", tx_err, fd_io_strerror( tx_err ) ));
416 0 : errno = tx_err;
417 0 : return -1;
418 0 : }
419 0 : client->metrics->stream_chunks_tx_bytes += client->frame_tx->lo_off - lo_0;
420 0 : return tx_err==EAGAIN ? 1 : 0;
421 0 : }
422 :
423 : int
424 : fd_grpc_client_rxtx_tls( fd_grpc_client_t * client,
425 : fd_tlsrec_conn_t * tls_conn,
426 : int sock_fd,
427 : long now,
428 0 : int * charge_busy ) {
429 0 : fd_h2_conn_t * conn = client->conn;
430 0 : fd_tlsrec_sock_t * sock = client->tls_sock;
431 :
432 : /* Busy means progress was made. A flush that moves nothing because
433 : the peer stopped reading is not progress: the caller would spin on
434 : send(2)/EAGAIN instead of waiting for the EPOLLOUT it armed. */
435 0 : ulong parked_0 = sock->tx_sz - sock->tx_off;
436 0 : if( FD_UNLIKELY( fd_tlsrec_sock_flush( sock, sock_fd )<0 ) ) return -1;
437 0 : if( parked_0 != sock->tx_sz - sock->tx_off ) *charge_busy = 1;
438 :
439 0 : ulong tcp_rx_sz;
440 0 : int rx_err = fd_tlsrec_sock_rx( sock, tls_conn, sock_fd, &tcp_rx_sz );
441 0 : if( FD_UNLIKELY( rx_err ) ) {
442 0 : if( rx_err==FD_TLSREC_SOCK_ERR_RECV || rx_err==FD_TLSREC_SOCK_ERR_SEND )
443 0 : FD_LOG_INFO(( "Disconnected: %s (%i-%s)", fd_tlsrec_sock_strerror( rx_err ), errno, fd_io_strerror( errno ) ));
444 0 : else
445 0 : FD_LOG_INFO(( "Disconnected: %s", fd_tlsrec_sock_strerror( rx_err ) ));
446 0 : return -1;
447 0 : }
448 0 : if( tcp_rx_sz ) *charge_busy = 1;
449 :
450 0 : if( FD_UNLIKELY( !fd_tlsrec_conn_is_ready( tls_conn ) ) ) {
451 0 : if( FD_UNLIKELY( fd_tlsrec_conn_is_failed( tls_conn ) ) ) {
452 0 : FD_LOG_WARNING(( "TLS handshake failed" ));
453 0 : return -1;
454 0 : }
455 0 : return 0;
456 0 : }
457 :
458 0 : if( FD_UNLIKELY( !client->h2_hs_done && !tls_conn->hs.cli.alpn_negotiated ) ) {
459 0 : FD_LOG_WARNING(( "TLS handshake failed: not a gRPC server (no h2 ALPN)" ));
460 0 : return -1;
461 0 : }
462 :
463 0 : ulong push_sz = fd_ulong_min( fd_tlsrec_sock_rx_avail( sock ),
464 0 : fd_h2_rbuf_free_sz( client->frame_rx ) );
465 0 : if( FD_LIKELY( push_sz ) ) {
466 0 : fd_h2_rbuf_push( client->frame_rx, fd_tlsrec_sock_rx_data( sock ), push_sz );
467 0 : fd_tlsrec_sock_rx_consume( sock, push_sz );
468 0 : client->metrics->stream_chunks_rx_bytes += push_sz;
469 0 : *charge_busy = 1;
470 0 : }
471 :
472 0 : if( FD_UNLIKELY( conn->flags ) ) fd_h2_tx_control( conn, client->frame_tx, &fd_grpc_client_h2_callbacks );
473 0 : fd_h2_rx( conn, client->frame_rx, client->frame_tx, client->frame_scratch, client->frame_scratch_max, &fd_grpc_client_h2_callbacks );
474 0 : if( FD_UNLIKELY( client->window_update_pending || client->request_stream ) ) {
475 0 : client->window_update_pending = 0;
476 0 : fd_grpc_client_request_continue( client ); /* credit or TX ring space may have freed */
477 0 : }
478 0 : fd_grpc_client_service_streams( client, now );
479 :
480 : /* HTTP/2 bytes wait behind parked ciphertext until EPOLLOUT */
481 0 : ulong tx_used = fd_h2_rbuf_used_sz( client->frame_tx );
482 0 : if( FD_LIKELY( tx_used && !fd_tlsrec_sock_tx_pending( sock ) ) ) {
483 0 : uchar plaintext[ FD_TLSREC_PLAINTEXT_MAX ];
484 0 : ulong pop_sz = fd_ulong_min( tx_used, sizeof(plaintext) );
485 0 : fd_h2_rbuf_pop_copy( client->frame_tx, plaintext, pop_sz );
486 :
487 0 : ulong consumed;
488 0 : int tx_err = fd_tlsrec_sock_tx( sock, tls_conn, sock_fd, plaintext, pop_sz, &consumed );
489 0 : if( FD_UNLIKELY( tx_err ) ) {
490 0 : FD_LOG_WARNING(( "fd_tlsrec_sock_tx failed: %s", fd_tlsrec_sock_strerror( tx_err ) ));
491 0 : return -1;
492 0 : }
493 0 : FD_CHECK_CRIT( consumed==pop_sz, "mismatched buffer sizes" );
494 :
495 0 : client->metrics->stream_chunks_tx_bytes += pop_sz;
496 0 : *charge_busy = 1;
497 0 : }
498 0 : return 0;
499 0 : }
500 :
501 : #endif /* FD_H2_HAS_SOCKETS */
502 :
503 : /* fd_grpc_client_request continue attempts to write a request data
504 : frame. */
505 :
506 : static int
507 0 : fd_grpc_client_request_continue1( fd_grpc_client_t * client ) {
508 0 : fd_grpc_h2_stream_t * stream = client->request_stream;
509 0 : fd_h2_stream_t * h2_stream = &stream->s;
510 0 : fd_h2_tx_op_copy( client->conn, h2_stream, client->frame_tx, client->request_tx_op );
511 0 : if( FD_UNLIKELY( client->request_tx_op->chunk_sz ) ) return 0;
512 0 : client->request_stream = NULL;
513 0 : if( FD_UNLIKELY( h2_stream->state != FD_H2_STREAM_STATE_CLOSING_TX ) ) return 0;
514 0 : client->metrics->stream_chunks_tx_cnt++;
515 : /* Request finished */
516 0 : client->callbacks->tx_complete( client->ctx, stream->request_ctx );
517 0 : return 1;
518 0 : }
519 :
520 : static int
521 0 : fd_grpc_client_request_continue( fd_grpc_client_t * client ) {
522 0 : if( FD_UNLIKELY( client->conn->flags & (FD_H2_CONN_FLAGS_DEAD|FD_H2_CONN_FLAGS_SEND_GOAWAY) ) ) return 0;
523 0 : if( FD_UNLIKELY( !client->request_stream ) ) return 0;
524 0 : if( FD_UNLIKELY( !client->request_tx_op->chunk_sz ) ) return 0;
525 0 : return fd_grpc_client_request_continue1( client );
526 0 : }
527 :
528 : int
529 0 : fd_grpc_client_is_connected( fd_grpc_client_t * client ) {
530 0 : if( FD_UNLIKELY( !client ) ) return 0;
531 0 : if( FD_UNLIKELY( client->conn->flags & FD_H2_CONN_FLAGS_DEAD ) ) return 0;
532 0 : if( FD_UNLIKELY( !client->h2_hs_done ) ) return 0;
533 0 : return 1;
534 0 : }
535 :
536 : int
537 0 : fd_grpc_client_request_is_blocked( fd_grpc_client_t * client ) {
538 0 : if( FD_UNLIKELY( !client ) ) return 1;
539 0 : if( FD_UNLIKELY( client->conn->flags & FD_H2_CONN_FLAGS_DEAD ) ) return 1;
540 0 : if( FD_UNLIKELY( !client->h2_hs_done ) ) return 1;
541 0 : if( FD_UNLIKELY( !fd_h2_rbuf_is_empty( client->frame_tx ) ) ) return 1;
542 0 : if( FD_UNLIKELY( client->request_tx_op->chunk_sz > 0UL ) ) return 1;
543 0 : if( FD_UNLIKELY( !fd_grpc_client_stream_acquire_is_safe( client ) ) ) return 1;
544 0 : return 0;
545 0 : }
546 :
547 : int
548 0 : fd_grpc_client_stream_send_is_blocked( fd_grpc_client_t * client ) {
549 0 : if( FD_UNLIKELY( !client ) ) return 1;
550 0 : if( FD_UNLIKELY( client->conn->flags & FD_H2_CONN_FLAGS_DEAD ) ) return 1;
551 0 : if( FD_UNLIKELY( !client->h2_hs_done ) ) return 1;
552 0 : if( FD_UNLIKELY( !fd_h2_rbuf_is_empty( client->frame_tx ) ) ) return 1;
553 0 : if( FD_UNLIKELY( client->request_tx_op->chunk_sz > 0UL ) ) return 1;
554 0 : return 0;
555 0 : }
556 :
557 : int
558 0 : fd_grpc_client_request_stream_busy( fd_grpc_client_t * client ) {
559 0 : return client->request_stream && client->request_tx_op->chunk_sz;
560 0 : }
561 :
562 : fd_grpc_h2_stream_t *
563 : fd_grpc_client_request_start(
564 : fd_grpc_client_t * client,
565 : char const * path,
566 : ulong path_len,
567 : ulong request_ctx,
568 : pb_msgdesc_t const * fields,
569 : void const * message,
570 : char const * auth_token,
571 : ulong auth_token_sz,
572 : int is_streaming
573 0 : ) {
574 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( client ) ) ) return NULL;
575 :
576 : /* Encode message */
577 0 : FD_TEST( client->nanopb_tx_max > sizeof(fd_grpc_hdr_t) );
578 0 : uchar * proto_buf = client->nanopb_tx + sizeof(fd_grpc_hdr_t);
579 0 : pb_ostream_t ostream = pb_ostream_from_buffer( proto_buf, client->nanopb_tx_max - sizeof(fd_grpc_hdr_t) );
580 0 : if( FD_UNLIKELY( !pb_encode( &ostream, fields, message ) ) ) {
581 0 : FD_LOG_WARNING(( "Failed to encode Protobuf message (%.*s). This is a bug (insufficient buffer space?)", (int)path_len, path ));
582 0 : return NULL;
583 0 : }
584 0 : ulong const serialized_sz = ostream.bytes_written;
585 :
586 : /* Create gRPC length prefix */
587 0 : fd_grpc_hdr_t hdr = {
588 0 : .compressed=0,
589 0 : .msg_sz=fd_uint_bswap( (uint)serialized_sz )
590 0 : };
591 0 : memcpy( client->nanopb_tx, &hdr, sizeof(fd_grpc_hdr_t) );
592 0 : ulong const payload_sz = serialized_sz + sizeof(fd_grpc_hdr_t);
593 :
594 : /* Allocate stream descriptor */
595 0 : fd_grpc_h2_stream_t * stream = fd_grpc_client_stream_acquire( client, request_ctx );
596 0 : uint const stream_id = stream->s.stream_id;
597 :
598 : /* Write HTTP/2 request headers */
599 0 : fd_h2_tx_prepare( client->conn, client->frame_tx, FD_H2_FRAME_TYPE_HEADERS, FD_H2_FLAG_END_HEADERS, stream_id );
600 0 : fd_grpc_req_hdrs_t req_meta = {
601 0 : .host = client->host,
602 0 : .host_len = client->host_len,
603 0 : .port = client->port,
604 0 : .path = path,
605 0 : .path_len = path_len,
606 0 : .https = 1, /* grpc_client assumes TLS encryption for now */
607 :
608 0 : .bearer_auth = auth_token,
609 0 : .bearer_auth_len = auth_token_sz
610 0 : };
611 0 : if( FD_UNLIKELY( !fd_grpc_h2_gen_request_hdrs(
612 0 : &req_meta,
613 0 : client->frame_tx,
614 0 : client->version,
615 0 : client->version_len
616 0 : ) ) ) {
617 0 : FD_LOG_WARNING(( "Failed to generate gRPC request headers (%.*s). This is a bug", (int)path_len, path ));
618 0 : fd_grpc_client_stream_release( client, stream );
619 0 : return NULL;
620 0 : }
621 0 : fd_h2_tx_commit( client->conn, client->frame_tx );
622 :
623 : /* Queue request payload for send
624 : (Protobuf message might have to be fragmented into multiple HTTP/2
625 : DATA frames if the client gets blocked)
626 : For streaming requests, don't set END_STREAM flag yet */
627 0 : uint flags = is_streaming ? 0U : FD_H2_FLAG_END_STREAM;
628 0 : fd_h2_tx_op_init( client->request_tx_op, client->nanopb_tx, payload_sz, flags );
629 0 : fd_grpc_client_request_continue1( client );
630 0 : client->metrics->requests_sent++;
631 0 : client->metrics->streams_active++;
632 :
633 0 : FD_LOG_DEBUG(( "gRPC request path=%.*s sz=%lu streaming=%d", (int)path_len, path, serialized_sz, is_streaming ));
634 :
635 0 : return stream;
636 0 : }
637 :
638 : fd_grpc_h2_stream_t *
639 : fd_grpc_client_request_start1(
640 : fd_grpc_client_t * client,
641 : char const * path,
642 : ulong path_len,
643 : ulong request_ctx,
644 : uchar const * protobuf,
645 : ulong protobuf_sz,
646 : char const * auth_token,
647 : ulong auth_token_sz,
648 : int is_streaming
649 0 : ) {
650 0 : if( FD_UNLIKELY( fd_grpc_client_request_is_blocked( client ) ) ) return NULL;
651 :
652 0 : int const headers_only = (protobuf==NULL);
653 0 : if( FD_UNLIKELY( headers_only && !is_streaming ) ) {
654 0 : FD_LOG_WARNING(( "headers-only request requires is_streaming (path %.*s). This is a bug", (int)path_len, path ));
655 0 : return NULL;
656 0 : }
657 :
658 : /* Validate protobuf size */
659 0 : FD_TEST( client->nanopb_tx_max > sizeof(fd_grpc_hdr_t) );
660 0 : ulong const max_proto_sz = client->nanopb_tx_max - sizeof(fd_grpc_hdr_t);
661 0 : if( FD_UNLIKELY( protobuf_sz > max_proto_sz ) ) {
662 0 : FD_LOG_WARNING(( "Protobuf message too large (%lu bytes) for path (%.*s). Max size is %lu bytes", protobuf_sz, (int)path_len, path, max_proto_sz ));
663 0 : return NULL;
664 0 : }
665 :
666 0 : if( FD_LIKELY( !headers_only ) ) {
667 : /* Copy protobuf to buffer after gRPC header */
668 0 : uchar * proto_buf = client->nanopb_tx + sizeof(fd_grpc_hdr_t);
669 0 : memcpy( proto_buf, protobuf, protobuf_sz );
670 :
671 : /* Create gRPC length prefix */
672 0 : fd_grpc_hdr_t hdr = {
673 0 : .compressed=0,
674 0 : .msg_sz=fd_uint_bswap( (uint)protobuf_sz )
675 0 : };
676 0 : memcpy( client->nanopb_tx, &hdr, sizeof(fd_grpc_hdr_t) );
677 0 : }
678 0 : ulong const payload_sz = protobuf_sz + sizeof(fd_grpc_hdr_t);
679 :
680 : /* Allocate stream descriptor */
681 0 : fd_grpc_h2_stream_t * stream = fd_grpc_client_stream_acquire( client, request_ctx );
682 0 : uint const stream_id = stream->s.stream_id;
683 :
684 : /* Write HTTP/2 request headers */
685 0 : fd_h2_tx_prepare( client->conn, client->frame_tx, FD_H2_FRAME_TYPE_HEADERS, FD_H2_FLAG_END_HEADERS, stream_id );
686 0 : fd_grpc_req_hdrs_t req_meta = {
687 0 : .host = client->host,
688 0 : .host_len = client->host_len,
689 0 : .port = client->port,
690 0 : .path = path,
691 0 : .path_len = path_len,
692 0 : .https = 1, /* grpc_client assumes TLS encryption for now */
693 :
694 0 : .bearer_auth = auth_token,
695 0 : .bearer_auth_len = auth_token_sz
696 0 : };
697 0 : if( FD_UNLIKELY( !fd_grpc_h2_gen_request_hdrs(
698 0 : &req_meta,
699 0 : client->frame_tx,
700 0 : client->version,
701 0 : client->version_len
702 0 : ) ) ) {
703 0 : FD_LOG_WARNING(( "Failed to generate gRPC request headers (%.*s). This is a bug", (int)path_len, path ));
704 0 : fd_grpc_client_stream_release( client, stream );
705 0 : return NULL;
706 0 : }
707 0 : fd_h2_tx_commit( client->conn, client->frame_tx );
708 :
709 0 : client->metrics->streams_active++;
710 :
711 0 : if( FD_UNLIKELY( headers_only ) ) {
712 0 : client->request_stream = NULL;
713 0 : return stream;
714 0 : }
715 :
716 : /* Queue request payload for send
717 : (Protobuf message might have to be fragmented into multiple HTTP/2
718 : DATA frames if the client gets blocked)
719 : For streaming requests, don't set END_STREAM flag yet */
720 0 : uint flags = is_streaming ? 0U : FD_H2_FLAG_END_STREAM;
721 0 : fd_h2_tx_op_init( client->request_tx_op, client->nanopb_tx, payload_sz, flags );
722 0 : fd_grpc_client_request_continue1( client );
723 0 : client->metrics->requests_sent++;
724 :
725 0 : FD_LOG_DEBUG(( "gRPC request path=%.*s sz=%lu streaming=%d", (int)path_len, path, protobuf_sz, is_streaming ));
726 :
727 0 : return stream;
728 0 : }
729 :
730 : int
731 : fd_grpc_client_stream_send_msg(
732 : fd_grpc_client_t * client,
733 : fd_grpc_h2_stream_t * stream,
734 : pb_msgdesc_t const * fields,
735 : void const * message
736 0 : ) {
737 0 : if( FD_UNLIKELY( !client || !stream ) ) return 0;
738 0 : if( FD_UNLIKELY( client->conn->flags & FD_H2_CONN_FLAGS_DEAD ) ) return 0;
739 0 : if( FD_UNLIKELY( !fd_h2_rbuf_is_empty( client->frame_tx ) ) ) return 0;
740 0 : if( FD_UNLIKELY( client->request_tx_op->chunk_sz > 0UL ) ) return 0;
741 0 : if( FD_UNLIKELY( client->request_stream != NULL && client->request_stream != stream ) ) return 0; /* Another stream has a request in progress */
742 0 : if( FD_UNLIKELY( stream->s.state!=FD_H2_STREAM_STATE_OPEN &&
743 0 : stream->s.state!=FD_H2_STREAM_STATE_CLOSING_RX ) ) return 0;
744 :
745 : /* Encode message */
746 0 : FD_TEST( client->nanopb_tx_max > sizeof(fd_grpc_hdr_t) );
747 0 : uchar * proto_buf = client->nanopb_tx + sizeof(fd_grpc_hdr_t);
748 0 : pb_ostream_t ostream = pb_ostream_from_buffer( proto_buf, client->nanopb_tx_max - sizeof(fd_grpc_hdr_t) );
749 0 : if( FD_UNLIKELY( !pb_encode( &ostream, fields, message ) ) ) {
750 0 : FD_LOG_WARNING(( "Failed to encode Protobuf message for stream_id=%u. This is a bug (insufficient buffer space?)", stream->s.stream_id ));
751 0 : return 0;
752 0 : }
753 0 : ulong const serialized_sz = ostream.bytes_written;
754 :
755 : /* Create gRPC length prefix */
756 0 : fd_grpc_hdr_t hdr = {
757 0 : .compressed=0,
758 0 : .msg_sz=fd_uint_bswap( (uint)serialized_sz )
759 0 : };
760 0 : memcpy( client->nanopb_tx, &hdr, sizeof(fd_grpc_hdr_t) );
761 0 : ulong const payload_sz = serialized_sz + sizeof(fd_grpc_hdr_t);
762 :
763 : /* Queue message payload for send (without END_STREAM flag) */
764 0 : client->request_stream = stream;
765 0 : fd_h2_tx_op_init( client->request_tx_op, client->nanopb_tx, payload_sz, 0U );
766 0 : fd_grpc_client_request_continue1( client );
767 :
768 0 : FD_LOG_DEBUG(( "gRPC stream_send_msg stream_id=%u sz=%lu", stream->s.stream_id, serialized_sz ));
769 :
770 0 : return 1;
771 0 : }
772 :
773 : int
774 : fd_grpc_client_stream_send_msg1(
775 : fd_grpc_client_t * client,
776 : fd_grpc_h2_stream_t * stream,
777 : uchar const * protobuf,
778 : ulong protobuf_sz
779 0 : ) {
780 0 : if( FD_UNLIKELY( !client || !stream ) ) return 0;
781 0 : if( FD_UNLIKELY( client->conn->flags & FD_H2_CONN_FLAGS_DEAD ) ) return 0;
782 0 : if( FD_UNLIKELY( !fd_h2_rbuf_is_empty( client->frame_tx ) ) ) return 0;
783 0 : if( FD_UNLIKELY( client->request_tx_op->chunk_sz > 0UL ) ) return 0;
784 0 : if( FD_UNLIKELY( client->request_stream != NULL && client->request_stream != stream ) ) return 0; /* Another stream has a request in progress */
785 0 : if( FD_UNLIKELY( stream->s.state!=FD_H2_STREAM_STATE_OPEN &&
786 0 : stream->s.state!=FD_H2_STREAM_STATE_CLOSING_RX ) ) return 0;
787 :
788 : /* Validate protobuf size */
789 0 : FD_TEST( client->nanopb_tx_max > sizeof(fd_grpc_hdr_t) );
790 0 : ulong const max_proto_sz = client->nanopb_tx_max - sizeof(fd_grpc_hdr_t);
791 0 : if( FD_UNLIKELY( protobuf_sz > max_proto_sz ) ) {
792 0 : FD_LOG_WARNING(( "Protobuf message too large (%lu bytes) for stream_id=%u. Max size is %lu bytes", protobuf_sz, stream->s.stream_id, max_proto_sz ));
793 0 : return 0;
794 0 : }
795 :
796 : /* Copy protobuf to buffer after gRPC header */
797 0 : uchar * proto_buf = client->nanopb_tx + sizeof(fd_grpc_hdr_t);
798 0 : if( protobuf!=proto_buf ) memcpy( proto_buf, protobuf, protobuf_sz );
799 :
800 : /* Create gRPC length prefix */
801 0 : fd_grpc_hdr_t hdr = {
802 0 : .compressed=0,
803 0 : .msg_sz=fd_uint_bswap( (uint)protobuf_sz )
804 0 : };
805 0 : memcpy( client->nanopb_tx, &hdr, sizeof(fd_grpc_hdr_t) );
806 0 : ulong const payload_sz = protobuf_sz + sizeof(fd_grpc_hdr_t);
807 :
808 : /* Queue message payload for send (without END_STREAM flag) */
809 0 : client->request_stream = stream;
810 0 : fd_h2_tx_op_init( client->request_tx_op, client->nanopb_tx, payload_sz, 0U );
811 0 : fd_grpc_client_request_continue1( client );
812 :
813 0 : FD_LOG_DEBUG(( "gRPC stream_send_msg stream_id=%u sz=%lu", stream->s.stream_id, protobuf_sz ));
814 :
815 0 : return 1;
816 0 : }
817 :
818 : int
819 : fd_grpc_client_stream_close(
820 : fd_grpc_client_t * client,
821 : fd_grpc_h2_stream_t * stream
822 0 : ) {
823 0 : if( FD_UNLIKELY( !client || !stream ) ) return 0;
824 0 : if( FD_UNLIKELY( client->conn->flags & FD_H2_CONN_FLAGS_DEAD ) ) return 0;
825 0 : if( FD_UNLIKELY( !fd_h2_rbuf_is_empty( client->frame_tx ) ) ) return 0;
826 0 : if( FD_UNLIKELY( client->request_stream != NULL ) ) return 0; /* Another request in progress */
827 0 : if( FD_UNLIKELY( stream->s.state!=FD_H2_STREAM_STATE_OPEN &&
828 0 : stream->s.state!=FD_H2_STREAM_STATE_CLOSING_RX ) ) return 0;
829 :
830 : /* Send empty DATA frame with END_STREAM flag to close the stream */
831 0 : fd_h2_stream_close_tx( &stream->s, client->conn );
832 0 : fd_h2_tx_prepare( client->conn, client->frame_tx, FD_H2_FRAME_TYPE_DATA, FD_H2_FLAG_END_STREAM, stream->s.stream_id );
833 0 : fd_h2_tx_commit( client->conn, client->frame_tx );
834 :
835 0 : FD_LOG_DEBUG(( "gRPC stream_close stream_id=%u", stream->s.stream_id ));
836 :
837 0 : return 1;
838 0 : }
839 :
840 : int
841 : fd_grpc_client_stream_cancel( fd_grpc_client_t * client,
842 0 : fd_grpc_h2_stream_t * stream ) {
843 0 : if( FD_UNLIKELY( fd_h2_rbuf_free_sz( client->frame_tx )<sizeof(fd_h2_rst_stream_t) ) ) return 0;
844 0 : fd_h2_stream_error( &stream->s, client->conn, client->frame_tx, FD_H2_ERR_CANCEL );
845 0 : fd_grpc_client_stream_release( client, stream );
846 0 : return 1;
847 0 : }
848 :
849 : void
850 : fd_grpc_client_deadline_set( fd_grpc_h2_stream_t * stream,
851 : int deadline_kind,
852 0 : long ts_nanos ) {
853 0 : switch( deadline_kind ) {
854 0 : case FD_GRPC_DEADLINE_HEADER:
855 0 : stream->header_deadline_nanos = ts_nanos;
856 0 : stream->has_header_deadline = 1;
857 0 : break;
858 0 : case FD_GRPC_DEADLINE_RX_END:
859 0 : stream->rx_end_deadline_nanos = ts_nanos;
860 0 : stream->has_rx_end_deadline = 1;
861 0 : break;
862 0 : }
863 0 : }
864 :
865 : /* Lookup stream by ID */
866 :
867 : static fd_h2_stream_t *
868 : fd_grpc_h2_stream_query( fd_h2_conn_t * conn,
869 0 : uint stream_id ) {
870 0 : fd_grpc_client_t * client = conn->ctx;
871 0 : for( ulong i=0UL; i<client->stream_cnt; i++ ) {
872 0 : if( client->stream_ids[ i ] == stream_id ) {
873 0 : return &client->streams[ i ]->s;
874 0 : }
875 0 : }
876 0 : return NULL;
877 0 : }
878 :
879 : static void
880 0 : fd_grpc_h2_conn_established( fd_h2_conn_t * conn ) {
881 0 : fd_grpc_client_t * client = conn->ctx;
882 0 : client->h2_hs_done = 1;
883 0 : client->callbacks->conn_established( client->ctx );
884 0 : }
885 :
886 : static void
887 : fd_grpc_h2_conn_final( fd_h2_conn_t * conn,
888 : uint h2_err,
889 0 : int closed_by ) {
890 0 : fd_grpc_client_t * client = conn->ctx;
891 0 : client->callbacks->conn_dead( client->ctx, h2_err, closed_by );
892 0 : }
893 :
894 : /* React to response data */
895 :
896 : void
897 : fd_grpc_h2_cb_headers(
898 : fd_h2_conn_t * conn,
899 : fd_h2_stream_t * h2_stream,
900 : void const * data,
901 : ulong data_sz,
902 : ulong flags
903 0 : ) {
904 0 : fd_grpc_h2_stream_t * stream = fd_grpc_h2_stream_upcast( h2_stream );
905 0 : fd_grpc_client_t * client = conn->ctx;
906 :
907 0 : int h2_status = fd_grpc_h2_read_response_hdrs( &stream->hdrs, client->matcher, data, data_sz );
908 0 : if( FD_UNLIKELY( h2_status!=FD_H2_SUCCESS ) ) {
909 : /* Failed to parse HTTP/2 headers */
910 0 : fd_h2_stream_error( h2_stream, conn, client->frame_tx, FD_H2_ERR_PROTOCOL );
911 0 : if( FD_LIKELY( !stream->rst_pending ) ) client->callbacks->rx_end( client->ctx, stream->request_ctx, &stream->hdrs ); /* invalidates stream->hdrs */
912 0 : fd_grpc_client_stream_release( client, stream );
913 0 : return;
914 0 : }
915 0 : if( FD_UNLIKELY( stream->rst_pending ) ) return; /* timed out */
916 :
917 0 : if( !( flags & FD_H2_FLAG_END_HEADERS ) ) return;
918 : /* END_STREAM may have arrived earlier in the block. Reset before
919 : callbacks consume the TX space reserved by fd_h2_rx. */
920 0 : int const rx_end = ( flags & FD_H2_FLAG_END_STREAM ) ||
921 0 : h2_stream->state==FD_H2_STREAM_STATE_CLOSING_RX || h2_stream->state==FD_H2_STREAM_STATE_CLOSED;
922 0 : if( rx_end ) fd_h2_stream_error( h2_stream, conn, client->frame_tx, FD_H2_SUCCESS );
923 :
924 0 : if( !stream->hdrs_received ) {
925 : /* Got initial response header */
926 0 : stream->hdrs_received = 1;
927 0 : stream->has_header_deadline = 0;
928 0 : if( FD_LIKELY( ( stream->hdrs.h2_status==200 ) &
929 0 : ( !!stream->hdrs.is_grpc_proto ) ) ) {
930 0 : client->callbacks->rx_start( client->ctx, stream->request_ctx );
931 0 : }
932 0 : }
933 :
934 0 : if( rx_end ) {
935 0 : client->callbacks->rx_end( client->ctx, stream->request_ctx, &stream->hdrs );
936 0 : fd_grpc_client_stream_release( client, stream );
937 0 : return;
938 0 : }
939 0 : }
940 :
941 : void
942 : fd_grpc_h2_cb_data(
943 : fd_h2_conn_t * conn,
944 : fd_h2_stream_t * h2_stream,
945 : void const * data,
946 : ulong data_sz,
947 : ulong flags
948 0 : ) {
949 0 : fd_grpc_client_t * client = conn->ctx;
950 0 : fd_grpc_h2_stream_t * stream = fd_grpc_h2_stream_upcast( h2_stream );
951 0 : if( FD_UNLIKELY( stream->rst_pending ) ) return; /* timed out */
952 0 : if( FD_UNLIKELY( ( stream->hdrs.h2_status!=200 ) |
953 0 : ( !stream->hdrs.is_grpc_proto ) ) ) {
954 0 : goto check_end_stream;
955 0 : }
956 :
957 0 : do {
958 :
959 : /* Read header bytes */
960 0 : if( stream->msg_buf_used < sizeof(fd_grpc_hdr_t) ) {
961 0 : ulong hdr_frag_sz = fd_ulong_min( sizeof(fd_grpc_hdr_t) - stream->msg_buf_used, data_sz );
962 0 : fd_memcpy( stream->msg_buf + stream->msg_buf_used, data, hdr_frag_sz );
963 0 : stream->msg_buf_used += hdr_frag_sz;
964 0 : data = (void const *)( (ulong)data + (ulong)hdr_frag_sz );
965 0 : data_sz -= hdr_frag_sz;
966 0 : if( FD_UNLIKELY( stream->msg_buf_used < sizeof(fd_grpc_hdr_t) ) ) goto check_end_stream;
967 :
968 : /* Header complete */
969 0 : stream->msg_sz = fd_uint_bswap( FD_LOAD( uint, (void *)( (ulong)stream->msg_buf+1 ) ) );
970 0 : if( FD_UNLIKELY( sizeof(fd_grpc_hdr_t) + stream->msg_sz > stream->msg_buf_max ) ) {
971 0 : FD_LOG_WARNING(( "Received oversized gRPC message (%lu bytes), killing request", stream->msg_sz ));
972 0 : fd_h2_stream_error( h2_stream, conn, client->frame_tx, FD_H2_ERR_INTERNAL );
973 0 : client->callbacks->rx_end( client->ctx, stream->request_ctx, &stream->hdrs );
974 0 : fd_grpc_client_stream_release( client, stream );
975 0 : return;
976 0 : }
977 0 : }
978 :
979 : /* Read payload bytes */
980 0 : ulong wmark = sizeof(fd_grpc_hdr_t) + stream->msg_sz;
981 0 : ulong chunk_sz = fd_ulong_min( stream->msg_buf_used+data_sz, wmark ) - stream->msg_buf_used;
982 0 : if( FD_UNLIKELY( chunk_sz>data_sz ) ) FD_LOG_CRIT(( "integer underflow" )); /* unreachable */
983 0 : fd_memcpy( stream->msg_buf + stream->msg_buf_used, data, chunk_sz );
984 0 : stream->msg_buf_used += chunk_sz;
985 0 : data = (void const *)( (ulong)data + (ulong)chunk_sz );
986 0 : data_sz -= chunk_sz;
987 :
988 0 : client->metrics->stream_chunks_rx_cnt++;
989 :
990 0 : if( stream->msg_buf_used >= wmark ) {
991 : /* Data complete */
992 0 : void const * msg_ptr = stream->msg_buf + sizeof(fd_grpc_hdr_t);
993 : /* The first wrapped callback has no END_STREAM flag, but the H2
994 : state already reflects the end of the carrying DATA frame. */
995 0 : if( FD_UNLIKELY( h2_stream->state==FD_H2_STREAM_STATE_CLOSING_RX ) ) fd_h2_stream_error( h2_stream, conn, client->frame_tx, FD_H2_SUCCESS );
996 0 : client->callbacks->rx_msg( client->ctx, msg_ptr, stream->msg_sz, stream->request_ctx );
997 0 : stream->msg_buf_used = 0UL;
998 0 : stream->msg_sz = 0UL;
999 0 : }
1000 :
1001 0 : } while( data_sz );
1002 :
1003 :
1004 : /* Check whether the server might want to end the stream in a DATA frame so no stream is leaked.
1005 : This shouldn't happen in gRPC as server responses should indicate end of stream in headers */
1006 0 : check_end_stream:
1007 0 : if( flags & FD_H2_FLAG_END_STREAM ) {
1008 : /* FIXME incomplete gRPC message */
1009 0 : if( FD_UNLIKELY( stream->msg_buf_used ) ) {
1010 0 : FD_LOG_WARNING(( "Received incomplete gRPC message" ));
1011 0 : }
1012 0 : fd_h2_stream_error( h2_stream, conn, client->frame_tx, FD_H2_SUCCESS );
1013 0 : client->callbacks->rx_end( client->ctx, stream->request_ctx, &stream->hdrs );
1014 0 : fd_grpc_client_stream_release( client, stream );
1015 0 : }
1016 0 : }
1017 :
1018 : /* Server might kill our request */
1019 :
1020 : static void
1021 : fd_grpc_h2_rst_stream( fd_h2_conn_t * conn,
1022 : fd_h2_stream_t * h2_stream,
1023 : uint error_code,
1024 0 : int closed_by ) {
1025 0 : fd_grpc_client_t * client = conn->ctx;
1026 0 : if( closed_by==1 ) {
1027 0 : FD_LOG_WARNING(( "server %s%.*s%s terminated request %s(%u-%s)%s",
1028 0 : fd_log_style_bold(), (int)client->host_len, client->host, fd_log_style_normal(),
1029 0 : fd_log_style_dim(), error_code, fd_h2_strerror( error_code ), fd_log_style_normal() ));
1030 0 : } else {
1031 0 : FD_LOG_WARNING(( "request to server %s%.*s%s failed %s(%u-%s)%s",
1032 0 : fd_log_style_bold(), (int)client->host_len, client->host, fd_log_style_normal(),
1033 0 : fd_log_style_dim(), error_code, fd_h2_strerror( error_code ), fd_log_style_normal() ));
1034 0 : }
1035 0 : fd_grpc_h2_stream_t * stream = fd_grpc_h2_stream_upcast( h2_stream );
1036 0 : if( FD_LIKELY( !stream->rst_pending ) ) client->callbacks->rx_end( client->ctx, stream->request_ctx, &stream->hdrs ); /* invalidates stream->hdrs */
1037 0 : fd_grpc_client_stream_release( client, stream );
1038 0 : }
1039 :
1040 : /* A HTTP/2 flow control change might unblock a queued request send op */
1041 :
1042 : void
1043 : fd_grpc_h2_window_update( fd_h2_conn_t * conn,
1044 0 : uint increment ) {
1045 0 : (void)increment;
1046 : /* Defer tx resumption to after fd_h2_rx: filling frame_tx mid-rx can
1047 : starve control-frame responses (SETTINGS ACK, PONG). */
1048 0 : ((fd_grpc_client_t *)conn->ctx)->window_update_pending = 1;
1049 0 : }
1050 :
1051 : void
1052 : fd_grpc_h2_stream_window_update( fd_h2_conn_t * conn,
1053 : fd_h2_stream_t * stream,
1054 0 : uint increment ) {
1055 0 : (void)increment;
1056 : /* Repay initial-window-shrink debt before the new credit counts. */
1057 0 : fd_grpc_h2_stream_t * st = fd_grpc_h2_stream_upcast( stream );
1058 0 : long repay = fd_long_min( st->tx_wnd_debt, (long)stream->tx_wnd );
1059 0 : if( FD_UNLIKELY( repay>0L ) ) {
1060 0 : stream->tx_wnd -= (uint)repay;
1061 0 : st->tx_wnd_debt -= repay;
1062 0 : }
1063 0 : ((fd_grpc_client_t *)conn->ctx)->window_update_pending = 1;
1064 0 : }
1065 :
1066 : void
1067 : fd_grpc_h2_initial_window_update( fd_h2_conn_t * conn,
1068 0 : long delta ) {
1069 0 : fd_grpc_client_t * client = conn->ctx;
1070 0 : for( uint i=0U; i<client->stream_cnt; i++ ) {
1071 0 : fd_grpc_h2_stream_t * stream = client->streams[ i ];
1072 0 : fd_h2_stream_t * s = &stream->s;
1073 : /* The effective window can go negative on a shrink below consumed
1074 : credit (RFC 9113 section 6.9.2). tx_wnd is unsigned, so carry
1075 : the deficit in tx_wnd_debt; WINDOW_UPDATEs repay it before
1076 : granting new credit. */
1077 0 : long wnd = (long)s->tx_wnd - stream->tx_wnd_debt + delta;
1078 0 : if( FD_UNLIKELY( wnd > 0x7fffffffL ) ) {
1079 0 : fd_h2_conn_error( conn, FD_H2_ERR_FLOW_CONTROL );
1080 0 : return;
1081 0 : }
1082 0 : s->tx_wnd = (uint)fd_long_max( wnd, 0L );
1083 0 : stream->tx_wnd_debt = fd_long_max( -wnd, 0L );
1084 0 : }
1085 : /* Do not continue the tx op here: this runs inside fd_h2_rx SETTINGS
1086 : processing, and filling frame_tx now can starve the SETTINGS ACK
1087 : push (conn death via FD_H2_ERR_INTERNAL). Defer to after rx. */
1088 0 : if( FD_UNLIKELY( delta>0L ) ) client->window_update_pending = 1;
1089 0 : }
1090 :
1091 : void
1092 0 : fd_grpc_h2_ping_ack( fd_h2_conn_t * conn ) {
1093 0 : fd_grpc_client_t * client = conn->ctx;
1094 0 : client->callbacks->ping_ack( client->ctx );
1095 0 : }
1096 :
1097 : fd_h2_rbuf_t *
1098 0 : fd_grpc_client_rbuf_tx( fd_grpc_client_t * client ) {
1099 0 : return client->frame_tx;
1100 0 : }
1101 :
1102 : fd_h2_rbuf_t *
1103 0 : fd_grpc_client_rbuf_rx( fd_grpc_client_t * client ) {
1104 0 : return client->frame_rx;
1105 0 : }
1106 :
1107 : fd_h2_conn_t *
1108 0 : fd_grpc_client_h2_conn( fd_grpc_client_t * client ) {
1109 0 : return client->conn;
1110 0 : }
1111 :
1112 : /* fd_grpc_client_h2_callbacks specifies h2->grpc_client callbacks.
1113 : Stored in .rodata for security. Must be kept in sync with fd_h2 to
1114 : avoid NULL pointers. */
1115 :
1116 : fd_h2_callbacks_t const fd_grpc_client_h2_callbacks = {
1117 : .stream_create = fd_h2_noop_stream_create,
1118 : .stream_query = fd_grpc_h2_stream_query,
1119 : .conn_established = fd_grpc_h2_conn_established,
1120 : .conn_final = fd_grpc_h2_conn_final,
1121 : .headers = fd_grpc_h2_cb_headers,
1122 : .data = fd_grpc_h2_cb_data,
1123 : .rst_stream = fd_grpc_h2_rst_stream,
1124 : .window_update = fd_grpc_h2_window_update,
1125 : .stream_window_update = fd_grpc_h2_stream_window_update,
1126 : .initial_window_update = fd_grpc_h2_initial_window_update,
1127 : .ping_ack = fd_grpc_h2_ping_ack,
1128 : };
|
