trunk/runtime/sv5ipc.inc

{Used by ISQLMonitor and implements System V IPC}

const
  IPCFileName: string = 'FB.SQL.MONITOR1_0';
  cNumberOfSemaphores = 10;
  cMutexSemaphore = 0;
  cMonitorCounter = 1;
  cReadReadyEventSemaphore = 2;
  cReadFinishedEventSemaphore = 4;
  cDataAvailableEventSemaphore = 6;
  cWriterBusyEventSemaphore = 8;
  cDefaultTimeout = 1; { 1 seconds }

{$IF FPC_FULLVERSION = 30000 }
{Fix regression in FPC 3.0.0 ipc.pp unit. Expected to be fixed in fpc 3.0.2}
{$IF defined(darwin) }
SEM_GETNCNT = 3;   { Return the value of sempid (READ)  }
SEM_GETPID  = 4;   { Return the value of semval (READ)  }
SEM_GETVAL  = 5;   { Return semvals into arg.array (READ)  }
SEM_GETALL  = 6;   { Return the value of semzcnt (READ)  }
SEM_GETZCNT = 7;   { Set the value of semval to arg.val (ALTER)  }
SEM_SETVAL  = 8;   { Set semvals from arg.array (ALTER)  }
SEM_SETALL  = 9;
{$ENDIF}
{$ENDIF}

{
  The call to semctl in ipc is broken in FPC release 2.4.2 and earlier. Hence
  need to replace with a working libc call. semtimedop is not present in 2.4.2 and earlier.
}

{$IF FPC_FULLVERSION <= 20402 }
Function real_semctl(semid:cint; semnum:cint; cmd:cint): cint; cdecl; varargs; external clib name 'semctl';

Function semctl(semid:cint; semnum:cint; cmd:cint; var arg: tsemun): cint;
  begin
    semctl := real_semctl(semid,semnum,cmd,pointer(arg));
  end;
{$IFDEF HAS_SEMTIMEDOP}
Function semtimedop(semid:cint; sops: psembuf; nsops: cuint; timeOut: ptimespec): cint; cdecl; external clib name 'semtimedop';
{$ENDIF}
Function semget(key:Tkey; nsems:cint; semflg:cint): cint; cdecl; external clib name 'semget';
Function semop(semid:cint; sops: psembuf; nsops: cuint): cint; cdecl; external clib name 'semop';

function GetLastErrno: cint;
begin
  Result := fpgetCerrno
end;
{$ELSE}
function GetLastErrno: cint;
begin
  Result := fpgetErrno
end;

{$ENDIF}

type
  TGlobalInterface = class;
  {Interprocess Communication Objects. All platform dependent IPC is abstracted
   into this set of objects }

  { TIpcCommon }

  TIpcCommon = class
   private
    function GetSa: PSecurityAttributes;
 protected
    FInitialiser: boolean;  static;
    FSemaphoreSetID: cint;  static;
    FSharedMemoryID: cint;  static;
    function sem_op(SemNum, op: integer; flags: cshort = 0): cint;
    function sem_timedop(SemNum, op: integer; timeout_secs: integer; flags: cshort = 0): cint;
    function GetSemValue(SemNum: integer): cint;
    procedure SemInit(SemNum, AValue: cint);
  public
    property Sa : PSecurityAttributes read GetSa;
  end;

  { TSharedMemory }

  {
    The shared memory segment is used for interprocess communication and
    holds both a message buffer and a number of shared variables. Shared
    memory is allocated to each shared variable using the Allocate function.
    An underlying assumption is that each process using the shared memory
    calls "Allocate" in the same order and for the same memory sizes.

    Linux:

    The Linux implementation uses Linux shared memory. IPC_PRIVATE is used
    to allocate the memory and the resulting memory id is written to a
    well known file. By default this is in the current user's home directory,
    but this can be over-ridden to specify a globally unique filename.

    Access to the shared memory is restricted to the current user/group.
    Note that the Linux semaphore set is also created with the shared memory.
  }

  TSharedMemory = class(TIpcCommon)
  private
    FBuffer: PChar;
    FLastAllocationSize: integer;
    FUnused: integer;
    FBufptr: PChar;
    procedure DropSharedMemory;
    procedure GetSharedMemory(MemSize: integer);
  public
    constructor Create(MemSize: integer);
    destructor Destroy; override;
    function Allocate(Size: integer): PChar;
    property LastAllocationSize: integer read FLastAllocationSize;
  end;

  {TMutex}

  TMutex = class(TIpcCommon)
  private
    FMutexSemaphore: cint;
    FLockCount: integer;
  public
    constructor Create(SemNumber: cint);
    procedure Lock;
    procedure Unlock;
  end;

  { TSingleLockGate }

  {
    A single lock gate is either open or closed. When open, any thread can pass
    through it while, when closed, all threads are blocked as they try to pass
    through the gate. When the gate is opened, all blocked threads are resumed.

    There is an implementation assumption that only one writer thread at
    a time (i.e. the thread which locks or unlocks the gate) can have access to
    it at any one time. I.e. an external Mutex prevents race conditions.

    Linux:

    In the Linux implementation, the gate is implemented by a semaphore
    and a share memory integer used as a bi-state variable. When the gate
    is open, the bi-state variable is non-zero. It is set to zero when closed.
    Another shared memory integer is used to count the number of waiting
    threads, and a second semaphore is used to protect access to this.

    The event semaphore is initialised to zero. When a thread passes through the gate
    it checks the state. If open, the thread continues. If closed then it
    increments the count of waiting threads and then decrements the semaphore
    and hence enters an indefinite wait state.

    When the gate is locked, the state is set to zero. When unlocked, the state
    is set to one and the semaphore incremented by the number of waiting threads,
    which itself is then zeroed.

    Always initialised to the Unlocked state
  }

  TSingleLockGate = class(TIpcCommon)
  private
    FOwner: TGlobalInterface;
    FSemaphore: cint;
    FMutex: cint;
    FSignalledState: PInteger;
    FWaitingThreads: PInteger;
    function GetWaitingThreads: integer;
  public
    constructor Create(SemNum: cint; AOwner: TGlobalInterface);
    property WaitingThreads: integer read GetWaitingThreads;
  public
    procedure PassthroughGate;
    procedure Unlock;
    procedure Lock;
  end;

  { TMultilockGate }

  { This type of Gate is used where several reader threads must pass
    through the gate before it can be opened for a writer thread.

    The reader threads register their interest by each locking the gate.
    The writer thread then waits on the locked gate until all the reader
    threads have separately unlocked the gate.

    There is an underlying assumption of a single writer. A Mutex must
    be used to control access to the gate from the writer side if this
    assumption is invalid.

    Linux:

    The Linux implementation uses a single semaphore to implement the gate,
    which is initialised to 1 (unlocked), and a count of the number of
    threads that have locked the gate (LockCount). A mutex semaphore
    protects access to the LockCount. When the gate is locked, the lockcount
    is incremented and, if the LockCount was originally zero, the semaphore is
    set to zero (Gate Closed).

    Unlocking the gate, is the reverse. The LockCount is decremented and, if it
    reaches zero, the semaphore is set to one (Gate Opened).

    When a writer passes through the gate, it checks the LockCount, if zero it
    proceeds to pass through the gate. Otherwise it decrements and waits on the
    semaphore. When the writer resumes, it increments the semaphore in order
    to return it to its unlocked state. The wait is a timed wait, as there is
    a risk that a reader thread may terminate while the gate is locked. If the
    LockCount is non-zero, it is decremented and the writer returns to wait on
    the gate.

    Always initialised to the Unlocked state
  }

  TMultilockGate = class(TIpcCommon)
  private
    FOnGateTimeout: TNotifyEvent;
    FOwner: TGlobalInterface;
    FSemaphore: cint;
    FMutex: cint;
    FLockCount: PInteger;
    function GetLockCount: integer;
  public
    constructor Create(SemNum: cint; AOwner: TGlobalInterface);
    procedure Lock;
    procedure Unlock;
    procedure PassthroughGate;
    property LockCount: integer read GetLockCount;
    property OnGateTimeout: TNotifyEvent read FOnGateTimeout write FOnGateTimeout;
  end;

  { TGlobalInterface }

  TGlobalInterface = class(TIpcCommon)
  private
    FMaxBufferSize: integer;
    FSharedMemory: TSharedMemory;
    FWriteLock: TMutex;
    FBuffer: PChar;
    FTraceDataType,
    FBufferSize: PInteger;
    FTimeStamp: PDateTime;
    FReadReadyEvent: TMultiLockGate;
    FReadFinishedEvent: TMultiLockGate;
    FDataAvailableEvent: TSingleLockGate;
    FWriterBusyEvent: TSingleLockGate;
    function GetMonitorCount: integer;
  public
    constructor Create;
    destructor Destroy; override;
    procedure IncMonitorCount;
    procedure DecMonitorCount;
    procedure SendTrace(TraceObject: TTraceObject);
    procedure ReceiveTrace(TraceObject: TTraceObject);
    property DataAvailableEvent: TSingleLockGate read FDataAvailableEvent;
    property WriterBusyEvent: TSingleLockGate read FWriterBusyEvent;
    property ReadReadyEvent: TMultiLockGate read FReadReadyEvent;
    property ReadFinishedEvent: TMultiLockGate read FReadFinishedEvent;
    property WriteLock: TMutex read FWriteLock;
    property MonitorCount: integer read GetMonitorCount;
    property SharedMemory: TSharedMemory read FSharedMemory;
    property MaxBufferSize: integer read FMaxBufferSize;
  end;

