Reduce contention on monitoring.

Keep statistics per-thread, and sum them only when we need to
display them.

Author: gbrail

src/BUILD

@@ -20,7 +20,6 @@ cc_library(
         "apib_lines.cc",
         "apib_message.cc",
         "apib_rand.cc",
-        "apib_reporting.cc",
         "apib_time.cc",
         "apib_url.cc",
         "apib_util.cc",
@@ -62,6 +61,7 @@ cc_library(
         "apib_io_socket.cc",
         "apib_iothread.cc",
         "apib_oauth.cc",
+        "apib_reporting.cc",
     ],
     hdrs = [
         "apib_commandqueue.h",
@@ -104,9 +104,6 @@ cc_binary(
     srcs = [
         "apib_main.cc",
     ],
-    copts = [
-        "-std=gnu++14",
-    ],
     deps = [
         ":io",
         "//third_party:base64",

src/apib_cpu_proc.cc

@@ -171,16 +171,21 @@ static int getTicks(CPUUsage* cpu) {
         if (!tok.empty()) {
           int64_t c;
           if (!absl::SimpleAtoi(tok, &c)) {
-            return 0;
+            continue;
           }
-          if ((i == 3) || (i == 4) || (i == 7)) {
-            /* The fourth and fifth columns are "idle" and "iowait".
+          switch (i) {
+            case 3:
+            case 4:
+            case 7:
+              /* The fourth and fifth columns are "idle" and "iowait".
                  We consider both to be idle CPU.
                The eigth is "steal", which is time lost to virtualization
                as a client -- that's idle too in our estimation */
-            idleCount += c;
-          } else {
-            nonIdleCount += c;
+              idleCount += c;
+              break;
+            default:
+              nonIdleCount += c;
+              break;
           }
           i++;
         }

src/apib_io_basic.cc

@@ -65,6 +65,10 @@ int ConnectionState::Connect() {
 
   size_t addrLen;
   const struct sockaddr* addr = url_->address(t_->threadIndex(), &addrLen);
+  if (addr == nullptr) {
+    io_Verbose(this, "No addresses to look up\n");
+    return -2;
+  }
 
   if (t_->verbose) {
     char hostName[512];

src/apib_iothread.cc

@@ -18,6 +18,7 @@ limitations under the License.
 
 #include <cassert>
 #include <functional>
+#include <iostream>
 #include <thread>
 
 #include "ev.h"
@@ -108,11 +109,15 @@ void ConnectionState::writeRequest() {
 void ConnectionState::ConnectAndSend() {
   startTime_ = GetTime();
   if (needsOpen_) {
-    int err = Connect();
-    mandatoryAssert(err == 0);
-    // Should only fail if we can't create a new socket --
-    // errors actually connecting will be handled during write.
-    RecordConnectionOpen();
+    const int err = Connect();
+    if (err == 0) {
+      RecordConnectionOpen();
+    } else {
+      std::cerr << "Error opening TCP connection: " << err << std::endl;
+      RecordSocketError();
+      sendAfterDelay(kConnectFailureDelay);
+      return;
+    }
   }
   writeRequest();
   SendWrite();
@@ -127,22 +132,26 @@ void ConnectionState::thinkingDone(struct ev_loop* loop, ev_timer* t,
 
 void ConnectionState::addThinkTime() {
   const double thinkTime = (double)(t_->thinkTime) / 1000.0;
-  io_Verbose(this, "Thinking for %.4lf seconds\n", thinkTime);
-  ev_timer_init(&thinkTimer_, thinkingDone, thinkTime, 0);
+  sendAfterDelay(thinkTime);
+}
+
+void ConnectionState::sendAfterDelay(double seconds) {
+  io_Verbose(this, "Thinking for %.4lf seconds\n", seconds);
+  ev_timer_init(&thinkTimer_, thinkingDone, seconds, 0);
   thinkTimer_.data = this;
   ev_timer_start(t_->loop(), &thinkTimer_);
 }
 
-void ConnectionState::recycle(int closeConn) {
+void ConnectionState::recycle(bool closeConn) {
   if (closeConn || t_->noKeepAlive || !t_->shouldKeepRunning()) {
-    needsOpen_ = 1;
+    needsOpen_ = true;
     // Close is async, especially for TLS. So we will
     // reconnect later...
     Close();
     return;
   }
 
-  needsOpen_ = 0;
+  needsOpen_ = false;
   if (t_->thinkTime > 0) {
     addThinkTime();
   } else {
@@ -152,7 +161,7 @@ void ConnectionState::recycle(int closeConn) {
 
 int ConnectionState::StartConnect() {
   url_ = URLInfo::GetNext(t_->rand());
-  needsOpen_ = 1;
+  needsOpen_ = true;
   ConnectAndSend();
   return 0;
 }
@@ -174,7 +183,7 @@ void ConnectionState::WriteDone(int err) {
   if (err != 0) {
     RecordSocketError();
     io_Verbose(this, "Error on write: %i\n", err);
-    recycle(1);
+    recycle(true);
   } else {
     io_Verbose(this, "Write complete. Starting to read\n");
     // Prepare to read.
@@ -191,32 +200,45 @@ void ConnectionState::ReadDone(int err) {
   if (err != 0) {
     io_Verbose(this, "Error on read: %i\n", err);
     RecordSocketError();
-    recycle(1);
+    recycle(true);
     return;
   }
 
-  RecordResult(parser_.status_code);
-  t_->recordLatency(GetTime() - startTime_);
+  t_->recordResult(parser_.status_code, GetTime() - startTime_);
   if (!http_should_keep_alive(&(parser_))) {
     io_Verbose(this, "Server does not want keep-alive\n");
-    recycle(1);
+    recycle(true);
   } else {
     const URLInfo* oldUrl = url_;
     url_ = URLInfo::GetNext(t_->rand());
     if (!URLInfo::IsSameServer(*oldUrl, *url_, t_->index)) {
       io_Verbose(this, "Switching to a different server\n");
       writeDirty_ = true;
-      recycle(1);
+      recycle(true);
     } else {
       // URLs are static throughout the run, so we can just compare pointer here
       if (url_ != oldUrl) {
         writeDirty_ = true;
       }
-      recycle(0);
+      recycle(false);
     }
   }
 }
 
+void IOThread::recordResult(int statusCode, int_fast64_t latency) {
+  Counters* c = getCounters();
+  if ((statusCode >= 200) && (statusCode < 300)) {
+    c->successfulRequests++;
+  } else {
+    c->failedRequests++;
+  }
+  c->latencies.push_back(latency);
+}
+
+void IOThread::recordRead(size_t c) { getCounters()->bytesRead += c; }
+
+void IOThread::recordWrite(size_t c) { getCounters()->bytesWritten += c; }
+
 void IOThread::setNumConnections(size_t newVal) {
   iothread_Verbose(this, "Current connections = %zu. New connections = %zu\n",
                    connections_.size(), newVal);
@@ -277,10 +299,6 @@ void IOThread::processCommands(struct ev_loop* loop, ev_async* a, int revents) {
 
 void IOThread::threadLoop() {
   int ret = 0;
-  readCount_ = 0;
-  writeCount_ = 0;
-  readBytes_ = 0;
-  writeBytes_ = 0;
 
   iothread_Verbose(this, "Starting new event loop %i for %i connection\n",
                    index, numConnections);
@@ -311,8 +329,6 @@ void IOThread::threadLoop() {
   }
   ret = ev_run(loop_, 0);
   iothread_Verbose(this, "ev_run finished: %i\n", ret);
-  RecordByteCounts(writeBytes_, readBytes_);
-  RecordLatencies(latencies_);
 
 finish:
   iothread_Verbose(this, "Cleaning up event loop %i\n", index);
@@ -322,13 +338,26 @@ void IOThread::threadLoop() {
   ev_loop_destroy(loop_);
 }
 
+IOThread::IOThread() {
+  Counters* c = new Counters();
+  counterPtr_.store(reinterpret_cast<uintptr_t>(c));
+}
+
 IOThread::~IOThread() {
   if (sslCtx != nullptr) {
     SSL_CTX_free(sslCtx);
   }
   if (thread_ != nullptr) {
     delete thread_;
   }
+  Counters* c = reinterpret_cast<Counters*>(counterPtr_.load());
+  delete c;
+}
+
+Counters* IOThread::exchangeCounters() {
+  Counters* newCounters = new Counters();
+  return reinterpret_cast<Counters*>(
+      counterPtr_.exchange(reinterpret_cast<uintptr_t>(newCounters)));
 }
 
 void IOThread::initializeParser() {
@@ -383,14 +412,4 @@ void IOThread::SetNumConnections(int newConnections) {
   ev_async_send(loop_, &async_);
 }
 
-void IOThread::recordRead(size_t c) {
-  readCount_++;
-  readBytes_ += c;
-}
-
-void IOThread::recordWrite(size_t c) {
-  writeCount_++;
-  writeBytes_ += c;
-}
-
 }  // namespace apib

src/apib_iothread.h

@@ -19,6 +19,8 @@ limitations under the License.
 
 #include <openssl/ssl.h>
 
+#include <atomic>
+#include <memory>
 #include <sstream>
 #include <string>
 #include <thread>
@@ -38,6 +40,7 @@ namespace apib {
 // already set
 
 class ConnectionState;
+class Counters;
 
 // This structure represents a single thread that runs a benchmark
 // across multiple connections.
@@ -67,6 +70,7 @@ class IOThread {
   static constexpr int kConnectionSet = (1 << 4);
   static constexpr int kUserAgentSet = (1 << 5);
 
+  IOThread();
   ~IOThread();
 
   // Start the thread. It's up to the caller to initialize everything
@@ -99,29 +103,32 @@ class IOThread {
 
   void recordRead(size_t c);
   void recordWrite(size_t c);
-  void recordLatency(int64_t l) { latencies_.push_back(l); }
+  void recordResult(int statusCode, int64_t latency);
+
+  // Swap the current set of performance counters and start new ones.
+  // The caller must free the result.
+  Counters* exchangeCounters();
 
  private:
   void threadLoop();
   static void initializeParser();
   static void processCommands(struct ev_loop* loop, ev_async* a, int revents);
   static void hardShutdown(struct ev_loop* loop, ev_timer* timer, int revents);
   void setNumConnections(size_t newVal);
+  Counters* getCounters() {
+    return reinterpret_cast<Counters*>(counterPtr_.load());
+  }
 
   static http_parser_settings parserSettings_;
 
-  long readCount_;
-  long writeCount_;
-  long long readBytes_;
-  long long writeBytes_;
   std::vector<ConnectionState*> connections_;
-  std::thread* thread_;
+  std::thread* thread_ = nullptr;
   RandomGenerator rand_;
-  struct ev_loop* loop_;
+  struct ev_loop* loop_ = nullptr;
   ev_async async_;
   CommandQueue commands_;
   ev_timer shutdownTimer_;
-  std::vector<int64_t> latencies_;
+  std::atomic_uintptr_t counterPtr_;
 };
 
 typedef enum {
@@ -166,11 +173,13 @@ class ConnectionState {
   static int httpComplete(http_parser* p);
 
  private:
-  static const int readBufSize = 512;
-  static const int writeBufSize = 1024;
+  static constexpr int readBufSize = 512;
+  static constexpr int writeBufSize = 1024;
+  static constexpr double kConnectFailureDelay = 0.25;
 
   void addThinkTime();
-  void recycle(int closeConn);
+  void sendAfterDelay(double seconds);
+  void recycle(bool closeConn);
   void writeRequest();
   void printSslError(const std::string& msg, int err) const;
 
@@ -207,6 +216,9 @@ class ConnectionState {
   long long startTime_ = 0LL;
 };
 
+// A typedef used to clean up some messy interfaces
+typedef std::vector<std::unique_ptr<IOThread>> ThreadList;
+
 // Debugging macro
 #define io_Verbose(c, ...) \
   if ((c)->t_->verbose) {  \