Various bug fixes

include/protocol.h

@@ -15,12 +15,13 @@ using namespace std;
 
 typedef struct {
     uint8_t protoVersion;
-    time_t clientStartTime; // Holds the creation time for the client session - used by the server to differentiate between runs
+    uint32_t clientStartTime; // Holds the creation time for the client session - used by the server to differentiate between runs
     char guid[MAX_GUID + 1];
     seqnum_t seqNum; // This should be some flavor of uint so that the math works out when it wraps.
-    struct timespec sent;
+    uint32_t sent_sec;
+    uint64_t sent_nsec;
     uint32_t size;
-} packet;
+} __attribute__((aligned(4))) packet;
 
 typedef struct {
     packet header;

include/serversession.h

@@ -22,7 +22,7 @@ class ServerSession {
     public:
         ServerSession (StatsWriter* writer, string peer, int port, packet* p);
         ~ServerSession ();
-        void writeStats();
+        void writeStats(time_t now);
         void recordSeq (seqnum_t seqNum);
         void setSuccessor (string successor) {this->successor = successor;}
         string getSuccessor () {return successor;}

include/statswriter.h

@@ -5,6 +5,7 @@
 #include "protocol.h"
 #include <string>
 #include <set>
+#include <memory>
 
 #ifndef STATSWRITER__H
 #define STATSWRITER__H 1
@@ -20,12 +21,13 @@ class StatsWriter {
         sockaddr* tags_sa;
         char* tags_metric;
         int quiet;
-        void writeConsoleStats (string guid, string peer, int port, Stats* stats);
+
+        void writeConsoleStats (string guid, string peer, int port, Stats* stats, time_t now);
         void writeStatsdStats (string guid, string peer, int port, Stats* stats);
         void writeTagsStats (string guid, string peer, int port, Stats* stats);
     public:
         StatsWriter (string listen_hostname, string receive_hostname, int listen_port, char* statsdInfo, char* statsdTagInfo, char* statsdTagMetric, int quiet);
-        void writeStats (string guid, string peer, int port, Stats* stats);
+        void writeStats (string guid, string peer, int port, Stats* stats, time_t now);
 };
 
 class StatsWriterSet {
@@ -36,7 +38,7 @@ class StatsWriterSet {
         int port;
     public:
         StatsWriterSet (StatsWriter* writer, string guid, string peer, int port);
-        void writeStats (Stats* stats);
+        void writeStats (Stats* stats, time_t now);
 };
 
 #endif

src/Makefile

@@ -1,16 +1,17 @@
-default: clean client server
+default: client server
 
-CXX = g++
+CXX = /usr/bin/c++
 COMMONOBJ = options.o delay.o protocol.o stats.o 
-CPPFLAGS = -I../include -g -std=c++0x
-LDFLAGS =
-LIBS = -lrt -lm
+CPPFLAGS = -I../include -g 
+LDFLAGS = -L../lib
+CLIBS = 
+SLIBS = 
 
 client: $(COMMONOBJ) client.o clientsession.o maclist.o ifinfo.o client.o options.o delay.o protocol.o
-	$(CXX) $(LDFLAGS) $(LIBS) -o udping_client client.o options.o delay.o protocol.o clientsession.o maclist.o ifinfo.o
+	$(CXX) $(LDFLAGS) $(CLIBS) -o udping_client client.o options.o delay.o protocol.o clientsession.o maclist.o ifinfo.o
 
 server: $(COMMONOBJ) server.o serversession.o statswriter.o delay.o options.o protocol.o stats.o
-	$(CXX) $(LDFLAGS) $(LIBS) -o udping_server server.o delay.o options.o protocol.o serversession.o stats.o statswriter.o
+	$(CXX) $(LDFLAGS) $(CLIBS) -o udping_server server.o delay.o options.o protocol.o serversession.o stats.o statswriter.o
 
 clean:
 	rm -f udping_client udping_server *.o

src/clientsession.cc

@@ -178,7 +178,11 @@ void ClientSession::increment() {
  * the ping header, sets the timestamp, and sends the ping
  */
 void ClientSession::sendPingPacket () {
-    clock_gettime(CLOCK_REALTIME, &(p.header.sent));
+    struct timespec ts;
+    //clock_gettime(CLOCK_REALTIME, &(p.header.sent));
+    clock_gettime(CLOCK_REALTIME, &ts);
+    p.header.sent_sec = ts.tv_sec;
+    p.header.sent_nsec = ts.tv_nsec;
     int size = rand() % this->parent->getMaxPacketSize();
     sendPacket(size);
 }

src/ifinfo.cc

@@ -90,4 +90,5 @@ int getIfInfo (string ip, int* ifIndex, uint8_t *srcMac) {
         return (-1);
     }
     printf ("Index is %i\n", *ifIndex);
+    return 0;
 }

src/protocol.cc

@@ -29,7 +29,7 @@ void dumpBuffer (char* buf) {
         return;
     }
     packet* ph = (packet*) buf;
-    printf ("%s:%ld:%d - %ld:%ld\n", ph->guid, ph->clientStartTime, ph->seqNum, ph->sent.tv_sec, ph->sent.tv_nsec);
+    printf ("%s:%d:%d - %d:%ld\n", ph->guid, ph->clientStartTime, ph->seqNum, ph->sent_sec, ph->sent_nsec);
 }
 
 int makeSocket (string host, int port) {
@@ -70,7 +70,8 @@ struct sockaddr* getSockAddr (string host, int port) {
 
     snprintf (portString, 100, "%d", port);
 
-    if (s = getaddrinfo (host.c_str(), portString, &hints, &addrinfo)) {
+    //if (s = getaddrinfo (host.c_str(), portString, &hints, &addrinfo)) {
+    if (s = getaddrinfo (host.c_str(), 0, &hints, &addrinfo)) {
         err(1, "getaddrinfo: %s\n", gai_strerror(s));
     } else {
         ret = new struct sockaddr;
@@ -181,7 +182,7 @@ int buildFrame (struct frame *etherFrame, uint8_t *srcMac, uint8_t *dstMac, uint
 uint16_t checksum (uint16_t *addr, int len)
 {
   int count = len;
-  register uint32_t sum = 0;
+  uint32_t sum = 0;
   uint16_t answer = 0;
 
   // Sum up 2-byte values until none or only one byte left.
@@ -211,7 +212,7 @@ uint16_t checksum (uint16_t *addr, int len)
 // This is a zero-copy rewrite of udp4 checksumming algorithm.
 uint16_t udp4_checksum2 (struct ip iphdr, struct udphdr udphdr, uint16_t *payload, int payloadlen)
 {
-    register uint32_t sum = 0;
+    uint32_t sum = 0;
 
     sum += (iphdr.ip_src.s_addr & 0xffff);
     sum += (iphdr.ip_src.s_addr >> 16);

src/serversession.cc

@@ -32,10 +32,10 @@ ServerSession::~ServerSession () {
     delete statsWriters;
 }
 
-void ServerSession::writeStats() {
+void ServerSession::writeStats(time_t now) {
     seqnum_t targetCount = 1 + maxSeq - minSeq;
     stats->setTargetCount(targetCount);
-    statsWriters->writeStats(stats);
+    statsWriters->writeStats(stats, now);
 }
 
 // This needs work to deal with out of order packets
@@ -76,7 +76,7 @@ ServerSession* ServerSessionManager::getServerSession (string peer, int port, pa
 void ServerSessionManager::sweepServerSessions () {
     int ix;
     time_t now = time(0);
-    for (map<string, ServerSession*>::iterator it = sessionMap.begin(); it != sessionMap.end(); it++) {
+    for (map<string, ServerSession*>::iterator it = sessionMap.begin(); it != sessionMap.end();) {
         ServerSession* ds = it->second;
         if (ds) {
             if (ds->getLastArrival() < now - keepalive) {
@@ -88,9 +88,11 @@ void ServerSessionManager::sweepServerSessions () {
                         ds->recordSeq(successor->getMinSeq() - 1);
                     }
                 }
-                ds->writeStats();
+                ds->writeStats(now);
                 delete ds;
-                sessionMap.erase(it->first);
+                it = sessionMap.erase(it);
+            } else {
+                it++;
             }
         } else {
             break;
@@ -103,8 +105,14 @@ void ServerSessionManager::receivePing (packet* ph, struct timespec* rcvd, strin
     ServerSession* ds = getServerSession(peer, port, ph);
     ds->setLastArrival(time(0));
     ds->recordSeq(ph->seqNum);
-    double elapsed = 1000000000L * (rcvd->tv_sec - ph->sent.tv_sec) + rcvd->tv_nsec - ph->sent.tv_nsec;
-    ds->getStats()->addDataPoint (elapsed/1000000);
+    double elapsed = rcvd->tv_nsec;
+    elapsed -= ph->sent_nsec;
+    elapsed /= 1000000000L;
+    elapsed += rcvd->tv_sec;
+    elapsed -= ph->sent_sec;
+    elapsed *= 1000;
+    //double elapsed = rcvd->tv_sec - ph->sent_sec + (rcvd->tv_nsec - ph->sent_nsec)/1000000000L;
+    ds->getStats()->addDataPoint (elapsed);
     sweepServerSessions();
 }
 
@@ -132,7 +140,7 @@ int ServerSessionManager::readNextPacket (int fd) {
             if (remoteName) {
                 hostName = hostMap[remote.sin_addr.s_addr] = remoteName->h_name;
             } else {
-                hostName = hostMap[remote.sin_addr.s_addr] =  inet_ntoa(remote.sin_addr);
+                hostName = hostMap[remote.sin_addr.s_addr] = inet_ntoa(remote.sin_addr);
             }
         }
         receivePing (ph, &tv, hostName, ntohs(remote.sin_port));

src/statswriter.cc

@@ -20,8 +20,8 @@ StatsWriterSet::StatsWriterSet (StatsWriter* writer, string guid, string peer, i
     this->port = port;
 }
 
-void StatsWriterSet::writeStats (Stats* stats) {
-    this->writer->writeStats(this->guid, this->peer, this->port, stats);
+void StatsWriterSet::writeStats (Stats* stats, time_t now) {
+    this->writer->writeStats(this->guid, this->peer, this->port, stats, now);
 }
 
 StatsWriter::StatsWriter (string listen_hostname, string receive_hostname, int listen_port, char* statsdInfo, char* statsdTagInfo, char* statsdTagMetric, int quiet) {
@@ -95,7 +95,7 @@ StatsWriter::StatsWriter (string listen_hostname, string receive_hostname, int l
     }
 }
 
-void StatsWriter::writeStats (string guid, string peer, int port, Stats *stats) {
+void StatsWriter::writeStats (string guid, string peer, int port, Stats *stats, time_t now) {
     int n = peer.length();
     char* peerCopy = new char[n + 1];
     strncpy(peerCopy, peer.c_str(), n);
@@ -109,7 +109,7 @@ void StatsWriter::writeStats (string guid, string peer, int port, Stats *stats)
     }
 
     if (!this->quiet) {
-        writeConsoleStats(guid, peerCopy, port, stats);
+        writeConsoleStats(guid, peerCopy, port, stats, now);
     }
     if (this->statsd_sa) {
         writeStatsdStats(guid, peerCopy, port, stats);
@@ -119,8 +119,9 @@ void StatsWriter::writeStats (string guid, string peer, int port, Stats *stats)
     }
 }
 
-void StatsWriter::writeConsoleStats (string guid, string peer, int port, Stats* stats) {
+void StatsWriter::writeConsoleStats (string guid, string peer, int port, Stats* stats, time_t now) {
     printf ("{");
+    printf ("\"time\":\"%ld\",", now);
     printf ("\"from_host\":\"%s\",", peer.c_str());
     printf ("\"from_port\":\"%d\",", port);
     printf ("\"to_host\":\"%s\",", this->receive_hostname);
@@ -152,4 +153,4 @@ void StatsWriter::writeTagsStats (string guid, string peer, int port, Stats* sta
     stat << this->tags_metric << ".sos,from_host=" << peer << ",from_port=" << port << ",to_host=" << this->receive_hostname << ",to_port=" << this->listen_port << ":" << stats->getSumOfSquares() << "|c" << endl;
     stat << this->tags_metric << ".max,from_host=" << peer << ",from_port=" << port << ",to_host=" << this->receive_hostname << ",to_port=" << this->listen_port << ":" << stats->getMax() << "|g" << endl;
     sendto(this->fd, stat.str().c_str(), stat.str().length(), 0, this->tags_sa, sizeof(sockaddr));
-}
+}