Author: aconway
Date: Mon Jul 16 10:52:01 2007
New Revision: 556679
URL: http://svn.apache.org/viewvc?view=rev&rev=556679
Log:
* src/qpid/sys/Serializer.h: Serialize calls to a function,
by queueing and dispatching in a separate thread if necessary.
* src/qpid/sys/*/Mutex.h: trylock to return a bool.
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h (with
props)
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h (with props)
incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp (with props)
Modified:
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/Mutex.h
incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Mutex.h
incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
Modified: incubator/qpid/trunk/qpid/cpp/src/Makefile.am
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/Makefile.am?view=diff&rev=556679&r1=556678&r2=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Mon Jul 16 10:52:01 2007
@@ -168,6 +168,7 @@
qpid/Url.h \
qpid/Url.cpp \
qpid/QpidError.cpp \
+ qpid/sys/Serializer.cpp \
qpid/sys/Runnable.cpp \
qpid/sys/Shlib.h \
qpid/sys/Shlib.cpp \
@@ -351,6 +352,7 @@
qpid/framing/Proxy.h \
qpid/framing/Requester.h \
qpid/framing/Responder.h \
+ qpid/framing/SerializeHandler.h \
qpid/framing/Value.h \
qpid/framing/Uuid.h \
qpid/framing/amqp_framing.h \
@@ -375,6 +377,7 @@
qpid/sys/Socket.h \
qpid/sys/Thread.h \
qpid/sys/ConcurrentQueue.h \
+ qpid/sys/Serializer.h \
qpid/sys/ThreadSafeQueue.h \
qpid/sys/Time.h \
qpid/sys/TimeoutHandler.h \
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h?view=auto&rev=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h Mon Jul
16 10:52:01 2007
@@ -0,0 +1,49 @@
+#ifndef QPID_FRAMING_SERIALIZEHANDLER_H
+#define QPID_FRAMING_SERIALIZEHANDLER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/sys/Serializer.h"
+#include "qpid/framing/Handler.h"
+
+#include <boost/bind.hpp>
+
+namespace qpid {
+namespace framing {
+
+
+/** Serializer that can be inserted into a Handler chain */
+template <class T>
+struct SerializeHandler : public framing::Handler<T>, public sys::Serializer {
+ SerializeHandler(typename framing::Handler<T>::Chain next)
+ : framing::Handler<T>(next) {}
+ void handle(T value) {
+ execute(boost::bind(&framing::Handler<T>::handle, this->next.get(),
value));
+ }
+};
+
+}} // namespace qpid::framing
+
+
+
+
+
+#endif /*!QPID_FRAMING_SERIALIZEHANDLER_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/SerializeHandler.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp?view=auto&rev=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp Mon Jul 16
10:52:01 2007
@@ -0,0 +1,125 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/sys/Serializer.h"
+#include "qpid/log/Statement.h"
+
+#include <boost/bind.hpp>
+
+#include <assert.h>
+
+namespace qpid {
+namespace sys {
+
+Serializer::Serializer(bool allowImmediate, Task notifyDispatchFn)
+ : state(IDLE), immediate(allowImmediate), notifyDispatch(notifyDispatchFn)
+{
+ if (notifyDispatch.empty())
+ notifyDispatch = boost::bind(&Serializer::notifyWorker, this);
+}
+
+Serializer::~Serializer() {
+ {
+ Mutex::ScopedLock l(lock);
+ state = SHUTDOWN;
+ lock.notify();
+ }
+ if (worker.id() != 0)
+ worker.join();
+}
+
+void Serializer::dispatch(Task& task) {
+ Mutex::ScopedUnlock u(lock);
+ // Preconditions: lock is held, state is EXECUTING or DISPATCHING
+ assert(state != IDLE);
+ assert(state != SHUTDOWN);
+ assert(state == EXECUTING || state == DISPATCHING);
+ try {
+ task();
+ } catch (const std::exception& e) {
+ QPID_LOG(error, "Unexpected exception in Serializer::dispatch"
+ << e.what());
+ assert(0); // Should not happen.
+ } catch (...) {
+ QPID_LOG(error, "Unexpected exception in Serializer::dispatch.");
+ assert(0); // Should not happen.
+ }
+}
+
+void Serializer::execute(Task task) {
+ bool needNotify = false;
+ {
+ Mutex::ScopedLock l(lock);
+ assert(state != SHUTDOWN);
+ if (immediate && state == IDLE) {
+ state = EXECUTING;
+ dispatch(task);
+ if (state != SHUTDOWN) {
+ assert(state == EXECUTING);
+ state = IDLE;
+ }
+ }
+ else
+ queue.push_back(task);
+
+ if (!queue.empty() && state == IDLE) {
+ state = DISPATCHING;
+ needNotify = true;
+ }
+ }
+ if (needNotify)
+ notifyDispatch(); // Not my function, call outside lock.
+}
+
+void Serializer::dispatch() {
+ Mutex::ScopedLock l(lock);
+ // TODO aconway 2007-07-16: This loop could be unbounded
+ // if other threads add work while we're in dispatch(Task&).
+ // If we need to bound it we could dispatch just the elements
+ // that were enqueued when dispatch() was first called - save
+ // begin() iterator and pop only up to that.
+ while (!queue.empty() && state != SHUTDOWN) {
+ assert(state == DISPATCHING);
+ dispatch(queue.front());
+ queue.pop_front();
+ }
+ if (state != SHUTDOWN) {
+ assert(state == DISPATCHING);
+ state = IDLE;
+ }
+}
+
+void Serializer::notifyWorker() {
+ if (!worker.id())
+ worker = Thread(*this);
+ else
+ lock.notify();
+}
+
+void Serializer::run() {
+ Mutex::ScopedLock l(lock);
+ while (state != SHUTDOWN) {
+ dispatch();
+ lock.wait();
+ }
+}
+
+}} // namespace qpid::sys
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h?view=auto&rev=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h Mon Jul 16 10:52:01
2007
@@ -0,0 +1,105 @@
+#ifndef SERIALIZER_H
+#define SERIALIZER_H
+
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/Monitor.h"
+#include "qpid/sys/Thread.h"
+
+#include <boost/function.hpp>
+#include <boost/noncopyable.hpp>
+
+#include <deque>
+
+namespace qpid {
+namespace sys {
+
+/**
+ * Execute tasks sequentially, queuing tasks when necessary to
+ * ensure only one thread at a time executes a task and tasks
+ * are executed in order.
+ */
+class Serializer : private boost::noncopyable, private Runnable
+{
+ public:
+ typedef boost::function<void()> Task;
+
+ /** Start a serializer.
+ *
+ * @param notifyDispatch Called when work is pending and there is no
+ * active dispatch thread. Must arrange for dispatch() to be called
+ * in some thread other than the calling thread and return.
+ * By default the Serializer supplies its own dispatch thread.
+ *
+ * @param immediate Allow execute() to execute a task immediatly
+ * in the current thread.
+ */
+ Serializer(bool immediate=true, Task notifyDispatch=Task());
+
+ ~Serializer();
+
+ /**
+ * Task may be executed immediately in the calling thread if there
+ * are no other tasks pending or executing and the "immediate"
+ * paramater to the constructor was true. Otherwise task will be
+ * enqueued for execution by a dispatch thread.
+ */
+ void execute(Task task);
+
+ /** Execute pending tasks sequentially in calling thread.
+ * Drains the task queue and returns, does not block for more tasks.
+ *
+ * @exception ShutdownException if the serializer is being destroyed.
+ */
+ void dispatch();
+
+ private:
+ enum State {
+ IDLE, ///< No threads are active.
+ EXECUTING, ///< execute() is executing a single task.
+ DISPATCHING, ///< dispatch() is draining the queue.
+ SHUTDOWN ///< Serializer is being destroyed.
+ };
+
+ void dispatch(Task&);
+ void notifyWorker();
+ void run();
+
+ Monitor lock;
+
+ State state;
+ bool immediate;
+ std::deque<Task> queue;
+ Thread worker;
+ Task notifyDispatch;
+};
+
+}} // namespace qpid::sys
+
+
+
+
+
+#endif /*!SERIALIZER_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/Serializer.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/Mutex.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/Mutex.h?view=diff&rev=556679&r1=556678&r2=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/Mutex.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/apr/Mutex.h Mon Jul 16 10:52:01
2007
@@ -42,7 +42,7 @@
inline ~Mutex();
inline void lock();
inline void unlock();
- inline void trylock();
+ inline bool trylock();
protected:
apr_thread_mutex_t* mutex;
@@ -64,8 +64,8 @@
CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
}
-void Mutex::trylock() {
- CHECK_APR_SUCCESS(apr_thread_mutex_trylock(mutex));
+bool Mutex::trylock() {
+ return apr_thread_mutex_trylock(mutex) == 0;
}
@@ -84,8 +84,8 @@
inline void wlock(); // will write-lock
inline void rlock(); // will read-lock
inline void unlock();
- inline void trywlock(); // will write-try
- inline void tryrlock(); // will read-try
+ inline bool trywlock(); // will write-try
+ inline bool tryrlock(); // will read-try
protected:
apr_thread_mutex_t* mutex;
@@ -111,12 +111,12 @@
CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex));
}
-void RWlock::trywlock() {
- CHECK_APR_SUCCESS(apr_thread_mutex_trylock(mutex));
+bool RWlock::trywlock() {
+ return apr_thread_mutex_trylock(mutex) == 0;
}
-void RWlock::tryrlock() {
- CHECK_APR_SUCCESS(apr_thread_mutex_trylock(mutex));
+bool RWlock::tryrlock() {
+ return apr_thread_mutex_trylock(mutex) == 0;
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Mutex.h
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Mutex.h?view=diff&rev=556679&r1=556678&r2=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Mutex.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/sys/posix/Mutex.h Mon Jul 16
10:52:01 2007
@@ -43,7 +43,7 @@
inline ~Mutex();
inline void lock();
inline void unlock();
- inline void trylock();
+ inline bool trylock();
protected:
@@ -127,7 +127,7 @@
inline void lock();
inline void unlock();
- inline void trylock();
+ inline bool trylock();
// Must be public to be a POD:
pthread_mutex_t mutex;
@@ -143,8 +143,8 @@
QPID_POSIX_THROW_IF(pthread_mutex_unlock(&mutex));
}
-void PODMutex::trylock() {
- QPID_POSIX_THROW_IF(pthread_mutex_trylock(&mutex));
+bool PODMutex::trylock() {
+ return pthread_mutex_trylock(&mutex) == 0;
}
Mutex::Mutex() {
@@ -163,8 +163,8 @@
QPID_POSIX_THROW_IF(pthread_mutex_unlock(&mutex));
}
-void Mutex::trylock() {
- QPID_POSIX_THROW_IF(pthread_mutex_trylock(&mutex));
+bool Mutex::trylock() {
+ return pthread_mutex_trylock(&mutex) == 0;
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am?view=diff&rev=556679&r1=556678&r2=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Makefile.am Mon Jul 16 10:52:01 2007
@@ -46,6 +46,10 @@
ConcurrentQueue_SOURCES=ConcurrentQueue.cpp
ConcurrentQueue_LDADD=-lboost_test_exec_monitor $(lib_common)
+check_PROGRAMS+=Serializer
+Serializer_SOURCES=Serializer.cpp
+Serializer_LDADD=-lboost_unit_test_framework $(lib_common)
+
include cluster.mk
# NB: CppUnit test libraries below will be migrated to boost test programs.
Added: incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp
URL:
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp?view=auto&rev=556679
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp Mon Jul 16 10:52:01
2007
@@ -0,0 +1,147 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/sys/Runnable.h"
+#include "qpid/sys/Thread.h"
+#include "qpid/sys/Mutex.h"
+#include "qpid/sys/Serializer.h"
+
+#define BOOST_AUTO_TEST_MAIN
+#include <boost/test/auto_unit_test.hpp>
+#include <boost/bind.hpp>
+#include <boost/utility/value_init.hpp>
+
+#include <set>
+
+#include <unistd.h>
+
+using namespace qpid;
+using namespace qpid::sys;
+using namespace qpid::framing;
+using namespace std;
+
+
+/** Test for concurrent calls */
+struct Tester {
+ Monitor lock;
+ size_t count;
+ size_t collisions;
+ set<long> threads;
+
+ Tester() : count(0), collisions(0) {}
+
+ void test() {
+ if (lock.trylock()) { // Check for concurrent calls.
+ ++count;
+ threads.insert(Thread::logId()); // Record thread.
+ usleep(1000); // Encourage overlap.
+ lock.notify();
+ lock.unlock();
+ }
+ else
+ ++collisions;
+ }
+};
+
+BOOST_AUTO_TEST_CASE(testSingleThread) {
+ // Verify that we call in the same thread by default.
+ Tester tester;
+ Serializer s;
+ for (int i = 0; i < 100; ++i)
+ s.execute(boost::bind(&Tester::test, &tester));
+ // All should be executed in this thread.
+ BOOST_CHECK_EQUAL(0u, tester.collisions);
+ BOOST_CHECK_EQUAL(100u, tester.count);
+ BOOST_REQUIRE_EQUAL(1u, tester.threads.size());
+ BOOST_CHECK_EQUAL(Thread::logId(), *tester.threads.begin());
+}
+
+
+BOOST_AUTO_TEST_CASE(testSingleThreadNoImmediate) {
+ // Verify that we call in different thread if immediate=false.
+ Tester tester;
+ Serializer s(false);
+ for (int i = 0; i < 100; ++i)
+ s.execute(boost::bind(&Tester::test, &tester));
+ {
+ // Wait for dispatch thread to complete.
+ Mutex::ScopedLock l(tester.lock);
+ while (tester.count != 100)
+ tester.lock.wait();
+ }
+ BOOST_CHECK_EQUAL(0u, tester.collisions);
+ BOOST_CHECK_EQUAL(100u, tester.count);
+ BOOST_REQUIRE_EQUAL(1u, tester.threads.size());
+ BOOST_CHECK(Thread::logId() != *tester.threads.begin());
+}
+
+struct Caller : public Runnable, public Tester {
+ Caller(Serializer& s) : serializer(s) {}
+ void run() { serializer.execute(boost::bind(&Tester::test, this)); }
+ Serializer& serializer;
+};
+
+BOOST_AUTO_TEST_CASE(testDispatchThread) {
+ Serializer s;
+ Caller caller(s);
+ Thread threads[100];
+ // Concurrent calls.
+ for (size_t i = 0; i < 100; ++i)
+ threads[i] = Thread(caller);
+ for (size_t i = 0; i < 100; ++i)
+ threads[i].join();
+
+ // At least one task should have been queued.
+ BOOST_CHECK_EQUAL(0u, caller.collisions);
+ BOOST_CHECK(caller.threads.size() > 2u);
+ BOOST_CHECK(caller.threads.size() < 100u);
+}
+
+
+std::auto_ptr<Serializer> serializer;
+
+struct CallDispatch : public Runnable {
+ void run() {
+ serializer->dispatch();
+ }
+};
+
+void notifyDispatch() {
+ static CallDispatch cd;
+ Thread t(cd);
+}
+
+// Use externally created threads.
+BOOST_AUTO_TEST_CASE(testExternalDispatch) {
+ serializer.reset(new Serializer(false, ¬ifyDispatch));
+ Tester tester;
+ for (int i = 0; i < 100; ++i)
+ serializer->execute(boost::bind(&Tester::test, &tester));
+ {
+ // Wait for dispatch thread to complete.
+ Mutex::ScopedLock l(tester.lock);
+ while (tester.count != 100)
+ tester.lock.wait();
+ }
+ BOOST_CHECK_EQUAL(0u, tester.collisions);
+ BOOST_CHECK_EQUAL(100u, tester.count);
+ BOOST_CHECK(Thread::logId() != *tester.threads.begin());
+}
Propchange: incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/tests/Serializer.cpp
------------------------------------------------------------------------------
svn:keywords = Rev Date
