program Srv;
{$mode objfpc}
uses
{$IFDEF UNIX}
cthreads,
{$ENDIF}
Classes, Sysutils, syncobjs, blcksock,synsock;
type
TThreadManager = class;
{ TManagedThread }
TManagedThread = class(TThread)
public
constructor Create(waiting : Boolean);
function isDone() : Boolean;
function isErroneus() : Boolean;
protected
done_,
erroneous_ : Boolean;
end;
{ TTCPThread }
TTCPThread = class(TManagedThread)
private
fSock: TTCPBlockSocket;
fIP: string;
FPort: integer;
FNumber: integer;
procedure SetSocket(aSock: TSocket);
protected
procedure Execute; override;
public
constructor Create();
destructor Destroy; override;
procedure ProcessingData(procSock: TSocket;Data: string);
Property Number: integer read Fnumber Write FNumber;
end;
{ TListenerThread }
TListenerThread = class(TThread)
private
ListenerSocket: TTCPBlockSocket;
FThreadManager: TThreadManager;
protected
procedure Execute; override;
public
constructor Create;
destructor Destroy; override;
end;
{ TThreadManager }
TThreadManager = class(TObject)
private
FItemList: TThreadList;
FAbort: Boolean;
FThreadList: TList;
FMaxThreadCount: Integer;
procedure SetMaxThreadCount(Count: Integer);
public
constructor Create(MaxThreads: integer);
destructor Destroy; override;
procedure AddItem(Item: TTCPThread);
function GetSuspendThread(aSock: TSocket): TTCPThread;
procedure clearFinishedThreads;
function GetActiveThreadCount: Integer;
property MaxThreadCount: Integer read FMaxThreadCount write SetMaxThreadCount;
end;
{ TThreadManager }
procedure TThreadManager.SetMaxThreadCount(Count: Integer);
begin
FMaxThreadCount := Count;
end;
constructor TThreadManager.Create(MaxThreads: integer);
begin
inherited Create;
FItemList := TThreadList.Create;
FThreadList := TList.Create;
FMaxThreadCount := MaxThreads;
end;
destructor TThreadManager.Destroy;
var
i: Integer;
begin
FThreadList.Pack;
for i := FThreadList.Count - 1 downto 0 do begin
TTCPThread(FThreadList[i]).Free;
end;
FThreadList.Capacity := FThreadList.Count;
FThreadList.Free;
FItemList.Clear;
FItemList.Free;
inherited;
end;
procedure TThreadManager.AddItem(Item: TTCPThread);
begin
FItemList.Add(Pointer(Item));
end;
function TThreadManager.GetSuspendThread(aSock: TSocket): TTCPThread;
var
i: Integer;
TCPThread: TTCPThread;
begin
Result := nil;
if GetActiveThreadCount >= FMaxThreadCount then Exit;
for i := 0 to FThreadList.Count - 1 do begin
if TTCPThread(FThreadList[i]).Suspended then
begin
TCPThread := TTCPThread(FThreadList[i]);
TCPThread.SetSocket(aSock);
TCPThread.Resume;
Break;
end;
end;
if (Result = nil) and (FMaxThreadCount > FThreadList.Count) then begin
TCPThread := TTCPThread.Create;
TCPThread.FreeOnTerminate := False;
TCPThread.SetSocket(aSock);
TCPThread.Number := FThreadList.Count;
FThreadList.Add(TCPThread);
Result := TCPThread;
end;
end;
procedure TThreadManager.clearFinishedThreads;
var
i: Integer;
begin
for i := 0 to FThreadList.Count - 1 do
begin
if (TTCPThread(FThreadList[i]) <> nil) and TTCPThread(FThreadList[i]).isDone() then
begin
TTCPThread(FThreadList[i]).WaitFor;
TTCPThread(FThreadList[i]).Free;
end;
end;
end;
function TThreadManager.GetActiveThreadCount: Integer;
var
i: Integer;
begin
Result := 0;
for i := 0 to FThreadList.Count - 1 do begin
if not TTCPThread(FThreadList[i]).Suspended then
Inc(Result);
end;
end;
{ TManagedThread }
constructor TManagedThread.Create(waiting : Boolean);
begin
inherited Create(waiting);
done_ := false;
erroneous_ := false;
end;
function TManagedThread.isDone() : Boolean;
begin
Result := done_;
end;
function TManagedThread.isErroneus() : Boolean;
begin
Result := erroneous_;
end;
{ TListenerThread }
procedure TListenerThread.Execute;
var
ClientSock: TSocket;
ClientThread: TTCPThread;
begin
with ListenerSocket do
begin
CreateSocket;
if LastError = 0 then
WriteLn('Socket successfully initialized')
else
WriteLn('An error occurred while initializing the socket: '+GetErrorDescEx);
Family := SF_IP4;
setLinger(true,10000);
bind('0.0.0.0', '5050');
if LastError = 0 then
WriteLn('Bind on 5050')
else
WriteLn('Bind error: '+GetErrorDescEx);
listen;
repeat
if CanRead(100) then
begin
ClientSock := Accept;
if LastError = 0
then
begin
//TTCPThread.Create()
ClientThread:=FThreadManager.GetSuspendThread(ClientSock);
WriteLn('We have '+ IntToStr(FThreadManager.GetActiveThreadCount)+#32+'client threads!');
end
else
WriteLn('TCP thread creation error: '+GetErrorDescEx);
end;
FThreadManager.clearFinishedThreads;
sleep(10);
until false;
end;
end;
constructor TListenerThread.Create;
begin
FreeOnTerminate := True;
ListenerSocket := TTCPBlockSocket.Create;
FThreadManager:=TThreadManager.Create(20000);
if ListenerSocket.LastError = 0
then
WriteLn('Listener has been created')
else
WriteLn('Listener creation error: '+ListenerSocket.GetErrorDescEx);
inherited Create(False);
end;
destructor TListenerThread.Destroy;
begin
ListenerSocket.Free;
if
ListenerSocket.LastError = 0
then
WriteLn('Listener has been deleted')
else
WriteLn('Listener deleting error: '+ListenerSocket.GetErrorDescEx);
inherited;
end;
{ TTCPThread }
procedure TTCPThread.SetSocket(aSock: TSocket);
begin
fSock.Socket := aSock;
fSock.GetSins;
end;
procedure TTCPThread.Execute;
var
s: ansistring;
begin
fIp:=fSock.GetRemoteSinIP;
fPort:=fSock.GetRemoteSinPort;
WriteLn(format('Accepted connection from %s:%d',[fIp,fPort]));
while not isDone do
begin
if fSock.WaitingData > 0 then
begin
s:=fSock.RecvPacket(2000);
if fSock.LastError <> 0 then
WriteLn(fSock.GetErrorDescEx);
ProcessingData(fSock.Socket,S);
end;
sleep(10);
end;
end;
constructor TTCPThread.Create();
begin
FreeOnTerminate := True;
fSock := TTCPBlockSocket.Create;
inherited Create(false);
end;
destructor TTCPThread.Destroy;
begin
WriteLn(format('Disconnect from %s:%d',[fIp,fPort]));
fSock.Free;
inherited;
end;
procedure TTCPThread.ProcessingData(procSock: TSocket; Data: string);
begin
if data <> '' then
WriteLn(data+#32+'we get it from '+IntToStr(number)+' thread');
end;
var
Server: TListenerThread;
begin
Server:=TListenerThread.Create;
ReadLn;
end.
WriteLn(data+#32+'we get it from '+IntToStr(number)+' thread');Isn't #32 just space-character? :P It could be added like ' we get...'.
Great! Your code is a very valuable resource.
Would you mind to post it to the Lazarus and Free Pascal Wiki, e.g. at http://wiki.lazarus.freepascal.org/Synapse (http://wiki.lazarus.freepascal.org/Synapse)?
Hey guys. I have a lot of time looking for a suitable example for this, but could not find it. So I had to write its own implementation. One thread listens to a socket, and if it recieved packet he make processing thread. Tested by telnet on Win and *nix. I'll just leave this here in case someone else is looking for it.Code: [Select]program Srv;
{$mode objfpc}
uses
{$IFDEF UNIX}
cthreads,
{$ENDIF}
Classes, Sysutils, syncobjs, blcksock,synsock;
type
TThreadManager = class;
{ TManagedThread }
TManagedThread = class(TThread)
public
constructor Create(waiting : Boolean);
function isDone() : Boolean;
function isErroneus() : Boolean;
protected
done_,
erroneous_ : Boolean;
end;
{ TTCPThread }
TTCPThread = class(TManagedThread)
private
fSock: TTCPBlockSocket;
fIP: string;
FPort: integer;
FNumber: integer;
procedure SetSocket(aSock: TSocket);
protected
procedure Execute; override;
public
constructor Create();
destructor Destroy; override;
procedure ProcessingData(procSock: TSocket;Data: string);
Property Number: integer read Fnumber Write FNumber;
end;
{ TListenerThread }
TListenerThread = class(TThread)
private
ListenerSocket: TTCPBlockSocket;
FThreadManager: TThreadManager;
protected
procedure Execute; override;
public
constructor Create;
destructor Destroy; override;
end;
{ TThreadManager }
TThreadManager = class(TObject)
private
FItemList: TThreadList;
FAbort: Boolean;
FThreadList: TList;
FMaxThreadCount: Integer;
procedure SetMaxThreadCount(Count: Integer);
public
constructor Create(MaxThreads: integer);
destructor Destroy; override;
procedure AddItem(Item: TTCPThread);
function GetSuspendThread(aSock: TSocket): TTCPThread;
procedure clearFinishedThreads;
function GetActiveThreadCount: Integer;
property MaxThreadCount: Integer read FMaxThreadCount write SetMaxThreadCount;
end;
{ TThreadManager }
procedure TThreadManager.SetMaxThreadCount(Count: Integer);
begin
FMaxThreadCount := Count;
end;
constructor TThreadManager.Create(MaxThreads: integer);
begin
inherited Create;
FItemList := TThreadList.Create;
FThreadList := TList.Create;
FMaxThreadCount := MaxThreads;
end;
destructor TThreadManager.Destroy;
var
i: Integer;
begin
FThreadList.Pack;
for i := FThreadList.Count - 1 downto 0 do begin
TTCPThread(FThreadList[i]).Free;
end;
FThreadList.Capacity := FThreadList.Count;
FThreadList.Free;
FItemList.Clear;
FItemList.Free;
inherited;
end;
procedure TThreadManager.AddItem(Item: TTCPThread);
begin
FItemList.Add(Pointer(Item));
end;
function TThreadManager.GetSuspendThread(aSock: TSocket): TTCPThread;
var
i: Integer;
TCPThread: TTCPThread;
begin
Result := nil;
if GetActiveThreadCount >= FMaxThreadCount then Exit;
for i := 0 to FThreadList.Count - 1 do begin
if TTCPThread(FThreadList[i]).Suspended then
begin
TCPThread := TTCPThread(FThreadList[i]);
TCPThread.SetSocket(aSock);
TCPThread.Resume;
Break;
end;
end;
if (Result = nil) and (FMaxThreadCount > FThreadList.Count) then begin
TCPThread := TTCPThread.Create;
TCPThread.FreeOnTerminate := False;
TCPThread.SetSocket(aSock);
TCPThread.Number := FThreadList.Count;
FThreadList.Add(TCPThread);
Result := TCPThread;
end;
end;
procedure TThreadManager.clearFinishedThreads;
var
i: Integer;
begin
for i := 0 to FThreadList.Count - 1 do
begin
if (TTCPThread(FThreadList[i]) <> nil) and TTCPThread(FThreadList[i]).isDone() then
begin
TTCPThread(FThreadList[i]).WaitFor;
TTCPThread(FThreadList[i]).Free;
end;
end;
end;
function TThreadManager.GetActiveThreadCount: Integer;
var
i: Integer;
begin
Result := 0;
for i := 0 to FThreadList.Count - 1 do begin
if not TTCPThread(FThreadList[i]).Suspended then
Inc(Result);
end;
end;
{ TManagedThread }
constructor TManagedThread.Create(waiting : Boolean);
begin
inherited Create(waiting);
done_ := false;
erroneous_ := false;
end;
function TManagedThread.isDone() : Boolean;
begin
Result := done_;
end;
function TManagedThread.isErroneus() : Boolean;
begin
Result := erroneous_;
end;
{ TListenerThread }
procedure TListenerThread.Execute;
var
ClientSock: TSocket;
ClientThread: TTCPThread;
begin
with ListenerSocket do
begin
CreateSocket;
if LastError = 0 then
WriteLn('Socket successfully initialized')
else
WriteLn('An error occurred while initializing the socket: '+GetErrorDescEx);
Family := SF_IP4;
setLinger(true,10000);
bind('0.0.0.0', '5050');
if LastError = 0 then
WriteLn('Bind on 5050')
else
WriteLn('Bind error: '+GetErrorDescEx);
listen;
repeat
if CanRead(100) then
begin
ClientSock := Accept;
if LastError = 0
then
begin
//TTCPThread.Create()
ClientThread:=FThreadManager.GetSuspendThread(ClientSock);
WriteLn('We have '+ IntToStr(FThreadManager.GetActiveThreadCount)+#32+'client threads!');
end
else
WriteLn('TCP thread creation error: '+GetErrorDescEx);
end;
FThreadManager.clearFinishedThreads;
sleep(10);
until false;
end;
end;
constructor TListenerThread.Create;
begin
FreeOnTerminate := True;
ListenerSocket := TTCPBlockSocket.Create;
FThreadManager:=TThreadManager.Create(20000);
if ListenerSocket.LastError = 0
then
WriteLn('Listener has been created')
else
WriteLn('Listener creation error: '+ListenerSocket.GetErrorDescEx);
inherited Create(False);
end;
destructor TListenerThread.Destroy;
begin
ListenerSocket.Free;
if
ListenerSocket.LastError = 0
then
WriteLn('Listener has been deleted')
else
WriteLn('Listener deleting error: '+ListenerSocket.GetErrorDescEx);
inherited;
end;
{ TTCPThread }
procedure TTCPThread.SetSocket(aSock: TSocket);
begin
fSock.Socket := aSock;
fSock.GetSins;
end;
procedure TTCPThread.Execute;
var
s: ansistring;
begin
fIp:=fSock.GetRemoteSinIP;
fPort:=fSock.GetRemoteSinPort;
WriteLn(format('Accepted connection from %s:%d',[fIp,fPort]));
while not isDone do
begin
if fSock.WaitingData > 0 then
begin
s:=fSock.RecvPacket(2000);
if fSock.LastError <> 0 then
WriteLn(fSock.GetErrorDescEx);
ProcessingData(fSock.Socket,S);
end;
sleep(10);
end;
end;
constructor TTCPThread.Create();
begin
FreeOnTerminate := True;
fSock := TTCPBlockSocket.Create;
inherited Create(false);
end;
destructor TTCPThread.Destroy;
begin
WriteLn(format('Disconnect from %s:%d',[fIp,fPort]));
fSock.Free;
inherited;
end;
procedure TTCPThread.ProcessingData(procSock: TSocket; Data: string);
begin
if data <> '' then
WriteLn(data+#32+'we get it from '+IntToStr(number)+' thread');
end;
var
Server: TListenerThread;
begin
Server:=TListenerThread.Create;
ReadLn;
end.
So my question is: how do I cleanup threads that are no longer in use?Simply set FreeOnTerminate to true
I'm using the code as shown on top of this thread (by CynicRus)..Not clear enough to me, how do you conclude that the thread counter is increasing despite to indicate that the thread is still alive after execute? Because I don't see any Dec against an ordinal var, nor a Remove against FThreadList (which seems to be used to get the number of threads). Note that dead thread won't get removed automatically from the list.
Hello,1) Is the FThreadlist of type TTcpThread? I do not have synapse at this time but I doubt it. Stop casting when its not needed eg
I have implemented a version of this Multithreaded Server, and I am experimenting a problem in this procedure:
procedure TThreadManager.clearFinishedThreads;
var
i: Integer;
begin
for i := 0 to FThreadList.Count - 1 do
begin
if (TTCPThread(FThreadList) <> nil) then
begin
if (TTCPThread(FThreadList.items).isDone = TRUE) then
begin
TTCPThread(FThreadList).WaitFor;
TTCPThread(FThreadList).Free;
end
end
end
end;
It does seem that it is unable to properly find the Thread (finished) that needs to be freed. It loops on each pointer in the FthreadList but is unable to find the one that is done and thus it never terminates it. As far I could understand, this doesn't trigger:
if (TTCPThread(FThreadList.items).isDone = TRUE)
I am unable to find out if this is supposed to work, but putting a breakpoint here never gets true for any thread.
That's the reason why the counter in the example never decrements, because the counter is a related to the number of items of the FThreadList (that never gets the relevant element removed).
Could you kindly tell me what is wrong?
Thanks in advance.
Giuseppe Marullo
@taazz, xardomain is doing all that.thanks rvk, I got weird out with all this italics but I am too bored to think it over I guess ;) . Well from a quick look around the srv unit, there is no logic to define when the thread is done, unless there is some code that I missed that property will always be false and rightfully so, there is no standard that defines when a tcp process is done every protocol on top of tcp has its own "is done" logic. In this case the user has to write a TTcpThread descendant override the ProcessingData method and in there if the data have the done marker on them set the done_ protected variable to true, in a thread safe manner.
But because the code isn't tagged with code, it isn't very clear (and [ i ] is filtered out as italic)
This is the code that's in the post:
The only thing I'm missing is setting the FThreadList[ i ] to nil when the thread is freed. Otherwise you end up doing this again when clearFinishedThreads is called again on a freed thread.
Fist of all, sorry for the mess about italics.
I assume that the Thread (one of the several I have, tied to a TCP connection) will stop executing.
When the code runs, I assume that looping on the list would at certain point meet the element(i) of the list that In reality is a pointer to that Thread.In the unit provided here there is a framework that manages data transfer through tcp for you but has no knowledge of the interpretation of that data. You need to code in that.
Now, why this construct will never be true for that element?
Yes and no. Yes its the correct property to use. No you do not check it your self, it is already part of the loop, you need to set to it to true to allow the consumer to exit its loop. If you have already set the free on terminate property to true also, it should automatically free the consumer for you when you set the protected field "Done_" to true your self.
Is correct way to test if a Thread which is pointed byshould be freed?
Keep in mind that my complete program is different, but it is based on this example. So we could talk specifically about this example that never "frees" any element from the ThreadList.Ok, so we established that this example never frees any elements.
Ok, but this test never works. Why? How could I detect that I need to free the corresponding thread that is ended?Like taazz already explained... you should build in tests to set the _done when you determine the connection is (or can be) ended. As it stands now the thread doesn't even check the "normal" TThread.Terminated in a while loop in TThread.Execute. It only checks for isDone. So if you want to terminate the thread from the outside with TThread.Terminate you also need to implement the check for TThread.Terminated (which is not in the example).
Ok, I think I mostly understand this, but I would like to free the Thread from this loop, and keep the FreeOnTerminate to false.You have much bigger problems than the FreeOnTerminate. As long as the thread/connection doesn't terminate it all doesn't matter. Once you fixed that you could keep FreeOnTerminate on false and call the clearFinishedThreads. Actually, because the threads are in a list you want to clean them up yourself because otherwise you can't check on isDone because the thread might already be finished and freed.
If I connect two telnet sessions on it, it keeps updating the number of threads and says it has two.That's because the TList elements (TTCPThread) are never removed from the TThreadManager. They are just freed (and not even set to nil what I consider a bug).
If I disconnect one, and then reconnect another one, it keeps saying I have three, not two.
>If you have already set the free on terminate property to true also, it should automatically free the >consumer for you when you set the protected field "Done_" to true your self.
I tried also this, I don't need to perform anything special after the thread is finished but I would rather do outside and explicitly, and BTW it didn't work either, this cleaning loop fails to detect that the Thread could be freed (or it has already freed itself) and thus the list never gets the chance to lower the number of elements.
At the moment, there is not the part that deletes the element in the list, but it doesn't matter because the condition is never verified, either if the Thread is freed before or not.It was already said you needed to build in this verification yourself.
If you could be so kind to test the example for yourself, it would be pretty clear that the number never decrements.Why test this example if we've already established that the threads are not removed from the list.
NOTE:Is that really necessary?
Project > options project > parsing
Disable "Use Ansistrings (-sh, {$H+})"