diff options
Diffstat (limited to 'research')
| -rw-r--r-- | research/experiments/019-idle-links/links.c | 42 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/run.py | 57 |
2 files changed, 88 insertions, 11 deletions
diff --git a/research/experiments/019-idle-links/links.c b/research/experiments/019-idle-links/links.c index d6cc28b..7c726c2 100644 --- a/research/experiments/019-idle-links/links.c +++ b/research/experiments/019-idle-links/links.c @@ -177,17 +177,22 @@ static PureLinkSender *sender_of(size_t device) { return s; } -/* What a sender has to send, to its socket. */ -static void send_packets(PureLinkSender *s, int fd, uint64_t now) { +/* What a sender has to send, to its socket: false when the system said + * there is no way there (no route, nobody at the port). */ +static bool send_packets(PureLinkSender *s, int fd, uint64_t now) { unsigned char packet[PURE_LINK_DATAGRAM_MOST]; uint64_t number; + bool way = true; for (size_t size; (size = pure_link_sender_packet(s, packet, false, now, &number));) { - if (send(fd, packet, size, 0) < 0 && (errno == EAGAIN || errno == ENOBUFS)) { + ssize_t sent = send(fd, packet, size, 0); + if (sent < 0 && (errno == EAGAIN || errno == ENOBUFS)) { pure_link_sender_refused(s, number); break; } + if (sent < 0 && (errno == ECONNREFUSED || errno == EHOSTUNREACH || errno == ENETUNREACH)) way = false; ++sends; } + return way; } /* Answers waiting on fd, to the sender. */ @@ -293,9 +298,32 @@ typedef struct { uint64_t made, stored; /* records made and stored: numbers 1..made, 1..stored */ uint64_t next_make; /* when the next record is made */ uint64_t due; /* its sender's next timer (0: none), as it was when last touched */ + unsigned looked; /* its probes unanswered when it last looked for the node */ + bool nowhere; /* a send failed: the system saw no way there */ uint64_t made_at[OUTBOX]; /* when each record not stored was made (ns), by number % OUTBOX */ } Device; +/* A device whose probes go unanswered three times in a row, or whose send + * the system said had no way there, looks for the node again, as + * link/link.c does: its name looked up, and a new socket connected there in + * place of the one before (before the next packet goes). */ +static void look(Device *d, int poller, size_t k, const char *host, const char *port) { + struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found; + if (getaddrinfo(host, port, &hints, &found)) return; + int fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK, 0); + bool ok = fd >= 0 && !connect(fd, found->ai_addr, found->ai_addrlen); + freeaddrinfo(found); + if (!ok) { + if (fd >= 0) close(fd); + return; + } + epoll_ctl(poller, EPOLL_CTL_DEL, d->fd, NULL); + close(d->fd); + d->fd = fd; + struct epoll_event e = {.events = EPOLLIN, .data.u64 = k}; + epoll_ctl(poller, EPOLL_CTL_ADD, d->fd, &e); +} + static int run_all(char **argv) { size_t first = strtoull(argv[8], NULL, 10), count = strtoull(argv[9], NULL, 10); uint64_t period = strtoull(argv[10], NULL, 10) * 1000000u, seconds = strtoull(argv[11], NULL, 10); @@ -342,8 +370,9 @@ static int run_all(char **argv) { ++stored; } if (heard > d->stored) d->stored = heard; - send_packets(d->sender, d->fd, now); /* what the answer let go */ + d->nowhere |= !send_packets(d->sender, d->fd, now); /* what the answer let go */ d->due = pure_link_sender_due(d->sender); + if (!pure_link_sender_probes(d->sender)) d->looked = 0; } if (now - last_scan < 5000000u) continue; /* the devices' clocks, every 5 ms */ last_scan = now; @@ -361,7 +390,10 @@ static int run_all(char **argv) { * sender is touched (so ten thousand at rest cost a glance each). */ if (d->due && d->due <= now) pure_link_sender_tick(d->sender, now), more = true; if (more) { - send_packets(d->sender, d->fd, now); + unsigned probes = pure_link_sender_probes(d->sender); + if ((probes >= 3 && probes != d->looked) || d->nowhere) look(d, poller, k, argv[5], argv[6]); + d->looked = probes; + d->nowhere = !send_packets(d->sender, d->fd, now); d->due = pure_link_sender_due(d->sender); } } diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py index 15ae2de..7f2647d 100644 --- a/research/experiments/019-idle-links/run.py +++ b/research/experiments/019-idle-links/run.py @@ -22,7 +22,11 @@ send it until it is stored; at --kill-at seconds the receiver is killed both sides before and after, the X25519 made (none when the node remembers its secrets; --no-memo forgets them), and that every order made was applied once. With --quic too, the same for quinn clients, which -connect again when their connection is lost. +connect again when their connection is lost. With --move too (M3), in +namespaces of its own (unshare -rmn), the devices send to receiver.test +(127.0.0.2), and the node is started again elsewhere: its store, key and +memo copied, 127.0.0.2 made unreachable, the name pointed at 127.0.0.3; +the devices look the name up again when their probes go unanswered. With --first-contact DELAYS (M4), for each one-way delay in ms, in a network namespace of its own (unshare -rn) whose loopback delays UDP: @@ -236,9 +240,10 @@ def recovery(timeline, killed_ms): return {'steady_per_100ms': round(steady, 1), 'silent_100ms_steps': silent, 'caught_up_ms': None} -def restart_pure(n, work, keys, module, period_ms, seconds, kill_at, memo): +def restart_pure(n, work, keys, module, period_ms, seconds, kill_at, memo, host='127.0.0.1', move=None): """M2 for Pure: n devices send a record every period_ms; the node is - killed (-9) at kill_at seconds and started again at once.""" + killed (-9) at kill_at seconds and started again at once (M3, with move: + elsewhere, move(args) giving its arguments there).""" senders = work / f'senders-{n}' with open(f'{keys}.senders') as source, open(senders, 'w') as out: for _, line in zip(range(n), source): @@ -249,7 +254,7 @@ def restart_pure(n, work, keys, module, period_ms, seconds, kill_at, memo): args = [ROOT / 'build/pure-serve', '--module', module, '--store', store, '--port', port, '--key', work / 'node.key', '--link-port', link_port, '--senders', senders] node, info = start(args) - generator = subprocess.Popen([str(a) for a in [ROOT / 'build/links', 'run', keys, module, 'order', '127.0.0.1', + generator = subprocess.Popen([str(a) for a in [ROOT / 'build/links', 'run', keys, module, 'order', host, link_port, info['public_key'], 0, n, period_ms, seconds]], stdout=subprocess.PIPE, stderr=subprocess.PIPE) for line in generator.stderr: # progress, a line a second @@ -267,8 +272,11 @@ def restart_pure(n, work, keys, module, period_ms, seconds, kill_at, memo): killed_ms = int((t + killed - seen) * 1000) # on the generator's clock if not memo: (work / 'node.key.links').unlink() + if move: + args = move(args) node, _ = start(args) - row = {'devices': n, 'period_ms': period_ms, 'memo': memo, 'restart_ms': round((time.monotonic() - killed) * 1000, 1), + row = {'devices': n, 'period_ms': period_ms, 'memo': memo, 'moved': bool(move), + 'restart_ms': round((time.monotonic() - killed) * 1000, 1), 'steady_cpu_ms_per_s': round(steady_cpu, 1), 'senders_steady_cpu_ms_per_s': round(senders_steady, 1)} cpu_start, senders_start = cpu_ms(node.pid), cpu_ms(generator.pid) time.sleep(10) @@ -290,6 +298,28 @@ def restart_pure(n, work, keys, module, period_ms, seconds, kill_at, memo): return row +def mover(hosts, ip): + """M3: the node's store, key and memo carried to another machine (copied, + in the namespace of --move), its old address (127.0.0.2) made + unreachable, and the name the devices send to (receiver.test) pointed at + the new one (127.0.0.3).""" + import shutil + + def move(args): + args = list(args) + store, key = Path(args[args.index('--store') + 1]), Path(args[args.index('--key') + 1]) + moved_store, moved_key = store.with_name(store.name + '-moved'), key.with_name('moved-' + key.name) + shutil.copytree(store, moved_store) + shutil.copy(key, moved_key) + if Path(f'{key}.links').exists(): + shutil.copy(f'{key}.links', f'{moved_key}.links') + subprocess.run([ip, 'route', 'add', 'unreachable', '127.0.0.2/32', 'table', 'local'], check=True) + hosts.write_text('127.0.0.1 localhost\n127.0.0.3 receiver.test\n') + args[args.index('--store') + 1], args[args.index('--key') + 1] = moved_store, moved_key + return args + return move + + def restart_quic(n, work, period_ms, seconds, kill_at): """M2 for quinn: the same load; the service is killed (-9) and started again on its port, with the same key (so it resets the old connections).""" @@ -539,6 +569,9 @@ def main(): parser.add_argument('--quic', action='store_true', help='measure the quinn service instead (quic/)') parser.add_argument('--restart', action='store_true', help='M2: kill -9 the receiver under load (with --quic, quinn)') parser.add_argument('--no-memo', action='store_true', help='M2: the node forgets the secrets it remembered') + parser.add_argument('--move', action='store_true', + help='M3 (with --restart): started again on another machine, the store carried there (a ' + 'namespace of its own: unshare -rmn)') parser.add_argument('--period-ms', type=int, default=5000) parser.add_argument('--seconds', type=int, default=40) parser.add_argument('--kill-at', type=int, default=20) @@ -551,6 +584,17 @@ def main(): parser.add_argument('--rounds', type=int, default=1) args = parser.parse_args() counts = [int(c) for c in args.counts.split(',')] + if args.move and not os.environ.get('PURE_MOVE_HOSTS'): # in namespaces of its own: its network, its /etc/hosts + with tempfile.TemporaryDirectory(prefix='pure-m3-', dir=args.store) as directory: + hosts = Path(directory) / 'hosts' + hosts.write_text('127.0.0.1 localhost\n127.0.0.2 receiver.test\n') + done = subprocess.run(['unshare', '-rmn', sys.executable] + sys.argv, env={**os.environ, 'PURE_MOVE_HOSTS': str(hosts)}) + sys.exit(done.returncode) + if args.move: + import shutil + ip = shutil.which('ip') or '/usr/sbin/ip' + subprocess.run([ip, 'link', 'set', 'lo', 'up'], check=True) + subprocess.run(['mount', '--bind', os.environ['PURE_MOVE_HOSTS'], '/etc/hosts'], check=True) if args.transport: subprocess.run(['make', '-s', '-C', str(ROOT), 'build/libpure.a', 'build/libpure-link.a', 'build/pure-crypto.so'], check=True) @@ -614,7 +658,8 @@ def main(): subprocess.run([str(ROOT / 'build/links'), 'keys', str(max(counts)), str(keys)], check=True) for n in counts: row = restart_quic(n, work, args.period_ms, args.seconds, args.kill_at) if args.quic else \ - restart_pure(n, work, keys, module, args.period_ms, args.seconds, args.kill_at, not args.no_memo) + restart_pure(n, work, keys, module, args.period_ms, args.seconds, args.kill_at, not args.no_memo, + *(('receiver.test', mover(Path(os.environ['PURE_MOVE_HOSTS']), ip)) if args.move else ())) line = json.dumps(row) print(line, flush=True) if args.out: |