Fixed a bug where a Connection would sometimes lose track of its localPort.

pelab/magent/CircularTraffic.cc

@@ -63,6 +63,7 @@ void CircularTraffic::writeToPeer(ConnectionModel * peer,
     current = (current + 1) % usedCount;
     peer->writeMessage(writes[current].size, result);
     result.nextWrite = previousTime + writes[current].delta;
+    result.isConnected = peer->isConnected();
   }
   else
   {
@@ -73,4 +74,3 @@ void CircularTraffic::writeToPeer(ConnectionModel * peer,
     result.nextWrite = Time();
   }
 }
-

pelab/magent/Connection.cc

@@ -96,6 +96,8 @@ void Connection::connect(void)
     isConnected = peer->isConnected();
     if (isConnected)
     {
+      logWrite(CONNECTION, "Inserting a connection with key %s",
+               planet.toString().c_str());
       global::planetMap.insert(make_pair(planet, this));
     }
   }
@@ -133,6 +135,7 @@ void Connection::addSensor(SensorCommand const & newSensor)
 
 void Connection::captureSend(PacketInfo * packet)
 {
+  logWrite(SENSOR, "Captured a send packet");
   Sensor * head = measurements.getHead();
   if (head != NULL && isConnected)
   {
@@ -144,6 +147,7 @@ void Connection::captureSend(PacketInfo * packet)
 
 void Connection::captureAck(PacketInfo * packet)
 {
+  logWrite(SENSOR, "Captured an ack packet");
   Sensor * head = measurements.getHead();
   if (head != NULL && isConnected)
   {
@@ -163,6 +167,10 @@ Time Connection::writeToConnection(Time const & previousTime)
   result.planet.transport = TCP_CONNECTION;
   result.planet.ip = elab.ip;
   result.planet.remotePort = global::peerServerPort;
+  if (isConnected)
+  {
+    result.planet.localPort = planet.localPort;
+  }
   traffic->writeToPeer(peer.get(), previousTime, result);
   if (!isConnected && result.isConnected)
   {
@@ -172,9 +180,11 @@ Time Connection::writeToConnection(Time const & previousTime)
   else if (isConnected && result.isConnected
            && planet != result.planet)
   {
+    logWrite(CONNECTION, "OldKey: %s", planet.toString().c_str());
     global::planetMap.erase(planet);
     planet = result.planet;
     global::planetMap.insert(make_pair(planet, this));
+    logWrite(CONNECTION, "NewKey: %s", planet.toString().c_str());
   }
   isConnected = result.isConnected;
   bufferFull = result.bufferFull;

pelab/magent/KernelTcp.cc

@@ -199,7 +199,11 @@ int KernelTcp::writeMessage(int size, WriteResult & result)
     }
     else if (error == -1)
     {
-      if (errno != EWOULDBLOCK)
+      if (errno == EWOULDBLOCK)
+      {
+        result.bufferFull = true;
+      }
+      else
       {
         logWrite(EXCEPTION, "Failed write to peer: %s", strerror(errno));
       }
@@ -391,6 +395,7 @@ namespace
                          struct pcap_pkthdr const * pcapInfo,
                          unsigned char const * packet)
   {
+    logWrite(PCAP, "Captured a packet");
     int packetType = getLinkLayer(pcapInfo, packet);
     if (packetType == -1)
     {
@@ -470,6 +475,7 @@ namespace
                  IpHeader const * ipPacket,
                  struct tcphdr const * tcpPacket)
   {
+    logWrite(PCAP, "Captured a TCP packet");
     struct tcp_info kernelInfo;
     bool isAck;
     if (tcpPacket->ack & 0x0001)

pelab/magent/PacketSensor.cc

@@ -50,7 +50,7 @@ void PacketSensor::localSend(PacketInfo * packet)
   else
   {
     SentPacket record;
-    record.seqStart = ntohl(startSequence);
+    record.seqStart = startSequence;
     unsigned int sequenceLength = packet->packetLength -
       sizeof(struct ether_header) - IP_HL(packet->ip)*4 -
       sizeof(struct tcphdr);
@@ -112,16 +112,25 @@ void PacketSensor::localAck(PacketInfo * packet)
 
 bool PacketSensor::SentPacket::inSequenceBlock(unsigned int sequence)
 {
+  logWrite(SENSOR,
+           "PacketSensor inSequencBlock(): Is %u between %u and %u?",
+           sequence, seqStart, seqEnd);
+  bool result = false;
   if (seqStart < seqEnd)
   {
-    return sequence >= seqStart && sequence < seqEnd;
+    result = sequence >= seqStart && sequence < seqEnd;
   }
   else if (seqStart > seqEnd)
   {
-    return sequence >= seqStart || sequence < seqEnd;
+    result = sequence >= seqStart || sequence < seqEnd;
+  }
+  if (result)
+  {
+    logWrite(SENSOR, "Yes!");
   }
   else
   {
-    return false;
+    logWrite(SENSOR, "No!");
   }
+  return result;
 }

pelab/magent/lib.h

@@ -44,6 +44,13 @@ extern char * optarg;
 
 #include "Time.h"
 
+void setDescriptor(int fd);
+void clearDescriptor(int fd);
+std::string ipToString(unsigned int ip);
+int createServer(int port, std::string const & debugString);
+int acceptServer(int acceptfd, struct sockaddr_in * remoteAddress,
+                 std::string const & debugString);
+
 // Enum of header types -- to-monitor (Events, etc.)
 enum
 {
@@ -125,6 +132,20 @@ struct Order
   {
     return !(*this == right);
   }
+  std::string toString(void)
+  {
+    std::ostringstream buffer;
+    if (transport == TCP_CONNECTION)
+    {
+      buffer << "TCP,";
+    }
+    else
+    {
+      buffer << "UDP,";
+    }
+    buffer << localPort << ":" << ipToString(htonl(ip)) << ":" << remotePort;
+    return buffer.str();
+  }
 };
 
 struct WriteResult
@@ -199,11 +220,4 @@ namespace global
   extern std::auto_ptr<CommandInput> input;
   extern std::auto_ptr<CommandOutput> output;
 }
-
-void setDescriptor(int fd);
-void clearDescriptor(int fd);
-std::string ipToString(unsigned int ip);
-int createServer(int port, std::string const & debugString);
-int acceptServer(int acceptfd, struct sockaddr_in * remoteAddress,
-                 std::string const & debugString);
 #endif

pelab/magent/log.cc

@@ -66,6 +66,10 @@ static void logPrefix(int flags)
   {
     fprintf(logFile, "CONNECTION ");
   }
+  if (flags & PCAP)
+  {
+    fprintf(logFile, "PCAP ");
+  }
   if (logTimestamp)
   {
     struct timeval now;

pelab/magent/log.h

@@ -47,9 +47,10 @@ enum LOG_TYPE
   MAIN_LOOP        = 0x040,
   COMMAND_INPUT    = 0x080,
   CONNECTION       = 0x100,
+  PCAP             = 0x200,
   // Shortcuts for common cases.
   LOG_NOTHING      = 0x000,
-  LOG_EVERYTHING   = 0x1ff
+  LOG_EVERYTHING   = 0x3ff
 };
 
 #endif

pelab/magent/main.cc

@@ -196,6 +196,7 @@ void mainLoop(void)
   // Select on file descriptors
   while (true)
   {
+    cerr << "(";
 //    struct timeval debugTimeout;
 //    debugTimeout.tv_sec = 0;
 //    debugTimeout.tv_usec = 100000;
@@ -216,9 +217,11 @@ void mainLoop(void)
       timeUntilWrite = Time();
       waitPeriod = NULL;
     }
+    cerr << "s";
     int error = select(global::maxReader + 1, &readable, NULL, NULL,
 //                       &debugTimeout);
                        waitPeriod);
+    cerr << "e";
     if (error == -1)
     {
       switch (errno)
@@ -239,6 +242,7 @@ void mainLoop(void)
       }
     }
 
+    cerr << "c";
     global::input->nextCommand(&readable);
     Command * current = global::input->getCommand();
     if (current != NULL)
@@ -247,21 +251,27 @@ void mainLoop(void)
 //      global::input->nextCommand(&readable);
 //      current = global::input->getCommand();
     }
+    cerr << "w";
     writeToConnections(schedule);
+    cerr << "a";
     addNewPeer(&readable);
+    cerr << "r";
     readFromPeers(&readable);
+    cerr << "c";
     packetCapture(&readable);
+    cerr << ")";
   }
 }
 
 void writeToConnections(multimap<Time, Connection *> & schedule)
 {
   Time now = getCurrentTime();
-  bool done = false;
+//  bool done = false;
     // Notify any connection which is due, then erase that entry from
     // the schedule and insert another entry at the new time specified
     // by the connection.
-  while (! schedule.empty() && !done)
+//  while (! schedule.empty() && !done)
+  if (! schedule.empty())
   {
     multimap<Time, Connection *>::iterator pos = schedule.begin();
     if (pos->first < now)
@@ -274,10 +284,10 @@ void writeToConnections(multimap<Time, Connection *> & schedule)
         schedule.insert(make_pair(nextTime, current));
       }
     }
-    else
-    {
-      done = true;
-    }
+//    else
+//    {
+//      done = true;
+//    }
   }
 }