finalize and integrate SSL support

Makefile

@@ -1,5 +1,5 @@
 CFLAGS  := -std=c99 -Wall -O2 -D_REENTRANT
-LIBS    := -lpthread -lm
+LIBS    := -lpthread -lm -lcrypto -lssl
 
 TARGET  := $(shell uname -s | tr [A-Z] [a-z] 2>/dev/null || echo unknown)
 
@@ -8,7 +8,7 @@ ifeq ($(TARGET), sunos)
 	LIBS   += -lsocket
 endif
 
-SRC  := wrk.c aprintf.c stats.c units.c ae.c zmalloc.c http_parser.c tinymt64.c
+SRC  := wrk.c net.c ssl.c aprintf.c stats.c units.c ae.c zmalloc.c http_parser.c tinymt64.c
 BIN  := wrk
 
 ODIR := obj

README

@@ -6,21 +6,21 @@ wrk - a HTTP benchmarking tool
 
 Basic Usage
 
-  wrk -t8 -c400 -d30 http://localhost:8080/index.html
+  wrk -t12 -c400 -d30s http://127.0.0.1:8080/index.html
 
-  This runs a benchmark for 30 seconds, using 8 threads, and keeping
+  This runs a benchmark for 30 seconds, using 12 threads, and keeping
   400 HTTP connections open.
 
   Output:
 
-  Running 30s test @ http://localhost:8080/index.html
-    8 threads and 400 connections
+  Running 30s test @ http://127.0.0.1:8080/index.html
+    12 threads and 400 connections
     Thread Stats   Avg      Stdev     Max   +/- Stdev
-      Latency   439.75us  350.49us   7.60ms   92.88%
-      Req/Sec    61.13k     8.26k   72.00k    87.54%
-    10000088 requests in 19.87s, 3.42GB read
-  Requests/sec: 503396.23
-  Transfer/sec:    176.16MB
+      Latency   635.91us    0.89ms  12.92ms   93.69%
+      Req/Sec    56.20k     8.07k   62.00k    86.54%
+    22464657 requests in 30.00s, 17.76GB read
+  Requests/sec: 748868.53
+  Transfer/sec:    606.33MB
 
 Benchmarking Tips
 

src/main.h

@@ -0,0 +1,55 @@
+#ifndef MAIN_H
+#define MAIN_H
+
+#include <ctype.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <getopt.h>
+#include <math.h>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <stdarg.h>
+#include <stdbool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <signal.h>
+#include <time.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <sys/uio.h>
+
+#include "ssl.h"
+#include "aprintf.h"
+#include "stats.h"
+#include "units.h"
+#include "zmalloc.h"
+
+struct config;
+
+static void *thread_main(void *);
+static int connect_socket(thread *, connection *);
+static int reconnect_socket(thread *, connection *);
+
+static int calibrate(aeEventLoop *, long long, void *);
+static int sample_rate(aeEventLoop *, long long, void *);
+static int check_timeouts(aeEventLoop *, long long, void *);
+
+static void socket_connected(aeEventLoop *, int, void *, int);
+static void socket_writeable(aeEventLoop *, int, void *, int);
+static void socket_readable(aeEventLoop *, int, void *, int);
+static int request_complete(http_parser *);
+
+static uint64_t time_us();
+
+static char *extract_url_part(char *, struct http_parser_url *, enum http_parser_url_fields);
+static char *format_request(char *, char *, char *, char **);
+
+static int parse_args(struct config *, char **, char **, int, char **);
+static void print_stats_header();
+static void print_stats(char *, stats *, char *(*)(long double));
+static void print_stats_latency(stats *);
+
+#endif /* MAIN_H */

src/net.c

