Line data Source code
1 : /**
2 :
3 : export RUST_LOG=solana_repair=TRACE
4 : cargo run --bin solana-test-validator
5 :
6 : build/native/gcc/bin/fd_repair_tool --peer_id 75dLVGm338wpo2SsfM7pWestidAjJL1Y9nw9Rb1x7yQQ --slot 1533:0,1534:0
7 :
8 : **/
9 :
10 : #define _GNU_SOURCE /* See feature_test_macros(7) */
11 :
12 : #include "fd_repair.h"
13 : #include "../fd_flamenco.h"
14 : #include "../../util/fd_util.h"
15 : #include "../../ballet/base58/fd_base58.h"
16 : #include "../types/fd_types_yaml.h"
17 : #include "../../util/net/fd_eth.h"
18 : #include <stdio.h>
19 : #include <unistd.h>
20 : #include <signal.h>
21 : #include <sys/socket.h>
22 : #include <netinet/in.h>
23 : #include <arpa/inet.h>
24 : #include <sys/random.h>
25 : #include <errno.h>
26 : #include <netdb.h>
27 : #include <stdlib.h>
28 :
29 : // SIGINT signal handler
30 : volatile int stopflag = 0;
31 0 : static void stop(int sig) { (void)sig; stopflag = 1; }
32 :
33 : static int sockfd = -1;
34 :
35 : /* Convert my style of address to UNIX style */
36 : static int
37 0 : repair_to_sockaddr( uchar * dst, fd_repair_peer_addr_t const * src ) {
38 0 : fd_memset(dst, 0, sizeof(struct sockaddr_in));
39 0 : struct sockaddr_in * t = (struct sockaddr_in *)dst;
40 0 : t->sin_family = AF_INET;
41 0 : t->sin_addr.s_addr = src->addr;
42 0 : t->sin_port = src->port;
43 0 : return sizeof(struct sockaddr_in);
44 0 : }
45 :
46 : /* Convert my style of address from UNIX style */
47 : static int
48 0 : repair_from_sockaddr( fd_repair_peer_addr_t * dst, uchar const * src ) {
49 0 : FD_STATIC_ASSERT(sizeof(fd_repair_peer_addr_t) == sizeof(ulong),"messed up size");
50 0 : dst->l = 0;
51 0 : const struct sockaddr_in * sa = (const struct sockaddr_in *)src;
52 0 : dst->addr = sa->sin_addr.s_addr;
53 0 : dst->port = sa->sin_port;
54 0 : return 0;
55 0 : }
56 :
57 : static void
58 : send_packet( uchar const * data,
59 : size_t sz,
60 : fd_repair_peer_addr_t const * addr,
61 : uint src_ip4_addr FD_PARAM_UNUSED,
62 0 : void * arg FD_PARAM_UNUSED ) {
63 : // FD_LOG_HEXDUMP_NOTICE(("send: ", data, sz));
64 0 : uchar saddr[sizeof(struct sockaddr_in)];
65 0 : int saddrlen = repair_to_sockaddr(saddr, addr);
66 0 : if ( sendto(sockfd, data, sz, MSG_DONTWAIT,
67 0 : (const struct sockaddr *)saddr, (socklen_t)saddrlen) < 0 ) {
68 0 : FD_LOG_WARNING(("sendto failed: %s", strerror(errno)));
69 0 : }
70 0 : }
71 :
72 : /* Convert a host:port string to a repair network address. If host is
73 : * missing, it assumes the local hostname. */
74 : static fd_repair_peer_addr_t *
75 0 : resolve_hostport(const char* str /* host:port */, fd_repair_peer_addr_t * res) {
76 0 : fd_memset(res, 0, sizeof(fd_repair_peer_addr_t));
77 :
78 : /* Find the : and copy out the host */
79 0 : char buf[128];
80 0 : uint i;
81 0 : for (i = 0; ; ++i) {
82 0 : if (str[i] == '\0' || i > sizeof(buf)-1U) {
83 0 : FD_LOG_ERR(("missing colon"));
84 0 : return NULL;
85 0 : }
86 0 : if (str[i] == ':') {
87 0 : buf[i] = '\0';
88 0 : break;
89 0 : }
90 0 : buf[i] = str[i];
91 0 : }
92 0 : if (i == 0)
93 : /* :port means $HOST:port */
94 0 : gethostname(buf, sizeof(buf));
95 :
96 0 : struct hostent * host = gethostbyname( buf );
97 0 : if (host == NULL) {
98 0 : FD_LOG_WARNING(("unable to resolve host %s", buf));
99 0 : return NULL;
100 0 : }
101 : /* Convert result to repair address */
102 0 : res->l = 0;
103 0 : res->addr = ((struct in_addr *)host->h_addr)->s_addr;
104 0 : int port = atoi(str + i + 1);
105 0 : if (port < 1024 || port > (int)USHORT_MAX) {
106 0 : FD_LOG_ERR(("invalid port number"));
107 0 : return NULL;
108 0 : }
109 0 : res->port = htons((ushort)port);
110 :
111 0 : return res;
112 0 : }
113 :
114 : static int
115 0 : main_loop( int * argc, char *** argv, fd_repair_t * glob, fd_repair_config_t * config, volatile int * stopflag ) {
116 0 : int fd;
117 0 : if ((fd = socket(AF_INET, SOCK_DGRAM, 0)) < 0) {
118 0 : FD_LOG_ERR(("socket failed: %s", strerror(errno)));
119 0 : return -1;
120 0 : }
121 0 : sockfd = fd;
122 0 : int optval = 1<<20;
123 0 : if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char *)&optval, sizeof(int)) < 0) {
124 0 : FD_LOG_ERR(("setsocketopt failed: %s", strerror(errno)));
125 0 : return -1;
126 0 : }
127 0 : if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char *)&optval, sizeof(int)) < 0) {
128 0 : FD_LOG_ERR(("setsocketopt failed: %s", strerror(errno)));
129 0 : return -1;
130 0 : }
131 0 : uchar saddr[sizeof(struct sockaddr_in6)];
132 0 : int saddrlen = repair_to_sockaddr(saddr, &config->intake_addr);
133 0 : if (saddrlen < 0 || bind(fd, (struct sockaddr*)saddr, (uint)saddrlen) < 0) {
134 0 : FD_LOG_ERR(("bind failed: %s", strerror(errno)));
135 0 : return -1;
136 0 : }
137 :
138 0 : fd_repair_settime(glob, fd_log_wallclock());
139 0 : fd_repair_start(glob);
140 :
141 0 : char const * id_cstr = fd_env_strip_cmdline_cstr ( argc, argv, "--peer_id", NULL, NULL );
142 0 : if ( id_cstr == NULL )
143 0 : FD_LOG_ERR(("--peer_id command line argument required"));
144 0 : fd_pubkey_t id;
145 0 : fd_base58_decode_32(id_cstr, id.uc);
146 0 : char const * addr_cstr = fd_env_strip_cmdline_cstr ( argc, argv, "--peer_addr", NULL, "127.0.0.1:1032" );
147 0 : fd_repair_peer_addr_t peeraddr;
148 0 : if ( fd_repair_add_active_peer(glob, resolve_hostport(addr_cstr, &peeraddr), &id) )
149 0 : return -1;
150 0 : fd_repair_add_sticky(glob, &id);
151 :
152 0 : char const * slot_cstr = fd_env_strip_cmdline_cstr ( argc, argv, "--slot", NULL, NULL );
153 0 : if ( slot_cstr == NULL )
154 0 : FD_LOG_ERR(("--slot command line argument required"));
155 :
156 0 : ulong smax = 8;
157 0 : ulong depth = 1<<17;
158 0 : char * smem = malloc(fd_scratch_smem_footprint(smax) + fd_scratch_fmem_footprint(depth));
159 0 : fd_scratch_attach( smem, smem + fd_scratch_smem_footprint(smax), smax, depth );
160 :
161 0 : #define VLEN 32U
162 0 : struct mmsghdr msgs[VLEN];
163 0 : struct iovec iovecs[VLEN];
164 0 : uchar bufs[VLEN][FD_ETH_PAYLOAD_MAX];
165 0 : uchar sockaddrs[VLEN][sizeof(struct sockaddr_in6)]; /* sockaddr is smaller than sockaddr_in6 */
166 :
167 0 : long lastreq = 0;
168 0 : while ( !*stopflag ) {
169 0 : long now = fd_log_wallclock();
170 :
171 0 : if (now - lastreq > (long)3e9) {
172 0 : lastreq = now;
173 0 : char* cstr = (char*)slot_cstr;
174 0 : do {
175 0 : ulong slot = strtoul(cstr, &cstr, 10);
176 0 : if ( *cstr != ':' )
177 0 : FD_LOG_ERR(("--slot takes <slot>:<idx>,<slot>:<idx>,<slot>:<idx>..."));
178 0 : ++cstr;
179 0 : long idx = strtol(cstr, &cstr, 10);
180 0 : if( idx == -1 ) {
181 0 : if( fd_repair_need_highest_window_index(glob, slot, 0U) )
182 0 : return -1;
183 0 : } else if( idx == -2 ) {
184 0 : if( fd_repair_need_orphan(glob, slot) )
185 0 : return -1;
186 0 : } else if( idx >= 0 ) {
187 0 : if( fd_repair_need_window_index(glob, slot, (uint)idx) != 1 )
188 0 : return -1;
189 0 : }
190 0 : if ( *cstr == '\0' )
191 0 : break;
192 0 : if ( *cstr != ',' )
193 0 : FD_LOG_ERR(("--slot takes <slot>:<idx>,<slot>:<idx>,<slot>:<idx>..."));
194 0 : ++cstr;
195 0 : } while (1);
196 0 : }
197 :
198 0 : fd_repair_settime(glob, now);
199 0 : fd_repair_continue(glob);
200 :
201 0 : fd_memset(msgs, 0, sizeof(msgs));
202 0 : for (uint i = 0; i < VLEN; i++) {
203 0 : iovecs[i].iov_base = bufs[i];
204 0 : iovecs[i].iov_len = FD_ETH_PAYLOAD_MAX;
205 0 : msgs[i].msg_hdr.msg_iov = &iovecs[i];
206 0 : msgs[i].msg_hdr.msg_iovlen = 1;
207 0 : msgs[i].msg_hdr.msg_name = sockaddrs[i];
208 0 : msgs[i].msg_hdr.msg_namelen = sizeof(struct sockaddr_in6);
209 0 : }
210 :
211 : /* Read more packets */
212 0 : long retval = recvmmsg(fd, msgs, VLEN, MSG_DONTWAIT, NULL);
213 0 : if (retval < 0) {
214 0 : if (errno == EINTR || errno == EWOULDBLOCK)
215 0 : continue;
216 0 : FD_LOG_ERR(("recvmmsg failed: %s", strerror(errno)));
217 0 : return -1;
218 0 : }
219 0 : if (retval == 0)
220 0 : continue;
221 :
222 0 : for (uint i = 0; i < (uint)retval; ++i) {
223 0 : fd_repair_peer_addr_t from;
224 0 : repair_from_sockaddr( &from, msgs[i].msg_hdr.msg_name );
225 0 : FD_LOG_HEXDUMP_NOTICE(("recv: ", bufs[i], (ulong)msgs[i].msg_len));
226 0 : fd_repair_recv_clnt_packet( glob, bufs[i], (ulong)msgs[i].msg_len, &from, 0 );
227 0 : }
228 0 : }
229 :
230 0 : fd_scratch_detach(NULL);
231 0 : free(smem);
232 :
233 0 : close(fd);
234 0 : return 0;
235 0 : }
236 :
237 : static void
238 0 : recv_shred(fd_shred_t const * shred, ulong shred_sz, fd_gossip_peer_addr_t const * from, fd_pubkey_t const * id, void * arg) {
239 0 : (void)from;
240 0 : (void)id;
241 0 : (void)arg;
242 0 : FD_LOG_NOTICE(( "shred variant=0x%02x sz=%lu slot=%lu idx=%u header_sz=0x%lx merkle_sz=0x%lx payload_sz=0x%lx",
243 0 : (uint)shred->variant, shred_sz, shred->slot, shred->idx, fd_shred_header_sz(shred->variant),
244 0 : fd_shred_merkle_sz(shred->variant), fd_shred_payload_sz(shred) ));
245 0 : }
246 :
247 : static void
248 : deliver_fail_fun( fd_pubkey_t const * id,
249 : ulong slot,
250 : uint shred_index,
251 : void * arg,
252 0 : int reason ) {
253 0 : (void)arg;
254 0 : FD_LOG_WARNING(( "repair_deliver_fail_fun - shred: %s, slot: %lu, idx: %u, reason: %d",
255 0 : FD_BASE58_ENC_32_ALLOCA( id ),
256 0 : slot,
257 0 : shred_index,
258 0 : reason ));
259 0 : }
260 :
261 : int main(int argc, char **argv) {
262 : fd_boot ( &argc, &argv );
263 : fd_flamenco_boot( &argc, &argv );
264 :
265 : fd_valloc_t valloc = fd_libc_alloc_virtual();
266 :
267 : fd_repair_config_t config;
268 : fd_memset(&config, 0, sizeof(config));
269 :
270 : uchar private_key[32];
271 : FD_TEST( 32UL==getrandom( private_key, 32UL, 0 ) );
272 : fd_sha512_t sha[1];
273 : fd_pubkey_t public_key;
274 : FD_TEST( fd_ed25519_public_from_private( public_key.uc, private_key, sha ) );
275 :
276 : config.private_key = private_key;
277 : config.public_key = &public_key;
278 :
279 : char hostname[64];
280 : gethostname(hostname, sizeof(hostname));
281 :
282 : char const * my_addr = fd_env_strip_cmdline_cstr ( &argc, &argv, "--my_addr", NULL, ":1125");
283 : FD_TEST( resolve_hostport(my_addr, &config.intake_addr) );
284 :
285 : config.deliver_fun = recv_shred;
286 : config.clnt_send_fun = send_packet;
287 : config.deliver_fail_fun = deliver_fail_fun;
288 :
289 : ulong seed = fd_hash(0, hostname, strnlen(hostname, sizeof(hostname)));
290 :
291 : void * shm = fd_valloc_malloc(valloc, fd_repair_align(), fd_repair_footprint());
292 : fd_repair_t * glob = fd_repair_join(fd_repair_new(shm, seed ));
293 :
294 : if ( fd_repair_set_config(glob, &config) )
295 : return 1;
296 :
297 : signal(SIGINT, stop);
298 : signal(SIGPIPE, SIG_IGN);
299 :
300 : if ( main_loop(&argc, &argv, glob, &config, &stopflag) )
301 : return 1;
302 :
303 : fd_valloc_free(valloc, fd_repair_delete(fd_repair_leave(glob) ));
304 :
305 : fd_halt();
306 :
307 : return 0;
308 : }
|
