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
#	Content
1	{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	{$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	{
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	if fpgetErrno = ESysEEXIST {EEXIST} then
273	begin
274	{ looks like it already exists}
275	Sleep(100);
276	F := fpOpen(IPCFileName,O_RdOnly);
277	if (F < 0) and (fpgetErrno = ESysENOENT {ENOENT}) then
278	{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