Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_ipecho_server.h"
3 : #include "fd_ipecho_server_port_check.h"
4 :
5 : #include "../../util/fd_util.h"
6 : #include "../../util/net/fd_ip4.h"
7 :
8 : #include <errno.h>
9 : #include <unistd.h>
10 : #include <poll.h>
11 : #include <sys/epoll.h>
12 : #include <sys/socket.h>
13 : #include <netinet/in.h>
14 :
15 0 : #define STATE_READING (0)
16 0 : #define STATE_WRITING (1)
17 :
18 0 : #define CLOSE_OK ( 0)
19 0 : #define CLOSE_EXPECTED_EOF (-1)
20 0 : #define CLOSE_PEER_RESET (-2)
21 0 : #define CLOSE_LARGE_REQUEST (-3)
22 0 : #define CLOSE_BAD_HEADER (-4)
23 0 : #define CLOSE_BAD_TRAILER (-5)
24 0 : #define CLOSE_BAD_LENGTH (-6)
25 0 : #define CLOSE_EVICTED (-7)
26 :
27 : struct fd_ipecho_server_connection {
28 : int state;
29 :
30 : uint ipv4;
31 :
32 : ushort parent;
33 :
34 : ulong request_bytes_read;
35 : uchar request_bytes[ 22UL ];
36 : ulong response_bytes_written;
37 : uchar response_bytes[ 27UL ];
38 : };
39 :
40 : typedef struct fd_ipecho_server_connection fd_ipecho_server_connection_t;
41 :
42 : #define POOL_NAME conn_pool
43 0 : #define POOL_T fd_ipecho_server_connection_t
44 : #define POOL_IDX_T ushort
45 0 : #define POOL_NEXT parent
46 : #include "../../util/tmpl/fd_pool.c"
47 :
48 : struct fd_ipecho_server {
49 : int sockfd;
50 :
51 : int epoll_fd;
52 :
53 : ushort shred_version;
54 :
55 : ulong evict_idx;
56 : ulong max_connection_cnt;
57 :
58 : fd_ipecho_server_connection_t * pool;
59 : struct pollfd * pollfds;
60 :
61 : fd_ipecho_server_port_check_t * port_check;
62 :
63 : fd_ipecho_server_metrics_t metrics[ 1 ];
64 :
65 : ulong magic;
66 : };
67 :
68 : FD_FN_CONST ulong
69 0 : fd_ipecho_server_align( void ) {
70 0 : return 128UL;
71 0 : }
72 :
73 : FD_FN_CONST ulong
74 0 : fd_ipecho_server_footprint( ulong max_connection_cnt ) {
75 0 : ulong l = FD_LAYOUT_INIT;
76 0 : l = FD_LAYOUT_APPEND( l, fd_ipecho_server_align(), sizeof(fd_ipecho_server_t) );
77 0 : l = FD_LAYOUT_APPEND( l, conn_pool_align(), conn_pool_footprint( max_connection_cnt ) );
78 0 : l = FD_LAYOUT_APPEND( l, alignof(struct pollfd), (1UL+max_connection_cnt)*sizeof(struct pollfd) );
79 0 : l = FD_LAYOUT_APPEND( l, fd_ipecho_server_port_check_align(), fd_ipecho_server_port_check_footprint() );
80 0 : return FD_LAYOUT_FINI( l, fd_ipecho_server_align() );
81 0 : }
82 :
83 : void *
84 : fd_ipecho_server_new( void * shmem,
85 0 : ulong max_connection_cnt ) {
86 0 : if( FD_UNLIKELY( !shmem ) ) {
87 0 : FD_LOG_WARNING(( "NULL shmem" ));
88 0 : return NULL;
89 0 : }
90 :
91 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_ipecho_server_align() ) ) ) {
92 0 : FD_LOG_WARNING(( "misaligned shmem" ));
93 0 : return NULL;
94 0 : }
95 :
96 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
97 0 : fd_ipecho_server_t * server = FD_SCRATCH_ALLOC_APPEND( l, fd_ipecho_server_align(), sizeof(fd_ipecho_server_t) );
98 0 : void * pool = FD_SCRATCH_ALLOC_APPEND( l, conn_pool_align(), conn_pool_footprint( max_connection_cnt ) );
99 0 : server->pollfds = FD_SCRATCH_ALLOC_APPEND( l, alignof(struct pollfd), (1UL+max_connection_cnt)*sizeof(struct pollfd) );
100 0 : void * port_check = FD_SCRATCH_ALLOC_APPEND( l, fd_ipecho_server_port_check_align(), fd_ipecho_server_port_check_footprint() );
101 :
102 0 : server->pool = conn_pool_join( conn_pool_new( pool, max_connection_cnt ) );
103 0 : FD_TEST( server->pool );
104 :
105 0 : server->port_check = fd_ipecho_server_port_check_join( fd_ipecho_server_port_check_new( port_check ) );
106 0 : FD_TEST( server->port_check );
107 :
108 0 : server->sockfd = -1;
109 0 : server->epoll_fd = -1;
110 :
111 0 : for( ulong i=0UL; i<max_connection_cnt; i++ ) {
112 0 : server->pollfds[ i ].fd = -1;
113 0 : server->pollfds[ i ].events = POLLIN;
114 0 : }
115 0 : server->pollfds[ max_connection_cnt ].fd = -1;
116 :
117 0 : server->evict_idx = 0UL;
118 0 : server->max_connection_cnt = max_connection_cnt;
119 :
120 0 : memset( &server->metrics, 0, sizeof(server->metrics) );
121 :
122 0 : FD_COMPILER_MFENCE();
123 0 : FD_VOLATILE( server->magic ) = FD_IPECHO_SERVER_MAGIC;
124 0 : FD_COMPILER_MFENCE();
125 :
126 0 : return server;
127 0 : }
128 :
129 : fd_ipecho_server_t *
130 0 : fd_ipecho_server_join( void * shipe ) {
131 0 : if( FD_UNLIKELY( !shipe ) ) {
132 0 : FD_LOG_WARNING(( "NULL shipe" ));
133 0 : return NULL;
134 0 : }
135 :
136 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shipe, fd_ipecho_server_align() ) ) ) {
137 0 : FD_LOG_WARNING(( "misaligned shipe" ));
138 0 : return NULL;
139 0 : }
140 :
141 0 : fd_ipecho_server_t * server = (fd_ipecho_server_t *)shipe;
142 :
143 0 : if( FD_UNLIKELY( server->magic!=FD_IPECHO_SERVER_MAGIC ) ) {
144 0 : FD_LOG_WARNING(( "bad magic" ));
145 0 : return NULL;
146 0 : }
147 :
148 0 : return server;
149 0 : }
150 :
151 : /* We handle epoll events for 3 kinds of sockets: already active
152 : clients, the listening socket, and the port check TCP connections.
153 :
154 : We use the high 32 bits of the data word to distinguish the socket
155 : type, and the low 32 bits to determine the connection.
156 : For active clients, the low 32 bits is the index into the
157 : server's connection pool. For the port check TCP sockets, the
158 : low 32 bits is the id returned by fd_ipecho_server_port_check_tcp. */
159 :
160 0 : #define EPOLL_DATA_FLAG_CONN (0UL<<32) /* active client */
161 0 : #define EPOLL_DATA_FLAG_LISTENER (1UL<<32) /* the listening socket */
162 0 : #define EPOLL_DATA_FLAG_PORT_CHECK (2UL<<32) /* port check TCP socket */
163 0 : #define EPOLL_DATA_FLAG_MASK (0xFFFFFFFFUL)
164 0 : #define EPOLL_DATA_FLAG( d ) ((d) & ~EPOLL_DATA_FLAG_MASK)
165 0 : #define EPOLL_DATA_CONN_IDX( d ) ((d) & EPOLL_DATA_FLAG_MASK)
166 :
167 : static void
168 0 : epoll_maybe_add_listener( fd_ipecho_server_t * server ) {
169 0 : if( FD_UNLIKELY( -1==server->sockfd ) ) return;
170 0 : if( FD_UNLIKELY( server->shred_version==0U ) ) return;
171 0 : struct epoll_event ev = { .events = EPOLLIN, .data.u64 = EPOLL_DATA_FLAG_LISTENER };
172 0 : if( FD_UNLIKELY( -1==epoll_ctl( server->epoll_fd, EPOLL_CTL_ADD, server->sockfd, &ev ) ) ) {
173 0 : if( FD_LIKELY( errno==EEXIST ) ) return;
174 0 : FD_LOG_ERR(( "epoll_ctl(ADD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
175 0 : }
176 0 : }
177 :
178 : void
179 : fd_ipecho_server_init( fd_ipecho_server_t * server,
180 : int epoll_fd,
181 : uint address,
182 : ushort port,
183 0 : ushort shred_version ) {
184 :
185 0 : FD_TEST( -1!=epoll_fd );
186 0 : server->epoll_fd = epoll_fd;
187 :
188 : /* If the shred version is 0 that means that the shred version has not
189 : been set yet. */
190 0 : server->shred_version = shred_version;
191 :
192 0 : server->sockfd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0 );
193 0 : if( FD_UNLIKELY( -1==server->sockfd ) ) FD_LOG_ERR(( "socket() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
194 :
195 0 : int optval = 1;
196 0 : if( FD_UNLIKELY( -1==setsockopt( server->sockfd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof( optval ) ) ) )
197 0 : FD_LOG_ERR(( "setsockopt failed (%i-%s)", errno, strerror( errno ) ));
198 :
199 0 : struct sockaddr_in addr = {
200 0 : .sin_family = AF_INET,
201 0 : .sin_port = fd_ushort_bswap( port ),
202 0 : .sin_addr.s_addr = address,
203 0 : };
204 :
205 0 : if( FD_UNLIKELY( -1==bind( server->sockfd, fd_type_pun( &addr ), sizeof( addr ) ) ) ) {
206 0 : FD_LOG_ERR(( "bind(%i,AF_INET," FD_IP4_ADDR_FMT ":%u) failed (%i-%s)",
207 0 : server->sockfd, FD_IP4_ADDR_FMT_ARGS( address ), port,
208 0 : errno, fd_io_strerror( errno ) ));
209 0 : }
210 0 : if( FD_UNLIKELY( -1==listen( server->sockfd, (int)server->max_connection_cnt ) ) ) FD_LOG_ERR(( "listen() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
211 :
212 0 : server->pollfds[ server->max_connection_cnt ] = (struct pollfd){ .fd = server->sockfd, .events = POLLIN, .revents = 0 };
213 0 : epoll_maybe_add_listener( server );
214 :
215 0 : fd_ipecho_server_port_check_init( server->port_check, address );
216 0 : }
217 :
218 : void
219 0 : fd_ipecho_server_fini( fd_ipecho_server_t * server ) {
220 0 : fd_ipecho_server_close_conns( server );
221 0 : if( FD_UNLIKELY( -1!=server->sockfd ) ) {
222 0 : FD_TEST( -1!=close( server->sockfd ) );
223 0 : server->sockfd = -1;
224 0 : server->pollfds[ server->max_connection_cnt ].fd = -1;
225 0 : }
226 0 : fd_ipecho_server_port_check_fini( server->port_check );
227 0 : }
228 :
229 : void
230 0 : fd_ipecho_server_close_conns( fd_ipecho_server_t * server ) {
231 0 : for( ulong i=0UL; i<server->max_connection_cnt; i++ ) {
232 0 : if( FD_UNLIKELY( -1!=server->pollfds[ i ].fd ) ) {
233 0 : FD_TEST( -1!=close( server->pollfds[ i ].fd ) );
234 0 : server->pollfds[ i ].fd = -1;
235 0 : conn_pool_ele_release( server->pool, &server->pool[ i ] );
236 0 : }
237 0 : }
238 0 : server->metrics->connection_cnt = 0UL;
239 0 : server->evict_idx = 0UL;
240 :
241 0 : fd_ipecho_server_port_check_close_all( server->port_check );
242 0 : }
243 :
244 : void
245 : fd_ipecho_server_set_shred_version( fd_ipecho_server_t * server,
246 0 : ushort shred_version ) {
247 0 : server->shred_version = shred_version;
248 0 : epoll_maybe_add_listener( server );
249 0 : }
250 :
251 : static inline int
252 0 : is_expected_network_error( int err ) {
253 0 : return
254 0 : err==ENETDOWN ||
255 0 : err==EPROTO ||
256 0 : err==ENOPROTOOPT ||
257 0 : err==EHOSTDOWN ||
258 0 : err==ENONET ||
259 0 : err==EHOSTUNREACH ||
260 0 : err==EOPNOTSUPP ||
261 0 : err==ENETUNREACH ||
262 0 : err==ETIMEDOUT ||
263 0 : err==ENETRESET ||
264 0 : err==ECONNABORTED ||
265 0 : err==ECONNRESET ||
266 0 : err==EPIPE ||
267 0 : err==EPERM || /* iptables */
268 0 : err==ENOMEM; /* net stack OOM */
269 0 : }
270 :
271 : static void
272 : close_conn( fd_ipecho_server_t * server,
273 : ulong conn_idx,
274 0 : int reason ) {
275 0 : (void)reason;
276 0 : FD_TEST( server->pollfds[ conn_idx ].fd!=-1 );
277 :
278 0 : if( FD_UNLIKELY( -1==close( server->pollfds[ conn_idx ].fd ) ) ) FD_LOG_ERR(( "close failed (%i-%s)", errno, strerror( errno ) ));
279 0 : server->pollfds[ conn_idx ].fd = -1;
280 0 : conn_pool_ele_release( server->pool, &server->pool[ conn_idx ] );
281 :
282 0 : FD_TEST( server->metrics->connection_cnt );
283 0 : server->metrics->connection_cnt--;
284 0 : if( FD_UNLIKELY( reason==CLOSE_OK ) ) server->metrics->connections_closed_ok++;
285 0 : else server->metrics->connections_closed_error++;
286 0 : }
287 :
288 : static void
289 : epoll_conn_add( fd_ipecho_server_t * server,
290 0 : ulong conn_idx ) {
291 0 : struct epoll_event ev = { .events = EPOLLIN, .data.u64 = EPOLL_DATA_FLAG_CONN|conn_idx };
292 0 : if( FD_UNLIKELY( -1==epoll_ctl( server->epoll_fd, EPOLL_CTL_ADD, server->pollfds[ conn_idx ].fd, &ev ) ) )
293 0 : FD_LOG_ERR(( "epoll_ctl(ADD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
294 0 : server->pollfds[ conn_idx ].events = POLLIN;
295 0 : }
296 :
297 : static void
298 : epoll_port_check_add( fd_ipecho_server_t * server,
299 0 : ulong check_id ) {
300 0 : int fd = fd_ipecho_server_port_check_conn_fd( server->port_check, check_id );
301 0 : struct epoll_event ev = { .events = EPOLLOUT, .data.u64 = EPOLL_DATA_FLAG_PORT_CHECK|check_id };
302 0 : if( FD_UNLIKELY( -1==epoll_ctl( server->epoll_fd, EPOLL_CTL_ADD, fd, &ev ) ) )
303 0 : FD_LOG_ERR(( "epoll_ctl(ADD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
304 0 : }
305 :
306 : static void
307 : epoll_update_out( fd_ipecho_server_t * server,
308 0 : ulong conn_idx ) {
309 0 : if( FD_UNLIKELY( -1==server->pollfds[ conn_idx ].fd ) ) return;
310 0 : short events = server->pool[ conn_idx ].state==STATE_WRITING ? POLLOUT : POLLIN;
311 0 : if( FD_LIKELY( server->pollfds[ conn_idx ].events==events ) ) return;
312 0 : struct epoll_event ev = {
313 0 : .events = ((events & POLLIN) ? EPOLLIN : 0U) | ((events & POLLOUT) ? EPOLLOUT : 0U),
314 0 : .data.u64 = EPOLL_DATA_FLAG_CONN|conn_idx,
315 0 : };
316 0 : if( FD_UNLIKELY( -1==epoll_ctl( server->epoll_fd, EPOLL_CTL_MOD, server->pollfds[ conn_idx ].fd, &ev ) ) )
317 0 : FD_LOG_ERR(( "epoll_ctl(MOD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
318 0 : server->pollfds[ conn_idx ].events = events;
319 0 : }
320 :
321 :
322 : static void
323 0 : accept_conns( fd_ipecho_server_t * server ) {
324 0 : for(;;) {
325 0 : struct sockaddr_in addr;
326 0 : socklen_t addr_len = sizeof(addr);
327 0 : int fd = accept4( server->pollfds[ server->max_connection_cnt ].fd, fd_type_pun( &addr ), &addr_len, SOCK_NONBLOCK|SOCK_CLOEXEC );
328 :
329 0 : if( FD_UNLIKELY( -1==fd ) ) {
330 0 : if( FD_LIKELY( EAGAIN==errno ) ) break;
331 0 : else if( FD_LIKELY( is_expected_network_error( errno ) ) ) continue;
332 0 : else FD_LOG_ERR(( "accept4() failed (%i-%s)", errno, strerror( errno ) ));
333 0 : }
334 :
335 0 : if( FD_UNLIKELY( !conn_pool_free( server->pool ) ) ) {
336 0 : close_conn( server, server->evict_idx, CLOSE_EVICTED );
337 0 : server->evict_idx = (server->evict_idx+1UL) % server->max_connection_cnt;
338 0 : }
339 0 : ulong conn_id = conn_pool_idx_acquire( server->pool );
340 :
341 0 : server->pollfds[ conn_id ].fd = fd;
342 0 : server->pollfds[ conn_id ].events = POLLIN;
343 0 : epoll_conn_add( server, conn_id );
344 0 : server->pool[ conn_id ].ipv4 = addr.sin_addr.s_addr;
345 0 : server->pool[ conn_id ].state = STATE_READING;
346 0 : server->pool[ conn_id ].request_bytes_read = 0UL;
347 0 : server->pool[ conn_id ].response_bytes_written = 0UL;
348 :
349 0 : server->metrics->connection_cnt++;
350 0 : }
351 0 : }
352 :
353 : static void
354 : read_conn( fd_ipecho_server_t * server,
355 : ulong conn_idx,
356 0 : long now ) {
357 0 : fd_ipecho_server_connection_t * conn = &server->pool[ conn_idx ];
358 :
359 0 : if( FD_UNLIKELY( conn->state!=STATE_READING ) ) {
360 0 : close_conn( server, conn_idx, CLOSE_EXPECTED_EOF );
361 0 : return;
362 0 : }
363 :
364 0 : long sz = read( server->pollfds[ conn_idx ].fd, conn->request_bytes+conn->request_bytes_read, sizeof(conn->request_bytes)-conn->request_bytes_read );
365 0 : if( FD_UNLIKELY( -1==sz && errno==EAGAIN ) ) return; /* No data to read, continue. */
366 0 : else if( -1==sz && is_expected_network_error( errno ) ) {
367 0 : close_conn( server, conn_idx, CLOSE_PEER_RESET );
368 0 : return;
369 0 : }
370 0 : else if( FD_UNLIKELY( -1==sz ) ) FD_LOG_ERR(( "read failed (%i-%s)", errno, strerror( errno ) )); /* Unexpected programmer error, abort */
371 :
372 0 : if( FD_UNLIKELY( !sz && conn->request_bytes_read!=21UL ) ) {
373 0 : close_conn( server, conn_idx, CLOSE_BAD_LENGTH );
374 0 : return;
375 0 : }
376 :
377 : /* New data was read... process it */
378 0 : server->metrics->bytes_read += (ulong)sz;
379 0 : conn->request_bytes_read += (ulong)sz;
380 0 : if( FD_UNLIKELY( conn->request_bytes_read==sizeof(conn->request_bytes) ) ) {
381 0 : close_conn( server, conn_idx, CLOSE_LARGE_REQUEST );
382 0 : return;
383 0 : }
384 :
385 0 : if( FD_UNLIKELY( conn->request_bytes_read<21UL ) ) return;
386 :
387 0 : if( FD_UNLIKELY( memcmp( conn->request_bytes, "\0\0\0\0", 4UL ) ) ) {
388 0 : close_conn( server, conn_idx, CLOSE_BAD_HEADER );
389 0 : return;
390 0 : }
391 :
392 0 : if( FD_UNLIKELY( conn->request_bytes[ 20UL ]!='\n' ) ) {
393 0 : close_conn( server, conn_idx, CLOSE_BAD_TRAILER );
394 0 : return;
395 0 : }
396 :
397 : /* Send a UDP datagram to each non-zero UDP port in the
398 : IpEchoServerMessage, and try to connect to each non-zero TCP port.
399 :
400 : These are the same semantics as Agave's IP echo server. If we
401 : don't do this Agave nodes without --no-port-check will fail to
402 : join our entrypoint.
403 :
404 : https://github.com/anza-xyz/agave/blob/v4.3.0/net-utils/src/ip_echo_server.rs#L132-L165 */
405 0 : for( ulong i=0UL; i<4UL; i++ ) {
406 : /* Safe because we read up to conn->request_bytes[19] here, and
407 : the checks above ensure that request_bytes is exactly 21 bytes. */
408 0 : ushort tcp_port = FD_LOAD( ushort, conn->request_bytes+4UL+2UL*i );
409 0 : ushort udp_port = FD_LOAD( ushort, conn->request_bytes+12UL+2UL*i );
410 :
411 : /* We expect most operators to have port checks disabled */
412 0 : if( FD_UNLIKELY( tcp_port ) ) {
413 0 : ulong check_id = fd_ipecho_server_port_check_tcp( server->port_check, conn->ipv4, tcp_port, now );
414 0 : if( FD_LIKELY( check_id!=ULONG_MAX ) ) epoll_port_check_add( server, check_id );
415 0 : }
416 0 : if( FD_UNLIKELY( udp_port ) ) {
417 0 : fd_ipecho_server_port_check_udp( server->port_check, conn->ipv4, udp_port );
418 0 : }
419 0 : }
420 :
421 0 : uchar response[ 27UL ] = {
422 0 : 0, 0, 0, 0, /* Magic */
423 0 : 0, 0, 0, 0, /* IP address variant */
424 0 : 0, 0, 0, 0, /* IP address */
425 0 : 1, /* Shred version option variant */
426 0 : 0, 0, /* Shred version */
427 0 : 0, /* [...] 12 bytes of trailing garbage, as in Agave */
428 0 : };
429 :
430 0 : FD_STORE( uint, response+8UL, conn->ipv4 );
431 0 : FD_STORE( ushort, response+13UL, server->shred_version );
432 :
433 : /* Now have a complete request ... buffer response */
434 0 : conn->state = STATE_WRITING;
435 0 : conn->response_bytes_written = 0UL;
436 0 : memcpy( conn->response_bytes, response, sizeof(response) );
437 0 : }
438 :
439 : static void
440 : write_conn( fd_ipecho_server_t * server,
441 0 : ulong conn_idx ) {
442 0 : fd_ipecho_server_connection_t * conn = &server->pool[ conn_idx ];
443 :
444 0 : if( FD_LIKELY( conn->state==STATE_READING ) ) return;
445 :
446 0 : long sz = sendto( server->pollfds[ conn_idx ].fd, conn->response_bytes+conn->response_bytes_written, sizeof(conn->response_bytes)-conn->response_bytes_written, MSG_NOSIGNAL, NULL, 0 );
447 0 : if( FD_UNLIKELY( -1==sz && errno==EAGAIN ) ) return; /* No data was written, continue. */
448 0 : if( FD_UNLIKELY( -1==sz && is_expected_network_error( errno ) ) ) {
449 0 : close_conn( server, conn_idx, CLOSE_PEER_RESET );
450 0 : return;
451 0 : }
452 0 : if( FD_UNLIKELY( -1==sz ) ) FD_LOG_ERR(( "write failed (%i-%s)", errno, strerror( errno ) )); /* Unexpected programmer error, abort */
453 :
454 0 : server->metrics->bytes_written += (ulong)sz;
455 0 : conn->response_bytes_written += (ulong)sz;
456 0 : if( FD_UNLIKELY( conn->response_bytes_written<sizeof(conn->response_bytes) ) ) return;
457 :
458 0 : close_conn( server, conn_idx, CLOSE_OK );
459 0 : }
460 :
461 :
462 : int
463 : fd_ipecho_server_epoll_poll( fd_ipecho_server_t * server,
464 : long now,
465 0 : int * charge_busy ) {
466 0 : FD_TEST( -1!=server->epoll_fd );
467 :
468 0 : if( FD_UNLIKELY( server->shred_version==0U ) ) return 0;
469 :
470 0 : struct epoll_event evs[ 64 ];
471 0 : int nfds = epoll_pwait( server->epoll_fd, evs, 64, 0, NULL );
472 0 : if( FD_UNLIKELY( -1==nfds ) ) {
473 0 : if( FD_LIKELY( errno==EINTR ) ) return 0;
474 0 : FD_LOG_ERR(( "epoll_pwait failed (%i-%s)", errno, fd_io_strerror( errno ) ));
475 0 : }
476 0 : if( FD_LIKELY( !nfds ) ) return 0;
477 :
478 0 : *charge_busy = 1;
479 0 : for( int i=0; i<nfds; i++ ) {
480 0 : ulong data = evs[ i ].data.u64;
481 0 : switch( EPOLL_DATA_FLAG( data ) ) {
482 0 : case EPOLL_DATA_FLAG_LISTENER:
483 : /* New connections to the ipecho server */
484 0 : accept_conns( server );
485 0 : break;
486 0 : case EPOLL_DATA_FLAG_PORT_CHECK:
487 : /* Port check TCP sockets (handshake succeeded or errored) */
488 0 : fd_ipecho_server_port_check_handle_event( server->port_check, EPOLL_DATA_CONN_IDX( data ) );
489 0 : break;
490 0 : case EPOLL_DATA_FLAG_CONN: {
491 : /* Existing non-port-check connections to the ipecho server */
492 0 : ulong conn_idx = EPOLL_DATA_CONN_IDX( data );
493 0 : if( FD_UNLIKELY( -1==server->pollfds[ conn_idx ].fd ) ) break;
494 0 : if( FD_LIKELY( evs[ i ].events & (EPOLLIN|EPOLLHUP|EPOLLERR) ) ) read_conn( server, conn_idx, now );
495 0 : if( FD_UNLIKELY( -1==server->pollfds[ conn_idx ].fd ) ) break;
496 0 : if( FD_LIKELY( evs[ i ].events & EPOLLOUT ) ) write_conn( server, conn_idx );
497 0 : if( FD_UNLIKELY( -1==server->pollfds[ conn_idx ].fd ) ) break;
498 0 : epoll_update_out( server, conn_idx );
499 0 : break;
500 0 : }
501 0 : default:
502 0 : FD_LOG_ERR(( "unexpected epoll data flag 0x%lx", EPOLL_DATA_FLAG( data ) ));
503 0 : }
504 0 : }
505 :
506 0 : return 1;
507 0 : }
508 :
509 : fd_ipecho_server_metrics_t *
510 0 : fd_ipecho_server_metrics( fd_ipecho_server_t * server ) {
511 0 : return server->metrics;
512 0 : }
513 :
514 : int
515 0 : fd_ipecho_server_sockfd( fd_ipecho_server_t * server ) {
516 0 : return server->sockfd;
517 0 : }
518 :
519 : fd_ipecho_server_port_check_t *
520 0 : fd_ipecho_server_port_check( fd_ipecho_server_t * server ) {
521 0 : return server->port_check;
522 0 : }
|
