// Copyright (C) 2012 - Will Glozer.  All rights reserved.
// -----------------------------------------------------------------------------
// Copyright (C) 2026 - G-WAN LLC.  All rights reserved.
// The purpose of our modifications is to fix several by-design wrk/wrk2 flaws
// documented here: http://gwan.com/blog/20260414.html
// Code changes are limited to the main.h, wrk.c and wrk.h files and signaled 
// by comments such as: "// GPG: xxx" (so just look for the string "// GPG:").
// Many more things would be worth enhancing, if only the wrk2 design was 
// not so flawed in the first place. At least, now, it is a bit less faulty
// and provides predictable, reproductible benchmarks that can be duplicated
// by other tools.
// -----------------------------------------------------------------------------
#define WRK_CALIBRATION 0 // GPG: set to 1 to abandon all hopes...

#include "wrk.h"
#include "script.h"
#include "main.h"
#include <hdr/hdr_histogram.h>
#include <limits.h>
#include "stats.h"

// -----------------------------------------------------------------------------
// GPG: add a crash handler to explin the great mysteries of the universe...
// -----------------------------------------------------------------------------
#include "libfault.h"
#include "libfault.c"

// Max recordable latency of 1 day
#define MAX_LATENCY 24L * 60 * 60 * 1e6

static struct config {
  uint64_t connections;
  uint64_t duration;
  uint64_t threads;
  uint64_t timeout;
  uint64_t pipeline;
  uint64_t rate;
  uint64_t delay_ms;
  bool     delay;
  bool     dynamic;
  bool     latency;
  bool     record_all_responses;
  bool     u_latency;
  bool     no_script_response_body;
  char    *host;
  char    *script;
  SSL_CTX *ctx;
} cfg;

static struct {
  stats *requests;
  pthread_mutex_t mutex;
} statistics;

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 s_stop = 0;

/*
static void handler(int sig) {
  char buf[32] = "==> signal: ",
       slen = log10(sig),
       blen = sizeof("==> signal: ")-1 + slen,
      *p = buf + blen;
  s_stop = 1;
  
	for(int i = slen; sig && i; --i, sig /= 10)
		p[i] = '0' + sig % 10;
  
  int r = write(STDOUT_FILENO, buf, blen);
  (void)r; // ditch gcc warning...
}
*/

// GPG: removed gazillions of pointless space characters...
static void usage() {
  puts("Usage: wrk3 <options> <url>\n"
  " Options:\n"
  " -c, --connections        <N>  Connections to keep open\n"
  " -d, --duration           <T>  Test duration to use\n"
  " -t, --threads            <N>  Number of threads to use\n"
  " -H, --header             <H>  Add header to HTTP requests\n"
  " -R, --rate               <T>  work rate (throughput) in requests/sec (total)\n"
  "\n"
  " -s, --script             <S>  Load Lua script file\n"
  " -N, --lua-dont-pass-body      Don't pass HTTP body to LUA script\n"
  "                               (faster for large response bodies)\n"
  "\n"
  " -L  --latency                 Print latency statistics\n"
  " -U  --u_latency               Print uncorrected latency statistics\n"
  "     --timeout            <T>  Socket/request timeout\n"
  " -B, --batch_latency           Measure latency of whole batches of pipelined ops (vs. each op)\n"
  "\n"
  " -v, --version                 Print version details\n"
  "\n"
  "  Numeric arguments may specify a quantity unit (1k, 1M, 1G)\n"
  "  Time arguments may specify a time unit (2s, 2m, 2h)\n"
  "\n"
  "Example: wrk3 -t1k -c1k -R10m 'http://127.0.0.1:8080/100.html'\n");
}

static void thread_reconnect_all(void*);

// =============================================================================
// GPG: add a crash handler to explain the great mysteries of the universe...
// -----------------------------------------------------------------------------
static void setup_crash_reports()
{ libfault_init(); /*
  libfault_set_app_name("wrk3");
  libfault_set_log_name("./wrk3_crash."); */
  libfault_install_handlers();
}

