stop after receiving SIGINT

.gitignore

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

Makefile

@@ -1,6 +1,12 @@
-CFLAGS  := -std=c99 -Wall -O2 -pthread
-LDFLAGS := -pthread
-LIBS    := -lm
+CFLAGS  := -std=c99 -Wall -O2 -D_REENTRANT
+LIBS    := -lpthread -lm
+
+TARGET  := $(shell uname -s | tr [A-Z] [a-z] 2>/dev/null || echo unknown)
+
+ifeq ($(TARGET), sunos)
+	CFLAGS += -D_PTHREADS -D_POSIX_C_SOURCE=200112L
+	LIBS   += -lsocket
+endif
 
 SRC  := wrk.c aprintf.c stats.c units.c ae.c zmalloc.c http_parser.c tinymt64.c
 BIN  := wrk

NOTICE

@@ -14,7 +14,9 @@ This product includes software developed by Salvatore Sanfilippo and
 other contributors to the redis project.
 
 Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
 Copyright (C) 2009 Harish Mallipeddi - harish.mallipeddi@gmail.com
+Copyright (c) 2012, Joyent, Inc. All rights reserved.
 
 Copyright (c) 2006-2009, Salvatore Sanfilippo
 All rights reserved.

src/ae.c

@@ -35,7 +35,10 @@
 #include <sys/types.h>
 #include <unistd.h>
 #include <stdlib.h>
+#include <poll.h>
 #include <string.h>
+#include <time.h>
+#include <errno.h>
 
 #include "ae.h"
 #include "zmalloc.h"
@@ -43,13 +46,17 @@
 
 /* Include the best multiplexing layer supported by this system.
  * The following should be ordered by performances, descending. */
-#ifdef HAVE_EPOLL
-#include "ae_epoll.c"
+#ifdef HAVE_EVPORT
+#include "ae_evport.c"
 #else
-    #ifdef HAVE_KQUEUE
-    #include "ae_kqueue.c"
+    #ifdef HAVE_EPOLL
+    #include "ae_epoll.c"
     #else
-    #include "ae_select.c"
+        #ifdef HAVE_KQUEUE
+        #include "ae_kqueue.c"
+        #else
+        #include "ae_select.c"
+        #endif
     #endif
 #endif
 