{ TSharedMemory }

procedure TSharedMemory.GetSharedMemory(MemSize: integer);
var F: cint;
begin
    {Get the Shared Memory and Semaphore IDs from the Global File if it exists
     or create them and the file otherwise }

    repeat
      F := fpOpen(IPCFileName, O_WrOnly or O_Creat or O_Excl);
      if F < 0 then
      begin
        if fpgetErrno = ESysEEXIST {EEXIST} then
        begin
          { looks like it already exists}
          Sleep(100);
          F := fpOpen(IPCFileName,O_RdOnly);
          if (F < 0) and (fpgetErrno = ESysENOENT {ENOENT}) then
            {probably just got deleted }
          else
          if F < 0 then
            IBError(ibxeCannotCreateSharedResource,['Error accessing IPC File - ' +
                                                 StrError(fpgetErrno)]);
        end
        else
            IBError(ibxeCannotCreateSharedResource,['Error creating IPC File  - ' +
                                                 StrError(fpgetErrno)]);
      end
      else
        FInitialiser := true
    until F >= 0;

    if FInitialiser then
    begin
      FSharedMemoryID := shmget(IPC_PRIVATE,MemSize, IPC_CREAT or
                           S_IRUSR or S_IWUSR or S_IRGRP or S_IWGRP);
      if FSharedMemoryID < 0 then
          IBError(ibxeCannotCreateSharedResource,['Cannot create shared memory segment - ' +
                                                 StrError(fpgetErrno)]);

      FSemaphoreSetID := semget(IPC_PRIVATE, cNumberOfSemaphores,IPC_CREAT or
                           S_IRUSR or S_IWUSR or S_IRGRP or S_IWGRP);
      if FSemaphoreSetID < 0 then
          IBError(ibxeCannotCreateSharedResource,['Cannot create shared semaphore set - ' +
                                                 StrError(fpgetErrno)]);

      fpWrite(F,FSharedMemoryID,sizeof(FSharedMemoryID));
      fpWrite(F,FSemaphoreSetID,sizeof(FSemaphoreSetID));
    end
    else
    begin
      fpRead(F,FSharedMemoryID,sizeof(FSharedMemoryID));
      fpRead(F,FSemaphoreSetID,sizeof(FSemaphoreSetID));
      if GetSemValue(cMonitorCounter) = 0 then
      begin
        FInitialiser := true;
        //writeln('Opened file and is initialiser');
      end
    end;
    fpClose(F);
end;

procedure TSharedMemory.DropSharedMemory;
var ds: TShmid_ds;
    arg: tsemun;
begin
  if shmctl(FSharedMemoryID,IPC_STAT,@ds) < 0 then
    IBError(ibxeSV5APIError,['Error getting shared memory info' + strError(fpgetErrno)]);
  if ds.shm_nattch = 0 then  { we are the last one out - so, turn off the lights }
  begin
    shmctl(FSharedMemoryID,IPC_RMID,nil);
    semctl(FSemaphoreSetID,0,IPC_RMID,arg);
    DeleteFile(IPCFileName);
  end;
