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