Added timestamps to tcp server and client workers for latency calcula…

…tions

Makefile

@@ -1,8 +1,8 @@
 CC=g++
 CFLAGS=-Wall -Wunused -Wextra -g3 -std=c++0x -pedantic
 LDFLAGS=
-LIBS=-lev
-SOURCES=main.cc flamethrower.cc params.cc tcp_factory.cc tcp_worker.cc payload.cc http_worker.cc
+LIBS=-lev -lrt
+SOURCES=main.cc common.cc flamethrower.cc params.cc events.cc tcp_factory.cc tcp_worker.cc payload.cc http_worker.cc
 OBJECTS=$(SOURCES:.cc=.o)
 EXECUTABLE=flamethrower
 

common.cc

@@ -0,0 +1,2 @@
+#include "common.h"
+

common.h

@@ -23,6 +23,7 @@
 #include <ev.h>
 
 #include "params.h"
+#include "events.h"
 
 #define DEBUG 1
 

events.cc

@@ -0,0 +1,12 @@
+#include "events.h"
+
+#define NS_PER_S 1000000000
+
+uint64_t timestamp_ns_now() {
+    uint64_t nsec;
+    timespec tspec;
+    clock_gettime(CLOCK_REALTIME, &tspec);
+    nsec = tspec.tv_sec * NS_PER_S + tspec.tv_nsec;
+    return nsec;
+}
+

events.h

@@ -0,0 +1,26 @@
+#ifndef EVENTS_H
+#define EVENTS_H
+
+#include <ctime>
+#include <cstdint>
+
+uint64_t timestamp_ns_now();
+
+class Event {
+public:
+    Event() {}
+    virtual ~Event() {}
+};
+
+class TcpClientEvent : public Event {
+public:
+    uint64_t established_time;
+
+    TcpClientEvent() : Event() {}
+    virtual ~TcpClientEvent() {}
+};
+
+
+
+
+#endif

tcp_worker.cc

@@ -4,7 +4,10 @@
 TcpWorker::TcpWorker(TcpFactory &factory, TcpWorkerParams &params, int sock)
     : factory(factory),
       params(params),
-      sock(sock) {
+      sock(sock),
+      readable_time(0),
+      writable_time(0),
+      close_time(0) {
 
     debug_socket_print(sock, "ctor\n");
 
@@ -27,6 +30,10 @@ TcpWorker::~TcpWorker() {
         perror("socket close error");
     }
 
+    if(!close_time) {
+        close_time = timestamp_ns_now();
+    }
+
     factory.worker_delete_cb(*this);
 }
 
@@ -124,7 +131,10 @@ TcpWorker::SockAct TcpWorker::recv_buf(char *buf, size_t buflen, size_t &recvlen
     }
 
     recvlen = ret;
+
+    // Update factory stats
     factory.bytes_in += ret;
+
     return SockAct::CONTINUE;
 }
 
@@ -152,7 +162,10 @@ TcpWorker::SockAct TcpWorker::send_buf(char *buf, size_t buflen, size_t &sentlen
     }
 
     sentlen = ret;
+
+    // Update factory stats
     factory.bytes_out += ret;
+
     return SockAct::CONTINUE;
 }
 
@@ -169,8 +182,14 @@ void TcpWorker::read_cb(struct ev_loop *loop, struct ev_io *watcher, int revents
         perror("invalid event");
         return;
     }
+
     TcpWorker *worker = (TcpWorker*)watcher->data;
 
+    // Timestamp
+    if(!worker->readable_time) {
+        worker->readable_time = timestamp_ns_now();
+    }
+
     worker->read_cb();
 }
 
@@ -205,8 +224,14 @@ void TcpWorker::write_cb(struct ev_loop *loop, struct ev_io *watcher, int revent
         perror("invalid event");
         return;
     }
+
     TcpWorker *worker = (TcpWorker*)watcher->data;
 
+    // Timestamp
+    if(!worker->writable_time) {
+        worker->writable_time = timestamp_ns_now();
+    }
+
     worker->write_cb();
 }
 
