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
|
/* An HTTP load generator speaking the tracker's own protocol.
Each request is its own connection, which is what a BitTorrent client
normally does and what the tracker's hot path is built for: accept, read
one request, write a short reply, close. That makes this a measurement of
connection cost rather than of request cost, and connection cost is where
the tracker's system calls are.
cc -O2 -pthread -o http_load tests/http_load.c
./http_load 127.0.0.1 6969 8 5 10000
Arguments: host, port, threads, seconds, and how many distinct torrents to
announce. Cycling a fixed set reaches a steady state where almost every
announce renews a peer that is already there, rather than measuring how
fast the tracker can insert torrents it will never see again. */
#include <arpa/inet.h>
#include <errno.h>
#include <netinet/in.h>
#include <netinet/tcp.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 <unistd.h>
static struct sockaddr_in g_tracker;
static long g_torrents = 10000;
static double g_seconds = 5;
static _Atomic long long g_replies = 0;
static _Atomic long long g_failed = 0;
static _Atomic int g_running = 1;
static double now_seconds(void) {
struct timeval tv;
gettimeofday(&tv, NULL);
return tv.tv_sec + tv.tv_usec / 1000000.0;
}
/* %xx for all twenty bytes, which is what a client sends and what the
tracker's urlencode reader has to walk */
static void hash_of(long which, char *out) {
int i;
for (i = 0; i < 20; ++i) {
unsigned byte = (unsigned)((which >> ((i % 4) * 8)) & 0xff) ^ (unsigned)(i * 37);
sprintf(out + i * 3, "%%%02x", byte & 0xff);
}
}
static void *worker(void *args) {
long index = (long)(intptr_t)args;
long which = index * 7919;
char request[512], hash[61], reply[2048];
while (atomic_load_explicit(&g_running, memory_order_relaxed)) {
int sock = socket(AF_INET, SOCK_STREAM, 0);
int one = 1;
ssize_t got, sent;
size_t length;
if (sock < 0) {
atomic_fetch_add_explicit(&g_failed, 1, memory_order_relaxed);
continue;
}
setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
if (connect(sock, (struct sockaddr *)&g_tracker, sizeof(g_tracker)) < 0) {
atomic_fetch_add_explicit(&g_failed, 1, memory_order_relaxed);
close(sock);
continue;
}
hash_of(which, hash);
which = (which + 1) % g_torrents;
length = (size_t)snprintf(request, sizeof(request),
"GET /announce?info_hash=%s&peer_id=-OT0001-%06ld%06ld&port=%u&left=0 HTTP/1.0\r\n\r\n",
hash, index, which, (unsigned)(6881 + index));
for (sent = 0; (size_t)sent < length;) {
ssize_t wrote = write(sock, request + sent, length - (size_t)sent);
if (wrote <= 0)
break;
sent += wrote;
}
/* Read to end of reply, which the tracker marks by closing */
got = 0;
for (;;) {
ssize_t piece = read(sock, reply, sizeof(reply));
if (piece <= 0)
break;
got += piece;
}
close(sock);
if (got > 0)
atomic_fetch_add_explicit(&g_replies, 1, memory_order_relaxed);
else
atomic_fetch_add_explicit(&g_failed, 1, memory_order_relaxed);
}
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]) : 4;
pthread_t *pool;
double started, elapsed;
long i;
g_seconds = argc > 4 ? atof(argv[4]) : 5;
g_torrents = argc > 5 ? atol(argv[5]) : 10000;
if (g_torrents < 1)
g_torrents = 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;
}
pool = calloc((size_t)threads, sizeof(pthread_t));
started = now_seconds();
for (i = 0; i < threads; ++i)
pthread_create(pool + i, NULL, worker, (void *)(intptr_t)i);
while (now_seconds() - started < g_seconds)
usleep(20000);
atomic_store(&g_running, 0);
for (i = 0; i < threads; ++i)
pthread_join(pool[i], NULL);
elapsed = now_seconds() - started;
printf("%.0f replies/s (%lld replies, %lld failed, %ld threads, %.1f s)\n",
(double)atomic_load(&g_replies) / elapsed, (long long)atomic_load(&g_replies),
(long long)atomic_load(&g_failed), threads, elapsed);
free(pool);
return 0;
}
|