@@ -0,0 +1,32 @@
+// Copyright (C) 2013 - Will Glozer.  All rights reserved.
+
+#include <errno.h>
+#include <unistd.h>
+
+#include "net.h"
+
+status sock_connect(connection *c) {
+    return OK;
+}
+
+status sock_close(connection *c) {
+    return OK;
+}
+
+status sock_read(connection *c, size_t *n) {
+    ssize_t r = read(c->fd, c->buf, sizeof(c->buf));
+    *n = (size_t) r;
+    return r > 0 ? OK : ERROR;
+}
+
+status sock_write(connection *c, char *buf, size_t len, size_t *n) {
+    ssize_t r;
+    if ((r = write(c->fd, buf, len)) == -1) {
+        switch (errno) {
+            case EAGAIN: return RETRY;
+            default:     return ERROR;
+        }
+    }
+    *n = (size_t) r;
+    return OK;
+}

src/net.h

@@ -0,0 +1,27 @@
+#ifndef NET_H
+#define NET_H
+
+#include <stdint.h>
+#include <openssl/ssl.h>
+
+#include "wrk.h"
+
+typedef enum {
+    OK,
+    ERROR,
+    RETRY
+} status;
+
+struct sock {
+    status (*connect)(connection *);
+    status (  *close)(connection *);
+    status (   *read)(connection *, size_t *);
+    status (  *write)(connection *, char *, size_t, size_t *);
+};
+
+status sock_connect(connection *);
+status sock_close(connection *);
+status sock_read(connection *, size_t *);
+status sock_write(connection *, char *, size_t, size_t *);
+
+#endif /* NET_H */

src/ssl.c

@@ -0,0 +1,95 @@
+// Copyright (C) 2013 - Will Glozer.  All rights reserved.
+
+#include <pthread.h>
+
+#include <openssl/evp.h>
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+
+#include "ssl.h"
+
+static pthread_mutex_t *locks;
+
+static void ssl_lock(int mode, int n, const char *file, int line) {
+    pthread_mutex_t *lock = &locks[n];
+    if (mode & CRYPTO_LOCK) {
+        pthread_mutex_lock(lock);
+    } else {
+        pthread_mutex_unlock(lock);
+    }
+}
+
+static unsigned long ssl_id() {
+    return (unsigned long) pthread_self();
+}
+
+SSL_CTX *ssl_init() {
+    SSL_CTX *ctx = NULL;
+
+    SSL_load_error_strings();
+    SSL_library_init();
+    OpenSSL_add_all_algorithms();
+
+    if ((locks = calloc(CRYPTO_num_locks(), sizeof(pthread_mutex_t)))) {
+        for (int i = 0; i < CRYPTO_num_locks(); i++) {
+            pthread_mutex_init(&locks[i], NULL);
+        }
+
+        CRYPTO_set_locking_callback(ssl_lock);
+        CRYPTO_set_id_callback(ssl_id);
+
+        if ((ctx = SSL_CTX_new(TLSv1_client_method()))) {
+            SSL_CTX_set_verify(ctx, SSL_VERIFY_NONE, NULL);
+            SSL_CTX_set_verify_depth(ctx, 0);
+            SSL_CTX_set_mode(ctx, SSL_MODE_AUTO_RETRY);
+            SSL_CTX_set_session_cache_mode(ctx, SSL_SESS_CACHE_CLIENT);
+        }
+    }
+
+    return ctx;
+}
+
+status ssl_connect(connection *c) {
+    int r;
+    SSL_set_fd(c->ssl, c->fd);
+    if ((r = SSL_connect(c->ssl)) != 1) {
+        switch (SSL_get_error(c->ssl, r)) {
+            case SSL_ERROR_WANT_READ:  return RETRY;
+            case SSL_ERROR_WANT_WRITE: return RETRY;
+            default:                   return ERROR;
+        }
+    }
+    return OK;
+}
+
+status ssl_close(connection *c) {
+    SSL_shutdown(c->ssl);
+    SSL_clear(c->ssl);
+    return OK;
+}
+
+status ssl_read(connection *c, size_t *n) {
+    int r;
+    if ((r = SSL_read(c->ssl, c->buf, sizeof(c->buf))) <= 0) {
+        switch (SSL_get_error(c->ssl, r)) {
+            case SSL_ERROR_WANT_READ:  return RETRY;
+            case SSL_ERROR_WANT_WRITE: return RETRY;
+            default:                   return ERROR;
+        }
+    }
+    *n = (size_t) r;
+    return OK;
+}
+
+status ssl_write(connection *c, char *buf, size_t len, size_t *n) {
+    int r;
+    if ((r = SSL_write(c->ssl, buf, len)) <= 0) {
+        switch (SSL_get_error(c->ssl, r)) {
+            case SSL_ERROR_WANT_READ:  return RETRY;
+            case SSL_ERROR_WANT_WRITE: return RETRY;
+            default:                   return ERROR;
+        }
+    }
+    *n = (size_t) r;
+    return OK;
+}