// =============================================================================
// GPG: main()
// -----------------------------------------------------------------------------
int main(int argc, char **argv)
{
  char *url, **headers = zmalloc(argc * sizeof(char*));
  struct http_parser_url parts = {};

  if(parse_args(&cfg, &url, &parts, headers, argc, argv))
  { usage();
    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;

  if(!strncmp("https", schema, 5)) 
  {
    if((cfg.ctx = ssl_init()) == NULL) 
    { 
      fprintf(stderr, "unable to initialize SSL\n");
      ERR_print_errors_fp(stderr);
      exit(1);
    }
    sock.connect  = ssl_connect;
    sock.close    = ssl_close;
    sock.read     = ssl_read;
    sock.write    = ssl_write;
    sock.readable = ssl_readable;
  }
  cfg.host = host;
  
/*// GPG: let's ditch this and use a proper crash handler instead:
  signal(SIGPIPE, SIG_IGN);
  signal(SIGINT,  SIG_IGN);
  struct sigaction sa = {
      .sa_handler = handler,
      .sa_flags   = 0,
  };
  sigfillset(&sa.sa_mask);
  sigaction(SIGINT, &sa, NULL); */
  setup_crash_reports(); // GPG: find WHY things go down...
  
  setlocale(LC_NUMERIC, ""); // GPG: get printf() pretty thousands...

  pthread_mutex_init(&statistics.mutex, NULL);
  statistics.requests = stats_alloc(10);
  thread *threads = zcalloc(cfg.threads * sizeof(thread));

  hdr_init(1, MAX_LATENCY, 3, &(statistics.requests->histogram));


  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);
  }
  
  uint64_t connections = cfg.connections / cfg.threads;
  double throughput    = (double)cfg.rate / cfg.threads;

  // -------------------------------------------------------------------------
  // GPG: event-loops creation
  // -------------------------------------------------------------------------
  uint64_t thread_init = time_us();
  for(uint64_t i = 0; i < cfg.threads; i++) {
      thread *t = &threads[i];
      t->loop        = aeCreateEventLoop(10 + cfg.connections * 3);
      t->connections = connections;
      t->throughput = throughput;
      t->reconnect_all = NULL;

      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;
          }
      }
  }
  // GPG: changed message about "event-loops" misrepresented as "threads":
  //     (this is a HIGHLY SUSPISCIOUS misrepresentation of the facts,
  //      probably designed to pretend thread creation is faster than
  //      what it is in reality...)
  //
