Changeset 163 for PinConnection/UCommProtocol.pas

Timestamp:

Feb 8, 2011, 11:02:02 AM (13 years ago)

Author:

george

Message:

• Modified: Units are dependent on MicroThreading system now.

Location:

PinConnection

Files:

• . (modified) (1 prop)
• UCommProtocol.pas (modified) (20 diffs)

PinConnection

@@ -1 +1 @@
 lib
+backup

@@ -1 +1 @@
 lib
+backup

PinConnection/UCommProtocol.pas

@@ -6 +6 @@
 
 uses
-  Classes, SysUtils, UVarBlockSerializer, syncobjs, UCommPin,
-  UDebugLog, UStreamHelper, StopWatch, SpecializedList, UCommon;
+  Classes, SysUtils, UVarBlockSerializer, syncobjs, UCommPin, UMicroThreading,
+  UDebugLog, UStreamHelper, StopWatch, SpecializedList, UCommon, UPlatform,
+  DateUtils;
 
 type
@@ -25 +26 @@
   private
     RepeatCounter: integer;
-    ReceiveEvent: TEvent;
+    ReceiveEvent: TMicroThreadEvent;
     Request: TStreamHelper;
     ResponseParameters: TVarBlockIndexed;
     TransmitTime: TDateTime;
-    StopWatch: TStopWatch;
   public
-    Lock: TCriticalSection;
+    Lock: TMicroThreadCriticalSection;
     SequenceNumber: Integer;
     ResponseCode: Integer;
@@ -38 +38 @@
     Timeouted: Boolean;
     CommandIndex: TListInteger;
-    Latency: Double;
+    Latency: TDateTime;
     constructor Create;
     destructor Destroy; override;
@@ -48 +48 @@
     SequenceNumber: integer;
     Parent: TCommProtocol;
-    Lock: TCriticalSection;
-    function GetFree: TDeviceProtocolSession;
+    Lock: TMicroThreadCriticalSection;
+    procedure Add(Session: TDeviceProtocolSession);
     function GetBySequence(Sequence: integer): TDeviceProtocolSession;
     procedure Remove(Session: TDeviceProtocolSession);
     constructor Create;
     destructor Destroy; override;
+    function GetSequenceNumber: Integer;
   end;
 
@@ -62 +63 @@
   { TRetransmitCheckThread }
 
-  TRetransmitCheckThread = class(TThread)
+  TRetransmitCheckThread = class(TMicroThread)
   public
     Parent: TCommProtocol;
@@ -94 +95 @@
     Pin: TCommPin;
     LastCommandResponseTime: TDateTime;