src/ssl.h

@@ -0,0 +1,13 @@
+#ifndef SSL_H
+#define SSL_H
+
+#include "net.h"
+
+SSL_CTX *ssl_init();
+
+status ssl_connect(connection *);
+status ssl_close(connection *);
+status ssl_read(connection *, size_t *);
+status ssl_write(connection *, char *, size_t, size_t *);
+
+#endif /* SSL_H */

src/stats.c

@@ -8,42 +8,43 @@
 #include "zmalloc.h"
 
 stats *stats_alloc(uint64_t samples) {
-    stats *stats = zcalloc(sizeof(stats) + sizeof(uint64_t) * samples);
-    stats->samples = samples;
-    return stats;
+    stats *s = zcalloc(sizeof(stats) + sizeof(uint64_t) * samples);
+    s->samples = samples;
+    s->min     = UINT64_MAX;
+    return s;
 }
 
 void stats_free(stats *stats) {
     zfree(stats);
 }
 
+void stats_reset(stats *stats) {
+    stats->limit = 0;
+    stats->index = 0;
+    stats->min   = UINT64_MAX;
+    stats->max   = 0;
+}
+
 void stats_record(stats *stats, uint64_t x) {
     stats->data[stats->index++] = x;
+    if (x < stats->min) stats->min = x;
+    if (x > stats->max) stats->max = x;
     if (stats->limit < stats->samples)  stats->limit++;
     if (stats->index == stats->samples) stats->index = 0;
 }
 