//printf("Initialised %d threads in %d ms.\n",
  printf("Created %'ld event-loop(s) in %'.3f seconds\n",
         cfg.threads, (time_us() - thread_init) / 1e6);

  // GPG: moved BEFORE threads creation -it was done AFTER threads creation!
  struct hdr_histogram* latency_histogram;
  hdr_init(1, MAX_LATENCY, 3, &latency_histogram);
  struct hdr_histogram* u_latency_histogram;
  hdr_init(1, MAX_LATENCY, 3, &u_latency_histogram);

  // -------------------------------------------------------------------------
  // GPG: threads creation
  // -------------------------------------------------------------------------
  // GPG: wrk2 was setting up STOP_AT before START and the thread creation
  //     and calibration. This was most often leaving no benchmarking time
  //     after thread calibration, so we now do it in thread_main():
  // -------------------------------------------------------------------------
  { uint64_t stop_at = time_us() + (10 * 1e6); // lifespan to be updated
    thread_init = time_us(); // GPG: time threads creation (was not done)
    for(uint64_t i = 0; i < cfg.threads; i++) {
      thread *t = &threads[i];
      t->stop_at = stop_at; // GPG: save the beast from premature death
      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);
        exit(2);
      }
    }
  }
  // GPG: added the (missing) message reporting the threads creation time:
  printf("Created %'ld thread(s)     in %'.3f seconds\n",
         cfg.threads, (time_us() - thread_init) / 1e6);

  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);

  // -------------------------------------------------------------------------
  // GPG: wait for all threads to complete their work
  // -------------------------------------------------------------------------
  for(uint64_t i = 0; i < cfg.threads; i++) {
      thread *t = &threads[i];
      pthread_join(t->thread, NULL);
  }

  // -------------------------------------------------------------------------
  // GPG: collect and report statistics from all threads contexts
  // -------------------------------------------------------------------------
  // GPG: 
  // wrk2's main() setups a stop_at time BEFORE creating the threads and a 
  // start time AFTER creating and calibrating the threads, so the 
  // effective_benchmark_time = benchmark_duration - calibration_duration;
  //
  // The thread calibration time lasts 10 + (5*thread_connections) seconds:
  //
  //  calibrate_delay = CALIBRATE_DELAY_MS + (thread->connections * 5);
  //
  // This calibration time is then substracted from the user-specified or 
  // default benchmark time (of 10 seconds)... which often becomes < 1 sec, 
  // turning the RPS division into a multiplication (with bogus results):
  //
  //  req_per_s = complete / runtime_s; // a mult. when runtime_s is < 1
  //
  // The exact way to do this is to have the thread START and STOP times 
  // created by each thread - AFTER their calibration has been done:
  //
  //  start   = time_us();
  //  stop_at = start + (cfg.duration * 1e6);
  //  aeMain(loop); // GPG: => the actual benchmark starts here!
  //  thread->stop_at = time_us(); // GPG: save the REAL (not planned) thread exit
  // 
  // Calculating exact RPS cannot be done by a main() START/STOP timer: 
  // it does not know about the exact thread starting and execution time!
  //
  // So, what we do instead is taking the data collected by each thread
  // to calculate the RPS from main() - after the benchmark is completed
  // (all the collected data were there, I have not added anything).
  // -------------------------------------------------------------------------
  // GPG: all calculated from thread-collected data
  long double runtime_us = 0, req_per_s = 0, bytes_per_s = 0;
  long double complete = 0, bytes = 0;
  errors errors = { 0 };

  for(uint64_t i = 0; i < cfg.threads; i++) {
      thread *t = &threads[i];
      long double tm   = t->stop_at - t->start; // GPG: thread benchmark time in us
      long double tm_s = tm  / 1e6;             // GPG: thread benchmark time in sec

      // GPG: tracing the REAL threads benchmark times
    //printf("- thread #%lu benchmark time: %'.3Lf sec (%'.Lf usec)\n", i, tm_s, tm);

      runtime_us     += tm;          // GPG: total benchmarking time in us
      complete       += t->complete; // GPG: error-free completed requests
      
      // GPG: here we use the exact numbers collected by EACH THREAD
      //     (not the previous wrk2 vague guess made by main() about 
      //     the total test duration, which included the very long
      //     thread_calibration time and the benchmark time)
      req_per_s      += t->complete / tm_s;
      bytes_per_s    += t->bytes    / tm_s;
      
      bytes          += t->bytes;
      errors.connect += t->errors.connect;
      errors.read    += t->errors.read;
      errors.write   += t->errors.write;
      errors.timeout += t->errors.timeout;
      errors.status  += t->errors.status;

      hdr_add(latency_histogram,   t->latency_histogram);
      hdr_add(u_latency_histogram, t->u_latency_histogram);

      if(script_has_teardown(t->L)) {
          script_teardown(t->L);
      }
  }
  // GPG: divide the sum of all thread workers' benchmark time by threads
  //      to get a 'printable' execution time
  runtime_us /= cfg.threads;
  
  /*// GPG: with many threads, the cumulated threads execution times is 
  //      most often inferior to 1 second, so RPS = reqs / time becomes
  //      a MULTIPLICATION (the case we want to avoid).
  //
  long double runtime_s = runtime_us / 1e6; // GPG: thread benchmark time in sec

  // double ........ -1.797693e+308 to 1.797693e+308
  // long double ... -1.18973e+4932 to 1.18973e+4932
  puts("-----------------------");
  printf("Requests/sec: %'9.2Lf\n"
         "              %'.3Lf sec,\n"
         "              %'.Lf usec,\n"
         "   MAX_ULINT: %'lu\n"
         "         now: %'.Lf usec,\n"
         " 10s as usec: %'.Lf usec,\n"
         " now +10 sec: %'.Lf usec\n", 
        complete / runtime_s, runtime_s, runtime_us, 
        ULONG_MAX,
        (long double)thread_init, // now: time_us()
        (long double)(10 * 1e6), 
        (long double)(thread_init + 10 * 1e6));
  printf("Transfer/sec: %15sB\n", format_binary(bytes / runtime_s));
  puts("-----------------------");
*/
  stats *latency_stats     = stats_alloc(10);
  latency_stats->min       = hdr_min(latency_histogram);
  latency_stats->max       = hdr_max(latency_histogram);
  latency_stats->histogram = latency_histogram;

  print_stats_header();
  print_stats("Latency", latency_stats, format_time_us);
  print_stats("Req/Sec", statistics.requests, format_metric);

  // GPG: this is absolute crap because even-queues latencies are way larger
  //      than anything planned to be measured here - so, as a consequence,
  //      the wrk2 CLIENT is measuring ITS OWN LATENCIES (and injecting
  //      "Coordinated Ommission Correction", that is, fake latencies that
  //      are generated bu wrk2, when wrk2 thinks that it is blocked by a
  //      server that in reality is desperately waiting for client requests
  //      to be sent by the metally-disabled wrk2 benchmark tool).
  //      
  /*if(cfg.latency) print_stats_latency(latency_stats);

  if(cfg.latency) {
      print_hdr_latency(latency_histogram,
              "Recorded Latency");
      printf("----------------------------------------------------------\n");
  }

  if(cfg.u_latency) {
      printf("\n");
      print_hdr_latency(u_latency_histogram,
              "Uncorrected Latency (measured without taking delayed starts into account)");
      printf("----------------------------------------------------------\n");
  }*/

  char *runtime_msg = format_time_us(runtime_us);

  printf("  %'.Lf requests in %s, %sB read\n", // GPG: pretty thousands
          complete, runtime_msg, format_binary(bytes));
  if(errors.connect || errors.read || errors.write || errors.timeout) {
      printf("  Socket errors: connect %d, read %d, write %d, timeout %d\n",
             errors.connect, errors.read, errors.write, errors.timeout);
  }

  if(errors.status) {
      printf("  Non-2xx or 3xx responses: %d\n", errors.status);
  }

  // GPG: make RPS reporting more readable with pretty thousands:
  printf("Requests/sec: %'9.2Lf\n", req_per_s);
  printf("Transfer/sec: %sB\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, latency_stats, statistics.requests);
  }
  return 0;
}

