Orocos Real-Time Toolkit  2.9.0
AtomicMWSRQueue.hpp
Go to the documentation of this file.
1 /***************************************************************************
2  tag: The SourceWorks Tue Sep 7 00:55:18 CEST 2010 AtomicMWSRQueue.hpp
3 
4  AtomicMWSRQueue.hpp - description
5  -------------------
6  begin : Tue September 07 2010
7  copyright : (C) 2010 The SourceWorks
8  email : peter@thesourceworks.com
9 
10  ***************************************************************************
11  * This library is free software; you can redistribute it and/or *
12  * modify it under the terms of the GNU General Public *
13  * License as published by the Free Software Foundation; *
14  * version 2 of the License. *
15  * *
16  * As a special exception, you may use this file as part of a free *
17  * software library without restriction. Specifically, if other files *
18  * instantiate templates or use macros or inline functions from this *
19  * file, or you compile this file and link it with other files to *
20  * produce an executable, this file does not by itself cause the *
21  * resulting executable to be covered by the GNU General Public *
22  * License. This exception does not however invalidate any other *
23  * reasons why the executable file might be covered by the GNU General *
24  * Public License. *
25  * *
26  * This library is distributed in the hope that it will be useful, *
27  * but WITHOUT ANY WARRANTY; without even the implied warranty of *
28  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU *
29  * Lesser General Public License for more details. *
30  * *
31  * You should have received a copy of the GNU General Public *
32  * License along with this library; if not, write to the Free Software *
33  * Foundation, Inc., 59 Temple Place, *
34  * Suite 330, Boston, MA 02111-1307 USA *
35  * *
36  ***************************************************************************/
37 
38 
39 #ifndef ORO_CORELIB_ATOMIC_MWSR_QUEUE_HPP
40 #define ORO_CORELIB_ATOMIC_MWSR_QUEUE_HPP
41 
42 #include "AtomicQueue.hpp"
43 #include "../os/CAS.hpp"
44 #include <utility>
45 
46 namespace RTT
47 {
48  namespace internal
49  {
59  template<class T>
60  class AtomicMWSRQueue : public AtomicQueue<T>
61  {
62  //typedef _T* T;
63  const int _size;
64  typedef T C;
65  typedef volatile C* CachePtrType;
66  typedef C* volatile CacheObjType;
67  typedef C ValueType;
68  typedef C* PtrType;
69 
77  union SIndexes
78  {
79  unsigned long _value;
80  unsigned short _index[2];
81  };
82 
87  CachePtrType _buf;
88 
93  volatile SIndexes _indxes;
94 
99  CachePtrType advance_w()
100  {
101  SIndexes oldval, newval;
102  do
103  {
104  oldval._value = _indxes._value; /*Points to a free writable pointer.*/
105  newval._value = oldval._value; /*Points to the next writable pointer.*/
106  // check for full :
107  if ((newval._index[0] == newval._index[1] - 1) || (newval._index[0] == newval._index[1] + _size - 1))
108  {
109  return 0;
110  }
111  newval._index[0]++;
112  if (newval._index[0] >= _size)
113  newval._index[0] = 0;
114  // if ptr is unchanged, replace it with newval.
115  } while (!os::CAS(&_indxes._value, oldval._value, newval._value));
116  // frome here on :
117  // oldval is 'unique', other preempting threads
118  // will have a different value for oldval, as
119  // _wptr advances. As long as oldval has not been written,
120  // rptr will not advance and wptr will remain stuck behind it.
121  // return the old position to write to :
122  return &_buf[oldval._index[0]];
123  }
124 
129  bool advance_r(T& result)
130  {
131  SIndexes oldval, newval;
132  // read it:
133  oldval._value = _indxes._value;
134  result = _buf[oldval._index[1]];
135  // return it if not yet written:
136  if ( !result )
137  return false;
138  // got it, clear field.
139  _buf[oldval._index[1]] = 0;
140 
141  // move pointer:
142  do
143  {
144  // re-read indxes, since we are the only reader,
145  // _index[1] will not have changed since entry of this function
146  oldval._value = _indxes._value;
147  newval._value = oldval._value;
148  ++newval._index[1];
149  if (newval._index[1] >= _size)
150  newval._index[1] = 0;
151 
152  // we need to CAS since the write pointer may have moved.
153  // this moves read pointer only:
154  } while (!os::CAS(&_indxes._value, oldval._value, newval._value));
155 
156  return true;
157  }
158 
159  // non-copyable !
161 
162  public:
164 
169  AtomicMWSRQueue(unsigned int size) :
170  _size(size + 1)
171  {
172  _buf = new C[_size];
173  this->clear();
174  }
175 
177  {
178  delete[] _buf;
179  }
180 
185  bool isFull() const
186  {
187  // two cases where the queue is full :
188  // if wptr is one behind rptr or if wptr is at end
189  // and rptr at beginning.
190  SIndexes val;
191  val._value = _indxes._value;
192  return val._index[0] == val._index[1] - 1 || val._index[0] == val._index[1] + _size - 1;
193  }
194 
199  bool isEmpty() const
200  {
201  // empty if nothing to read.
202  SIndexes val;
203  val._value = _indxes._value;
204  return val._index[0] == val._index[1];
205  }
206 
210  size_type capacity() const
211  {
212  return _size - 1;
213  }
214 
218  size_type size() const
219  {
220  SIndexes val;
221  val._value = _indxes._value;
222  int c = (val._index[0] - val._index[1]);
223  return c >= 0 ? c : c + _size;
224  }
225 
231  bool enqueue(const T& value)
232  {
233  if (value == 0)
234  return false;
235  CachePtrType loc = advance_w();
236  if (loc == 0)
237  return false;
238  *loc = value;
239  return true;
240  }
241 
249  bool dequeue(T& result)
250  {
251  T tmpresult;
252  if (advance_r(tmpresult) ) {
253  result = tmpresult;
254  return true;
255  }
256  return false;
257  }
258 
262  const T front() const
263  {
264  return _buf[_indxes._index[1]];
265  }
266 
270  void clear()
271  {
272  for (int i = 0; i != _size; ++i)
273  {
274  _buf[i] = 0;
275  }
276  _indxes._value = 0;
277  }
278 
279  };
280 
281  }
282 }
283 #endif
AtomicMWSRQueue(unsigned int size)
Create an AtomicMWSRQueue with queue size size.
const T front() const
Return the next to be read value.
bool dequeue(T &result)
Dequeue an item.
size_type capacity() const
Return the maximum number of items this queue can contain.
An atomic, non-blocking single ended queue (FIFO) for storing a pointer to T.
Definition: AtomicQueue.hpp:57
size_type size() const
Return the number of elements in the queue.
bool isEmpty() const
Inspect if the Queue is empty.
bool CAS(volatile T *addr, const V &expected, const W &value)
Compare And Swap.
Definition: CAS.hpp:54
bool enqueue(const T &value)
Enqueue an item.
void clear()
Clear all contents of the Queue and thus make it empty.
Contains TaskContext, Activity, OperationCaller, Operation, Property, InputPort, OutputPort, Attribute.
Definition: Activity.cpp:52
bool isFull() const
Inspect if the Queue is full.
Create an atomic, non-blocking Multi-Writer Single-Reader FIFO for storing a pointer T by value...
AtomicQueue< T >::size_type size_type