Pool of Objects - Synchronize - Delphi

问题:

I am implementing a pool of objects in Delphi. I need to synchronize the threads to get the objects from the pool.

Thread Code:

uClientQueryPool.CLIENT_POOL_GUARD.Acquire();
QueryClient := QUERY_POOL.GetClient();
uClientQueryPool.CLIENT_POOL_GUARD.Release;

Pool Code:

var
   CLIENT_POOL_GUARD: TCriticalSection;

type
   TClientQueryPool = class
public
   function GetClient(): TQueryClient;
end;

The CLIENT_POOL_GUARD is a unit variable. The pool is working well, but can I use "uClientQueryPool.CLIENT_POOL_GUARD.Acquire();" and "uClientQueryPool.CLIENT_POOL_GUARD.Release;" inside the GetClient method?

Like this:

function TClientQueryPool.GetClient: TQueryClient;
begin
    CLIENT_POOL_GUARD.Acquire();
    ...
    CLIENT_POOL_GUARD.Release;
end;

回答1:

Moving the lock inside the get/pop/whatever method is just fine, as is making the CriticalSection instance a private member of the pool class. Use the same CS in the release() call that pushes the objects back onto the pool.

Been doing this for decades, usually with TObjectQueue as the pool queue, a CS to protect it and a semaphore to count the pool contents and something for requesting threads to block on if the pool empties temporarily.

Don't know where that 'double acquire' thread came from. Either the lock is inside the pool class, or outside. I really can't imagine why anyone would code up both!

Example classes:

First, thread-safe P-C queue, for holding the pooled objects:

unit tinySemaphoreQueue;

interface

uses
  Windows, Messages, SysUtils, Classes,syncObjs,contnrs;


type

pObject=^Tobject;


TsemaphoreMailbox=class(TobjectQueue)
private
  countSema:Thandle;
protected
  access:TcriticalSection;
public
  property semaHandle:Thandle read countSema;
  constructor create; virtual;
  procedure push(aObject:Tobject); virtual;
  function pop(pResObject:pObject;timeout:DWORD):boolean;  virtual;
end;


implementation

{ TsemaphoreMailbox }

constructor TsemaphoreMailbox.create;
begin
  inherited Create;
  access:=TcriticalSection.create;
  countSema:=createSemaphore(nil,0,maxInt,nil);
end;

function TsemaphoreMailbox.pop(pResObject: pObject;
  timeout: DWORD): boolean;
begin // wait for a unit from the semaphore
  result:=(WAIT_OBJECT_0=waitForSingleObject(countSema,timeout));
  if result then // if a unit was supplied before the timeout,
  begin
    access.acquire;
    try
      pResObject^:=inherited pop; // get an object from the queue
    finally
      access.release;
    end;
  end;
end;

procedure TsemaphoreMailbox.push(aObject: Tobject);
begin
  access.acquire;
  try
    inherited push(aObject); // shove the object onto the queue
  finally
    access.release;
  end;
  releaseSemaphore(countSema,1,nil); // release one unit to semaphore
end;

end.

then object pool:

unit tinyObjectPool;

interface

uses
  Windows, Messages, SysUtils, Classes,syncObjs,contnrs,
  tinySemaphoreQueue;

type
  TobjectPool=class;

  TpooledObject=class(TObject)
  private
    FmyPool:TObjectPool;
  protected
    Fparameter:TObject;
  public
    procedure release;
    constructor create(parameter:TObject); virtual;
  end;

  TpooledObjectClass=class of TpooledObject;

  TobjectPool=class(TsemaphoreMailbox)
  private
    Fparameter:TObject;
    function getPoolLevel: integer;
  public
    property poolLevel:integer read getPoolLevel;
    constructor create(poolDepth:integer;
      pooledObjectClass:TpooledObjectClass;parameter:TObject); reintroduce; virtual;
  end;

implementation

{ TobjectPool }

constructor TobjectPool.create(poolDepth: integer;
  pooledObjectClass: TpooledObjectClass;parameter:TObject);
var objectCount:integer;
    thisObject:TpooledObject;
begin
  inherited create;
  Fparameter:=parameter; // a user parameter passed to all objects
  for objectCount:=0 to poolDepth-1 do // fill up the pool with objects
  begin
    thisObject:=pooledObjectClass.create(parameter);
    thisObject.FmyPool:=self;
    inherited push(thisObject);
  end;
end;

function TobjectPool.getPoolLevel: integer;
begin
  access.acquire;
  result:=inherited count;
  access.release;
end;



{ TpooledObject }

constructor TpooledObject.create(parameter: TObject);
begin
  inherited create;
  Fparameter:=parameter;
end;