end;

constructor TSharedMemory.Create(MemSize: integer);
begin
  inherited Create;
  FInitialiser := false;
  GetSharedMemory(MemSize);
  FBuffer := shmat(FSharedMemoryID,nil,0);
  if PtrInt(FBuffer) = -1 then
    IBError(ibxeCannotCreateSharedResource,[StrError(Errno)]);
  FBufPtr := FBuffer;
  FUnused := MemSize
end;

destructor TSharedMemory.Destroy;
begin
  shmdt(FBuffer);
  DropSharedMemory;
  inherited Destroy;
end;

function TSharedMemory.Allocate(Size: integer): PChar;
begin
  if Size > FUnused then
      IBError(ibxeCannotCreateSharedResource, ['Not enough shared memory']);
  Result := FBufPtr;

  if Size = 0 then
  begin
    FLastAllocationSize := FUnused;
    FUnused := 0
  end
  else
  begin
    FLastAllocationSize := Size;
    Dec(FUnused,Size);
  end;
  Inc(FBufPtr,Size)
end;

{ TIpcCommon }

function TIpcCommon.GetSa: PSecurityAttributes;
begin
  Result := nil
end;

function TIpcCommon.sem_op(SemNum, op: integer; flags: cshort): cint;
var sembuf: TSEMbuf;
begin
    sembuf.sem_num := SemNum;
    sembuf.sem_op:= op;
    sembuf.sem_flg := flags or SEM_UNDO;
    Result := semop(FSemaphoreSetID,@sembuf,1);
end;

function TIpcCommon.sem_timedop(SemNum, op: integer; timeout_secs: integer;
  flags: cshort): cint;
var sembuf: TSEMbuf;
    timeout: TimeSpec;
begin
    sembuf.sem_num := SemNum;
    sembuf.sem_op:= op;
    sembuf.sem_flg := flags or SEM_UNDO;
    timeout.tv_sec := timeout_secs;
    timeout.tv_nsec := 0;
{$IFDEF HAS_SEMTIMEDOP}
    Result := semtimedop(FSemaphoreSetID,@sembuf,1,@timeout);
{$ELSE}
    Result := semop(FSemaphoreSetID,@sembuf,1);    {May hang on race condition}
{$ENDIF}
end;

function TIpcCommon.GetSemValue(SemNum: integer): cint;
var args :TSEMun;
begin
  Result := semctl(FSemaphoreSetID,SemNum,SEM_GETVAL,args);
  if Result < 0 then
     IBError(ibxeSV5APIError,['GetSemValue: '+strError(GetLastErrno)]);
end;

procedure TIpcCommon.SemInit(SemNum, AValue: cint);
var args :TSEMun;
begin
  //writeln('Initialising ',SemNum,' to ',AValue);
  args.val := AValue;
  if semctl(FSemaphoreSetID,SemNum,SEM_SETVAL,args)  < 0 then
     IBError(ibxeCannotCreateSharedResource,['Unable to initialise Semaphone ' +
                          IntToStr(SemNum) + '- ' + StrError(GetLastErrno)]);

end;

  { TMutex }

constructor TMutex.Create(SemNumber: cint);
begin
  inherited Create;
  FMutexSemaphore := SemNumber;
  if FInitialiser then
    SemInit(FMutexSemaphore,1)
end;

{ Obtain ownership of the Mutex and prevent other threads from accessing protected resource }

procedure TMutex.Lock;
begin
  //writeln('Lock: Entering Mutex ',FMutexSemaphore,' LockCount=',FLockCount,' State = ',GetSemValue(FMutexSemaphore));
  if FLockCount = 0 then
    sem_op(FMutexSemaphore,-1);
  Inc(FLockCount);
  //writeln('Lock: Mutex Exit');
end;

{Give up ownership of the Mutex and allow other threads access }

procedure TMutex.Unlock;
begin
  //writeln('UnLock: Entering Mutex, LockCount=',FLockCount);
  if FLockCount = 0 then Exit;
  Dec(FLockCount);
  if FLockCount = 0 then
    sem_op(FMutexSemaphore,1);
  //writeln('UnLock: Mutex Exit',' State = ',GetSemValue(FMutexSemaphore));
end;

{ TSingleLockGate }

function TSingleLockGate.GetWaitingThreads: integer;
begin
  Result := FWaitingThreads^
end;

constructor TSingleLockGate.Create(SemNum: cint; AOwner: TGlobalInterface);
begin
  inherited Create;
  FOwner := AOwner;
  FSignalledState := PInteger(FOwner.SharedMemory.Allocate(sizeof(FSignalledState)));
  FWaitingThreads := PInteger(FOwner.SharedMemory.Allocate(sizeof(FWaitingThreads)));
  FSemaphore := SemNum;
  FMutex := SemNum + 1;
  if FInitialiser then
  begin
    FSignalledState^ := 1;
    FWaitingThreads^ := 0;
    SemInit(FSemaphore,0);
    SemInit(FMutex,1);
  end;
end;

procedure TSingleLockGate.PassthroughGate;
begin
  if FSignalledState^ = 0 then
  begin
    sem_op(FMutex,-1,0); //Acquire Mutex
    Inc(FWaitingThreads^);
    sem_op(FMutex,1,0); //Release Mutex
    //writeln(ClassName + ': Wait State Entered ',FSemaphore,' = ',GetSemValue(FSemaphore));
    sem_op(FSemaphore,-1,0); //Enter Wait
    //writeln(ClassName + ': Wait State Ends ',FSemaphore);
  end;
end;

procedure TSingleLockGate.Unlock;
begin
  if FSignalledState^ = 0 then
  begin
    FSignalledState^ := 1;
    sem_op(FMutex,-1,0); //Acquire Mutex
    //writeln(ClassName + ': Unlocking' ,FSemaphore);
    sem_op(FSemaphore,FWaitingThreads^,0);
    FWaitingThreads^ := 0;
    sem_op(FMutex,1,0); //Release Mutex
  end;
end;