-    LastLatency: Double;
+    LastLatency: TDateTime;
     procedure SendCommand(Command: array of integer;
       ResponseParameters: TVarBlockIndexed = nil;
@@ -149 +150 @@
                 if TestIndex(4) then CommandError := ReadVarUInt(4)
                   else CommandError := 0;
-                StopWatch.Stop;
-                Latency := StopWatch.ElapsedMiliseconds;
+                Latency := NowPrecise - TransmitTime;
                 ReceiveEvent.SetEvent;
               finally
@@ -254 +254 @@
       Parent := Self;
       FreeOnTerminate := False;
+      Name := 'CommProtocol';
       Start;
     end;
@@ -268 +269 @@
   NewRequest: TVarBlockIndexed;
 begin
-  Session := Sessions.GetFree;
   try
+    Session := TDeviceProtocolSession.Create;
+    Sessions.Add(Session);
     NewRequest := TVarBlockIndexed.Create;
 
@@ -276 +278 @@
       try
         Session.Lock.Acquire;
+        Session.SequenceNumber := Sessions.GetSequenceNumber;
         CommandIndex.Clear;
         CommandIndex.AddArray(Command);
@@ -290 +293 @@
 
         // Wait for free remote buffer
-        repeat
-          Sleep(1);
-        until (RemoteBufferUsed + Request.Size) <= RemoteBufferSize;
-
-        StopWatch.Start;
-        TransmitTime := Now;
+        while (RemoteBufferUsed + Request.Size) > RemoteBufferSize do
+          MTSleep(1 * OneMillisecond);
+
+        //StopWatch.Start;
+        TransmitTime := NowPrecise;
         Pin.Send(Request);
       finally
@@ -301 +303 @@
       end;
       try
-        RemoteBufferUsed := RemoteBufferUsed + Request.Size;
-        while ReceiveEvent.WaitFor(10) = wrTimeout do begin
+        try
+          Sessions.Lock.Acquire;
+          RemoteBufferUsed := RemoteBufferUsed + Request.Size;
+        finally
+          Sessions.Lock.Release;
+        end;
+        while MTWaitForEvent(ReceiveEvent, 10 * OneMillisecond) = wrTimeout do begin
           if Timeouted then
             raise ECommTimeout.Create(SResponseTimeout);
@@ -311 +318 @@
           raise ECommResponseCodeError.Create(Format(SResponseError, [CommandIndex.Implode('.', IntToStr), IntToStr(ResponseCode)]));
         end;
-        LastCommandResponseTime := Now;
+        LastCommandResponseTime := NowPrecise;
         LastLatency := Latency;
       finally
-        RemoteBufferUsed := RemoteBufferUsed - Session.Request.Size;
-        if RemoteBufferUsed < 0 then RemoteBufferUsed := 0;
+        try
+          Sessions.Lock.Acquire;
+          RemoteBufferUsed := RemoteBufferUsed - Session.Request.Size;
+          if RemoteBufferUsed < 0 then RemoteBufferUsed := 0;
+        finally
+          Sessions.Lock.Release;
+        end;
         Sessions.Remove(Session);
       end;
@@ -332 +344 @@
   MaxSessionCount := 10;
   MaxSequenceNumber := 127;
-  RetransmitTimeout := 1 / 24 / 3600 * 1;
+  RetransmitTimeout := 2 * OneSecond;
   RetransmitRepeatCount := 3;
   RetransmitTotalCount := 0;
@@ -350 +362 @@
 constructor TDeviceProtocolSession.Create;
 begin
-  StopWatch := TStopWatch.Create;
   ResponseCode := 0;
-  Lock := TCriticalSection.Create;
-  ReceiveEvent := TEvent.Create(nil, False, False, '');
+  Lock := TMicroThreadCriticalSection.Create;
+  ReceiveEvent := TMicroThreadEvent.Create;
+  ReceiveEvent.AutoReset := False;
   Request := TStreamHelper.Create;
   ResponseParameters := nil;
   CommandIndex := TListInteger.Create;
   Latency := 0;
+  TransmitTime := 0;
 end;
 
@@ -366 +379 @@
   ReceiveEvent.Free;
   Lock.Free;
-  StopWatch.Free;
   inherited Destroy;
 end;
@@ -372 +384 @@
 { TDeviceProtocolSessionList }
 
-function TDeviceProtocolSessionList.GetFree: TDeviceProtocolSession;
+procedure TDeviceProtocolSessionList.Add(Session: TDeviceProtocolSession);
 begin
   // Block if no free session available
-  Lock.Acquire;
   try
+    Lock.Acquire;
     while Count >= Parent.MaxSessionCount do
     begin
-      Lock.Release;
-      Sleep(1);
-      Lock.Acquire;
-    end;
-    Result := TDeviceProtocolSession.Create;
-    Result.SequenceNumber := SequenceNumber;
-    Inc(SequenceNumber);
-    if SequenceNumber > Parent.MaxSequenceNumber then
-      SequenceNumber := 0;
-    Add(Result);
+      try
+        Lock.Release;
+        MTSleep(1 * OneMillisecond);
+      finally
+        Lock.Acquire;
+      end;
+    end;
+    inherited Add(Session);
   finally
     Lock.Release;
@@ -421 +431 @@
 begin
   inherited Create;
-  Lock := TCriticalSection.Create;
+  Lock := TMicroThreadCriticalSection.Create;
 end;
 
@@ -428 +438 @@
   // Free session list before freeing Lock
   // instead of freeing in inherited Destroy in TListObject
-  Lock.Acquire;
-  Clear;
-  Lock.Release;
+  try
+    Lock.Acquire;
+    Clear;
+  finally
+    Lock.Release;
+  end;
 
   Lock.Free;
@@ -436 +449 @@
 end;
 
+function TDeviceProtocolSessionList.GetSequenceNumber: Integer;
+begin
+  try
+    Lock.Acquire;
+    Inc(SequenceNumber);
+    if SequenceNumber > Parent.MaxSequenceNumber then
+      SequenceNumber := 0;
+    Result := SequenceNumber;
+  finally
+    Lock.Release;
+  end;
+end;
+
 { TRetransmitCheckThread }
 
 procedure TRetransmitCheckThread.Execute;
 var
-  I: integer;
-begin
-  with Parent do repeat
-      try
-        try
-          Parent.Sessions.Lock.Acquire;
-          for I := 0 to Sessions.Count - 1 do begin
-          with TDeviceProtocolSession(Sessions[I]) do begin
-              try
-                Lock.Acquire;
-                if Now > (TransmitTime + RetransmitTimeout) then begin
-                  if RepeatCounter < RetransmitRepeatCount then begin
-                    Pin.Send(Request);
-                    StopWatch.Start;
-                    TransmitTime := Now;
-                    Inc(RepeatCounter);
-                    Inc(RetransmitTotalCount);
-                  end else
-                    Timeouted := True;
-                end;
-              finally
-                Lock.Release;
-              end;
+  I: Integer;
+  C: Integer;
+  Session: TDeviceProtocolSession;
+begin
+  with Parent do
+  repeat
+    try
+      Parent.Sessions.Lock.Acquire;
+      I := 0;
+      while I < Sessions.Count do begin
+        Session := TDeviceProtocolSession(Sessions[I]);
+        with TDeviceProtocolSession(Sessions[I]) do begin
+          try
+            Session.Lock.Acquire;
+            if (TransmitTime > 0) and (NowPrecise > (TransmitTime + RetransmitTimeout)) then begin
+              if RepeatCounter < RetransmitRepeatCount then begin
+                Pin.Send(Request);
+                TransmitTime := NowPrecise;
+                Inc(RepeatCounter);
+                Inc(RetransmitTotalCount);
+              end else
+                Timeouted := True;
             end;
+          finally
+            Session.Lock.Release;
           end;
-        finally
-          Parent.Sessions.Lock.Release;
-        end;
-
-      if not Terminated then
-        Sleep(CheckPeriod);
-      except
-        on E: Exception do begin
-          if Assigned(ExceptionHandler) then ExceptionHandler(Self, E);
-        end;
+        end;
+        Inc(I);
       end;
-    until Terminated;
+    finally
+      Parent.Sessions.Lock.Release;
+    end;
+
+    if not Terminated then
+      MTSleep(CheckPeriod * OneMillisecond);
+  until Terminated;
 end;
 

@@ -1 +1 @@
 lib
+backup