Orocos Real-Time Toolkit  2.8.3
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 "../os/CAS.hpp"
43 #include <utility>
44 
45 namespace RTT
46 {
47  namespace internal
48  {
58  template<class T>
60  {
61  //typedef _T* T;
62  const int _size;
63  typedef T C;
64  typedef volatile C* CachePtrType;
65  typedef C* volatile CacheObjType;
66  typedef C ValueType;
67  typedef C* PtrType;
68 
76  union SIndexes
77  {
78  unsigned long _value;
79  unsigned short _index[2];
80  };
81 
86  CachePtrType _buf;
87 
92  volatile SIndexes _indxes;
93 
98  CachePtrType advance_w()
99  {
100  SIndexes oldval, newval;
101  do
102  {
103  oldval._value = _indxes._value; /*Points to a free writable pointer.*/
104  newval._value = oldval._value; /*Points to the next writable pointer.*/
105  // check for full :
106  if ((newval._index[0] == newval._index[1] - 1) || (newval._index[0] == newval._index[1] + _size - 1))
107  {
108  return 0;
109  }
110  newval._index[0]++;
111  if (newval._index[0] >= _size)
112  newval._index[0] = 0;
113  // if ptr is unchanged, replace it with newval.
114  } while (!os::CAS(&_indxes._value, oldval._value, newval._value));
115  // frome here on :
116  // oldval is 'unique', other preempting threads
117  // will have a different value for oldval, as
118  // _wptr advances. As long as oldval has not been written,
119  // rptr will not advance and wptr will remain stuck behind it.
120  // return the old position to write to :
121  return &_buf[oldval._index[0]];
122  }
123 
128  bool advance_r(T& result)
129  {
130  SIndexes oldval, newval;
131  // read it:
132  oldval._value = _indxes._value;
133  result = _buf[oldval._index[1]];
134  // return it if not yet written:
135  if ( !result )
136  return false;
137  // got it, clear field.
138  _buf[oldval._index[1]] = 0;
139 
140  // move pointer:
141  do
142  {
143  // re-read indxes, since we are the only reader,
144  // _index[1] will not have changed since entry of this function
145  oldval._value = _indxes._value;
146  newval._value = oldval._value;
147  ++newval._index[1];
148  if (newval._index[1] >= _size)
149  newval._index[1] = 0;
150 
151  // we need to CAS since the write pointer may have moved.
152  // this moves read pointer only:
153  } while (!os::CAS(&_indxes._value, oldval._value, newval._value));
154 
155  return true;
156  }
157 
158  // non-copyable !
160  public:
161  typedef unsigned int size_type;
162 
167  AtomicMWSRQueue(unsigned int size) :
168  _size(size + 1)
169  {
170  _buf = new C[_size];
171  this->clear();
172  }
173 
175  {
176  delete[] _buf;
177  }
178 
183  bool isFull() const
184  {
185  // two cases where the queue is full :
186  // if wptr is one behind rptr or if wptr is at end
187  // and rptr at beginning.
188  SIndexes val;
189  val._value = _indxes._value;
190  return val._index[0] == val._index[1] - 1 || val._index[0] == val._index[1] + _size - 1;
191  }
192 
197  bool isEmpty() const
198  {
199  // empty if nothing to read.
200  SIndexes val;
201  val._value = _indxes._value;
202  return val._index[0] == val._index[1];
203  }
204 
208  size_type capacity() const
209  {
210  return _size - 1;
211  }
212 
216  size_type size() const
217  {
218  SIndexes val;
219  val._value = _indxes._value;
220  int c = (val._index[0] - val._index[1]);
221  return c >= 0 ? c : c + _size;
222  }
223 
229  bool enqueue(const T& value)
230  {
231  if (value == 0)
232  return false;
233  CachePtrType loc = advance_w();
234  if (loc == 0)
235  return false;
236  *loc = value;
237  return true;
238  }
239 
247  bool dequeue(T& result)
248  {
249  T tmpresult;
250  if (advance_r(tmpresult) ) {
251  result = tmpresult;
252  return true;
253  }
254  return false;
255  }
256 
260  const T front() const
261  {
262  return _buf[_indxes._index[1]];
263  }
264 
268  void clear()
269  {
270  for (int i = 0; i != _size; ++i)
271  {
272  _buf[i] = 0;
273  }
274  _indxes._value = 0;
275  }
276 
277  };
278 
279  }
280 }
281 #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.
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:51
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...