procedure TSingleLockGate.Lock;
begin
  if FSignalledState^ = 1 then
  begin
    //writeln(ClassName + ': Locking Gate ',FSemaphore);
    SemInit(FSemaphore,0);
    FSignalledState^ := 0;
  end;
end;

{ TMultilockGate }

constructor TMultilockGate.Create(SemNum: cint; AOwner: TGlobalInterface);
begin
  inherited Create;
  FOwner := AOwner;
  FSemaphore := SemNum;
  FMutex := SemNum + 1;
  FLockCount := PInteger(FOwner.SharedMemory.Allocate(sizeof(FLockCount)));
  if FInitialiser then
  begin
    FLockCount^ := 0;
    SemInit(FSemaphore,1);
    SemInit(FMutex,1);
  end;
end;

function TMultilockGate.GetLockCount: integer;
begin
  Result := FLockCount^
end;

procedure TMultilockGate.Lock;
begin
    sem_op(FMutex,-1,0); //Acquire Mutex
    if FLockCount^ = 0 then
    begin
      //writeln(ClassName,': Locking ',FSemaphore);
      SemInit(FSemaphore,0);
    end;
    Inc(FLockCount^);
    sem_op(FMutex,1,0); //Release Mutex
end;

procedure TMultilockGate.Unlock;
begin
    sem_op(FMutex,-1,0); //Acquire Mutex
    Dec(FLockCount^);
    if FLockCount^ <= 0 then
    begin
      //writeln(ClassName,': UnLocking ',FSemaphore);
      SemInit(FSemaphore,1);
      FLockCount^ := 0
    end;
    sem_op(FMutex,1,0); //Release Mutex
end;

procedure TMultilockGate.PassthroughGate;
begin
  if FLockCount^ = 0 then
    Exit;
  //writeln(ClassName,': Waiting on ',FSemaphore);
  while sem_timedop(FSemaphore,-1,cDefaultTimeout) < 0 do
  {looks like we lost a reader}
  begin
    if FLockCount^ > 0 then
    begin
      UnLock;
      if assigned(FOnGateTimeout) then
        OnGateTimeout(self)
    end
  end;
  sem_op(FSemaphore,1);
  //writeln(ClassName,': Wait done on ',FSemaphore);
end;


{ TGlobalInterface }

function TGlobalInterface.GetMonitorCount: integer;
begin
  Result := GetSemValue(cMonitorCounter)
end;

constructor TGlobalInterface.Create;
begin
  inherited Create;
  FSharedMemory := TSharedMemory.Create(cMonitorHookSize);

  FWriteLock := TMutex.Create(cMutexSemaphore);

  FDataAvailableEvent := TSingleLockGate.Create(cDataAvailableEventSemaphore,self);
  FWriterBusyEvent := TSingleLockGate.Create(cWriterBusyEventSemaphore,self);
  FReadReadyEvent := TMultiLockGate.Create(cReadReadyEventSemaphore,self);
  FReadFinishedEvent := TMultiLockGate.Create(cReadFinishedEventSemaphore,self);

  if FInitialiser then
    SemInit(cMonitorCounter,0);
  FTraceDataType := PInteger(FSharedMemory.Allocate(sizeof(Integer)));
  FTimeStamp := PDateTime(FSharedMemory.Allocate(sizeof(TDateTime)));
  FBufferSize := PInteger(FSharedMemory.Allocate(sizeof(Integer)));
  FBuffer := FSharedMemory.Allocate(0); //All remaining
  FMaxBufferSize := FSharedMemory.LastAllocationSize;

  if FInitialiser then
  begin
    FBufferSize^ := 0;
    FDataAvailableEvent.Lock
  end;
end;

destructor TGlobalInterface.Destroy;
begin
  if assigned(FWriteLock) then FWriteLock.Free;
  if assigned(FDataAvailableEvent) then FDataAvailableEvent.Free;
  if assigned(FWriterBusyEvent) then FWriterBusyEvent.Free;
  if assigned(FReadReadyEvent) then FReadReadyEvent.Free;
  if assigned(FReadFinishedEvent) then FReadFinishedEvent.Free;
  if assigned(FSharedMemory) then FSharedMemory.Free;
  inherited Destroy;
end;

procedure TGlobalInterface.IncMonitorCount;
begin
  sem_op(cMonitorCounter,1);
end;

procedure TGlobalInterface.DecMonitorCount;
begin
  sem_op(cMonitorCounter,-1,IPC_NOWAIT);
end;

procedure TGlobalInterface.SendTrace(TraceObject: TTraceObject);
begin
  FTraceDataType^ := Integer(TraceObject.FDataType);
  FTimeStamp^ := TraceObject.FTimeStamp;
  FBufferSize^ := Min(Length(TraceObject.FMsg), MaxBufferSize);
  Move(TraceObject.FMsg[1], FBuffer^, FBufferSize^);
end;

procedure TGlobalInterface.ReceiveTrace(TraceObject: TTraceObject);
begin
  SetString(TraceObject.FMsg, FBuffer, FBufferSize^);
  TraceObject.FDataType := TTraceFlag(FTraceDataType^);
  TraceObject.FTimeStamp := TDateTime(FTimeStamp^);
end;


