Alex Hultman
277 lines
8.1 KiB
C
277 lines
8.1 KiB
C
/* This is a scalability test for testing million(s) of pinging connections */
|
|
|
|
#include <libusockets.h>
|
|
int SSL;
|
|
|
|
#include <stdio.h>
|
|
#include <stdlib.h>
|
|
#include <string.h>
|
|
#include <time.h>
|
|
#include <stdint.h>
|
|
|
|
unsigned char web_socket_request[26] = {130, 128 | 20, 1, 2, 3, 4};
|
|
|
|
char request[] = "GET / HTTP/1.1\r\n"
|
|
"Upgrade: websocket\r\n"
|
|
"Connection: Upgrade\r\n"
|
|
"Sec-WebSocket-Key: x3JJHMbDL1EzLkh9GBhXDw==\r\n"
|
|
"Host: server.example.com\r\n"
|
|
"Sec-WebSocket-Version: 13\r\n\r\n";
|
|
char *host;
|
|
int port;
|
|
int connections;
|
|
|
|
/* All the ips we as client can use */
|
|
char **ips;
|
|
int num_ips;
|
|
|
|
/* Send ping every 16 seconds */
|
|
int WEBSOCKET_PING_INTERVAL = 8;
|
|
|
|
/* We only establish 20k connections per address */
|
|
int CONNECTIONS_PER_ADDRESS = 20000;
|
|
|
|
/* How many connections a time */
|
|
int BATCH_CONNECT = 1;
|
|
|
|
/* Currently open and alive connections */
|
|
int opened_connections;
|
|
/* Dead connections */
|
|
int closed_connections;
|
|
|
|
struct http_socket {
|
|
/* How far we have streamed our websocket request */
|
|
int offset;
|
|
|
|
/* How far we have streamed our upgrade request */
|
|
int upgrade_offset;
|
|
};
|
|
|
|
struct us_socket_t *next_connection_failed = 0;
|
|
|
|
/* We don't need any of these */
|
|
void on_pre(struct us_loop_t *loop) {
|
|
|
|
}
|
|
|
|
/* This is not HTTP POST, it is merely an event emitted post loop iteration */
|
|
void on_post(struct us_loop_t *loop) {
|
|
|
|
}
|
|
|
|
void next_connection(struct us_socket_t *s) {
|
|
/* We could wait with this until properly upgraded */
|
|
if (--connections/* > BATCH_CONNECT*/) {
|
|
/* Swap address */
|
|
int address = opened_connections / CONNECTIONS_PER_ADDRESS;
|
|
|
|
if (us_socket_context_connect(SSL, us_socket_context(SSL, s), host, port, ips[address], 0, sizeof(struct http_socket)) == 0) {
|
|
printf("Next connection failed immediately\n");
|
|
|
|
/* Try agsin next event loop iteration */
|
|
next_connection_failed = s;
|
|
us_wakeup_loop(us_socket_context_loop(0, us_socket_context(0, s)));
|
|
|
|
}
|
|
}
|
|
}
|
|
|
|
void on_wakeup(struct us_loop_t *loop) {
|
|
if (next_connection_failed) {
|
|
struct us_socket_t *s = next_connection_failed;
|
|
next_connection_failed = 0;
|
|
next_connection(s);
|
|
}
|
|
}
|
|
|
|
struct us_socket_t *on_http_socket_writable(struct us_socket_t *s) {
|
|
struct http_socket *http_socket = (struct http_socket *) us_socket_ext(SSL, s);
|
|
|
|
/* Are we still not upgraded yet? */
|
|
if (http_socket->upgrade_offset < sizeof(request) - 1) {
|
|
http_socket->upgrade_offset += us_socket_write(SSL, s, request + http_socket->upgrade_offset, sizeof(request) - 1 - http_socket->upgrade_offset, 0);
|
|
|
|
/* Now we should be */
|
|
if (http_socket->upgrade_offset == sizeof(request) - 1) {
|
|
next_connection(s);
|
|
|
|
/* Make sure to send ping */
|
|
us_socket_timeout(SSL, s, WEBSOCKET_PING_INTERVAL);
|
|
}
|
|
} else {
|
|
/* Stream whatever is remaining of the request */
|
|
http_socket->offset += us_socket_write(SSL, s, (char *) web_socket_request + http_socket->offset, sizeof(web_socket_request) - http_socket->offset, 0);
|
|
if (http_socket->offset == sizeof(web_socket_request)) {
|
|
/* Reset timeout if we managed to */
|
|
us_socket_timeout(SSL, s, WEBSOCKET_PING_INTERVAL);
|
|
}
|
|
}
|
|
|
|
return s;
|
|
}
|
|
|
|
struct us_socket_t *on_http_socket_close(struct us_socket_t *s, int code, void *reason) {
|
|
|
|
closed_connections++;
|
|
if (closed_connections % 1000 == 0) {
|
|
printf("Alive: %d, dead: %d\n", opened_connections, closed_connections);
|
|
}
|
|
|
|
return s;
|
|
}
|
|
|
|
struct us_socket_t *on_http_socket_end(struct us_socket_t *s) {
|
|
return us_socket_close(SSL, s, 0, NULL);
|
|
}
|
|
|
|
// should never get a response!
|
|
struct us_socket_t *on_http_socket_data(struct us_socket_t *s, char *data, int length) {
|
|
|
|
// is this a broadcasted unix time in millis?
|
|
if (length % 10 == 0) {
|
|
|
|
// data sent first will come first, so it is oldest
|
|
|
|
struct timespec ts;
|
|
timespec_get(&ts, TIME_UTC);
|
|
|
|
int64_t millis = ts.tv_sec * 1000 + ts.tv_nsec / 1000000;
|
|
|
|
|
|
int64_t received_millis;
|
|
memcpy(&received_millis, data + 2, 8);
|
|
|
|
static int counter;
|
|
static int max_latency;
|
|
static long average_latency;
|
|
|
|
int latency = millis - received_millis;
|
|
|
|
average_latency += latency;
|
|
|
|
if (max_latency < latency) {
|
|
max_latency = latency;
|
|
}
|
|
|
|
if (++counter % 10000 == 0) {
|
|
printf("Alive: %d, dead: %d\n", opened_connections, closed_connections);
|
|
printf("Max latency: %d ms\n", max_latency);
|
|
printf("Average latency: %ld ms\n\n", average_latency / 10000);
|
|
max_latency = 0;
|
|
average_latency = 0;
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
return s;
|
|
}
|
|
|
|
struct us_socket_t *on_http_socket_open(struct us_socket_t *s, int is_client, char *ip, int ip_length) {
|
|
struct http_socket *http_socket = (struct http_socket *) us_socket_ext(SSL, s);
|
|
|
|
/* Display number of opened connections */
|
|
opened_connections++;
|
|
if (opened_connections % 1000 == 0) {
|
|
printf("Alive: %d, dead: %d\n", opened_connections, closed_connections);
|
|
}
|
|
|
|
/* Send an upgrade request */
|
|
http_socket->upgrade_offset = us_socket_write(SSL, s, request, sizeof(request) - 1, 0);
|
|
if (http_socket->upgrade_offset == sizeof(request) - 1) {
|
|
next_connection(s);
|
|
|
|
/* Make sure to send ping */
|
|
us_socket_timeout(SSL, s, WEBSOCKET_PING_INTERVAL);
|
|
}
|
|
|
|
return s;
|
|
}
|
|
|
|
// here we should send a message as ping (part of the test)
|
|
struct us_socket_t *on_http_socket_timeout(struct us_socket_t *s) {
|
|
struct http_socket *http_socket = (struct http_socket *) us_socket_ext(SSL, s);
|
|
|
|
/* Send ping here */
|
|
http_socket->offset = us_socket_write(SSL, s, (char *) web_socket_request, sizeof(web_socket_request), 0);
|
|
if (http_socket->offset == sizeof(web_socket_request)) {
|
|
/* Reset timeout if we managed to */
|
|
us_socket_timeout(SSL, s, WEBSOCKET_PING_INTERVAL);
|
|
}
|
|
|
|
return s;
|
|
}
|
|
|
|
struct us_socket_t *on_http_socket_connect_error(struct us_socket_t *s, int code) {
|
|
|
|
printf("Connection failed\n");
|
|
|
|
next_connection(s);
|
|
|
|
return s;
|
|
}
|
|
|
|
int main(int argc, char **argv) {
|
|
|
|
/* Parse host and port */
|
|
if (argc < 5) {
|
|
printf("Usage: connections host port ssl [ip ...]\n");
|
|
return 0;
|
|
}
|
|
|
|
port = atoi(argv[3]);
|
|
host = malloc(strlen(argv[2]) + 1);
|
|
memcpy(host, argv[2], strlen(argv[2]) + 1);
|
|
connections = atoi(argv[1]);
|
|
SSL = atoi(argv[4]);
|
|
|
|
/* Do we have ip addresses? */
|
|
if (argc > 5) {
|
|
ips = &argv[5];
|
|
num_ips = argc - 5;
|
|
|
|
for (int i = 0; i < num_ips; i++) {
|
|
printf("%s\n", ips[i]);
|
|
}
|
|
} else {
|
|
static char *default_ips[] = {""};
|
|
ips = default_ips;
|
|
num_ips = 1;
|
|
}
|
|
|
|
/* Check so that we have enough ip addresses */
|
|
if (num_ips <= connections / CONNECTIONS_PER_ADDRESS) {
|
|
printf("You'll need more IP addresses for this run\n");
|
|
return 0;
|
|
}
|
|
|
|
/* Create the event loop */
|
|
struct us_loop_t *loop = us_create_loop(0, on_wakeup, on_pre, on_post, 0);
|
|
|
|
/* Create a socket context for HTTP */
|
|
struct us_socket_context_options_t options = {};
|
|
struct us_socket_context_t *http_context = us_create_socket_context(SSL, loop, 0, options);
|
|
|
|
/* Set up event handlers */
|
|
us_socket_context_on_open(SSL, http_context, on_http_socket_open);
|
|
us_socket_context_on_data(SSL, http_context, on_http_socket_data);
|
|
us_socket_context_on_writable(SSL, http_context, on_http_socket_writable);
|
|
us_socket_context_on_close(SSL, http_context, on_http_socket_close);
|
|
us_socket_context_on_timeout(SSL, http_context, on_http_socket_timeout);
|
|
us_socket_context_on_end(SSL, http_context, on_http_socket_end);
|
|
us_socket_context_on_connect_error(SSL, http_context, on_http_socket_connect_error);
|
|
|
|
/* Start making HTTP connections */
|
|
for (int i = 0; i < BATCH_CONNECT; i++) {
|
|
if (us_socket_context_connect(SSL, http_context, host, port, ips[0], 0, sizeof(struct http_socket)) == 0) {
|
|
printf("Connection failed immediately\n");
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
us_loop_run(loop);
|
|
}
|