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 2011/06/05 07:24:54 UTC
svn commit: r1131746 - in /incubator/mesos/trunk/src: Makefile.in
third_party/libprocess/process.cpp third_party/libprocess/process.hpp
Author: benh
Date: Sun Jun 5 05:24:54 2011
New Revision: 1131746
URL: http://svn.apache.org/viewvc?rev=1131746&view=rev
Log:
Changes to libprocess to make it have subsecond timers.
Modified:
incubator/mesos/trunk/src/Makefile.in
incubator/mesos/trunk/src/third_party/libprocess/process.cpp
incubator/mesos/trunk/src/third_party/libprocess/process.hpp
Modified: incubator/mesos/trunk/src/Makefile.in
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/Makefile.in?rev=1131746&r1=1131745&r2=1131746&view=diff
==============================================================================
--- incubator/mesos/trunk/src/Makefile.in (original)
+++ incubator/mesos/trunk/src/Makefile.in Sun Jun 5 05:24:54 2011
@@ -24,8 +24,6 @@ LIBEV = third_party/libprocess/third_par
GLOG = third_party/glog-0.3.0
GTEST = third_party/gtest-1.4.0-patched
-ZOOKEEPER = third_party/zookeeper-3.3.1/src/c
-
# Ensure that we get better debugging info.
CFLAGS += -g
CXXFLAGS += -g
@@ -50,13 +48,11 @@ LDFLAGS += -Lthird_party/libprocess
# Add libev to LDFLAGS.
LDFLAGS += -L$(LIBEV)/.libs
-# Add glog and gtest and zookeeper to include and lib paths
+# Add glog and gtest to include and lib paths
CXXFLAGS += -I$(GLOG)/src
CXXFLAGS += -I$(GTEST)/include
-CXXFLAGS += -I$(ZOOKEEPER)/include -I$(ZOOKEEPER)/generated
LDFLAGS += -L$(GLOG)/.libs
LDFLAGS += -L$(GTEST)/lib/.libs
-LDFLAGS += -L$(ZOOKEEPER)/.libs
# Add dependency tracking to CFLAGS, CXXFLAGS.
CFLAGS += -MMD -MP
@@ -71,7 +67,7 @@ CFLAGS += -DBUILD_USER="\"$$USER\""
CXXFLAGS += -DBUILD_USER="\"$$USER\""
# Add libev, libprocess, pthread, and dl to LIBS.
-LIBS += -lglog -lgtest -lprocess -lev -lpthread -ldl -lzookeeper_st
+LIBS += -lglog -lgtest -lprocess -lev -lpthread -ldl
NEXUS_EXES = nexus-master nexus-slave nexus-local nexus-launcher alltests \
test-framework test-executor cpp-test-framework cpp-test-executor \
@@ -92,7 +88,7 @@ NEXUS_LIBS = $(SCHED_LIB) $(EXEC_LIB) $(
MASTER_OBJ = master.o allocator_factory.o simple_allocator.o
SLAVE_OBJ = slave.o launcher.o isolation_module_factory.o \
process_based_isolation_module.o
-COMMON_OBJ = fatal.o hash_pid.o messages.o lock.o master_detector.o url_processor.o ft_messaging.o zookeeper.o
+COMMON_OBJ = fatal.o hash_pid.o lock.o messages.o
EXEC_LIB_OBJ = nexus_exec.o
SCHED_LIB_OBJ = nexus_sched.o nexus_local.o params.o
TEST_OBJ = tests/main.o tests/test_master.o tests/test_resources.o
@@ -133,7 +129,7 @@ $(NEXUS_LIBS): $(COMMON_OBJ) third_party
$(NEXUS_EXES): $(COMMON_OBJ) third_party/libprocess/libprocess.a
$(SCHED_LIB): $(SCHED_LIB_OBJ) $(MASTER_OBJ) $(SLAVE_OBJ) $(COMMON_OBJ)
- $(AR) rcs $@ $^ third_party/libprocess/libprocess.a $(GLOG)/.libs/libglog.a $(LIBEV)/.libs/libev.a $(ZOOKEEPER)/.libs/libzookeeper_st.a
+ $(AR) rcs $@ $^ third_party/libprocess/libprocess.a $(GLOG)/.libs/libglog.a $(LIBEV)/.libs/libev.a
$(EXEC_LIB): $(EXEC_LIB_OBJ) $(COMMON_OBJ)
$(AR) rcs $@ $^
@@ -252,7 +248,6 @@ third_party:
$(MAKE) -C third_party/libprocess
$(MAKE) -C $(GLOG)
$(MAKE) -C $(GTEST)
- $(MAKE) -C $(ZOOKEEPER)
all: third_party $(NEXUS_LIBS) $(NEXUS_EXES) java python ruby
@@ -260,7 +255,6 @@ clean:
$(MAKE) -C third_party/libprocess clean
$(MAKE) -C $(GLOG) clean
$(MAKE) -C $(GTEST) clean
- $(MAKE) -C $(ZOOKEEPER) clean
rm -f $(patsubst %.o, %.d, $(ALL_OBJ))
rm -f $(ALL_OBJ)
rm -f $(NEXUS_LIBS)
@@ -276,11 +270,21 @@ clean:
rm -f swig/java/*.class
rm -f swig/python/nexus.py
rm -f swig/python/nexus.pyc
+ rm -f master_webui.o
+ rm -f master_webui.d
rm -f webui/master/swig/master_wrap.h
rm -f webui/master/swig/master_wrap.cpp
rm -f webui/master/swig/master_wrap.o
+ rm -f webui/master/swig/master_wrap.d
rm -f webui/master/swig/master.py
rm -f webui/master/swig/master.pyc
- rm -f TAGS
+ rm -f slave_webui.o
+ rm -f slave_webui.d
+ rm -f webui/slave/swig/slave_wrap.h
+ rm -f webui/slave/swig/slave_wrap.cpp
+ rm -f webui/slave/swig/slave_wrap.o
+ rm -f webui/slave/swig/slave_wrap.d
+ rm -f webui/slave/swig/slave.py
+ rm -f webui/slave/swig/slave.pyc
.PHONY: default third_party all clean java python ruby test
Modified: incubator/mesos/trunk/src/third_party/libprocess/process.cpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/third_party/libprocess/process.cpp?rev=1131746&r1=1131745&r2=1131746&view=diff
==============================================================================
--- incubator/mesos/trunk/src/third_party/libprocess/process.cpp (original)
+++ incubator/mesos/trunk/src/third_party/libprocess/process.cpp Sun Jun 5 05:24:54 2011
@@ -1085,7 +1085,7 @@ public:
/* Free any pending messages. */
while (!process->msgs.empty()) {
struct msg *msg = process->msgs.front();
- process->msgs.pop_front();
+ process->msgs.pop();
free(msg);
}
@@ -1600,7 +1600,7 @@ public:
return !interrupted;
}
#else
- bool await(Process *process, int fd, int op, ev_tstamp interval, bool ignore)
+ bool await(Process *process, int fd, int op, double secs, bool ignore)
{
assert(process != NULL);
@@ -1633,9 +1633,9 @@ public:
timeout_t timeout;
- if (interval > 0) {
+ if (secs > 0) {
/* Create timeout. */
- timeout = create_timeout(process, interval);
+ timeout = create_timeout(process, secs);
/* Start the timeout. */
start_timeout(timeout);
}
@@ -1658,7 +1658,7 @@ public:
process->state == Process::INTERRUPTED);
/* Attempt to cancel the timer if necessary. */
- if (interval > 0 && process->state != Process::TIMEDOUT)
+ if (secs > 0 && process->state != Process::TIMEDOUT)
cancel_timeout(timeout);
if (process->state == Process::INTERRUPTED)
@@ -1744,10 +1744,10 @@ public:
}
- timeout_t create_timeout(Process *process, ev_tstamp interval)
+ timeout_t create_timeout(Process *process, double secs)
{
assert(process != NULL);
- ev_tstamp tstamp = ev_now(loop) + interval;
+ ev_tstamp tstamp = ev_time() + secs;
return make_tuple(tstamp, process, process->generation);
}
@@ -2863,7 +2863,7 @@ void Process::enqueue(struct msg *msg)
{
if (state != EXITED) {
//cout << "enqueing pending message: " << msg << endl;
- msgs.push_back(msg);
+ msgs.push(msg);
if (state == RECEIVING) {
state = READY;
@@ -2895,7 +2895,7 @@ struct msg * Process::dequeue()
assert (state == RUNNING);
if (!msgs.empty()) {
msg = msgs.front();
- msgs.pop_front();
+ msgs.pop();
//cout << "dequeueing pending message: " << msg << endl;
}
}
@@ -2905,51 +2905,19 @@ struct msg * Process::dequeue()
}
-PID Process::self() const
+PID Process::self()
{
return pid;
}
-PID Process::from() const
+PID Process::from()
{
PID pid = { 0, 0, 0 };
return current != NULL ? current->from : pid;
}
-void Process::inject(const PID &from, MSGID id, const char *data, size_t length)
-{
- if (replaying)
- return;
-
- /* Disallow sending messages using an internal id. */
- if (id < PROCESS_MSGID)
- return;
-
- /* Allocate/Initialize outgoing message. */
- struct msg *msg = (struct msg *) malloc(sizeof(struct msg) + length);
-
- msg->from.pipe = from.pipe;
- msg->from.ip = from.ip;
- msg->from.port = from.port;
- msg->to.pipe = pid.pipe;
- msg->to.ip = pid.ip;
- msg->to.port = pid.port;
- msg->id = id;
- msg->len = length;
-
- if (length > 0)
- memcpy((char *) msg + sizeof(struct msg), data, length);
-
- lock();
- {
- msgs.push_front(msg);
- }
- unlock();
-}
-
-
void Process::send(const PID &to, MSGID id, const char *data, size_t length)
{
//cout << "Process::send" << endl;
@@ -2999,7 +2967,7 @@ MSGID Process::receive(double secs)
{
//cout << "Process::receive(" << secs << ")" << endl;
/* Free current message. */
- if (current) {
+ if (current != NULL) {
free(current);
current = NULL;
}
@@ -3077,7 +3045,7 @@ MSGID Process::call(const PID &to, MSGID
}
-const char * Process::body(size_t *length) const
+const char * Process::body(size_t *length)
{
if (current != NULL && current->len > 0) {
if (length != NULL)
@@ -3124,8 +3092,8 @@ bool Process::await(int fd, int op, cons
bool Process::await(int fd, int op, const timeval& tv, bool ignore)
{
- ev_tstamp interval = tv.tv_sec + (tv.tv_usec * 1e-6);
- return ProcessManager::instance()->await(this, fd, op, interval, ignore);
+ double secs = tv.tv_sec + (tv.tv_usec * 1e-6);
+ return ProcessManager::instance()->await(this, fd, op, secs, ignore);
}
Modified: incubator/mesos/trunk/src/third_party/libprocess/process.hpp
URL: http://svn.apache.org/viewvc/incubator/mesos/trunk/src/third_party/libprocess/process.hpp?rev=1131746&r1=1131745&r2=1131746&view=diff
==============================================================================
--- incubator/mesos/trunk/src/third_party/libprocess/process.hpp (original)
+++ incubator/mesos/trunk/src/third_party/libprocess/process.hpp Sun Jun 5 05:24:54 2011
@@ -111,7 +111,7 @@ private:
EXITED } state;
/* Queue of messages received. */
- std::deque<struct msg *> msgs;
+ std::queue<struct msg *> msgs;
/* Current message. */
struct msg *current;
@@ -158,67 +158,66 @@ protected:
virtual void operator() () = 0;
/* Returns the PID describing this process. */
- PID self() const;
+ PID self();
/* Returns the sender's PID of the last dequeued (current) message. */
- PID from() const;
+ PID from();
- /* Returns the id of the current message. */
- MSGID msgid() const;
-
- /* Returns pointer and length of body of last dequeued (current) message. */
- const char * body(size_t *length) const;
-
- /* Put a message at front of queue (will not reschedule process). */
- virtual void inject(const PID &from, MSGID id, const char *data, size_t length);
+ MSGID msgid();
/* Sends a message to PID. */
- virtual void send(const PID &to , MSGID);
+ void send(const PID &, MSGID);
/* Sends a message with data to PID. */
- virtual void send(const PID &to, MSGID id, const char *data, size_t length);
+ void send(const PID &, MSGID, const char *data, size_t length);
/* Blocks for message indefinitely. */
- virtual MSGID receive();
+ MSGID receive();
/* Blocks for message at most specified seconds. */
- virtual MSGID receive(time_t secs);
+ MSGID receive(double);
/* Sends a message to PID and then blocks for a message indefinitely. */
- virtual MSGID call(const PID &to , MSGID id);
+ MSGID call(const PID &, MSGID);
/* Sends a message with data to PID and then blocks for a message. */
- virtual MSGID call(const PID &to, MSGID id, const char *data, size_t length);
+ MSGID call(const PID &, MSGID, const char *data, size_t length);
/* Sends, and then blocks for a message at most specified seconds. */
- virtual MSGID call(const PID &to, MSGID id, const char *data, size_t length, time_t secs);
+ MSGID call(const PID &, MSGID, const char *data, size_t length, double);
+
+ /* Returns pointer and length of body of last dequeued (current) message. */
+ const char * body(size_t *length);
/* Blocks at least specified seconds (may block longer). */
- virtual void pause(time_t secs);
+ void pause(double);
/* Links with the specified PID. */
- virtual PID link(const PID &pid);
+ PID link(const PID &);
/* IO operations for awaiting. */
enum { RDONLY = 01, WRONLY = 02, RDWR = 03 };
/* Wait until operation is ready for file descriptor (or message received). */
- virtual bool await(int fd, int op);
+ bool await(int fd, int op, const timeval& tv);
+
+ /* Wait until operation is ready for file descriptor (or message received if not ignored). */
+ bool await(int fd, int op, const timeval& tv, bool ignore);
/* Returns true if operation on file descriptor is ready. */
- virtual bool ready(int fd, int op);
+ bool ready(int fd, int op);
public:
virtual ~Process();
/* Returns pid of process; valid even before calling spawn. */
- PID getPID() const;
+ PID getPID();
/* Sends a message to PID without a return address. */
- static void post(const PID &to, MSGID id);
+ static void post(const PID &, MSGID);
/* Sends a message with data to PID without a return address. */
- static void post(const PID &to, MSGID id, const char *data, size_t length);
+ static void post(const PID &, MSGID, const char *data, size_t length);
/* Spawn a new process. */
static PID spawn(Process *process);
@@ -234,7 +233,7 @@ public:
};
-inline MSGID Process::msgid() const
+inline MSGID Process::msgid()
{
return current != NULL ? current->id : PROCESS_ERROR;
}
@@ -255,7 +254,7 @@ inline MSGID Process::call(const PID &to
inline MSGID Process::call(const PID &to, MSGID id,
const char *data, size_t length)
{
- return call(to, id, NULL, 0, 0);
+ return call(to, id, data, length, 0);
}
@@ -265,7 +264,7 @@ inline MSGID Process::receive()
}
-inline PID Process::getPID() const
+inline PID Process::getPID()
{
return self();
}