diff options
Diffstat (limited to 'raul')
| -rw-r--r-- | raul/AtomicInt.h | 75 | ||||
| -rw-r--r-- | raul/AtomicPtr.h | 43 | ||||
| -rw-r--r-- | raul/Makefile.am | 6 | ||||
| -rw-r--r-- | raul/SRMWQueue.h | 201 | ||||
| -rw-r--r-- | raul/SRSWQueue.h (renamed from raul/Queue.h) | 107 |
5 files changed, 373 insertions, 59 deletions
diff --git a/raul/AtomicInt.h b/raul/AtomicInt.h new file mode 100644 index 0000000..ba8e4b0 --- /dev/null +++ b/raul/AtomicInt.h @@ -0,0 +1,75 @@ +/* This file is part of Ingen. Copyright (C) 2007 Dave Robillard. + * + * Ingen is free software; you can redistribute it and/or modify it under the + * terms of the GNU General Public License as published by the Free Software + * Foundation; either version 2 of the License, or (at your option) any later + * version. + * + * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for details. + * + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef RAUL_ATOMIC_INT_H +#define RAUL_ATOMIC_INT_H + +#include <glib.h> + +class AtomicInt { +public: + + inline AtomicInt(int val) + { g_atomic_int_set(&_val, val); } + + inline AtomicInt(const AtomicInt& copy) + { g_atomic_int_set(&_val, copy.get()); } + + inline int get() const + { return g_atomic_int_get(&_val); } + + inline void operator=(int val) + { g_atomic_int_set(&_val, val); } + + inline void operator+=(int val) + { g_atomic_int_add(&_val, val); } + + inline void operator-=(int val) + { g_atomic_int_add(&_val, -val); } + + inline bool operator==(int val) const + { return get() == val; } + + inline int operator+(int val) const + { return get() + val; } + + inline AtomicInt& operator++() // prefix + { g_atomic_int_inc(&_val); return *this; } + + inline AtomicInt& operator--() // prefix + { g_atomic_int_add(&_val, -1); return *this; } + + /** Set value to newval iff current value is oldval */ + inline bool compare_and_exchange(int oldval, int newval) + { return g_atomic_int_compare_and_exchange(&_val, oldval, newval); } + + /** Add val to value. + * @return value immediately before addition took place. + */ + inline int exchange_and_add(int val) + { return g_atomic_int_exchange_and_add(&_val, val); } + + /** Decrement value. + * @return true if value is now 0, otherwise false. + */ + inline bool decrement_and_test() + { return g_atomic_int_dec_and_test(&_val); } + +private: + volatile int _val; +}; + +#endif // RAUL_ATOMIC_INT_H diff --git a/raul/AtomicPtr.h b/raul/AtomicPtr.h new file mode 100644 index 0000000..8209ca6 --- /dev/null +++ b/raul/AtomicPtr.h @@ -0,0 +1,43 @@ +/* This file is part of Ingen. Copyright (C) 2007 Dave Robillard. + * + * Ingen is free software; you can redistribute it and/or modify it under the + * terms of the GNU General Public License as published by the Free Software + * Foundation; either version 2 of the License, or (at your option) any later + * version. + * + * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for details. + * + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef RAUL_ATOMIC_PTR_H +#define RAUL_ATOMIC_PTR_H + +#include <glib.h> + +template<typename T> +class AtomicPtr { +public: + + inline AtomicPtr(const AtomicPtr& copy) + { g_atomic_pointer_set(&_val, copy.get()); } + + inline T* get() const + { return (T*)g_atomic_pointer_get(&_val); } + + inline void operator=(T* val) + { g_atomic_pointer_set(&_val, val); } + + /** Set value to newval iff current value is oldval */ + inline bool compare_and_exchange(int oldval, int newval) + { return g_atomic_pointer_compare_and_exchange(&_val, oldval, newval); } + +private: + volatile T* _val; +}; + +#endif // RAUL_ATOMIC_PTR_H diff --git a/raul/Makefile.am b/raul/Makefile.am index 64cc7ed..a536a0e 100644 --- a/raul/Makefile.am +++ b/raul/Makefile.am @@ -4,7 +4,8 @@ raulinclude_HEADERS = \ SharedPtr.h \ WeakPtr.h \ Path.h \ - Queue.h \ + SRSWQueue.h \ + SRMWQueue.h \ Semaphore.h \ Mutex.h \ Condition.h \ @@ -14,7 +15,8 @@ raulinclude_HEADERS = \ Atom.h \ JackDriver.h \ RDFQuery.h \ - Namespaces.h + Namespaces.h \ + AtomicInt.h if WITH_LIBLO raulinclude_HEADERS += AtomLiblo.h diff --git a/raul/SRMWQueue.h b/raul/SRMWQueue.h new file mode 100644 index 0000000..2a4955c --- /dev/null +++ b/raul/SRMWQueue.h @@ -0,0 +1,201 @@ +/* This file is part of Ingen. Copyright (C) 2006 Dave Robillard. + * + * Ingen is free software; you can redistribute it and/or modify it under the + * terms of the GNU General Public License as published by the Free Software + * Foundation; either version 2 of the License, or (at your option) any later + * version. + * + * Ingen is distributed in the hope that it will be useful, but WITHOUT ANY + * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS + * FOR A PARTICULAR PURPOSE. See the GNU General Public License for details. + * + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#ifndef RAUL_SRMW_QUEUE_H +#define RAUL_SRMW_QUEUE_H + +#include <cassert> +#include <cstdlib> +#include <boost/utility.hpp> +#include "raul/AtomicInt.h" + +#include <iostream> +using namespace std; + + +/** Realtime-safe single-reader multi-writer queue (aka lock-free ringbuffer) + * + * Implemented as a dequeue in a fixed array. Both push and pop are realtime + * safe, but only push is threadsafe. In other words, multiple threads can push + * data into this queue for a single thread to consume. + * + * The interface is intentionally as similar to std::queue as possible, but + * note the additional thread restrictions imposed (e.g. empty() is only + * legal to call in the read thread). + * + * Obey the threading restrictions documented here, or horrible nasty (possibly + * undetected) errors will occur. + * + * This is slightly more expensive than the SRMWSRMWQueue. Use that if you don't + * require multiple writers. + * + * \ingroup raul + */ +template <typename T> +class SRMWQueue : boost::noncopyable +{ +public: + SRMWQueue(size_t size); + ~SRMWQueue(); + + + // Any thread: + + inline size_t capacity() const { return _size-1; } + + + // Write thread(s): + + inline bool full() const; + inline bool push(const T& obj); + + + // Read thread: + + inline bool empty() const; + inline T& front() const; + inline void pop(); + +private: + + // Note that _front doesn't need to be an AtomicInt since it's only accessed + // by the (single) reader thread + + int _front; ///< Circular index of element at front of queue (READER ONLY) + AtomicInt _back; ///< Circular index 1 past element at back of queue (WRITERS ONLY) + AtomicInt _space; ///< Remaining free space for new elements (all threads) + const int _size; ///< Size of @ref _objects (you can store _size-1 objects) + T* const _objects; ///< Fixed array containing queued elements +}; + + +template<typename T> +SRMWQueue<T>::SRMWQueue(size_t size) + : _front(0) + , _back(0) + , _space(size) + , _size(size+1) + , _objects((T*)calloc(_size, sizeof(T))) +{ + assert(size > 1); + assert(_size-1 == _space.get()); +} + + +template <typename T> +SRMWQueue<T>::~SRMWQueue() +{ + free(_objects); +} + + +/** Return whether the queue is full. + * + * Write thread(s) only. + */ +template <typename T> +inline bool +SRMWQueue<T>::full() const +{ + return ( _space.get() <= 0 ); +} + + +/** Push an item onto the back of the SRMWQueue - realtime-safe, not thread-safe. + * + * Write thread(s) only. + * + * @returns true if @a elem was successfully pushed onto the queue, + * false otherwise (queue is full). + */ +template <typename T> +inline bool +SRMWQueue<T>::push(const T& elem) +{ + const int old_space = _space.exchange_and_add(-1); + + const bool already_full = ( old_space <= 0 ); + + /* Technically right here pop could be called in the reader thread and + * make space available, but no harm in failing anyway - this queue + * really isn't designed to be filled... */ + + if (already_full) { + + /* if multiple threads simultaneously get here, _space may be 0 + * or negative. The next call to pop() will set _space back to + * a sane value. Note that _space is not exposed, so this is okay + * (... assuming this code is correct) */ + + return false; + + } else { + + const int write_index = _back.exchange_and_add(1) % _size; + _objects[write_index] = elem; + + return true; + + } +} + + +/** Return whether the queue is empty. + * + * Read thread only. + */ +template <typename T> +inline bool +SRMWQueue<T>::empty() const +{ + return ( _space == capacity() ); +} + + +/** Return the element at the front of the queue without removing it. + * + * It is a fatal error to call front() when the queue is empty. + * Read thread only. + */ +template <typename T> +inline T& +SRMWQueue<T>::front() const +{ + return _objects[_front]; +} + + +/** Pop an item off the front of the queue - realtime-safe, NOT thread-safe. + * + * It is a fatal error to call pop() if the queue is empty. + * Read thread only. + * + * @return true if queue is now empty, otherwise false. + */ +template <typename T> +inline void +SRMWQueue<T>::pop() +{ + _front = (_front + 1) % (_size); + + if (_space.get() < 0) + _space = 1; + else + ++_space; +} + + +#endif // RAUL_SRMW_QUEUE_H diff --git a/raul/Queue.h b/raul/SRSWQueue.h index 36d1cca..e7dd60d 100644 --- a/raul/Queue.h +++ b/raul/SRSWQueue.h @@ -14,65 +14,71 @@ * 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA */ -#ifndef RAUL_QUEUE_H -#define RAUL_QUEUE_H +#ifndef RAUL_SRSW_QUEUE_H +#define RAUL_SRSW_QUEUE_H #include <cassert> #include <cstdlib> #include <boost/utility.hpp> +#include "raul/AtomicInt.h" /** Realtime-safe single-reader single-writer queue (aka lock-free ringbuffer) * * Implemented as a dequeue in a fixed array. This is read/write thread-safe, * pushing and popping may occur simultaneously by seperate threads, but - * the push and pop operations themselves are not thread-safe. - * - * FIXME: Verify atomicity of everything here. + * the push and pop operations themselves are not thread-safe (ie. there can + * be at most 1 read and at most 1 writer thread). * * \ingroup raul */ template <typename T> -class Queue : boost::noncopyable +class SRSWQueue : boost::noncopyable { public: - Queue(size_t size); - ~Queue(); + SRSWQueue(size_t size); + ~SRSWQueue(); - inline bool is_empty() const; - inline bool is_full() const; + // Any thread: - inline size_t capacity() const { return m_size-1; } - inline size_t fill() const; + inline size_t capacity() const { return _size-1; } - inline T& front() const; + + // Write thread(s): + + inline bool full() const; + inline bool push(const T& obj); + - inline bool push(T obj); - inline T& pop(); + // Read thread: + + inline bool empty() const; + inline T& front() const; + inline void pop(); private: - volatile size_t m_front; ///< Index to front of queue (circular) - volatile size_t m_back; ///< Index to back of queue (one past last element) (circular) - const size_t m_size; ///< Size of @ref m_objects (you can store m_size-1 objects) - T* const m_objects; ///< Fixed array containing queued elements + volatile size_t _front; ///< Index to front of queue (circular) + volatile size_t _back; ///< Index to back of queue (one past last element) (circular) + const size_t _size; ///< Size of @ref _objects (you can store _size-1 objects) + T* const _objects; ///< Fixed array containing queued elements }; template<typename T> -Queue<T>::Queue(size_t size) -: m_front(0), - m_back(0), - m_size(size+1), - m_objects((T*)calloc(m_size, sizeof(T))) +SRSWQueue<T>::SRSWQueue(size_t size) +: _front(0), + _back(0), + _size(size+1), + _objects((T*)calloc(_size, sizeof(T))) { assert(size > 1); } template <typename T> -Queue<T>::~Queue() +SRSWQueue<T>::~SRSWQueue() { - free(m_objects); + free(_objects); } @@ -80,9 +86,9 @@ Queue<T>::~Queue() */ template <typename T> inline bool -Queue<T>::is_empty() const +SRSWQueue<T>::empty() const { - return (m_back == m_front); + return (_back == _front); } @@ -90,20 +96,10 @@ Queue<T>::is_empty() const */ template <typename T> inline bool -Queue<T>::is_full() const +SRSWQueue<T>::full() const { - // FIXME: This can probably be faster - return (fill() == capacity()); -} - - -/** Returns how many elements are currently in the queue. - */ -template <typename T> -inline size_t -Queue<T>::fill() const -{ - return (m_back + m_size - m_front) % m_size; + // FIXME: uses both _front and _back - thread safe? + return ( ((_front - _back + _size) % _size) == 1 ); } @@ -111,26 +107,26 @@ Queue<T>::fill() const */ template <typename T> inline T& -Queue<T>::front() const +SRSWQueue<T>::front() const { - return m_objects[m_front]; + return _objects[_front]; } -/** Push an item onto the back of the Queue - realtime-safe, not thread-safe. +/** Push an item onto the back of the SRSWQueue - realtime-safe, not thread-safe. * * @returns true if @a elem was successfully pushed onto the queue, * false otherwise (queue is full). */ template <typename T> inline bool -Queue<T>::push(T elem) +SRSWQueue<T>::push(const T& elem) { - if (is_full()) { + if (full()) { return false; } else { - m_objects[m_back] = elem; - m_back = (m_back + 1) % (m_size); + _objects[_back] = elem; + _back = (_back + 1) % (_size); return true; } } @@ -143,17 +139,14 @@ Queue<T>::push(T elem) * @returns the element popped. */ template <typename T> -inline T& -Queue<T>::pop() +inline void +SRSWQueue<T>::pop() { - assert(!is_empty()); - assert(m_size > 0); + assert(!empty()); + assert(_size > 0); - T& r = m_objects[m_front]; - m_front = (m_front + 1) % (m_size); - - return r; + _front = (_front + 1) % (_size); } -#endif // RAUL_QUEUE_H +#endif // RAUL_SRSW_QUEUE_H |