Added rudimentary error checking for sensors. Each sensor has an ackValid and...

Added rudimentary error checking for sensors. Each sensor has an ackValid and a sendValid boolean value which says whether the data from a recent ack or send is valid. These should be checked before any access to data in a sensor.

pelab/magent/DelaySensor.cc

@@ -18,19 +18,37 @@ DelaySensor::DelaySensor(PacketSensor const * newPacketHistory,
 
 int DelaySensor::getLastDelay(void) const
 {
-  return lastDelay;
+  if (ackValid)
+  {
+    return lastDelay;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "DelaySensor::getLastDelay() called with invalid ack data");
+    return 0;
+  }
 }
 
 void DelaySensor::localSend(PacketInfo *)
 {
+  ackValid = false;
+  sendValid = true;
 }
 
 void DelaySensor::localAck(PacketInfo * packet)
 {
-  if (state->getState() == StateSensor::ESTABLISHED)
+  sendValid = false;
+  if (state->isAckValid() && packetHistory->isAckValid()
+      && state->getState() == StateSensor::ESTABLISHED)
   {
     Time diff = packet->packetTime - packetHistory->getAckedSendTime();
     lastDelay = diff.toMilliseconds();
     logWrite(SENSOR, "DELAY: %d ms", lastDelay);
+    ackValid = true;
+  }
+  else
+  {
+    ackValid = false;
   }
 }

pelab/magent/EwmaThroughputSensor.cc

