Adding -r and -C options to h2load

src/h2load.cc

@@ -75,7 +75,8 @@ Config::Config()
     : data_length(-1), addrs(nullptr), nreqs(1), nclients(1), nthreads(1),
       max_concurrent_streams(-1), window_bits(30), connection_window_bits(30),
       no_tls_proto(PROTO_HTTP2), data_fd(-1), port(0), default_port(0),
-      verbose(false) {}
+      verbose(false), 
+      nconns(0), rate(0), current_worker(0) {}
 
 Config::~Config() {
   freeaddrinfo(addrs);
@@ -85,6 +86,10 @@ Config::~Config() {
   }
 }
 
+bool Config::is_rate_mode() {    
+  return (this->rate != 0);    
+}
+
 Config config;
 
 namespace {
@@ -261,7 +266,7 @@ void Client::process_abandoned_streams() {
 }
 
 void Client::report_progress() {
-  if (worker->id == 0 &&
+  if (!worker->config->is_rate_mode() && worker->id == 0 &&
       worker->stats.req_done % worker->progress_interval == 0) {
     std::cout << "progress: "
               << worker->stats.req_done * 100 / worker->stats.req_todo
@@ -962,6 +967,33 @@ std::vector<std::string> read_uri_from_file(std::istream &infile) {
 }
 } // namespace
 
+namespace {
+// Called every second when rate mode is being used
+void second_timeout_cb(EV_P_ ev_timer *w, int revents) {
+  auto config = static_cast<Config *>(w->data);
+
+  auto nclients_per_worker = config->rate;
+  auto nreqs_per_worker = config->max_concurrent_streams * config->rate;
+
+  if(config->current_worker >= std::max(0. , (config->seconds - 1.))) {
+    nclients_per_worker = config->rate + config->conns_remainder;
+    nreqs_per_worker = (int)config->max_concurrent_streams *
+                       (config->rate + config->conns_remainder);
+    ev_timer_stop(config->rate_loop, w);
+  }
+
+  config->workers.push_back(make_unique<Worker>(config->current_worker,
+                                                config->ssl_ctx,
+                                                nreqs_per_worker,
+                                                nclients_per_worker,
+                                                config));
+
+  config->current_worker++;
+
+  config->workers.back()->run();
+}
+} // namespace
+
 namespace {
 void print_version(std::ostream &out) {
   out << "h2load nghttp2/" NGHTTP2_VERSION << std::endl;
@@ -1027,6 +1059,7 @@ Options:
   -p, --no-tls-proto=<PROTOID>
               Specify ALPN identifier of the  protocol to be used when
               accessing http URI without SSL/TLS.)";
+
 #ifdef HAVE_SPDYLAY
   out << R"(
               Available protocols: spdy/2, spdy/3, spdy/3.1 and )";
@@ -1039,6 +1072,24 @@ Options:
   -d, --data=<FILE>
               Post FILE to  server.  The request method  is changed to
               POST.
+  -r, --rate=<N>
+              Specified the fixed rate at which connections are
+              created. The rate must be a positive integer,
+              representing the number of connections to be made per
+              second. When the rate is 0, the program will run as it
+              normally does, creating connections at whatever variable
+              rate it wants. The default value for this option is 0.
+  -C, --num-conns=<N>
+              Specifies the total number of connections to create. The
+              total number of connections must be a positive integer.
+              On each connection, '-m' requests are made. The test
+              stops once as soon as the N connections have either
+              completed or failed. When the number of connections is
+              0, the program will run as it normally does, creating as
+              many connections as it needs in order to make the '-n'
+              requests specified. The defauly value for this option is
+              0. The '-n' option is not required if the '-C' option
+              is being used.
   -v, --verbose
               Output debug information.
   --version   Display version information and exit.
@@ -1073,10 +1124,12 @@ int main(int argc, char **argv) {
         {"help", no_argument, nullptr, 'h'},
         {"version", no_argument, &flag, 1},
         {"ciphers", required_argument, &flag, 2},
+        {"rate", required_argument, nullptr, 'r'},
+        {"num-conns", required_argument, nullptr, 'C'},
         {nullptr, 0, nullptr, 0}};
     int option_index = 0;
-    auto c = getopt_long(argc, argv, "hvW:c:d:m:n:p:t:w:H:i:", long_options,
-                         &option_index);
+    auto c = getopt_long(argc, argv, "hvW:c:d:m:n:p:t:w:H:i:r:C:", 
+                         long_options, &option_index);
     if (c == -1) {
       break;
     }
@@ -1170,6 +1223,24 @@ int main(int argc, char **argv) {
         exit(EXIT_FAILURE);
       }
       break;
+    case 'r':
+      config.rate = strtoul(optarg, nullptr, 10);
+      if (config.rate <= 0) {
+        std::cerr << "-r: the rate at which connections are made "
+                  << "must be positive."
+                  << std::endl;
+        exit(EXIT_FAILURE);
+      }
+      break;
+    case 'C':
+      config.nconns = strtoul(optarg, nullptr, 10);
+      if (config.nconns <= 0) {
+        std::cerr << "-C: the total number of connections made "
+                  << "must be positive."
+                  << std::endl;
+        exit(EXIT_FAILURE);
+      }
+      break;
     case 'v':
       config.verbose = true;
       break;
@@ -1238,6 +1309,25 @@ int main(int argc, char **argv) {
               << "cores." << std::endl;
   }
 
+  if (config.nconns < 0) {
+    std::cerr << "-C: the total number of connections made "
+              << "cannot be negative."
+              << std::endl;
+    exit(EXIT_FAILURE);
+  }
+
+  if (config.rate < 0) {
+    std::cerr << "-r: the rate at which connections are made "
+              << "cannot be negative."
+              << std::endl;
+    exit(EXIT_FAILURE);
+  }
+
+  if (config.rate != 0 && config.nthreads != 1) {
+    std::cerr << "-r, -t: warning: the -t option will be ignored when the -r "
+              << "option is in use." << std::endl;
+  }
+
   if (!datafile.empty()) {
     config.data_fd = open(datafile.c_str(), O_RDONLY | O_BINARY);
     if (config.data_fd == -1) {
@@ -1405,46 +1495,92 @@ int main(int argc, char **argv) {
 
   auto start = std::chrono::steady_clock::now();
 
-  std::vector<std::unique_ptr<Worker>> workers;
-  workers.reserve(config.nthreads);
-
+  // if not in rate mode, continue making workers and clients normally
+  if (!config.is_rate_mode()) {
+    config.workers.reserve(config.nthreads);
 #ifndef NOTHREADS
-  std::vector<std::future<void>> futures;
-  for (size_t i = 0; i < config.nthreads - 1; ++i) {
-    auto nreqs = nreqs_per_thread + (nreqs_rem-- > 0);
-    auto nclients = nclients_per_thread + (nclients_rem-- > 0);
-    std::cout << "spawning thread #" << i << ": " << nclients
-              << " concurrent clients, " << nreqs << " total requests"
-              << std::endl;
-    workers.push_back(
-        make_unique<Worker>(i, ssl_ctx, nreqs, nclients, &config));
-    auto &worker = workers.back();
-    futures.push_back(
-        std::async(std::launch::async, [&worker]() { worker->run(); }));
-  }
+    std::vector<std::future<void>> futures;
+    for (size_t i = 0; i < config.nthreads - 1; ++i) {
+      auto nreqs = nreqs_per_thread + (nreqs_rem-- > 0);
+      auto nclients = nclients_per_thread + (nclients_rem-- > 0);
+      std::cout << "spawning thread #" << i << ": " << nclients
+                << " concurrent clients, " << nreqs << " total requests"
+                << std::endl;
+      config.workers.push_back(
+          make_unique<Worker>(i, ssl_ctx, nreqs, nclients, &config));
+      auto &worker = config.workers.back();
+      futures.push_back(
+          std::async(std::launch::async, [&worker]() { worker->run(); }));
+    }
 #endif // NOTHREADS
 
-  auto nreqs_last = nreqs_per_thread + (nreqs_rem-- > 0);
-  auto nclients_last = nclients_per_thread + (nclients_rem-- > 0);
-  std::cout << "spawning thread #" << (config.nthreads - 1) << ": "
-            << nclients_last << " concurrent clients, " << nreqs_last
-            << " total requests" << std::endl;
-  workers.push_back(make_unique<Worker>(config.nthreads - 1, ssl_ctx,
-                                        nreqs_last, nclients_last, &config));
-  workers.back()->run();
+    auto nreqs_last = nreqs_per_thread + (nreqs_rem-- > 0);
+    auto nclients_last = nclients_per_thread + (nclients_rem-- > 0);
+    std::cout << "spawning thread #" << (config.nthreads - 1) << ": "
+              << nclients_last << " concurrent clients, " << nreqs_last
+              << " total requests" << std::endl;
+    config.workers.push_back(make_unique<Worker>(config.nthreads - 1, ssl_ctx,
+                                          nreqs_last, nclients_last, &config));
+    config.workers.back()->run();
 
 #ifndef NOTHREADS
-  for (auto &fut : futures) {
-    fut.get();
-  }
+    for (auto &fut : futures) {
+      fut.get();
+    }
 #endif // NOTHREADS
+  } //!config.is_rate_mode()
+  // if in rate mode, create a new worker each second
+  else {
+    // set various config values
+    config.seconds = std::min(n_time, c_time);
+
+    if ((int)config.nreqs < config.nconns) {
+      config.seconds = c_time;
+      config.workers.reserve(config.seconds);
+    }
+    else if (config.nconns == 0) {
+      config.seconds = n_time;
+    }
+    else {
+      config.workers.reserve(config.seconds);
+    }
+
+    config.conns_remainder = config.nconns % config.rate;
+
+    // config.seconds must be positive or else an exception is thrown
+    if (config.seconds <= 0) {
+      std::cerr << "Test cannot be run with current option values."
+                << " Please look at documentation for -r option for"
+                << " more information." << std::endl;
+      exit(EXIT_FAILURE);
+    }
+    config.current_worker = 0;
+
+    config.ssl_ctx = ssl_ctx;
+
+    // create timer that will go off every second
+    ev_timer timeout_watcher;
+
+    // create loop for running the timer
+    struct ev_loop *rate_loop = EV_DEFAULT;
+
+    config.rate_loop = rate_loop;
+
+    // giving the second_timeout_cb access to config
+    timeout_watcher.data = &config;
+
+    ev_init(&timeout_watcher, second_timeout_cb);
+    timeout_watcher.repeat = 1.;
+    ev_timer_again(rate_loop, &timeout_watcher);
+    ev_run(rate_loop, 0);
+  } // end rate mode section
 
   auto end = std::chrono::steady_clock::now();
   auto duration =
       std::chrono::duration_cast<std::chrono::microseconds>(end - start);
 
   Stats stats(0);
-  for (const auto &w : workers) {
+  for (const auto &w : config.workers) {
     const auto &s = w->stats;
 
     stats.req_todo += s.req_todo;
@@ -1463,7 +1599,7 @@ int main(int argc, char **argv) {
     }
   }
 
-  auto ts = process_time_stats(workers);
+  auto ts = process_time_stats(config.workers);
 
   // Requests which have not been issued due to connection errors, are
   // counted towards req_failed and req_error.
@@ -1476,7 +1612,7 @@ int main(int argc, char **argv) {
   //
   // [1] https://github.com/lighttpd/weighttp
   // [2] https://github.com/wg/wrk
-  size_t rps = 0;
+  double rps = 0;
   int64_t bps = 0;
   if (duration.count() > 0) {
     auto secd = static_cast<double>(duration.count()) / (1000 * 1000);

src/h2load.h

@@ -76,6 +76,10 @@ struct Config {
   ssize_t max_concurrent_streams;
   size_t window_bits;
   size_t connection_window_bits;
+  // rate at which connections should be made
+  ssize_t rate;
+  // number of connections made
+  ssize_t nconns;
   enum { PROTO_HTTP2, PROTO_SPDY2, PROTO_SPDY3, PROTO_SPDY3_1 } no_tls_proto;
   // file descriptor for upload data
   int data_fd;
@@ -83,8 +87,17 @@ struct Config {
   uint16_t default_port;
   bool verbose;
 
+  ssize_t current_worker;
+  std::vector<std::unique_ptr<Worker>> workers;
+  SSL_CTX *ssl_ctx;
+  struct ev_loop *rate_loop;
+  ssize_t seconds;
+  ssize_t conns_remainder;
+
   Config();
   ~Config();
+
+  bool is_rate_mode();
 };
 
 struct RequestStat {