Line data Source code
1 : #define _GNU_SOURCE /* ppoll */
2 : #include "fd_ssping.h"
3 : #include "fd_sspeer_selector.h"
4 :
5 : #include "../../../util/fd_util.h"
6 : #include "../../../util/bits/fd_bits.h"
7 : #include "../../../util/log/fd_log.h"
8 :
9 : #include <errno.h>
10 : #include <fcntl.h>
11 : #include <unistd.h>
12 : #include <sys/epoll.h>
13 : #include <sys/socket.h>
14 : #include <netinet/in.h>
15 : #include <netinet/tcp.h>
16 : #include <poll.h>
17 :
18 630 : #define PEER_STATE_UNPINGED 0
19 0 : #define PEER_STATE_PINGED 1
20 0 : #define PEER_STATE_VALID 2
21 0 : #define PEER_STATE_REFRESHING 3
22 504 : #define PEER_STATE_INVALID 4
23 :
24 0 : #define PEER_DEADLINE_NANOS_PING (1L*1000L*1000L*1000L) /* 1 second */
25 0 : #define PEER_DEADLINE_NANOS_VALID (2L*60L*1000L*1000L*1000L) /* 2 minutes */
26 162 : #define PEER_DEADLINE_NANOS_INVALID (5L*60L*1000L*1000L*1000L) /* 5 minutes */
27 :
28 72 : #define PING_BURST_MAX (16UL) /* Limit how many pings we can burst at once. */
29 :
30 : struct fd_ssping_peer {
31 : ulong refcnt;
32 : fd_ip4_port_t addr;
33 :
34 : struct {
35 : ulong next;
36 : } pool;
37 :
38 : struct {
39 : ulong next;
40 : ulong prev;
41 : } map;
42 :
43 : struct {
44 : ulong next;
45 : ulong prev;
46 : } deadline;
47 :
48 : int state;
49 : ulong latency_nanos;
50 : long deadline_nanos;
51 : ulong used_fd_idx;
52 : };
53 :
54 : typedef struct fd_ssping_peer fd_ssping_peer_t;
55 :
56 : #define POOL_NAME peer_pool
57 36 : #define POOL_T fd_ssping_peer_t
58 : #define POOL_IDX_T ulong
59 918 : #define POOL_NEXT pool.next
60 : #include "../../../util/tmpl/fd_pool.c"
61 :
62 : #define MAP_NAME peer_map
63 630 : #define MAP_KEY addr
64 306 : #define MAP_ELE_T fd_ssping_peer_t
65 : #define MAP_KEY_T fd_ip4_port_t
66 324 : #define MAP_PREV map.prev
67 630 : #define MAP_NEXT map.next
68 684 : #define MAP_KEY_EQ(k0,k1) ((k0)->l==(k1)->l)
69 1890 : #define MAP_KEY_HASH(key,seed) (fd_ulong_hash( (seed)^(key)->l ))
70 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
71 : #include "../../../util/tmpl/fd_map_chain.c"
72 :
73 : #define DLIST_NAME deadline_list
74 : #define DLIST_ELE_T fd_ssping_peer_t
75 954 : #define DLIST_PREV deadline.prev
76 954 : #define DLIST_NEXT deadline.next
77 : #include "../../../util/tmpl/fd_dlist.c"
78 :
79 : struct fd_ssping_private {
80 : fd_ssping_peer_t * pool;
81 : peer_map_t * map;
82 :
83 : deadline_list_t * unpinged;
84 : deadline_list_t * pinged;
85 : deadline_list_t * valid;
86 : deadline_list_t * refreshing;
87 : deadline_list_t * invalid;
88 :
89 : fd_ssping_on_ping_fn_t on_ping_cb;
90 : void * cb_arg;
91 :
92 : ulong magic; /* ==FD_SSPING_MAGIC */
93 :
94 : int epoll_fd;
95 :
96 : /* Invariant: The pool elements with an associated file descriptor are
97 : exactly those that are PINGED or REFRESHING. */
98 : ulong used_fd_cnt;
99 : struct pollfd used_fds[ FD_SSPING_FD_CNT ]; /* indexed [0, used_fd_cnt) */
100 : int idle_fds[ FD_SSPING_FD_CNT ]; /* indexed [0, FD_SSPING_FD_CNT-used_fd_cnt) */
101 : /* ping_to_pool[ i ]==x means that used_fds[ i ].fd is in use for
102 : pinging the peer in pool[ x ]. */
103 : ulong ping_to_pool[ FD_SSPING_FD_CNT ]; /* indexed [0, used_fd_cnt) */
104 : };
105 :
106 :
107 : FD_FN_CONST ulong
108 3015 : fd_ssping_align( void ) {
109 3015 : return fd_ulong_max( alignof(fd_ssping_t),
110 3015 : fd_ulong_max( peer_pool_align(),
111 3015 : fd_ulong_max( peer_map_align(),
112 3015 : deadline_list_align() ) ) );
113 3015 : }
114 :
115 : FD_FN_CONST ulong
116 435 : fd_ssping_footprint( ulong max_peers ) {
117 435 : ulong l;
118 435 : l = FD_LAYOUT_INIT;
119 435 : l = FD_LAYOUT_APPEND( l, alignof(fd_ssping_t), sizeof(fd_ssping_t) );
120 435 : l = FD_LAYOUT_APPEND( l, peer_pool_align(), peer_pool_footprint( max_peers ) );
121 435 : l = FD_LAYOUT_APPEND( l, peer_map_align(), peer_map_footprint( peer_map_chain_cnt_est( max_peers ) ) );
122 435 : l = FD_LAYOUT_APPEND( l, deadline_list_align(), deadline_list_footprint() );
123 435 : l = FD_LAYOUT_APPEND( l, deadline_list_align(), deadline_list_footprint() );
124 435 : l = FD_LAYOUT_APPEND( l, deadline_list_align(), deadline_list_footprint() );
125 435 : l = FD_LAYOUT_APPEND( l, deadline_list_align(), deadline_list_footprint() );
126 435 : l = FD_LAYOUT_APPEND( l, deadline_list_align(), deadline_list_footprint() );
127 435 : return FD_LAYOUT_FINI( l, fd_ssping_align() );
128 435 : }
129 :
130 : void *
131 : fd_ssping_new( void * shmem,
132 : ulong max_peers,
133 : ulong seed,
134 : fd_ssping_on_ping_fn_t on_ping_cb,
135 : void * cb_arg,
136 18 : int epoll_fd ) {
137 18 : if( FD_UNLIKELY( !shmem ) ) {
138 0 : FD_LOG_WARNING(( "NULL shmem" ));
139 0 : return NULL;
140 0 : }
141 :
142 18 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_ssping_align() ) ) ) {
143 0 : FD_LOG_WARNING(( "unaligned shmem" ));
144 0 : return NULL;
145 0 : }
146 :
147 18 : if( FD_UNLIKELY( max_peers < 1UL ) ) {
148 0 : FD_LOG_WARNING(( "max_peers must be at least 1" ));
149 0 : return NULL;
150 0 : }
151 :
152 18 : if( FD_UNLIKELY( epoll_fd==-1 ) ) {
153 0 : FD_LOG_WARNING(( "invalid epoll_fd" ));
154 0 : return NULL;
155 0 : }
156 :
157 18 : FD_SCRATCH_ALLOC_INIT( l, shmem );
158 18 : fd_ssping_t * ssping = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_ssping_t), sizeof(fd_ssping_t) );
159 18 : void * _pool = FD_SCRATCH_ALLOC_APPEND( l, peer_pool_align(), peer_pool_footprint( max_peers ) );
160 18 : void * _map = FD_SCRATCH_ALLOC_APPEND( l, peer_map_align(), peer_map_footprint( peer_map_chain_cnt_est( max_peers ) ) );
161 18 : void * _unpinged = FD_SCRATCH_ALLOC_APPEND( l, deadline_list_align(), deadline_list_footprint() );
162 18 : void * _pinged = FD_SCRATCH_ALLOC_APPEND( l, deadline_list_align(), deadline_list_footprint() );
163 18 : void * _valid = FD_SCRATCH_ALLOC_APPEND( l, deadline_list_align(), deadline_list_footprint() );
164 18 : void * _refreshing = FD_SCRATCH_ALLOC_APPEND( l, deadline_list_align(), deadline_list_footprint() );
165 18 : void * _invalid = FD_SCRATCH_ALLOC_APPEND( l, deadline_list_align(), deadline_list_footprint() );
166 :
167 18 : ssping->pool = peer_pool_join( peer_pool_new( _pool, max_peers ) );
168 18 : ssping->map = peer_map_join( peer_map_new( _map, peer_map_chain_cnt_est( max_peers ), seed ) );
169 :
170 18 : ssping->unpinged = deadline_list_join( deadline_list_new( _unpinged ) );
171 18 : ssping->pinged = deadline_list_join( deadline_list_new( _pinged ) );
172 18 : ssping->valid = deadline_list_join( deadline_list_new( _valid ) );
173 18 : ssping->refreshing = deadline_list_join( deadline_list_new( _refreshing ) );
174 18 : ssping->invalid = deadline_list_join( deadline_list_new( _invalid ) );
175 :
176 : /* Allocate a contiguous range of file descriptors */
177 18 : int next_fd = FD_SSPING_FD_MIN;
178 :
179 4464 : for( ulong i=0UL; i<FD_SSPING_FD_CNT; i++, next_fd++ ) {
180 4446 : int fd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK, IPPROTO_TCP );
181 4446 : if( FD_UNLIKELY( -1==fd ) ) FD_LOG_ERR(( "socket(SOCK_STREAM,IPPROTO_TCP) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
182 :
183 4446 : int actual_fd = fcntl( fd, F_DUPFD_CLOEXEC, next_fd );
184 4446 : if( FD_UNLIKELY( actual_fd<0 ) ) {
185 0 : FD_LOG_ERR(( "fcntl(F_DUPFD_CLOEXEC,%d) failed (%i-%s)", next_fd, errno, fd_io_strerror( errno ) ));
186 0 : }
187 4446 : if( FD_UNLIKELY( actual_fd!=next_fd ) ) {
188 0 : FD_LOG_ERR(( "file descriptor collision at %d", next_fd ));
189 0 : }
190 4446 : if( FD_UNLIKELY( 0!=close( fd ) ) ) {
191 0 : FD_LOG_ERR(( "close(%d) failed (%i-%s)", fd, errno, fd_io_strerror( errno ) ));
192 0 : }
193 :
194 4446 : int tcp_nodelay = 1;
195 4446 : if( FD_UNLIKELY( setsockopt( next_fd, SOL_TCP, TCP_NODELAY, &tcp_nodelay, sizeof(int) ) ) ) {
196 0 : FD_LOG_ERR(( "setsockopt(SOL_TCP,TCP_NODELAY,1) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
197 0 : }
198 4446 : ssping->idle_fds[ i ] = next_fd;
199 :
200 4446 : ssping->used_fds[ i ].fd = -1;
201 4446 : ssping->used_fds[ i ].events = POLLOUT|POLLRDHUP|POLLPRI;
202 4446 : ssping->used_fds[ i ].revents = 0;
203 4446 : }
204 :
205 18 : ssping->used_fd_cnt = 0UL;
206 18 : ssping->epoll_fd = epoll_fd;
207 :
208 18 : ssping->on_ping_cb = on_ping_cb;
209 18 : ssping->cb_arg = cb_arg;
210 :
211 18 : FD_COMPILER_MFENCE();
212 18 : FD_VOLATILE( ssping->magic ) = FD_SSPING_MAGIC;
213 18 : FD_COMPILER_MFENCE();
214 :
215 18 : return (void *)ssping;
216 18 : }
217 :
218 : fd_ssping_t *
219 18 : fd_ssping_join( void * shping ) {
220 18 : if( FD_UNLIKELY( !shping ) ) {
221 0 : FD_LOG_WARNING(( "NULL shping" ));
222 0 : return NULL;
223 0 : }
224 :
225 18 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shping, fd_ssping_align() ) ) ) {
226 0 : FD_LOG_WARNING(( "misaligned shping" ));
227 0 : return NULL;
228 0 : }
229 :
230 18 : fd_ssping_t * ssping = (fd_ssping_t *)shping;
231 :
232 18 : if( FD_UNLIKELY( ssping->magic!=FD_SSPING_MAGIC ) ) {
233 0 : FD_LOG_WARNING(( "bad magic" ));
234 0 : return NULL;
235 0 : }
236 :
237 18 : return ssping;
238 18 : }
239 :
240 : void *
241 3 : fd_ssping_leave( fd_ssping_t * ssping ) {
242 3 : if( FD_UNLIKELY( !ssping ) ) {
243 0 : FD_LOG_WARNING(( "NULL ssping" ));
244 0 : return NULL;
245 0 : }
246 :
247 3 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ssping, fd_ssping_align() ) ) ) {
248 0 : FD_LOG_WARNING(( "misaligned ssping" ));
249 0 : return NULL;
250 0 : }
251 :
252 3 : if( FD_UNLIKELY( ssping->magic!=FD_SSPING_MAGIC ) ) {
253 0 : FD_LOG_WARNING(( "bad magic" ));
254 0 : return NULL;
255 0 : }
256 :
257 3 : ssping->pool = peer_pool_leave( ssping->pool );
258 3 : ssping->map = peer_map_leave( ssping->map );
259 3 : ssping->unpinged = deadline_list_leave( ssping->unpinged );
260 3 : ssping->pinged = deadline_list_leave( ssping->pinged );
261 3 : ssping->valid = deadline_list_leave( ssping->valid );
262 3 : ssping->refreshing = deadline_list_leave( ssping->refreshing );
263 3 : ssping->invalid = deadline_list_leave( ssping->invalid );
264 :
265 3 : return (void *)ssping;
266 3 : }
267 :
268 : void *
269 3 : fd_ssping_delete( void * shping ) {
270 3 : if( FD_UNLIKELY( !shping ) ) {
271 0 : FD_LOG_WARNING(( "NULL shping" ));
272 0 : return NULL;
273 0 : }
274 :
275 3 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shping, fd_ssping_align() ) ) ) {
276 0 : FD_LOG_WARNING(( "misaligned shping" ));
277 0 : return NULL;
278 0 : }
279 :
280 3 : fd_ssping_t * ssping = (fd_ssping_t *)shping;
281 :
282 3 : if( FD_UNLIKELY( ssping->magic!=FD_SSPING_MAGIC ) ) {
283 0 : FD_LOG_WARNING(( "bad magic" ));
284 0 : return NULL;
285 0 : }
286 :
287 : /* Close all file descriptors opened by fd_ssping_new. */
288 3 : for( ulong i=0UL; i<ssping->used_fd_cnt; i++ ) {
289 0 : close( ssping->used_fds[ i ].fd );
290 0 : }
291 744 : for( ulong i=0UL; i<FD_SSPING_FD_CNT-ssping->used_fd_cnt; i++ ) {
292 741 : close( ssping->idle_fds[ i ] );
293 741 : }
294 :
295 3 : FD_COMPILER_MFENCE();
296 3 : FD_VOLATILE( ssping->magic ) = 0UL;
297 3 : FD_COMPILER_MFENCE();
298 :
299 3 : return (void *)ssping;
300 3 : }
301 :
302 : void
303 : fd_ssping_add( fd_ssping_t * ssping,
304 342 : fd_ip4_port_t addr ) {
305 342 : fd_ssping_peer_t * peer = peer_map_ele_query( ssping->map, &addr, NULL, ssping->pool );
306 342 : if( FD_LIKELY( !peer ) ) {
307 324 : if( FD_UNLIKELY( !peer_pool_free( ssping->pool ) ) ) {
308 0 : FD_LOG_WARNING(( "ping peer pool exhausted" ));
309 0 : return;
310 0 : }
311 324 : peer = peer_pool_ele_acquire( ssping->pool );
312 324 : memset( peer, 0, sizeof(fd_ssping_peer_t) );
313 324 : peer->refcnt = 0UL;
314 324 : peer->state = PEER_STATE_UNPINGED;
315 324 : peer->addr = addr;
316 324 : peer->latency_nanos = ULONG_MAX;
317 324 : peer->used_fd_idx = ULONG_MAX;
318 324 : peer_map_ele_insert( ssping->map, peer, ssping->pool );
319 324 : deadline_list_ele_push_tail( ssping->unpinged, peer, ssping->pool );
320 324 : }
321 342 : peer->refcnt++;
322 342 : }
323 :
324 : static void
325 : remove_fdesc_idx( fd_ssping_t * ssping,
326 0 : ulong fdesc_idx ) {
327 0 : FD_TEST( fdesc_idx<FD_SSPING_FD_CNT );
328 0 : FD_TEST( fdesc_idx<ssping->used_fd_cnt );
329 0 : ulong pool_idx = ssping->ping_to_pool[ fdesc_idx ];
330 :
331 0 : int fdesc = ssping->used_fds[ fdesc_idx ].fd;
332 : /* Abort the connection attempt or close the connection by connecting
333 : to AF_UNSPEC. */
334 0 : struct sockaddr_in addr[1] = {{
335 0 : .sin_family = AF_UNSPEC,
336 0 : .sin_addr = { .s_addr = 0U },
337 0 : .sin_port = 0
338 0 : }};
339 0 : if( FD_UNLIKELY( connect( fdesc, fd_type_pun_const( addr ), sizeof(addr) ) ) ) FD_LOG_ERR(( "connect(AF_UNSPEC) failed (%d-%s)", errno, fd_io_strerror( errno ) ));
340 0 : if( FD_UNLIKELY( -1==epoll_ctl( ssping->epoll_fd, EPOLL_CTL_DEL, fdesc, NULL ) ) ) FD_LOG_ERR(( "epoll_ctl(DEL) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
341 :
342 : /* Mark that the pool element no longer has an associated index. */
343 0 : ssping->pool[ pool_idx ].used_fd_idx = ULONG_MAX;
344 :
345 : /* Now swap the last used_fd into this position, updating all the
346 : relevant bookkeeping info. */
347 0 : ulong last = ssping->used_fd_cnt-1UL;
348 0 : if( FD_LIKELY( fdesc_idx!=last ) ) {
349 0 : ssping->used_fds[ fdesc_idx ] = ssping->used_fds[ last ];
350 0 : ulong last_pool_idx = ssping->ping_to_pool[ fdesc_idx ] = ssping->ping_to_pool[ last ];
351 0 : ssping->pool[ last_pool_idx ].used_fd_idx = fdesc_idx;
352 0 : }
353 :
354 0 : ssping->idle_fds[ FD_SSPING_FD_CNT - ssping->used_fd_cnt ] = fdesc;
355 0 : ssping->used_fd_cnt--;
356 0 : }
357 :
358 : int
359 : fd_ssping_remove( fd_ssping_t * ssping,
360 378 : fd_ip4_port_t addr ) {
361 378 : fd_ssping_peer_t * peer = peer_map_ele_query( ssping->map, &addr, NULL, ssping->pool );
362 378 : if( FD_UNLIKELY( !peer ) ) return 0;
363 324 : if( FD_UNLIKELY( !peer->refcnt ) ) return 0;
364 324 : peer->refcnt--;
365 324 : if( FD_LIKELY( !peer->refcnt ) ) {
366 306 : switch( peer->state ) {
367 144 : case PEER_STATE_UNPINGED:
368 144 : deadline_list_ele_remove( ssping->unpinged, peer, ssping->pool );
369 144 : break;
370 0 : case PEER_STATE_PINGED:
371 0 : deadline_list_ele_remove( ssping->pinged, peer, ssping->pool );
372 0 : remove_fdesc_idx( ssping, peer->used_fd_idx );
373 0 : break;
374 0 : case PEER_STATE_VALID:
375 0 : deadline_list_ele_remove( ssping->valid, peer, ssping->pool );
376 0 : break;
377 0 : case PEER_STATE_REFRESHING:
378 0 : deadline_list_ele_remove( ssping->refreshing, peer, ssping->pool );
379 0 : remove_fdesc_idx( ssping, peer->used_fd_idx );
380 0 : break;
381 162 : case PEER_STATE_INVALID:
382 162 : deadline_list_ele_remove( ssping->invalid, peer, ssping->pool );
383 162 : break;
384 306 : }
385 306 : peer_map_ele_remove_fast( ssping->map, peer, ssping->pool );
386 306 : peer_pool_ele_release( ssping->pool, peer );
387 306 : return 1;
388 306 : }
389 18 : return 0;
390 324 : }
391 :
392 : void
393 : fd_ssping_invalidate( fd_ssping_t * ssping,
394 : fd_ip4_port_t addr,
395 234 : long now ) {
396 234 : if( FD_UNLIKELY( !ssping ) ) return;
397 234 : fd_ssping_peer_t * peer = peer_map_ele_query( ssping->map, &addr, NULL, ssping->pool );
398 234 : if( FD_UNLIKELY( !peer ) ) return;
399 180 : switch( peer->state ) {
400 162 : case PEER_STATE_UNPINGED:
401 162 : deadline_list_ele_remove( ssping->unpinged, peer, ssping->pool );
402 162 : break;
403 0 : case PEER_STATE_PINGED:
404 0 : deadline_list_ele_remove( ssping->pinged, peer, ssping->pool );
405 0 : remove_fdesc_idx( ssping, peer->used_fd_idx );
406 0 : break;
407 0 : case PEER_STATE_VALID:
408 0 : deadline_list_ele_remove( ssping->valid, peer, ssping->pool );
409 0 : break;
410 0 : case PEER_STATE_REFRESHING:
411 0 : deadline_list_ele_remove( ssping->refreshing, peer, ssping->pool );
412 0 : remove_fdesc_idx( ssping, peer->used_fd_idx );
413 0 : break;
414 18 : case PEER_STATE_INVALID:
415 18 : return;
416 180 : }
417 162 : peer->state = PEER_STATE_INVALID;
418 162 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_INVALID;
419 162 : deadline_list_ele_push_tail( ssping->invalid, peer, ssping->pool );
420 162 : }
421 :
422 : static inline void
423 : recv_pings( fd_ssping_t * ssping,
424 18 : fd_sspeer_selector_t * selector) {
425 18 : int pollv = fd_syscall_poll( ssping->used_fds, (uint)ssping->used_fd_cnt, 0 );
426 18 : if( FD_UNLIKELY( pollv<0 ) ) {
427 0 : FD_LOG_WARNING(( "poll(used_fds,%lu,0) failed (%d-%s)", ssping->used_fd_cnt, errno, fd_io_strerror( errno ) ));
428 0 : return;
429 0 : }
430 18 : long now = fd_log_wallclock();
431 18 : ulong processed = 0UL;
432 18 : ulong processed_idx[ PING_BURST_MAX ];
433 18 : for( ulong i=0UL; i<ssping->used_fd_cnt; i++ ) {
434 0 : if( FD_UNLIKELY( processed >= fd_ulong_min( (ulong)pollv, PING_BURST_MAX ) ) ) break;
435 0 : if( FD_UNLIKELY( ssping->used_fds[ i ].revents ) ) {
436 0 : ulong pool_idx = ssping->ping_to_pool[ i ];
437 0 : fd_ssping_peer_t * peer = ssping->pool+pool_idx;
438 :
439 0 : FD_TEST( peer->state==PEER_STATE_PINGED || peer->state==PEER_STATE_REFRESHING );
440 :
441 :
442 0 : deadline_list_ele_remove( peer->state==PEER_STATE_PINGED ? ssping->pinged : ssping->refreshing, peer, ssping->pool );
443 0 : int is_err = ssping->used_fds[ i ].revents & (POLLRDHUP|POLLERR|POLLHUP);
444 0 : if( FD_LIKELY( !is_err ) ) {
445 0 : peer->latency_nanos = (ulong)fd_long_max( now - (peer->deadline_nanos - PEER_DEADLINE_NANOS_PING), 1L );
446 0 : peer->state = PEER_STATE_VALID;
447 0 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_VALID;
448 0 : deadline_list_ele_push_tail( ssping->valid, peer, ssping->pool );
449 :
450 0 : FD_LOG_INFO(( "pinged " FD_IP4_ADDR_FMT ":%hu in %lu nanos",
451 0 : FD_IP4_ADDR_FMT_ARGS( peer->addr.addr ), fd_ushort_bswap( peer->addr.port ), peer->latency_nanos ));
452 0 : ssping->on_ping_cb( ssping->cb_arg, peer->addr, peer->latency_nanos );
453 0 : } else {
454 : /* This is pretty unlikely, but the host could respond with an
455 : RST packet I suppose. */
456 0 : peer->state = PEER_STATE_INVALID;
457 0 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_INVALID;
458 0 : deadline_list_ele_push_tail( ssping->invalid, peer, ssping->pool );
459 0 : fd_sspeer_selector_remove_by_addr( selector, peer->addr );
460 0 : }
461 0 : processed_idx[ processed ] = i;
462 0 : processed++;
463 0 : }
464 0 : }
465 : /* Now we need to call remove_fdesc_idx on the processed ones in
466 : reverse order (largest to smallest) so that we don't trip on
467 : ourself as we shuffle the array. */
468 18 : while( processed ) remove_fdesc_idx( ssping, processed_idx[ --processed ] );
469 18 : }
470 :
471 : static uint
472 : send_pings( fd_ssping_t * ssping,
473 : deadline_list_t * list,
474 36 : long until ) {
475 36 : uint msg_cnt = 0U;
476 36 : for( deadline_list_iter_t iter = deadline_list_iter_fwd_init( list, ssping->pool );
477 36 : msg_cnt<PING_BURST_MAX && ssping->used_fd_cnt<FD_SSPING_FD_CNT && !deadline_list_iter_done( iter, list, ssping->pool );
478 36 : iter = deadline_list_iter_fwd_next( iter, list, ssping->pool ) ) {
479 0 : ulong peer_idx = deadline_list_iter_idx( iter, list, ssping->pool );
480 0 : fd_ssping_peer_t * peer = peer_pool_ele( ssping->pool, peer_idx );
481 0 : if( peer->deadline_nanos>until ) break;
482 :
483 0 : int fdesc = ssping->idle_fds[ FD_SSPING_FD_CNT-ssping->used_fd_cnt-1UL ];
484 :
485 0 : struct sockaddr_in addr[1] = {{
486 0 : .sin_family = AF_INET,
487 0 : .sin_addr = { .s_addr = peer->addr.addr },
488 0 : .sin_port = peer->addr.port
489 0 : }};
490 :
491 0 : if( FD_UNLIKELY( connect( fdesc, fd_type_pun_const( addr ), sizeof(addr) ) && errno!=EINPROGRESS ) ) {
492 0 : FD_LOG_WARNING(( "connect(" FD_IP4_ADDR_FMT ":%hu) failed (%d-%s)", FD_IP4_ADDR_FMT_ARGS( peer->addr.addr ), fd_ushort_bswap( peer->addr.port ), errno, fd_io_strerror( errno ) ));
493 : /* Nothing to do. It will get "reaped" later. */
494 0 : }
495 0 : struct epoll_event ev = { .events = EPOLLOUT|EPOLLRDHUP|EPOLLPRI, .data.fd = fdesc };
496 0 : if( FD_UNLIKELY( -1==epoll_ctl( ssping->epoll_fd, EPOLL_CTL_ADD, fdesc, &ev ) ) ) FD_LOG_ERR(( "epoll_ctl(ADD) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
497 :
498 0 : ssping->used_fds [ ssping->used_fd_cnt ].fd = fdesc;
499 0 : ssping->ping_to_pool[ ssping->used_fd_cnt ] = peer_idx;
500 0 : peer->used_fd_idx = ssping->used_fd_cnt;
501 0 : ssping->used_fd_cnt++;
502 0 : msg_cnt++;
503 0 : }
504 :
505 36 : if( msg_cnt==0U ) return 0U;
506 0 : return (uint)msg_cnt;
507 36 : }
508 :
509 :
510 : long
511 18 : fd_ssping_next_deadline( fd_ssping_t const * ssping ) {
512 18 : long next = LONG_MAX;
513 18 : deadline_list_t const * lists[] = { ssping->pinged, ssping->refreshing, ssping->valid, ssping->invalid };
514 90 : for( ulong i=0UL; i<sizeof(lists)/sizeof(lists[0]); i++ ) {
515 72 : if( FD_LIKELY( deadline_list_is_empty( lists[ i ], ssping->pool ) ) ) continue;
516 0 : next = fd_long_min( next, deadline_list_ele_peek_head_const( lists[ i ], ssping->pool )->deadline_nanos );
517 0 : }
518 18 : if( FD_LIKELY( !deadline_list_is_empty( ssping->unpinged, ssping->pool ) && ssping->used_fd_cnt<FD_SSPING_FD_CNT ) ) next = 0L; /* sends now */
519 18 : return next;
520 18 : }
521 :
522 : void
523 : fd_ssping_advance( fd_ssping_t * ssping,
524 : long now,
525 18 : fd_sspeer_selector_t * selector) {
526 18 : uint sent = send_pings( ssping, ssping->unpinged, LONG_MAX );
527 18 : for( uint i=0U; i<sent; i++ ) {
528 0 : fd_ssping_peer_t * peer = deadline_list_ele_pop_head( ssping->unpinged, ssping->pool );
529 0 : FD_TEST( peer );
530 0 : peer->state = PEER_STATE_PINGED;
531 0 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_PING;
532 0 : deadline_list_ele_push_tail( ssping->pinged, peer, ssping->pool );
533 0 : }
534 :
535 18 : while( !deadline_list_is_empty( ssping->pinged, ssping->pool ) ) {
536 0 : fd_ssping_peer_t * peer = deadline_list_ele_peek_head( ssping->pinged, ssping->pool );
537 0 : if( FD_LIKELY( peer->deadline_nanos>now ) ) break;
538 :
539 0 : deadline_list_ele_pop_head( ssping->pinged, ssping->pool );
540 :
541 0 : remove_fdesc_idx( ssping, peer->used_fd_idx );
542 :
543 0 : peer->state = PEER_STATE_INVALID;
544 0 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_INVALID;
545 0 : deadline_list_ele_push_tail( ssping->invalid, peer, ssping->pool );
546 0 : fd_sspeer_selector_remove_by_addr( selector, peer->addr );
547 0 : }
548 :
549 18 : sent = send_pings( ssping, ssping->valid, now );
550 18 : for( uint i=0U; i<sent; i++ ) {
551 0 : fd_ssping_peer_t * peer = deadline_list_ele_pop_head( ssping->valid, ssping->pool );
552 0 : FD_TEST( peer );
553 0 : peer->state = PEER_STATE_REFRESHING;
554 0 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_PING;
555 0 : deadline_list_ele_push_tail( ssping->refreshing, peer, ssping->pool );
556 0 : }
557 :
558 18 : while( !deadline_list_is_empty( ssping->refreshing, ssping->pool ) ) {
559 0 : fd_ssping_peer_t * peer = deadline_list_ele_peek_head( ssping->refreshing, ssping->pool );
560 0 : if( FD_LIKELY( peer->deadline_nanos>now ) ) break;
561 :
562 0 : deadline_list_ele_pop_head( ssping->refreshing, ssping->pool );
563 :
564 0 : remove_fdesc_idx( ssping, peer->used_fd_idx );
565 :
566 0 : peer->state = PEER_STATE_INVALID;
567 0 : peer->deadline_nanos = now + PEER_DEADLINE_NANOS_INVALID;
568 0 : deadline_list_ele_push_tail( ssping->invalid, peer, ssping->pool );
569 0 : fd_sspeer_selector_remove_by_addr( selector, peer->addr );
570 0 : }
571 :
572 18 : while( !deadline_list_is_empty( ssping->invalid, ssping->pool ) ) {
573 0 : fd_ssping_peer_t * peer = deadline_list_ele_peek_head( ssping->invalid, ssping->pool );
574 0 : if( FD_LIKELY( peer->deadline_nanos>now ) ) break;
575 :
576 0 : deadline_list_ele_pop_head( ssping->invalid, ssping->pool );
577 :
578 0 : peer->state = PEER_STATE_UNPINGED;
579 0 : peer->deadline_nanos = 0L;
580 0 : deadline_list_ele_push_tail( ssping->unpinged, peer, ssping->pool );
581 0 : }
582 :
583 18 : recv_pings( ssping, selector );
584 18 : }
585 :
586 : int
587 : fd_ssping_is_invalidated( fd_ssping_t * ssping,
588 324 : fd_ip4_port_t addr ) {
589 324 : if( FD_UNLIKELY( !ssping ) ) return 0;
590 306 : fd_ssping_peer_t * peer = peer_map_ele_query( ssping->map, &addr, NULL, ssping->pool );
591 306 : if( FD_UNLIKELY( !peer ) ) return 0;
592 162 : return peer->state==PEER_STATE_INVALID;
593 306 : }
|
