freepascal/Delphi thread-safe queue?

[paul_nicholls]

Hi all,
I have some code for a thread-safe queue that works just fine in Delphi, but I want see if I can make it work under freepascal, and if possible, cross-platform compatible too which would be even better :)

Any ideas?

PS. I haven't actually tried compiling it under freepascal yet, but I have suspicions that it won't due to the units being used...


Unit threadsafe_queue;
{$IFDEF fpc}
{$MODE DELPHI} {$H+}
{$ENDIF}
Interface

Uses
Windows,
Contnrs,
Syncobjs;
{
As many consumer threads as you like can wait on the queue with no CPU
waste. When objects come in, waiting threads are woken up. Note that the
'pop' method takes a pointer to the variable where an incoming object is to
be placed, (avoids need for casting):

Var
inObject: TobjectOfSomeClass;
..
..
if (myThreadSafeQueue.pop(@inObject,INFINITE))then
handleObject
else
// only handled if timeout not INFINITE !!
handleTimeout;
}

Type
{................................................. .............................}
PObject = ^TObject;
TThreadSafeQueue = Class(TObjectQueue)
Private
FCountSema : Thandle;
FAccess : TCriticalSection;
Public
Constructor Create;
Procedure Push(Const AObject : TObject);
Function Pop (Const AObject : PObject;
Const Timeout : LongWord) : Boolean;
Destructor Destroy; Override;
End;
{................................................. .............................}

Implementation

{................................................. .............................}

{................................................. .............................}
Constructor TThreadSafeQueue.Create;
Begin
Inherited create;

FAccess := TCriticalSection.Create;
FCountSema := CreateSemaphore(Nil,0,MaxInt,Nil);
End;
{................................................. .............................}

{................................................. .............................}
Destructor TThreadSafeQueue.Destroy;
Begin
CloseHandle(FCountSema);
FAccess.Free;

Inherited Destroy;
End;
{................................................. .............................}

{................................................. .............................}
Function TThreadSafeQueue.Pop(Const AObject : PObject;
Const Timeout : LongWord) : Boolean;
Begin
Result := (WAIT_OBJECT_0 = WaitForSingleObject(FCountSema,Timeout));
If Result Then
Begin
FAccess.Acquire;
Try
AObject^ := Inherited Pop;
Finally
FAccess.Release;
End;
End;
End;
{................................................. .............................}

{................................................. .............................}
Procedure TThreadSafeQueue.Push(Const AObject : TObject);
Begin
FAccess.Acquire;
Try
Inherited Push(AObject);
Finally
FAccess.Release;
End;
ReleaseSemaphore(FCountSema,1,Nil);
End;
{................................................. .............................}

{................................................. .............................}
End.

Cheers,
Paul

[paul_nicholls]

Update: I can compile it using Lazarus under Windows ok

As I expected, when I try compiling it for the arm-linux GP2X handheld, it doesn't compile as it uses the windows Windows unit :(

Any ideas on a replacement?
cheers,
Paul

[technomage]

Paul

Nice work , not sure about a replacement for Windows, depending on what you use from it you might just be able to $DEFINE it out. I suspect you will need to fine a free pascal cross platform alternative for WaitForSingleObject and THandle as this will definately not work under linux.

If I get a chance I'll have a quick look for you :)

Dean

[paul_nicholls]

Paul

Nice work , not sure about a replacement for Windows, depending on what you use from it you might just be able to $DEFINE it out. I suspect you will need to fine a free pascal cross platform alternative for WaitForSingleObject and THandle as this will definately not work under linux.

If I get a chance I'll have a quick look for you :)

Dean

Thanks Dean :-)
In the meanwhile I will see what I can find too...
cheers,
Paul

[AthenaOfDelphi]

Hi Paul,

From what I know of semaphores, your using it to stop the queue growing too large... correct?

If this is the case, ditch it. You can maintain the count yourself within the critical section. Get the critical section, check the count and then either add and return true or return false, then leave the critical section.

I believe, TCriticalSection is well supported and so you shouldn't have any problems converting to other platforms/compilers.

[JSoftware]

Why don't you use the inbuilt semaphore handling in FPC? You don't have access to any timeouts when aquiring locks but you can do that by some combination of Lock and a timer

[paul_nicholls]

Why don't you use the inbuilt semaphore handling in FPC? You don't have access to any timeouts when aquiring locks but you can do that by some combination of Lock and a timer

Not sure about the built-in semaphore stuff in FPC, but I found this page http://community.freepascal.org:10000/bboards/message?message_id=219310&forum_id=24083 which says about WaitfForSingleObject equivalents under Linux...

cheers,
Paul

[paul_nicholls]

Hi Paul,

From what I know of semaphores, your using it to stop the queue growing too large... correct?

If this is the case, ditch it. You can maintain the count yourself within the critical section. Get the critical section, check the count and then either add and return true or return false, then leave the critical section.

I believe, TCriticalSection is well supported and so you shouldn't have any problems converting to other platforms/compilers.

I'm pretty sure TCriticalSection IS well supported as you say :-)
To be honest, I am not sure what the semaphore is being used for in the code (it isn't mine, but from someone on the Delphi newsgroups)

cheers,
Paul

[marcov]

Have a look at unit syncobjs for multiplatform synchronization objects.

On Unices, don't forget to add the cthreads unit as first unit in your project to plugin threading support.

[paul_nicholls]

Have a look at unit syncobjs for multiplatform synchronization objects.

On Unices, don't forget to add the cthreads unit as first unit in your project to plugin threading support.

Thanks for the info marcov :-)
cheers,
Paul