send hostname in TLS SNI extension

.gitignore

@@ -1,2 +1,2 @@
-obj/*
+obj/
 wrk

CHANGES

@@ -0,0 +1,20 @@
+wrk 4.0.2
+
+ * Send hostname using TLS SNI.
+ * Add optional WITH_OPENSSL and WITH_LUAJIT to use system libs.
+ * Bundle OpenSSL 1.0.2.
+ * delay() can return milliseconds to delay sending next request.
+
+wrk 4.0.0
+
+ * The wrk global variable is the only global defined by default.
+ * wrk.init() calls the global init(), remove calls to wrk.init().
+ * Add wrk.lookup(host, port) and wrk.connect(addr) functions.
+ * Add setup phase that calls the global setup() for each thread.
+ * Allow assignment to thread.addr to specify the server address.
+ * Add thread:set(name, value), thread:get(name), and thread:stop().
+ * Record latency for every request instead of random samples.
+ * Latency and requests in done() are now callable, not indexable.
+ * Only record timeouts when a response is actually received.
+ * Remove calibration phase and record rate at fixed interval.
+ * Improve correction of coordinated omission.

INSTALL

@@ -0,0 +1,29 @@
+Overview
+
+  wrk should build on most UNIX-like operating systems and
+  architectures that have GNU make and are supported by LuaJIT and
+  OpenSSL. Some systems may require additional CFLAGS or LDFLAGS,
+  see the top of the Makefile for examples
+
+  In many cases simply running `make` (often `gmake` on *BSD) will
+  do the trick.
+
+Dependencies
+
+  wrk requires LuaJIT and OpenSSL and is distributed with appropriate
+  versions that will be unpacked and built as necessary.
+
+  If you are building wrk packages for an OS distribution or otherwise
+  prefer to use system versions of dependencies you may specify their
+  location when invoking make with one or more of:
+
+    WITH_LUAJIT
+    WITH_OPENSSL
+
+  For example to use the system version of both libraries on Linux:
+
+    make WITH_LUAJIT=/usr WITH_OPENSSL=/usr
+
+  Or to use the Homebrew version of OpenSSL on Mac OS X:
+
+    make WITH_OPENSSL=/usr/local/opt/openssl

LICENSE

@@ -1,7 +1,6 @@
 
-                                 Apache License
-                           Version 2.0, January 2004
-                        http://www.apache.org/licenses/
+                          Modified Apache 2.0 License
+                         Version 2.0.1, February 2015
 
    TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
 
@@ -121,6 +120,12 @@
           that such additional attribution notices cannot be construed
           as modifying the License.
 
+      (e) If the Derivative Work includes substantial changes to features
+          or functionality of the Work, then you must remove the name of
+          the Work, and any derivation thereof, from all copies that you
+          distribute, whether in Source or Object form, except as required
+          in copyright, patent, trademark, and attribution notices.
+
       You may add Your own copyright statement to Your modifications and
       may provide additional or different license terms and conditions
       for use, reproduction, or distribution of Your modifications, or

Makefile

@@ -1,38 +1,108 @@
-CFLAGS  := -std=c99 -Wall -O2 -D_REENTRANT
-LIBS    := -lpthread -lm
+CFLAGS  += -std=c99 -Wall -O2 -D_REENTRANT
+LIBS    := -lpthread -lm -lssl -lcrypto
 
-TARGET  := $(shell uname -s | tr [A-Z] [a-z] 2>/dev/null || echo unknown)
+TARGET  := $(shell uname -s | tr '[A-Z]' '[a-z]' 2>/dev/null || echo unknown)
 
 ifeq ($(TARGET), sunos)
-	CFLAGS += -D_PTHREADS
+	CFLAGS += -D_PTHREADS -D_POSIX_C_SOURCE=200112L
 	LIBS   += -lsocket
+else ifeq ($(TARGET), darwin)
+	LDFLAGS += -pagezero_size 10000 -image_base 100000000
+else ifeq ($(TARGET), linux)
+	CFLAGS  += -D_POSIX_C_SOURCE=200112L -D_BSD_SOURCE
+	LIBS    += -ldl
+	LDFLAGS += -Wl,-E
+else ifeq ($(TARGET), freebsd)
+	CFLAGS  += -D_DECLARE_C99_LDBL_MATH
+	LDFLAGS += -Wl,-E
 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 script.c units.c \
+		ae.c zmalloc.c http_parser.c
 BIN  := wrk
+VER  ?= $(shell git describe --tags --always --dirty)
 
 ODIR := obj
-OBJ  := $(patsubst %.c,$(ODIR)/%.o,$(SRC))
+OBJ  := $(patsubst %.c,$(ODIR)/%.o,$(SRC)) $(ODIR)/bytecode.o $(ODIR)/version.o
+LIBS := -lluajit-5.1 $(LIBS)
+
+DEPS    :=
+CFLAGS  += -I$(ODIR)/include
+LDFLAGS += -L$(ODIR)/lib
+
+ifneq ($(WITH_LUAJIT),)
+	CFLAGS  += -I$(WITH_LUAJIT)/include
+	LDFLAGS += -L$(WITH_LUAJIT)/lib
+else
+	DEPS += $(ODIR)/lib/libluajit-5.1.a
+endif
+
+ifneq ($(WITH_OPENSSL),)
+	CFLAGS  += -I$(WITH_OPENSSL)/include
+	LDFLAGS += -L$(WITH_OPENSSL)/lib
+else
+	DEPS += $(ODIR)/lib/libssl.a
+endif
 
 all: $(BIN)
 
 clean:
-	$(RM) $(BIN) obj/*
+	$(RM) -rf $(BIN) obj/*
 
 $(BIN): $(OBJ)
-	$(CC) $(LDFLAGS) -o $@ $^ $(LIBS)
+	@echo LINK $(BIN)
+	@$(CC) $(LDFLAGS) -o $@ $^ $(LIBS)
 
-$(OBJ): config.h Makefile | $(ODIR)
+$(OBJ): config.h Makefile $(DEPS) | $(ODIR)
 
 $(ODIR):
-	@mkdir $@
+	@mkdir -p $@
+
+$(ODIR)/bytecode.o: src/wrk.lua
+	@echo LUAJIT $<
+	@$(SHELL) -c 'PATH=obj/bin:$(PATH) luajit -b $(CURDIR)/$< $(CURDIR)/$@'
+
+$(ODIR)/version.o:
+	@echo 'const char *VERSION="$(VER)";' | $(CC) -xc -c -o $@ -
 
 $(ODIR)/%.o : %.c
-	$(CC) $(CFLAGS) -c -o $@ $<
+	@echo CC $<
+	@$(CC) $(CFLAGS) -c -o $@ $<
+
+# Dependencies
+
+LUAJIT  := $(notdir $(patsubst %.tar.gz,%,$(wildcard deps/LuaJIT*.tar.gz)))
+OPENSSL := $(notdir $(patsubst %.tar.gz,%,$(wildcard deps/openssl*.tar.gz)))
+
+OPENSSL_OPTS = no-shared no-ssl2 no-psk no-srp no-dtls no-idea --prefix=$(abspath $(ODIR))
+
+$(ODIR)/$(LUAJIT):  deps/$(LUAJIT).tar.gz  | $(ODIR)
+	@tar -C $(ODIR) -xf $<
+
+$(ODIR)/$(OPENSSL): deps/$(OPENSSL).tar.gz | $(ODIR)
+	@tar -C $(ODIR) -xf $<
+
+$(ODIR)/lib/libluajit-5.1.a: $(ODIR)/$(LUAJIT)
+	@echo Building LuaJIT...
+	@$(MAKE) -C $< PREFIX=$(abspath $(ODIR)) BUILDMODE=static install
+
+$(ODIR)/lib/libssl.a: $(ODIR)/$(OPENSSL)
+	@echo Building OpenSSL...
+ifeq ($(TARGET), darwin)
+	@$(SHELL) -c "cd $< && ./Configure $(OPENSSL_OPTS) darwin64-x86_64-cc"
+else
+	@$(SHELL) -c "cd $< && ./config $(OPENSSL_OPTS)"
+endif
+	@$(MAKE) -C $< depend install
+
+# ------------
 
 .PHONY: all clean
-.SUFFIXES:
-.SUFFIXES: .c .o
+.PHONY: $(ODIR)/version.o
 
-vpath %.c src
-vpath %.h src
+.SUFFIXES:
+.SUFFIXES: .c .o .lua
+
+vpath %.c   src
+vpath %.h   src
+vpath %.lua scripts

NOTICE

@@ -82,36 +82,27 @@ OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
 USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 =========================================================================
-==  Tiny Mersenne Twister (TinyMT) Notice                              ==
+==  LuaJIT Notice                                                      ==
 =========================================================================
 
-Copyright (c) 2011 Mutsuo Saito, Makoto Matsumoto, Hiroshima University
-and The University of Tokyo. All rights reserved.
+LuaJIT -- a Just-In-Time Compiler for Lua. http://luajit.org/
 
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are
-met:
+Copyright (C) 2005-2013 Mike Pall. All rights reserved.
 
-    * Redistributions of source code must retain the above copyright
-      notice, this list of conditions and the following disclaimer.
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
 
-    * Redistributions in binary form must reproduce the above copyright
-      notice, this list of conditions and the following disclaimer in the
-      documentation and/or other materials provided with the
-      distribution.
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
 
-    * Neither the name of the Hiroshima University nor the names of its
-      contributors may be used to endorse or promote products derived
-      from this software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
-IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
-THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
-PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
-CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
-EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
-PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
-PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
-LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
-NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
-SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

README

@@ -4,23 +4,27 @@ wrk - a HTTP benchmarking tool
   load when run on a single multi-core CPU. It combines a multithreaded
   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. Details are available in SCRIPTING and
+  several examples are located in scripts/
+
 Basic Usage
 
-  wrk -t8 -c400 -r10m http://localhost:8080/index.html
+  wrk -t12 -c400 -d30s http://127.0.0.1:8080/index.html
 
-  This runs wrk with 8 threads, keeping 400 connections open, and making a
-  total of 10 million HTTP GET requests to http://localhost:8080/index.html
+  This runs a benchmark for 30 seconds, using 12 threads, and keeping
+  400 HTTP connections open.
 
   Output:
 
-  Making 10000000 requests to 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
 
@@ -29,9 +33,35 @@ Benchmarking Tips
   initial connection burst the server's listen(2) backlog should be greater
   than the number of concurrent connections being tested.
 
+  A user script that only changes the HTTP method, path, adds headers or
+  a body, will have no performance impact. Per-request actions, particularly
+  building a new HTTP request, and use of response() will necessarily reduce
+  the amount of load that can be generated.
+
 Acknowledgements
 
   wrk contains code from a number of open source projects including the
-  'ae' event loop from redis, the nginx/joyent/node.js 'http-parser' and
-  the Tiny Mersenne Twister PRNG. Please consult the NOTICE file for
-  licensing details.
+  'ae' event loop from redis, the nginx/joyent/node.js 'http-parser',
+  and Mike Pall's LuaJIT. Please consult the NOTICE file for licensing
+  details.
+
+Cryptography Notice
+
+  This distribution includes cryptographic software. The country in
+  which you currently reside may have restrictions on the import,
+  possession, use, and/or re-export to another country, of encryption
+  software. BEFORE using any encryption software, please check your
+  country's laws, regulations and policies concerning the import,
+  possession, or use, and re-export of encryption software, to see if
+  this is permitted. See <http://www.wassenaar.org/> for more
+  information.
+
+  The U.S. Government Department of Commerce, Bureau of Industry and
+  Security (BIS), has classified this software as Export Commodity
+  Control Number (ECCN) 5D002.C.1, which includes information security
+  software using or performing cryptographic functions with symmetric
+  algorithms. The form and manner of this distribution makes it
+  eligible for export under the License Exception ENC Technology
+  Software Unrestricted (TSU) exception (see the BIS Export
+  Administration Regulations, Section 740.13) for both object code and
+  source code.

SCRIPTING

@@ -0,0 +1,117 @@
+Overview
+
+  wrk supports executing a LuaJIT script during three distinct phases: setup,
+  running, and done. Each wrk thread has an independent scripting environment
+  and the setup & done phases execute in a separate environment which does
+  not participate in the running phase.
+
+  The public Lua API consists of a global table and a number of global
+  functions:
+
+  wrk = {
+    scheme  = "http",
+    host    = "localhost",
+    port    = nil,
+    method  = "GET",
+    path    = "/",
+    headers = {},
+    body    = nil,
+    thread  = <userdata>,
+  }
+
+  function wrk.format(method, path, headers, body)
+
+    wrk.format returns a HTTP request string containing the passed parameters
+    merged with values from the wrk table.
+
+  function wrk.lookup(host, service)
+
+    wrk.lookup returns a table containing all known addresses for the host
+    and service pair. This corresponds to the POSIX getaddrinfo() function.
+
+  function wrk.connect(addr)
+
+    wrk.connect returns true if the address can be connected to, otherwise
+    it returns false. The address must be one returned from wrk.lookup().
+
+  The following globals are optional, and if defined must be functions:
+
+    global setup    -- called during thread setup
+    global init     -- called when the thread is starting
+    global delay    -- called to get the request delay
+    global request  -- called to generate the HTTP request
+    global response -- called with HTTP response data
+    global done     -- called with results of run
+
+Setup
+
+  function setup(thread)
+
+  The setup phase begins after the target IP address has been resolved and all
+  threads have been initialized but not yet started.
+
+  setup() is called once for each thread and receives a userdata object
+  representing the thread.
+
+    thread.addr             - get or set the thread's server address
+    thread:get(name)        - get the value of a global in the thread's env
+    thread:set(name, value) - set the value of a global in the thread's env
+    thread:stop()           - stop the thread
+
+  Only boolean, nil, number, and string values or tables of the same may be
+  transfered via get()/set() and thread:stop() can only be called while the
+  thread is running.
+
+Running
+
+  function init(args)
+  function delay()
+  function request()
+  function response(status, headers, body)
+
+  The running phase begins with a single call to init(), followed by
+  a call to request() and response() for each request cycle.
+
+  The init() function receives any extra command line arguments for the
+  script which must be separated from wrk arguments with "--".
+
+  delay() returns the number of milliseconds to delay sending the next
+  request.
+
+  request() returns a string containing the HTTP request. Building a new
+  request each time is expensive, when testing a high performance server
+  one solution is to pre-generate all requests in init() and do a quick
+  lookup in request().
+
+  response() is called with the HTTP response status, headers, and body.
+  Parsing the headers and body is expensive, so if the response global is
+  nil after the call to init() wrk will ignore the headers and body.
+
+Done
+
+  function done(summary, latency, requests)
+
+  The done() function receives a table containing result data, and two
+  statistics objects representing the per-request latency and per-thread
+  request rate. Duration and latency are microsecond values and rate is
+  measured in requests per second.
+
+  latency.min              -- minimum value seen
+  latency.max              -- maximum value seen
+  latency.mean             -- average value seen
+  latency.stdev            -- standard deviation
+  latency:percentile(99.0) -- 99th percentile value
+  latency(i)               -- raw value and count
+
+  summary = {
+    duration = N,  -- run duration in microseconds
+    requests = N,  -- total completed requests
+    bytes    = N,  -- total bytes received
+    errors   = {
+      connect = N, -- total socket connection errors
+      read    = N, -- total socket read errors
+      write   = N, -- total socket write errors
+      status  = N, -- total HTTP status codes > 399
+      timeout = N  -- total request timeouts
+    }
+  }

scripts/addr.lua

@@ -0,0 +1,22 @@
+-- example script that demonstrates use of setup() to pass
+-- a random server address to each thread
+
+local addrs = nil
+
+function setup(thread)
+   if not addrs then
+      addrs = wrk.lookup(wrk.host, wrk.port or "http")
+      for i = #addrs, 1, -1 do
+         if not wrk.connect(addrs[i]) then
+            table.remove(addrs, i)
+         end
+      end
+   end
+
+   thread.addr = addrs[math.random(#addrs)]
+end
+
+function init(args)
+   local msg = "thread addr: %s"
+   print(msg:format(wrk.thread.addr))
+end

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/delay.lua

@@ -0,0 +1,6 @@
+-- example script that demonstrates adding a random
+-- 10-50ms delay before each request
+
+function delay()
+   return math.random(10, 50)
+end

scripts/pipeline.lua

@@ -0,0 +1,14 @@
+-- example script demonstrating HTTP pipelining
+
+init = function(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

scripts/setup.lua

@@ -0,0 +1,38 @@
+-- example script that demonstrates use of setup() to pass
+-- data to and from the threads
+
+local counter = 1
+local threads = {}
+
+function setup(thread)
+   thread:set("id", counter)
+   table.insert(threads, thread)
+   counter = counter + 1
+end
+
+function init(args)
+   requests  = 0
+   responses = 0
+
+   local msg = "thread %d created"
+   print(msg:format(id))
+end
+
+function request()
+   requests = requests + 1
+   return wrk.request()
+end
+
+function response(status, headers, body)
+   responses = responses + 1
+end
+
+function done(summary, latency, requests)
+   for index, thread in ipairs(threads) do
+      local id        = thread:get("id")
+      local requests  = thread:get("requests")
+      local responses = thread:get("responses")
+      local msg = "thread %d made %d requests and got %d responses"
+      print(msg:format(id, requests, responses))
+   end
+end

scripts/stop.lua

@@ -0,0 +1,10 @@
+-- example script that demonstrates use of thread:stop()
+
+local counter = 1
+
+function response()
+   if counter == 100 then
+      wrk.thread:stop()
+   end
+   counter = counter + 1
+end

src/config.h

@@ -5,9 +5,13 @@
 #define HAVE_KQUEUE
 #elif defined(__linux__)
 #define HAVE_EPOLL
-#define _POSIX_C_SOURCE 200809L
 #elif defined (__sun)
 #define HAVE_EVPORT
+#define _XPG6
+#define __EXTENSIONS__
+#include <stropts.h>
+#include <sys/filio.h>
+#include <sys/time.h>
 #endif
 
 #endif /* CONFIG_H */