void *thread_main(void *arg) {
  thread *thread = arg;
  aeEventLoop *loop = thread->loop;

  thread->cs = zcalloc(thread->connections * sizeof(connection));
//tinymt64_init(&thread->rand, time_us()); // GPG: unused, wasting time...
  hdr_init(1, MAX_LATENCY, 3, &thread->latency_histogram);
  hdr_init(1, MAX_LATENCY, 3, &thread->u_latency_histogram);

  char *request = NULL;
  size_t length = 0;

  if(!cfg.dynamic) {
      script_request(thread->L, &request, &length);
  }

  double throughput = (thread->throughput / 1e6) / thread->connections;

  connection *c = thread->cs;

  for(uint64_t i = 0; i < thread->connections; i++, c++) {
      c->fd         = -1;
      c->thread     = thread;
      c->ssl        = cfg.ctx ? SSL_new(cfg.ctx) : NULL;
      c->request    = request;
      c->length     = length;
      c->delayed = cfg.delay;
      c->throughput = throughput;
      c->catch_up_throughput = throughput * 2;
      c->complete   = 0;
      c->caught_up  = true;
      // Stagger connects 5 msec apart within thread:
      aeCreateTimeEvent(loop, i * 5, delayed_initial_connect, c, NULL);
  }
  thread->reconnect_all = &thread_reconnect_all;

  // GPG: the original code made nonsensical choices for high concurrencies:
//uint64_t calibrate_delay = CALIBRATE_DELAY_MS + (thread->connections * 5);
#if WRK_CALIBRATION
  uint64_t calibrate_delay = CALIBRATE_DELAY_MS; // GPG: less of a nuisance
#endif
  uint64_t timeout_delay   = TIMEOUT_INTERVAL_MS + (thread->connections * 5);

  // GPG: let's give away the "Crow Jewels" for the sake of operability:
  //aeCreateTimeEvent(loop, calibrate_delay, calibrate, thread, NULL);
  aeCreateTimeEvent(loop, timeout_delay, check_timeouts, thread, NULL);

  // GPG: make worker threads control their own start/stop time
  //     so that their actual execution time is matching the 
  //     default or command-line specified benchmark duration.
  //     (wrk2 did this in main(), by setting up stop_at before
  //     start and the thread creation/calibration, most often
  //     leaving no benchmarking time after thread calibration!)
  thread->start   = time_us();
  thread->stop_at = thread->start + (cfg.duration * 1e6);
/*
  printf("- thread #? PLANNED start: %'.3Lf sec, stop:%'.3Lf sec, duration:%'.3Lf sec, cfg.dur:%lu\n", 
     (long double)(thread->start / 1e6), 
     (long double)(thread->stop_at / 1e6), 
     (long double)((thread->stop_at - thread->start) / 1e6), cfg.duration); */

  aeMain(loop); // GPG: => the actual benchmark starts here!
  thread->stop_at = time_us(); // GPG: save the REAL (not planned) thread exit
/*
  printf("- thread #? ACTUAL STOP: %'.3Lf sec, thread lifespan: %'.3Lf sec\n", 
     (long double)(thread->stop_at / 1e6),
     (long double)((thread->stop_at - thread->start) / 1e6)); */

  aeDeleteEventLoop(loop);
  zfree(thread->cs);
  return NULL;
}