-uint64_t stats_min(stats *stats) {
-    uint64_t min = 0;
-    for (uint64_t i = 0; i < stats->limit; i++) {
-        uint64_t x = stats->data[i];
-        if (x < min || min == 0) min = x;
-    }
-    return min;
+static int stats_compare(const void *a, const void *b) {
+    uint64_t *x = (uint64_t *) a;
+    uint64_t *y = (uint64_t *) b;
+    return *x - *y;
 }
 
-uint64_t stats_max(stats *stats) {
-    uint64_t max = 0;
-    for (uint64_t i = 0; i < stats->limit; i++) {
-        uint64_t x = stats->data[i];
-        if (x > max || max == 0) max = x;
-    }
-    return max;
-}
+long double stats_summarize(stats *stats) {
+    qsort(stats->data, stats->limit, sizeof(uint64_t), &stats_compare);
 
-long double stats_mean(stats *stats) {
-    uint64_t sum = 0;
     if (stats->limit == 0) return 0.0;
+
+    uint64_t sum = 0;
     for (uint64_t i = 0; i < stats->limit; i++) {
         sum += stats->data[i];
     }
@@ -71,3 +72,24 @@ long double stats_within_stdev(stats *stats, long double mean, long double stdev
 
     return (sum / (long double) stats->limit) * 100;
 }
+
+uint64_t stats_percentile(stats *stats, long double p) {
+    uint64_t rank = round((p / 100.0) * stats->limit + 0.5);
+    return stats->data[rank - 1];
+}
+
+void stats_sample(stats *dst, tinymt64_t *state, uint64_t count, stats *src) {
+    for (uint64_t i = 0; i < count; i++) {
+        uint64_t n = rand64(state, src->limit);
+        stats_record(dst, src->data[n]);
+    }
+}
+
+uint64_t rand64(tinymt64_t *state, uint64_t n) {
+    uint64_t x, max = ~UINT64_C(0);
+    max -= max % n;
+    do {
+        x = tinymt64_generate_uint64(state);
+    } while (x >= max);
+    return x % n;
+}

src/stats.h

@@ -1,21 +1,33 @@
 #ifndef STATS_H
 #define STATS_H
 
+#include <stdbool.h>
+#include "tinymt64.h"
+
+#define MAX(X, Y) ((X) > (Y) ? (X) : (Y))
+#define MIN(X, Y) ((X) < (Y) ? (X) : (Y))
+
 typedef struct {
     uint64_t samples;
     uint64_t index;
     uint64_t limit;
+    uint64_t min;
+    uint64_t max;
     uint64_t data[];
 } stats;
 
 stats *stats_alloc(uint64_t);
 void stats_free(stats *);
+void stats_reset(stats *);
+
 void stats_record(stats *, uint64_t);
-uint64_t stats_min(stats *);
-uint64_t stats_max(stats *);
-long double stats_mean(stats *);
+
+long double stats_summarize(stats *);
 long double stats_stdev(stats *stats, long double);
 long double stats_within_stdev(stats *, long double, long double, uint64_t);
+uint64_t stats_percentile(stats *, long double);
+
+void stats_sample(stats *, tinymt64_t *, uint64_t, stats *);
+uint64_t rand64(tinymt64_t *, uint64_t);
 
 #endif /* STATS_H */
-

src/wrk.c

@@ -1,31 +1,7 @@
 // Copyright (C) 2012 - Will Glozer.  All rights reserved.
 
 #include "wrk.h"
-
-#include <ctype.h>
-#include <errno.h>
-#include <fcntl.h>
-#include <getopt.h>
-#include <math.h>
-#include <netdb.h>
-#include <netinet/in.h>
-#include <netinet/tcp.h>
-#include <stdarg.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <signal.h>
-#include <time.h>
-#include <unistd.h>
-#include <sys/socket.h>
-#include <sys/time.h>
-#include <sys/uio.h>
-
-#include "aprintf.h"
-#include "stats.h"
-#include "units.h"
-#include "zmalloc.h"
-#include "tinymt64.h"
+#include "main.h"
 
 static struct config {
     struct addrinfo addr;
@@ -33,12 +9,16 @@ static struct config {
     uint64_t connections;
     uint64_t duration;
     uint64_t timeout;
+    bool     latency;
+    SSL_CTX *ctx;
 } cfg;
 
 static struct {
+    char *method;
+    char *body;
     size_t size;
     char *buf;
-} request;
+} req;
 
 static struct {
     stats *latency;
@@ -46,6 +26,13 @@ static struct {
     pthread_mutex_t mutex;
 } statistics;
 
+static struct sock sock = {
+    .connect = sock_connect,
+    .close   = sock_close,
+    .read    = sock_read,
+    .write   = sock_write
+};
+
 static const struct http_parser_settings parser_settings = {
     .on_message_complete = request_complete
 };
@@ -64,6 +51,9 @@ static void usage() {
            "    -t, --threads     <N>  Number of threads to use   \n"
            "                                                      \n"
            "    -H, --header      <H>  Add header to request      \n"
+           "    -M, --method      <M>  HTTP method                \n"
+           "        --body        <B>  Request body               \n"
+           "        --latency          Print latency statistics   \n"
            "        --timeout     <T>  Socket/request timeout     \n"
            "    -v, --version          Print version details      \n"
            "                                                      \n"
@@ -89,10 +79,11 @@ int main(int argc, char **argv) {
         exit(1);
     }
 
-    char *host = extract_url_part(url, &parser_url, UF_HOST);
-    char *port = extract_url_part(url, &parser_url, UF_PORT);
-    char *service = port ? port : extract_url_part(url, &parser_url, UF_SCHEMA);
-    char *path = "/";
+    char *schema  = extract_url_part(url, &parser_url, UF_SCHEMA);
+    char *host    = extract_url_part(url, &parser_url, UF_HOST);
+    char *port    = extract_url_part(url, &parser_url, UF_PORT);
+    char *service = port ? port : schema;
+    char *path    = "/";
 
     if (parser_url.field_set & (1 << UF_PATH)) {
         path = &url[parser_url.field_data[UF_PATH].off];
@@ -123,11 +114,23 @@ int main(int argc, char **argv) {
         exit(1);
     }
 
+    if (!strncmp("https", schema, 5)) {
+        if ((cfg.ctx = ssl_init()) == NULL) {
+            fprintf(stderr, "unable to initialize SSL\n");
+            ERR_print_errors_fp(stderr);
+            exit(1);
+        }
+        sock.connect = ssl_connect;
+        sock.close   = ssl_close;
+        sock.read    = ssl_read;
+        sock.write   = ssl_write;
+    }
+
     signal(SIGPIPE, SIG_IGN);
     signal(SIGINT,  SIG_IGN);
-    cfg.addr     = *addr;
-    request.buf  = format_request(host, port, path, headers);
-    request.size = strlen(request.buf);
+    cfg.addr = *addr;
+    req.buf  = format_request(host, port, path, headers);
+    req.size = strlen(req.buf);
 
     pthread_mutex_init(&statistics.mutex, NULL);
     statistics.latency  = stats_alloc(SAMPLES);
@@ -187,6 +190,7 @@ int main(int argc, char **argv) {
     print_stats_header();
     print_stats("Latency", statistics.latency, format_time_us);
     print_stats("Req/Sec", statistics.requests, format_metric);
+    if (cfg.latency) print_stats_latency(statistics.latency);
 
     char *runtime_msg = format_time_us(runtime_us);
 
@@ -213,16 +217,17 @@ void *thread_main(void *arg) {
     thread->cs   = zmalloc(thread->connections * sizeof(connection));
     thread->loop = loop;
     tinymt64_init(&thread->rand, time_us());
+    thread->latency = stats_alloc(100000);
 
     connection *c = thread->cs;
 
     for (uint64_t i = 0; i < thread->connections; i++, c++) {
-        c->thread  = thread;
-        c->latency = 0;
+        c->thread = thread;
+        c->ssl = cfg.ctx ? SSL_new(cfg.ctx) : NULL;
         connect_socket(thread, c);
     }
 
-    aeCreateTimeEvent(loop, SAMPLE_INTERVAL_MS, sample_rate, thread, NULL);
+    aeCreateTimeEvent(loop, CALIBRATE_DELAY_MS, calibrate, thread, NULL);
     aeCreateTimeEvent(loop, TIMEOUT_INTERVAL_MS, check_timeouts, thread, NULL);
 
     thread->start = time_us();
@@ -231,6 +236,15 @@ void *thread_main(void *arg) {
     aeDeleteEventLoop(loop);
     zfree(thread->cs);
 
+    uint64_t max = thread->latency->max;
+    stats_free(thread->latency);
+
+    pthread_mutex_lock(&statistics.mutex);
+    for (uint64_t i = 0; i < thread->missed; i++) {
+        stats_record(statistics.latency, max);
+    }
+    pthread_mutex_unlock(&statistics.mutex);
+
     return NULL;
 }
 
@@ -245,50 +259,71 @@ static int connect_socket(thread *thread, connection *c) {
     fcntl(fd, F_SETFL, flags | O_NONBLOCK);
 
     if (connect(fd, addr.ai_addr, addr.ai_addrlen) == -1) {
-        if (errno != EINPROGRESS) {
-            thread->errors.connect++;
-            goto error;
-        }
+        if (errno != EINPROGRESS) goto error;
     }
 
     flags = 1;
     setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags));
 
-    if (aeCreateFileEvent(loop, fd, AE_WRITABLE, socket_writeable, c) != AE_OK) {
-        goto error;
+    flags = AE_READABLE | AE_WRITABLE;
+    if (aeCreateFileEvent(loop, fd, flags, socket_connected, c) == AE_OK) {
+        c->parser.data = c;
+        c->fd = fd;
+        return fd;
     }
 
-    http_parser_init(&c->parser, HTTP_RESPONSE);
-    c->parser.data = c;
-    c->fd = fd;
-
-    return fd;
-
   error:
+    thread->errors.connect++;
     close(fd);
     return -1;
 }
 
 static int reconnect_socket(thread *thread, connection *c) {
     aeDeleteFileEvent(thread->loop, c->fd, AE_WRITABLE | AE_READABLE);
+    sock.close(c);
     close(c->fd);
     return connect_socket(thread, c);
 }
 
+static int calibrate(aeEventLoop *loop, long long id, void *data) {
+    thread *thread = data;
+
+    uint64_t elapsed_ms = (time_us() - thread->start) / 1000;
+    uint64_t req_per_ms = thread->requests / elapsed_ms;
+
+    if (!req_per_ms) return CALIBRATE_DELAY_MS / 2;
+
+    thread->rate  = (req_per_ms * SAMPLE_INTERVAL_MS) / 10;
+    thread->start = time_us();
+    thread->requests = 0;
+    stats_reset(thread->latency);
+
+    aeCreateTimeEvent(loop, SAMPLE_INTERVAL_MS, sample_rate, thread, NULL);
+
+    return AE_NOMORE;
+}
+
 static int sample_rate(aeEventLoop *loop, long long id, void *data) {
     thread *thread = data;
 
-    uint64_t n = rand64(&thread->rand, thread->connections);
     uint64_t elapsed_ms = (time_us() - thread->start) / 1000;
-    connection *c = thread->cs + n;
-    uint64_t requests = (thread->complete / elapsed_ms) * 1000;
+    uint64_t requests = (thread->requests / elapsed_ms) * 1000;
+    uint64_t missed = thread->rate - MIN(thread->rate, thread->latency->limit);
+    uint64_t count = thread->rate - missed;
 
     pthread_mutex_lock(&statistics.mutex);
-    stats_record(statistics.latency,  c->latency);
+    stats_sample(statistics.latency, &thread->rand, count, thread->latency);
     stats_record(statistics.requests, requests);
     pthread_mutex_unlock(&statistics.mutex);
 
-    return SAMPLE_INTERVAL_MS + rand64(&thread->rand, SAMPLE_INTERVAL_MS);
+    uint64_t max = thread->latency->max;
+    thread->missed  += missed;
+    thread->requests = 0;
+    thread->start    = time_us();
+    stats_reset(thread->latency);
+    thread->latency->max = max;
+
+    return SAMPLE_INTERVAL_MS;
 }
 
 static int request_complete(http_parser *parser) {
@@ -297,7 +332,9 @@ static int request_complete(http_parser *parser) {
     uint64_t now = time_us();
 
     thread->complete++;
-    c->latency = now - c->start;
+    thread->requests++;
+
+    stats_record(thread->latency, now - c->start);
 
     if (parser->status_code > 399) {
         thread->errors.status++;
@@ -311,7 +348,6 @@ static int request_complete(http_parser *parser) {
     if (!http_should_keep_alive(parser)) goto reconnect;
 
     http_parser_init(parser, HTTP_RESPONSE);
-    aeDeleteFileEvent(thread->loop, c->fd, AE_READABLE);
     aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
 
     goto done;
@@ -343,13 +379,47 @@ static int check_timeouts(aeEventLoop *loop, long long id, void *data) {
     return TIMEOUT_INTERVAL_MS;
 }
 
-static void socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
+static void socket_connected(aeEventLoop *loop, int fd, void *data, int mask) {
     connection *c = data;
 
-    if (write(fd, request.buf, request.size) < request.size) goto error;
-    c->start = time_us();
-    aeDeleteFileEvent(loop, fd, AE_WRITABLE);
-    aeCreateFileEvent(loop, fd, AE_READABLE, socket_readable, c);
+    switch (sock.connect(c)) {
+        case OK:    break;
+        case ERROR: goto error;
+        case RETRY: return;
+    }
+
+    http_parser_init(&c->parser, HTTP_RESPONSE);
+    c->written = 0;
+
+    aeCreateFileEvent(c->thread->loop, c->fd, AE_READABLE, socket_readable, c);
+    aeCreateFileEvent(c->thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
+
+    return;
+
+  error:
+    c->thread->errors.connect++;
+    reconnect_socket(c->thread, c);
+
+}
+
+static void socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
+    connection *c = data;
+    size_t len = req.size - c->written;
+    size_t n;
+
+    switch (sock.write(c, req.buf + c->written, len, &n)) {
+        case OK:    break;
+        case ERROR: goto error;
+        case RETRY: return;
+    }
+
+    if (!c->written) c->start = time_us();
+
+    c->written += n;
+    if (c->written == req.size) {
+        c->written = 0;
+        aeDeleteFileEvent(loop, fd, AE_WRITABLE);
+    }
 
     return;
 
@@ -358,11 +428,17 @@ static void socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
     reconnect_socket(c->thread, c);
 }
 
+
 static void socket_readable(aeEventLoop *loop, int fd, void *data, int mask) {
     connection *c = data;
-    ssize_t n;
+    size_t n;
+
+    switch (sock.read(c, &n)) {
+        case OK:    break;
+        case ERROR: goto error;
+        case RETRY: return;
+    }
 
-    if ((n = read(fd, c->buf, sizeof(c->buf))) <= 0) goto error;
     if (http_parser_execute(&c->parser, &parser_settings, c->buf, n) != n) goto error;
     c->thread->bytes += n;
 
@@ -379,15 +455,6 @@ static uint64_t time_us() {
     return (t.tv_sec * 1000000) + t.tv_usec;
 }
 
-static uint64_t rand64(tinymt64_t *state, uint64_t n) {
-    uint64_t x, max = ~UINT64_C(0);
-    max -= max % n;
-    do {
-        x = tinymt64_generate_uint64(state);
-    } while (x >= max);
-    return x % n;
-}
-
 static char *extract_url_part(char *url, struct http_parser_url *parser_url, enum http_parser_url_fields field) {
     char *part = NULL;
 
@@ -402,7 +469,7 @@ static char *extract_url_part(char *url, struct http_parser_url *parser_url, enu
 }
 
 static char *format_request(char *host, char *port, char *path, char **headers) {
-    char *req  = NULL;
+    char *buf  = NULL;
     char *head = NULL;
 
     for (char **h = headers; *h != NULL; h++) {
@@ -413,14 +480,21 @@ static char *format_request(char *host, char *port, char *path, char **headers)
         }
     }
 
-    aprintf(&req, "GET %s HTTP/1.1\r\n", path);
-    if (host) aprintf(&req, "Host: %s", host);
-    if (port) aprintf(&req, ":%s", port);
-    if (host) aprintf(&req, "\r\n");
-    aprintf(&req, "%s\r\n", head ? head : "");
+    if (req.body) {
+        size_t len = strlen(req.body);
+        aprintf(&head, "Content-Length: %zd\r\n", len);
+    }
+
+    aprintf(&buf, "%s %s HTTP/1.1\r\n", req.method, path);
+    if (host) aprintf(&buf, "Host: %s", host);
+    if (port) aprintf(&buf, ":%s", port);
+    if (host) aprintf(&buf, "\r\n");
+
+    aprintf(&buf, "%s\r\n", head ? head : "");
+    aprintf(&buf, "%s", req.body ? req.body : "");
 
     free(head);
-    return req;
+    return buf;
 }
 
 static struct option longopts[] = {
@@ -428,6 +502,9 @@ static struct option longopts[] = {
     { "duration",    required_argument, NULL, 'd' },
     { "threads",     required_argument, NULL, 't' },
     { "header",      required_argument, NULL, 'H' },
+    { "method",      required_argument, NULL, 'M' },
+    { "body",        required_argument, NULL, 'B' },
+    { "latency",     no_argument,       NULL, 'L' },
     { "timeout",     required_argument, NULL, 'T' },
     { "help",        no_argument,       NULL, 'h' },
     { "version",     no_argument,       NULL, 'v' },
@@ -442,8 +519,9 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
     cfg->connections = 10;
     cfg->duration    = 10;
     cfg->timeout     = SOCKET_TIMEOUT_MS;
+    req.method       = "GET";
 
-    while ((c = getopt_long(argc, argv, "t:c:d:H:T:rv?", longopts, NULL)) != -1) {
+    while ((c = getopt_long(argc, argv, "t:c:d:H:M:B:T:Lrv?", longopts, NULL)) != -1) {
         switch (c) {
             case 't':
                 if (scan_metric(optarg, &cfg->threads)) return -1;
@@ -457,6 +535,15 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
             case 'H':
                 *header++ = optarg;
                 break;
+            case 'M':
+                req.method = optarg;
+                break;
+            case 'B':
+                req.body = optarg;
+                break;
+            case 'L':
+                cfg->latency = true;
+                break;
             case 'T':
                 if (scan_time(optarg, &cfg->timeout)) return -1;
                 cfg->timeout *= 1000;
@@ -506,8 +593,8 @@ static void print_units(long double n, char *(*fmt)(long double), int width) {
 }
 
 static void print_stats(char *name, stats *stats, char *(*fmt)(long double)) {
-    long double mean  = stats_mean(stats);
-    long double max   = stats_max(stats);
+    uint64_t max = stats->max;
+    long double mean  = stats_summarize(stats);
     long double stdev = stats_stdev(stats, mean);
 
     printf("    %-10s", name);
@@ -516,3 +603,15 @@ static void print_stats(char *name, stats *stats, char *(*fmt)(long double)) {
     print_units(max,   fmt, 9);
     printf("%8.2Lf%%\n", stats_within_stdev(stats, mean, stdev, 1));
 }
+
+static void print_stats_latency(stats *stats) {
+    long double percentiles[] = { 50.0, 75.0, 90.0, 99.0 };
+    printf("  Latency Distribution\n");
+    for (size_t i = 0; i < sizeof(percentiles) / sizeof(long double); i++) {
+        long double p = percentiles[i];
+        uint64_t n = stats_percentile(stats, p);
+        printf("%7.0Lf%%", p);
+        print_units(n, format_time_us, 10);
+        printf("\n");
+    }
+}

src/wrk.h

@@ -6,17 +6,20 @@
 #include <inttypes.h>
 #include <sys/types.h>
 
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+
 #include "stats.h"
 #include "ae.h"
 #include "http_parser.h"
-#include "tinymt64.h"
 
-#define VERSION  "2.0.0"
+#define VERSION  "2.2.0"
 #define RECVBUF  8192
-#define SAMPLES  100000
+#define SAMPLES  100000000
 
 #define SOCKET_TIMEOUT_MS   2000
-#define SAMPLE_INTERVAL_MS  100
+#define SAMPLE_INTERVAL_MS  10
+#define CALIBRATE_DELAY_MS  500
 #define TIMEOUT_INTERVAL_MS 2000
 
 typedef struct {
@@ -33,8 +36,12 @@ typedef struct {
     uint64_t connections;
     uint64_t stop_at;
     uint64_t complete;
+    uint64_t requests;
     uint64_t bytes;
     uint64_t start;
+    uint64_t rate;
+    uint64_t missed;
+    stats *latency;
     tinymt64_t rand;
     errors errors;
     struct connection *cs;
@@ -44,32 +51,10 @@ typedef struct connection {
     thread *thread;
     http_parser parser;
     int fd;
+    SSL *ssl;
     uint64_t start;
-    uint64_t latency;
+    size_t written;
     char buf[RECVBUF];
 } connection;
 
-struct config;
-
-static void *thread_main(void *);
-static int connect_socket(thread *, connection *);
-static int reconnect_socket(thread *, connection *);
-
-static int sample_rate(aeEventLoop *, long long, void *);
-static int check_timeouts(aeEventLoop *, long long, void *);
-
-static void socket_writeable(aeEventLoop *, int, void *, int);
-static void socket_readable(aeEventLoop *, int, void *, int);
-static int request_complete(http_parser *);
-
-static uint64_t time_us();
-static uint64_t rand64(tinymt64_t *, uint64_t);
-
-static char *extract_url_part(char *, struct http_parser_url *, enum http_parser_url_fields);
-static char *format_request(char *, char *, char *, char **);
-
-static int parse_args(struct config *, char **, char **, int, char **);
-static void print_stats_header();
-static void print_stats(char *, stats *, char *(*)(long double));
-
 #endif /* WRK_H */