blob: 4fd0cda9b0a2015c4de6a4a2f5e62914fef516c7 (
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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
/*
* 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 = 10000)
{
m_size = size;
m_semaphore.acquire(); // reset count to zero
};
bool push(T &);
T pop();
int numElements();
void setSize(int size);
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()
{
// #### Fixme: when timeout specified under Linux CentOS 4.2, acquire returns immediately instead of waiting for the timeout -> causes CPU to spike at 100%
#ifdef WIN32
ACE_Time_Value timeout(time(NULL)+2);
m_semaphore.acquire(&timeout);
#else
m_semaphore.acquire();
#endif
MutexSentinel mutexSentinel(m_mutex);
T element;
if(m_queue.size() > 0)
{
element = m_queue.front();
m_queue.pop();
}
return element;
}
template <class T> int ThreadSafeQueue<T>::numElements()
{
return m_queue.size();
}
template <class T> void ThreadSafeQueue<T>::setSize(int size)
{
m_size = size;
}
#endif // __THREADSAFEQUEUE_H__
|
