cleanup & pipelining for 3.1.0 release

Makefile

@@ -21,7 +21,7 @@ SRC  := wrk.c net.c ssl.c aprintf.c stats.c script.c units.c \
 BIN  := wrk
 
 ODIR := obj
-OBJ  := $(patsubst %.c,$(ODIR)/%.o,$(SRC))
+OBJ  := $(patsubst %.c,$(ODIR)/%.o,$(SRC)) $(ODIR)/bytecode.o
 
 LDIR     = deps/luajit/src
 LIBS    := -lluajit $(LIBS)
@@ -34,25 +34,25 @@ clean:
 	$(RM) $(BIN) obj/*
 	@$(MAKE) -C deps/luajit clean
 
-$(BIN): $(OBJ) $(ODIR)/bytecode.o
+$(BIN): $(OBJ)
 	@echo LINK $(BIN)
 	@$(CC) $(LDFLAGS) -o $@ $^ $(LIBS)
 
-$(OBJ): config.h Makefile | $(ODIR)
+$(OBJ): config.h Makefile $(LDIR)/libluajit.a | $(ODIR)
 
-$(ODIR): $(LDIR)/libluajit.a
+$(ODIR):
 	@mkdir -p $@
 
-$(ODIR)/bytecode.o: scripts/wrk.lua
+$(ODIR)/bytecode.o: src/wrk.lua
 	@echo LUAJIT $<
-	@$(SHELL) -c 'cd $(LDIR) && ./luajit -b $(PWD)/$< $(PWD)/$@'
+	@$(SHELL) -c 'cd $(LDIR) && ./luajit -b $(CURDIR)/$< $(CURDIR)/$@'
 
 $(ODIR)/%.o : %.c
 	@echo CC $<
 	@$(CC) $(CFLAGS) -c -o $@ $<
 
 $(LDIR)/libluajit.a:
-	@echo Building LuaJit...
+	@echo Building LuaJIT...
 	@$(MAKE) -C $(LDIR) BUILDMODE=static
 
 .PHONY: all clean

README

@@ -5,7 +5,8 @@ wrk - a HTTP benchmarking tool
   design with scalable event notification systems such as epoll and kqueue.
 
   An optional LuaJIT script can perform HTTP request generation, response
-  processing, and custom reporting.
+  processing, and custom reporting. Several example scripts are located in
+  scripts/
 
 Basic Usage
 

scripts/auth.lua

@@ -0,0 +1,18 @@
+-- example script that demonstrates response handling and
+-- retrieving an authentication token to set on all future
+-- requests
+
+token = nil
+path  = "/authenticate"
+
+request = function()
+   return wrk.format("GET", path)
+end
+
+response = function(status, headers, body)
+   if not token and status == 200 then
+      token = headers["X-Token"]
+      path  = "/resource"
+      wrk.headers["X-Token"] = token
+   end
+end

scripts/counter.lua

@@ -0,0 +1,14 @@
+-- example dynamic request script which demonstrates changing
+-- the request path and a header for each request
+-------------------------------------------------------------
+-- NOTE: each wrk thread has an independent Lua scripting
+-- context and thus there will be one counter per thread
+
+counter = 0
+
+request = function()
+   path = "/" .. counter
+   wrk.headers["X-Counter"] = counter
+   counter = counter + 1
+   return wrk.format(nil, path)
+end

scripts/pipeline.lua

@@ -0,0 +1,16 @@
+-- example script demonstrating HTTP pipelining
+
+init = function(args)
+   wrk.init(args)
+
+   local r = {}
+   r[1] = wrk.format(nil, "/?foo")
+   r[2] = wrk.format(nil, "/?bar")
+   r[3] = wrk.format(nil, "/?baz")
+
+   req = table.concat(r)
+end
+
+request = function()
+   return req
+end

scripts/post.lua

@@ -0,0 +1,6 @@
+-- example HTTP POST script which demonstrates setting the
+-- HTTP method, body, and adding a header
+
+wrk.method = "POST"
+wrk.body   = "foo=bar&baz=quux"
+wrk.headers["Content-Type"] = "application/x-www-form-urlencoded"

scripts/report.lua

@@ -0,0 +1,10 @@
+-- example reporting script which demonstrates a custom
+-- done() function that prints latency percentiles as CSV
+
+done = function(summary, latency, requests)
+   io.write("------------------------------\n")
+   for _, p in pairs({ 50, 90, 99, 99.999 }) do
+      n = latency:percentile(p)
+      io.write(string.format("%g%%,%d\n", p, n))
+   end
+end

src/net.c

@@ -2,6 +2,7 @@
 
 #include <errno.h>
 #include <unistd.h>
+#include <sys/ioctl.h>
 
 #include "net.h"
 
@@ -30,3 +31,9 @@ status sock_write(connection *c, char *buf, size_t len, size_t *n) {
     *n = (size_t) r;
     return OK;
 }
+
+size_t sock_readable(connection *c) {
+    int n, rc;
+    rc = ioctl(c->fd, FIONREAD, &n);
+    return rc == -1 ? 0 : n;
+}

src/net.h

@@ -13,15 +13,17 @@ typedef enum {
 } status;
 
 struct sock {
-    status (*connect)(connection *);
-    status (  *close)(connection *);
-    status (   *read)(connection *, size_t *);
-    status (  *write)(connection *, char *, size_t, size_t *);
+    status ( *connect)(connection *);
+    status (   *close)(connection *);
+    status (    *read)(connection *, size_t *);
+    status (   *write)(connection *, char *, size_t, size_t *);
+    size_t (*readable)(connection *);
 };
 
 status sock_connect(connection *);
 status sock_close(connection *);
 status sock_read(connection *, size_t *);
 status sock_write(connection *, char *, size_t, size_t *);
+size_t sock_readable(connection *);
 
 #endif /* NET_H */