@@ -20,44 +20,55 @@ EwmaThroughputSensor::EwmaThroughputSensor(
 
 void EwmaThroughputSensor::localSend(PacketInfo * packet)
 {
+  ackValid = false;
+  sendValid = true;
 }
 
 void EwmaThroughputSensor::localAck(PacketInfo * packet)
 {
-  int latest = throughputSource->getThroughputInKbps();
-  if (state->isSaturated())
+  sendValid = false;
+  if (throughputSource->isAckValid() && state->isAckValid())
   {
-    // The link is saturated, so we know that the throughput
-    // measurement is the real bandwidth.
-    if (bandwidth == 0.0)
+    int latest = throughputSource->getThroughputInKbps();
+    if (state->isSaturated())
     {
-      bandwidth = latest;
+      // The link is saturated, so we know that the throughput
+      // measurement is the real bandwidth.
+      if (bandwidth == 0.0)
+      {
+        bandwidth = latest;
+      }
+      else
+      {
+        static const double alpha = 0.1;
+        bandwidth = bandwidth*(1.0-alpha) + latest*alpha;
+      }
+      // We have got an actual bandwidth measurement, so reset
+      // maxThroughput accordingly.
+      maxThroughput = static_cast<int>(bandwidth);
+      ostringstream buffer;
+      buffer << static_cast<int>(bandwidth);
+      global::output->genericMessage(AUTHORITATIVE_BANDWIDTH, buffer.str(),
+                                     packet->elab);
     }
     else
     {
-      static const double alpha = 0.1;
-      bandwidth = bandwidth*(1.0-alpha) + latest*alpha;
+      // The link isn't saturated, so we don't know whether this
+      // throughput measurement represents real bandwidth or not.
+      if (latest > maxThroughput)
+      {
+        maxThroughput = latest;
+        // Send out a tentative number
+        ostringstream buffer;
+        buffer << maxThroughput;
+        global::output->genericMessage(TENTATIVE_THROUGHPUT, buffer.str(),
+                                       packet->elab);
+      }
     }
-    // We have got an actual bandwidth measurement, so reset
-    // maxThroughput accordingly.
-    maxThroughput = static_cast<int>(bandwidth);
-    ostringstream buffer;
-    buffer << static_cast<int>(bandwidth);
-    global::output->genericMessage(AUTHORITATIVE_BANDWIDTH, buffer.str(),
-                            packet->elab);
+    ackValid = true;
   }
   else
   {
-    // The link isn't saturated, so we don't know whether this
-    // throughput measurement represents real bandwidth or not.
-    if (latest > maxThroughput)
-    {
-      maxThroughput = latest;
-      // Send out a tentative number
-      ostringstream buffer;
-      buffer << maxThroughput;
-      global::output->genericMessage(TENTATIVE_THROUGHPUT, buffer.str(),
-                              packet->elab);
-    }
+    ackValid = false;
   }
 }

pelab/magent/MaxDelaySensor.cc

@@ -23,11 +23,12 @@ MaxDelaySensor::MaxDelaySensor(DelaySensor const * newDelay,
 
 void MaxDelaySensor::localSend(PacketInfo * packet)
 {
+  ackValid = false;
   /*
    * If we detect packet loss, report the maximum delay we've recently seen on
    * the forward path
    */
-  if (packetsensor->getIsRetransmit()) {
+  if (packetsensor->isSendValid() && packetsensor->getIsRetransmit()) {
     if (lastreported != maximum.get()) {
       logWrite(SENSOR,"MaxDelaySensor::localSend() reporting new max %d",
                maximum.get());
@@ -39,16 +40,25 @@ void MaxDelaySensor::localSend(PacketInfo * packet)
       logWrite(SENSOR_DETAIL,"MaxDelaySensor::localSend() suppressing max %d",
                maximum.get());
     }
+    sendValid = true;
+  }
+  else
+  {
+    sendValid = false;
   }
 }
 
 void MaxDelaySensor::localAck(PacketInfo * packet)
 {
+  sendValid = false;
 
   /*
    * Only try to make this calculation on established connections
    */
-  if (state->getState() != StateSensor::ESTABLISHED) {
+  if (! state->isAckValid() || ! delay->isAckValid()
+      || ! mindelay->isAckValid()
+      ||  state->getState() != StateSensor::ESTABLISHED) {
+    ackValid = false;
     return;
   }
 
@@ -67,6 +77,7 @@ void MaxDelaySensor::localAck(PacketInfo * packet)
            minimumDelay, queueingDelay, state->isSaturated());
   if (queueingDelay < 0) {
     logWrite(ERROR,"Queueing delay is less than zero!");
+    ackValid = false;
     return;
   }
   /*
@@ -82,4 +93,5 @@ void MaxDelaySensor::localAck(PacketInfo * packet)
   {
     maximum.decay();
   }
+  ackValid = true;
 }

pelab/magent/MinDelaySensor.cc

@@ -13,28 +13,53 @@ MinDelaySensor::MinDelaySensor(DelaySensor const * newDelay)
   delay = newDelay;
 }
 
+int MinDelaySensor::getMinDelay(void) const
+{
+  if (ackValid)
+  {
+    return minDelay;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "MinDelaySensor::getMinDelay() called with invalid ack data");
+    return 0;
+  }
+}
+
 void MinDelaySensor::localSend(PacketInfo *)
 {
+  ackValid = false;
+  sendValid = true;
 }
 
 void MinDelaySensor::localAck(PacketInfo * packet)
 {
-  int current = delay->getLastDelay();
-  int oneway = current / 2;
-  if (current < minimum && current != 0 && oneway != lastreported)
+  sendValid = false;
+  if (delay->isAckValid())
   {
-    minDelay = current;
-    ostringstream buffer;
-    buffer << "delay=" << oneway;
-    minimum.reset(current);
-    global::output->eventMessage(buffer.str(), packet->elab,
-                                 CommandOutput::FORWARD_PATH);
-    global::output->eventMessage(buffer.str(), packet->elab,
-                                 CommandOutput::BACKWARD_PATH);
-    lastreported = oneway;
+    int current = delay->getLastDelay();
+    int oneway = current / 2;
+    if (current < minimum && current != 0 && oneway != lastreported)
+    {
+      minDelay = current;
+      ostringstream buffer;
+      buffer << "delay=" << oneway;
+      minimum.reset(current);
+      global::output->eventMessage(buffer.str(), packet->elab,
+                                   CommandOutput::FORWARD_PATH);
+      global::output->eventMessage(buffer.str(), packet->elab,
+                                   CommandOutput::BACKWARD_PATH);
+      lastreported = oneway;
+    }
+    else
+    {
+      minimum.decay();
+    }
+    ackValid = false;
   }
   else
   {
-    minimum.decay();
+    ackValid = false;
   }
 }

pelab/magent/MinDelaySensor.h

@@ -16,7 +16,7 @@ class MinDelaySensor : public Sensor
 public:
   MinDelaySensor(DelaySensor const * newDelay);
   // Note: This is the minimum RTT, not one-way delay
-  int getMinDelay(void) const { return minDelay; }
+  int getMinDelay(void) const;
 protected:
   virtual void localSend(PacketInfo * packet);
   virtual void localAck(PacketInfo * packet);

pelab/magent/PacketSensor.cc

@@ -17,26 +17,45 @@ PacketSensor::PacketSensor(StateSensor const * newState)
 
 int PacketSensor::getAckedSize(void) const
 {
-  if (ackedSendTime == Time())
+  if (ackValid)
   {
-    logWrite(ERROR, "PacketSensor::getAckedSize() called before localAck()");
+    return ackedSize;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "PacketSensor::getAckedSize() called with invalid ack data");
+    return 0;
   }
-  return ackedSize;
 }
 
 bool PacketSensor::getIsRetransmit(void) const
 {
-  return isRetransmit;
+  if (sendValid)
+  {
+    return isRetransmit;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "PacketSensor::getIsRetransmit() called with invalid send data");
+    return false;
+  }
 }
 
 Time const & PacketSensor::getAckedSendTime(void) const
 {
-  if (ackedSendTime == Time())
+  if (ackValid)
   {
-    logWrite(ERROR, "PacketSensor::getAckedSendTime() called before"
-             " localAck()");
+    return ackedSendTime;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "PacketSensor::getAckedSendTime() called with invalid ack data");
+    static Time invalidTime = Time();
+    return invalidTime;
   }
-  return ackedSendTime;
 }
 
 void PacketSensor::localSend(PacketInfo * packet)
@@ -57,9 +76,12 @@ void PacketSensor::localSend(PacketInfo * packet)
     }
   }
   // Assume this packet is not a retransmit unless proven otherwise
+  ackValid = false;
   isRetransmit = false;
-  if (state->getState() == StateSensor::ESTABLISHED)
+  if (state->isSendValid() && state->getState() == StateSensor::ESTABLISHED)
   {
+    // Set it to true, and then set it to false if we encounter an error.
+    sendValid = true;
     logWrite(SENSOR_DETAIL,
              "PacketSensor::localSend() for sequence number %u",
              ntohl(packet->tcp->seq));
@@ -86,7 +108,9 @@ void PacketSensor::localSend(PacketInfo * packet)
       }
 
       if (!done) {
-        logWrite(ERROR, "localSend() unable to find packet record to update.");
+        logWrite(ERROR,
+                 "localSend() unable to find packet record to update.");
+        sendValid = false;
       }
     }
     else
@@ -99,8 +123,8 @@ void PacketSensor::localSend(PacketInfo * packet)
       record.seqStart = startSequence;
 
       /*
-       * Calculate the packet payload size - we have to make sure to take into
-       * account IP and TCP option headers
+       * Calculate the packet payload size - we have to make sure to
+       * take into account IP and TCP option headers
        */
       unsigned int sequenceLength =
         // Total length of the IP part of the packet
@@ -132,9 +156,9 @@ void PacketSensor::localSend(PacketInfo * packet)
         if (record.seqStart != (globalSequence.seqEnd + 1))
         {
           fprintf(stderr,"PacketSensor::localSend() may have missed a "
-                   "packet - last seq seen: %u, new seq: %u (lost %d)",
-                   globalSequence.seqEnd,record.seqStart,
-                   record.seqStart - globalSequence.seqEnd);
+                  "packet - last seq seen: %u, new seq: %u (lost %d)",
+                  globalSequence.seqEnd,record.seqStart,
+                  record.seqStart - globalSequence.seqEnd);
         }
         globalSequence.seqEnd = record.seqEnd;
       }
@@ -147,15 +171,22 @@ void PacketSensor::localSend(PacketInfo * packet)
     ackedSize = 0;
     ackedSendTime = Time();
   }
+  else
+  {
+    sendValid = false;
+  }
 }
 
 void PacketSensor::localAck(PacketInfo * packet)
 {
+  sendValid = false;
   // Right now, we don't know whether or not this ACK is for a retransmitted
   // packet (we could tell, but there doesn't seem to be a need yet)
   isRetransmit = false;
-  if (state->getState() == StateSensor::ESTABLISHED)
+  if (state->isAckValid() && state->getState() == StateSensor::ESTABLISHED)
   {
+    // Set it to true, and then set it to false if we encounter an error.
+    ackValid = true;
     /*
      * When we get an ACK, the sequence number is really the next one the peer
      * excects to see: thus, the last sequence number it's ACKing is one less
@@ -202,6 +233,7 @@ void PacketSensor::localAck(PacketInfo * packet)
          ++opt) {
       if (opt->type == TCPOPT_SACK) {
         logWrite(ERROR,"Packet has a SACK option!");
+        ackValid = false;
       }
     }
 
@@ -228,6 +260,7 @@ void PacketSensor::localAck(PacketInfo * packet)
                "of unacked packets.");
       ackedSize = 0;
       ackedSendTime = Time();
+      ackValid = false;
       return;
     }
     unacked.erase(unacked.begin(), pos);
@@ -244,6 +277,10 @@ void PacketSensor::localAck(PacketInfo * packet)
     logWrite(SENSOR_DETAIL, "PacketSensor::localAck() decided on size %u",
              ackedSize);
   }
+  else
+  {
+    ackValid = false;
+  }
 }
 
 bool PacketSensor::SentPacket::inSequenceBlock(unsigned int sequence)

pelab/magent/Sensor.cc

@@ -5,6 +5,12 @@
 
 using namespace std;
 
+Sensor::Sensor()
+  : sendValid(false)
+  , ackValid(false)
+{
+}
+
 Sensor::~Sensor()
 {
 }
@@ -49,17 +55,32 @@ void Sensor::capturePacket(PacketInfo * packet)
   }
 }
 
+bool Sensor::isSendValid(void) const
+{
+  return sendValid;
+}
+
+bool Sensor::isAckValid(void) const
+{
+  return ackValid;
+}
+
+
 NullSensor::~NullSensor()
 {
 }
 
 void NullSensor::localSend(PacketInfo *)
 {
+  ackValid = false;
+  sendValid = true;
   logWrite(SENSOR, "Send received");
 }
 
 void NullSensor::localAck(PacketInfo * packet)
 {
+  sendValid = false;
+  ackValid = true;
   logWrite(SENSOR, "Ack received");
   list<Option>::iterator pos = packet->tcpOptions->begin();
   list<Option>::iterator limit = packet->tcpOptions->end();

pelab/magent/Sensor.h

@@ -12,15 +12,23 @@
 class Sensor
 {
 public:
+  Sensor();
   virtual ~Sensor();
   Sensor * getTail(void);
   void addNode(std::auto_ptr<Sensor> node);
   void capturePacket(PacketInfo * packet);
+  bool isSendValid(void) const;
+  bool isAckValid(void) const;
 private:
   std::auto_ptr<Sensor> next;
 protected:
   virtual void localSend(PacketInfo * packet)=0;
   virtual void localAck(PacketInfo * packet)=0;
+protected:
+  // This is used for functions which only yield data on a send.
+  bool sendValid;
+  // This is used for functions which only yield data on an ack.
+  bool ackValid;
 };
 
 class NullSensor : public Sensor

pelab/magent/StateSensor.cc

@@ -17,16 +17,37 @@ StateSensor::~StateSensor()
 
 int StateSensor::getState(void) const
 {
-  return state;
+  if (sendValid || ackValid)
+  {
+    return state;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "StateSensor::getState() called with invalid data");
+    return INITIAL;
+
+  }
 }
 
 bool StateSensor::isSaturated(void) const
 {
-  return saturated;
+  if (sendValid || ackValid)
+  {
+    return saturated;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "StateSensor::isSaturated() called with invalid data");
+    return false;
+  }
 }
 
 void StateSensor::localSend(PacketInfo * packet)
 {
+  ackValid = false;
+  sendValid = true;
   if (packet->tcp->syn && state == INITIAL)
   {
     state = AFTER_SYN;
@@ -35,6 +56,7 @@ void StateSensor::localSend(PacketInfo * packet)
   else if (packet->tcp->syn)
   {
     logWrite(ERROR, "Sent a SYN packet out of order");
+    sendValid = false;
   }
   else if (! packet->tcp->syn && state == AFTER_SYN_ACK)
   {
@@ -46,6 +68,8 @@ void StateSensor::localSend(PacketInfo * packet)
 
 void StateSensor::localAck(PacketInfo * packet)
 {
+  sendValid = false;
+  ackValid = true;
   if (packet->tcp->syn && packet->tcp->ack && state == AFTER_SYN)
   {
     state = AFTER_SYN_ACK;
@@ -54,6 +78,7 @@ void StateSensor::localAck(PacketInfo * packet)
   else if (packet->tcp->syn && packet->tcp->ack)
   {
     logWrite(ERROR, "Received a SYNACK packet out of order");
+    ackValid = false;
   }
   calculateSaturated(packet);
 }

pelab/magent/ThroughputSensor.cc

@@ -18,17 +18,32 @@ ThroughputSensor::ThroughputSensor(PacketSensor const * newPacketHistory,
 
 int ThroughputSensor::getThroughputInKbps(void) const
 {
-  return throughputInKbps;
+  if (ackValid)
+  {
+    return throughputInKbps;
+  }
+  else
+  {
+    logWrite(ERROR,
+             "ThroughputSensor::getThroughputInKbps() "
+             "called with invalid data");
+    return 0;
+  }
 }
 
 void ThroughputSensor::localSend(PacketInfo *)
 {
+  ackValid = false;
+  sendValid = true;
 }
 
 void ThroughputSensor::localAck(PacketInfo * packet)
 {
-  if (state->getState() == StateSensor::ESTABLISHED)
+  sendValid = false;
+  if (state->isAckValid() && packetHistory->isAckValid() &&
+      state->getState() == StateSensor::ESTABLISHED)
   {
+    ackValid = true;
     Time currentAckTime = packet->packetTime;
     if (lastAckTime != Time() && currentAckTime != lastAckTime)
     {
@@ -46,6 +61,7 @@ void ThroughputSensor::localAck(PacketInfo * packet)
     else
     {
       throughputInKbps = 0;
+      ackValid = false;
     }
     lastAckTime = currentAckTime;
   }
@@ -53,5 +69,6 @@ void ThroughputSensor::localAck(PacketInfo * packet)
   {
     throughputInKbps = 0;
     lastAckTime = Time();
+    ackValid = false;
   }
 }