blob: 4fcc33095e127ebd7dbc53d122081b25e02e8f10 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
/*
* Oreka -- A media capture and retrieval platform
*
* Copyright (C) 2005, orecx LLC
*
* http://www.orecx.com
*
* This program is free software, distributed under the terms of
* the GNU General Public License.
* Please refer to http://www.gnu.org/copyleft/gpl.html
*
*/
#ifndef __THREADSAFEQUEUE_H__
#define __THREADSAFEQUEUE_H__
#include <queue>
#include "ace/Thread_Mutex.h"
#include "ace/Thread_Semaphore.h"
#include "Utils.h"
/** Thread safe queue holding objects of arbitrary class.
Enqueuing is never blocking
Dequeuing is blocking when the queue is empty
*/
template <class T> class ThreadSafeQueue
{
public:
ThreadSafeQueue(int size = 1000)
{
m_size = size;
m_semaphore.acquire(); // reset count to zero
};
bool push(T &);
T pop();
private:
int m_size;
ACE_Thread_Mutex m_mutex;
ACE_Thread_Semaphore m_semaphore;
std::queue<T> m_queue;
};
/** Push an element onto the queue, returns false if queue full (never blocks) */
template <class T> bool ThreadSafeQueue<T>::push(T &element)
{
bool result = false;
MutexSentinel mutexSentinel(m_mutex);
if (m_queue.size() < (unsigned int)m_size)
{
m_queue.push(element);
result = true;
}
m_semaphore.release();
return result;
}
/** Pop and element from the queue, or blocks until one available */
template <class T> T ThreadSafeQueue<T>::pop()
{
m_semaphore.acquire();
MutexSentinel mutexSentinel(m_mutex);
T element = m_queue.front();
m_queue.pop();
return element;
}
#endif // __THREADSAFEQUEUE_H__
|