static void thread_reconnect_all(void *_t) {
  thread * t = (thread*)_t;
  connection *c = t->cs;
  for(uint64_t i = 0; i < t->connections; i++, c++) {
      if(c && 0 < c->fd) reconnect_socket(t, c);
  }
}

static int connect_socket(thread *thread, connection *c) {
  struct addrinfo *addr = thread->addr;
  struct aeEventLoop *loop = thread->loop;
  int fd, flags;

  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) goto error;
  }

  flags = 1;
  setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags));

  c->latest_connect = time_us();

  flags = AE_READABLE | AE_WRITABLE;
  if(aeCreateFileEvent(loop, fd, flags, socket_connected, c) == AE_OK) {
      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 delayed_initial_connect(aeEventLoop *loop, long long id, void *data) {
  connection* c = data;
  c->thread_start = time_us();
  connect_socket(c->thread, c);
  return AE_NOMORE;
}

#if WRK_CALIBRATION // GPG: thanks but no thanks...
static int calibrate(aeEventLoop *loop, long long id, void *data) {
  thread *thread = data;

  long double mean = hdr_mean(thread->latency_histogram);
  long double latency = hdr_value_at_percentile(
          thread->latency_histogram, 90.0) / 1000.0L;
  long double interval = MAX(latency * 2, 10);

  if(mean == 0) return CALIBRATE_DELAY_MS;

  thread->mean     = (uint64_t) mean;
  hdr_reset(thread->latency_histogram);
  hdr_reset(thread->u_latency_histogram);

  thread->start    = time_us();
  thread->interval = interval;
  thread->requests = 0;

  // GPG: gazillions of these, trashing CPU caches for no purpose...
  printf("  Thread calibration: mean lat.: %.3fms, rate sampling interval: %dms\n",
          (thread->mean)/1000.0,
          thread->interval);*/

  aeCreateTimeEvent(loop, thread->interval, sample_rate, thread, NULL);

  return AE_NOMORE;
}

static int sample_rate(aeEventLoop *loop, long long id, void *data) {
  thread *thread = data;

  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.requests, requests);
  pthread_mutex_unlock(&statistics.mutex);

  thread->requests = 0;
  thread->start    = time_us();

  return thread->interval;
}
#endif

static int check_timeouts(aeEventLoop *loop, long long id, void *data) {
  thread *thread = data;
  connection *c  = thread->cs;
  uint64_t now   = time_us();

  uint64_t maxAge = now - (cfg.timeout * 1000);

  for(uint64_t i = 0; i < thread->connections; i++, c++)
      if(maxAge > c->start)
          thread->errors.timeout++;

  if(s_stop || now >= thread->stop_at) {
     aeStop(loop);
     // GPG: tracing
     //printf("** timeout id: %lld ***\n", id);
  }

  return TIMEOUT_INTERVAL_MS;
}

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) {
  if(cfg.no_script_response_body)
      return 0;
  connection *c = parser->data;
  buffer_append(&c->body, at, len);
  return 0;
}

