diff options
Diffstat (limited to 'cpp/src')
| -rw-r--r-- | cpp/src/Makefile.am | 3 | ||||
| -rw-r--r-- | cpp/src/qpid/framing/SerializeHandler.h | 49 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/Serializer.cpp | 125 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/Serializer.h | 105 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/apr/Mutex.h | 18 | ||||
| -rw-r--r-- | cpp/src/qpid/sys/posix/Mutex.h | 12 | ||||
| -rw-r--r-- | cpp/src/tests/Makefile.am | 4 | ||||
| -rw-r--r-- | cpp/src/tests/Serializer.cpp | 147 |
8 files changed, 448 insertions, 15 deletions
diff --git a/cpp/src/Makefile.am b/cpp/src/Makefile.am index db913d601a..09d3f6185d 100644 --- a/cpp/src/Makefile.am +++ b/cpp/src/Makefile.am @@ -168,6 +168,7 @@ libqpidcommon_la_SOURCES = \ 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 @@ nobase_include_HEADERS = \ 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 @@ nobase_include_HEADERS = \ 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 \ diff --git a/cpp/src/qpid/framing/SerializeHandler.h b/cpp/src/qpid/framing/SerializeHandler.h new file mode 100644 index 0000000000..55bd7da08c --- /dev/null +++ b/cpp/src/qpid/framing/SerializeHandler.h @@ -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*/ diff --git a/cpp/src/qpid/sys/Serializer.cpp b/cpp/src/qpid/sys/Serializer.cpp new file mode 100644 index 0000000000..db2b3cab6d --- /dev/null +++ b/cpp/src/qpid/sys/Serializer.cpp @@ -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 diff --git a/cpp/src/qpid/sys/Serializer.h b/cpp/src/qpid/sys/Serializer.h new file mode 100644 index 0000000000..eba8e48555 --- /dev/null +++ b/cpp/src/qpid/sys/Serializer.h @@ -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*/ diff --git a/cpp/src/qpid/sys/apr/Mutex.h b/cpp/src/qpid/sys/apr/Mutex.h index 6679adeebb..51089c98ff 100644 --- a/cpp/src/qpid/sys/apr/Mutex.h +++ b/cpp/src/qpid/sys/apr/Mutex.h @@ -42,7 +42,7 @@ class Mutex : private boost::noncopyable { inline ~Mutex(); inline void lock(); inline void unlock(); - inline void trylock(); + inline bool trylock(); protected: apr_thread_mutex_t* mutex; @@ -64,8 +64,8 @@ void Mutex::unlock() { 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 @@ public: 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 @@ void RWlock::unlock() { 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; } diff --git a/cpp/src/qpid/sys/posix/Mutex.h b/cpp/src/qpid/sys/posix/Mutex.h index b29219235d..4cf0c3a3b0 100644 --- a/cpp/src/qpid/sys/posix/Mutex.h +++ b/cpp/src/qpid/sys/posix/Mutex.h @@ -43,7 +43,7 @@ public: inline ~Mutex(); inline void lock(); inline void unlock(); - inline void trylock(); + inline bool trylock(); protected: @@ -127,7 +127,7 @@ struct PODMutex 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 @@ void PODMutex::unlock() { 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 @@ void Mutex::unlock() { 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; } diff --git a/cpp/src/tests/Makefile.am b/cpp/src/tests/Makefile.am index 64543268ee..69d1d8def1 100644 --- a/cpp/src/tests/Makefile.am +++ b/cpp/src/tests/Makefile.am @@ -46,6 +46,10 @@ check_PROGRAMS+=ConcurrentQueue 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. diff --git a/cpp/src/tests/Serializer.cpp b/cpp/src/tests/Serializer.cpp new file mode 100644 index 0000000000..8c0ee7b85c --- /dev/null +++ b/cpp/src/tests/Serializer.cpp @@ -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()); +} |