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