static uint64_t usec_to_next_send(connection *c) {
  uint64_t now = time_us();
  uint64_t next_start_time = c->thread_start + (c->complete / c->throughput);
  bool send_now = true;

  if(next_start_time > now) {
      // We are on pace. Indicate caught_up and don't send now.
      c->caught_up = true;
      send_now = false;
  } else {
      // We are behind
      if(c->caught_up) {
          // This is the first fall-behind since we were last caught up
          c->caught_up = false;
          c->catch_up_start_time = now;
          c->complete_at_catch_up_start = c->complete;
      }

      // Figure out if it's time to send, per catch up throughput:
      uint64_t complete_since_catch_up_start =
              c->complete - c->complete_at_catch_up_start;

      next_start_time = c->catch_up_start_time +
              (complete_since_catch_up_start / c->catch_up_throughput);

      if(next_start_time > now) {
          // Not yet time to send, even at catch-up throughout:
          send_now = false;
      }
  }

  if(send_now) {
      c->latest_should_send_time = now;
      c->latest_expected_start = next_start_time;
  }

  return send_now ? 0 : (next_start_time - now);
}

static int delay_request(aeEventLoop *loop, long long id, void *data) {
  connection* c = data;
  c->delayed = false;
  uint64_t time_usec_to_wait = usec_to_next_send(c);
  if(time_usec_to_wait) {
    return round((time_usec_to_wait / 1000.0L) + 0.5); /* don't send, wait */
  }
  aeCreateFileEvent(c->thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
  return AE_NOMORE;
}

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;

  thread->complete++;
  thread->requests++;

  if(status > 399) {
    thread->errors.status++;
  }

  if(c->headers.buffer) {
    *c->headers.cursor++ = '\0';
    script_response(thread->L, status, &c->headers, &c->body);
    c->state = FIELD;
  }

  if(now >= thread->stop_at) {
    aeStop(thread->loop);
    goto done;
  }

  // Count all responses (including pipelined ones:)
  c->complete++;

  // Note that expected start time is computed based on the completed
  // response count seen at the beginning of the last request batch sent.
  // A single request batch send may contain multiple requests, and
  // result in multiple responses. If we incorrectly calculated expect
  // start time based on the completion count of these individual pipelined
  // requests we can easily end up "gifting" them time and seeing
  // negative latencies.
  uint64_t expected_latency_start = c->thread_start +
          (c->complete_at_last_batch_start / c->throughput);

  int64_t expected_latency_timing = now - expected_latency_start;

  if(expected_latency_timing < 0) {
      printf("\n\n ---------- \n\n");
      printf("We are about to crash and die (recoridng a negative #)");
      printf("This wil never ever ever happen...");
      printf("But when it does. The following information will help in debugging");
      printf("response_complete:\n");
      printf("  expected_latency_timing = %ld\n", expected_latency_timing);
      printf("  now = %ld\n", now);
      printf("  expected_latency_start = %ld\n", expected_latency_start);
      printf("  c->thread_start = %ld\n", c->thread_start);
      printf("  c->complete = %ld\n", c->complete);
      printf("  throughput = %g\n", c->throughput);
      printf("  latest_should_send_time = %ld\n", c->latest_should_send_time);
      printf("  latest_expected_start = %ld\n", c->latest_expected_start);
      printf("  latest_connect = %ld\n", c->latest_connect);
      printf("  latest_write = %ld\n", c->latest_write);

      expected_latency_start = c->thread_start +
              ((c->complete ) / c->throughput);
      printf("  next expected_latency_start = %ld\n", expected_latency_start);
  }

  c->latest_should_send_time = 0;
  c->latest_expected_start = 0;

  if(--c->pending == 0) {
      c->has_pending = false;
      c->delayed = cfg.delay;
      aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);
  }

  // Record if needed, either last in batch or all, depending in cfg:
  if(cfg.record_all_responses || !c->has_pending) {
      hdr_record_value(thread->latency_histogram, expected_latency_timing);

      uint64_t actual_latency_timing = now - c->actual_latency_start;
      hdr_record_value(thread->u_latency_histogram, actual_latency_timing);
  }

  if(!http_should_keep_alive(parser)) {
      reconnect_socket(thread, c);
      goto done;
  }

  http_parser_init(parser, HTTP_RESPONSE);