src/ssl.c

@@ -38,7 +38,7 @@ SSL_CTX *ssl_init() {
         CRYPTO_set_locking_callback(ssl_lock);
         CRYPTO_set_id_callback(ssl_id);
 
-        if ((ctx = SSL_CTX_new(TLSv1_client_method()))) {
+        if ((ctx = SSL_CTX_new(SSLv23_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);
@@ -93,3 +93,7 @@ status ssl_write(connection *c, char *buf, size_t len, size_t *n) {
     *n = (size_t) r;
     return OK;
 }
+
+size_t ssl_readable(connection *c) {
+    return SSL_pending(c->ssl);
+}

src/ssl.h

@@ -9,5 +9,6 @@ status ssl_connect(connection *);
 status ssl_close(connection *);
 status ssl_read(connection *, size_t *);
 status ssl_write(connection *, char *, size_t, size_t *);
+size_t ssl_readable(connection *);
 
 #endif /* SSL_H */

src/wrk.c

@@ -9,6 +9,7 @@ static struct config {
     uint64_t connections;
     uint64_t duration;
     uint64_t timeout;
+    uint64_t pipeline;
     bool     latency;
     bool     dynamic;
     char    *script;
@@ -22,10 +23,11 @@ static struct {
 } statistics;
 
 static struct sock sock = {
-    .connect = sock_connect,
-    .close   = sock_close,
-    .read    = sock_read,
-    .write   = sock_write
+    .connect  = sock_connect,
+    .close    = sock_close,
+    .read     = sock_read,
+    .write    = sock_write,
+    .readable = sock_readable
 };
 
 static struct http_parser_settings parser_settings = {
@@ -114,10 +116,11 @@ int main(int argc, char **argv) {
             ERR_print_errors_fp(stderr);
             exit(1);
         }
-        sock.connect = ssl_connect;
-        sock.close   = ssl_close;
-        sock.read    = ssl_read;
-        sock.write   = ssl_write;
+        sock.connect  = ssl_connect;
+        sock.close    = ssl_close;
+        sock.read     = ssl_read;
+        sock.write    = ssl_write;
+        sock.readable = ssl_readable;
     }
 
     signal(SIGPIPE, SIG_IGN);
@@ -134,6 +137,7 @@ int main(int argc, char **argv) {
 
     for (uint64_t i = 0; i < cfg.threads; i++) {
         thread *t = &threads[i];
+        t->loop        = aeCreateEventLoop(10 + cfg.connections * 3);
         t->connections = connections;
         t->stop_at     = stop_at;
 
@@ -142,7 +146,7 @@ int main(int argc, char **argv) {
         script_init(t->L, cfg.script, argc - optind, &argv[optind]);
 
         if (i == 0) {
-            script_verify_request(t->L);
+            cfg.pipeline = script_verify_request(t->L);
             cfg.dynamic = !script_is_static(t->L);
             if (script_want_response(t->L)) {
                 parser_settings.on_header_field = header_field;
@@ -151,9 +155,9 @@ int main(int argc, char **argv) {
             }
         }
 
-        if (pthread_create(&t->thread, NULL, &thread_main, t)) {
+        if (!t->loop || pthread_create(&t->thread, NULL, &thread_main, t)) {
             char *msg = strerror(errno);
-            fprintf(stderr, "unable to create thread %"PRIu64" %s\n", i, msg);
+            fprintf(stderr, "unable to create thread %"PRIu64": %s\n", i, msg);
             exit(2);
         }
     }
@@ -225,10 +229,9 @@ int main(int argc, char **argv) {
 
 void *thread_main(void *arg) {
     thread *thread = arg;
+    aeEventLoop *loop = thread->loop;
 
-    aeEventLoop *loop = aeCreateEventLoop(10 + cfg.connections * 3);
-    thread->cs   = zmalloc(thread->connections * sizeof(connection));
-    thread->loop = loop;
+    thread->cs = zmalloc(thread->connections * sizeof(connection));
     tinymt64_init(&thread->rand, time_us());
     thread->latency = stats_alloc(100000);
 
@@ -328,6 +331,26 @@ static int calibrate(aeEventLoop *loop, long long id, void *data) {
     return AE_NOMORE;
 }
 
+static int check_timeouts(aeEventLoop *loop, long long id, void *data) {
+    thread *thread = data;
+    connection *c  = thread->cs;
+    uint64_t now   = time_us();
+
+    uint64_t maxAge = now - (cfg.timeout * 1000);
+
+    for (uint64_t i = 0; i < thread->connections; i++, c++) {
+        if (maxAge > c->start) {
+            thread->errors.timeout++;
+        }
+    }
+
+    if (stop || now >= thread->stop_at) {
+        aeStop(loop);
+    }
+
+    return TIMEOUT_INTERVAL_MS;
+}
+
 static int sample_rate(aeEventLoop *loop, long long id, void *data) {
     thread *thread = data;
 
@@ -384,8 +407,6 @@ static int response_complete(http_parser *parser) {
     thread->complete++;
     thread->requests++;
 
-    stats_record(thread->latency, now - c->start);
-
     if (status > 399) {
         thread->errors.status++;
     }
@@ -401,40 +422,22 @@ static int response_complete(http_parser *parser) {
         goto done;
     }
 
-    if (!http_should_keep_alive(parser)) goto reconnect;
+    if (--c->pending == 0) {
+        stats_record(thread->latency, now - c->start);
+        aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
+    }
+
+    if (!http_should_keep_alive(parser)) {
+        reconnect_socket(thread, c);
+        goto done;
+    }
 
     http_parser_init(parser, HTTP_RESPONSE);
-    aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
-
-    goto done;
-
-  reconnect:
-    reconnect_socket(thread, c);
 
   done:
     return 0;
 }
 
-static int check_timeouts(aeEventLoop *loop, long long id, void *data) {
-    thread *thread = data;
-    connection *c  = thread->cs;
-    uint64_t now   = time_us();
-
-    uint64_t maxAge = now - (cfg.timeout * 1000);
-
-    for (uint64_t i = 0; i < thread->connections; i++, c++) {
-        if (maxAge > c->start) {
-            thread->errors.timeout++;
-        }
-    }
-
-    if (stop || now >= thread->stop_at) {
-        aeStop(loop);
-    }
-
-    return TIMEOUT_INTERVAL_MS;
-}
-
 static void socket_connected(aeEventLoop *loop, int fd, void *data, int mask) {
     connection *c = data;
 
@@ -447,8 +450,8 @@ static void socket_connected(aeEventLoop *loop, int fd, void *data, int mask) {
     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);
+    aeCreateFileEvent(c->thread->loop, fd, AE_READABLE, socket_readable, c);
+    aeCreateFileEvent(c->thread->loop, fd, AE_WRITABLE, socket_writeable, c);
 
     return;
 
@@ -476,7 +479,10 @@ static void socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
         case RETRY: return;
     }
 
-    if (!c->written) c->start = time_us();
+    if (!c->written) {
+        c->start = time_us();
+        c->pending = cfg.pipeline;
+    }
 
     c->written += n;
     if (c->written == c->length) {
@@ -496,14 +502,16 @@ static void socket_readable(aeEventLoop *loop, int fd, void *data, int mask) {
     connection *c = data;
     size_t n;
 
-    switch (sock.read(c, &n)) {
-        case OK:    break;
-        case ERROR: goto error;
-        case RETRY: return;
-    }
+    do {
+        switch (sock.read(c, &n)) {
+            case OK:    break;
+            case ERROR: goto error;
+            case RETRY: return;
+        }
 
-    if (http_parser_execute(&c->parser, &parser_settings, c->buf, n) != n) goto error;
-    c->thread->bytes += n;
+        if (http_parser_execute(&c->parser, &parser_settings, c->buf, n) != n) goto error;
+        c->thread->bytes += n;
+    } while (n == RECVBUF && sock.readable(c) > 0);
 
     return;
 

src/wrk.h

@@ -14,7 +14,7 @@
 #include "script.h"
 #include "http_parser.h"
 
-#define VERSION  "3.0.0"
+#define VERSION  "3.1.0"
 #define RECVBUF  8192
 #define SAMPLES  100000000
 
@@ -53,6 +53,7 @@ typedef struct connection {
     char *request;
     size_t length;
     size_t written;
+    uint64_t pending;
     buffer headers;
     buffer body;
     char buf[RECVBUF];

src/wrk.lua

@@ -9,14 +9,16 @@ local wrk = {
 }
 
 function wrk.format(method, path, headers, body)
-   local host    = wrk.host
    local method  = method  or wrk.method
    local path    = path    or wrk.path
    local headers = headers or wrk.headers
    local body    = body    or wrk.body
    local s       = {}
 
-   headers["Host"]           = port and (host .. ":" .. port) or host
+   if not headers["Host"] then
+      headers["Host"] = wrk.headers["Host"]
+   end
+
    headers["Content-Length"] = body and string.len(body)
 
    s[1] = string.format("%s %s HTTP/1.1", method, path)
@@ -30,8 +32,22 @@ function wrk.format(method, path, headers, body)
    return table.concat(s, "\r\n")
 end
 
-function wrk.init(args) req = wrk.format() end
-function wrk.request()  return req         end
+function wrk.init(args)
+   if not wrk.headers["Host"] then
+      local host = wrk.host
+      local port = wrk.port
+
+      host = host:find(":") and ("[" .. host .. "]")  or host
+      host = port           and (host .. ":" .. port) or host
+
+      wrk.headers["Host"] = host
+   end
+   req = wrk.format()
+end
+
+function wrk.request()
+   return req
+end
 
 init     = wrk.init
 request  = wrk.request