svn commit: r1419936 - /incubator/mesos/branches/0.10.x/src/sched/sched.cpp

[be...@apache.org]

Author: benh
Date: Tue Dec 11 00:03:43 2012
New Revision: 1419936

URL: http://svn.apache.org/viewvc?rev=1419936&view=rev
Log:
*** MODIFIED FOR 0.10.0 ***
Fixed bug where a status update acknowledgment is sent after the
driver has been aborted.

From: Vinod Kone <vi...@gmail.com>
Review: https://reviews.apache.org/r/7782

Modified:
    incubator/mesos/branches/0.10.x/src/sched/sched.cpp

Modified: incubator/mesos/branches/0.10.x/src/sched/sched.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/branches/0.10.x/src/sched/sched.cpp?rev=1419936&r1=1419935&r2=1419936&view=diff
==============================================================================
--- incubator/mesos/branches/0.10.x/src/sched/sched.cpp (original)
+++ incubator/mesos/branches/0.10.x/src/sched/sched.cpp Tue Dec 11 00:03:43 2012
@@ -263,7 +263,7 @@ protected:
       send(master, message);
     }
 
-    delay(1.0, self(), &SchedulerProcess::doReliableRegistration);
+    delay(1.0, self(), &Self::doReliableRegistration);
   }
 
   void resourceOffers(const vector<Offer>& offers,
@@ -337,19 +337,30 @@ protected:
 
     scheduler->statusUpdate(driver, status);
 
-    // Send a status update acknowledgement ONLY if not aborted!
-    if (!aborted && pid) {
-      // Acknowledge the message (we do this last, after we invoked
-      // the scheduler, if we did at all, in case it causes a crash,
-      // since this way the message might get resent/routed after the
-      // scheduler comes back online).
-      StatusUpdateAcknowledgementMessage message;
-      message.mutable_framework_id()->MergeFrom(framework.id());
-      message.mutable_slave_id()->MergeFrom(update.slave_id());
-      message.mutable_task_id()->MergeFrom(status.task_id());
-      message.set_uuid(update.uuid());
-      send(pid, message);
+    // Acknowledge the status update.
+    // NOTE: We do a dispatch here instead of directly sending the ACK because,
+    // we want to avoid sending the ACK if the driver was aborted when we
+    // made the statusUpdate call. This works because, the 'abort' message will
+    // be enqueued before the ACK message is processed.
+    if (pid) {
+      dispatch(self(), &Self::statusUpdateAcknowledgement, update, pid);
+    }
+  }
+
+  void statusUpdateAcknowledgement(const StatusUpdate& update, const UPID& pid)
+  {
+    if (aborted) {
+      VLOG(1) << "Not sending status update acknowledgment message because "
+              << "the driver is aborted!";
+      return;
     }
+
+    StatusUpdateAcknowledgementMessage message;
+    message.mutable_framework_id()->MergeFrom(framework.id());
+    message.mutable_slave_id()->MergeFrom(update.slave_id());
+    message.mutable_task_id()->MergeFrom(update.status().task_id());
+    message.set_uuid(update.uuid());
+    send(pid, message);
   }
 
   void lostSlave(const SlaveID& slaveId)