done:
  return 0;
}

static void socket_connected(aeEventLoop *loop, int fd, void *data, int mask) {
  connection *c = data;

  switch (sock.connect(c, cfg.host)) {
      case OK:    break;
      case ERROR: goto error;
      case RETRY: return;
      default   : return; // GPG: ditch GCC warning (stupid enum{})...
  }

  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.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) {
      uint64_t time_usec_to_wait = usec_to_next_send(c);
      if(time_usec_to_wait) {
          int msec_to_wait = round((time_usec_to_wait / 1000.0L) + 0.5);

          // Not yet time to send. Delay:
          aeDeleteFileEvent(loop, fd, AE_WRITABLE);
          aeCreateTimeEvent(
                  thread->loop, msec_to_wait, delay_request, c, NULL);
          return;
      }
      c->latest_write = time_us();
  }

  if(!c->written && cfg.dynamic) {
      script_request(thread->L, &c->request, &c->length);
  }

  char  *buf = c->request + c->written;
  size_t len = c->length  - c->written;
  size_t n;

  if(!c->written) {
      c->start = time_us();
      if(!c->has_pending) {
          c->actual_latency_start = c->start;
          c->complete_at_last_batch_start = c->complete;
          c->has_pending = true;
      }
      c->pending = cfg.pipeline;
  }

  switch (sock.write(c, buf, len, &n)) {
      case OK:    break;
      case ERROR: goto error;
      case RETRY: return;
      default   : return; // GPG: ditch GCC warning (stupid enum{})...
  }

  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;
  size_t n;
  int read_status = OK;

  do {
      switch (read_status = sock.read(c, &n)) {
          case OK:    break;
          case ERROR: goto error;
          case RETRY: return;
          case READ_EOF: break;
      }

      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);

  if(read_status == READ_EOF) goto reconnect;
  return;

error:
  c->thread->errors.read++;
reconnect:
  reconnect_socket(c->thread, c);
}
// -----------------------------------------------------------------------------
static uint64_t time_us() {
  struct timeval t;
  gettimeofday(&t, NULL);
  return (t.tv_sec * 1e6) + t.tv_usec;
}
// -----------------------------------------------------------------------------
static char *copy_url_part(char *url, struct http_parser_url *parts, enum http_parser_url_fields field) {
  char *part = NULL;

  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);
  }
  return part;
}
// -----------------------------------------------------------------------------
static struct option longopts[] = {
  { "connections",    required_argument, NULL, 'c' },
  { "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' },
  { "u_latency",           no_argument,  NULL, 'U' },
  { "batch_latency",       no_argument,  NULL, 'B' },
  { "lua-dont-pass-body",  no_argument,  NULL, 'N' },
  { "timeout",        required_argument, NULL, 'T' },
  { "help",                no_argument,  NULL, 'h' },
  { "version",             no_argument,  NULL, 'v' },
  { "rate",           required_argument, NULL, 'R' },
  { NULL,             0,                 NULL,  0  }
};

static int 
parse_args(struct config *cfg, char **url, struct http_parser_url *parts, 
           char **headers, int argc, char **argv) 
{
  char c, **header = headers;

  memset(cfg, 0, sizeof(struct config));
  cfg->threads     = 2;
  cfg->connections = 10;
  cfg->duration    = 10;
  cfg->timeout     = SOCKET_TIMEOUT_MS;
  cfg->rate        = 0;
  cfg->record_all_responses = true;
  cfg->no_script_response_body = false;

  // GPG: getopt_long() has a weird behavior influenced with ebvironment
  //      variables and/or leading '++-' chars in the options string, but
  //      none of the possible documented ways seems to work on GLIbC for
  //      my RaspberryPi 500+ - so I have patched the "creative" variable
  //      GLibC behavior to get something that just works...
  while((c = getopt_long(argc, argv, "t:c:d:s:H:T:R:LUBNrv?", longopts, NULL)) != -1) 
  { switch(c) 
    { case 't': if(scan_metric(optarg, &cfg->threads))     return -1; break;
      case 'c': if(scan_metric(optarg, &cfg->connections)) return -1; break;
      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 'B': cfg->record_all_responses    = false;  break;
      case 'N': cfg->no_script_response_body = true;   break;
      case 'U': cfg->latency = true; cfg->u_latency = true; break;
      case 'T': if(scan_time(optarg, &cfg->timeout)) return -1; 
                cfg->timeout *= 1000; break; // GPG: tmo in ms
      case 'R': if(scan_metric(optarg, &cfg->rate))  return -1; break;
      case 'v': printf("wrk3 %s [%s] ", VERSION, aeGetApiName());
                printf("Copyright (C) 2012 Will Glozer, Copyright"
                       " (C) 2026 - G-WAN LLC.\n\n");
                break;
    //case 'h': case '?': case ':': // GPG: unused so commented...
      default: //return -1; // GPG: GLibC bug fix for the RaspberryPi 500+
               goto end_parsing; // GPG: GLibC bug fix for the RaspberryPi 500+
    }
  }
end_parsing: // GPG: GLibC bug fix for the RaspberryPi 500+

  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");
    return -1;
  }