procedure TpooledObject.release;
begin
  FmyPool.push(self);
end;

end.

回答2:

Yes you can. Note, though that although you can pull an object from the pool in a thread-safe manner, it may not be thread-safe to use it if the object itself isn't thread-safe. For instance, in the example below, the pool is thread safe and even makes threads wait if all objects in the pool are in use, but once an object is in use, using it still is not thread safe, because it uses global data.

uses
  SyncObjs;

var
  GlobalData: Integer = 0;

type
  TDataObject = class
    Used: Boolean;
    procedure UpdateData;
  end;

type
  TPool = class
    FLock: TCriticalSection;
    FSemaphore: TSemaphore;
    FDataObjects: array[0..9] of TDataObject;
    constructor Create;
    destructor Destroy; override;
    function GetDataObject: TDataObject;
    procedure ReleaseDataObject(AObject: TDataObject);
  end;

var
  Pool: TPool;

type
  TDataThread = class(TThread)
    constructor Create;
    procedure Execute; override;
  end;

{ TPool }

constructor TPool.Create;
var
  i: Integer;
begin
  inherited Create;
  FLock := TCriticalSection.Create;
  FSemaphore := TSemaphore.Create(nil, Length(FDataObjects), Length(FDataObjects), '', False);

  for i := Low(FDataObjects) to High(FDataObjects) do
    FDataObjects[i] := TDataObject.Create;
end;

destructor TPool.Destroy;
var
  i: Integer;
begin
  for i := Low(FDataObjects) to High(FDataObjects) do
    FDataObjects[i].Free;

  FSemaphore.Free;
  FLock.Free;
end;

function TPool.GetDataObject: TDataObject;
var
  i: Integer;
begin
  Result := nil;

  FLock.Acquire;
  try
    FSemaphore.Acquire;
    for i := Low(FDataObjects) to High(FDataObjects) do
      if not FDataObjects[i].Used then
      begin
        Result := FDataObjects[i];
        Result.Used := True;
        Exit;
      end;

    Assert(Result <> nil, 'Pool did not return an object');
  finally
    FLock.Release;
  end;
end;

procedure TPool.ReleaseDataObject(AObject: TDataObject);
begin
  if not AObject.Used then
    raise Exception.Create('Data object cannot be released, because it is not in use.');

  AObject.Used := False;
  FSemaphore.Release;
end;

{ TDataObject }

procedure TDataObject.UpdateData;
begin
  Inc(GlobalData);
end;

procedure TForm1.Button1Click(Sender: TObject);
begin
  TDataThread.Create;
end;

{ TDataThread }

constructor TDataThread.Create;
begin
  inherited Create(True);
  FreeOnTerminate := True;
  Resume;
end;

procedure TDataThread.Execute;
var
  DataObject: TDataObject;
begin
  DataObject := Pool.GetDataObject;

  DataObject.UpdateData; // <-- Not thread-safe!

  Pool.ReleaseDataObject(DataObject);
end;

initialization
  Pool := TPool.Create;
finalization
  Pool.Free;
end.

回答3:

1) I'd remove Acquire/Release code from threads code - it is fragile. In one thread you forget to call it - and ba-bang! Security measures, as a rule of thumb, should be centralized and enforced by server, not distributed in fuzzy way in clients.

2) Acquire/Release calls should be guarded from errors, else any stray exception would forever lock all the threads.

 function TClientQueryPool.GetClient: TQueryClient;
 begin
   CS.Acquire;
   try
     // actually getting object, preferably just calling
     // internal non-public thread-unsafe method for it
   finally
     CS.Release;
   end;
  end;

3) Critical section itself should better be a Pool's internal, non-public member. That way you would be allowed in future, when you forget of implementation details, easy refactoring, like:

3.1) implementing several pools

3.2) moving pool code to another unit

3.3) ensuring any stray erroneous code outside pool would not be able to crash the application be randomly acquiring or releasing the CS

4) Double calling of acquire/release over TCriticalSection object puts all your bets over implications from a single note in TCriticalSection documentation, pointed to by The_Fox. "Each call to Release should be balance by an earlier call to Acquire" http://docwiki.embarcadero.com/Libraries/en/System.SyncObjs.TCriticalSection.Release

And over the hope that all other Pascal implementations today and tomorrow would not miss it.

That is fragile practice. And multi-threading code is famous for creating Heisenbugs, when there are problems at clients sites, but you can not reproduce and find it in house. If in future your company would expand to different platform or different language implementation, that puts a potential land mine. And the kind of mine, that would be hard to find by testing in house. Multithreading code is the place where you'd better be over-defeinsive and just do not allow ANY uncertainty to happen.