@@ -62,6 +69,7 @@ aeEventLoop *aeCreateEventLoop(int setsize) {
     eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
     if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
     eventLoop->setsize = setsize;
+    eventLoop->lastTime = time(NULL);
     eventLoop->timeEventHead = NULL;
     eventLoop->timeEventNextId = 0;
     eventLoop->stop = 0;
@@ -97,7 +105,10 @@ void aeStop(aeEventLoop *eventLoop) {
 int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
         aeFileProc *proc, void *clientData)
 {
-    if (fd >= eventLoop->setsize) return AE_ERR;
+    if (fd >= eventLoop->setsize) {
+        errno = ERANGE;
+        return AE_ERR;
+    }
     aeFileEvent *fe = &eventLoop->events[fd];
 
     if (aeApiAddEvent(eventLoop, fd, mask) == -1)
@@ -231,6 +242,24 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
     int processed = 0;
     aeTimeEvent *te;
     long long maxId;
+    time_t now = time(NULL);
+
+    /* If the system clock is moved to the future, and then set back to the
+     * right value, time events may be delayed in a random way. Often this
+     * means that scheduled operations will not be performed soon enough.
+     *
+     * Here we try to detect system clock skews, and force all the time
+     * events to be processed ASAP when this happens: the idea is that
+     * processing events earlier is less dangerous than delaying them
+     * indefinitely, and practice suggests it is. */
+    if (now < eventLoop->lastTime) {
+        te = eventLoop->timeEventHead;
+        while(te) {
+            te->when_sec = 0;
+            te = te->next;
+        }
+    }
+    eventLoop->lastTime = now;
 
     te = eventLoop->timeEventHead;
     maxId = eventLoop->timeEventNextId-1;
@@ -369,21 +398,19 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
 /* Wait for millseconds until the given file descriptor becomes
  * writable/readable/exception */
 int aeWait(int fd, int mask, long long milliseconds) {
-    struct timeval tv;
-    fd_set rfds, wfds, efds;
+    struct pollfd pfd;
     int retmask = 0, retval;
 
-    tv.tv_sec = milliseconds/1000;
-    tv.tv_usec = (milliseconds%1000)*1000;
-    FD_ZERO(&rfds);
-    FD_ZERO(&wfds);
-    FD_ZERO(&efds);
+    memset(&pfd, 0, sizeof(pfd));
+    pfd.fd = fd;
+    if (mask & AE_READABLE) pfd.events |= POLLIN;
+    if (mask & AE_WRITABLE) pfd.events |= POLLOUT;
 
-    if (mask & AE_READABLE) FD_SET(fd,&rfds);
-    if (mask & AE_WRITABLE) FD_SET(fd,&wfds);
-    if ((retval = select(fd+1, &rfds, &wfds, &efds, &tv)) > 0) {
-        if (FD_ISSET(fd,&rfds)) retmask |= AE_READABLE;
-        if (FD_ISSET(fd,&wfds)) retmask |= AE_WRITABLE;
+    if ((retval = poll(&pfd, 1, milliseconds))== 1) {
+        if (pfd.revents & POLLIN) retmask |= AE_READABLE;
+        if (pfd.revents & POLLOUT) retmask |= AE_WRITABLE;
+	if (pfd.revents & POLLERR) retmask |= AE_WRITABLE;
+        if (pfd.revents & POLLHUP) retmask |= AE_WRITABLE;
         return retmask;
     } else {
         return retval;

src/ae.h

@@ -2,7 +2,7 @@
  * for the Jim's event-loop (Jim is a Tcl interpreter) but later translated
  * it in form of a library for easy reuse.
  *
- * Copyright (c) 2006-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * Copyright (c) 2006-2012, Salvatore Sanfilippo <antirez at gmail dot com>
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -88,6 +88,7 @@ typedef struct aeEventLoop {
     int maxfd;   /* highest file descriptor currently registered */
     int setsize; /* max number of file descriptors tracked */
     long long timeEventNextId;
+    time_t lastTime;     /* Used to detect system clock skew */
     aeFileEvent *events; /* Registered events */
     aeFiredEvent *fired; /* Fired events */
     aeTimeEvent *timeEventHead;

src/ae_epoll.c

@@ -1,6 +1,33 @@
 /* Linux epoll(2) based ae.c module
- * Copyright (C) 2009-2010 Salvatore Sanfilippo - antirez@gmail.com
- * Released under the BSD license. See the COPYING file for more info. */
+ *
+ * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * 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.
+ *   * Neither the name of Redis 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.
+ */
+
 
 #include <sys/epoll.h>
 
@@ -89,6 +116,8 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
 
             if (e->events & EPOLLIN) mask |= AE_READABLE;
             if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
+            if (e->events & EPOLLERR) mask |= AE_WRITABLE;
+            if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
             eventLoop->fired[j].fd = e->data.fd;
             eventLoop->fired[j].mask = mask;
         }

src/ae_evport.c

@@ -0,0 +1,315 @@
+/* ae.c module for illumos event ports.
+ *
+ * Copyright (c) 2012, Joyent, Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * 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.
+ *   * Neither the name of Redis 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.
+ */
+
+
+#include <assert.h>
+#include <errno.h>
+#include <port.h>
+#include <poll.h>
+
+#include <sys/types.h>
+#include <sys/time.h>
+
+#include <stdio.h>
+
+static int evport_debug = 0;
+
+/*
+ * This file implements the ae API using event ports, present on Solaris-based
+ * systems since Solaris 10.  Using the event port interface, we associate file
+ * descriptors with the port.  Each association also includes the set of poll(2)
+ * events that the consumer is interested in (e.g., POLLIN and POLLOUT).
+ *
+ * There's one tricky piece to this implementation: when we return events via
+ * aeApiPoll, the corresponding file descriptors become dissociated from the
+ * port.  This is necessary because poll events are level-triggered, so if the
+ * fd didn't become dissociated, it would immediately fire another event since
+ * the underlying state hasn't changed yet.  We must reassociate the file
+ * descriptor, but only after we know that our caller has actually read from it.
+ * The ae API does not tell us exactly when that happens, but we do know that
+ * it must happen by the time aeApiPoll is called again.  Our solution is to
+ * keep track of the last fds returned by aeApiPoll and reassociate them next
+ * time aeApiPoll is invoked.
+ *
+ * To summarize, in this module, each fd association is EITHER (a) represented
+ * only via the in-kernel assocation OR (b) represented by pending_fds and
+ * pending_masks.  (b) is only true for the last fds we returned from aeApiPoll,
+ * and only until we enter aeApiPoll again (at which point we restore the
+ * in-kernel association).
+ */
+#define MAX_EVENT_BATCHSZ 512
+
+typedef struct aeApiState {
+    int     portfd;                             /* event port */
+    int     npending;                           /* # of pending fds */
+    int     pending_fds[MAX_EVENT_BATCHSZ];     /* pending fds */
+    int     pending_masks[MAX_EVENT_BATCHSZ];   /* pending fds' masks */
+} aeApiState;
+
+static int aeApiCreate(aeEventLoop *eventLoop) {
+    int i;
+    aeApiState *state = zmalloc(sizeof(aeApiState));
+    if (!state) return -1;
+
+    state->portfd = port_create();
+    if (state->portfd == -1) {
+        zfree(state);
+        return -1;
+    }
+
+    state->npending = 0;
+
+    for (i = 0; i < MAX_EVENT_BATCHSZ; i++) {
+        state->pending_fds[i] = -1;
+        state->pending_masks[i] = AE_NONE;
+    }
+
+    eventLoop->apidata = state;
+    return 0;
+}
+
+static void aeApiFree(aeEventLoop *eventLoop) {
+    aeApiState *state = eventLoop->apidata;
+
+    close(state->portfd);
+    zfree(state);
+}
+
+static int aeApiLookupPending(aeApiState *state, int fd) {
+    int i;
+
+    for (i = 0; i < state->npending; i++) {
+        if (state->pending_fds[i] == fd)
+            return (i);
+    }
+
+    return (-1);
+}
+
+/*
+ * Helper function to invoke port_associate for the given fd and mask.
+ */
+static int aeApiAssociate(const char *where, int portfd, int fd, int mask) {
+    int events = 0;
+    int rv, err;
+
+    if (mask & AE_READABLE)
+        events |= POLLIN;
+    if (mask & AE_WRITABLE)
+        events |= POLLOUT;
+
+    if (evport_debug)
+        fprintf(stderr, "%s: port_associate(%d, 0x%x) = ", where, fd, events);
+
+    rv = port_associate(portfd, PORT_SOURCE_FD, fd, events,
+        (void *)(uintptr_t)mask);
+    err = errno;
+
+    if (evport_debug)
+        fprintf(stderr, "%d (%s)\n", rv, rv == 0 ? "no error" : strerror(err));
+
+    if (rv == -1) {
+        fprintf(stderr, "%s: port_associate: %s\n", where, strerror(err));
+
+        if (err == EAGAIN)
+            fprintf(stderr, "aeApiAssociate: event port limit exceeded.");
+    }
+
+    return rv;
+}
+
+static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
+    aeApiState *state = eventLoop->apidata;
+    int fullmask, pfd;
+
+    if (evport_debug)
+        fprintf(stderr, "aeApiAddEvent: fd %d mask 0x%x\n", fd, mask);
+
+    /*
+     * Since port_associate's "events" argument replaces any existing events, we
+     * must be sure to include whatever events are already associated when
+     * we call port_associate() again.
+     */
+    fullmask = mask | eventLoop->events[fd].mask;
+    pfd = aeApiLookupPending(state, fd);
+
+    if (pfd != -1) {
+        /*
+         * This fd was recently returned from aeApiPoll.  It should be safe to
+         * assume that the consumer has processed that poll event, but we play
+         * it safer by simply updating pending_mask.  The fd will be
+         * reassociated as usual when aeApiPoll is called again.
+         */
+        if (evport_debug)
+            fprintf(stderr, "aeApiAddEvent: adding to pending fd %d\n", fd);
+        state->pending_masks[pfd] |= fullmask;
+        return 0;
+    }
+
+    return (aeApiAssociate("aeApiAddEvent", state->portfd, fd, fullmask));
+}
+
+static void aeApiDelEvent(aeEventLoop *eventLoop, int fd, int mask) {
+    aeApiState *state = eventLoop->apidata;
+    int fullmask, pfd;
+
+    if (evport_debug)
+        fprintf(stderr, "del fd %d mask 0x%x\n", fd, mask);
+
+    pfd = aeApiLookupPending(state, fd);
+
+    if (pfd != -1) {
+        if (evport_debug)
+            fprintf(stderr, "deleting event from pending fd %d\n", fd);
+
+        /*
+         * This fd was just returned from aeApiPoll, so it's not currently
+         * associated with the port.  All we need to do is update
+         * pending_mask appropriately.
+         */
+        state->pending_masks[pfd] &= ~mask;
+
+        if (state->pending_masks[pfd] == AE_NONE)
+            state->pending_fds[pfd] = -1;
+
+        return;
+    }
+
+    /*
+     * The fd is currently associated with the port.  Like with the add case
+     * above, we must look at the full mask for the file descriptor before
+     * updating that association.  We don't have a good way of knowing what the
+     * events are without looking into the eventLoop state directly.  We rely on
+     * the fact that our caller has already updated the mask in the eventLoop.
+     */
+
+    fullmask = eventLoop->events[fd].mask;
+    if (fullmask == AE_NONE) {
+        /*
+         * We're removing *all* events, so use port_dissociate to remove the
+         * association completely.  Failure here indicates a bug.
+         */
+        if (evport_debug)
+            fprintf(stderr, "aeApiDelEvent: port_dissociate(%d)\n", fd);
+
+        if (port_dissociate(state->portfd, PORT_SOURCE_FD, fd) != 0) {
+            perror("aeApiDelEvent: port_dissociate");
+            abort(); /* will not return */
+        }
+    } else if (aeApiAssociate("aeApiDelEvent", state->portfd, fd,
+        fullmask) != 0) {
+        /*
+         * ENOMEM is a potentially transient condition, but the kernel won't
+         * generally return it unless things are really bad.  EAGAIN indicates
+         * we've reached an resource limit, for which it doesn't make sense to
+         * retry (counterintuitively).  All other errors indicate a bug.  In any
+         * of these cases, the best we can do is to abort.
+         */
+        abort(); /* will not return */
+    }
+}
+
+static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
+    aeApiState *state = eventLoop->apidata;
+    struct timespec timeout, *tsp;
+    int mask, i;
+    uint_t nevents;
+    port_event_t event[MAX_EVENT_BATCHSZ];
+
+    /*
+     * If we've returned fd events before, we must reassociate them with the
+     * port now, before calling port_get().  See the block comment at the top of
+     * this file for an explanation of why.
+     */
+    for (i = 0; i < state->npending; i++) {
+        if (state->pending_fds[i] == -1)
+            /* This fd has since been deleted. */
+            continue;
+
+        if (aeApiAssociate("aeApiPoll", state->portfd,
+            state->pending_fds[i], state->pending_masks[i]) != 0) {
+            /* See aeApiDelEvent for why this case is fatal. */
+            abort();
+        }
+
+        state->pending_masks[i] = AE_NONE;
+        state->pending_fds[i] = -1;
+    }
+
+    state->npending = 0;
+
+    if (tvp != NULL) {
+        timeout.tv_sec = tvp->tv_sec;
+        timeout.tv_nsec = tvp->tv_usec * 1000;
+        tsp = &timeout;
+    } else {
+        tsp = NULL;
+    }
+
+    /*
+     * port_getn can return with errno == ETIME having returned some events (!).
+     * So if we get ETIME, we check nevents, too.
+     */
+    nevents = 1;
+    if (port_getn(state->portfd, event, MAX_EVENT_BATCHSZ, &nevents,
+        tsp) == -1 && (errno != ETIME || nevents == 0)) {
+        if (errno == ETIME || errno == EINTR)
+            return 0;
+
+        /* Any other error indicates a bug. */
+        perror("aeApiPoll: port_get");
+        abort();
+    }
+
+    state->npending = nevents;
+
+    for (i = 0; i < nevents; i++) {
+            mask = 0;
+            if (event[i].portev_events & POLLIN)
+                mask |= AE_READABLE;
+            if (event[i].portev_events & POLLOUT)
+                mask |= AE_WRITABLE;
+
+            eventLoop->fired[i].fd = event[i].portev_object;
+            eventLoop->fired[i].mask = mask;
+
+            if (evport_debug)
+                fprintf(stderr, "aeApiPoll: fd %d mask 0x%x\n",
+                    (int)event[i].portev_object, mask);
+
+            state->pending_fds[i] = event[i].portev_object;
+            state->pending_masks[i] = (uintptr_t)event[i].portev_user;
+    }
+
+    return nevents;
+}
+
+static char *aeApiName(void) {
+    return "evport";
+}

src/ae_kqueue.c

@@ -1,6 +1,33 @@
 /* Kqueue(2)-based ae.c module
+ *
  * Copyright (C) 2009 Harish Mallipeddi - harish.mallipeddi@gmail.com
- * Released under the BSD license. See the COPYING file for more info. */
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * 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.
+ *   * Neither the name of Redis 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.
+ */
+
 
 #include <sys/types.h>
 #include <sys/event.h>

src/ae_select.c

@@ -1,6 +1,33 @@
-/* Select()-based ae.c module
- * Copyright (C) 2009-2010 Salvatore Sanfilippo - antirez@gmail.com
- * Released under the BSD license. See the COPYING file for more info. */
+/* Select()-based ae.c module.
+ *
+ * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * 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.
+ *   * Neither the name of Redis 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.
+ */
+
 
 #include <string.h>
 

src/aprintf.h

@@ -1,6 +1,6 @@
-#ifndef __APRINTF_H
-#define __APRINTF_H
+#ifndef APRINTF_H
+#define APRINTF_H
 
 char *aprintf(char **, const char *, ...);
 
-#endif /* __APRINTF_H */
+#endif /* APRINTF_H */

src/config.h

@@ -1,11 +1,13 @@
-#ifndef __CONFIG_H
-#define __CONFIG_H
+#ifndef CONFIG_H
+#define CONFIG_H
 
 #if defined(__FreeBSD__) || defined(__APPLE__)
 #define HAVE_KQUEUE
 #elif defined(__linux__)
 #define HAVE_EPOLL
 #define _POSIX_C_SOURCE 200809L
+#elif defined (__sun)
+#define HAVE_EVPORT
 #endif
 
-#endif /* __CONFIG_H */
+#endif /* CONFIG_H */

src/stats.c

@@ -7,7 +7,7 @@
 #include "stats.h"
 #include "zmalloc.h"
 
-stats *stats_alloc(int samples) {
+stats *stats_alloc(uint64_t samples) {
     stats *stats = zcalloc(sizeof(stats) + sizeof(uint64_t) * samples);
     stats->samples = samples;
     return stats;
@@ -25,7 +25,7 @@ void stats_record(stats *stats, uint64_t x) {
 
 uint64_t stats_min(stats *stats) {
     uint64_t min = 0;
-    for (int i = 0; i < stats->limit; i++) {
+    for (uint64_t i = 0; i < stats->limit; i++) {
         uint64_t x = stats->data[i];
         if (x < min || min == 0) min = x;
     }
@@ -34,7 +34,7 @@ uint64_t stats_min(stats *stats) {
 
 uint64_t stats_max(stats *stats) {
     uint64_t max = 0;
-    for (int i = 0; i < stats->limit; i++) {
+    for (uint64_t i = 0; i < stats->limit; i++) {
         uint64_t x = stats->data[i];
         if (x > max || max == 0) max = x;
     }
@@ -44,7 +44,7 @@ uint64_t stats_max(stats *stats) {
 long double stats_mean(stats *stats) {
     uint64_t sum = 0;
     if (stats->limit == 0) return 0.0;
-    for (int i = 0; i < stats->limit; i++) {
+    for (uint64_t i = 0; i < stats->limit; i++) {
         sum += stats->data[i];
     }
     return sum / (long double) stats->limit;
@@ -53,7 +53,7 @@ long double stats_mean(stats *stats) {
 long double stats_stdev(stats *stats, long double mean) {
     long double sum = 0.0;
     if (stats->limit < 2) return 0.0;
-    for (int i = 0; i < stats->limit; i++) {
+    for (uint64_t i = 0; i < stats->limit; i++) {
         sum += powl(stats->data[i] - mean, 2);
     }
     return sqrtl(sum / (stats->limit - 1));
@@ -64,7 +64,7 @@ long double stats_within_stdev(stats *stats, long double mean, long double stdev
     long double lower = mean - (stdev * n);
     uint64_t sum = 0;
 
-    for (int i = 0; i < stats->limit; i++) {
+    for (uint64_t i = 0; i < stats->limit; i++) {
         uint64_t x = stats->data[i];
         if (x >= lower && x <= upper) sum++;
     }

src/stats.h

@@ -1,14 +1,14 @@
-#ifndef __STATS_H
-#define __STATS_H
+#ifndef STATS_H
+#define STATS_H
 
 typedef struct {
-    int samples;
-    int index;
-    int limit;
+    uint64_t samples;
+    uint64_t index;
+    uint64_t limit;
     uint64_t data[];
 } stats;
 
-stats *stats_alloc(int);
+stats *stats_alloc(uint64_t);
 void stats_free(stats *);
 void stats_record(stats *, uint64_t);
 uint64_t stats_min(stats *);
@@ -17,5 +17,5 @@ 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);
 
-#endif /* __STATS_H */
+#endif /* STATS_H */
 

src/units.h

@@ -1,5 +1,5 @@
-#ifndef __UNITS_H
-#define __UNITS_H
+#ifndef UNITS_H
+#define UNITS_H
 
 char *format_binary(long double);
 char *format_metric(long double);
@@ -7,5 +7,4 @@ char *format_time_us(long double);
 
 int scan_metric(char *, uint64_t *);
 
-#endif /* __UNITS_H */
-
+#endif /* UNITS_H */

src/wrk.c

@@ -14,6 +14,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <signal.h>
 #include <time.h>
 #include <unistd.h>
 #include <sys/socket.h>
@@ -32,6 +33,7 @@ static struct config {
     uint64_t connections;
     uint64_t requests;
     uint64_t timeout;
+    uint64_t errors;
 } cfg;
 
 static struct {
@@ -49,6 +51,12 @@ static const struct http_parser_settings parser_settings = {
     .on_message_complete = request_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"
@@ -58,6 +66,7 @@ static void usage() {
            "                                                      \n"
            "    -H, --header      <h>  Add header to request      \n"
            "    -v, --version          Print version details      \n"
+           "        --errors      <n>  Abort after N errors       \n"
            "                                                      \n"
            "  Numeric arguments may include a SI unit (2k, 2M, 2G)\n");
 }
@@ -83,7 +92,11 @@ int main(int argc, char **argv) {
     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 = &url[parser_url.field_data[UF_PATH].off];
+    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,
@@ -99,14 +112,9 @@ int main(int argc, char **argv) {
     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;
-            }
-        }
+        rc = connect(fd, addr->ai_addr, addr->ai_addrlen);
         close(fd);
-        break;
+        if (rc == 0) break;
     }
 
     if (addr == NULL) {
@@ -115,6 +123,8 @@ int main(int argc, char **argv) {
         exit(1);
     }
 
+    signal(SIGPIPE, SIG_IGN);
+    signal(SIGINT,  SIG_IGN);
     cfg.addr     = *addr;
     request.buf  = format_request(host, port, path, headers);
     request.size = strlen(request.buf);
@@ -127,27 +137,34 @@ int main(int argc, char **argv) {
     uint64_t connections = cfg.connections / cfg.threads;
     uint64_t requests    = cfg.requests    / cfg.threads;
 
-    for (int i = 0; i < cfg.threads; i++) {
+    for (uint64_t i = 0; i < cfg.threads; i++) {
         thread *t = &threads[i];
         t->connections = connections;
         t->requests    = requests;
 
         if (pthread_create(&t->thread, NULL, &thread_main, t)) {
             char *msg = strerror(errno);
-            fprintf(stderr, "unable to create thread %d %s\n", i, msg);
+            fprintf(stderr, "unable to create thread %"PRIu64" %s\n", i, msg);
             exit(2);
         }
     }
 
+    struct sigaction sa = {
+        .sa_handler = handler,
+        .sa_flags   = 0,
+    };
+    sigfillset(&sa.sa_mask);
+    sigaction(SIGINT, &sa, NULL);
+
     printf("Making %"PRIu64" requests to %s\n", cfg.requests, url);
     printf("  %"PRIu64" threads and %"PRIu64" connections\n", cfg.threads, cfg.connections);
 
     uint64_t start    = time_us();
     uint64_t complete = 0;
     uint64_t bytes    = 0;
-    errors_t errors   = { 0 };
+    errors errors     = { 0 };
 
-    for (int i = 0; i < cfg.threads; i++) {
+    for (uint64_t i = 0; i < cfg.threads; i++) {
         thread *t = &threads[i];
         pthread_join(t->thread, NULL);
 
@@ -198,7 +215,7 @@ void *thread_main(void *arg) {
 
     connection *c = thread->cs;
 
-    for (int i = 0; i < thread->connections; i++, c++) {
+    for (uint64_t i = 0; i < thread->connections; i++, c++) {
         c->thread  = thread;
         c->latency = 0;
         connect_socket(thread, c);
@@ -308,12 +325,22 @@ static int check_timeouts(aeEventLoop *loop, long long id, void *data) {
 
     uint64_t maxAge = time_us() - (cfg.timeout * 1000);
 
-    for (int i = 0; i < thread->connections; i++, c++) {
+    for (uint64_t i = 0; i < thread->connections; i++, c++) {
         if (maxAge > c->start) {
             thread->errors.timeout++;
         }
     }
 
+    uint64_t errors = 0;
+    errors += thread->errors.connect;
+    errors += thread->errors.read;
+    errors += thread->errors.write;
+    errors += thread->errors.timeout;
+
+    if (stop || errors >= cfg.errors) {
+        aeStop(loop);
+    }
+
     return TIMEOUT_INTERVAL_MS;
 }
 
@@ -334,9 +361,9 @@ static void socket_writeable(aeEventLoop *loop, int fd, void *data, int mask) {
 
 static void socket_readable(aeEventLoop *loop, int fd, void *data, int mask) {
     connection *c = data;
-    int n;
+    ssize_t n;
 
-    if ((n = read(fd, c->buf, sizeof(c->buf))) == -1) goto error;
+    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;
 
@@ -376,18 +403,24 @@ static char *extract_url_part(char *url, struct http_parser_url *parser_url, enu
 }
 
 static char *format_request(char *host, char *port, char *path, char **headers) {
-    char *req = NULL;
-
-    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");
+    char *req  = NULL;
+    char *head = NULL;
 
     for (char **h = headers; *h != NULL; h++) {
-        aprintf(&req, "%s\r\n", *h);
+        aprintf(&head, "%s\r\n", *h);
+        if (!strncasecmp(*h, "Host:", 5)) {
+            host = NULL;
+            port = NULL;
+        }
     }
 
-    aprintf(&req, "\r\n");
+    aprintf(&req, "GET %s HTTP/1.1\r\n", path);
+    if (host) aprintf(&req, "Host: %s", host);
+    if (port) aprintf(&req, ":%s", port);
+    if (host) aprintf(&req, "\r\n");
+    aprintf(&req, "%s\r\n", head ? head : "");
+
+    free(head);
     return req;
 }
 
@@ -396,6 +429,7 @@ static struct option longopts[] = {
     { "requests",    required_argument, NULL, 'r' },
     { "threads",     required_argument, NULL, 't' },
     { "header",      required_argument, NULL, 'H' },
+    { "errors",      required_argument, NULL, 'E' },
     { "help",        no_argument,       NULL, 'h' },
     { "version",     no_argument,       NULL, 'v' },
     { NULL,          0,                 NULL,  0  }
@@ -410,7 +444,7 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
     cfg->requests    = 100;
     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:r:E:H:v?", longopts, NULL)) != -1) {
         switch (c) {
             case 't':
                 if (scan_metric(optarg, &cfg->threads)) return -1;
@@ -421,6 +455,9 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
             case 'r':
                 if (scan_metric(optarg, &cfg->requests)) return -1;
                 break;
+            case 'E':
+                if (scan_metric(optarg, &cfg->errors)) return -1;
+                break;
             case 'H':
                 *header++ = optarg;
                 break;
@@ -443,6 +480,10 @@ static int parse_args(struct config *cfg, char **url, char **headers, int argc,
         return -1;
     }
 
+    if (cfg->errors == 0) {
+        cfg->errors = cfg->requests / cfg->threads / 10;
+    }
+
     *url    = argv[optind];
     *header = NULL;
 

src/wrk.h

@@ -1,5 +1,5 @@
-#ifndef __WRK_H
-#define __WRK_H
+#ifndef WRK_H
+#define WRK_H
 
 #include "config.h"
 #include <pthread.h>
@@ -11,7 +11,7 @@
 #include "http_parser.h"
 #include "tinymt64.h"
 
-#define VERSION  "1.0.0"
+#define VERSION  "1.2.0"
 #define RECVBUF  8192
 #define SAMPLES  100000
 
@@ -25,7 +25,7 @@ typedef struct {
     uint32_t write;
     uint32_t status;
     uint32_t timeout;
-} errors_t;
+} errors;
 
 typedef struct {
     pthread_t thread;
@@ -36,11 +36,11 @@ typedef struct {
     uint64_t bytes;
     uint64_t start;
     tinymt64_t rand;
-    errors_t errors;
-    struct _connection *cs;
+    errors errors;
+    struct connection *cs;
 } thread;
 
-typedef struct _connection {
+typedef struct connection {
     thread *thread;
     http_parser parser;
     int fd;
@@ -72,4 +72,4 @@ 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 */
+#endif /* WRK_H */