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.
}

function GetLastErrno: cint;
begin
  Result := fpgetErrno
end;

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