diff options
author | Andrew Stitcher <astitcher@apache.org> | 2007-05-22 15:18:08 +0000 |
---|---|---|
committer | Andrew Stitcher <astitcher@apache.org> | 2007-05-22 15:18:08 +0000 |
commit | f646350b5e59ccf49f1253bd55f98d062769f2ee (patch) | |
tree | ba8143aa842ced96eaa450cc236a96abdd8b9c05 /cpp/src/qpid/sys | |
parent | b8f00ac2a358a02d0cdae2dc098f2bacb2af44d5 (diff) | |
download | qpid-python-f646350b5e59ccf49f1253bd55f98d062769f2ee.tar.gz |
* Split apart platform (threading etc.) from network io
you can now use a posix platform implementation by configuring
--disable-apr-platform
* Changed Time classes to distinguish between absolute times (AbsTime)
and durations (Duration). This should avoid bugs caused by confusing
the two types of time.
git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@540608 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'cpp/src/qpid/sys')
31 files changed, 1140 insertions, 589 deletions
diff --git a/cpp/src/qpid/sys/Condition.h b/cpp/src/qpid/sys/Condition.h index 455b179683..961c15e1ee 100644 --- a/cpp/src/qpid/sys/Condition.h +++ b/cpp/src/qpid/sys/Condition.h @@ -22,107 +22,10 @@ * */ -#include <sys/errno.h> -#include <boost/noncopyable.hpp> -#include "Mutex.h" -#include "Time.h" - -#ifdef USE_APR -#include <apr_thread_cond.h> -#endif - -namespace qpid { -namespace sys { - -/** - * A condition variable for thread synchronization. - */ -class Condition -{ - public: - inline Condition(); - inline ~Condition(); - inline void wait(Mutex&); - inline bool wait(Mutex&, const Time& absoluteTime); - inline void notify(); - inline void notifyAll(); - - private: -#ifdef USE_APR - apr_thread_cond_t* condition; +#ifdef USE_APR_PLATFORM +#include "apr/Condition.h" #else - pthread_cond_t condition; +#include "posix/Condition.h" #endif -}; - - -// APR ================================================================ -#ifdef USE_APR - -Condition::Condition() { - CHECK_APR_SUCCESS(apr_thread_cond_create(&condition, APRPool::get())); -} - -Condition::~Condition() { - CHECK_APR_SUCCESS(apr_thread_cond_destroy(condition)); -} - -void Condition::wait(Mutex& mutex) { - CHECK_APR_SUCCESS(apr_thread_cond_wait(condition, mutex.mutex)); -} - -bool Condition::wait(Mutex& mutex, const Time& absoluteTime){ - // APR uses microseconds. - apr_status_t status = - apr_thread_cond_timedwait( - condition, mutex.mutex, absoluteTime/TIME_USEC); - if(status != APR_TIMEUP) CHECK_APR_SUCCESS(status); - return status == 0; -} - -void Condition::notify(){ - CHECK_APR_SUCCESS(apr_thread_cond_signal(condition)); -} - -void Condition::notifyAll(){ - CHECK_APR_SUCCESS(apr_thread_cond_broadcast(condition)); -} - -#else -// POSIX ================================================================ - -Condition::Condition() { - QPID_POSIX_THROW_IF(pthread_cond_init(&condition, 0)); -} - -Condition::~Condition() { - QPID_POSIX_THROW_IF(pthread_cond_destroy(&condition)); -} - -void Condition::wait(Mutex& mutex) { - QPID_POSIX_THROW_IF(pthread_cond_wait(&condition, &mutex.mutex)); -} - -bool Condition::wait(Mutex& mutex, const Time& absoluteTime){ - struct timespec ts; - toTimespec(ts, absoluteTime); - int status = pthread_cond_timedwait(&condition, &mutex.mutex, &ts); - if (status != 0) { - if (status == ETIMEDOUT) return false; - throw QPID_POSIX_ERROR(status); - } - return true; -} - -void Condition::notify(){ - QPID_POSIX_THROW_IF(pthread_cond_signal(&condition)); -} - -void Condition::notifyAll(){ - QPID_POSIX_THROW_IF(pthread_cond_broadcast(&condition)); -} -#endif /*USE_APR*/ - -}} #endif /*!_sys_Condition_h*/ diff --git a/cpp/src/qpid/sys/Module.h b/cpp/src/qpid/sys/Module.h index cb31018fdc..89cea36a74 100644 --- a/cpp/src/qpid/sys/Module.h +++ b/cpp/src/qpid/sys/Module.h @@ -21,141 +21,12 @@ * under the License. * */ -#include <boost/noncopyable.hpp> -#include <iostream> -#include "qpid/QpidError.h" - -namespace qpid { -namespace sys { -#if USE_APR -#include <apr_dso.h> - typedef apr_dso_handle_t* dso_handle_t; + +#ifdef USE_APR_PLATFORM +#include "apr/Module.h" #else - typedef void* dso_handle_t; +#include "posix/Module.h" #endif - template <class T> class Module : private boost::noncopyable - { - typedef T* create_t(); - typedef void destroy_t(T*); - - dso_handle_t handle; - destroy_t* destroy; - T* ptr; - - void load(const std::string& name); - void unload(); - void* getSymbol(const std::string& name); - - public: - Module(const std::string& name); - T* operator->(); - T* get(); - ~Module() throw(); - }; - -} -} - -using namespace qpid::sys; - -template <class T> Module<T>::Module(const std::string& module) : destroy(0), ptr(0) -{ - load(module); - //TODO: need a better strategy for symbol names to allow multiple - //modules to be loaded without clashes... - - //Note: need the double cast to avoid errors in casting from void* to function pointer with -pedantic - create_t* create = reinterpret_cast<create_t*>(reinterpret_cast<intptr_t>(getSymbol("create"))); - destroy = reinterpret_cast<destroy_t*>(reinterpret_cast<intptr_t>(getSymbol("destroy"))); - ptr = create(); -} - -template <class T> T* Module<T>::operator->() -{ - return ptr; -} - -template <class T> T* Module<T>::get() -{ - return ptr; -} - -template <class T> Module<T>::~Module() throw() -{ - try { - if (handle && ptr) { - destroy(ptr); - } - if (handle) unload(); - } catch (std::exception& e) { - std::cout << "Error while destroying module: " << e.what() << std::endl; - } - destroy = 0; - handle = 0; - ptr = 0; -} - -// APR ================================================================ -#if USE_APR - -#include "apr/APRBase.h" -#include "apr/APRPool.h" - -template <class T> void Module<T>::load(const std::string& name) -{ - CHECK_APR_SUCCESS(apr_dso_load(&handle, name.c_str(), APRPool::get())); -} - -template <class T> void Module<T>::unload() -{ - CHECK_APR_SUCCESS(apr_dso_unload(handle)); -} - -template <class T> void* Module<T>::getSymbol(const std::string& name) -{ - apr_dso_handle_sym_t symbol; - CHECK_APR_SUCCESS(apr_dso_sym(&symbol, handle, name.c_str())); - return (void*) symbol; -} - -// POSIX================================================================ -#else - -#include <dlfcn.h> - -template <class T> void Module<T>::load(const std::string& name) -{ - dlerror(); - handle = dlopen(name.c_str(), RTLD_NOW); - const char* error = dlerror(); - if (error) { - THROW_QPID_ERROR(INTERNAL_ERROR, error); - } -} - -template <class T> void Module<T>::unload() -{ - dlerror(); - dlclose(handle); - const char* error = dlerror(); - if (error) { - THROW_QPID_ERROR(INTERNAL_ERROR, error); - } -} - -template <class T> void* Module<T>::getSymbol(const std::string& name) -{ - dlerror(); - void* sym = dlsym(handle, name.c_str()); - const char* error = dlerror(); - if (error) { - THROW_QPID_ERROR(INTERNAL_ERROR, error); - } - return sym; -} - -#endif //if USE_APR - #endif //ifndef _sys_Module_h diff --git a/cpp/src/qpid/sys/Monitor.h b/cpp/src/qpid/sys/Monitor.h index 1b8ae1a527..1d9835675c 100644 --- a/cpp/src/qpid/sys/Monitor.h +++ b/cpp/src/qpid/sys/Monitor.h @@ -25,10 +25,6 @@ #include <sys/errno.h> #include "Condition.h" -#ifdef USE_APR -#include <apr_thread_cond.h> -#endif - namespace qpid { namespace sys { @@ -39,7 +35,7 @@ class Monitor : public Mutex, public Condition { public: using Condition::wait; inline void wait(); - inline bool wait(const Time& absoluteTime); + inline bool wait(const AbsTime& absoluteTime); }; @@ -47,7 +43,7 @@ void Monitor::wait() { Condition::wait(*this); } -bool Monitor::wait(const Time& absoluteTime) { +bool Monitor::wait(const AbsTime& absoluteTime) { return Condition::wait(*this, absoluteTime); } diff --git a/cpp/src/qpid/sys/Mutex.h b/cpp/src/qpid/sys/Mutex.h index 825b519039..4eff6078ae 100644 --- a/cpp/src/qpid/sys/Mutex.h +++ b/cpp/src/qpid/sys/Mutex.h @@ -19,21 +19,9 @@ * */ -#ifdef USE_APR -#include <apr_thread_mutex.h> -#include "apr/APRBase.h" -#include "apr/APRPool.h" -#else -#include <pthread.h> -#include "posix/check.h" -#endif -#include <boost/noncopyable.hpp> - namespace qpid { namespace sys { -class Condition; - /** * Scoped lock template: calls lock() in ctor, unlock() in dtor. * L can be any class with lock() and unlock() functions. @@ -57,109 +45,13 @@ class ScopedUnlock private: L& mutex; }; - -/** - * Mutex lock. - */ -class Mutex : private boost::noncopyable { - public: - typedef ScopedLock<Mutex> ScopedLock; - typedef ScopedUnlock<Mutex> ScopedUnlock; - - inline Mutex(); - inline ~Mutex(); - inline void lock(); - inline void unlock(); - inline void trylock(); - protected: -#ifdef USE_APR - apr_thread_mutex_t* mutex; +}} + +#ifdef USE_APR_PLATFORM +#include "apr/Mutex.h" #else - pthread_mutex_t mutex; +#include "posix/Mutex.h" #endif - friend class Condition; -}; - -#ifdef USE_APR -// APR ================================================================ - -Mutex::Mutex() { - CHECK_APR_SUCCESS(apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_NESTED, APRPool::get())); -} - -Mutex::~Mutex(){ - CHECK_APR_SUCCESS(apr_thread_mutex_destroy(mutex)); -} - -void Mutex::lock() { - CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex)); -} -void Mutex::unlock() { - CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex)); -} - -void Mutex::trylock() { - CHECK_APR_SUCCESS(apr_thread_mutex_trylock(mutex)); -} - -#else -// POSIX ================================================================ - -/** - * PODMutex is a POD, can be static-initialized with - * PODMutex m = QPID_PODMUTEX_INITIALIZER - */ -struct PODMutex -{ - typedef ScopedLock<PODMutex> ScopedLock; - - inline void lock(); - inline void unlock(); - inline void trylock(); - - // Must be public to be a POD: - pthread_mutex_t mutex; -}; - -#define QPID_MUTEX_INITIALIZER { PTHREAD_MUTEX_INITIALIZER } - - -void PODMutex::lock() { - QPID_POSIX_THROW_IF(pthread_mutex_lock(&mutex)); -} -void PODMutex::unlock() { - QPID_POSIX_THROW_IF(pthread_mutex_unlock(&mutex)); -} - -void PODMutex::trylock() { - QPID_POSIX_THROW_IF(pthread_mutex_trylock(&mutex)); -} - - -Mutex::Mutex() { - QPID_POSIX_THROW_IF(pthread_mutex_init(&mutex, 0)); -} - -Mutex::~Mutex(){ - QPID_POSIX_THROW_IF(pthread_mutex_destroy(&mutex)); -} - -void Mutex::lock() { - QPID_POSIX_THROW_IF(pthread_mutex_lock(&mutex)); -} -void Mutex::unlock() { - QPID_POSIX_THROW_IF(pthread_mutex_unlock(&mutex)); -} - -void Mutex::trylock() { - QPID_POSIX_THROW_IF(pthread_mutex_trylock(&mutex)); -} - -#endif // USE_APR - -}} - - #endif /*!_sys_Mutex_h*/ diff --git a/cpp/src/qpid/sys/ProducerConsumer.cpp b/cpp/src/qpid/sys/ProducerConsumer.cpp index 56b4c0cdb9..e892f60794 100644 --- a/cpp/src/qpid/sys/ProducerConsumer.cpp +++ b/cpp/src/qpid/sys/ProducerConsumer.cpp @@ -103,7 +103,7 @@ ProducerConsumer::ConsumerLock::ConsumerLock(ProducerConsumer& p) : Lock(p) } ProducerConsumer::ConsumerLock::ConsumerLock( - ProducerConsumer& p, const Time& timeout) : Lock(p) + ProducerConsumer& p, const Duration& timeout) : Lock(p) { if (isOk()) { // Don't wait if timeout==0 @@ -113,7 +113,7 @@ ProducerConsumer::ConsumerLock::ConsumerLock( return; } else { - Time deadline = now() + timeout; + AbsTime deadline(now(), timeout); ScopedIncrement<size_t> inc(pc.waiters); while (pc.items == 0 && !pc.shutdownFlag) { if (!pc.monitor.wait(deadline)) { diff --git a/cpp/src/qpid/sys/ProducerConsumer.h b/cpp/src/qpid/sys/ProducerConsumer.h index fc6434ef48..2a02dab503 100644 --- a/cpp/src/qpid/sys/ProducerConsumer.h +++ b/cpp/src/qpid/sys/ProducerConsumer.h @@ -143,7 +143,7 @@ class ProducerConsumer * If isTimedOut() there was a timeout. * If neither then we were shutdown. */ - ConsumerLock(ProducerConsumer& p, const Time& timeout); + ConsumerLock(ProducerConsumer& p, const Duration& timeout); /** Release locks */ ~ConsumerLock(); diff --git a/cpp/src/qpid/sys/Socket.h b/cpp/src/qpid/sys/Socket.h index ea03222aed..bb1ef27e65 100644 --- a/cpp/src/qpid/sys/Socket.h +++ b/cpp/src/qpid/sys/Socket.h @@ -22,67 +22,10 @@ * */ -#include <string> -#include "Time.h" - -#ifdef USE_APR -#include <apr_network_io.h> -#endif - -namespace qpid { -namespace sys { - -class Socket -{ - public: - /** Create an initialized TCP socket */ - static Socket createTcp(); - - /** Create a socket wrapper for descriptor. */ -#ifdef USE_APR - Socket(apr_socket_t* descriptor = 0); -#else - Socket(int descriptor = 0); -#endif - - /** Set timeout for read and write */ - void setTimeout(Time interval); - - void connect(const std::string& host, int port); - - void close(); - - enum { SOCKET_TIMEOUT=-2, SOCKET_EOF=-3 } ErrorCode; - - /** Returns bytes sent or an ErrorCode value < 0. */ - ssize_t send(const void* data, size_t size); - - /** - * Returns bytes received, an ErrorCode value < 0 or 0 - * if the connection closed in an orderly manner. - */ - ssize_t recv(void* data, size_t size); - - /** Bind to a port and start listening. - *@param port 0 means choose an available port. - *@param backlog maximum number of pending connections. - *@return The bound port. - */ - int listen(int port = 0, int backlog = 10); - - /** Get file descriptor */ - int fd(); - - private: -#ifdef USE_APR - apr_socket_t* socket; +#ifdef USE_APR_PLATFORM +#include "apr/Socket.h" #else - void init() const; - mutable int socket; // Initialized on demand. +#include "posix/Socket.h" #endif -}; - -}} - #endif /*!_sys_Socket_h*/ diff --git a/cpp/src/qpid/sys/Thread.h b/cpp/src/qpid/sys/Thread.h index e52f2a1b3e..fd9be5617e 100644 --- a/cpp/src/qpid/sys/Thread.h +++ b/cpp/src/qpid/sys/Thread.h @@ -22,121 +22,10 @@ * */ -#include "Runnable.h" - -#ifdef USE_APR -#include <apr_thread_proc.h> -#include <apr_portable.h> -#include "apr/APRPool.h" -#include "apr/APRBase.h" -#else -#include "posix/check.h" -#include <pthread.h> -#endif - -namespace qpid { -namespace sys { - -class Thread -{ - public: - inline static Thread current(); - inline static void yield(); - - inline Thread(); - inline explicit Thread(qpid::sys::Runnable*); - inline explicit Thread(qpid::sys::Runnable&); - - inline void join(); - - inline long id(); - - private: -#ifdef USE_APR - static void* APR_THREAD_FUNC runRunnable(apr_thread_t* thread, void *data); - inline Thread(apr_thread_t* t); - apr_thread_t* thread; -#else - static void* runRunnable(void* runnable); - inline Thread(pthread_t); - pthread_t thread; -#endif -}; - - -Thread::Thread() : thread(0) {} - -// APR ================================================================ -#ifdef USE_APR - -Thread::Thread(Runnable* runnable) { - CHECK_APR_SUCCESS( - apr_thread_create(&thread, 0, runRunnable, runnable, APRPool::get())); -} - -Thread::Thread(Runnable& runnable) { - CHECK_APR_SUCCESS( - apr_thread_create(&thread, 0, runRunnable, &runnable, APRPool::get())); -} - -void Thread::join(){ - apr_status_t status; - if (thread != 0) - CHECK_APR_SUCCESS(apr_thread_join(&status, thread)); -} - -long Thread::id() { - return long(thread); -} - -Thread::Thread(apr_thread_t* t) : thread(t) {} - -Thread Thread::current(){ - apr_thread_t* thr; - apr_os_thread_t osthr = apr_os_thread_current(); - CHECK_APR_SUCCESS(apr_os_thread_put(&thr, &osthr, APRPool::get())); - return Thread(thr); -} - -void Thread::yield() -{ - apr_thread_yield(); -} - - -// POSIX ================================================================ +#ifdef USE_APR_PLATFORM +#include "apr/Thread.h" #else - -Thread::Thread(Runnable* runnable) { - QPID_POSIX_THROW_IF(pthread_create(&thread, NULL, runRunnable, runnable)); -} - -Thread::Thread(Runnable& runnable) { - QPID_POSIX_THROW_IF(pthread_create(&thread, NULL, runRunnable, &runnable)); -} - -void Thread::join(){ - QPID_POSIX_THROW_IF(pthread_join(thread, 0)); -} - -long Thread::id() { - return long(thread); -} - -Thread::Thread(pthread_t thr) : thread(thr) {} - -Thread Thread::current() { - return Thread(pthread_self()); -} - -void Thread::yield() -{ - QPID_POSIX_THROW_IF(pthread_yield()); -} - - +#include "posix/Thread.h" #endif -}} - #endif /*!_sys_Thread_h*/ diff --git a/cpp/src/qpid/sys/Time.h b/cpp/src/qpid/sys/Time.h index 47609d51df..314f4b5bbb 100644 --- a/cpp/src/qpid/sys/Time.h +++ b/cpp/src/qpid/sys/Time.h @@ -23,36 +23,76 @@ */ #include <stdint.h> - -#ifdef USE_APR -#include <apr_time.h> -#else -#include <time.h> -#endif +#include <limits> namespace qpid { namespace sys { -/** Time in nanoseconds */ -typedef int64_t Time; +class Duration; + +/** Times in nanoseconds */ +class AbsTime { + int64_t time_ns; + + friend class Duration; + +public: + inline AbsTime() {} + inline AbsTime(const AbsTime& time0, const Duration& duration); + // Default asignment operation fine + // Default copy constructor fine + + static AbsTime now(); + inline static AbsTime FarFuture(); +}; + +class Duration { + int64_t nanosecs; + + friend class AbsTime; + +public: + inline Duration(int64_t time0); + inline explicit Duration(const AbsTime& time0); + inline explicit Duration(const AbsTime& start, const AbsTime& finish); + inline operator int64_t() const; +}; + -Time now(); +AbsTime::AbsTime(const AbsTime& time0, const Duration& duration0) : + time_ns(time0.time_ns+duration0.nanosecs) +{} + +AbsTime AbsTime::FarFuture() { AbsTime ff; ff.time_ns = std::numeric_limits<int64_t>::max(); return ff;} + +inline AbsTime now() { return AbsTime::now(); } + +Duration::Duration(int64_t time0) : + nanosecs(time0) +{} + +Duration::Duration(const AbsTime& time0) : + nanosecs(time0.time_ns) +{} + +Duration::Duration(const AbsTime& start, const AbsTime& finish) : + nanosecs(finish.time_ns - start.time_ns) +{} + +Duration::operator int64_t() const +{ return nanosecs; } /** Nanoseconds per second. */ -const Time TIME_SEC = 1000*1000*1000; +const Duration TIME_SEC = 1000*1000*1000; /** Nanoseconds per millisecond */ -const Time TIME_MSEC = 1000*1000; +const Duration TIME_MSEC = 1000*1000; /** Nanoseconds per microseconds. */ -const Time TIME_USEC = 1000; +const Duration TIME_USEC = 1000; /** Nanoseconds per nanosecond. */ -const Time TIME_NSEC = 1; - -#ifndef USE_APR -struct timespec toTimespec(const Time& t); -struct timespec& toTimespec(struct timespec& ts, const Time& t); -Time toTime(const struct timespec& ts); -#endif +const Duration TIME_NSEC = 1; +/** Time greater than any other time */ +const AbsTime FAR_FUTURE = AbsTime::FarFuture(); }} #endif /*!_sys_Time_h*/ diff --git a/cpp/src/qpid/sys/apr/APRAcceptor.cpp b/cpp/src/qpid/sys/apr/APRAcceptor.cpp index dc021d2a3f..e9ce24ac2d 100644 --- a/cpp/src/qpid/sys/apr/APRAcceptor.cpp +++ b/cpp/src/qpid/sys/apr/APRAcceptor.cpp @@ -20,6 +20,7 @@ */ #include "qpid/sys/Acceptor.h" #include "qpid/sys/ConnectionInputHandlerFactory.h" +#include "qpid/sys/Mutex.h" #include "LFProcessor.h" #include "LFSessionContext.h" #include "APRBase.h" diff --git a/cpp/src/qpid/sys/apr/Condition.h b/cpp/src/qpid/sys/apr/Condition.h new file mode 100644 index 0000000000..5e544219ab --- /dev/null +++ b/cpp/src/qpid/sys/apr/Condition.h @@ -0,0 +1,84 @@ +#ifndef _sys_apr_Condition_h +#define _sys_apr_Condition_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 "APRPool.h" +#include "qpid/sys/Mutex.h" +#include "qpid/sys/Time.h" + +#include <sys/errno.h> +#include <boost/noncopyable.hpp> +#include <apr_thread_cond.h> + +namespace qpid { +namespace sys { + +/** + * A condition variable for thread synchronization. + */ +class Condition +{ + public: + inline Condition(); + inline ~Condition(); + inline void wait(Mutex&); + inline bool wait(Mutex&, const AbsTime& absoluteTime); + inline void notify(); + inline void notifyAll(); + + private: + apr_thread_cond_t* condition; +}; + + +Condition::Condition() { + CHECK_APR_SUCCESS(apr_thread_cond_create(&condition, APRPool::get())); +} + +Condition::~Condition() { + CHECK_APR_SUCCESS(apr_thread_cond_destroy(condition)); +} + +void Condition::wait(Mutex& mutex) { + CHECK_APR_SUCCESS(apr_thread_cond_wait(condition, mutex.mutex)); +} + +bool Condition::wait(Mutex& mutex, const AbsTime& absoluteTime){ + // APR uses microseconds. + apr_status_t status = + apr_thread_cond_timedwait( + condition, mutex.mutex, Duration(now(), absoluteTime)/TIME_USEC); + if(status != APR_TIMEUP) CHECK_APR_SUCCESS(status); + return status == 0; +} + +void Condition::notify(){ + CHECK_APR_SUCCESS(apr_thread_cond_signal(condition)); +} + +void Condition::notifyAll(){ + CHECK_APR_SUCCESS(apr_thread_cond_broadcast(condition)); +} + +}} +#endif /*!_sys_apr_Condition_h*/ diff --git a/cpp/src/qpid/sys/apr/LFProcessor.cpp b/cpp/src/qpid/sys/apr/LFProcessor.cpp index 0d8ac425fe..9e139c874c 100644 --- a/cpp/src/qpid/sys/apr/LFProcessor.cpp +++ b/cpp/src/qpid/sys/apr/LFProcessor.cpp @@ -20,6 +20,7 @@ */ #include <sstream> #include "qpid/QpidError.h" +#include "qpid/sys/Mutex.h" #include "LFProcessor.h" #include "APRBase.h" #include "LFSessionContext.h" diff --git a/cpp/src/qpid/sys/apr/LFSessionContext.h b/cpp/src/qpid/sys/apr/LFSessionContext.h index ed97b23645..3c90c4a381 100644 --- a/cpp/src/qpid/sys/apr/LFSessionContext.h +++ b/cpp/src/qpid/sys/apr/LFSessionContext.h @@ -30,6 +30,7 @@ #include "qpid/framing/AMQFrame.h" #include "qpid/framing/Buffer.h" #include "qpid/sys/Monitor.h" +#include "qpid/sys/Mutex.h" #include "qpid/sys/ConnectionOutputHandler.h" #include "qpid/sys/ConnectionInputHandler.h" diff --git a/cpp/src/qpid/sys/apr/Module.h b/cpp/src/qpid/sys/apr/Module.h new file mode 100644 index 0000000000..d77cc0f388 --- /dev/null +++ b/cpp/src/qpid/sys/apr/Module.h @@ -0,0 +1,114 @@ +#ifndef _sys_apr_Module_h +#define _sys_apr_Module_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/QpidError.h" +#include "APRBase.h" +#include "APRPool.h" + +#include <boost/noncopyable.hpp> +#include <iostream> +#include <apr_dso.h> + +namespace qpid { +namespace sys { + +typedef apr_dso_handle_t* dso_handle_t; + +template <class T> class Module : private boost::noncopyable +{ + typedef T* create_t(); + typedef void destroy_t(T*); + + dso_handle_t handle; + destroy_t* destroy; + T* ptr; + + void load(const std::string& name); + void unload(); + void* getSymbol(const std::string& name); + +public: + Module(const std::string& name); + T* operator->(); + T* get(); + ~Module() throw(); +}; + +template <class T> Module<T>::Module(const std::string& module) : destroy(0), ptr(0) +{ + load(module); + //TODO: need a better strategy for symbol names to allow multiple + //modules to be loaded without clashes... + + //Note: need the double cast to avoid errors in casting from void* to function pointer with -pedantic + create_t* create = reinterpret_cast<create_t*>(reinterpret_cast<intptr_t>(getSymbol("create"))); + destroy = reinterpret_cast<destroy_t*>(reinterpret_cast<intptr_t>(getSymbol("destroy"))); + ptr = create(); +} + +template <class T> T* Module<T>::operator->() +{ + return ptr; +} + +template <class T> T* Module<T>::get() +{ + return ptr; +} + +template <class T> Module<T>::~Module() throw() +{ + try { + if (handle && ptr) { + destroy(ptr); + } + if (handle) unload(); + } catch (std::exception& e) { + std::cout << "Error while destroying module: " << e.what() << std::endl; + } + destroy = 0; + handle = 0; + ptr = 0; +} + +template <class T> void Module<T>::load(const std::string& name) +{ + CHECK_APR_SUCCESS(apr_dso_load(&handle, name.c_str(), APRPool::get())); +} + +template <class T> void Module<T>::unload() +{ + CHECK_APR_SUCCESS(apr_dso_unload(handle)); +} + +template <class T> void* Module<T>::getSymbol(const std::string& name) +{ + apr_dso_handle_sym_t symbol; + CHECK_APR_SUCCESS(apr_dso_sym(&symbol, handle, name.c_str())); + return (void*) symbol; +} + +}} +#endif //ifndef _sys_apr_Module_h + diff --git a/cpp/src/qpid/sys/apr/Mutex.h b/cpp/src/qpid/sys/apr/Mutex.h new file mode 100644 index 0000000000..700b5b910b --- /dev/null +++ b/cpp/src/qpid/sys/apr/Mutex.h @@ -0,0 +1,72 @@ +#ifndef _sys_apr_Mutex_h +#define _sys_apr_Mutex_h + +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed 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 "APRBase.h" +#include "APRPool.h" + +#include <boost/noncopyable.hpp> +#include <apr_thread_mutex.h> + +namespace qpid { +namespace sys { + +class Condition; + +/** + * Mutex lock. + */ +class Mutex : private boost::noncopyable { + public: + typedef ScopedLock<Mutex> ScopedLock; + typedef ScopedUnlock<Mutex> ScopedUnlock; + + inline Mutex(); + inline ~Mutex(); + inline void lock(); + inline void unlock(); + inline void trylock(); + + protected: + apr_thread_mutex_t* mutex; + friend class Condition; +}; + +Mutex::Mutex() { + CHECK_APR_SUCCESS(apr_thread_mutex_create(&mutex, APR_THREAD_MUTEX_NESTED, APRPool::get())); +} + +Mutex::~Mutex(){ + CHECK_APR_SUCCESS(apr_thread_mutex_destroy(mutex)); +} + +void Mutex::lock() { + CHECK_APR_SUCCESS(apr_thread_mutex_lock(mutex)); +} +void Mutex::unlock() { + CHECK_APR_SUCCESS(apr_thread_mutex_unlock(mutex)); +} + +void Mutex::trylock() { + CHECK_APR_SUCCESS(apr_thread_mutex_trylock(mutex)); +} + +}} +#endif /*!_sys_apr_Mutex_h*/ diff --git a/cpp/src/qpid/sys/apr/Socket.cpp b/cpp/src/qpid/sys/apr/Socket.cpp index 8cd98161d2..6e64d656d2 100644 --- a/cpp/src/qpid/sys/apr/Socket.cpp +++ b/cpp/src/qpid/sys/apr/Socket.cpp @@ -20,7 +20,7 @@ */ -#include "qpid/sys/Socket.h" +#include "Socket.h" #include "APRBase.h" #include "APRPool.h" @@ -40,7 +40,7 @@ Socket::Socket(apr_socket_t* s) { socket = s; } -void Socket::setTimeout(Time interval) { +void Socket::setTimeout(const Duration& interval) { apr_socket_timeout_set(socket, interval/TIME_USEC); } diff --git a/cpp/src/qpid/sys/apr/Socket.h b/cpp/src/qpid/sys/apr/Socket.h new file mode 100644 index 0000000000..c20c36dcd9 --- /dev/null +++ b/cpp/src/qpid/sys/apr/Socket.h @@ -0,0 +1,75 @@ +#ifndef _sys_apr_Socket_h +#define _sys_apr_Socket_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 <string> +#include "qpid/sys/Time.h" + +#include <apr_network_io.h> + +namespace qpid { +namespace sys { + +class Socket +{ + public: + /** Create an initialized TCP socket */ + static Socket createTcp(); + + /** Create a socket wrapper for descriptor. */ + Socket(apr_socket_t* descriptor = 0); + + /** Set timeout for read and write */ + void setTimeout(const Duration& interval); + + void connect(const std::string& host, int port); + + void close(); + + enum { SOCKET_TIMEOUT=-2, SOCKET_EOF=-3 } ErrorCode; + + /** Returns bytes sent or an ErrorCode value < 0. */ + ssize_t send(const void* data, size_t size); + + /** + * Returns bytes received, an ErrorCode value < 0 or 0 + * if the connection closed in an orderly manner. + */ + ssize_t recv(void* data, size_t size); + + /** Bind to a port and start listening. + *@param port 0 means choose an available port. + *@param backlog maximum number of pending connections. + *@return The bound port. + */ + int listen(int port = 0, int backlog = 10); + + /** Get file descriptor */ + int fd(); + + private: + apr_socket_t* socket; +}; + +}} +#endif /*!_sys_Socket_h*/ diff --git a/cpp/src/qpid/sys/apr/Thread.cpp b/cpp/src/qpid/sys/apr/Thread.cpp index 0f9aeca186..3369ef7eb1 100644 --- a/cpp/src/qpid/sys/apr/Thread.cpp +++ b/cpp/src/qpid/sys/apr/Thread.cpp @@ -19,7 +19,8 @@ * */ -#include "qpid/sys/Thread.h" +#include "Thread.h" +#include "qpid/sys/Runnable.h" using namespace qpid::sys; using qpid::sys::Runnable; diff --git a/cpp/src/qpid/sys/apr/Thread.h b/cpp/src/qpid/sys/apr/Thread.h new file mode 100644 index 0000000000..ce876efbdf --- /dev/null +++ b/cpp/src/qpid/sys/apr/Thread.h @@ -0,0 +1,93 @@ +#ifndef _sys_apr_Thread_h +#define _sys_apr_Thread_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 "APRPool.h" +#include "APRBase.h" + +#include <apr_thread_proc.h> +#include <apr_portable.h> + +namespace qpid { +namespace sys { + +class Runnable; + +class Thread +{ + public: + inline static Thread current(); + inline static void yield(); + + inline Thread(); + inline explicit Thread(qpid::sys::Runnable*); + inline explicit Thread(qpid::sys::Runnable&); + + inline void join(); + + inline long id(); + + private: + static void* APR_THREAD_FUNC runRunnable(apr_thread_t* thread, void *data); + inline Thread(apr_thread_t* t); + apr_thread_t* thread; +}; + +Thread::Thread() : thread(0) {} + +Thread::Thread(Runnable* runnable) { + CHECK_APR_SUCCESS( + apr_thread_create(&thread, 0, runRunnable, runnable, APRPool::get())); +} + +Thread::Thread(Runnable& runnable) { + CHECK_APR_SUCCESS( + apr_thread_create(&thread, 0, runRunnable, &runnable, APRPool::get())); +} + +void Thread::join(){ + apr_status_t status; + if (thread != 0) + CHECK_APR_SUCCESS(apr_thread_join(&status, thread)); +} + +long Thread::id() { + return long(thread); +} + +Thread::Thread(apr_thread_t* t) : thread(t) {} + +Thread Thread::current(){ + apr_thread_t* thr; + apr_os_thread_t osthr = apr_os_thread_current(); + CHECK_APR_SUCCESS(apr_os_thread_put(&thr, &osthr, APRPool::get())); + return Thread(thr); +} + +void Thread::yield() +{ + apr_thread_yield(); +} + +}} +#endif /*!_sys_apr_Thread_h*/ diff --git a/cpp/src/qpid/sys/apr/Time.cpp b/cpp/src/qpid/sys/apr/Time.cpp new file mode 100644 index 0000000000..34e740b144 --- /dev/null +++ b/cpp/src/qpid/sys/apr/Time.cpp @@ -0,0 +1,36 @@ +/* + * + * 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/Time.h" + +#include <apr_time.h> + +namespace qpid { +namespace sys { + +AbsTime AbsTime::now() { + AbsTime time_now; + time_now.time_ns = apr_time_now() * TIME_USEC; + return time_now; +} + +}} + diff --git a/cpp/src/qpid/sys/posix/Condition.h b/cpp/src/qpid/sys/posix/Condition.h new file mode 100644 index 0000000000..1c8d1a80b1 --- /dev/null +++ b/cpp/src/qpid/sys/posix/Condition.h @@ -0,0 +1,86 @@ +#ifndef _sys_posix_Condition_h +#define _sys_posix_Condition_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 "PrivatePosix.h" + +#include "qpid/sys/Mutex.h" +#include "qpid/sys/Time.h" + +#include <time.h> +#include <sys/errno.h> +#include <boost/noncopyable.hpp> + +namespace qpid { +namespace sys { + +/** + * A condition variable for thread synchronization. + */ +class Condition +{ + public: + inline Condition(); + inline ~Condition(); + inline void wait(Mutex&); + inline bool wait(Mutex&, const AbsTime& absoluteTime); + inline void notify(); + inline void notifyAll(); + + private: + pthread_cond_t condition; +}; + +Condition::Condition() { + QPID_POSIX_THROW_IF(pthread_cond_init(&condition, 0)); +} + +Condition::~Condition() { + QPID_POSIX_THROW_IF(pthread_cond_destroy(&condition)); +} + +void Condition::wait(Mutex& mutex) { + QPID_POSIX_THROW_IF(pthread_cond_wait(&condition, &mutex.mutex)); +} + +bool Condition::wait(Mutex& mutex, const AbsTime& absoluteTime){ + struct timespec ts; + toTimespec(ts, Duration(absoluteTime)); + int status = pthread_cond_timedwait(&condition, &mutex.mutex, &ts); + if (status != 0) { + if (status == ETIMEDOUT) return false; + throw QPID_POSIX_ERROR(status); + } + return true; +} + +void Condition::notify(){ + QPID_POSIX_THROW_IF(pthread_cond_signal(&condition)); +} + +void Condition::notifyAll(){ + QPID_POSIX_THROW_IF(pthread_cond_broadcast(&condition)); +} + +}} +#endif /*!_sys_posix_Condition_h*/ diff --git a/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp b/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp index 2676985cc6..d5a2c238d9 100644 --- a/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp +++ b/cpp/src/qpid/sys/posix/EventChannelAcceptor.cpp @@ -51,9 +51,9 @@ class EventChannelAcceptor : public Acceptor { int16_t port_, int backlog, int nThreads, bool trace_ ); - int getPort() const; + uint16_t getPort() const; - void run(ConnectionInputHandlerFactory& factory); + void run(ConnectionInputHandlerFactory* factory); void shutdown(); @@ -96,17 +96,17 @@ EventChannelAcceptor::EventChannelAcceptor( threads(EventChannelThreads::create(EventChannel::create(), nThreads)) { } -int EventChannelAcceptor::getPort() const { +uint16_t EventChannelAcceptor::getPort() const { return port; // Immutable no need for lock. } -void EventChannelAcceptor::run(ConnectionInputHandlerFactory& f) { +void EventChannelAcceptor::run(ConnectionInputHandlerFactory* f) { { Mutex::ScopedLock l(lock); if (!isRunning && !isShutdown) { isRunning = true; - factory = &f; - threads->post(acceptEvent); + factory = f; + threads->postEvent(acceptEvent); } } threads->join(); // Wait for shutdown. @@ -143,7 +143,7 @@ void EventChannelAcceptor::accept() int fd = acceptEvent.getAcceptedDesscriptor(); connections.push_back( new EventChannelConnection(threads, *factory, fd, fd, isTrace)); - threads->post(acceptEvent); // Keep accepting. + threads->postEvent(acceptEvent); // Keep accepting. } }} // namespace qpid::sys diff --git a/cpp/src/qpid/sys/posix/EventChannelConnection.cpp b/cpp/src/qpid/sys/posix/EventChannelConnection.cpp index 9a7d53fa8a..73e617ea83 100644 --- a/cpp/src/qpid/sys/posix/EventChannelConnection.cpp +++ b/cpp/src/qpid/sys/posix/EventChannelConnection.cpp @@ -140,7 +140,7 @@ void EventChannelConnection::startWrite() { writeFd, out.start(), out.available(), boost::bind(&EventChannelConnection::closeOnException, this, &EventChannelConnection::endWrite)); - threads->post(writeEvent); + threads->postEvent(writeEvent); } // ScopedBusy ctor increments busyThreads. @@ -180,7 +180,7 @@ void EventChannelConnection::startRead() { // Non blocking read, as much as we can swallow. readEvent = ReadEvent( readFd, in.start(), in.available(), readCallback,true); - threads->post(readEvent); + threads->postEvent(readEvent); } // Completion of initial read, expect protocolInit. diff --git a/cpp/src/qpid/sys/posix/Module.h b/cpp/src/qpid/sys/posix/Module.h new file mode 100644 index 0000000000..af3d6ac6ef --- /dev/null +++ b/cpp/src/qpid/sys/posix/Module.h @@ -0,0 +1,126 @@ +#ifndef _sys_posix_Module_h +#define _sys_posix_Module_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/QpidError.h" + +#include <boost/noncopyable.hpp> +#include <iostream> +#include <dlfcn.h> + +namespace qpid { +namespace sys { + +typedef void* dso_handle_t; + +template <class T> class Module : private boost::noncopyable +{ + typedef T* create_t(); + typedef void destroy_t(T*); + + dso_handle_t handle; + destroy_t* destroy; + T* ptr; + + void load(const std::string& name); + void unload(); + void* getSymbol(const std::string& name); + +public: + Module(const std::string& name); + T* operator->(); + T* get(); + ~Module() throw(); +}; + +template <class T> Module<T>::Module(const std::string& module) : destroy(0), ptr(0) +{ + load(module); + //TODO: need a better strategy for symbol names to allow multiple + //modules to be loaded without clashes... + + //Note: need the double cast to avoid errors in casting from void* to function pointer with -pedantic + create_t* create = reinterpret_cast<create_t*>(reinterpret_cast<intptr_t>(getSymbol("create"))); + destroy = reinterpret_cast<destroy_t*>(reinterpret_cast<intptr_t>(getSymbol("destroy"))); + ptr = create(); +} + +template <class T> T* Module<T>::operator->() +{ + return ptr; +} + +template <class T> T* Module<T>::get() +{ + return ptr; +} + +template <class T> Module<T>::~Module() throw() +{ + try { + if (handle && ptr) { + destroy(ptr); + } + if (handle) unload(); + } catch (std::exception& e) { + std::cout << "Error while destroying module: " << e.what() << std::endl; + } + destroy = 0; + handle = 0; + ptr = 0; +} + +template <class T> void Module<T>::load(const std::string& name) +{ + dlerror(); + handle = dlopen(name.c_str(), RTLD_NOW); + const char* error = dlerror(); + if (error) { + THROW_QPID_ERROR(INTERNAL_ERROR, error); + } +} + +template <class T> void Module<T>::unload() +{ + dlerror(); + dlclose(handle); + const char* error = dlerror(); + if (error) { + THROW_QPID_ERROR(INTERNAL_ERROR, error); + } +} + +template <class T> void* Module<T>::getSymbol(const std::string& name) +{ + dlerror(); + void* sym = dlsym(handle, name.c_str()); + const char* error = dlerror(); + if (error) { + THROW_QPID_ERROR(INTERNAL_ERROR, error); + } + return sym; +} + +}} +#endif //ifndef _sys_posix_Module_h + diff --git a/cpp/src/qpid/sys/posix/Mutex.h b/cpp/src/qpid/sys/posix/Mutex.h new file mode 100644 index 0000000000..b278c6b14a --- /dev/null +++ b/cpp/src/qpid/sys/posix/Mutex.h @@ -0,0 +1,127 @@ +#ifndef _sys_posix_Mutex_h +#define _sys_posix_Mutex_h + +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed 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 "check.h" + +#include <pthread.h> +#include <boost/noncopyable.hpp> + +namespace qpid { +namespace sys { + +class Condition; + +/** + * Mutex lock. + */ +class Mutex : private boost::noncopyable { + friend class Condition; + +public: + typedef ScopedLock<Mutex> ScopedLock; + typedef ScopedUnlock<Mutex> ScopedUnlock; + + inline Mutex(); + inline ~Mutex(); + inline void lock(); + inline void unlock(); + inline void trylock(); + +protected: + pthread_mutex_t mutex; +}; + +/** + * Initialise a recursive mutex attr for use in creating mutexes later + * (we use pthread_once to make sure it is initialised exactly once) + */ +namespace { + pthread_once_t onceControl = PTHREAD_ONCE_INIT; + pthread_mutexattr_t mutexattr; + + void initMutexattr() { + pthread_mutexattr_init(&mutexattr); + pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_RECURSIVE); + } + + struct RecursiveMutexattr { + RecursiveMutexattr() { + pthread_once(&onceControl, initMutexattr); + } + + operator const pthread_mutexattr_t*() const { + return &mutexattr; + } + }; + + RecursiveMutexattr recursiveMutexattr; +} + +/** + * PODMutex is a POD, can be static-initialized with + * PODMutex m = QPID_PODMUTEX_INITIALIZER + */ +struct PODMutex +{ + typedef ScopedLock<PODMutex> ScopedLock; + + inline void lock(); + inline void unlock(); + inline void trylock(); + + // Must be public to be a POD: + pthread_mutex_t mutex; +}; + +#define QPID_MUTEX_INITIALIZER { PTHREAD_MUTEX_INITIALIZER } + +void PODMutex::lock() { + QPID_POSIX_THROW_IF(pthread_mutex_lock(&mutex)); +} +void PODMutex::unlock() { + QPID_POSIX_THROW_IF(pthread_mutex_unlock(&mutex)); +} + +void PODMutex::trylock() { + QPID_POSIX_THROW_IF(pthread_mutex_trylock(&mutex)); +} + +Mutex::Mutex() { + QPID_POSIX_THROW_IF(pthread_mutex_init(&mutex, recursiveMutexattr)); +} + +Mutex::~Mutex(){ + QPID_POSIX_THROW_IF(pthread_mutex_destroy(&mutex)); +} + +void Mutex::lock() { + QPID_POSIX_THROW_IF(pthread_mutex_lock(&mutex)); +} +void Mutex::unlock() { + QPID_POSIX_THROW_IF(pthread_mutex_unlock(&mutex)); +} + +void Mutex::trylock() { + QPID_POSIX_THROW_IF(pthread_mutex_trylock(&mutex)); +} + +}} +#endif /*!_sys_posix_Mutex_h*/ diff --git a/cpp/src/qpid/sys/posix/PrivatePosix.h b/cpp/src/qpid/sys/posix/PrivatePosix.h new file mode 100644 index 0000000000..2707057ef0 --- /dev/null +++ b/cpp/src/qpid/sys/posix/PrivatePosix.h @@ -0,0 +1,39 @@ +#ifndef _sys_posix_PrivatePosix_h +#define _sys_posix_PrivatePosix_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/Time.h" + +struct timespec; +struct timeval; + +namespace qpid { +namespace sys { + +struct timespec& toTimespec(struct timespec& ts, const Duration& t); +struct timeval& toTimeval(struct timeval& tv, const Duration& t); +Duration toTime(const struct timespec& ts); + +}} + +#endif /*!_sys_posix_PrivatePosix_h*/ diff --git a/cpp/src/qpid/sys/posix/Socket.cpp b/cpp/src/qpid/sys/posix/Socket.cpp index e9bd2eeb6b..39651fa821 100644 --- a/cpp/src/qpid/sys/posix/Socket.cpp +++ b/cpp/src/qpid/sys/posix/Socket.cpp @@ -19,6 +19,12 @@ * */ +#include "qpid/sys/Socket.h" + +#include "qpid/QpidError.h" +#include "check.h" +#include "PrivatePosix.h" + #include <sys/socket.h> #include <sys/errno.h> #include <netinet/in.h> @@ -26,10 +32,6 @@ #include <boost/format.hpp> -#include "qpid/QpidError.h" -#include "check.h" -#include "qpid/sys/Socket.h" - using namespace qpid::sys; Socket Socket::createTcp() @@ -41,11 +43,10 @@ Socket Socket::createTcp() Socket::Socket(int descriptor) : socket(descriptor) {} -void Socket::setTimeout(Time interval) +void Socket::setTimeout(const Duration& interval) { struct timeval tv; - tv.tv_sec = interval/TIME_SEC; - tv.tv_usec = (interval%TIME_SEC)/TIME_USEC; + toTimeval(tv, interval); setsockopt(socket, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); setsockopt(socket, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); } diff --git a/cpp/src/qpid/sys/posix/Socket.h b/cpp/src/qpid/sys/posix/Socket.h new file mode 100644 index 0000000000..614221354f --- /dev/null +++ b/cpp/src/qpid/sys/posix/Socket.h @@ -0,0 +1,76 @@ +#ifndef _sys_posix_Socket_h +#define _sys_posix_Socket_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 <string> +#include "qpid/sys/Time.h" + +namespace qpid { +namespace sys { + +class Socket +{ + public: + /** Create an initialized TCP socket */ + static Socket createTcp(); + + /** Create a socket wrapper for descriptor. */ + Socket(int descriptor = 0); + + /** Set timeout for read and write */ + void setTimeout(const Duration& interval); + + void connect(const std::string& host, int port); + + void close(); + + enum { SOCKET_TIMEOUT=-2, SOCKET_EOF=-3 } ErrorCode; + + /** Returns bytes sent or an ErrorCode value < 0. */ + ssize_t send(const void* data, size_t size); + + /** + * Returns bytes received, an ErrorCode value < 0 or 0 + * if the connection closed in an orderly manner. + */ + ssize_t recv(void* data, size_t size); + + /** Bind to a port and start listening. + *@param port 0 means choose an available port. + *@param backlog maximum number of pending connections. + *@return The bound port. + */ + int listen(int port = 0, int backlog = 10); + + /** Get file descriptor */ + int fd(); + + private: + void init() const; + mutable int socket; // Initialized on demand. +}; + +}} + + +#endif /*!_sys_Socket_h*/ diff --git a/cpp/src/qpid/sys/posix/Thread.cpp b/cpp/src/qpid/sys/posix/Thread.cpp index fc85e35028..dc9b21448f 100644 --- a/cpp/src/qpid/sys/posix/Thread.cpp +++ b/cpp/src/qpid/sys/posix/Thread.cpp @@ -19,7 +19,8 @@ * */ -#include "qpid/sys/Thread.h" +#include "Thread.h" +#include "qpid/sys/Runnable.h" void* qpid::sys::Thread::runRunnable(void* p) { diff --git a/cpp/src/qpid/sys/posix/Thread.h b/cpp/src/qpid/sys/posix/Thread.h new file mode 100644 index 0000000000..9de7299f5a --- /dev/null +++ b/cpp/src/qpid/sys/posix/Thread.h @@ -0,0 +1,86 @@ +#ifndef _sys_posix_Thread_h +#define _sys_posix_Thread_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 "check.h" +#include <pthread.h> + +namespace qpid { +namespace sys { + +class Runnable; + +class Thread +{ + public: + inline static Thread current(); + inline static void yield(); + + inline Thread(); + inline explicit Thread(qpid::sys::Runnable*); + inline explicit Thread(qpid::sys::Runnable&); + + inline void join(); + + inline long id(); + + private: + static void* runRunnable(void* runnable); + inline Thread(pthread_t); + pthread_t thread; +}; + + +Thread::Thread() : thread(0) {} + +Thread::Thread(Runnable* runnable) { + QPID_POSIX_THROW_IF(pthread_create(&thread, NULL, runRunnable, runnable)); +} + +Thread::Thread(Runnable& runnable) { + QPID_POSIX_THROW_IF(pthread_create(&thread, NULL, runRunnable, &runnable)); +} + +void Thread::join(){ + if (thread != 0) + QPID_POSIX_THROW_IF(pthread_join(thread, 0)); +} + +long Thread::id() { + return long(thread); +} + +Thread::Thread(pthread_t thr) : thread(thr) {} + +Thread Thread::current() { + return Thread(pthread_self()); +} + +void Thread::yield() +{ + QPID_POSIX_THROW_IF(pthread_yield()); +} + + +}} +#endif /*!_sys_posix_Thread_h*/ diff --git a/cpp/src/qpid/sys/Time.cpp b/cpp/src/qpid/sys/posix/Time.cpp index ad6185b966..2228caea58 100644 --- a/cpp/src/qpid/sys/Time.cpp +++ b/cpp/src/qpid/sys/posix/Time.cpp @@ -19,42 +19,39 @@ * */ -#include "Time.h" +#include "PrivatePosix.h" -namespace qpid { -namespace sys { - -// APR ================================================================ -#if USE_APR +#include "qpid/sys/Time.h" -Time now() { return apr_time_now() * TIME_USEC; } +#include <time.h> +#include <sys/time.h> -// POSIX================================================================ -#else +namespace qpid { +namespace sys { -Time now() { +AbsTime AbsTime::now() { struct timespec ts; clock_gettime(CLOCK_REALTIME, &ts); - return toTime(ts); + AbsTime time_now; + time_now.time_ns = toTime(ts).nanosecs; + return time_now; } -struct timespec toTimespec(const Time& t) { - struct timespec ts; - toTimespec(ts, t); +struct timespec& toTimespec(struct timespec& ts, const Duration& t) { + ts.tv_sec = t / TIME_SEC; + ts.tv_nsec = t % TIME_SEC; return ts; } -struct timespec& toTimespec(struct timespec& ts, const Time& t) { - ts.tv_sec = t / TIME_SEC; - ts.tv_nsec = t % TIME_SEC; - return ts; +struct timeval& toTimeval(struct timeval& tv, const Duration& t) { + tv.tv_sec = t/TIME_SEC; + tv.tv_usec = (t%TIME_SEC)/TIME_USEC; + return tv; } -Time toTime(const struct timespec& ts) { +Duration toTime(const struct timespec& ts) { return ts.tv_sec*TIME_SEC + ts.tv_nsec; } - -#endif }} |