Line data Source code
1 : #include "fd_genesis_client_private.h"
2 :
3 : #include "../../third_party/picohttpparser/picohttpparser.h"
4 : #include "../../util/fd_util.h"
5 : #include "../../waltz/http/fd_http.h"
6 : #include "../../ballet/sha256/fd_sha256.h"
7 :
8 : #include <errno.h>
9 : #include <netinet/in.h>
10 : #include <string.h>
11 : #include <strings.h>
12 : #include <sys/epoll.h>
13 : #include <sys/socket.h>
14 : #include <unistd.h>
15 : #include <stdlib.h>
16 :
17 : FD_FN_CONST ulong
18 0 : fd_genesis_client_align( void ) {
19 0 : return alignof(fd_genesis_client_t);
20 0 : }
21 :
22 : FD_FN_CONST ulong
23 0 : fd_genesis_client_footprint( void ) {
24 0 : return sizeof(fd_genesis_client_t);
25 0 : }
26 :
27 : void *
28 0 : fd_genesis_client_new( void * shmem ) {
29 0 : fd_genesis_client_t * gen = (fd_genesis_client_t *)shmem;
30 :
31 0 : if( FD_UNLIKELY( !shmem ) ) {
32 0 : FD_LOG_WARNING(( "NULL shmem" ));
33 0 : return NULL;
34 0 : }
35 :
36 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_genesis_client_align() ) ) ) {
37 0 : FD_LOG_WARNING(( "misaligned shmem" ));
38 0 : return NULL;
39 0 : }
40 :
41 0 : FD_COMPILER_MFENCE();
42 0 : FD_VOLATILE( gen->magic ) = FD_GENESIS_CLIENT_MAGIC;
43 0 : FD_COMPILER_MFENCE();
44 :
45 0 : return (void *)gen;
46 0 : }
47 :
48 : fd_genesis_client_t *
49 0 : fd_genesis_client_join( void * shgen ) {
50 0 : if( FD_UNLIKELY( !shgen ) ) {
51 0 : FD_LOG_WARNING(( "NULL shgen" ));
52 0 : return NULL;
53 0 : }
54 :
55 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shgen, fd_genesis_client_align() ) ) ) {
56 0 : FD_LOG_WARNING(( "misaligned shgen" ));
57 0 : return NULL;
58 0 : }
59 :
60 0 : fd_genesis_client_t * gen = (fd_genesis_client_t *)shgen;
61 :
62 0 : if( FD_UNLIKELY( gen->magic!=FD_GENESIS_CLIENT_MAGIC ) ) {
63 0 : FD_LOG_WARNING(( "bad magic" ));
64 0 : return NULL;
65 0 : }
66 :
67 0 : return gen;
68 0 : }
69 :
70 : void
71 : fd_genesis_client_init( fd_genesis_client_t * client,
72 : fd_ip4_port_t const * servers,
73 : ulong servers_len,
74 0 : int epoll_fd ) {
75 0 : FD_TEST( servers_len<=FD_TOPO_GOSSIP_ENTRYPOINTS_MAX );
76 0 : FD_TEST( epoll_fd!=-1 );
77 :
78 0 : client->epoll_fd = epoll_fd;
79 :
80 0 : ulong peer_cnt = 0UL;
81 :
82 0 : for( ulong i=0UL; i<servers_len; i++ ) {
83 0 : fd_ip4_port_t server = servers[ i ];
84 0 : server.port = fd_ushort_bswap( 8899 ); // TODO: SPECIFY IN CONFIG
85 :
86 0 : int sockfd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0 );
87 0 : if( FD_UNLIKELY( -1==sockfd ) ) FD_LOG_ERR(( "socket() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
88 :
89 0 : struct sockaddr_in addr = {
90 0 : .sin_family = AF_INET,
91 0 : .sin_port = server.port,
92 0 : .sin_addr = { .s_addr = server.addr }
93 0 : };
94 :
95 0 : if( FD_UNLIKELY( -1==connect( sockfd, fd_type_pun( &addr ), sizeof(addr) ) && errno!=EINPROGRESS ) ) {
96 0 : if( FD_UNLIKELY( -1==close( sockfd ) ) ) FD_LOG_ERR(( "close() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
97 0 : continue;
98 0 : }
99 :
100 0 : struct epoll_event ev = { .events = EPOLLIN|EPOLLOUT, .data.fd = sockfd };
101 0 : if( FD_UNLIKELY( -1==epoll_ctl( epoll_fd, EPOLL_CTL_ADD, sockfd, &ev ) ) ) FD_LOG_ERR(( "epoll_ctl(ADD) failed (%d-%s)", errno, fd_io_strerror( errno ) ));
102 :
103 0 : client->pollfds[ peer_cnt ] = (struct pollfd){
104 0 : .fd = sockfd,
105 0 : .events = POLLIN | POLLOUT,
106 0 : .revents = 0
107 0 : };
108 0 : client->peers[ peer_cnt ].addr = server;
109 0 : client->peers[ peer_cnt ].writing = 1;
110 0 : client->peers[ peer_cnt ].request_bytes_sent = 0UL;
111 0 : client->peers[ peer_cnt ].response_bytes_read = 0UL;
112 0 : peer_cnt++;
113 0 : }
114 :
115 0 : for( ulong i=peer_cnt; i<FD_TOPO_GOSSIP_ENTRYPOINTS_MAX; i++ ) client->pollfds[ i ].fd = -1;
116 :
117 0 : client->peer_cnt = peer_cnt;
118 0 : client->remaining_peer_cnt = peer_cnt;
119 0 : client->start_time_nanos = LONG_MAX;
120 0 : }
121 :
122 : static void
123 : close_one( fd_genesis_client_t * client,
124 0 : ulong idx ) {
125 0 : if( FD_UNLIKELY( -1==epoll_ctl( client->epoll_fd, EPOLL_CTL_DEL, client->pollfds[ idx ].fd, NULL ) ) ) FD_LOG_ERR(( "epoll_ctl(DEL) failed (%d-%s)", errno, fd_io_strerror( errno ) ));
126 :
127 0 : if( FD_UNLIKELY( -1==close( client->pollfds[ idx ].fd ) ) ) FD_LOG_ERR(( "close() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
128 0 : client->pollfds[ idx ].fd = -1;
129 0 : client->remaining_peer_cnt--;
130 0 : }
131 :
132 : static void
133 0 : close_all( fd_genesis_client_t * client ) {
134 0 : for( ulong i=0UL; i<client->peer_cnt; i++ ) {
135 0 : if( FD_UNLIKELY( -1==client->pollfds[ i ].fd ) ) continue;
136 0 : close_one( client, i );
137 0 : }
138 0 : }
139 :
140 : static void
141 : write_conn( fd_genesis_client_t * client,
142 0 : ulong conn_idx ) {
143 0 : fd_genesis_client_peer_t * peer = &client->peers[ conn_idx ];
144 :
145 0 : if( FD_LIKELY( !peer->writing ) ) return;
146 :
147 0 : char request[ 1024UL ];
148 0 : ulong request_sz = 0UL;
149 0 : FD_TEST( fd_cstr_printf_check( request, sizeof(request), &request_sz, "GET /genesis.tar.bz2 HTTP/1.1\r\n"
150 0 : "Cache-Control: no-cache\r\n"
151 0 : "Connection: keep-alive\r\n"
152 0 : "Pragma: no-cache\r\n"
153 0 : "User-Agent: Firedancer\r\n"
154 0 : "Host: " FD_IP4_ADDR_FMT ":%hu\r\n\r\n",
155 0 : FD_IP4_ADDR_FMT_ARGS( peer->addr.addr ), fd_ushort_bswap( peer->addr.port ) ) );
156 0 : FD_TEST( request_sz > peer->request_bytes_sent );
157 :
158 0 : long written = sendto( client->pollfds[ conn_idx ].fd,
159 0 : request+peer->request_bytes_sent,
160 0 : request_sz-peer->request_bytes_sent,
161 0 : MSG_NOSIGNAL,
162 0 : NULL,
163 0 : 0 );
164 0 : if( FD_UNLIKELY( -1==written && errno==EAGAIN ) ) return; /* No data was written, continue. */
165 0 : else if( FD_UNLIKELY( -1==written ) ) {
166 0 : close_one( client, conn_idx );
167 0 : return;
168 0 : }
169 :
170 0 : peer->request_bytes_sent += (ulong)written;
171 0 : if( FD_UNLIKELY( peer->request_bytes_sent==request_sz ) ) {
172 0 : peer->writing = 0;
173 0 : peer->response_bytes_read = 0UL;
174 : /* A connected socket is always writable so drop EPOLLOUT or the
175 : waker refires on every rearm while we wait for the response */
176 0 : client->pollfds[ conn_idx ].events = POLLIN;
177 0 : struct epoll_event ev = { .events = EPOLLIN, .data.fd = client->pollfds[ conn_idx ].fd };
178 0 : if( FD_UNLIKELY( -1==epoll_ctl( client->epoll_fd, EPOLL_CTL_MOD, client->pollfds[ conn_idx ].fd, &ev ) ) ) FD_LOG_ERR(( "epoll_ctl(MOD) failed (%d-%s)", errno, fd_io_strerror( errno ) ));
179 0 : }
180 0 : }
181 :
182 : static ulong
183 : rpc_phr_content_length( struct phr_header * headers,
184 0 : ulong num_headers ) {
185 0 : for( ulong i=0UL; i<num_headers; i++ ) {
186 0 : if( FD_LIKELY( headers[i].name_len!=14UL ) ) continue;
187 0 : if( FD_LIKELY( strncasecmp( headers[i].name, "Content-Length", 14UL ) ) ) continue;
188 0 : ulong content_length = 0UL;
189 0 : if( FD_UNLIKELY( fd_http_parse_content_len( headers[i].value, (ulong)headers[i].value_len, &content_length ) ) ) return ULONG_MAX;
190 0 : if( FD_UNLIKELY( content_length>UINT_MAX ) ) return ULONG_MAX; /* prevent overflow */
191 0 : return content_length;
192 0 : }
193 0 : return ULONG_MAX;
194 0 : }
195 :
196 : static int
197 : read_conn( fd_genesis_client_t * client,
198 : ulong conn_idx,
199 : uchar ** buffer,
200 0 : ulong * buffer_sz ) {
201 0 : fd_genesis_client_peer_t * peer = &client->peers[ conn_idx ];
202 :
203 0 : if( FD_UNLIKELY( peer->writing ) ) return 1;
204 0 : long read = recvfrom( client->pollfds[ conn_idx ].fd,
205 0 : peer->response+peer->response_bytes_read,
206 0 : sizeof(peer->response)-peer->response_bytes_read,
207 0 : 0,
208 0 : NULL,
209 0 : NULL );
210 0 : if( FD_UNLIKELY( -1==read && (errno==EAGAIN || errno==EINTR) ) ) return 1;
211 0 : else if( FD_UNLIKELY( -1==read ) ) {
212 0 : close_one( client, conn_idx );
213 0 : return 1;
214 0 : }
215 0 : if( FD_UNLIKELY( !read ) ) {
216 0 : close_one( client, conn_idx );
217 0 : return 1;
218 0 : }
219 :
220 0 : peer->response_bytes_read += (ulong)read;
221 :
222 0 : int minor_version;
223 0 : int status;
224 0 : const char * message;
225 0 : ulong message_len;
226 0 : struct phr_header headers[ 32 ];
227 0 : ulong num_headers = 32UL;
228 0 : int len = phr_parse_response( (char*)peer->response, peer->response_bytes_read,
229 0 : &minor_version, &status, &message, &message_len,
230 0 : headers, &num_headers, 0L );
231 0 : if( FD_UNLIKELY( -1==len ) ) {
232 0 : close_one( client, conn_idx );
233 0 : return 1;
234 0 : } else if( FD_UNLIKELY( -2==len ) ) {
235 0 : return 1;
236 0 : }
237 :
238 0 : if( FD_UNLIKELY( status!=200 ) ) {
239 0 : close_one( client, conn_idx );
240 0 : return 1;
241 0 : }
242 :
243 0 : ulong content_length = rpc_phr_content_length( headers, num_headers );
244 0 : if( FD_UNLIKELY( content_length==ULONG_MAX ) ) {
245 0 : close_one( client, conn_idx );
246 0 : return 1;
247 0 : }
248 0 : if( FD_UNLIKELY( content_length+(ulong)len>sizeof(peer->response) ) ) {
249 0 : close_one( client, conn_idx );
250 0 : return 1;
251 0 : }
252 0 : if( FD_LIKELY( content_length+(ulong)len>peer->response_bytes_read ) ) {
253 0 : return 1;
254 0 : }
255 :
256 0 : *buffer_sz = content_length;
257 0 : *buffer = peer->response + (ulong)len;
258 :
259 0 : uchar hash[ 32UL ] = {0};
260 0 : fd_sha256_hash( *buffer, *buffer_sz, hash );
261 :
262 0 : return 0;
263 0 : }
264 :
265 : int
266 : fd_genesis_client_poll( fd_genesis_client_t * client,
267 : long now,
268 : fd_ip4_port_t * peer,
269 : uchar ** buffer,
270 : ulong * buffer_sz,
271 0 : int * charge_busy ) {
272 0 : if( FD_UNLIKELY( !client->remaining_peer_cnt ) ) return -1;
273 0 : if( FD_UNLIKELY( LONG_MAX==client->start_time_nanos ) ) client->start_time_nanos = now;
274 0 : if( FD_UNLIKELY( now>fd_genesis_client_deadline_nanos( client ) ) ) {
275 0 : close_all( client );
276 0 : return -1;
277 0 : }
278 :
279 0 : int nfds = fd_syscall_poll( client->pollfds, (uint)client->peer_cnt, 0 );
280 0 : if( FD_UNLIKELY( 0==nfds ) ) return 1;
281 0 : else if( FD_UNLIKELY( -1==nfds && errno==EINTR ) ) return 1;
282 0 : else if( FD_UNLIKELY( -1==nfds ) ) FD_LOG_ERR(( "poll() failed (%i-%s)", errno, strerror( errno ) ));
283 :
284 0 : for( ulong i=0UL; i<FD_TOPO_GOSSIP_ENTRYPOINTS_MAX; i++ ) {
285 0 : if( FD_UNLIKELY( -1==client->pollfds[ i ].fd ) ) continue;
286 :
287 0 : if( FD_LIKELY( client->pollfds[ i ].revents & POLLOUT ) ) {
288 0 : if( FD_UNLIKELY( client->peers[ i ].writing ) ) *charge_busy = 1;
289 0 : write_conn( client, i );
290 0 : }
291 0 : if( FD_UNLIKELY( -1==client->pollfds[ i ].fd ) ) continue;
292 0 : if( FD_LIKELY( client->pollfds[ i ].revents & POLLIN ) ) {
293 0 : if( FD_UNLIKELY( !client->peers[ i ].writing ) ) *charge_busy = 1;
294 0 : if( FD_LIKELY( !read_conn( client, i, buffer, buffer_sz ) ) ) {
295 0 : close_all( client );
296 0 : *peer = client->peers[ i ].addr;
297 0 : return 0;
298 0 : }
299 0 : }
300 0 : }
301 :
302 0 : return 1;
303 0 : }
304 :
305 : struct pollfd const *
306 0 : fd_genesis_client_get_pollfds( fd_genesis_client_t * client ) {
307 0 : return client->pollfds;
308 0 : }
309 :
310 : long
311 0 : fd_genesis_client_deadline_nanos( fd_genesis_client_t const * client ) {
312 0 : if( FD_UNLIKELY( LONG_MAX==client->start_time_nanos ) ) return LONG_MAX;
313 0 : return client->start_time_nanos+20L*1000L*1000L*1000L;
314 0 : }
|
