You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2013/02/24 06:05:54 UTC
svn commit: r1449438 - in /incubator/mesos/trunk: src/linux/cgroups.cpp
third_party/libprocess/include/process/future.hpp
third_party/libprocess/src/tests/process_tests.cpp
Author: benh
Date: Sun Feb 24 05:05:53 2013
New Revision: 1449438
URL: http://svn.apache.org/r1449438
Log:
Added Future::then which takes result of defer.
Review: https://reviews.apache.org/r/9569
Modified:
incubator/mesos/trunk/src/linux/cgroups.cpp
incubator/mesos/trunk/third_party/libprocess/include/process/future.hpp
incubator/mesos/trunk/third_party/libprocess/src/tests/process_tests.cpp
Modified: incubator/mesos/trunk/src/linux/cgroups.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/linux/cgroups.cpp?rev=1449438&r1=1449437&r2=1449438&view=diff
==============================================================================
--- incubator/mesos/trunk/src/linux/cgroups.cpp (original)
+++ incubator/mesos/trunk/src/linux/cgroups.cpp Sun Feb 24 05:05:53 2013
@@ -1443,7 +1443,7 @@ protected:
{
// Stop when no one cares.
promise.future().onDiscarded(lambda::bind(
- static_cast<void (*)(const UPID&, bool)>(terminate), self(), true));
+ static_cast<void(*)(const UPID&, bool)>(terminate), self(), true));
CHECK(interval >= Seconds(0));
@@ -1460,24 +1460,14 @@ protected:
private:
void killTasks() {
- lambda::function<Future<bool>(const bool&)>
- funcFreeze = defer(self(), &Self::freeze);
- lambda::function<Future<Nothing>(const bool&)>
- funcKill = defer(self(), &Self::kill);
- lambda::function<Future<bool>(const Nothing&)>
- funcThaw = defer(self(), &Self::thaw);
- lambda::function<Future<bool>(const bool&)>
- funcEmpty = defer(self(), &Self::empty);
-
// Chain together the steps needed to kill the tasks. Note that we
- // ignore he return values of freeze, kill, and thaw because,
+ // ignore the return values of freeze, kill, and thaw because,
// provided there are no errors, we'll just retry the chain as
// long as tasks still exist.
- chain = Future<bool>(true)
- .then(funcFreeze) // Freeze the cgroup.
- .then(funcKill) // Send kill signals to all tasks in the cgroup.
- .then(funcThaw) // Thaw the cgroup to let kill signals be received.
- .then(funcEmpty); // Wait until no task is in the cgroup.
+ chain = freeze() // Freeze the cgroup.
+ .then(defer(self(), &Self::kill)) // Send kill signals to all tasks.
+ .then(defer(self(), &Self::thaw)) // Thaw cgroup to deliver signals.
+ .then(defer(self(), &Self::empty)); // Wait until cgroup is empty.
chain.onAny(defer(self(), &Self::finished, lambda::_1));
}
@@ -1554,7 +1544,7 @@ protected:
{
// Stop when no one cares.
promise.future().onDiscarded(lambda::bind(
- static_cast<void (*)(const UPID&, bool)>(terminate), self(), true));
+ static_cast<void(*)(const UPID&, bool)>(terminate), self(), true));
CHECK(interval >= Seconds(0));
Modified: incubator/mesos/trunk/third_party/libprocess/include/process/future.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/third_party/libprocess/include/process/future.hpp?rev=1449438&r1=1449437&r2=1449438&view=diff
==============================================================================
--- incubator/mesos/trunk/third_party/libprocess/include/process/future.hpp (original)
+++ incubator/mesos/trunk/third_party/libprocess/include/process/future.hpp Sun Feb 24 05:05:53 2013
@@ -15,12 +15,17 @@
#include <tr1/memory> // TODO(benh): Replace shared_ptr with unique_ptr.
#include <process/latch.hpp>
+#include <process/pid.hpp>
+#include <process/preprocessor.hpp>
#include <stout/duration.hpp>
#include <stout/option.hpp>
namespace process {
+// Forward declaration (instead of include to break circular dependency).
+template <typename _F> struct _Defer;
+
namespace internal {
template <typename T>
@@ -101,6 +106,12 @@ public:
// and associates the result of the callback with the future that is
// returned to the caller (which may be of a different type).
template <typename X>
+ Future<X> then(const std::tr1::function<Future<X>(const T&)>& f) const;
+
+ template <typename X>
+ Future<X> then(const std::tr1::function<X(const T&)>& f) const;
+
+ template <typename X>
Future<X> then(
const std::tr1::_Bind<
Future<X>(*(std::tr1::_Placeholder<1>))(const T&)>& b) const;
@@ -110,11 +121,57 @@ public:
const std::tr1::_Bind<
X(*(std::tr1::_Placeholder<1>))(const T&)>& b) const;
- template <typename X>
- Future<X> then(const std::tr1::function<Future<X>(const T&)>& f) const;
+ template <typename X, typename U>
+ Future<X> then(const _Defer<Future<X>(*(PID<U>, X(U::*)(void)))
+ (const PID<U>&, X(U::*)(void))>& d) const
+ {
+ return then(std::tr1::function<Future<X>(const T&)>(d));
+ }
- template <typename X>
- Future<X> then(const std::tr1::function<X(const T&)>& f) const;
+#define TEMPLATE(Z, N, DATA) \
+ template <typename X, \
+ typename U, \
+ ENUM_PARAMS(N, typename P), \
+ ENUM_PARAMS(N, typename A)> \
+ Future<X> then( \
+ const _Defer<Future<X>(*(PID<U>, \
+ X(U::*)(ENUM_PARAMS(N, P)), \
+ ENUM_PARAMS(N, A))) \
+ (const PID<U>&, \
+ X(U::*)(ENUM_PARAMS(N, P)), \
+ ENUM_PARAMS(N, P))>& d) const \
+ { \
+ return then(std::tr1::function<Future<X>(const T&)>(d)); \
+ }
+
+ REPEAT_FROM_TO(1, 11, TEMPLATE, _) // Args A0 -> A9.
+#undef TEMPLATE
+
+ template <typename X, typename U>
+ Future<X> then(const _Defer<Future<X>(*(PID<U>, Future<X>(U::*)(void)))
+ (const PID<U>&, Future<X>(U::*)(void))>& d) const
+ {
+ return then(std::tr1::function<Future<X>(const T&)>(d));
+ }
+
+#define TEMPLATE(Z, N, DATA) \
+ template <typename X, \
+ typename U, \
+ ENUM_PARAMS(N, typename P), \
+ ENUM_PARAMS(N, typename A)> \
+ Future<X> then( \
+ const _Defer<Future<X>(*(PID<U>, \
+ Future<X>(U::*)(ENUM_PARAMS(N, P)), \
+ ENUM_PARAMS(N, A))) \
+ (const PID<U>&, \
+ Future<X>(U::*)(ENUM_PARAMS(N, P)), \
+ ENUM_PARAMS(N, P))>& d) const \
+ { \
+ return then(std::tr1::function<Future<X>(const T&)>(d)); \
+ }
+
+ REPEAT_FROM_TO(1, 11, TEMPLATE, _) // Args A0 -> A9.
+#undef TEMPLATE
#if __cplusplus >= 201103L
template <typename F>
@@ -670,11 +727,11 @@ namespace internal {
template <typename T, typename X>
void thenf(const std::tr1::shared_ptr<Promise<X> >& promise,
- const std::tr1::function<Future<X>(const T&)>& callback,
+ const std::tr1::function<Future<X>(const T&)>& f,
const Future<T>& future)
{
if (future.isReady()) {
- promise->associate(callback(future.get()));
+ promise->associate(f(future.get()));
} else if (future.isFailed()) {
promise->fail(future.failure());
} else if (future.isDiscarded()) {
@@ -685,11 +742,11 @@ void thenf(const std::tr1::shared_ptr<Pr
template <typename T, typename X>
void then(const std::tr1::shared_ptr<Promise<X> >& promise,
- const std::tr1::function<X(const T&)>& callback,
+ const std::tr1::function<X(const T&)>& f,
const Future<T>& future)
{
if (future.isReady()) {
- promise->set(callback(future.get()));
+ promise->set(f(future.get()));
} else if (future.isFailed()) {
promise->fail(future.failure());
} else if (future.isDiscarded()) {
@@ -764,12 +821,12 @@ Future<X> Future<T>::then(
{
std::tr1::shared_ptr<Promise<X> > promise(new Promise<X>());
- std::tr1::function<Future<X>(const T&)> callback = b;
+ std::tr1::function<Future<X>(const T&)> f = b;
std::tr1::function<void(const Future<T>&)> thenf =
std::tr1::bind(&internal::thenf<T, X>,
promise,
- callback,
+ f,
std::tr1::placeholders::_1);
onAny(thenf);
@@ -796,12 +853,12 @@ Future<X> Future<T>::then(
{
std::tr1::shared_ptr<Promise<X> > promise(new Promise<X>());
- std::tr1::function<X(const T&)> callback = b;
+ std::tr1::function<X(const T&)> f = b;
std::tr1::function<void(const Future<T>&)> then =
std::tr1::bind(&internal::then<T, X>,
promise,
- callback,
+ f,
std::tr1::placeholders::_1);
onAny(then);
Modified: incubator/mesos/trunk/third_party/libprocess/src/tests/process_tests.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/third_party/libprocess/src/tests/process_tests.cpp?rev=1449438&r1=1449437&r2=1449438&view=diff
==============================================================================
--- incubator/mesos/trunk/third_party/libprocess/src/tests/process_tests.cpp (original)
+++ incubator/mesos/trunk/third_party/libprocess/src/tests/process_tests.cpp Sun Feb 24 05:05:53 2013
@@ -24,6 +24,7 @@
#include <stout/duration.hpp>
#include <stout/os.hpp>
+#include <stout/stringify.hpp>
#include "encoder.hpp"
@@ -373,42 +374,29 @@ TEST(Process, defer1)
}
-template <typename T>
-void set(T* t1, const T& t2)
-{
- *t1 = t2;
-}
-
-
class DeferProcess : public Process<DeferProcess>
{
public:
- DeferProcess(volatile bool* _bool1, volatile bool* _bool2)
- : bool1(_bool1), bool2(_bool2) {}
-
-protected:
- virtual void initialize()
+ Future<std::string> func1(const Future<int>& f)
{
- deferred<void(bool)> set1 =
- defer(std::tr1::function<void(bool)>(
- std::tr1::bind(&set<volatile bool>,
- bool1,
- std::tr1::placeholders::_1)));
-
- set1(true);
-
- deferred<void(bool)> set2 =
- defer(std::tr1::function<void(bool)>(
- std::tr1::bind(&set<volatile bool>,
- bool2,
- std::tr1::placeholders::_1)));
+ return f.then(defer(self(), &Self::_func1, std::tr1::placeholders::_1));
+ }
- set2(true);
+ Future<std::string> func2(const Future<int>& f)
+ {
+ return f.then(defer(self(), &Self::_func2));
}
private:
- volatile bool* bool1;
- volatile bool* bool2;
+ Future<std::string> _func1(int i)
+ {
+ return stringify(i);
+ }
+
+ Future<std::string> _func2()
+ {
+ return std::string("42");
+ }
};
@@ -416,21 +404,42 @@ TEST(Process, defer2)
{
ASSERT_TRUE(GTEST_IS_THREADSAFE);
- volatile bool bool1 = false;
- volatile bool bool2 = false;
-
- DeferProcess process(&bool1, &bool2);
+ DeferProcess process;
PID<DeferProcess> pid = spawn(process);
- while (!bool1);
- while (!bool2);
+ Future<std::string> f = dispatch(pid, &DeferProcess::func1, 41);
+
+ f.await();
+
+ ASSERT_TRUE(f.isReady());
+ EXPECT_EQ("41", f.get());
+
+ f = dispatch(pid, &DeferProcess::func2, 41);
+
+ f.await();
+
+ ASSERT_TRUE(f.isReady());
+ EXPECT_EQ("42", f.get());
terminate(pid);
wait(pid);
+}
+
+
+template <typename T>
+void set(T* t1, const T& t2)
+{
+ *t1 = t2;
+}
+
- bool1 = false;
- bool2 = false;
+TEST(Process, defer3)
+{
+ ASSERT_TRUE(GTEST_IS_THREADSAFE);
+
+ volatile bool bool1 = false;
+ volatile bool bool2 = false;
deferred<void(bool)> set1 =
defer(std::tr1::function<void(bool)>(