Line data Source code
1 : #include "fd_ipecho_client.h"
2 : #include "fd_ipecho_client_private.h"
3 :
4 : #include "../../util/fd_util.h"
5 :
6 : #include <errno.h>
7 : #include <netinet/in.h>
8 : #include <string.h>
9 : #include <sys/epoll.h>
10 : #include <sys/socket.h>
11 : #include <unistd.h>
12 :
13 : FD_FN_CONST ulong
14 0 : fd_ipecho_client_align( void ) {
15 0 : return alignof(fd_ipecho_client_t);
16 0 : }
17 :
18 : FD_FN_CONST ulong
19 0 : fd_ipecho_client_footprint( void ) {
20 0 : return sizeof(fd_ipecho_client_t);
21 0 : }
22 :
23 : void *
24 0 : fd_ipecho_client_new( void * shmem ) {
25 0 : fd_ipecho_client_t * ipe = (fd_ipecho_client_t *)shmem;
26 :
27 0 : if( FD_UNLIKELY( !shmem ) ) {
28 0 : FD_LOG_WARNING(( "NULL shmem" ));
29 0 : return NULL;
30 0 : }
31 :
32 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_ipecho_client_align() ) ) ) {
33 0 : FD_LOG_WARNING(( "misaligned shmem" ));
34 0 : return NULL;
35 0 : }
36 :
37 0 : FD_COMPILER_MFENCE();
38 0 : FD_VOLATILE( ipe->magic ) = FD_IPECHO_CLIENT_MAGIC;
39 0 : FD_COMPILER_MFENCE();
40 :
41 0 : return (void *)ipe;
42 0 : }
43 :
44 : fd_ipecho_client_t *
45 0 : fd_ipecho_client_join( void * shipe ) {
46 0 : if( FD_UNLIKELY( !shipe ) ) {
47 0 : FD_LOG_WARNING(( "NULL shipe" ));
48 0 : return NULL;
49 0 : }
50 :
51 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shipe, fd_ipecho_client_align() ) ) ) {
52 0 : FD_LOG_WARNING(( "misaligned shipe" ));
53 0 : return NULL;
54 0 : }
55 :
56 0 : fd_ipecho_client_t * ipe = (fd_ipecho_client_t *)shipe;
57 :
58 0 : if( FD_UNLIKELY( ipe->magic!=FD_IPECHO_CLIENT_MAGIC ) ) {
59 0 : FD_LOG_WARNING(( "bad magic" ));
60 0 : return NULL;
61 0 : }
62 :
63 0 : return ipe;
64 0 : }
65 :
66 : void
67 : fd_ipecho_client_init( fd_ipecho_client_t * client,
68 : fd_ip4_port_t const * servers,
69 : ulong servers_len,
70 0 : int epoll_fd ) {
71 0 : FD_TEST( epoll_fd!=-1 );
72 0 : ulong peer_cnt = 0UL;
73 :
74 0 : client->epoll_fd = epoll_fd;
75 :
76 0 : for( ulong i=0UL; i<servers_len; i++ ) {
77 0 : fd_ip4_port_t const * server = &servers[ i ];
78 :
79 0 : int sockfd = socket( AF_INET, SOCK_STREAM|SOCK_NONBLOCK, 0 );
80 0 : if( FD_UNLIKELY( -1==sockfd ) ) FD_LOG_ERR(( "socket() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
81 :
82 0 : struct sockaddr_in addr = {
83 0 : .sin_family = AF_INET,
84 0 : .sin_port = server->port,
85 0 : .sin_addr = { .s_addr = server->addr }
86 0 : };
87 :
88 0 : if( FD_UNLIKELY( -1==connect( sockfd, fd_type_pun( &addr ), sizeof(addr) ) && errno!=EINPROGRESS ) ) {
89 0 : if( FD_UNLIKELY( -1==close( sockfd ) ) ) FD_LOG_ERR(( "close() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
90 0 : continue;
91 0 : }
92 :
93 0 : struct epoll_event ev = { .events = EPOLLIN|EPOLLOUT, .data.fd = sockfd };
94 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 ) ));
95 :
96 0 : client->pollfds[ peer_cnt ] = (struct pollfd){
97 0 : .fd = sockfd,
98 0 : .events = POLLIN | POLLOUT,
99 0 : .revents = 0
100 0 : };
101 0 : client->peers[ peer_cnt ].writing = 1;
102 0 : client->peers[ peer_cnt ].request_bytes_sent = 0UL;
103 0 : client->peers[ peer_cnt ].response_bytes_read = 0UL;
104 0 : peer_cnt++;
105 0 : }
106 :
107 0 : for( ulong i=peer_cnt; i<16UL; i++ ) client->pollfds[ i ].fd = -1;
108 :
109 0 : client->peer_cnt = peer_cnt;
110 0 : client->remaining_peer_cnt = peer_cnt;
111 0 : client->start_time_nanos = LONG_MAX;
112 0 : }
113 :
114 : static void
115 : close_one( fd_ipecho_client_t * client,
116 0 : ulong idx ) {
117 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 ) ));
118 :
119 0 : if( FD_UNLIKELY( -1==close( client->pollfds[ idx ].fd ) ) ) FD_LOG_ERR(( "close() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
120 0 : client->pollfds[ idx ].fd = -1;
121 0 : client->remaining_peer_cnt--;
122 0 : }
123 :
124 : static void
125 0 : close_all( fd_ipecho_client_t * client ) {
126 0 : for( ulong i=0UL; i<client->peer_cnt; i++ ) {
127 0 : if( FD_UNLIKELY( -1==client->pollfds[ i ].fd ) ) continue;
128 0 : close_one( client, i );
129 0 : }
130 0 : }
131 :
132 : int
133 : fd_ipecho_client_parse_response( uchar const * response,
134 : ulong response_len,
135 0 : ushort * _shred_version ) {
136 0 : if( FD_UNLIKELY( response_len<8UL ) ) return FD_IPECHO_PARSE_ERR;
137 0 : if( FD_UNLIKELY( memcmp( response, "\0\0\0\0", 4UL ) ) ) return FD_IPECHO_PARSE_ERR;
138 :
139 0 : uint ip_variant = fd_uint_load_4_fast( response+4UL );
140 0 : if( FD_UNLIKELY( ip_variant>1U ) ) return FD_IPECHO_PARSE_ERR;
141 :
142 0 : ulong offset = 8UL;
143 0 : if( FD_LIKELY( !ip_variant ) ) offset+=4UL;
144 0 : else offset+=16UL;
145 :
146 0 : if( FD_UNLIKELY( response_len<offset+3UL ) ) return FD_IPECHO_PARSE_ERR;
147 :
148 0 : if( FD_UNLIKELY( response[ offset ]!=1 ) ) return FD_IPECHO_PARSE_ERR;
149 :
150 0 : ushort shred_version = fd_ushort_load_2_fast( response+offset+1UL );
151 0 : if( FD_UNLIKELY( !shred_version ) ) return FD_IPECHO_PARSE_ERR;
152 :
153 0 : *_shred_version = shred_version;
154 0 : return FD_IPECHO_PARSE_OK;
155 0 : }
156 :
157 : static void
158 : write_conn( fd_ipecho_client_t * client,
159 0 : ulong conn_idx ) {
160 0 : fd_ipecho_client_peer_t * peer = &client->peers[ conn_idx ];
161 :
162 0 : if( FD_LIKELY( !peer->writing ) ) return;
163 :
164 0 : uchar request[ 21UL ] = {
165 0 : 0, 0, 0, 0, /* Magic */
166 0 : 0, 0, 0, 0, 0, 0, 0, 0, /* TCP ports */
167 0 : 0, 0, 0, 0, 0, 0, 0, 0, /* UDP ports */
168 0 : '\n', /* End of request */
169 0 : };
170 :
171 0 : long written = sendto( client->pollfds[ conn_idx ].fd,
172 0 : request+peer->request_bytes_sent,
173 0 : sizeof(request)-peer->request_bytes_sent,
174 0 : MSG_NOSIGNAL,
175 0 : NULL,
176 0 : 0 );
177 0 : if( FD_UNLIKELY( -1==written && errno==EAGAIN ) ) return; /* No data was written, continue. */
178 0 : else if( FD_UNLIKELY( -1==written ) ) {
179 0 : close_one( client, conn_idx );
180 0 : return;
181 0 : }
182 :
183 0 : peer->request_bytes_sent += (ulong)written;
184 0 : if( FD_UNLIKELY( peer->request_bytes_sent==sizeof(request) ) ) {
185 0 : peer->writing = 0;
186 0 : peer->response_bytes_read = 0UL;
187 : /* A connected socket is always writable so drop EPOLLOUT or the
188 : waker refires on every rearm while we wait for the response */
189 0 : client->pollfds[ conn_idx ].events = POLLIN;
190 0 : struct epoll_event ev = { .events = EPOLLIN, .data.fd = client->pollfds[ conn_idx ].fd };
191 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 ) ));
192 0 : }
193 0 : }
194 :
195 : static int
196 : read_conn( fd_ipecho_client_t * client,
197 : ulong conn_idx,
198 0 : ushort * shred_version ) {
199 0 : fd_ipecho_client_peer_t * peer = &client->peers[ conn_idx ];
200 :
201 0 : if( FD_UNLIKELY( peer->writing ) ) return 1;
202 0 : long read = recvfrom( client->pollfds[ conn_idx ].fd,
203 0 : peer->response+peer->response_bytes_read,
204 0 : sizeof(peer->response)-peer->response_bytes_read,
205 0 : 0,
206 0 : NULL,
207 0 : NULL );
208 0 : if( FD_UNLIKELY( -1==read && (errno==EAGAIN || errno==EINTR) ) ) return 1;
209 0 : else if( FD_UNLIKELY( -1==read ) ) {
210 0 : close_one( client, conn_idx );
211 0 : return 1;
212 0 : }
213 :
214 0 : peer->response_bytes_read += (ulong)read;
215 0 : if( FD_UNLIKELY( read ) ) return 1;
216 :
217 0 : int response = fd_ipecho_client_parse_response( peer->response,
218 0 : peer->response_bytes_read,
219 0 : shred_version );
220 0 : if( FD_LIKELY( response==FD_IPECHO_PARSE_OK ) ) return 0;
221 0 : else {
222 0 : close_one( client, conn_idx );
223 0 : return 1;
224 0 : }
225 0 : }
226 :
227 : int
228 : fd_ipecho_client_poll( fd_ipecho_client_t * client,
229 : long now,
230 : ushort * shred_version,
231 0 : int * charge_busy ) {
232 0 : if( FD_UNLIKELY( !client->remaining_peer_cnt ) ) return -1;
233 0 : if( FD_UNLIKELY( LONG_MAX==client->start_time_nanos ) ) client->start_time_nanos = now;
234 0 : if( FD_UNLIKELY( now>fd_ipecho_client_deadline_nanos( client ) ) ) {
235 0 : close_all( client );
236 0 : return -1;
237 0 : }
238 :
239 0 : int nfds = fd_syscall_poll( client->pollfds, (uint)client->peer_cnt, 0 );
240 0 : if( FD_UNLIKELY( 0==nfds ) ) return 1;
241 0 : else if( FD_UNLIKELY( -1==nfds && errno==EINTR ) ) return 1;
242 0 : else if( FD_UNLIKELY( -1==nfds ) ) FD_LOG_ERR(( "poll() failed (%i-%s)", errno, strerror( errno ) ));
243 :
244 0 : for( ulong i=0UL; i<16UL; i++ ) {
245 0 : if( FD_UNLIKELY( -1==client->pollfds[ i ].fd ) ) continue;
246 :
247 0 : if( FD_LIKELY( client->pollfds[ i ].revents & POLLOUT ) ) {
248 0 : if( FD_UNLIKELY( client->peers[ i ].writing ) ) *charge_busy = 1;
249 0 : write_conn( client, i );
250 0 : }
251 0 : if( FD_UNLIKELY( -1==client->pollfds[ i ].fd ) ) continue;
252 0 : if( FD_LIKELY( client->pollfds[ i ].revents & POLLIN ) ) {
253 0 : if( FD_UNLIKELY( !client->peers[ i ].writing ) ) *charge_busy = 1;
254 0 : if( FD_LIKELY( !read_conn( client, i, shred_version ) ) ) {
255 0 : close_all( client );
256 0 : return 0;
257 0 : }
258 0 : }
259 0 : }
260 :
261 0 : return 1;
262 0 : }
263 :
264 : struct pollfd const *
265 0 : fd_ipecho_client_get_pollfds( fd_ipecho_client_t * client ) {
266 0 : return client->pollfds;
267 0 : }
268 :
269 : long
270 0 : fd_ipecho_client_deadline_nanos( fd_ipecho_client_t const * client ) {
271 0 : if( FD_UNLIKELY( LONG_MAX==client->start_time_nanos ) ) return LONG_MAX;
272 0 : return client->start_time_nanos+2L*1000L*1000L*1000L;
273 0 : }
|