  if(cfg->rate == 0) 
  {  fprintf(stderr,
     "Throughput MUST be specified with the --rate or -R option\n");
     return -1;
  }

  *url    = argv[optind];
  *header = NULL;
  return 0;
}

static void print_stats_header() {
    printf("  Thread Stats%6s%11s%8s%12s\n", "Avg", "Stdev", "Max", "+/- Stdev");
}

static void print_units(long double n, char *(*fmt)(long double), int width) {
  char *msg = fmt(n);
  int len = strlen(msg), pad = 2;

  if(isalpha(msg[len-1])) pad--;
  if(isalpha(msg[len-2])) pad--;
  width -= pad;

  printf("%*.*s%.*s", width, width, msg, pad, "  ");
  free(msg);
}

static void print_stats(char *name, stats *stats, char *(*fmt)(long double)) {
  uint64_t max = stats->max;
  long double mean  = stats_summarize(stats);
  long double stdev = stats_stdev(stats, mean);

  printf("    %-10s", name);
  print_units(mean,  fmt, 8);
  print_units(stdev, fmt, 10);
  print_units(max,   fmt, 9);
  printf("%8.2Lf%%\n", stats_within_stdev(stats, mean, stdev, 1));
}

/*
// GPG: this is absolute crap because even-queues latencies are way larger
//      than anything planned to be measured here - so, as a consequence,
//      the wrk2 CLIENT is measuring ITS OWN LATENCIES (and injecting
//      "Coordinated Ommission Correction", that is, fake latencies, when
//      it thinks that it is blocked by a server desperately waiting for
//      requests to be sent by the metally-disabled wrk2 benchmark tool).
//      
static void print_hdr_latency(struct hdr_histogram* histogram, const char* description) {
  long double percentiles[] = { 50.0, 75.0, 90.0, 99.0, 99.9, 99.99, 99.999, 100.0};
  printf("  Latency Distribution (HdrHistogram - %s)\n", description);
  for(size_t i = 0; i < sizeof(percentiles) / sizeof(long double); i++) {
      long double p = percentiles[i];
      int64_t n = hdr_value_at_percentile(histogram, p);
      printf("%7.3Lf%%", p);
      print_units(n, format_time_us, 10);
      printf("\n");
  }
  printf("\n%s\n", "  Detailed Percentile spectrum:");
  hdr_percentiles_print(histogram, stdout, 5, 1000.0, CLASSIC);
}

static void print_stats_latency(stats *stats) {
  long double percentiles[] = { 50.0, 75.0, 90.0, 99.0, 99.9, 99.99, 99.999, 100.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.3Lf%%", p);
      print_units(n, format_time_us, 10);
      printf("\n");
  }
}*/
// =============================================================================
// End of Source Code
// =============================================================================