src/http_parser.h

@@ -24,11 +24,15 @@
 extern "C" {
 #endif
 
-#define HTTP_PARSER_VERSION_MAJOR 1
-#define HTTP_PARSER_VERSION_MINOR 0
+/* Also update SONAME in the Makefile whenever you change these. */
+#define HTTP_PARSER_VERSION_MAJOR 2
+#define HTTP_PARSER_VERSION_MINOR 4
+#define HTTP_PARSER_VERSION_PATCH 2
 
 #include <sys/types.h>
 #if defined(_WIN32) && !defined(__MINGW32__) && (!defined(_MSC_VER) || _MSC_VER<1600)
+#include <BaseTsd.h>
+#include <stddef.h>
 typedef __int8 int8_t;
 typedef unsigned __int8 uint8_t;
 typedef __int16 int16_t;
@@ -37,9 +41,6 @@ typedef __int32 int32_t;
 typedef unsigned __int32 uint32_t;
 typedef __int64 int64_t;
 typedef unsigned __int64 uint64_t;
-
-typedef unsigned int size_t;
-typedef int ssize_t;
 #else
 #include <stdint.h>
 #endif
@@ -51,18 +52,17 @@ typedef int ssize_t;
 # define HTTP_PARSER_STRICT 1
 #endif
 
-/* Compile with -DHTTP_PARSER_DEBUG=1 to add extra debugging information to
- * the error reporting facility.
+/* Maximium header size allowed. If the macro is not defined
+ * before including this header then the default is used. To
+ * change the maximum header size, define the macro in the build
+ * environment (e.g. -DHTTP_MAX_HEADER_SIZE=<value>). To remove
+ * the effective limit on the size of the header, define the macro
+ * to a very large number (e.g. -DHTTP_MAX_HEADER_SIZE=0x7fffffff)
  */
-#ifndef HTTP_PARSER_DEBUG
-# define HTTP_PARSER_DEBUG 0
+#ifndef HTTP_MAX_HEADER_SIZE
+# define HTTP_MAX_HEADER_SIZE (80*1024)
 #endif
 
-
-/* Maximium header size allowed */
-#define HTTP_MAX_HEADER_SIZE (80*1024)
-
-
 typedef struct http_parser http_parser;
 typedef struct http_parser_settings http_parser_settings;
 
@@ -76,8 +76,8 @@ typedef struct http_parser_settings http_parser_settings;
  * HEAD request which may contain 'Content-Length' or 'Transfer-Encoding:
  * chunked' headers that indicate the presence of a body.
  *
- * http_data_cb does not return data chunks. It will be call arbitrarally
- * many times for each string. E.G. you might get 10 callbacks for "on_path"
+ * http_data_cb does not return data chunks. It will be called arbitrarily
+ * many times for each string. E.G. you might get 10 callbacks for "on_url"
  * each providing just a few characters more data.
  */
 typedef int (*http_data_cb) (http_parser*, const char *at, size_t length);
@@ -85,43 +85,46 @@ typedef int (*http_cb) (http_parser*);
 
 
 /* Request Methods */
-#define HTTP_METHOD_MAP(XX) \
-  XX(0, DELETE) \
-  XX(1, GET) \
-  XX(2, HEAD) \
-  XX(3, POST) \
-  XX(4, PUT) \
-  /* pathological */ \
-  XX(5, CONNECT) \
-  XX(6, OPTIONS) \
-  XX(7, TRACE) \
-  /* webdav */ \
-  XX(8, COPY) \
-  XX(9, LOCK) \
-  XX(10, MKCOL) \
-  XX(11, MOVE) \
-  XX(12, PROPFIND) \
-  XX(13, PROPPATCH) \
-  XX(14, UNLOCK) \
-  /* subversion */ \
-  XX(15, REPORT) \
-  XX(16, MKACTIVITY) \
-  XX(17, CHECKOUT) \
-  XX(18, MERGE) \
-  /* upnp */ \
-  XX(19, MSEARCH) \
-  XX(20, NOTIFY) \
-  XX(21, SUBSCRIBE) \
-  XX(22, UNSUBSCRIBE) \
-  /* RFC-5789 */ \
-  XX(23, PATCH) \
-  XX(24, PURGE) \
+#define HTTP_METHOD_MAP(XX)         \
+  XX(0,  DELETE,      DELETE)       \
+  XX(1,  GET,         GET)          \
+  XX(2,  HEAD,        HEAD)         \
+  XX(3,  POST,        POST)         \
+  XX(4,  PUT,         PUT)          \
+  /* pathological */                \
+  XX(5,  CONNECT,     CONNECT)      \
+  XX(6,  OPTIONS,     OPTIONS)      \
+  XX(7,  TRACE,       TRACE)        \
+  /* webdav */                      \
+  XX(8,  COPY,        COPY)         \
+  XX(9,  LOCK,        LOCK)         \
+  XX(10, MKCOL,       MKCOL)        \
+  XX(11, MOVE,        MOVE)         \
+  XX(12, PROPFIND,    PROPFIND)     \
+  XX(13, PROPPATCH,   PROPPATCH)    \
+  XX(14, SEARCH,      SEARCH)       \
+  XX(15, UNLOCK,      UNLOCK)       \
+  /* subversion */                  \
+  XX(16, REPORT,      REPORT)       \
+  XX(17, MKACTIVITY,  MKACTIVITY)   \
+  XX(18, CHECKOUT,    CHECKOUT)     \
+  XX(19, MERGE,       MERGE)        \
+  /* upnp */                        \
+  XX(20, MSEARCH,     M-SEARCH)     \
+  XX(21, NOTIFY,      NOTIFY)       \
+  XX(22, SUBSCRIBE,   SUBSCRIBE)    \
+  XX(23, UNSUBSCRIBE, UNSUBSCRIBE)  \
+  /* RFC-5789 */                    \
+  XX(24, PATCH,       PATCH)        \
+  XX(25, PURGE,       PURGE)        \
+  /* CalDAV */                      \
+  XX(26, MKCALENDAR,  MKCALENDAR)   \
 
 enum http_method
   {
-#define XX(num, name) HTTP_##name = num,
+#define XX(num, name, string) HTTP_##name = num,
   HTTP_METHOD_MAP(XX)
-#undef X
+#undef XX
   };
 
 
@@ -133,9 +136,10 @@ enum flags
   { F_CHUNKED               = 1 << 0
   , F_CONNECTION_KEEP_ALIVE = 1 << 1
   , F_CONNECTION_CLOSE      = 1 << 2
-  , F_TRAILING              = 1 << 3
-  , F_UPGRADE               = 1 << 4
-  , F_SKIPBODY              = 1 << 5
+  , F_CONNECTION_UPGRADE    = 1 << 3
+  , F_TRAILING              = 1 << 4
+  , F_UPGRADE               = 1 << 5
+  , F_SKIPBODY              = 1 << 6
   };
 
 
@@ -155,6 +159,7 @@ enum flags
   XX(CB_headers_complete, "the on_headers_complete callback failed") \
   XX(CB_body, "the on_body callback failed")                         \
   XX(CB_message_complete, "the on_message_complete callback failed") \
+  XX(CB_status, "the on_status callback failed")                     \
                                                                      \
   /* Parsing-related errors */                                       \
   XX(INVALID_EOF_STATE, "stream ended at an unexpected time")        \
@@ -195,21 +200,14 @@ enum http_errno {
 /* Get an http_errno value from an http_parser */
 #define HTTP_PARSER_ERRNO(p)            ((enum http_errno) (p)->http_errno)
 
-/* Get the line number that generated the current error */
-#if HTTP_PARSER_DEBUG
-#define HTTP_PARSER_ERRNO_LINE(p)       ((p)->error_lineno)
-#else
-#define HTTP_PARSER_ERRNO_LINE(p)       0
-#endif
-
 
 struct http_parser {
   /** PRIVATE **/
-  unsigned char type : 2;     /* enum http_parser_type */
-  unsigned char flags : 6;    /* F_* values from 'flags' enum; semi-public */
-  unsigned char state;        /* enum state from http_parser.c */
-  unsigned char header_state; /* enum header_state from http_parser.c */
-  unsigned char index;        /* index into current matcher */
+  unsigned int type : 2;         /* enum http_parser_type */
+  unsigned int flags : 6;        /* F_* values from 'flags' enum; semi-public */
+  unsigned int state : 8;        /* enum state from http_parser.c */
+  unsigned int header_state : 8; /* enum header_state from http_parser.c */
+  unsigned int index : 8;        /* index into current matcher */
 
   uint32_t nread;          /* # bytes read in various scenarios */
   uint64_t content_length; /* # bytes in body (0 if no Content-Length header) */
@@ -217,20 +215,16 @@ struct http_parser {
   /** READ-ONLY **/
   unsigned short http_major;
   unsigned short http_minor;
-  unsigned short status_code; /* responses only */
-  unsigned char method;       /* requests only */
-  unsigned char http_errno : 7;
+  unsigned int status_code : 16; /* responses only */
+  unsigned int method : 8;       /* requests only */
+  unsigned int http_errno : 7;
 
   /* 1 = Upgrade header was present and the parser has exited because of that.
    * 0 = No upgrade header present.
    * Should be checked when http_parser_execute() returns in addition to
    * error checking.
    */
-  unsigned char upgrade : 1;
-
-#if HTTP_PARSER_DEBUG
-  uint32_t error_lineno;
-#endif
+  unsigned int upgrade : 1;
 
   /** PUBLIC **/
   void *data; /* A pointer to get hook to the "connection" or "socket" object */
@@ -240,6 +234,7 @@ struct http_parser {
 struct http_parser_settings {
   http_cb      on_message_begin;
   http_data_cb on_url;
+  http_data_cb on_status;
   http_data_cb on_header_field;
   http_data_cb on_header_value;
   http_cb      on_headers_complete;
@@ -255,7 +250,8 @@ enum http_parser_url_fields
   , UF_PATH             = 3
   , UF_QUERY            = 4
   , UF_FRAGMENT         = 5
-  , UF_MAX              = 6
+  , UF_USERINFO         = 6
+  , UF_MAX              = 7
   };
 
 
@@ -277,9 +273,23 @@ struct http_parser_url {
 };
 
 
+/* Returns the library version. Bits 16-23 contain the major version number,
+ * bits 8-15 the minor version number and bits 0-7 the patch level.
+ * Usage example:
+ *
+ *   unsigned long version = http_parser_version();
+ *   unsigned major = (version >> 16) & 255;
+ *   unsigned minor = (version >> 8) & 255;
+ *   unsigned patch = version & 255;
+ *   printf("http_parser v%u.%u.%u\n", major, minor, patch);
+ */
+unsigned long http_parser_version(void);
+
 void http_parser_init(http_parser *parser, enum http_parser_type type);
 
 
+/* Executes the parser. Returns number of parsed bytes. Sets
+ * `parser->http_errno` on error. */
 size_t http_parser_execute(http_parser *parser,
                            const http_parser_settings *settings,
                            const char *data,
@@ -287,12 +297,12 @@ size_t http_parser_execute(http_parser *parser,
 
 
 /* If http_should_keep_alive() in the on_headers_complete or
- * on_message_complete callback returns true, then this will be should be
+ * on_message_complete callback returns 0, then this should be
  * the last message on the connection.
  * If you are the server, respond with the "Connection: close" header.
  * If you are the client, close the connection.
  */
-int http_should_keep_alive(http_parser *parser);
+int http_should_keep_alive(const http_parser *parser);
 
 /* Returns a string version of the HTTP method. */
 const char *http_method_str(enum http_method m);
@@ -311,6 +321,9 @@ int http_parser_parse_url(const char *buf, size_t buflen,
 /* Pause or un-pause the parser; a nonzero value pauses */
 void http_parser_pause(http_parser *parser, int paused);
 
+/* Checks if this is the final chunk of the body. */
+int http_body_is_final(const http_parser *parser);
+
 #ifdef __cplusplus
 }
 #endif

src/main.h

@@ -0,0 +1,54 @@
+#ifndef MAIN_H
+#define MAIN_H
+
+#include <ctype.h>
+#include <errno.h>
+#include <fcntl.h>
+#include <getopt.h>
+#include <math.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/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 record_rate(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 response_complete(http_parser *);
+static int header_field(http_parser *, const char *, size_t);
+static int header_value(http_parser *, const char *, size_t);
+static int response_body(http_parser *, const char *, size_t);
+
+static uint64_t time_us();
+
+static int parse_args(struct config *, char **, struct http_parser_url *, char **, int, char **);
+static char *copy_url_part(char *, struct http_parser_url *, enum http_parser_url_fields);
+
+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,39 @@
+// Copyright (C) 2013 - Will Glozer.  All rights reserved.
+
+#include <errno.h>
+#include <unistd.h>
+#include <sys/ioctl.h>
+
+#include "net.h"
+
+status sock_connect(connection *c, char *host) {
+    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;
+}
+
+size_t sock_readable(connection *c) {
+    int n, rc;
+    rc = ioctl(c->fd, FIONREAD, &n);
+    return rc == -1 ? 0 : n;
+}

src/net.h

@@ -0,0 +1,29 @@
+#ifndef NET_H
+#define NET_H
+
+#include "config.h"
+#include <stdint.h>
+#include <openssl/ssl.h>
+#include "wrk.h"
+
+typedef enum {
+    OK,
+    ERROR,
+    RETRY
+} status;
+
+struct sock {
+    status ( *connect)(connection *, char *);
+    status (   *close)(connection *);
+    status (    *read)(connection *, size_t *);
+    status (   *write)(connection *, char *, size_t, size_t *);
+    size_t (*readable)(connection *);
+};
+
+status sock_connect(connection *, char *);
+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/script.c

@@ -0,0 +1,576 @@
+// Copyright (C) 2013 - Will Glozer.  All rights reserved.
+
+#include <stdlib.h>
+#include <string.h>
+#include "script.h"
+#include "http_parser.h"
+#include "zmalloc.h"
+
+typedef struct {
+    char *name;
+    int   type;
+    void *value;
+} table_field;
+
+static int script_addr_tostring(lua_State *);
+static int script_addr_gc(lua_State *);
+static int script_stats_call(lua_State *);
+static int script_stats_len(lua_State *);
+static int script_stats_index(lua_State *);
+static int script_thread_index(lua_State *);
+static int script_thread_newindex(lua_State *);
+static int script_wrk_lookup(lua_State *);
+static int script_wrk_connect(lua_State *);
+
+static void set_fields(lua_State *, int, const table_field *);
+static void set_field(lua_State *, int, char *, int);
+static int push_url_part(lua_State *, char *, struct http_parser_url *, enum http_parser_url_fields);
+
+static const struct luaL_reg addrlib[] = {
+    { "__tostring", script_addr_tostring   },
+    { "__gc"    ,   script_addr_gc         },
+    { NULL,         NULL                   }
+};
+
+static const struct luaL_reg statslib[] = {
+    { "__call",     script_stats_call      },
+    { "__index",    script_stats_index     },
+    { "__len",      script_stats_len       },
+    { NULL,         NULL                   }
+};
+
+static const struct luaL_reg threadlib[] = {
+    { "__index",    script_thread_index    },
+    { "__newindex", script_thread_newindex },
+    { NULL,         NULL                   }
+};
+
+lua_State *script_create(char *file, char *url, char **headers) {
+    lua_State *L = luaL_newstate();
+    luaL_openlibs(L);
+    (void) luaL_dostring(L, "wrk = require \"wrk\"");
+
+    luaL_newmetatable(L, "wrk.addr");
+    luaL_register(L, NULL, addrlib);
+    luaL_newmetatable(L, "wrk.stats");
+    luaL_register(L, NULL, statslib);
+    luaL_newmetatable(L, "wrk.thread");
+    luaL_register(L, NULL, threadlib);
+
+    struct http_parser_url parts = {};
+    script_parse_url(url, &parts);
+    char *path = "/";
+
+    if (parts.field_set & (1 << UF_PATH)) {
+        path = &url[parts.field_data[UF_PATH].off];
+    }
+
+    const table_field fields[] = {
+        { "lookup",  LUA_TFUNCTION, script_wrk_lookup  },
+        { "connect", LUA_TFUNCTION, script_wrk_connect },
+        { "path",    LUA_TSTRING,   path               },
+        { NULL,      0,             NULL               },
+    };
+
+    lua_getglobal(L, "wrk");
+
+    set_field(L, 4, "scheme", push_url_part(L, url, &parts, UF_SCHEMA));
+    set_field(L, 4, "host",   push_url_part(L, url, &parts, UF_HOST));
+    set_field(L, 4, "port",   push_url_part(L, url, &parts, UF_PORT));
+    set_fields(L, 4, fields);
+
+    lua_getfield(L, 4, "headers");
+    for (char **h = headers; *h; h++) {
+        char *p = strchr(*h, ':');
+        if (p && p[1] == ' ') {
+            lua_pushlstring(L, *h, p - *h);
+            lua_pushstring(L, p + 2);
+            lua_settable(L, 5);
+        }
+    }
+    lua_pop(L, 5);
+
+    if (file && luaL_dofile(L, file)) {
+        const char *cause = lua_tostring(L, -1);
+        fprintf(stderr, "%s: %s\n", file, cause);
+    }
+
+    return L;
+}
+
+bool script_resolve(lua_State *L, char *host, char *service) {
+    lua_getglobal(L, "wrk");
+
+    lua_getfield(L, -1, "resolve");
+    lua_pushstring(L, host);
+    lua_pushstring(L, service);
+    lua_call(L, 2, 0);
+
+    lua_getfield(L, -1, "addrs");
+    size_t count = lua_objlen(L, -1);
+    lua_pop(L, 2);
+    return count > 0;
+}
+
+void script_push_thread(lua_State *L, thread *t) {
+    thread **ptr = (thread **) lua_newuserdata(L, sizeof(thread **));
+    *ptr = t;
+    luaL_getmetatable(L, "wrk.thread");
+    lua_setmetatable(L, -2);
+}
+
+void script_init(lua_State *L, thread *t, int argc, char **argv) {
+    lua_getglobal(t->L, "wrk");
+
+    script_push_thread(t->L, t);
+    lua_setfield(t->L, -2, "thread");
+
+    lua_getglobal(L, "wrk");
+    lua_getfield(L, -1, "setup");
+    script_push_thread(L, t);
+    lua_call(L, 1, 0);
+    lua_pop(L, 1);
+
+    lua_getfield(t->L, -1, "init");
+    lua_newtable(t->L);
+    for (int i = 0; i < argc; i++) {
+        lua_pushstring(t->L, argv[i]);
+        lua_rawseti(t->L, -2, i);
+    }
+    lua_call(t->L, 1, 0);
+    lua_pop(t->L, 1);
+}
+
+uint64_t script_delay(lua_State *L) {
+    lua_getglobal(L, "delay");
+    lua_call(L, 0, 1);
+    uint64_t delay = lua_tonumber(L, -1);
+    lua_pop(L, 1);
+    return delay;
+}
+
+void script_request(lua_State *L, char **buf, size_t *len) {
+    int pop = 1;
+    lua_getglobal(L, "request");
+    if (!lua_isfunction(L, -1)) {
+        lua_getglobal(L, "wrk");
+        lua_getfield(L, -1, "request");
+        pop += 2;
+    }
+    lua_call(L, 0, 1);
+    const char *str = lua_tolstring(L, -1, len);
+    *buf = realloc(*buf, *len);
+    memcpy(*buf, str, *len);
+    lua_pop(L, pop);
+}
+
+void script_response(lua_State *L, int status, buffer *headers, buffer *body) {
+    lua_getglobal(L, "response");
+    lua_pushinteger(L, status);
+    lua_newtable(L);
+
+    for (char *c = headers->buffer; c < headers->cursor; ) {
+        c = buffer_pushlstring(L, c);
+        c = buffer_pushlstring(L, c);
+        lua_rawset(L, -3);
+    }
+
+    lua_pushlstring(L, body->buffer, body->cursor - body->buffer);
+    lua_call(L, 3, 0);
+
+    buffer_reset(headers);
+    buffer_reset(body);
+}
+
+bool script_is_function(lua_State *L, char *name) {
+    lua_getglobal(L, name);
+    bool is_function = lua_isfunction(L, -1);
+    lua_pop(L, 1);
+    return is_function;
+}
+
+bool script_is_static(lua_State *L) {
+    return !script_is_function(L, "request");
+}
+
+bool script_want_response(lua_State *L) {
+    return script_is_function(L, "response");
+}
+
+bool script_has_delay(lua_State *L) {
+    return script_is_function(L, "delay");
+}
+
+bool script_has_done(lua_State *L) {
+    return script_is_function(L, "done");
+}
+
+void script_header_done(lua_State *L, luaL_Buffer *buffer) {
+    luaL_pushresult(buffer);
+}
+
+void script_summary(lua_State *L, uint64_t duration, uint64_t requests, uint64_t bytes) {
+    const table_field fields[] = {
+        { "duration", LUA_TNUMBER, &duration },
+        { "requests", LUA_TNUMBER, &requests },
+        { "bytes",    LUA_TNUMBER, &bytes    },
+        { NULL,       0,           NULL      },
+    };
+    lua_newtable(L);
+    set_fields(L, 1, fields);
+}
+
+void script_errors(lua_State *L, errors *errors) {
+    uint64_t e[] = {
+        errors->connect,
+        errors->read,
+        errors->write,
+        errors->status,
+        errors->timeout
+    };
+    const table_field fields[] = {
+        { "connect", LUA_TNUMBER, &e[0] },
+        { "read",    LUA_TNUMBER, &e[1] },
+        { "write",   LUA_TNUMBER, &e[2] },
+        { "status",  LUA_TNUMBER, &e[3] },
+        { "timeout", LUA_TNUMBER, &e[4] },
+        { NULL,      0,           NULL  },
+    };
+    lua_newtable(L);
+    set_fields(L, 2, fields);
+    lua_setfield(L, 1, "errors");
+}
+
+void script_push_stats(lua_State *L, stats *s) {
+    stats **ptr = (stats **) lua_newuserdata(L, sizeof(stats **));
+    *ptr = s;
+    luaL_getmetatable(L, "wrk.stats");
+    lua_setmetatable(L, -2);
+}
+
+void script_done(lua_State *L, stats *latency, stats *requests) {
+    lua_getglobal(L, "done");
+    lua_pushvalue(L, 1);
+
+    script_push_stats(L, latency);
+    script_push_stats(L, requests);
+
+    lua_call(L, 3, 0);
+    lua_pop(L, 1);
+}
+
+static int verify_request(http_parser *parser) {
+    size_t *count = parser->data;
+    (*count)++;
+    return 0;
+}
+
+size_t script_verify_request(lua_State *L) {
+    http_parser_settings settings = {
+        .on_message_complete = verify_request
+    };
+    http_parser parser;
+    char *request = NULL;
+    size_t len, count = 0;
+
+    script_request(L, &request, &len);
+    http_parser_init(&parser, HTTP_REQUEST);
+    parser.data = &count;
+
+    size_t parsed = http_parser_execute(&parser, &settings, request, len);
+
+    if (parsed != len || count == 0) {
+        enum http_errno err = HTTP_PARSER_ERRNO(&parser);
+        const char *desc = http_errno_description(err);
+        const char *msg = err != HPE_OK ? desc : "incomplete request";
+        int line = 1, column = 1;
+
+        for (char *c = request; c < request + parsed; c++) {
+            column++;
+            if (*c == '\n') {
+                column = 1;
+                line++;
+            }
+        }
+
+        fprintf(stderr, "%s at %d:%d\n", msg, line, column);
+        exit(1);
+    }
+
+    return count;
+}
+
+static struct addrinfo *checkaddr(lua_State *L) {
+    struct addrinfo *addr = luaL_checkudata(L, -1, "wrk.addr");
+    luaL_argcheck(L, addr != NULL, 1, "`addr' expected");
+    return addr;
+}
+
+void script_addr_copy(struct addrinfo *src, struct addrinfo *dst) {
+    *dst = *src;
+    dst->ai_addr = zmalloc(src->ai_addrlen);
+    memcpy(dst->ai_addr, src->ai_addr, src->ai_addrlen);
+}
+
+struct addrinfo *script_addr_clone(lua_State *L, struct addrinfo *addr) {
+    struct addrinfo *udata = lua_newuserdata(L, sizeof(*udata));
+    luaL_getmetatable(L, "wrk.addr");
+    lua_setmetatable(L, -2);
+    script_addr_copy(addr, udata);
+    return udata;
+}
+
+static int script_addr_tostring(lua_State *L) {
+    struct addrinfo *addr = checkaddr(L);
+    char host[NI_MAXHOST];
+    char service[NI_MAXSERV];
+
+    int flags = NI_NUMERICHOST | NI_NUMERICSERV;
+    int rc = getnameinfo(addr->ai_addr, addr->ai_addrlen, host, NI_MAXHOST, service, NI_MAXSERV, flags);
+    if (rc != 0) {
+        const char *msg = gai_strerror(rc);
+        return luaL_error(L, "addr tostring failed %s", msg);
+    }
+
+    lua_pushfstring(L, "%s:%s", host, service);
+    return 1;
+}
+
+static int script_addr_gc(lua_State *L) {
+    struct addrinfo *addr = checkaddr(L);
+    zfree(addr->ai_addr);
+    return 0;
+}
+
+static stats *checkstats(lua_State *L) {
+    stats **s = luaL_checkudata(L, 1, "wrk.stats");
+    luaL_argcheck(L, s != NULL, 1, "`stats' expected");
+    return *s;
+}
+
+static int script_stats_percentile(lua_State *L) {
+    stats *s = checkstats(L);
+    lua_Number p = luaL_checknumber(L, 2);
+    lua_pushnumber(L, stats_percentile(s, p));
+    return 1;
+}
+
+static int script_stats_call(lua_State *L) {
+    stats *s = checkstats(L);
+    uint64_t index = lua_tonumber(L, 2);
+    uint64_t count;
+    lua_pushnumber(L, stats_value_at(s, index - 1, &count));
+    lua_pushnumber(L, count);
+    return 2;
+}
+
+static int script_stats_index(lua_State *L) {
+    stats *s = checkstats(L);
+    const char *method = lua_tostring(L, 2);
+    if (!strcmp("min",   method)) lua_pushnumber(L, s->min);
+    if (!strcmp("max",   method)) lua_pushnumber(L, s->max);
+    if (!strcmp("mean",  method)) lua_pushnumber(L, stats_mean(s));
+    if (!strcmp("stdev", method)) lua_pushnumber(L, stats_stdev(s, stats_mean(s)));
+    if (!strcmp("percentile", method)) {
+        lua_pushcfunction(L, script_stats_percentile);
+    }
+    return 1;
+}
+
+static int script_stats_len(lua_State *L) {
+    stats *s = checkstats(L);
+    lua_pushinteger(L, stats_popcount(s));
+    return 1;
+}
+
+static thread *checkthread(lua_State *L) {
+    thread **t = luaL_checkudata(L, 1, "wrk.thread");
+    luaL_argcheck(L, t != NULL, 1, "`thread' expected");
+    return *t;
+}
+
+static int script_thread_get(lua_State *L) {
+    thread *t = checkthread(L);
+    const char *key = lua_tostring(L, -1);
+    lua_getglobal(t->L, key);
+    script_copy_value(t->L, L, -1);
+    lua_pop(t->L, 1);
+    return 1;
+}
+
+static int script_thread_set(lua_State *L) {
+    thread *t = checkthread(L);
+    const char *name = lua_tostring(L, -2);
+    script_copy_value(L, t->L, -1);
+    lua_setglobal(t->L, name);
+    return 0;
+}
+
+static int script_thread_stop(lua_State *L) {
+    thread *t = checkthread(L);
+    aeStop(t->loop);
+    return 0;
+}
+
+static int script_thread_index(lua_State *L) {
+    thread *t = checkthread(L);
+    const char *key = lua_tostring(L, 2);
+    if (!strcmp("get",  key)) lua_pushcfunction(L, script_thread_get);
+    if (!strcmp("set",  key)) lua_pushcfunction(L, script_thread_set);
+    if (!strcmp("stop", key)) lua_pushcfunction(L, script_thread_stop);
+    if (!strcmp("addr", key)) script_addr_clone(L, t->addr);
+    return 1;
+}
+
+static int script_thread_newindex(lua_State *L) {
+    thread *t = checkthread(L);
+    const char *key = lua_tostring(L, -2);
+    if (!strcmp("addr", key)) {
+        struct addrinfo *addr = checkaddr(L);
+        if (t->addr) zfree(t->addr->ai_addr);
+        t->addr = zrealloc(t->addr, sizeof(*addr));
+        script_addr_copy(addr, t->addr);
+    } else {
+        luaL_error(L, "cannot set '%s' on thread", luaL_typename(L, -1));
+    }
+    return 0;
+}
+
+static int script_wrk_lookup(lua_State *L) {
+    struct addrinfo *addrs;
+    struct addrinfo hints = {
+        .ai_family   = AF_UNSPEC,
+        .ai_socktype = SOCK_STREAM
+    };
+    int rc, index = 1;
+
+    const char *host    = lua_tostring(L, -2);
+    const char *service = lua_tostring(L, -1);
+
+    if ((rc = getaddrinfo(host, service, &hints, &addrs)) != 0) {
+        const char *msg = gai_strerror(rc);
+        fprintf(stderr, "unable to resolve %s:%s %s\n", host, service, msg);
+        exit(1);
+    }
+
+    lua_newtable(L);
+    for (struct addrinfo *addr = addrs; addr != NULL; addr = addr->ai_next) {
+        script_addr_clone(L, addr);
+        lua_rawseti(L, -2, index++);
+    }
+
+    freeaddrinfo(addrs);
+    return 1;
+}
+
+static int script_wrk_connect(lua_State *L) {
+    struct addrinfo *addr = checkaddr(L);
+    int fd, connected = 0;
+    if ((fd = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol)) != -1) {
+        connected = connect(fd, addr->ai_addr, addr->ai_addrlen) == 0;
+        close(fd);
+    }
+    lua_pushboolean(L, connected);
+    return 1;
+}
+
+void script_copy_value(lua_State *src, lua_State *dst, int index) {
+    switch (lua_type(src, index)) {
+        case LUA_TBOOLEAN:
+            lua_pushboolean(dst, lua_toboolean(src, index));
+            break;
+        case LUA_TNIL:
+            lua_pushnil(dst);
+            break;
+        case LUA_TNUMBER:
+            lua_pushnumber(dst, lua_tonumber(src, index));
+            break;
+        case LUA_TSTRING:
+            lua_pushstring(dst, lua_tostring(src, index));
+            break;
+        case LUA_TTABLE:
+            lua_newtable(dst);
+            lua_pushnil(src);
+            while (lua_next(src, index - 1)) {
+                script_copy_value(src, dst, -2);
+                script_copy_value(src, dst, -1);
+                lua_settable(dst, -3);
+                lua_pop(src, 1);
+            }
+            lua_pop(src, 1);
+            break;
+        default:
+            luaL_error(src, "cannot transfer '%s' to thread", luaL_typename(src, index));
+    }
+}
+
+int script_parse_url(char *url, struct http_parser_url *parts) {
+    if (!http_parser_parse_url(url, strlen(url), 0, parts)) {
+        if (!(parts->field_set & (1 << UF_SCHEMA))) return 0;
+        if (!(parts->field_set & (1 << UF_HOST)))   return 0;
+        return 1;
+    }
+    return 0;
+}
+
+static int push_url_part(lua_State *L, char *url, struct http_parser_url *parts, enum http_parser_url_fields field) {
+    int type = parts->field_set & (1 << field) ? LUA_TSTRING : LUA_TNIL;
+    uint16_t off, len;
+    switch (type) {
+        case LUA_TSTRING:
+            off = parts->field_data[field].off;
+            len = parts->field_data[field].len;
+            lua_pushlstring(L, &url[off], len);
+            break;
+        case LUA_TNIL:
+            lua_pushnil(L);
+    }
+    return type;
+}
+
+static void set_field(lua_State *L, int index, char *field, int type) {
+    (void) type;
+    lua_setfield(L, index, field);
+}
+
+static void set_fields(lua_State *L, int index, const table_field *fields) {
+    for (int i = 0; fields[i].name; i++) {
+        table_field f = fields[i];
+        switch (f.value == NULL ? LUA_TNIL : f.type) {
+            case LUA_TFUNCTION:
+                lua_pushcfunction(L, (lua_CFunction) f.value);
+                break;
+            case LUA_TNUMBER:
+                lua_pushinteger(L, *((lua_Integer *) f.value));
+                break;
+            case LUA_TSTRING:
+                lua_pushstring(L, (const char *) f.value);
+                break;
+            case LUA_TNIL:
+                lua_pushnil(L);
+                break;
+        }
+        lua_setfield(L, index, f.name);
+    }
+}
+
+void buffer_append(buffer *b, const char *data, size_t len) {
+    size_t used = b->cursor - b->buffer;
+    while (used + len + 1 >= b->length) {
+        b->length += 1024;
+        b->buffer  = realloc(b->buffer, b->length);
+        b->cursor  = b->buffer + used;
+    }
+    memcpy(b->cursor, data, len);
+    b->cursor += len;
+}
+
+void buffer_reset(buffer *b) {
+    b->cursor = b->buffer;
+}
+
+char *buffer_pushlstring(lua_State *L, char *start) {
+    char *end = strchr(start, 0);
+    lua_pushlstring(L, start, end - start);
+    return end + 1;
+}

src/script.h

@@ -0,0 +1,38 @@
+#ifndef SCRIPT_H
+#define SCRIPT_H
+
+#include <stdbool.h>
+#include <luajit-2.0/lua.h>
+#include <luajit-2.0/lualib.h>
+#include <luajit-2.0/lauxlib.h>
+#include <unistd.h>
+#include "stats.h"
+#include "wrk.h"
+
+lua_State *script_create(char *, char *, char **);
+
+bool script_resolve(lua_State *, char *, char *);
+void script_setup(lua_State *, thread *);
+void script_done(lua_State *, stats *, stats *);
+
+void script_init(lua_State *, thread *, int, char **);
+uint64_t script_delay(lua_State *);
+void script_request(lua_State *, char **, size_t *);
+void script_response(lua_State *, int, buffer *, buffer *);
+size_t script_verify_request(lua_State *L);
+
+bool script_is_static(lua_State *);
+bool script_want_response(lua_State *L);
+bool script_has_delay(lua_State *L);
+bool script_has_done(lua_State *L);
+void script_summary(lua_State *, uint64_t, uint64_t, uint64_t);
+void script_errors(lua_State *, errors *);
+
+void script_copy_value(lua_State *, lua_State *, int);
+int script_parse_url(char *, struct http_parser_url *);
+
+void buffer_append(buffer *, const char *, size_t);
+void buffer_reset(buffer *);
+char *buffer_pushlstring(lua_State *, char *);
+
+#endif /* SCRIPT_H */

src/ssl.c

@@ -0,0 +1,100 @@
+// 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(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);
+            SSL_CTX_set_session_cache_mode(ctx, SSL_SESS_CACHE_CLIENT);
+        }
+    }
+
+    return ctx;
+}
+
+status ssl_connect(connection *c, char *host) {
+    int r;
+    SSL_set_fd(c->ssl, c->fd);
+    SSL_set_tlsext_host_name(c->ssl, host);
+    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;
+}
+
+size_t ssl_readable(connection *c) {
+    return SSL_pending(c->ssl);
+}

src/ssl.h

@@ -0,0 +1,14 @@
+#ifndef SSL_H
+#define SSL_H
+
+#include "net.h"
+
+SSL_CTX *ssl_init();
+
+status ssl_connect(connection *, char *);
+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/stats.c

@@ -7,56 +7,60 @@
 #include "stats.h"
 #include "zmalloc.h"
 
-stats *stats_alloc(uint64_t samples) {
-    stats *stats = zcalloc(sizeof(stats) + sizeof(uint64_t) * samples);
-    stats->samples = samples;
-    return stats;
+stats *stats_alloc(uint64_t max) {
+    uint64_t limit = max + 1;
+    stats *s = zcalloc(sizeof(stats) + sizeof(uint64_t) * limit);
+    s->limit = limit;
+    s->min   = UINT64_MAX;
+    return s;
 }
 
 void stats_free(stats *stats) {
     zfree(stats);
 }
 
-void stats_record(stats *stats, uint64_t x) {
-    stats->data[stats->index++] = x;
-    if (stats->limit < stats->samples)  stats->limit++;
-    if (stats->index == stats->samples) stats->index = 0;
+int stats_record(stats *stats, uint64_t n) {
+    if (n >= stats->limit) return 0;
+    __sync_fetch_and_add(&stats->data[n], 1);
+    __sync_fetch_and_add(&stats->count, 1);
+    uint64_t min = stats->min;
+    uint64_t max = stats->max;
+    while (n < min) min = __sync_val_compare_and_swap(&stats->min, min, n);
+    while (n > max) max = __sync_val_compare_and_swap(&stats->max, max, n);
+    return 1;
 }
 
-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;
+void stats_correct(stats *stats, int64_t expected) {
+    for (uint64_t n = expected * 2; n <= stats->max; n++) {
+        uint64_t count = stats->data[n];
+        int64_t m = (int64_t) n - expected;
+        while (count && m > expected) {
+            stats->data[m] += count;
+            stats->count += count;
+            m -= expected;
+        }
     }
-    return min;
-}
-
-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_mean(stats *stats) {
+    if (stats->count == 0) return 0.0;
+
     uint64_t sum = 0;
-    if (stats->limit == 0) return 0.0;
-    for (uint64_t i = 0; i < stats->limit; i++) {
-        sum += stats->data[i];
+    for (uint64_t i = stats->min; i <= stats->max; i++) {
+        sum += stats->data[i] * i;
     }
-    return sum / (long double) stats->limit;
+    return sum / (long double) stats->count;
 }
 
 long double stats_stdev(stats *stats, long double mean) {
     long double sum = 0.0;
-    if (stats->limit < 2) return 0.0;
-    for (uint64_t i = 0; i < stats->limit; i++) {
-        sum += powl(stats->data[i] - mean, 2);
+    if (stats->count < 2) return 0.0;
+    for (uint64_t i = stats->min; i <= stats->max; i++) {
+        if (stats->data[i]) {
+            sum += powl(i - mean, 2) * stats->data[i];
+        }
     }
-    return sqrtl(sum / (stats->limit - 1));
+    return sqrtl(sum / (stats->count - 1));
 }
 
 long double stats_within_stdev(stats *stats, long double mean, long double stdev, uint64_t n) {
@@ -64,10 +68,40 @@ long double stats_within_stdev(stats *stats, long double mean, long double stdev
     long double lower = mean - (stdev * n);
     uint64_t sum = 0;
 
-    for (uint64_t i = 0; i < stats->limit; i++) {
-        uint64_t x = stats->data[i];
-        if (x >= lower && x <= upper) sum++;
+    for (uint64_t i = stats->min; i <= stats->max; i++) {
+        if (i >= lower && i <= upper) {
+            sum += stats->data[i];
+        }
     }
 
-    return (sum / (long double) stats->limit) * 100;
+    return (sum / (long double) stats->count) * 100;
+}
+
+uint64_t stats_percentile(stats *stats, long double p) {
+    uint64_t rank = round((p / 100.0) * stats->count + 0.5);
+    uint64_t total = 0;
+    for (uint64_t i = stats->min; i <= stats->max; i++) {
+        total += stats->data[i];
+        if (total >= rank) return i;
+    }
+    return 0;
+}
+
+uint64_t stats_popcount(stats *stats) {
+    uint64_t count = 0;
+    for (uint64_t i = stats->min; i <= stats->max; i++) {
+        if (stats->data[i]) count++;
+    }
+    return count;
+}
+
+uint64_t stats_value_at(stats *stats, uint64_t index, uint64_t *count) {
+    *count = 0;
+    for (uint64_t i = stats->min; i <= stats->max; i++) {
+        if (stats->data[i] && (*count)++ == index) {
+            *count = stats->data[i];
+            return i;
+        }
+    }
+    return 0;
 }

src/stats.h

@@ -1,21 +1,40 @@
 #ifndef STATS_H
 #define STATS_H
 
+#include <stdbool.h>
+#include <stdint.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;
+    uint32_t connect;
+    uint32_t read;
+    uint32_t write;
+    uint32_t status;
+    uint32_t timeout;
+} errors;
+
+typedef struct {
+    uint64_t count;
     uint64_t limit;
+    uint64_t min;
+    uint64_t max;
     uint64_t data[];
 } stats;
 
 stats *stats_alloc(uint64_t);
 void stats_free(stats *);
-void stats_record(stats *, uint64_t);
-uint64_t stats_min(stats *);
-uint64_t stats_max(stats *);
+
+int stats_record(stats *, uint64_t);
+void stats_correct(stats *, int64_t);
+
 long double stats_mean(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);
+
+uint64_t stats_popcount(stats *);
+uint64_t stats_value_at(stats *stats, uint64_t, uint64_t *);
 
 #endif /* STATS_H */
-

src/tinymt64.c

@@ -1,129 +0,0 @@
-/**
- * @file tinymt64.c
- *
- * @brief 64-bit Tiny Mersenne Twister only 127 bit internal state
- *
- * @author Mutsuo Saito (Hiroshima University)
- * @author Makoto Matsumoto (The University of Tokyo)
- *
- * Copyright (C) 2011 Mutsuo Saito, Makoto Matsumoto,
- * Hiroshima University and The University of Tokyo.
- * All rights reserved.
- *
- * The 3-clause BSD License is applied to this software, see
- * LICENSE.txt
- */
-#include "tinymt64.h"
-
-#define MIN_LOOP 8
-
-/**
- * This function represents a function used in the initialization
- * by init_by_array
- * @param[in] x 64-bit integer
- * @return 64-bit integer
- */
-static uint64_t ini_func1(uint64_t x) {
-    return (x ^ (x >> 59)) * UINT64_C(2173292883993);
-}
-
-/**
- * This function represents a function used in the initialization
- * by init_by_array
- * @param[in] x 64-bit integer
- * @return 64-bit integer
- */
-static uint64_t ini_func2(uint64_t x) {
-    return (x ^ (x >> 59)) * UINT64_C(58885565329898161);
-}
-
-/**
- * This function certificate the period of 2^127-1.
- * @param random tinymt state vector.
- */
-static void period_certification(tinymt64_t * random) {
-    if ((random->status[0] & TINYMT64_MASK) == 0 &&
-	random->status[1] == 0) {
-	random->status[0] = 'T';
-	random->status[1] = 'M';
-    }
-}
-
-/**
- * This function initializes the internal state array with a 64-bit
- * unsigned integer seed.
- * @param random tinymt state vector.
- * @param seed a 64-bit unsigned integer used as a seed.
- */
-void tinymt64_init(tinymt64_t * random, uint64_t seed) {
-    random->status[0] = seed ^ ((uint64_t)random->mat1 << 32);
-    random->status[1] = random->mat2 ^ random->tmat;
-    for (int i = 1; i < MIN_LOOP; i++) {
-	random->status[i & 1] ^= i + UINT64_C(6364136223846793005)
-	    * (random->status[(i - 1) & 1]
-	       ^ (random->status[(i - 1) & 1] >> 62));
-    }
-    period_certification(random);
-}
-
-/**
- * This function initializes the internal state array,
- * with an array of 64-bit unsigned integers used as seeds
- * @param random tinymt state vector.
- * @param init_key the array of 64-bit integers, used as a seed.
- * @param key_length the length of init_key.
- */
-void tinymt64_init_by_array(tinymt64_t * random, const uint64_t init_key[],
-			    int key_length) {
-    const int lag = 1;
-    const int mid = 1;
-    const int size = 4;
-    int i, j;
-    int count;
-    uint64_t r;
-    uint64_t st[4];
-
-    st[0] = 0;
-    st[1] = random->mat1;
-    st[2] = random->mat2;
-    st[3] = random->tmat;
-    if (key_length + 1 > MIN_LOOP) {
-	count = key_length + 1;
-    } else {
-	count = MIN_LOOP;
-    }
-    r = ini_func1(st[0] ^ st[mid % size]
-		  ^ st[(size - 1) % size]);
-    st[mid % size] += r;
-    r += key_length;
-    st[(mid + lag) % size] += r;
-    st[0] = r;
-    count--;
-    for (i = 1, j = 0; (j < count) && (j < key_length); j++) {
-	r = ini_func1(st[i] ^ st[(i + mid) % size] ^ st[(i + size - 1) % size]);
-	st[(i + mid) % size] += r;
-	r += init_key[j] + i;
-	st[(i + mid + lag) % size] += r;
-	st[i] = r;
-	i = (i + 1) % size;
-    }
-    for (; j < count; j++) {
-	r = ini_func1(st[i] ^ st[(i + mid) % size] ^ st[(i + size - 1) % size]);
-	st[(i + mid) % size] += r;
-	r += i;
-	st[(i + mid + lag) % size] += r;
-	st[i] = r;
-	i = (i + 1) % size;
-    }
-    for (j = 0; j < size; j++) {
-	r = ini_func2(st[i] + st[(i + mid) % size] + st[(i + size - 1) % size]);
-	st[(i + mid) % size] ^= r;
-	r -= i;
-	st[(i + mid + lag) % size] ^= r;
-	st[i] = r;
-	i = (i + 1) % size;
-    }
-    random->status[0] = st[0] ^ st[1];
-    random->status[1] = st[2] ^ st[3];
-    period_certification(random);
-}

src/tinymt64.h

@@ -1,210 +0,0 @@
-#ifndef TINYMT64_H
-#define TINYMT64_H
-/**
- * @file tinymt64.h
- *
- * @brief Tiny Mersenne Twister only 127 bit internal state
- *
- * @author Mutsuo Saito (Hiroshima University)
- * @author Makoto Matsumoto (The University of Tokyo)
- *
- * Copyright (C) 2011 Mutsuo Saito, Makoto Matsumoto,
- * Hiroshima University and The University of Tokyo.
- * All rights reserved.
- *
- * The 3-clause BSD License is applied to this software, see
- * LICENSE.txt
- */
-
-#include <stdint.h>
-#include <inttypes.h>
-
-#define TINYMT64_MEXP 127
-#define TINYMT64_SH0 12
-#define TINYMT64_SH1 11
-#define TINYMT64_SH8 8
-#define TINYMT64_MASK UINT64_C(0x7fffffffffffffff)
-#define TINYMT64_MUL (1.0 / 18446744073709551616.0)
-
-/*
- * tinymt64 internal state vector and parameters
- */
-struct TINYMT64_T {
-    uint64_t status[2];
-    uint32_t mat1;
-    uint32_t mat2;
-    uint64_t tmat;
-};
-
-typedef struct TINYMT64_T tinymt64_t;
-
-void tinymt64_init(tinymt64_t * random, uint64_t seed);
-void tinymt64_init_by_array(tinymt64_t * random, const uint64_t init_key[],
-			    int key_length);
-
-#if defined(__GNUC__)
-/**
- * This function always returns 127
- * @param random not used
- * @return always 127
- */
-inline static int tinymt64_get_mexp(
-    tinymt64_t * random  __attribute__((unused))) {
-    return TINYMT64_MEXP;
-}
-#else
-inline static int tinymt64_get_mexp(tinymt64_t * random) {
-    return TINYMT64_MEXP;
-}
-#endif
-
-/**
- * This function changes internal state of tinymt64.
- * Users should not call this function directly.
- * @param random tinymt internal status
- */
-inline static void tinymt64_next_state(tinymt64_t * random) {
-    uint64_t x;
-
-    random->status[0] &= TINYMT64_MASK;
-    x = random->status[0] ^ random->status[1];
-    x ^= x << TINYMT64_SH0;
-    x ^= x >> 32;
-    x ^= x << 32;
-    x ^= x << TINYMT64_SH1;
-    random->status[0] = random->status[1];
-    random->status[1] = x;
-    random->status[0] ^= -((int64_t)(x & 1)) & random->mat1;
-    random->status[1] ^= -((int64_t)(x & 1)) & (((uint64_t)random->mat2) << 32);
-}
-
-/**
- * This function outputs 64-bit unsigned integer from internal state.
- * Users should not call this function directly.
- * @param random tinymt internal status
- * @return 64-bit unsigned pseudorandom number
- */
-inline static uint64_t tinymt64_temper(tinymt64_t * random) {
-    uint64_t x;
-#if defined(LINEARITY_CHECK)
-    x = random->status[0] ^ random->status[1];
-#else
-    x = random->status[0] + random->status[1];
-#endif
-    x ^= random->status[0] >> TINYMT64_SH8;
-    x ^= -((int64_t)(x & 1)) & random->tmat;
-    return x;
-}
-
-/**
- * This function outputs floating point number from internal state.
- * Users should not call this function directly.
- * @param random tinymt internal status
- * @return floating point number r (1.0 <= r < 2.0)
- */
-inline static double tinymt64_temper_conv(tinymt64_t * random) {
-    uint64_t x;
-    union {
-	uint64_t u;
-	double d;
-    } conv;
-#if defined(LINEARITY_CHECK)
-    x = random->status[0] ^ random->status[1];
-#else
-    x = random->status[0] + random->status[1];
-#endif
-    x ^= random->status[0] >> TINYMT64_SH8;
-    conv.u = ((x ^ (-((int64_t)(x & 1)) & random->tmat)) >> 12)
-	| UINT64_C(0x3ff0000000000000);
-    return conv.d;
-}
-
-/**
- * This function outputs floating point number from internal state.
- * Users should not call this function directly.
- * @param random tinymt internal status
- * @return floating point number r (1.0 < r < 2.0)
- */
-inline static double tinymt64_temper_conv_open(tinymt64_t * random) {
-    uint64_t x;
-    union {
-	uint64_t u;
-	double d;
-    } conv;
-#if defined(LINEARITY_CHECK)
-    x = random->status[0] ^ random->status[1];
-#else
-    x = random->status[0] + random->status[1];
-#endif
-    x ^= random->status[0] >> TINYMT64_SH8;
-    conv.u = ((x ^ (-((int64_t)(x & 1)) & random->tmat)) >> 12)
-	| UINT64_C(0x3ff0000000000001);
-    return conv.d;
-}
-
-/**
- * This function outputs 64-bit unsigned integer from internal state.
- * @param random tinymt internal status
- * @return 64-bit unsigned integer r (0 <= r < 2^64)
- */
-inline static uint64_t tinymt64_generate_uint64(tinymt64_t * random) {
-    tinymt64_next_state(random);
-    return tinymt64_temper(random);
-}
-
-/**
- * This function outputs floating point number from internal state.
- * This function is implemented using multiplying by 1 / 2^64.
- * @param random tinymt internal status
- * @return floating point number r (0.0 <= r < 1.0)
- */
-inline static double tinymt64_generate_double(tinymt64_t * random) {
-    tinymt64_next_state(random);
-    return tinymt64_temper(random) * TINYMT64_MUL;
-}
-
-/**
- * This function outputs floating point number from internal state.
- * This function is implemented using union trick.
- * @param random tinymt internal status
- * @return floating point number r (0.0 <= r < 1.0)
- */
-inline static double tinymt64_generate_double01(tinymt64_t * random) {
-    tinymt64_next_state(random);
-    return tinymt64_temper_conv(random) - 1.0;
-}
-
-/**
- * This function outputs floating point number from internal state.
- * This function is implemented using union trick.
- * @param random tinymt internal status
- * @return floating point number r (1.0 <= r < 2.0)
- */
-inline static double tinymt64_generate_double12(tinymt64_t * random) {
-    tinymt64_next_state(random);
-    return tinymt64_temper_conv(random);
-}
-
-/**
- * This function outputs floating point number from internal state.
- * This function is implemented using union trick.
- * @param random tinymt internal status
- * @return floating point number r (0.0 < r <= 1.0)
- */
-inline static double tinymt64_generate_doubleOC(tinymt64_t * random) {
-    tinymt64_next_state(random);
-    return 2.0 - tinymt64_temper_conv(random);
-}
-
-/**
- * This function outputs floating point number from internal state.
- * This function is implemented using union trick.
- * @param random tinymt internal status
- * @return floating point number r (0.0 < r < 1.0)
- */
-inline static double tinymt64_generate_doubleOO(tinymt64_t * random) {
-    tinymt64_next_state(random);
-    return tinymt64_temper_conv_open(random) - 1.0;
-}
-
-#endif

src/units.c

@@ -62,10 +62,10 @@ static int scan_units(char *s, uint64_t *n, units *m) {
 
     if ((c = sscanf(s, "%"SCNu64"%2s", &base, unit)) < 1) return -1;
 
-    if (c == 2) {
+    if (c == 2 && strncasecmp(unit, m->base, 3)) {
         for (i = 0; m->units[i] != NULL; i++) {
             scale *= m->scale;
-            if (!strncasecmp(unit, m->units[i], sizeof(unit))) break;
+            if (!strncasecmp(unit, m->units[i], 3)) break;
         }
         if (m->units[i] == NULL) return -1;
     }
@@ -91,6 +91,14 @@ char *format_time_us(long double n) {
     return format_units(n, units, 2);
 }
 
+char *format_time_s(long double n) {
+    return format_units(n, &time_units_s, 0);
+}
+
 int scan_metric(char *s, uint64_t *n) {
     return scan_units(s, n, &metric_units);
 }
+
+int scan_time(char *s, uint64_t *n) {
+    return scan_units(s, n, &time_units_s);
+}

src/units.h

@@ -4,7 +4,9 @@
 char *format_binary(long double);
 char *format_metric(long double);
 char *format_time_us(long double);
+char *format_time_s(long double);
 
 int scan_metric(char *, uint64_t *);
+int scan_time(char *, uint64_t *);
 
 #endif /* UNITS_H */

src/wrk.c

@@ -1,151 +1,141 @@
 // 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 "script.h"
+#include "main.h"
 
 static struct config {
-    struct addrinfo addr;
-    uint64_t threads;
     uint64_t connections;
-    uint64_t requests;
+    uint64_t duration;
+    uint64_t threads;
     uint64_t timeout;
+    uint64_t pipeline;
+    bool     delay;
+    bool     dynamic;
+    bool     latency;
+    char    *host;
+    char    *script;
+    SSL_CTX *ctx;
 } cfg;
 
-static struct {
-    size_t size;
-    char *buf;
-} request;
-
 static struct {
     stats *latency;
     stats *requests;
-    pthread_mutex_t mutex;
 } statistics;
 
-static const struct http_parser_settings parser_settings = {
-    .on_message_complete = request_complete
+static struct sock sock = {
+    .connect  = sock_connect,
+    .close    = sock_close,
+    .read     = sock_read,
+    .write    = sock_write,
+    .readable = sock_readable
 };
 
+static struct http_parser_settings parser_settings = {
+    .on_message_complete = response_complete
+};
+
+static volatile sig_atomic_t stop = 0;
+
+static void handler(int sig) {
+    stop = 1;
+}
+
 static void usage() {
     printf("Usage: wrk <options> <url>                            \n"
            "  Options:                                            \n"
-           "    -c, --connections <n>  Connections to keep open   \n"
-           "    -r, --requests    <n>  Total requests to make     \n"
-           "    -t, --threads     <n>  Number of threads to use   \n"
+           "    -c, --connections <N>  Connections to keep open   \n"
+           "    -d, --duration    <T>  Duration of test           \n"
+           "    -t, --threads     <N>  Number of threads to use   \n"
            "                                                      \n"
-           "    -H, --header      <h>  Add header to request      \n"
+           "    -s, --script      <S>  Load Lua script file       \n"
+           "    -H, --header      <H>  Add header to request      \n"
+           "        --latency          Print latency statistics   \n"
+           "        --timeout     <T>  Socket/request timeout     \n"
            "    -v, --version          Print version details      \n"
            "                                                      \n"
-           "  Numeric arguments may include a SI unit (2k, 2M, 2G)\n");
+           "  Numeric arguments may include a SI unit (1k, 1M, 1G)\n"
+           "  Time arguments may include a time unit (2s, 2m, 2h)\n");
 }
 
 int main(int argc, char **argv) {
-    struct addrinfo *addrs, *addr;
-    struct http_parser_url parser_url;
-    char *url, **headers;
-    int rc;
+    char *url, **headers = zmalloc(argc * sizeof(char *));
+    struct http_parser_url parts = {};
 
-    headers = zmalloc((argc / 2) * sizeof(char *));
-
-    if (parse_args(&cfg, &url, headers, argc, argv)) {
+    if (parse_args(&cfg, &url, &parts, headers, argc, argv)) {
         usage();
         exit(1);
     }
 
-    if (http_parser_parse_url(url, strlen(url), 0, &parser_url)) {
-        fprintf(stderr, "invalid URL: %s\n", url);
-        exit(1);
-    }
+    char *schema  = copy_url_part(url, &parts, UF_SCHEMA);
+    char *host    = copy_url_part(url, &parts, UF_HOST);
+    char *port    = copy_url_part(url, &parts, UF_PORT);
+    char *service = port ? port : 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 : extract_url_part(url, &parser_url, UF_SCHEMA);
-    char *path = "/";
-
-    if (parser_url.field_set & (1 << UF_PATH)) {
-        path = &url[parser_url.field_data[UF_PATH].off];
-    }
-
-    struct addrinfo hints = {
-        .ai_family   = AF_UNSPEC,
-        .ai_socktype = SOCK_STREAM
-    };
-
-    if ((rc = getaddrinfo(host, service, &hints, &addrs)) != 0) {
-        const char *msg = gai_strerror(rc);
-        fprintf(stderr, "unable to resolve %s:%s %s\n", host, service, msg);
-        exit(1);
-    }
-
-    for (addr = addrs; addr != NULL; addr = addr->ai_next) {
-        int fd = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
-        if (fd == -1) continue;
-        if (connect(fd, addr->ai_addr, addr->ai_addrlen) == -1) {
-            if (errno == EHOSTUNREACH || errno == ECONNREFUSED) {
-                close(fd);
-                continue;
-            }
+    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);
         }
-        close(fd);
-        break;
+        sock.connect  = ssl_connect;
+        sock.close    = ssl_close;
+        sock.read     = ssl_read;
+        sock.write    = ssl_write;
+        sock.readable = ssl_readable;
     }
 
-    if (addr == NULL) {
+    signal(SIGPIPE, SIG_IGN);
+    signal(SIGINT,  SIG_IGN);
+
+    statistics.latency  = stats_alloc(cfg.timeout * 1000);
+    statistics.requests = stats_alloc(MAX_THREAD_RATE_S);
+    thread *threads     = zcalloc(cfg.threads * sizeof(thread));
+
+    lua_State *L = script_create(cfg.script, url, headers);
+    if (!script_resolve(L, host, service)) {
         char *msg = strerror(errno);
         fprintf(stderr, "unable to connect to %s:%s %s\n", host, service, msg);
         exit(1);
     }
 
-    signal(SIGPIPE, SIG_IGN);
-    cfg.addr     = *addr;
-    request.buf  = format_request(host, port, path, headers);
-    request.size = strlen(request.buf);
-
-    pthread_mutex_init(&statistics.mutex, NULL);
-    statistics.latency  = stats_alloc(SAMPLES);
-    statistics.requests = stats_alloc(SAMPLES);
-
-    thread *threads = zcalloc(cfg.threads * sizeof(thread));
-    uint64_t connections = cfg.connections / cfg.threads;
-    uint64_t requests    = cfg.requests    / cfg.threads;
+    cfg.host = host;
 
     for (uint64_t i = 0; i < cfg.threads; i++) {
-        thread *t = &threads[i];
-        t->connections = connections;
-        t->requests    = requests;
+        thread *t      = &threads[i];
+        t->loop        = aeCreateEventLoop(10 + cfg.connections * 3);
+        t->connections = cfg.connections / cfg.threads;
 
-        if (pthread_create(&t->thread, NULL, &thread_main, t)) {
+        t->L = script_create(cfg.script, url, headers);
+        script_init(L, t, argc - optind, &argv[optind]);
+
+        if (i == 0) {
+            cfg.pipeline = script_verify_request(t->L);
+            cfg.dynamic  = !script_is_static(t->L);
+            cfg.delay    = script_has_delay(t->L);
+            if (script_want_response(t->L)) {
+                parser_settings.on_header_field = header_field;
+                parser_settings.on_header_value = header_value;
+                parser_settings.on_body         = response_body;
+            }
+        }
+
+        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);
         }
     }
 
-    printf("Making %"PRIu64" requests to %s\n", cfg.requests, url);
+    struct sigaction sa = {
+        .sa_handler = handler,
+        .sa_flags   = 0,
+    };
+    sigfillset(&sa.sa_mask);
+    sigaction(SIGINT, &sa, NULL);
+
+    char *time = format_time_s(cfg.duration);
+    printf("Running %s test @ %s\n", time, url);
     printf("  %"PRIu64" threads and %"PRIu64" connections\n", cfg.threads, cfg.connections);
 
     uint64_t start    = time_us();
@@ -153,6 +143,9 @@ int main(int argc, char **argv) {
     uint64_t bytes    = 0;
     errors errors     = { 0 };
 
+    sleep(cfg.duration);
+    stop = 1;
+
     for (uint64_t i = 0; i < cfg.threads; i++) {
         thread *t = &threads[i];
         pthread_join(t->thread, NULL);
@@ -172,9 +165,15 @@ int main(int argc, char **argv) {
     long double req_per_s   = complete   / runtime_s;
     long double bytes_per_s = bytes      / runtime_s;
 
+    if (complete / cfg.connections > 0) {
+        int64_t interval = runtime_us / (complete / cfg.connections);
+        stats_correct(statistics.latency, interval);
+    }
+
     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);
 
@@ -191,27 +190,39 @@ int main(int argc, char **argv) {
     printf("Requests/sec: %9.2Lf\n", req_per_s);
     printf("Transfer/sec: %10sB\n", format_binary(bytes_per_s));
 
+    if (script_has_done(L)) {
+        script_summary(L, runtime_us, complete, bytes);
+        script_errors(L, &errors);
+        script_done(L, statistics.latency, statistics.requests);
+    }
+
     return 0;
 }
 
 void *thread_main(void *arg) {
     thread *thread = arg;
 
-    aeEventLoop *loop = aeCreateEventLoop(10 + cfg.connections * 3);
-    thread->cs   = zmalloc(thread->connections * sizeof(connection));
-    thread->loop = loop;
-    tinymt64_init(&thread->rand, time_us());
+    char *request = NULL;
+    size_t length = 0;
 
+    if (!cfg.dynamic) {
+        script_request(thread->L, &request, &length);
+    }
+
+    thread->cs = zcalloc(thread->connections * sizeof(connection));
     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;
+        c->request = request;
+        c->length  = length;
+        c->delayed = cfg.delay;
         connect_socket(thread, c);
     }
 
-    aeCreateTimeEvent(loop, SAMPLE_INTERVAL_MS, sample_rate, thread, NULL);
-    aeCreateTimeEvent(loop, TIMEOUT_INTERVAL_MS, check_timeouts, thread, NULL);
+    aeEventLoop *loop = thread->loop;
+    aeCreateTimeEvent(loop, RECORD_INTERVAL_MS, record_rate, thread, NULL);
 
     thread->start = time_us();
     aeMain(loop);
@@ -223,128 +234,211 @@ void *thread_main(void *arg) {
 }
 
 static int connect_socket(thread *thread, connection *c) {
-    struct addrinfo addr = cfg.addr;
+    struct addrinfo *addr = thread->addr;
     struct aeEventLoop *loop = thread->loop;
     int fd, flags;
 
-    fd = socket(addr.ai_family, addr.ai_socktype, addr.ai_protocol);
+    fd = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
 
     flags = fcntl(fd, F_GETFL, 0);
     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 (connect(fd, addr->ai_addr, addr->ai_addrlen) == -1) {
+        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 sample_rate(aeEventLoop *loop, long long id, void *data) {
+static int record_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;
+    if (thread->requests > 0) {
+        uint64_t elapsed_ms = (time_us() - thread->start) / 1000;
+        uint64_t requests = (thread->requests / (double) elapsed_ms) * 1000;
 
-    pthread_mutex_lock(&statistics.mutex);
-    stats_record(statistics.latency,  c->latency);
-    stats_record(statistics.requests, requests);
-    pthread_mutex_unlock(&statistics.mutex);
+        stats_record(statistics.requests, requests);
 
-    return SAMPLE_INTERVAL_MS + rand64(&thread->rand, SAMPLE_INTERVAL_MS);
+        thread->requests = 0;
+        thread->start    = time_us();
+    }
+
+    if (stop) aeStop(loop);
+
+    return RECORD_INTERVAL_MS;
 }
 
-static int request_complete(http_parser *parser) {
+static int delay_request(aeEventLoop *loop, long long id, void *data) {
+    connection *c = data;
+    c->delayed = false;
+    aeCreateFileEvent(loop, c->fd, AE_WRITABLE, socket_writeable, c);
+    return AE_NOMORE;
+}
+
+static int header_field(http_parser *parser, const char *at, size_t len) {
+    connection *c = parser->data;
+    if (c->state == VALUE) {
+        *c->headers.cursor++ = '\0';
+        c->state = FIELD;
+    }
+    buffer_append(&c->headers, at, len);
+    return 0;
+}
+
+static int header_value(http_parser *parser, const char *at, size_t len) {
+    connection *c = parser->data;
+    if (c->state == FIELD) {
+        *c->headers.cursor++ = '\0';
+        c->state = VALUE;
+    }
+    buffer_append(&c->headers, at, len);
+    return 0;
+}
+
+static int response_body(http_parser *parser, const char *at, size_t len) {
+    connection *c = parser->data;
+    buffer_append(&c->body, at, len);
+    return 0;
+}
+
+static int response_complete(http_parser *parser) {
     connection *c = parser->data;
     thread *thread = c->thread;
+    uint64_t now = time_us();
+    int status = parser->status_code;
 
-    if (parser->status_code > 399) {
+    thread->complete++;
+    thread->requests++;
+
+    if (status > 399) {
         thread->errors.status++;
     }
 
-    if (++thread->complete >= thread->requests) {
-        aeStop(thread->loop);
+    if (c->headers.buffer) {
+        *c->headers.cursor++ = '\0';
+        script_response(thread->L, status, &c->headers, &c->body);
+        c->state = FIELD;
+    }
+
+    if (--c->pending == 0) {
+        if (!stats_record(statistics.latency, now - c->start)) {
+            thread->errors.timeout++;
+        }
+        c->delayed = cfg.delay;
+        aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
+    }
+
+    if (!http_should_keep_alive(parser)) {
+        reconnect_socket(thread, c);
         goto done;
     }
 
-    c->latency = time_us() - c->start;
-    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;
-
-  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 maxAge = time_us() - (cfg.timeout * 1000);
-
-    for (uint64_t i = 0; i < thread->connections; i++, c++) {
-        if (maxAge > c->start) {
-            thread->errors.timeout++;
-        }
-    }
-
-    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, cfg.host)) {
+        case OK:    break;
+        case ERROR: goto error;
+        case RETRY: return;
+    }
+
+    http_parser_init(&c->parser, HTTP_RESPONSE);
+    c->written = 0;
+
+    aeCreateFileEvent(c->thread->loop, fd, AE_READABLE, socket_readable, c);
+    aeCreateFileEvent(c->thread->loop, fd, AE_WRITABLE, socket_writeable, c);
 
     return;
 
   error:
-    c->thread->errors.write++;
+    c->thread->errors.connect++;
     reconnect_socket(c->thread, c);
 }
 
+static void socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
+    connection *c = data;
+    thread *thread = c->thread;
+
+    if (c->delayed) {
+        uint64_t delay = script_delay(thread->L);
+        aeDeleteFileEvent(loop, fd, AE_WRITABLE);
+        aeCreateTimeEvent(loop, delay, delay_request, c, NULL);
+        return;
+    }
+
+    if (!c->written) {
+        if (cfg.dynamic) {
+            script_request(thread->L, &c->request, &c->length);
+        }
+        c->start   = time_us();
+        c->pending = cfg.pipeline;
+    }
+
+    char  *buf = c->request + c->written;
+    size_t len = c->length  - c->written;
+    size_t n;
+
+    switch (sock.write(c, buf, len, &n)) {
+        case OK:    break;
+        case ERROR: goto error;
+        case RETRY: return;
+    }
+
+    c->written += n;
+    if (c->written == c->length) {
+        c->written = 0;
+        aeDeleteFileEvent(loop, fd, AE_WRITABLE);
+    }
+
+    return;
+
+  error:
+    thread->errors.write++;
+    reconnect_socket(thread, c);
+}
+
 static void socket_readable(aeEventLoop *loop, int fd, void *data, int mask) {
     connection *c = data;
-    ssize_t n;
+    size_t n;
 
-    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;
+    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;
+        if (n == 0 && !http_body_is_final(&c->parser)) goto error;
+
+        c->thread->bytes += n;
+    } while (n == RECVBUF && sock.readable(c) > 0);
 
     return;
 
@@ -359,21 +453,12 @@ 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) {
+static char *copy_url_part(char *url, struct http_parser_url *parts, enum http_parser_url_fields field) {
     char *part = NULL;
 
-    if (parser_url->field_set & (1 << field)) {
-        uint16_t off = parser_url->field_data[field].off;
-        uint16_t len = parser_url->field_data[field].len;
+    if (parts->field_set & (1 << field)) {
+        uint16_t off = parts->field_data[field].off;
+        uint16_t len = parts->field_data[field].len;
         part = zcalloc(len + 1 * sizeof(char));
         memcpy(part, &url[off], len);
     }
@@ -381,42 +466,30 @@ static char *extract_url_part(char *url, struct http_parser_url *parser_url, enu
     return part;
 }
 
-static char *format_request(char *host, char *port, char *path, char **headers) {
-    char *req = NULL;
-
-    aprintf(&req, "GET %s HTTP/1.1\r\n", path);
-    aprintf(&req, "Host: %s", host);
-    if (port) aprintf(&req, ":%s", port);
-    aprintf(&req, "\r\n");
-
-    for (char **h = headers; *h != NULL; h++) {
-        aprintf(&req, "%s\r\n", *h);
-    }
-
-    aprintf(&req, "\r\n");
-    return req;
-}
-
 static struct option longopts[] = {
     { "connections", required_argument, NULL, 'c' },
-    { "requests",    required_argument, NULL, 'r' },
+    { "duration",    required_argument, NULL, 'd' },
     { "threads",     required_argument, NULL, 't' },
+    { "script",      required_argument, NULL, 's' },
     { "header",      required_argument, NULL, 'H' },
+    { "latency",     no_argument,       NULL, 'L' },
+    { "timeout",     required_argument, NULL, 'T' },
     { "help",        no_argument,       NULL, 'h' },
     { "version",     no_argument,       NULL, 'v' },
     { NULL,          0,                 NULL,  0  }
 };
 
-static int parse_args(struct config *cfg, char **url, char **headers, int argc, char **argv) {
-    char c, **header = headers;
+static int parse_args(struct config *cfg, char **url, struct http_parser_url *parts, char **headers, int argc, char **argv) {
+    char **header = headers;
+    int c;
 
     memset(cfg, 0, sizeof(struct config));
     cfg->threads     = 2;
     cfg->connections = 10;
-    cfg->requests    = 100;
+    cfg->duration    = 10;
     cfg->timeout     = SOCKET_TIMEOUT_MS;
 
-    while ((c = getopt_long(argc, argv, "t:c:r:H:v?", longopts, NULL)) != -1) {
+    while ((c = getopt_long(argc, argv, "t:c:d:s:H:T:Lrv?", longopts, NULL)) != -1) {
         switch (c) {
             case 't':
                 if (scan_metric(optarg, &cfg->threads)) return -1;
@@ -424,12 +497,22 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
             case 'c':
                 if (scan_metric(optarg, &cfg->connections)) return -1;
                 break;
-            case 'r':
-                if (scan_metric(optarg, &cfg->requests)) return -1;
+            case 'd':
+                if (scan_time(optarg, &cfg->duration)) return -1;
+                break;
+            case 's':
+                cfg->script = optarg;
                 break;
             case 'H':
                 *header++ = optarg;
                 break;
+            case 'L':
+                cfg->latency = true;
+                break;
+            case 'T':
+                if (scan_time(optarg, &cfg->timeout)) return -1;
+                cfg->timeout *= 1000;
+                break;
             case 'v':
                 printf("wrk %s [%s] ", VERSION, aeGetApiName());
                 printf("Copyright (C) 2012 Will Glozer\n");
@@ -442,7 +525,12 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
         }
     }
 
-    if (optind == argc || !cfg->threads || !cfg->requests) return -1;
+    if (optind == argc || !cfg->threads || !cfg->duration) return -1;
+
+    if (!script_parse_url(argv[optind], parts)) {
+        fprintf(stderr, "invalid URL: %s\n", argv[optind]);
+        return -1;
+    }
 
     if (!cfg->connections || cfg->connections < cfg->threads) {
         fprintf(stderr, "number of connections must be >= threads\n");
@@ -473,8 +561,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)) {
+    uint64_t max = stats->max;
     long double mean  = stats_mean(stats);
-    long double max   = stats_max(stats);
     long double stdev = stats_stdev(stats, mean);
 
     printf("    %-10s", name);
@@ -483,3 +571,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

@@ -5,71 +5,62 @@
 #include <pthread.h>
 #include <inttypes.h>
 #include <sys/types.h>
+#include <netdb.h>
+#include <sys/socket.h>
+
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+#include <luajit-2.0/lua.h>
 
 #include "stats.h"
 #include "ae.h"
 #include "http_parser.h"
-#include "tinymt64.h"
 
-#define VERSION  "1.1.0"
 #define RECVBUF  8192
-#define SAMPLES  100000
 
+#define MAX_THREAD_RATE_S   10000000
 #define SOCKET_TIMEOUT_MS   2000
-#define SAMPLE_INTERVAL_MS  100
-#define TIMEOUT_INTERVAL_MS 2000
+#define RECORD_INTERVAL_MS  100
 
-typedef struct {
-    uint32_t connect;
-    uint32_t read;
-    uint32_t write;
-    uint32_t status;
-    uint32_t timeout;
-} errors;
+extern const char *VERSION;
 
 typedef struct {
     pthread_t thread;
     aeEventLoop *loop;
+    struct addrinfo *addr;
     uint64_t connections;
-    uint64_t requests;
     uint64_t complete;
+    uint64_t requests;
     uint64_t bytes;
     uint64_t start;
-    tinymt64_t rand;
+    lua_State *L;
     errors errors;
     struct connection *cs;
 } thread;
 
+typedef struct {
+    char  *buffer;
+    size_t length;
+    char  *cursor;
+} buffer;
+
 typedef struct connection {
     thread *thread;
     http_parser parser;
+    enum {
+        FIELD, VALUE
+    } state;
     int fd;
+    SSL *ssl;
+    bool delayed;
     uint64_t start;
-    uint64_t latency;
+    char *request;
+    size_t length;
+    size_t written;
+    uint64_t pending;
+    buffer headers;
+    buffer body;
     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 */

src/wrk.lua

@@ -0,0 +1,74 @@
+local wrk = {
+   scheme  = "http",
+   host    = "localhost",
+   port    = nil,
+   method  = "GET",
+   path    = "/",
+   headers = {},
+   body    = nil,
+   thread  = nil,
+}
+
+function wrk.resolve(host, service)
+   local addrs = wrk.lookup(host, service)
+   for i = #addrs, 1, -1 do
+      if not wrk.connect(addrs[i]) then
+         table.remove(addrs, i)
+      end
+   end
+   wrk.addrs = addrs
+end
+
+function wrk.setup(thread)
+   thread.addr = wrk.addrs[1]
+   if type(setup) == "function" then
+      setup(thread)
+   end
+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
+
+   if type(init) == "function" then
+      init(args)
+   end
+
+   local req = wrk.format()
+   wrk.request = function()
+      return req
+   end
+end
+
+function wrk.format(method, path, headers, body)
+   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       = {}
+
+   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)
+   for name, value in pairs(headers) do
+      s[#s+1] = string.format("%s: %s", name, value)
+   end
+
+   s[#s+1] = ""
+   s[#s+1] = body or ""
+
+   return table.concat(s, "\r\n")
+end
+
+return wrk