mmringbf.pas

一套及时通讯的原码

{========================================================================}
{=                (c) 1995-98 SwiftSoft Ronald Dittrich                 =}
{========================================================================}
{=                          All Rights Reserved                         =}
{========================================================================}
{=  D 01099 Dresden             = Fax.: [phone]              =}
{=  Loewenstr.7a                = [e-mail]                     =}
{========================================================================}
{=  Actual versions on http://www.swiftsoft.de/mmtools.html             =}
{========================================================================}
{=  This code is for reference purposes only and may not be copied or   =}
{=  distributed in any format electronic or otherwise except one copy   =}
{=  for backup purposes.                                                =}
{=                                                                      =}
{=  No Delphi Component Kit or Component individually or in a collection=}
{=  subclassed or otherwise from the code in this unit, or associated   =}
{=  .pas, .dfm, .dcu, .asm or .obj files may be sold or distributed     =}
{=  without express permission from SwiftSoft.                          =}
{=                                                                      =}
{=  For more licence informations please refer to the associated        =}
{=  HelpFile.                                                           =}
{========================================================================}
{=  $Date: 09.09.98 - 12:04:56 $                                        =}
{========================================================================}
unit MMRingBf;

{$C FIXED PRELOAD PERMANENT}

{$I COMPILER.INC}

{.$DEFINE _MMDEBUG}

interface

uses
  Windows,
  SysUtils,
  Messages,
  Classes,
  Controls,
  Forms,
  MMSystem,
  MMObj,
  MMDSPObj,
  MMDSPMtr,
  MMUtils,
  MMAlloc,
  MMstring
  {$IFDEF _MMDEBUG}
  ,MMDebug
  {$ENDIF};

{$IFDEF _MMDEBUG}
   const
        DEBUGLEVEL = 0;  { 0,1,2 }
{$ENDIF}

const
  CACHE_PRIORITY   : TThreadPriority = tpHigher;
  SLEEPTIME        : integer = 1;
  MINBUFFERSIZE    = 32;

type
  EMMBufferError = class(Exception);

  PBufferItem    = ^TBufferItem;
  TBufferItem    = record
      lpData     : PChar;
      dwSize     : DWORD;
      dwUser     : DWORD;
      dwFlags    : DWORD;   { set to 1 if this is the last buffer }
  end;

  PBufferPool    = ^TBufferPool;
  TBufferPool    = array[0..0] of PBufferItem;

  TMMRingBuffer  = class;

  {-- TMMQueueThread ----------------------------------------------------}
  TMMQueueThread = class(TMMDSPThread)
  private
     procedure Execute; override;
  end;

  TMMQueueMode   = (qmNone,qmRead,qmWrite,qmDuplex);

  {-- TMMRingBuffer -----------------------------------------------------}
  TMMRingBuffer  = class(TMMDSPComponent)
  private
      FBufferPool  : PBufferPool;   { array of buffers used for queue    }
      FTempBuffer  : PWaveHdr;      { buffer for queue handling          }
      FNumBuffers  : integer;       { number of buffers used for queue   }
      FQueueCount  : integer;       { number of buffers in the queue     }
      FBytesQueued : Longint;       { number of bytes in the queue       }
      FInIndex     : integer;       { index for next push operation      }
      FOutIndex    : integer;       { index for next get operation       }
      FLastIndex   : integer;       { index for last valid get operation }
      FOpen        : Boolean;       { ringbuffer is ready to go now      }
      FStarted     : Boolean;       { flag to check if queue is started  }
      FStopping    : Boolean;       { flag to check if queue is stopping }
      FQueueMode   : TMMQueueMode;  { queue mode for current operation   }
      FLoopRec     : TMMLoopRec;    { record for loop handling           }
      FReadSize    : Longint;       { rounded read size in auto mode     }
      FWriteSize   : Longint;       { rounded write size in auto mode    }
      FEnabled     : Boolean;       { Enable/Disable the Queue           }
      FWaitIfFull  : Boolean;       { wait with new buffers if queue full}
      FLoopIfEmpty : Boolean;       { loop the last buffer if empty      }
      FMainWaiting : Boolean;       { the mainthread is currently waiting}
      FTimeOut     : integer;       { TimeOut value                      }

      FThreadError : Boolean;        { Error in Thread Handler           }
      FQueueThread : TMMQueueThread;{ Thread for buffer handling         }
      DataSection  : TRtlCriticalSection;{ CriticalSection Object        }
      DataSectionOK: Boolean;       { CriticalSection prepared           }
      FGeneralEvent: THandle;       { event for thread notification      }
      FQueueEvent  : THandle;       { event object for notify handling   }
      FDoneEvent   : THandle;       { current queue operation is done    }
      FCloseEvent  : THandle;       { event object to close the thread   }
      FSyncProc    : TThreadMethod;
      FAllocator   : TMMAllocator;

      FOnOverflow      : TNotifyEvent;
      FOnEmpty         : TNotifyEvent;
      FOnFillQueue     : TNotifyEvent;
      FOnFillQueueEnd  : TNotifyEvent;
      FOnFlushQueue    : TNotifyEvent;
      FOnFlushQueueEnd : TNotifyEvent;

      function  GetQueueCount: integer;
      function  GetBytesQueued: Longint;
      function  GetBytesFree: Longint;
      procedure SetNumBuffers(aValue: integer);
      procedure SetQueueMode(aValue: TMMQueueMode);

      function  AllocBuffer: PBufferItem;
      procedure CreateBuffers;
      procedure FreeBuffers;
      procedure FillQueue;
      procedure FlushQueue;

      procedure InitThread;
      procedure DoneThread;

      procedure InitCritical;
      procedure EnterCritical;
      procedure LeaveCritical;
      procedure DoneCritical;
      procedure WaitForDoneEvent(WaitEver: Boolean);

  protected
      procedure SetBufferSize(aValue: Longint); override;
      function  GetBufferSize: Longint; override;

      procedure Opened; override;
      procedure Closed; override;
      procedure Started; override;
      procedure Stopped; override;
      procedure Reseting; override;

      procedure BufferReady(lpwh: PWaveHdr); override;
      procedure BufferLoad(lpwh: PWaveHdr; var MoreBuffers: Boolean); override;

      procedure DoOverflow; dynamic;
      procedure DoEmpty; dynamic;
      procedure DoFillQueue; dynamic;
      procedure DoFillQueueEnd; dynamic;
      procedure DoFlushQueue; dynamic;
      procedure DoFlushQueueEnd; dynamic;

  public
      constructor Create(AOwner: TComponent); override;
      destructor Destroy; override;

      procedure Open;
      procedure Close;
      procedure Flush;
      procedure Clear;

      { maybe you must syncronize anything in the events ? }
      procedure SynchronizeVCL(VCLProc: TThreadMethod);

      function  Empty: Boolean;
      function  Full: Boolean;

      procedure PutBuffer(lpData: PChar; dwSize,dwUser,dwFlags: DWORD);
      procedure GetBuffer(lpData: PChar; var dwSize,dwUser,dwFlags: DWORD);

      function PutBlock(lpData: PChar; dwSize,dwFlags: DWORD): DWORD;
      function GetBlock(lpData: PChar; dwSize: DWORD): DWORD;

      property QueueCount: integer read GetQueueCount;
      property BytesQueued: Longint read GetBytesQueued;
      property BytesFree: Longint read GetBytesFree;

  published
      property OnOverlow: TNotifyEvent read FOnOverflow write FOnOverflow;
      property OnEmpty: TNotifyEvent read FOnEmpty write FOnEmpty;
      property OnFillQueue: TNotifyEvent read FOnFillQueue write FOnFillQueue;
      property OnFillQueueEnd: TNotifyEvent read FOnFillQueueEnd write FOnFillQueueEnd;
      property OnFlushQueue: TNotifyEvent read FOnFlushQueue write FOnFlushQueue;
      property OnFlushQueueEnd: TNotifyEvent read FOnFlushQueueEnd write FOnFlushQueueEnd;
      property OnBufferReady;
      property OnBufferLoad;

      property Input;
      property Output;
      property BufferSize;
      property NumBuffers: integer read FNumBuffers write SetNumBuffers default 20;
      property QueueMode: TMMQueueMode read FQueueMode write SetQueueMode default qmNone;
      property Enabled: Boolean read FEnabled write FEnabled default True;
      property WaitIfFull: Boolean read FWaitIfFull write FWaitIfFull default False;
      property LoopIfEmpty: Boolean read FLoopIfEmpty write FLoopIfEmpty default False;
      property TimeOut: integer read FTimeOut write FTimeOut default 10000;
  end;

implementation

const
     { private flags for queue handling }
     HDR_END  = $0001;  { last header flag }
     HDR_LOOP = $0002;  { looping flag     }

{$IFDEF _MMDEBUG}
{-------------------------------------------------------------------------}
procedure DebugStr(Level: integer; s: String);
var
   found: Boolean;
   ID,i: integer;
begin
   if (s <> ' ') then s := 'Queue: '+s;

   if (ThreadList <> nil) then
   begin
      ID := GetCurrentThreadID;
      Found := False;
      for i := 0 to ThreadList.Count-1 do
      begin
         if TMMDSPThread(ThreadList[i]).ThreadID = ID then
         begin
            s := TMMDSPThread(ThreadList[i]).Owner.ClassName + ' '+s;
            Found := True;
            break;
         end;
      end;
   end;

   if not Found then s := 'MainThread '+s;

   DB_WriteStrLn(Level,s);
end;
{$ENDIF}

{== TMMRingBuffer ========================================================}
constructor TMMRingBuffer.Create(aOwner: TComponent);
begin
   inherited Create(aOwner);

   FBufferPool := nil;
   FTempBuffer := nil;
   FNumBuffers := 20;
   FEnabled    := True;
   FOpen       := False;
   FStarted    := False;
   FStopping   := False;
   FQueueMode  := qmNone;
   FMainWaiting:= False;
   FWaitIfFull := False;
   FLoopIfEmpty:= False;
   FTimeOut    := 10000;

   FAllocator  := TMMAllocator.Create;

   Clear;

   DataSectionOK := False;

   ErrorCode := ComponentRegistered(InitCode, Self, ClassName);
   if (ErrorCode <> 0) then RegisterFailed(InitCode, Self , ClassName);
end;

{-- TMMRingBuffer --------------------------------------------------------}
destructor TMMRingBuffer.Destroy;
begin
   Close;

   if assigned(FAllocator) then FAllocator.Free;

   inherited Destroy;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.SynchronizeVCL(VCLProc: TThreadMethod);
begin
   {$IFDEF _MMDEBUG}
   DebugStr(0,'Enter Sync...');
   {$ENDIF}

   if FMainWaiting then
   begin
      FSyncProc := VCLProc;
      while assigned(FSyncProc) do Sleep(SLEEPTIME);
   end
   else GlobalSynchronize(VCLProc);

   {$IFDEF _MMDEBUG}
   DebugStr(0,'Leave Sync...');
   {$ENDIF}
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.InitThread;
begin
   if (FQueueMode in [qmRead,qmWrite]) then
   begin
      EnterCritical;
      try
         FThreadError := False;

         { create event objects }
         FGeneralEvent := CreateEvent(nil, False, False, nil);
         FQueueEvent   := CreateEvent(nil, True, False, nil);
         FDoneEvent    := CreateEvent(nil, False, False, nil);
         FCloseEvent   := CreateEvent(nil, False, False, nil);

         { create the thread }
         FQueueThread := TMMQueueThread.CreateSuspended(Self);
         if (FQueueThread = nil) then
            raise EMMBufferError.Create('RingBuffer:'#10#13+LoadResStr(IDS_THREADERROR));

         FQueueThread.FreeOnTerminate := True;
         FQueueThread.Resume;

         { Wait for it to start... }
         if WaitForSingleObject(FGeneralEvent, 5000) <> WAIT_OBJECT_0 then
            raise EMMBufferError.Create('RingBuffer:'#10#13+LoadResStr(IDS_THREADERROR));

         {$IFDEF _MMDEBUG}
         DebugStr(0,'Thread Started');
         {$ENDIF}

         {$IFDEF TRIAL}
         {$DEFINE _HACK1}
         {$I MMHACK.INC}
         {$ENDIF}

      finally
         LeaveCritical;
      end;
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoneThread;
begin
   if (FQueueMode in [qmRead,qmWrite]) and not FThreadError then
   begin
      {$IFDEF _MMDEBUG}
      DebugStr(0,'Shot down Tread');
      {$ENDIF}

      { Wait until the thread is ready with the work }
      WaitForDoneEvent(False);

      { Force the output thread to close... }
      SetEvent(FCloseEvent);

      { ...and wait for it to die }
      WaitForSingleObject(FGeneralEvent, 5000);

      { release events }
      CloseHandle(FGeneralEvent);
      CloseHandle(FQueueEvent);
      CloseHandle(FDoneEvent);
      CloseHandle(FCloseEvent);

      {$IFDEF _MMDEBUG}
      DebugStr(0,'Thread Terminated');
      {$ENDIF}
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.InitCritical;
begin
   if not DataSectionOK then
   begin
      { create critical section object }
      FillChar(DataSection, SizeOf(DataSection), 0);
      InitializeCriticalSection(DataSection);
      DataSectionOK := True;
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.EnterCritical;
begin
   if FOpen and DataSectionOK then EnterCriticalSection(DataSection);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.LeaveCritical;
begin
   if FOpen and DataSectionOK then LeaveCriticalSection(DataSection);
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.DoneCritical;
begin
   if DataSectionOK then
   begin
      DataSectionOK := False;
      DeleteCriticalSection(DataSection);
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}
procedure TMMRingBuffer.WaitForDoneEvent(WaitEver: Boolean);
begin
   if (FQueueMode in [qmRead,qmWrite]) then
   begin
      EnterCritical;

      { wait until the thread is ready with the work }
      if WaitEver or (WaitForSingleObject(FQueueEvent,0) = WAIT_OBJECT_0) then
      begin
         LeaveCritical;

         {$IFDEF _MMDEBUG}
         DebugStr(0,'Wait until thread is ready...');
         {$ENDIF}

         while WaitForSingleObject(FDoneEvent,0) = WAIT_OBJECT_0 do Sleep(SLEEPTIME);

         {$IFDEF _MMDEBUG}
         DebugStr(0,'Thread now ready...');
         {$ENDIF}
      end
      else LeaveCritical;
      ResetEvent(FDoneEvent);
   end;
end;

{-- TMMRingBuffer --------------------------------------------------------}