Re: Accessing stl queue in worker thread
R Samuel Klatchko
> I've created a thread safe queue using the STL queue, in the main thread I
> push objects onto the queue and within my thread I interrogate the queue to
> see if the size increases and if so I then pop the object off and perform
> something.
>
> My problem seems to be simple in that I can't seem to access the same queue
> pointer from within my thread as I do within my main thread.
>
> In my main thread as I pop objects onto the queue, I can see the size
> increase but whenever I interrogate the queue from the worker thread it
> always returns zero....and I'm sure I'm doing something stupidly simple but
> i can't seem to see what that is...can someone point out my stupidity.
>
> ....LoggingServiceSource.h...
> //--------------------------------------------------------------------------
> #ifndef LoggingServiceSourceH
> #define LoggingServiceSourceH
> //--------------------------------------------------------------------------
> #include <SysUtils.hpp>
> #include <Classes.hpp>
> #include <SvcMgr.hpp>
> #include <vcl.h>
> #include "thdLoggingSource.h"
> #include <OleServer.hpp>
> #include <queue>
> using namespace std;
> //--------------------------------------------------------------------------
> TLogQueue<LogStruct> LogQueue; //declaring my queue
> //--------------------------------------------------------------------------
You have the definition of the queue in a header file. That means each
file that includes this header file see a different definition. I'm not
even sure why this is linking, perhaps some Microsoft magic that I don't
know.
Try putting a declaration in the header file:
extern TLogQueue<LogStruct> LogQueue;
And putting the definition in a source file. If you want a more modern
solution, you can investigate wrapping your LogQueue in a singleton object.
samuel
[ See http://www.gotw.ca/resources/clcm.htm for info about ]
[ comp.lang.c++.moderated. First time posters: Do this! ]
Wheasle
push objects onto the queue and within my thread I interrogate the queue to
see if the size increases and if so I then pop the object off and perform
something.
My problem seems to be simple in that I can't seem to access the same queue
pointer from within my thread as I do within my main thread.
In my main thread as I pop objects onto the queue, I can see the size
increase but whenever I interrogate the queue from the worker thread it
always returns zero....and I'm sure I'm doing something stupidly simple but
i can't seem to see what that is...can someone point out my stupidity.
I have summarized by code below...
...LogQueueSource.h file...
//--------------------------------------------------------------------------
-
#ifndef TLogQueueSourceH
#define TLogQueueSourceH
//--------------------------------------------------------------------------
-
#include <queue>
//--------------------------------------------------------------------------
-
//--------------------------------------------------------------------------
-
typedef struct LogStruct
{
long StatusType;
AnsiString Origin;
long Level;
AnsiString Description;
TDateTime DateStamp;
TDateTime TimeStamp;
AnsiString CallNumber;
AnsiString DisplayMessage;
long Priority;
long PagingID;
AnsiString Subject;
AnsiString Body;
long IndicationCode;
bool BreakThrough;
long ConfirmationType;
bool IsErase;
long TimeToLive;
bool AllowRemoteDelete;
} LogInfo;
//--------------------------------------------------------------------------
-
class CriticalSection
{
private:
CRITICAL_SECTION m_cs;
public:
CriticalSection()
{
::InitializeCriticalSection(&m_cs);
}
CriticalSection(const CriticalSection& rhs)
{
// ignore rhs
::InitializeCriticalSection(&m_cs);
}
~CriticalSection()
{
::DeleteCriticalSection(&m_cs);
}
CriticalSection& operator=(const CriticalSection& rhs)
{
// ignore rhs & do nothing
return *this;
}
void Enter()
{
::EnterCriticalSection(&m_cs);
}
void Leave()
{
::LeaveCriticalSection(&m_cs);
}
};
//--------------------------------------------------------------------------
-
class CriticalSectionLock
{
private:
CriticalSection& m_cs;
// unimplemented
CriticalSectionLock& operator=(const CriticalSectionLock&);
CriticalSectionLock(const CriticalSectionLock&);
public:
CriticalSectionLock(CriticalSection& cs)
: m_cs(cs)
{
m_cs.Enter();
}
~CriticalSectionLock()
{
m_cs.Leave();
}
};
//--------------------------------------------------------------------------
-
template <class T>
class TLogQueue
{
private:
std::queue<T> m_q;
CriticalSection m_cs;
public:
typedef T value_type;
// .. other typedefs as needed
void push(const value_type& v)
{
CriticalSectionLock lock(m_cs);
m_q.push(v);
}
void pop()
{
CriticalSectionLock lock(m_cs);
m_q.pop();
}
unsigned int size()
{
CriticalSectionLock lock(m_cs);
unsigned int size = m_q.size();
return size;
}
const value_type& front() const
{
return m_q.front();
}
const value_type& top() const
{
// CriticalSectionLock lock(m_cs);
return m_q.top();
}
};
//--------------------------------------------------------------------------
-
#endif
I don't think there is anything wrong with the above declaration, I included
this to reduce the number of questions in this area.
I think the problem is in how I've declared my queue but I'm having a brain
freeze and can't think straight to work out my problem as seen below.
....LoggingServiceSource.h...
//--------------------------------------------------------------------------
-
#ifndef LoggingServiceSourceH
#define LoggingServiceSourceH
//--------------------------------------------------------------------------
-
#include <SysUtils.hpp>
#include <Classes.hpp>
#include <SvcMgr.hpp>
#include <vcl.h>
#include "thdLoggingSource.h"
#include <OleServer.hpp>
#include <queue>
using namespace std;
//--------------------------------------------------------------------------
-
TLogQueue<LogStruct> LogQueue; //declaring my queue
//--------------------------------------------------------------------------
-
class TLoggingService : public TService
{
__published: // IDE-managed Components
TatSystemStatus *atSystemStatus;
TatPagingSubscription *atPagingSubscription;
void __fastcall atSystemStatusSystemStatus(TObject *Sender,
atSystemStatusType StatusType, BSTR Origin,
atSystemStatusLevel Level, BSTR Description);
void __fastcall atPagingSubscriptionPaging(TObject *Sender,
BSTR CallNumber, BSTR DisplayMessage, long Priority,
PagingInfoStruct *AdditionalInfo);
private: // User declarations
TLoggingThread *LoggingThread;
String ConnectionStatus;
void __fastcall UpdateServiceDescription(TService *Sender);
void __fastcall CreateWorkerThread();
void __fastcall CheckSystemStatus();
public: // User declarations
__fastcall TLoggingService(TComponent* Owner);
TServiceController __fastcall GetServiceController(void);
friend void __stdcall ServiceController(unsigned CtrlCode);
};
//--------------------------------------------------------------------------
-
extern PACKAGE TLoggingService *LoggingService;
//--------------------------------------------------------------------------
-
#endif
...LoggingServiceSource.cpp....
//--------------------------------------------------------------------------
-
#include "LoggingServiceSource.h"
#include <deque>
#include "LogQueueSource.h"
//--------------------------------------------------------------------------
-
#pragma package(smart_init)
#pragma resource "*.dfm"
TLoggingService *LoggingService;
//--------------------------------------------------------------------------
-
__fastcall TLoggingService::TLoggingService(TComponent* Owner)
: TService(Owner)
{
CreateWorkerThread();
CheckSystemStatus();
}
//--------------------------------------------------------------------------
-
TServiceController __fastcall TLoggingService::GetServiceController(void)
{
return (TServiceController) ServiceController;
}
//--------------------------------------------------------------------------
-
void __stdcall ServiceController(unsigned CtrlCode)
{
LoggingService->Controller(CtrlCode);
}
//--------------------------------------------------------------------------
-
void __fastcall TLoggingService::atSystemStatusSystemStatus(TObject *Sender,
atSystemStatusType StatusType, BSTR Origin,
atSystemStatusLevel Level, BSTR Description)
{
//log all this information in the queue
LogInfo NewLog;
NewLog.StatusType = StatusType;
NewLog.Origin = Origin;
NewLog.Level = Level;
NewLog.Description = Description;
NewLog.DateStamp = TDateTime::CurrentDate();
NewLog.TimeStamp = TDateTime::CurrentTime();
LogQueue.push(NewLog);
unsigned int size = LogQueue.size(); //this value is
incremented correctly after each push
}
//--------------------------------------------------------------------------
-
void __fastcall TLoggingService::CreateWorkerThread()
{
LoggingThread = new TLoggingThread(false);
LoggingThread->FreeOnTerminate = true;
LoggingThread->Resume();
}
//--------------------------------------------------------------------------
-
void __fastcall TLoggingService::CheckSystemStatus()
{
atSystemStatus->QueryConnectionStatus();
}
//--------------------------------------------------------------------------
-
all of the above seems to work fine (please ignore the fact that I will have
problems with my struct definition and what is actually stored on the queue,
I am more interested in that fact that when something is stored on the queue
I can see that from my worker thread as seem below...
...LoggingSource.h...
//--------------------------------------------------------------------------
-
#ifndef thdLoggingSourceH
#define thdLoggingSourceH
//--------------------------------------------------------------------------
-
#include <Classes.hpp>
#include "LogQueueSource.h"
//--------------------------------------------------------------------------
-
class TLoggingThread : public TThread
{
private:
void __fastcall ProcessQueueData();
protected:
void __fastcall Execute();
public:
__fastcall TLoggingThread(bool CreateSuspended);
};
//--------------------------------------------------------------------------
-
#endif
...LoggingSource.cpp...
//--------------------------------------------------------------------------
-
#include <vcl.h>
#pragma hdrstop
#include "thdLoggingSource.h"
#include "frmLoggingServiceSource.h"
#include "LogQueueSource.h"
#pragma package(smart_init)
//--------------------------------------------------------------------------
-
__fastcall TLoggingThread::TLoggingThread(bool CreateSuspended)
: TThread(CreateSuspended)
{
}
//--------------------------------------------------------------------------
-
void __fastcall TLoggingThread::Execute()
{
while (!Terminated)
{
unsigned int size = LogQueue.size(); //this returns zero no
matter how big the queue is in the main thread
if (size > 0)
{
int i = 0;
i = i;
//ProcessQueueData();
}
else
{
Sleep(5000);
}
}
}
//--------------------------------------------------------------------------
-
void __fastcall TLoggingThread::ProcessQueueData()
{
LogInfo NewLog = LogQueue.front();
LogQueue.pop();
///do something here
}
//--------------------------------------------------------------------------
-
Maxim Yegorushkin
> I've created a thread safe queue using the STL queue, in the main thread
> I
> push objects onto the queue and within my thread I interrogate the queue
> to
> see if the size increases and if so I then pop the object off and perform
> something.
[]
> I don't think there is anything wrong with the above declaration, I
> included
> this to reduce the number of questions in this area.
> I think the problem is in how I've declared my queue but I'm having a
> brain
> freeze and can't think straight to work out my problem as seen below.
IMO, the main problem is that such a thread safe container is pretty much
useless.
Consider the typical usage:
TLogQueue<Whatever> q;
thread 1:
q.push(whatever);
thread 2:
Whatever whatever(q.top()); // 1
q.pop(); // 2
While your container's member functions use a lock, this kind of locking
is too fine-grained. For thread 2 to atomically pop an element from the
queue it needs external locking as the action requires calling two member
functions atomically. Of course, this particular problem can be fixed by
changing queue's pop function:
void pop(T* t)
{
CriticalSectionLock lock(m_cs);
*t = this->top();
this->pop();
}
But in general, most of the problems can not be fixed this way. Most of
the time you certainly want external locking.
--
Maxim Yegorushkin
James Kanze
> Wheasle <whea...@yahoo.com> wrote:
>>I've created a thread safe queue using the STL queue, in the
>>main thread I push objects onto the queue and within my thread
>>I interrogate the queue to see if the size increases and if so
>>I then pop the object off and perform something.
> []
>>I don't think there is anything wrong with the above
>>declaration, I included this to reduce the number of questions
>>in this area. I think the problem is in how I've declared my
>>queue but I'm having a brain freeze and can't think straight
>>to work out my problem as seen below.
> IMO, the main problem is that such a thread safe container is
> pretty much useless.
99% of the time, I'd agree with you. Not that that is is the
main problem -- the main problem here is that he defined a
variable in a header file:-). But that 99% of the time, such
fine grained locking is pretty much useless.
As it happens, however, he specifically spoke of using the queue
to transmit data from one thread to another. Which probably
falls into the 1%; although I don't do it quite the way he does
(I use a queue of pointers, and use auto_ptr in the interface),
this is a case where the queue itself should handle the locking.
> Consider the typical usage:
> TLogQueue<Whatever> q;
> thread 1:
> q.push(whatever);
> thread 2:
> Whatever whatever(q.top()); // 1
> q.pop(); // 2
> While your container's member functions use a lock, this kind
> of locking is too fine-grained. For thread 2 to atomically
> pop an element from the queue it needs external locking as the
> action requires calling two member functions atomically.
The most serious problem in his case here was that he was
returning references. Regardless of how often such thread safe
queues are appropriate, they are not thread safe if they return
a reference, rather than a copy.
If the data involved is small, and consists only of basic types,
using copies everywhere is a valid solution (supposing you're in
the 1% of the cases where this idiom applies); of course, in
that case, you'll probably want pop to return the value
(although it isn't essential). For larger or more complex data,
it's probably preferable to pass around a pointer.
> Of course, this particular problem can be fixed by changing
> queue's pop function:
> void pop(T* t)
> {
> CriticalSectionLock lock(m_cs);
> *t = this->top();
> this->pop();
> }
> But in general, most of the problems can not be fixed this
> way. Most of the time you certainly want external locking.
In general, just making a standard container "thread safe" is
meaningless. Or impossible, since they do return references to
internal elements. That doesn't mean that there can't be cases
where what is basically a container should be thread safe; a
queue to communicate between threads would be a good example.
His application seems to be more of this second category,
although I'm not quite sure that he realizes the distinction.
--
James Kanze mailto: james...@free.fr
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 pl. Pierre Sémard, 78210 St.-Cyr-l'École, France +33 (0)1 30 23 00 34
James Kanze
> I've created a thread safe queue using the STL queue, in the
> main thread I push objects onto the queue and within my thread
> I interrogate the queue to see if the size increases and if so
> I then pop the object off and perform something.
> My problem seems to be simple in that I can't seem to access
> the same queue pointer from within my thread as I do within my
> main thread.
You're not accessing the same queue, as far as I can see.
> In my main thread as I pop objects onto the queue, I can see
> the size increase but whenever I interrogate the queue from
> the worker thread it always returns zero....and I'm sure I'm
> doing something stupidly simple but i can't seem to see what
> that is...can someone point out my stupidity.
> I have summarized by code below...
That's a lot of code, with a lot of system dependencies. I'm
surprised the moderators let it past. Still, there are two
definite errors in it (at least if we suppose that the
undocumented semantics of Windows threads are roughly the same
as those for Posix threads).
> ...LogQueueSource.h file...
[...]
> template <class T>
> class TLogQueue
> {
> private:
> std::queue<T> m_q;
> CriticalSection m_cs;
>
> public:
> typedef T value_type;
> // .. other typedefs as needed
> void push(const value_type& v)
> {
> CriticalSectionLock lock(m_cs);
> m_q.push(v);
> }
> void pop()
> {
> CriticalSectionLock lock(m_cs);
> m_q.pop();
> }
> unsigned int size()
> {
> CriticalSectionLock lock(m_cs);
> unsigned int size = m_q.size();
>
> return size;
> }
> const value_type& front() const
> {
You need a lock here. I'm also suspicious of returning a
reference in this case -- accesses through the reference are NOT
locked; if another thread calls pop() at the wrong moment,
you're in deep trouble.
(My own preference for passing objects between threads is to use
auto_ptr at the interface. That way, the sending thread
automatically looses access to the object, and the receiving
thread acquires exclusive access.)
> return m_q.front();
> }
> const value_type& top() const
> {
> // CriticalSectionLock lock(m_cs);
Same thing here. Why did you comment out the lock? And it is
very risky returning references.
> return m_q.top();
> }
> };
>
//--------------------------------------------------------------------------
> -
> #endif
Note that the above errors won't manifest themselves
systematically, so that you can effectively test for them. Just
once in a while, something funny will happen, that you won't be
able to reproduce.
> I don't think there is anything wrong with the above
> declaration, I included this to reduce the number of questions
> in this area. I think the problem is in how I've declared my
> queue but I'm having a brain freeze and can't think straight
> to work out my problem as seen below.
> ....LoggingServiceSource.h...
A .h, thus included separately in each of the .cpp files.
[...]
> TLogQueue<LogStruct> LogQueue; //declaring my queue
Not only declaring, but defining. Each source file which
includes this header will have its own instance of LogQueue.
What you probably want is:
extern TLogQueue<LogStruct> LogQueue ;
with a declaration (without the extern) in a single source file.
This error, of course, explains perfectly the symptoms you are
seeing.
Also, a personal preference (but more in keeping with the
philosophy of the group): abstract the system dependant parts
into classes which you don't post (or even better, use the Boost
equivalents), and cut out the extensions (like __fastcall)
before posting. For people like myself, who don't even have
access to your platform, it reduces the noise, and makes it much
easier to evaluate your code.
--
James Kanze mailto: james...@free.fr
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 pl. Pierre Sémard, 78210 St.-Cyr-l'École, France +33 (0)1 30 23 00 34
[ See http://www.gotw.ca/resources/clcm.htm for info about ]
Maxim Yegorushkin
[]
>> IMO, the main problem is that such a thread safe container is
>> pretty much useless.
>
> 99% of the time, I'd agree with you. Not that that is is the
> main problem -- the main problem here is that he defined a
> variable in a header file:-). But that 99% of the time, such
> fine grained locking is pretty much useless.
>
> As it happens, however, he specifically spoke of using the queue
> to transmit data from one thread to another. Which probably
> falls into the 1%; although I don't do it quite the way he does
> (I use a queue of pointers, and use auto_ptr in the interface),
> this is a case where the queue itself should handle the locking.
For the remaining 1% I use a queue like container based on std::list. I
found that list::splice function is unique across standard containers and
it allows one to split node allocation and actual insertion. This helps to
do as little inside critical sections in the queue as possible. The only
assumption is that list::splice really moves a node between lists, rather
than copying it, and I believe any sane implementation do so. Here is the
code:
template<class T>
class atomic_queue : private boost::noncopyable
{
private:
typedef std::list<T> items_type;
public:
struct cancelled : std::logic_error
{
cancelled() : std::logic_error("cancelled") {}
};
public:
atomic_queue() : cancel_(false) {}
public:
void cancel()
{
++cancel_;
cnd_.notify_all();
}
public:
void push(T const& t)
{
// this avoids an allocation inside the critical section bellow
items_type tmp(&t, &t + 1);
{
boost::mutex::scoped_lock l(mtx_);
items_.splice(items_.end(), tmp);
}
cnd_.notify_one();
}
// this function provides only basic exception safety if T's copy ctor
can
// throw or strong exception safety if T's copy ctor is nothrow
T pop()
{
// this avoids copying T inside the critical section bellow
items_type tmp;
{
boost::mutex::scoped_lock l(mtx_);
while(!cancel_ && items_.empty())
cnd_.wait(l);
if(cancel_)
throw cancelled();
tmp.splice(tmp.end(), items_, items_.begin());
}
return tmp.front();
}
private:
boost::mutex mtx_;
boost::condition cnd_;
boost::detail::atomic_count cancel_;
items_type items_;
};
--
Maxim Yegorushkin
James Kanze
> On 27 Mar 2005 05:28:04 -0500, James Kanze <ka...@none.news.free.fr>
wrote:
> []
>>>IMO, the main problem is that such a thread safe container is
>>>pretty much useless.
>>99% of the time, I'd agree with you. Not that that is is the
>>main problem -- the main problem here is that he defined a
>>variable in a header file:-). But that 99% of the time, such
>>fine grained locking is pretty much useless.
>>As it happens, however, he specifically spoke of using the
>>queue to transmit data from one thread to another. Which
>>probably falls into the 1%; although I don't do it quite the
>>way he does (I use a queue of pointers, and use auto_ptr in
>>the interface), this is a case where the queue itself should
>>handle the locking.
> For the remaining 1% I use a queue like container based on
> std::list. I found that list::splice function is unique across
> standard containers and it allows one to split node allocation
> and actual insertion. This helps to do as little inside
> critical sections in the queue as possible. The only
> assumption is that list::splice really moves a node between
> lists, rather than copying it, and I believe any sane
> implementation do so. Here is the code:
[code deleted...]
I'm not sure that this is optimal. You still have to allocate
the nodes, and it is probable that the implementation locks in
the allocation routines as well. My own solution here has been
to use a dequeue -- once the maximum size has been attained,
there is no more allocation, and the lock protects a few simple
pointer manipulations.
--
James Kanze mailto: james...@free.fr
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 pl. Pierre Sémard, 78210 St.-Cyr-l'École, France +33 (0)1 30 23 00 34
[ See http://www.gotw.ca/resources/clcm.htm for info about ]
Maxim Yegorushkin
>> For the remaining 1% I use a queue like container based on
>> std::list. I found that list::splice function is unique across
>> standard containers and it allows one to split node allocation
>> and actual insertion. This helps to do as little inside
>> critical sections in the queue as possible. The only
>> assumption is that list::splice really moves a node between
>> lists, rather than copying it, and I believe any sane
>> implementation do so. Here is the code:
>
> [code deleted...]
>
> I'm not sure that this is optimal. You still have to allocate
> the nodes, and it is probable that the implementation locks in
> the allocation routines as well.
Somebody has to allocate the nodes somehow. It's still possible to use a
custom allocator.
> My own solution here has been
> to use a dequeue -- once the maximum size has been attained,
> there is no more allocation, and the lock protects a few simple
> pointer manipulations.
I think the most efficient way for a dynamic (rather than a fixed sized)
locking queue is to use an intrusive list. That is, a user allocates node
successors and puts (push()) them in the queue passing ownership. push()
does only trivial list node insertion, pop() is implemented as a trivial
list node deletion. That way you only get one allocation per element and
avoid copying. Of course, a preallocated fixed size queue is superior in
terms of performance, but it sometimes hard to predict or constrain the
queue size beforehand.
--
Maxim Yegorushkin
James Kanze
> On 28 Mar 2005 18:55:37 -0500, James Kanze <ka...@none.news.free.fr>
wrote:
>>>For the remaining 1% I use a queue like container based on
>>>std::list. I found that list::splice function is unique
>>>across standard containers and it allows one to split node
>>>allocation and actual insertion. This helps to do as little
>>>inside critical sections in the queue as possible. The only
>>>assumption is that list::splice really moves a node between
>>>lists, rather than copying it, and I believe any sane
>>>implementation do so. Here is the code:
>> [code deleted...]
>>I'm not sure that this is optimal. You still have to allocate
>>the nodes, and it is probable that the implementation locks in
>>the allocation routines as well.
> Somebody has to allocate the nodes somehow. It's still
> possible to use a custom allocator.
>>My own solution here has been to use a dequeue -- once the
>>maximum size has been attained, there is no more allocation,
>>and the lock protects a few simple pointer manipulations.
> I think the most efficient way for a dynamic (rather than a
> fixed sized) locking queue is to use an intrusive list.
Possibly. It was generally my preferred solution in the past.
Today, it suffers from a lack of support from the standard
library, although I still use it in particular cases. (Not
related to threads, but you can easily arrange for the delete of
an object to automatically remove it from the list.)
> That is, a user allocates node successors and puts (push())
> them in the queue passing ownership. push() does only trivial
> list node insertion, pop() is implemented as a trivial list
> node deletion. That way you only get one allocation per
> element and avoid copying. Of course, a preallocated fixed
> size queue is superior in terms of performance, but it
> sometimes hard to predict or constrain the queue size
> beforehand.
But typically, in most applications, the queue rapidly hits a
maximum size, or at least, most of the insertions will occur
when there is still place in the queue. And if the deque
contains pointers, there is still just one allocation.
But I think we agree that there isn't one universal solution,
which is best in every case.
--
James Kanze mailto: james...@free.fr
Conseils en informatique orientée objet/
Beratung in objektorientierter Datenverarbeitung
9 pl. Pierre Sémard, 78210 St.-Cyr-l'École, France +33 (0)1 30 23 00 34
[ See http://www.gotw.ca/resources/clcm.htm for info about ]