mesos-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bmah...@apache.org
Subject [2/3] git commit: Updated users of ProtobufProcess::from.
Date Thu, 24 Oct 2013 22:11:28 GMT
Updated users of ProtobufProcess::from.

Review: https://reviews.apache.org/r/14901


Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/1657a845
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/1657a845
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/1657a845

Branch: refs/heads/master
Commit: 1657a84569e8455e7c39b6529a70a248dbab08ac
Parents: f6793de
Author: Benjamin Mahler <bmahler@twitter.com>
Authored: Wed Oct 23 20:29:00 2013 -0700
Committer: Benjamin Mahler <bmahler@twitter.com>
Committed: Thu Oct 24 14:31:07 2013 -0700

----------------------------------------------------------------------
 src/exec/exec.cpp          |   2 +-
 src/master/master.cpp      |  93 +++++++++++++--------------------
 src/master/master.hpp      | 110 ++++++++++++++++++++++++----------------
 src/sasl/authenticatee.hpp |   4 +-
 src/sched/sched.cpp        |   4 +-
 src/slave/slave.cpp        |  16 +++---
 src/slave/slave.hpp        |  12 +++--
 src/tests/cluster.hpp      |   2 +-
 8 files changed, 127 insertions(+), 116 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/exec/exec.cpp
----------------------------------------------------------------------
diff --git a/src/exec/exec.cpp b/src/exec/exec.cpp
index 7ca21fa..4a598f5 100644
--- a/src/exec/exec.cpp
+++ b/src/exec/exec.cpp
@@ -233,7 +233,7 @@ protected:
     VLOG(1) << "Executor::reregistered took " << stopwatch.elapsed();
   }
 