@@ -299,7 +324,8 @@ TcpWorker::SockAct TcpWorker::write_payloads(PayloadList &payloads, size_t sendl
 ////////////////////////////////////////////////////////////////////////////////
 TcpServerWorker::TcpServerWorker(TcpServerFactory &factory, TcpServerWorkerParams &params, int sock)
     : TcpWorker(factory, params, sock),
-      params(params) {
+      params(params),
+      established_time(0) {
 
     debug_socket_print(sock, "ctor\n");
 
@@ -325,10 +351,26 @@ TcpServerWorker::TcpServerWorker(TcpServerFactory &factory, TcpServerWorkerParam
     ev_io_init(&sock_w_ev, write_cb, sock, EV_WRITE);
     ev_io_start(factory.loop, &sock_w_ev);
 
+    // Accept timestamp
+    established_time = timestamp_ns_now();
 }
 
 TcpServerWorker::~TcpServerWorker() {
     debug_socket_print(sock, "dtor\n");
+
+    if(!close_time) {
+        close_time = timestamp_ns_now();
+    }
+
+    debug_print("latency: "
+                "readable=%luns "
+                "writable=%luns "
+                "close=%luns "
+                "\n",
+                readable_time - established_time,
+                writable_time - established_time,
+                close_time - established_time);
+
 }
 
 void TcpServerWorker::finish() {
@@ -358,7 +400,9 @@ TcpServerWorker* TcpServerWorker::maker(TcpServerFactory &factory, TcpServerWork
 ////////////////////////////////////////////////////////////////////////////////
 TcpClientWorker::TcpClientWorker(TcpClientFactory &factory, TcpClientWorkerParams &params)
     : TcpWorker(factory, params),
-      params(params) {
+      params(params),
+      connect_time(0),
+      established_time(0) {
 
     debug_print("ctor\n");
 
@@ -427,12 +471,31 @@ TcpClientWorker::TcpClientWorker(TcpClientFactory &factory, TcpClientWorkerParam
     sock_timeout.data = this;
     ev_timer_init(&sock_timeout, timeout_cb, factory.params.connect_timeout, 0);
     ev_timer_start(factory.loop, &sock_timeout);
+
+    // Connect timestamp
+    connect_time = timestamp_ns_now();
 }
 
 
 TcpClientWorker::~TcpClientWorker() {
     debug_socket_print(sock, "dtor\n");
     ev_timer_stop(factory.loop, &sock_timeout);
+
+    if(!close_time) {
+        close_time = timestamp_ns_now();
+    }
+
+    debug_print("latency: "
+                "connect=%luns "
+                "readable=%luns "
+                "writable=%luns "
+                "close=%luns "
+                "\n",
+                established_time - connect_time,
+                readable_time - established_time,
+                writable_time - established_time,
+                close_time - established_time);
+
 }
 
 void TcpClientWorker::finish() {
@@ -465,7 +528,13 @@ void TcpClientWorker::connected_cb(struct ev_loop *loop, struct ev_io *watcher,
         perror("invalid event");
         return;
     }
+
     TcpClientWorker *worker = (TcpClientWorker*)watcher->data;
+
+    if(!worker->established_time) {
+        worker->established_time = timestamp_ns_now();
+    }
+
     worker->connected_cb();
 }
 
@@ -509,6 +578,7 @@ void TcpClientWorker::timeout_cb(struct ev_loop *loop, struct ev_timer *watcher,
         return;
     }
     TcpClientWorker *worker = (TcpClientWorker*)watcher->data;
+
     worker->timeout_cb();
 }
 

tcp_worker.h

@@ -40,6 +40,10 @@ class TcpWorker {
         ERROR
     };
 
+    uint64_t readable_time;
+    uint64_t writable_time;
+    uint64_t close_time;
+
     TcpWorker(TcpFactory &factory, TcpWorkerParams &params, int sock=-1);
     virtual ~TcpWorker();
 
@@ -75,6 +79,9 @@ class TcpServerWorker : public TcpWorker {
 private:
     TcpServerWorkerParams &params;
 public:
+
+    uint64_t established_time;
+
     TcpServerWorker(TcpServerFactory &factory, TcpServerWorkerParams &params, int sock);
     virtual ~TcpServerWorker();
 
@@ -92,6 +99,10 @@ class TcpClientWorker : public TcpWorker {
     TcpClientWorkerParams &params;
     struct ev_timer sock_timeout;
 public:
+
+    uint64_t connect_time;
+    uint64_t established_time;
+
     TcpClientWorker(TcpClientFactory &factory, TcpClientWorkerParams &params);
     virtual ~TcpClientWorker();