1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
/* A UDP load generator speaking the tracker's own protocol.
Each thread keeps a socket of its own, does the BEP 15 connect handshake
once and then announces in a loop, counting replies. Separate sockets
matter: the kernel spreads datagrams across a SO_REUSEPORT group by a hash
of the four tuple, so a generator sending everything from one source port
would pile every packet onto one of the tracker's workers and measure the
wrong thing.
cc -O2 -pthread -o udp_load tests/udp_load.c
./udp_load 127.0.0.1 6969 4 5 10000 64
The last argument is how many announces each thread keeps in flight. With
one, a thread sends and waits, so the offered load is threads divided by
the round trip and the figure says more about scheduling than about the
tracker: measured here, a tracker answering 60k/s that way was using
three quarters of one core. Raising the window offers load without
needing a thread per outstanding request, which is what lets the tracker
be pushed to the point where it is actually the limit.
The last argument is how many distinct torrents to announce. It is what
keeps the figure honest: a fresh info hash every time measures how fast
the tracker can insert a million torrents into sorted vectors, which is
not what a tracker spends its life doing, and which punishes whichever
configuration is quickest because it reaches the expensive end of the run
soonest. Cycling a fixed set reaches a steady state where almost every
announce renews a peer that is already there. */
#include <arpa/inet.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#define CONNECT_MAGIC 0x41727101980ULL
static struct sockaddr_in g_tracker;
static long g_torrents = 10000;
static long g_window = 64;
static _Atomic long long g_sent = 0;
static _Atomic int g_running = 1;
static _Atomic long long g_replies = 0;
static _Atomic long long g_timeouts = 0;
static double seconds(void) {
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
return now.tv_sec + now.tv_nsec / 1e9;
}
static void put64(unsigned char *at, uint64_t value) {
int i;
for (i = 0; i < 8; ++i)
at[i] = (unsigned char)(value >> (56 - 8 * i));
}
static void put32(unsigned char *at, uint32_t value) {
at[0] = (unsigned char)(value >> 24);
at[1] = (unsigned char)(value >> 16);
at[2] = (unsigned char)(value >> 8);
at[3] = (unsigned char)value;
}
static uint64_t get64(const unsigned char *at) {
uint64_t value = 0;
int i;
for (i = 0; i < 8; ++i)
value = (value << 8) | at[i];
return value;
}
static void *loader(void *args) {
long index = (long)args;
unsigned char request[98], reply[1500];
uint64_t connection_id = 0;
int sock = socket(AF_INET, SOCK_DGRAM, 0);
struct timeval patience = {0, 50000}; /* 50 ms */
unsigned seed = (unsigned)index * 2654435761u + 1;
if (sock < 0)
return NULL;
setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &patience, sizeof(patience));
/* connect handshake */
memset(request, 0, sizeof(request));
put64(request, CONNECT_MAGIC);
put32(request + 8, 0);
put32(request + 12, (uint32_t)index);
if (sendto(sock, request, 16, 0, (struct sockaddr *)&g_tracker, sizeof(g_tracker)) != 16) {
close(sock);
return NULL;
}
if (recvfrom(sock, reply, sizeof(reply), 0, NULL, NULL) < 16) {
fprintf(stderr, " thread %ld: no answer to connect\n", index);
close(sock);
return NULL;
}
connection_id = get64(reply + 8);
/* Keep g_window announces outstanding: send that many, then drain what has
come back, rather than waiting for each reply in turn */
while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
ssize_t got;
int i;
long outstanding;
for (outstanding = 0; outstanding < g_window; ++outstanding) {
memset(request, 0, sizeof(request));
put64(request, connection_id);
put32(request + 8, 1); /* announce */
put32(request + 12, (uint32_t)index);
/* One of a fixed set, so that after the first pass almost every announce
renews a peer rather than creating a torrent. The index is hashed into
all twenty bytes so the buckets, which key off the first four, are
spread rather than walked in order. */
{
unsigned which = (unsigned)((seed = seed * 1103515245u + 12345u) >> 8) % (unsigned)g_torrents;
unsigned spread;
for (i = 0; i < 20; ++i) {
spread = which * 2654435761u + (unsigned)i * 40503u;
spread ^= spread >> 13;
request[16 + i] = (unsigned char)(spread >> 11);
}
}
for (i = 0; i < 20; ++i)
request[36 + i] = (unsigned char)('a' + (i % 26));
put64(request + 56, 0); /* downloaded */
put64(request + 64, 0); /* left: a seeder */
put64(request + 72, 0); /* uploaded */
put32(request + 80, 0); /* event */
put32(request + 84, 0); /* ip */
put32(request + 88, (uint32_t)index); /* key */
put32(request + 92, (uint32_t)-1); /* num_want */
/* A port of this thread's own, so each thread is a distinct peer and the
swarms hold more than one */
request[96] = (unsigned char)((6881 + index) >> 8);
request[97] = (unsigned char)((6881 + index) & 0xff);
if (sendto(sock, request, 98, 0, (struct sockaddr *)&g_tracker, sizeof(g_tracker)) != 98)
break;
atomic_fetch_add_explicit(&g_sent, 1, memory_order_relaxed);
}
for (outstanding = 0; outstanding < g_window; ++outstanding) {
got = recvfrom(sock, reply, sizeof(reply), 0, NULL, NULL);
if (got >= 20)
atomic_fetch_add_explicit(&g_replies, 1, memory_order_relaxed);
else {
atomic_fetch_add_explicit(&g_timeouts, 1, memory_order_relaxed);
break; /* nothing more is coming; go round and send again */
}
}
}
close(sock);
return NULL;
}
int main(int argc, char **argv) {
const char *host = argc > 1 ? argv[1] : "127.0.0.1";
int port = argc > 2 ? atoi(argv[2]) : 6969;
long threads = argc > 3 ? atol(argv[3]) : 16;
double run_for = argc > 4 ? atof(argv[4]) : 5.0;
long torrents = argc > 5 ? atol(argv[5]) : 10000;
long window = argc > 6 ? atol(argv[6]) : 64;
pthread_t *ids = calloc((size_t)threads, sizeof(pthread_t));
long i;
double started, elapsed;
g_torrents = torrents > 0 ? torrents : 1;
g_window = window > 0 ? window : 1;
memset(&g_tracker, 0, sizeof(g_tracker));
g_tracker.sin_family = AF_INET;
g_tracker.sin_port = htons((uint16_t)port);
if (inet_pton(AF_INET, host, &g_tracker.sin_addr) != 1) {
fprintf(stderr, "cannot parse address %s\n", host);
return 1;
}
for (i = 0; i < threads; ++i)
pthread_create(ids + i, NULL, loader, (void *)i);
started = seconds();
usleep((useconds_t)(run_for * 1e6));
atomic_store(&g_running, 0);
elapsed = seconds() - started;
for (i = 0; i < threads; ++i)
pthread_join(ids[i], NULL);
printf("%.0f replies/s (%.0f sent/s, %lld replies, %lld lost, %ld threads x %ld in flight, %.1f s)\n",
(double)atomic_load(&g_replies) / elapsed, (double)atomic_load(&g_sent) / elapsed,
(long long)atomic_load(&g_replies), (long long)atomic_load(&g_timeouts), threads, g_window, elapsed);
free(ids);
return 0;
}
|