-  void reconnect(const SlaveID& slaveId)
+  void reconnect(const UPID& from, const SlaveID& slaveId)
   {
     if (aborted) {
       VLOG(1) << "Ignoring reconnect message from slave " << slaveId

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index f838a9d..1147cc6 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -714,37 +714,24 @@ void Master::noMasterDetected()
 }
 
 
-void Master::registerFramework(const FrameworkInfo& frameworkInfo)
+void Master::registerFramework(
+    const UPID& from,
+    const FrameworkInfo& frameworkInfo)
 {
   if (authenticating.contains(from)) {
     LOG(INFO) << "Queuing up registration request from " << from
               << " because authentication is still in progress";
 
     authenticating[from]
-      .onReady(defer(self(), &Self::_registerFramework, frameworkInfo, from));
-  } else {
-    _registerFramework(frameworkInfo, from);
+      .onReady(defer(self(), &Self::registerFramework, from, frameworkInfo));
+    return;
   }
-}
-
 
-void Master::_registerFramework(
-    const FrameworkInfo& frameworkInfo,
-    const UPID& from)
-{
   if (!elected) {
     LOG(WARNING) << "Ignoring register framework message since not elected yet";
     return;
   }
 
-  if (authenticating.contains(from)) {
-    // This could happen if another authentication request came
-    // through before we are here.
-    LOG(WARNING) << "Ignoring registration request from " << from
-                 << " because authentication is back in progress";
-    return;
-  }
-
   if (flags.authenticate && !authenticated.contains(from)) {
     // This could happen if another authentication request came
     // through before we are here or if a framework tried to register
@@ -754,14 +741,14 @@ void Master::_registerFramework(
     FrameworkErrorMessage message;
     message.set_message("Framework at " + stringify(from) +
                         " is not authenticated.");
-    reply(message);
+    send(from, message);
     return;
   }
 
   if (!roles.contains(frameworkInfo.role())) {
     FrameworkErrorMessage message;
     message.set_message("Role '" + frameworkInfo.role() + "' is not valid.");
-    reply(message);
+    send(from, message);
     return;
   }
 
@@ -775,7 +762,7 @@ void Master::_registerFramework(
       FrameworkRegisteredMessage message;
       message.mutable_framework_id()->MergeFrom(framework->id);
       message.mutable_master_info()->MergeFrom(info);
-      reply(message);
+      send(from, message);
       return;
     }
   }
@@ -792,7 +779,7 @@ void Master::_registerFramework(
               << "root submissions are disabled on this cluster";
     FrameworkErrorMessage message;
     message.set_message("User 'root' is not allowed to run frameworks");
-    reply(message);
+    send(from, message);
     delete framework;
     return;
   }
@@ -802,6 +789,7 @@ void Master::_registerFramework(
 
 
 void Master::reregisterFramework(
+    const UPID& from,
     const FrameworkInfo& frameworkInfo,
     bool failover)
 {
@@ -811,21 +799,13 @@ void Master::reregisterFramework(
 
     authenticating[from]
       .onReady(defer(self(),
-                     &Self::_reregisterFramework,
+                     &Self::reregisterFramework,
+                     from,
                      frameworkInfo,
-                     failover,
-                     from));
-  } else {
-    _reregisterFramework(frameworkInfo, failover, from);
+                     failover));
+    return;
   }
-}
-
 
-void Master::_reregisterFramework(
-    const FrameworkInfo& frameworkInfo,
-    bool failover,
-    const UPID& from)
-{
   if (!elected) {
     LOG(WARNING) << "Ignoring re-register framework message since "
                  << "not elected yet";
@@ -836,16 +816,7 @@ void Master::_reregisterFramework(
     LOG(ERROR) << "Framework re-registering without an id!";
     FrameworkErrorMessage message;
     message.set_message("Framework reregistered without a framework id");
-    reply(message);
-    return;
-  }
-
-  if (authenticating.contains(from)) {
-    // This could happen if another authentication request came
-    // through before we are here.
-    LOG(WARNING)
-      << "Ignoring re-registration request for framework " << frameworkInfo.id()
-      << " from " << from << " because authentication is back in progress";
+    send(from, message);
     return;
   }
 
@@ -858,14 +829,14 @@ void Master::_reregisterFramework(
     FrameworkErrorMessage message;
     message.set_message("Framework '" + frameworkInfo.id().value() + "' at " +
                         stringify(from) + " is not authenticated.");
-    reply(message);
+    send(from, message);
     return;
   }
 
   if (!roles.contains(frameworkInfo.role())) {
     FrameworkErrorMessage message;
     message.set_message("Role '" + frameworkInfo.role() + "' is not valid.");
-    reply(message);
+    send(from, message);
     return;
   }
 
@@ -915,7 +886,7 @@ void Master::_reregisterFramework(
       FrameworkReregisteredMessage message;
       message.mutable_framework_id()->MergeFrom(frameworkInfo.id());
       message.mutable_master_info()->MergeFrom(info);
-      reply(message);
+      send(from, message);
       return;
     }
   } else {
@@ -975,7 +946,9 @@ void Master::_reregisterFramework(
 }
 
 
-void Master::unregisterFramework(const FrameworkID& frameworkId)
+void Master::unregisterFramework(
+    const UPID& from,
+    const FrameworkID& frameworkId)
 {
   LOG(INFO) << "Asked to unregister framework " << frameworkId;
 
@@ -991,7 +964,9 @@ void Master::unregisterFramework(const FrameworkID& frameworkId)
 }
 
 
-void Master::deactivateFramework(const FrameworkID& frameworkId)
+void Master::deactivateFramework(
+    const UPID& from,
+    const FrameworkID& frameworkId)
 {
   Framework* framework = getFramework(frameworkId);
 
@@ -1211,7 +1186,7 @@ void Master::schedulerMessage(const SlaveID& slaveId,
 }
 
 
-void Master::registerSlave(const SlaveInfo& slaveInfo)
+void Master::registerSlave(const UPID& from, const SlaveInfo& slaveInfo)
 {
   if (!elected) {
     LOG(WARNING) << "Ignoring register slave message from "
@@ -1266,10 +1241,12 @@ void Master::registerSlave(const SlaveInfo& slaveInfo)
 }
 
 
-void Master::reregisterSlave(const SlaveID& slaveId,
-                             const SlaveInfo& slaveInfo,
-                             const vector<ExecutorInfo>& executorInfos,
-                             const vector<Task>& tasks)
+void Master::reregisterSlave(
+    const UPID& from,
+    const SlaveID& slaveId,
+    const SlaveInfo& slaveInfo,
+    const vector<ExecutorInfo>& executorInfos,
+    const vector<Task>& tasks)
 {
   if (!elected) {
     LOG(WARNING) << "Ignoring re-register slave message from "
@@ -1460,6 +1437,7 @@ void Master::statusUpdate(const StatusUpdate& update, const UPID&
pid)
 
 
 void Master::exitedExecutor(
+    const UPID& from,
     const SlaveID& slaveId,
     const FrameworkID& frameworkId,
     const ExecutorID& executorId,
@@ -1488,7 +1466,7 @@ void Master::exitedExecutor(
 
   CHECK(!deactivatedSlaves.contains(from))
     << "Received exited message for executor " << executorId << " from
" << from
-    << " which is deactivated slave " << slaveId
+    << " which is the deactivated slave " << slaveId
     << "(" << slave->info.hostname() << ")";
 
   // Tell the allocator about the recovered resources.
@@ -1545,6 +1523,7 @@ void Master::deactivateSlave(const SlaveID& slaveId)
 
 
 void Master::reconcileTasks(
+    const UPID& from,
     const FrameworkID& frameworkId,
     const std::vector<TaskStatus>& statuses)
 {
@@ -1693,7 +1672,7 @@ void Master::offer(const FrameworkID& frameworkId,
 }
 
 
-void Master::authenticate(const UPID& pid)
+void Master::authenticate(const UPID& from, const UPID& pid)
 {
   // Deactivate the framework if it's already registered.
   foreachvalue (Framework* framework, frameworks) {
@@ -1715,7 +1694,7 @@ void Master::authenticate(const UPID& pid)
 
     // Retry after the current authenticator finishes.
     authenticating[pid]
-      .onAny(defer(self(), &Self::authenticate, pid));
+      .onAny(defer(self(), &Self::authenticate, from, pid));
 
     return;
   }

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index 9f5e25b..1eba03f 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -89,67 +89,91 @@ public:
 
   virtual ~Master();
 
-  void submitScheduler(const std::string& name);
-  void newMasterDetected(const UPID& pid);
+  void submitScheduler(
+      const std::string& name);
+  void newMasterDetected(
+      const UPID& pid);
   void noMasterDetected();
   void masterDetectionFailure();
-  void registerFramework(const FrameworkInfo& frameworkInfo);
-  void reregisterFramework(const FrameworkInfo& frameworkInfo,
-                           bool failover);
-  void unregisterFramework(const FrameworkID& frameworkId);
-  void deactivateFramework(const FrameworkID& frameworkId);
-  void resourceRequest(const FrameworkID& frameworkId,
-                       const std::vector<Request>& requests);
-  void launchTasks(const FrameworkID& frameworkId,
-                   const OfferID& offerId,
-                   const std::vector<TaskInfo>& tasks,
-                   const Filters& filters);
-  void reviveOffers(const FrameworkID& frameworkId);
-  void killTask(const FrameworkID& frameworkId, const TaskID& taskId);
-  void schedulerMessage(const SlaveID& slaveId,
-                        const FrameworkID& frameworkId,
-                        const ExecutorID& executorId,
-                        const std::string& data);
-  void registerSlave(const SlaveInfo& slaveInfo);
-  void reregisterSlave(const SlaveID& slaveId,
-                       const SlaveInfo& slaveInfo,
-                       const std::vector<ExecutorInfo>& executorInfos,
-                       const std::vector<Task>& tasks);
-  void unregisterSlave(const SlaveID& slaveId);
-  void statusUpdate(const StatusUpdate& update, const UPID& pid);
-  void exitedExecutor(const SlaveID& slaveId,
-                      const FrameworkID& frameworkId,
-                      const ExecutorID& executorId,
-                      int32_t status);
-  void deactivateSlave(const SlaveID& slaveId);
+  void registerFramework(
+      const process::UPID& from,
+      const FrameworkInfo& frameworkInfo);
+  void reregisterFramework(
+      const process::UPID& from,
+      const FrameworkInfo& frameworkInfo,
+      bool failover);
+  void unregisterFramework(
+      const process::UPID& from,
+      const FrameworkID& frameworkId);
+  void deactivateFramework(
+      const process::UPID& from,
+      const FrameworkID& frameworkId);
+  void resourceRequest(
+      const FrameworkID& frameworkId,
+      const std::vector<Request>& requests);
+  void launchTasks(
+      const FrameworkID& frameworkId,
+      const OfferID& offerId,
+      const std::vector<TaskInfo>& tasks,
+      const Filters& filters);
+  void reviveOffers(
+      const FrameworkID& frameworkId);
+  void killTask(
+      const FrameworkID& frameworkId,
+      const TaskID& taskId);
+  void schedulerMessage(
+      const SlaveID& slaveId,
+      const FrameworkID& frameworkId,
+      const ExecutorID& executorId,
+      const std::string& data);
+  void registerSlave(
+      const process::UPID& from,
+      const SlaveInfo& slaveInfo);
+  void reregisterSlave(
+      const process::UPID& from,
+      const SlaveID& slaveId,
+      const SlaveInfo& slaveInfo,
+      const std::vector<ExecutorInfo>& executorInfos,
+      const std::vector<Task>& tasks);
+  void unregisterSlave(
+      const SlaveID& slaveId);
+  void statusUpdate(
+      const StatusUpdate& update,
+      const UPID& pid);
+  void exitedExecutor(
+      const process::UPID& from,
+      const SlaveID& slaveId,
+      const FrameworkID& frameworkId,
+      const ExecutorID& executorId,
+      int32_t status);
+  void deactivateSlave(
+      const SlaveID& slaveId);
 
   // TODO(bmahler): It would be preferred to use a unique libprocess
   // Process identifier (PID is not sufficient) for identifying the
   // framework instance, rather than relying on re-registration time.
-  void frameworkFailoverTimeout(const FrameworkID& frameworkId,
-                                const Time& reregisteredTime);
+  void frameworkFailoverTimeout(
+      const FrameworkID& frameworkId,
+      const Time& reregisteredTime);
 
-  void offer(const FrameworkID& framework,
-             const hashmap<SlaveID, Resources>& resources);
+  void offer(
+      const FrameworkID& framework,
+      const hashmap<SlaveID, Resources>& resources);
 
   void reconcileTasks(
+      const process::UPID& from,
       const FrameworkID& frameworkId,
       const std::vector<TaskStatus>& statuses);
 
-  void authenticate(const UPID& pid);
+  void authenticate(
+      const process::UPID& from,
+      const process::UPID& pid);
 
 protected:
   virtual void initialize();
   virtual void finalize();
   virtual void exited(const UPID& pid);
 
-  void _registerFramework(const FrameworkInfo& frameworkInfo, const UPID& pid);
-
-  void _reregisterFramework(
-      const FrameworkInfo& frameworkInfo,
-      bool failover,
-      const UPID& pid);
-
   void deactivate(Framework* framework);
 
   // 'promise' is used to signal finish of authentication.

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/sasl/authenticatee.hpp
----------------------------------------------------------------------
diff --git a/src/sasl/authenticatee.hpp b/src/sasl/authenticatee.hpp
index 09ef018..f1a677f 100644
--- a/src/sasl/authenticatee.hpp
+++ b/src/sasl/authenticatee.hpp
@@ -242,7 +242,7 @@ protected:
     message.set_mechanism(mechanism);
     message.set_data(output, length);
 
-    send(from, message);
+    reply(message);
 
     status = STEPPING;
   }
@@ -279,7 +279,7 @@ protected:
       if (output != NULL && length > 0) {
         message.set_data(output, length);
       }
-      send(from, message);
+      reply(message);
     } else {
       status = ERROR;
       std::string error(sasl_errdetail(connection));

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/sched/sched.cpp
----------------------------------------------------------------------
diff --git a/src/sched/sched.cpp b/src/sched/sched.cpp
index 824b4b7..3049096 100644
--- a/src/sched/sched.cpp
+++ b/src/sched/sched.cpp
@@ -554,7 +554,7 @@ protected:
     send(pid, message);
   }
 
-  void lostSlave(const SlaveID& slaveId)
+  void lostSlave(const UPID& from, const SlaveID& slaveId)
   {
     if (aborted) {
       VLOG(1) << "Ignoring lost slave message because the driver is aborted!";
@@ -563,7 +563,7 @@ protected:
 
     if (from != master) {
       LOG(WARNING) << "Ignoring lost slave message from " << from
-                   << "because it is not from the registered master ("
+                   << " because it is not from the registered master ("
                    << master << ")";
       return;
     }

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 5272600..243a830 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -402,7 +402,7 @@ void Slave::finalize()
     // checkpointing. This is because slave recovery tests terminate
     // the slave to simulate slave restart.
     if (!frameworks[frameworkId]->info.checkpoint()) {
-      shutdownFramework(frameworkId);
+      shutdownFramework(UPID(), frameworkId);
     }
   }
 
@@ -421,7 +421,7 @@ void Slave::finalize()
 }
 
 
-void Slave::shutdown()
+void Slave::shutdown(const UPID& from)
 {
   // Allow shutdown message only if
   // 1) Its a message received from the registered master or
@@ -451,7 +451,7 @@ void Slave::shutdown()
     // NOTE: We use 'frameworks.keys()' here because 'shutdownFramework'
     // can potentially remove a framework from 'frameworks'.
     foreach (const FrameworkID& frameworkId, frameworks.keys()) {
-      shutdownFramework(frameworkId);
+      shutdownFramework(from, frameworkId);
     }
   }
 }
@@ -534,7 +534,7 @@ void Slave::noMasterDetected()
 }
 
 
-void Slave::registered(const SlaveID& slaveId)
+void Slave::registered(const UPID& from, const SlaveID& slaveId)
 {
   if (from != master) {
     LOG(WARNING) << "Ignoring registration message from " << from
@@ -581,7 +581,7 @@ void Slave::registered(const SlaveID& slaveId)
 }
 
 
-void Slave::reregistered(const SlaveID& slaveId)
+void Slave::reregistered(const UPID& from, const SlaveID& slaveId)
 {
   if (from != master) {
     LOG(WARNING) << "Ignoring re-registration message from " << from
@@ -1102,7 +1102,9 @@ void Slave::killTask(const FrameworkID& frameworkId, const TaskID&
taskId)
 // sending back a shut down acknowledgement, because otherwise you
 // could get into a state where a shut down was sent, dropped, and
 // therefore never processed.
-void Slave::shutdownFramework(const FrameworkID& frameworkId)
+void Slave::shutdownFramework(
+    const UPID& from,
+    const FrameworkID& frameworkId)
 {
   // Allow shutdownFramework() only if
   // its called directly (e.g. Slave::finalize()) or
@@ -1394,6 +1396,7 @@ void Slave::_statusUpdateAcknowledgement(
 
 
 void Slave::registerExecutor(
+    const UPID& from,
     const FrameworkID& frameworkId,
     const ExecutorID& executorId)
 {
@@ -1545,6 +1548,7 @@ void Slave::registerExecutor(
 
 
 void Slave::reregisterExecutor(
+    const UPID& from,
     const FrameworkID& frameworkId,
     const ExecutorID& executorId,
     const vector<TaskInfo>& tasks,

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 5211b46..b9f9b04 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -82,13 +82,13 @@ public:
 
   virtual ~Slave();
 
-  void shutdown();
+  void shutdown(const process::UPID& from);
 
   void newMasterDetected(const UPID& pid);
   void noMasterDetected();
   void masterDetectionFailure();
-  void registered(const SlaveID& slaveId);
-  void reregistered(const SlaveID& slaveId);
+  void registered(const process::UPID& from, const SlaveID& slaveId);
+  void reregistered(const process::UPID& from, const SlaveID& slaveId);
   void doReliableRegistration();
 
   void runTask(
@@ -108,7 +108,9 @@ public:
 
   void killTask(const FrameworkID& frameworkId, const TaskID& taskId);
 
-  void shutdownFramework(const FrameworkID& frameworkId);
+  void shutdownFramework(
+      const process::UPID& from,
+      const FrameworkID& frameworkId);
 
   void schedulerMessage(
       const SlaveID& slaveId,
@@ -119,6 +121,7 @@ public:
   void updateFramework(const FrameworkID& frameworkId, const std::string& pid);
 
   void registerExecutor(
+      const process::UPID& from,
       const FrameworkID& frameworkId,
       const ExecutorID& executorId);
 
@@ -127,6 +130,7 @@ public:
   //           driver never received an ACK for.)
   // 'updates' : Unacknowledged updates.
   void reregisterExecutor(
+      const process::UPID& from,
       const FrameworkID& frameworkId,
       const ExecutorID& executorId,
       const std::vector<TaskInfo>& tasks,

http://git-wip-us.apache.org/repos/asf/mesos/blob/1657a845/src/tests/cluster.hpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.hpp b/src/tests/cluster.hpp
index f743bb3..5621139 100644
--- a/src/tests/cluster.hpp
+++ b/src/tests/cluster.hpp
@@ -379,7 +379,7 @@ inline Try<Nothing> Cluster::Slaves::stop(
   Slave slave = slaves[pid];
 
   if (shutdown) {
-    process::dispatch(slave.slave, &slave::Slave::shutdown);
+    process::dispatch(slave.slave, &slave::Slave::shutdown, process::UPID());
   } else {
     process::terminate(slave.slave);
   }


Mime
View raw message