Revision:	39
Committed:	Tue May 17 08:14:52 2016 UTC (8 years, 6 months ago) by tony
File size:	20213 byte(s)
Log Message:	Committing updates for Release R1-4-1
#	User	Rev	Content
1	tony	33	{Used by ISQLMonitor and implements System V IPC}
2
3			const
4			IPCFileName: string = 'FB.SQL.MONITOR1_0';
5			cNumberOfSemaphores = 10;
6			cMutexSemaphore = 0;
7			cMonitorCounter = 1;
8			cReadReadyEventSemaphore = 2;
9			cReadFinishedEventSemaphore = 4;
10			cDataAvailableEventSemaphore = 6;
11			cWriterBusyEventSemaphore = 8;
12			cDefaultTimeout = 1; { 1 seconds }
13
14	tony	39	{$IF FPC_FULLVERSION = 30000 }
15			{Fix regression in FPC 3.0.0 ipc.pp unit. Expected to be fixed in fpc 3.0.2}
16			{$IF defined(darwin) }
17			SEM_GETNCNT = 3; { Return the value of sempid (READ) }
18			SEM_GETPID = 4; { Return the value of semval (READ) }
19			SEM_GETVAL = 5; { Return semvals into arg.array (READ) }
20			SEM_GETALL = 6; { Return the value of semzcnt (READ) }
21			SEM_GETZCNT = 7; { Set the value of semval to arg.val (ALTER) }
22			SEM_SETVAL = 8; { Set semvals from arg.array (ALTER) }
23			SEM_SETALL = 9;
24			{$ENDIF}
25			{$ENDIF}
26
27	tony	33	{
28			The call to semctl in ipc is broken in FPC release 2.4.2 and earlier. Hence
29			need to replace with a working libc call. semtimedop is not present in 2.4.2 and earlier.
30			}
31
32			{$IF FPC_FULLVERSION <= 20402 }
33			Function real_semctl(semid:cint; semnum:cint; cmd:cint): cint; cdecl; varargs; external clib name 'semctl';
34
35			Function semctl(semid:cint; semnum:cint; cmd:cint; var arg: tsemun): cint;
36			begin
37			semctl := real_semctl(semid,semnum,cmd,pointer(arg));
38			end;
39			{$IFDEF HAS_SEMTIMEDOP}
40			Function semtimedop(semid:cint; sops: psembuf; nsops: cuint; timeOut: ptimespec): cint; cdecl; external clib name 'semtimedop';
41			{$ENDIF}
42			Function semget(key:Tkey; nsems:cint; semflg:cint): cint; cdecl; external clib name 'semget';
43			Function semop(semid:cint; sops: psembuf; nsops: cuint): cint; cdecl; external clib name 'semop';
44
45			function GetLastErrno: cint;
46			begin
47			Result := fpgetCerrno
48			end;
49			{$ELSE}
50			function GetLastErrno: cint;
51			begin
52			Result := fpgetErrno
53			end;
54
55			{$ENDIF}
56
57			type
58			TGlobalInterface = class;
59			{Interprocess Communication Objects. All platform dependent IPC is abstracted
60			into this set of objects }
61
62			{ TIpcCommon }
63
64			TIpcCommon = class
65			private
66			function GetSa: PSecurityAttributes;
67			protected
68			FInitialiser: boolean; static;
69			FSemaphoreSetID: cint; static;
70			FSharedMemoryID: cint; static;
71			function sem_op(SemNum, op: integer; flags: cshort = 0): cint;
72			function sem_timedop(SemNum, op: integer; timeout_secs: integer; flags: cshort = 0): cint;
73			function GetSemValue(SemNum: integer): cint;
74			procedure SemInit(SemNum, AValue: cint);
75			public
76			property Sa : PSecurityAttributes read GetSa;
77			end;
78
79			{ TSharedMemory }
80
81			{
82			The shared memory segment is used for interprocess communication and
83			holds both a message buffer and a number of shared variables. Shared
84			memory is allocated to each shared variable using the Allocate function.
85			An underlying assumption is that each process using the shared memory
86			calls "Allocate" in the same order and for the same memory sizes.
87
88			Linux:
89
90			The Linux implementation uses Linux shared memory. IPC_PRIVATE is used
91			to allocate the memory and the resulting memory id is written to a
92			well known file. By default this is in the current user's home directory,
93			but this can be over-ridden to specify a globally unique filename.
94
95			Access to the shared memory is restricted to the current user/group.
96			Note that the Linux semaphore set is also created with the shared memory.
97			}
98
99			TSharedMemory = class(TIpcCommon)
100			private
101			FBuffer: PChar;
102			FLastAllocationSize: integer;
103			FUnused: integer;
104			FBufptr: PChar;
105			procedure DropSharedMemory;
106			procedure GetSharedMemory(MemSize: integer);
107			public
108			constructor Create(MemSize: integer);
109			destructor Destroy; override;
110			function Allocate(Size: integer): PChar;
111			property LastAllocationSize: integer read FLastAllocationSize;
112			end;
113
114			{TMutex}
115
116			TMutex = class(TIpcCommon)
117			private
118			FMutexSemaphore: cint;
119			FLockCount: integer;
120			public
121			constructor Create(SemNumber: cint);
122			procedure Lock;
123			procedure Unlock;
124			end;
125
126			{ TSingleLockGate }
127
128			{
129			A single lock gate is either open or closed. When open, any thread can pass
130			through it while, when closed, all threads are blocked as they try to pass
131			through the gate. When the gate is opened, all blocked threads are resumed.
132
133			There is an implementation assumption that only one writer thread at
134			a time (i.e. the thread which locks or unlocks the gate) can have access to
135			it at any one time. I.e. an external Mutex prevents race conditions.
136
137			Linux:
138
139			In the Linux implementation, the gate is implemented by a semaphore
140			and a share memory integer used as a bi-state variable. When the gate
141			is open, the bi-state variable is non-zero. It is set to zero when closed.
142			Another shared memory integer is used to count the number of waiting
143			threads, and a second semaphore is used to protect access to this.
144
145			The event semaphore is initialised to zero. When a thread passes through the gate
146			it checks the state. If open, the thread continues. If closed then it
147			increments the count of waiting threads and then decrements the semaphore
148			and hence enters an indefinite wait state.
149
150			When the gate is locked, the state is set to zero. When unlocked, the state
151			is set to one and the semaphore incremented by the number of waiting threads,
152			which itself is then zeroed.
153
154			Always initialised to the Unlocked state
155			}
156
157			TSingleLockGate = class(TIpcCommon)
158			private
159			FOwner: TGlobalInterface;
160			FSemaphore: cint;
161			FMutex: cint;
162			FSignalledState: PInteger;
163			FWaitingThreads: PInteger;
164			function GetWaitingThreads: integer;
165			public
166			constructor Create(SemNum: cint; AOwner: TGlobalInterface);
167			property WaitingThreads: integer read GetWaitingThreads;
168			public
169			procedure PassthroughGate;
170			procedure Unlock;
171			procedure Lock;
172			end;
173
174			{ TMultilockGate }
175
176			{ This type of Gate is used where several reader threads must pass
177			through the gate before it can be opened for a writer thread.
178
179			The reader threads register their interest by each locking the gate.
180			The writer thread then waits on the locked gate until all the reader
181			threads have separately unlocked the gate.
182
183			There is an underlying assumption of a single writer. A Mutex must
184			be used to control access to the gate from the writer side if this
185			assumption is invalid.
186
187			Linux:
188
189			The Linux implementation uses a single semaphore to implement the gate,
190			which is initialised to 1 (unlocked), and a count of the number of
191			threads that have locked the gate (LockCount). A mutex semaphore
192			protects access to the LockCount. When the gate is locked, the lockcount
193			is incremented and, if the LockCount was originally zero, the semaphore is
194			set to zero (Gate Closed).
195
196			Unlocking the gate, is the reverse. The LockCount is decremented and, if it
197			reaches zero, the semaphore is set to one (Gate Opened).
198
199			When a writer passes through the gate, it checks the LockCount, if zero it
200			proceeds to pass through the gate. Otherwise it decrements and waits on the
201			semaphore. When the writer resumes, it increments the semaphore in order
202			to return it to its unlocked state. The wait is a timed wait, as there is
203			a risk that a reader thread may terminate while the gate is locked. If the
204			LockCount is non-zero, it is decremented and the writer returns to wait on
205			the gate.
206
207			Always initialised to the Unlocked state
208			}
209
210			TMultilockGate = class(TIpcCommon)
211			private
212			FOnGateTimeout: TNotifyEvent;
213			FOwner: TGlobalInterface;
214			FSemaphore: cint;
215			FMutex: cint;
216			FLockCount: PInteger;
217			function GetLockCount: integer;
218			public
219			constructor Create(SemNum: cint; AOwner: TGlobalInterface);
220			procedure Lock;
221			procedure Unlock;
222			procedure PassthroughGate;
223			property LockCount: integer read GetLockCount;
224			property OnGateTimeout: TNotifyEvent read FOnGateTimeout write FOnGateTimeout;
225			end;
226
227			{ TGlobalInterface }
228
229			TGlobalInterface = class(TIpcCommon)
230			private
231			FMaxBufferSize: integer;
232			FSharedMemory: TSharedMemory;
233			FWriteLock: TMutex;
234			FBuffer: PChar;
235			FTraceDataType,
236			FBufferSize: PInteger;
237			FTimeStamp: PDateTime;
238			FReadReadyEvent: TMultiLockGate;
239			FReadFinishedEvent: TMultiLockGate;
240			FDataAvailableEvent: TSingleLockGate;
241			FWriterBusyEvent: TSingleLockGate;
242			function GetMonitorCount: integer;
243			public
244			constructor Create;
245			destructor Destroy; override;
246			procedure IncMonitorCount;
247			procedure DecMonitorCount;
248			procedure SendTrace(TraceObject: TTraceObject);
249			procedure ReceiveTrace(TraceObject: TTraceObject);
250			property DataAvailableEvent: TSingleLockGate read FDataAvailableEvent;
251			property WriterBusyEvent: TSingleLockGate read FWriterBusyEvent;
252			property ReadReadyEvent: TMultiLockGate read FReadReadyEvent;
253			property ReadFinishedEvent: TMultiLockGate read FReadFinishedEvent;
254			property WriteLock: TMutex read FWriteLock;
255			property MonitorCount: integer read GetMonitorCount;
256			property SharedMemory: TSharedMemory read FSharedMemory;
257			property MaxBufferSize: integer read FMaxBufferSize;
258			end;
259
260			{ TSharedMemory }
261
262			procedure TSharedMemory.GetSharedMemory(MemSize: integer);
263			var F: cint;
264			begin
265			{Get the Shared Memory and Semaphore IDs from the Global File if it exists
266			or create them and the file otherwise }
267
268			repeat
269			F := fpOpen(IPCFileName, O_WrOnly or O_Creat or O_Excl);
270			if F < 0 then
271			begin
272	tony	39	if fpgetErrno = ESysEEXIST {EEXIST} then
273	tony	33	begin
274			{ looks like it already exists}
275			Sleep(100);
276			F := fpOpen(IPCFileName,O_RdOnly);
277	tony	39	if (F < 0) and (fpgetErrno = ESysENOENT {ENOENT}) then
278	tony	33	{probably just got deleted }
279			else
280			if F < 0 then
281			IBError(ibxeCannotCreateSharedResource,['Error accessing IPC File - ' +
282			StrError(fpgetErrno)]);
283			end
284			else
285			IBError(ibxeCannotCreateSharedResource,['Error creating IPC File - ' +
286			StrError(fpgetErrno)]);
287			end
288			else
289			FInitialiser := true
290			until F >= 0;
291
292			if FInitialiser then
293			begin
294			FSharedMemoryID := shmget(IPC_PRIVATE,MemSize, IPC_CREAT or
295			S_IRUSR or S_IWUSR or S_IRGRP or S_IWGRP);
296			if FSharedMemoryID < 0 then
297			IBError(ibxeCannotCreateSharedResource,['Cannot create shared memory segment - ' +
298			StrError(fpgetErrno)]);
299
300			FSemaphoreSetID := semget(IPC_PRIVATE, cNumberOfSemaphores,IPC_CREAT or
301			S_IRUSR or S_IWUSR or S_IRGRP or S_IWGRP);
302			if FSemaphoreSetID < 0 then
303			IBError(ibxeCannotCreateSharedResource,['Cannot create shared semaphore set - ' +
304			StrError(fpgetErrno)]);
305
306			fpWrite(F,FSharedMemoryID,sizeof(FSharedMemoryID));
307			fpWrite(F,FSemaphoreSetID,sizeof(FSemaphoreSetID));
308			end
309			else
310			begin
311			fpRead(F,FSharedMemoryID,sizeof(FSharedMemoryID));
312			fpRead(F,FSemaphoreSetID,sizeof(FSemaphoreSetID));
313			if GetSemValue(cMonitorCounter) = 0 then
314			begin
315			FInitialiser := true;
316			//writeln('Opened file and is initialiser');
317			end
318			end;
319			fpClose(F);
320			end;
321
322			procedure TSharedMemory.DropSharedMemory;
323			var ds: TShmid_ds;
324			arg: tsemun;
325			begin
326			if shmctl(FSharedMemoryID,IPC_STAT,@ds) < 0 then
327			IBError(ibxeSV5APIError,['Error getting shared memory info' + strError(fpgetErrno)]);
328			if ds.shm_nattch = 0 then { we are the last one out - so, turn off the lights }
329			begin
330			shmctl(FSharedMemoryID,IPC_RMID,nil);
331			semctl(FSemaphoreSetID,0,IPC_RMID,arg);
332			DeleteFile(IPCFileName);
333			end;
334			end;
335
336			constructor TSharedMemory.Create(MemSize: integer);
337			begin
338			inherited Create;
339			FInitialiser := false;
340			GetSharedMemory(MemSize);
341			FBuffer := shmat(FSharedMemoryID,nil,0);
342			if PtrInt(FBuffer) = -1 then
343			IBError(ibxeCannotCreateSharedResource,[StrError(Errno)]);
344			FBufPtr := FBuffer;
345			FUnused := MemSize
346			end;
347
348			destructor TSharedMemory.Destroy;
349			begin
350			shmdt(FBuffer);
351			DropSharedMemory;
352			inherited Destroy;
353			end;
354
355			function TSharedMemory.Allocate(Size: integer): PChar;
356			begin
357			if Size > FUnused then
358			IBError(ibxeCannotCreateSharedResource, ['Not enough shared memory']);
359			Result := FBufPtr;
360
361			if Size = 0 then
362			begin
363			FLastAllocationSize := FUnused;
364			FUnused := 0
365			end
366			else
367			begin
368			FLastAllocationSize := Size;
369			Dec(FUnused,Size);
370			end;
371			Inc(FBufPtr,Size)
372			end;
373
374			{ TIpcCommon }
375
376			function TIpcCommon.GetSa: PSecurityAttributes;
377			begin
378			Result := nil
379			end;
380
381			function TIpcCommon.sem_op(SemNum, op: integer; flags: cshort): cint;
382			var sembuf: TSEMbuf;
383			begin
384			sembuf.sem_num := SemNum;
385			sembuf.sem_op:= op;
386			sembuf.sem_flg := flags or SEM_UNDO;
387			Result := semop(FSemaphoreSetID,@sembuf,1);
388			end;
389
390			function TIpcCommon.sem_timedop(SemNum, op: integer; timeout_secs: integer;
391			flags: cshort): cint;
392			var sembuf: TSEMbuf;
393			timeout: TimeSpec;
394			begin
395			sembuf.sem_num := SemNum;
396			sembuf.sem_op:= op;
397			sembuf.sem_flg := flags or SEM_UNDO;
398			timeout.tv_sec := timeout_secs;
399			timeout.tv_nsec := 0;
400			{$IFDEF HAS_SEMTIMEDOP}
401			Result := semtimedop(FSemaphoreSetID,@sembuf,1,@timeout);
402			{$ELSE}
403			Result := semop(FSemaphoreSetID,@sembuf,1); {May hang on race condition}
404			{$ENDIF}
405			end;
406
407			function TIpcCommon.GetSemValue(SemNum: integer): cint;
408			var args :TSEMun;
409			begin
410			Result := semctl(FSemaphoreSetID,SemNum,SEM_GETVAL,args);
411			if Result < 0 then
412			IBError(ibxeSV5APIError,['GetSemValue: '+strError(GetLastErrno)]);
413			end;
414
415			procedure TIpcCommon.SemInit(SemNum, AValue: cint);
416			var args :TSEMun;
417			begin
418			//writeln('Initialising ',SemNum,' to ',AValue);
419			args.val := AValue;
420			if semctl(FSemaphoreSetID,SemNum,SEM_SETVAL,args) < 0 then
421			IBError(ibxeCannotCreateSharedResource,['Unable to initialise Semaphone ' +
422			IntToStr(SemNum) + '- ' + StrError(GetLastErrno)]);
423
424			end;
425
426			{ TMutex }
427
428			constructor TMutex.Create(SemNumber: cint);
429			begin
430			inherited Create;
431			FMutexSemaphore := SemNumber;
432			if FInitialiser then
433			SemInit(FMutexSemaphore,1)
434			end;
435
436			{ Obtain ownership of the Mutex and prevent other threads from accessing protected resource }
437
438			procedure TMutex.Lock;
439			begin
440			//writeln('Lock: Entering Mutex ',FMutexSemaphore,' LockCount=',FLockCount,' State = ',GetSemValue(FMutexSemaphore));
441			if FLockCount = 0 then
442			sem_op(FMutexSemaphore,-1);
443			Inc(FLockCount);
444			//writeln('Lock: Mutex Exit');
445			end;
446
447			{Give up ownership of the Mutex and allow other threads access }
448
449			procedure TMutex.Unlock;
450			begin
451			//writeln('UnLock: Entering Mutex, LockCount=',FLockCount);
452			if FLockCount = 0 then Exit;
453			Dec(FLockCount);
454			if FLockCount = 0 then
455			sem_op(FMutexSemaphore,1);
456			//writeln('UnLock: Mutex Exit',' State = ',GetSemValue(FMutexSemaphore));
457			end;
458
459			{ TSingleLockGate }
460
461			function TSingleLockGate.GetWaitingThreads: integer;
462			begin
463			Result := FWaitingThreads^
464			end;
465
466			constructor TSingleLockGate.Create(SemNum: cint; AOwner: TGlobalInterface);
467			begin
468			inherited Create;
469			FOwner := AOwner;
470			FSignalledState := PInteger(FOwner.SharedMemory.Allocate(sizeof(FSignalledState)));
471			FWaitingThreads := PInteger(FOwner.SharedMemory.Allocate(sizeof(FWaitingThreads)));
472			FSemaphore := SemNum;
473			FMutex := SemNum + 1;
474			if FInitialiser then
475			begin
476			FSignalledState^ := 1;
477			FWaitingThreads^ := 0;
478			SemInit(FSemaphore,0);
479			SemInit(FMutex,1);
480			end;
481			end;
482
483			procedure TSingleLockGate.PassthroughGate;
484			begin
485			if FSignalledState^ = 0 then
486			begin
487			sem_op(FMutex,-1,0); //Acquire Mutex
488			Inc(FWaitingThreads^);
489			sem_op(FMutex,1,0); //Release Mutex
490			//writeln(ClassName + ': Wait State Entered ',FSemaphore,' = ',GetSemValue(FSemaphore));
491			sem_op(FSemaphore,-1,0); //Enter Wait
492			//writeln(ClassName + ': Wait State Ends ',FSemaphore);
493			end;
494			end;
495
496			procedure TSingleLockGate.Unlock;
497			begin
498			if FSignalledState^ = 0 then
499			begin
500			FSignalledState^ := 1;
501			sem_op(FMutex,-1,0); //Acquire Mutex
502			//writeln(ClassName + ': Unlocking' ,FSemaphore);
503			sem_op(FSemaphore,FWaitingThreads^,0);
504			FWaitingThreads^ := 0;
505			sem_op(FMutex,1,0); //Release Mutex
506			end;
507			end;
508
509			procedure TSingleLockGate.Lock;
510			begin
511			if FSignalledState^ = 1 then
512			begin
513			//writeln(ClassName + ': Locking Gate ',FSemaphore);
514			SemInit(FSemaphore,0);
515			FSignalledState^ := 0;
516			end;
517			end;
518
519			{ TMultilockGate }
520
521			constructor TMultilockGate.Create(SemNum: cint; AOwner: TGlobalInterface);
522			begin
523			inherited Create;
524			FOwner := AOwner;
525			FSemaphore := SemNum;
526			FMutex := SemNum + 1;
527			FLockCount := PInteger(FOwner.SharedMemory.Allocate(sizeof(FLockCount)));
528			if FInitialiser then
529			begin
530			FLockCount^ := 0;
531			SemInit(FSemaphore,1);
532			SemInit(FMutex,1);
533			end;
534			end;
535
536			function TMultilockGate.GetLockCount: integer;
537			begin
538			Result := FLockCount^
539			end;
540
541			procedure TMultilockGate.Lock;
542			begin
543			sem_op(FMutex,-1,0); //Acquire Mutex
544			if FLockCount^ = 0 then
545			begin
546			//writeln(ClassName,': Locking ',FSemaphore);
547			SemInit(FSemaphore,0);
548			end;
549			Inc(FLockCount^);
550			sem_op(FMutex,1,0); //Release Mutex
551			end;
552
553			procedure TMultilockGate.Unlock;
554			begin
555			sem_op(FMutex,-1,0); //Acquire Mutex
556			Dec(FLockCount^);
557			if FLockCount^ <= 0 then
558			begin
559			//writeln(ClassName,': UnLocking ',FSemaphore);
560			SemInit(FSemaphore,1);
561			FLockCount^ := 0
562			end;
563			sem_op(FMutex,1,0); //Release Mutex
564			end;
565
566			procedure TMultilockGate.PassthroughGate;
567			begin
568			if FLockCount^ = 0 then
569			Exit;
570			//writeln(ClassName,': Waiting on ',FSemaphore);
571			while sem_timedop(FSemaphore,-1,cDefaultTimeout) < 0 do
572			{looks like we lost a reader}
573			begin
574			if FLockCount^ > 0 then
575			begin
576			UnLock;
577			if assigned(FOnGateTimeout) then
578			OnGateTimeout(self)
579			end
580			end;
581			sem_op(FSemaphore,1);
582			//writeln(ClassName,': Wait done on ',FSemaphore);
583			end;
584
585
586			{ TGlobalInterface }
587
588			function TGlobalInterface.GetMonitorCount: integer;
589			begin
590			Result := GetSemValue(cMonitorCounter)
591			end;
592
593			constructor TGlobalInterface.Create;
594			begin
595			inherited Create;
596			FSharedMemory := TSharedMemory.Create(cMonitorHookSize);
597
598			FWriteLock := TMutex.Create(cMutexSemaphore);
599
600			FDataAvailableEvent := TSingleLockGate.Create(cDataAvailableEventSemaphore,self);
601			FWriterBusyEvent := TSingleLockGate.Create(cWriterBusyEventSemaphore,self);
602			FReadReadyEvent := TMultiLockGate.Create(cReadReadyEventSemaphore,self);
603			FReadFinishedEvent := TMultiLockGate.Create(cReadFinishedEventSemaphore,self);
604
605			if FInitialiser then
606			SemInit(cMonitorCounter,0);
607			FTraceDataType := PInteger(FSharedMemory.Allocate(sizeof(Integer)));
608			FTimeStamp := PDateTime(FSharedMemory.Allocate(sizeof(TDateTime)));
609			FBufferSize := PInteger(FSharedMemory.Allocate(sizeof(Integer)));
610			FBuffer := FSharedMemory.Allocate(0); //All remaining
611			FMaxBufferSize := FSharedMemory.LastAllocationSize;
612
613			if FInitialiser then
614			begin
615			FBufferSize^ := 0;
616			FDataAvailableEvent.Lock
617			end;
618			end;
619
620			destructor TGlobalInterface.Destroy;
621			begin
622			if assigned(FWriteLock) then FWriteLock.Free;
623			if assigned(FDataAvailableEvent) then FDataAvailableEvent.Free;
624			if assigned(FWriterBusyEvent) then FWriterBusyEvent.Free;
625			if assigned(FReadReadyEvent) then FReadReadyEvent.Free;
626			if assigned(FReadFinishedEvent) then FReadFinishedEvent.Free;
627			if assigned(FSharedMemory) then FSharedMemory.Free;
628			inherited Destroy;
629			end;
630
631			procedure TGlobalInterface.IncMonitorCount;
632			begin
633			sem_op(cMonitorCounter,1);
634			end;
635
636			procedure TGlobalInterface.DecMonitorCount;
637			begin
638			sem_op(cMonitorCounter,-1,IPC_NOWAIT);
639			end;
640
641			procedure TGlobalInterface.SendTrace(TraceObject: TTraceObject);
642			begin
643			FTraceDataType^ := Integer(TraceObject.FDataType);
644			FTimeStamp^ := TraceObject.FTimeStamp;
645			FBufferSize^ := Min(Length(TraceObject.FMsg), MaxBufferSize);
646			Move(TraceObject.FMsg[1], FBuffer^, FBufferSize^);
647			end;
648
649			procedure TGlobalInterface.ReceiveTrace(TraceObject: TTraceObject);
650			begin
651			SetString(TraceObject.FMsg, FBuffer, FBufferSize^);
652			TraceObject.FDataType := TTraceFlag(FTraceDataType^);
653			TraceObject.FTimeStamp := TDateTime(FTimeStamp^);
654			end;
655
656
657