Hi Guys,
I've done a mulithread queue implementation on stl<queue>, my
developement environment is on VC6 as well as FC3. Let's talks about
the win32 side.
The suspected memory leak is find after I've run through my unit test
cases.
Test Case:
1. start 2 threads, thread A and thread B.
2. thread A enqueue 10000 dummy objects, meanwhile thread B dequeue
10000 dummy objects (delete object at the same time), these operations
are called a "cycle".
3. there is a while loop to loop the cycle, in between each cycle,
there is a 5 seconds sleep time.
4. run this for 10 minutes
5. stop insert any dummy objects
I've run this test case for 15 mins, and monitor its private_bytes,
working_set and virtual_bytes with the windows performance monitor.
My concern after running this test case is that,
1. virtual_bytes increase double its size frequently (seems to me it is
due to the queue memory allocation), and the virtual_bytes never drops
into its original position (I mean at the beginning when the program is
started)
2. both private_bytes and working_set are increasing as well, just the
fact that, duing the 5 seconds wait time, (step 3), these values drop
to nearly zeros, but after the 5 seconds wait time, these values
increase rapidly back to a state where it was dropped. and, of course
keep increasing.
3. during the (step 5), I can tell virtual_bytes is at its max, and
private_bytes and working_set are set at a values (1/3 below the max
point) and won't be able to drop any more.
My questions are:
1. shall windows OS handle the virtual_bytes, private_bytes and
working_bytes dellocation?
2. or is it because there is error on my implementation of multithread
queue?
Thank you very much. I am really appreciate yours help,
Best wishes,
Kit
--------------- My queue --------------------------
#ifndef __QUEUELOCK_HEADER__
#define __QUEUELOCK_HEADER__
#ifdef WIN32
# define _WINSOCKAPI_
# include <Windows.h>
#else
extern "C" {
# include <pthread.h>
};
#endif
#include <queue>
#include <iostream>
#define EXIT_IF(EXP) if (EXP) { cout << "Failed qlock: " #EXP "\n";
exit(1); }
#define QUEUE_MAX_SIZE 30000
using namespace std;
/* Queue with a lock, a template extending queue template
*/
template <typename _Tp, typename _Sequence = deque<_Tp
class qlock : public queue<_Tp,_Sequence{
typedef typename _Sequence::value_type value_type;
typedef typename _Sequence::reference reference;
typedef typename _Sequence::const_reference const_reference;
CRITICAL_SECTION CriticalSection;// Lock for modification
HANDLE hSynEvent; // Empty queue condition
public:
qlock() {
// Initialize the critical section one time only.
InitializeCriticalSection(&CriticalSection);
hSynEvent = CreateSemaphore(NULL, 0, QUEUE_MAX_SIZE, NULL);
if (hSynEvent == NULL)
{
printf("CreateEvent failed: %d\n", GetLastError());
exit(1);
}
};
~qlock() {
// Release resources used by the critical section object.
DeleteCriticalSection(&CriticalSection);
EXIT_IF(CloseHandle(hSynEvent));
};
void enqueue(const value_type& __x) {
EnterCriticalSection(&CriticalSection); // Lock the queue
if (this->size() QUEUE_MAX_SIZE)
{
while(this->size() 1) // Clear the queue
{
if(!this->empty()) {
value_type r;
r = this->front();
this->pop();
if(r != NULL) delete r;
}
}
}
this->push(__x);
// Release Semaphore
ReleaseSemaphore(hSynEvent, 1, NULL);
LeaveCriticalSection(&CriticalSection); // Unlock the queue
};
//reference
value_type dequeue(int i=0,int j=0) { // Blocking-call
bool empty = true;
value_type r = {0};
EnterCriticalSection(&CriticalSection);
empty = this->empty();
if(!empty) {
r = this->front();
this->pop();
}
LeaveCriticalSection(&CriticalSection);
while (empty) {
// Wait until something is written into the queue
if(WaitForSingleObject(hSynEvent, INFINITE) == WAIT_FAILED)
{
printf("WaitForSingleObject failed (%d)\n", GetLastError());
exit(1);
};
EnterCriticalSection(&CriticalSection);
empty = this->empty();
if(!empty) {
r = this->front();
this->pop();
}
LeaveCriticalSection(&CriticalSection);
};
return r;
}
const_reference dequeue() const { // Blocking-call
bool empty = true;
const_reference r;
EnterCriticalSection(&CriticalSection);
empty = this->empty();
if(!empty) {
r = this->front();
this->pop();
}
LeaveCriticalSection(&CriticalSection);
while (empty) {
// Wait until something is written into the queue
if(WaitForSingleObject(hSynEvent, INFINITE) == WAIT_FAILED)
{
printf("WaitForSingleObject failed (%d)\n", GetLastError());
LOG(LV_EMERG,"Wait for Single Queue Object failed.",1);
exit(1);
};
EnterCriticalSection(&CriticalSection);
empty = this->empty();
if(!empty) {
r = this->front();
this->pop();
}
LeaveCriticalSection(&CriticalSection);
};
return r;
}
int qSize() {
int size = 0;
EnterCriticalSection(&CriticalSection);
size = this->size();
LeaveCriticalSection(&CriticalSection);
return size;
}
};
