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 }

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