Use OperationsBlockValidator on LoadSafeBoxFromStream

Will use multithread operations block validator when loading the safebox for the first time (when "checkall" variable is True)

src/core/UAccounts.pas

@@ -478,7 +478,7 @@ Procedure Check_Safebox_Integrity(sb : TPCSafebox; title: String);
 implementation
 
 uses
-  ULog, UAccountKeyStorage, math, UCommon;
+  ULog, UAccountKeyStorage, math, UCommon, UPCOperationsBlockValidator;
 
 { This function is for testing purpose only.
   Will check if Account Names are well assigned and stored }
@@ -2685,16 +2685,24 @@ Var
   nPos,posOffsetZone : Int64;
   offsets : Array of Cardinal;
   sbHeader : TPCSafeBoxHeader;
-  tc : TTickCount;
+  tc, LStartTickCount : TTickCount;
   previous_Block : TBlockAccount;
   do_check_blockchain_info : Boolean;
   aux_errors : String;
-begin
+  LUseMultiThreadOperationsBlockValidator, LAddToMultiThreadOperationsBlockValidator : Boolean;
+  LPCOperationsBlockValidator : TPCOperationsBlockValidator;
+  LValidatedOPOk, LValidatedOPError, LValidatedOPPending : Integer;
+begin
+  LPCOperationsBlockValidator := Nil;
+  if checkAll then begin
+    LUseMultiThreadOperationsBlockValidator := TLogicalCPUCount.GetLogicalCPUCount>1;
+  end else LUseMultiThreadOperationsBlockValidator := False;
   If Assigned(FPreviousSafeBox) then Raise Exception.Create('Cannot loadSafeBoxFromStream on a Safebox in a Separate chain');
   if (previousCheckedSafebox = Self) then previousCheckedSafebox := Nil; // Protection
   tc := TPlatform.GetTickCount;
   StartThreadSafe;
   try
+    LStartTickCount := tc;
     Clear;
     Result := false;
     Try
@@ -2733,6 +2741,12 @@ begin
       FBufferBlocksHash.SetLength(sbHeader.blocksCount*32);
       errors := 'Corrupted stream';
       do_check_blockchain_info := Not Assigned(previousCheckedSafebox);
+      if checkAll then
+        LPCOperationsBlockValidator := TPCOperationsBlockValidator.Create
+      else LPCOperationsBlockValidator := Nil;
+      try
+        if Assigned(LPCOperationsBlockValidator) then
+          LPCOperationsBlockValidator.StartThreads;
       for iblock := 0 to sbHeader.blockscount-1 do begin
         if (Assigned(progressNotify)) and ((TPlatform.GetElapsedMilliseconds(tc)>=500)) then begin
           tc := TPlatform.GetTickCount;
@@ -2786,6 +2800,7 @@ begin
         errors := 'Corrupted stream reading block '+inttostr(iblock+1)+'/'+inttostr(sbHeader.blockscount);
         If TStreamOp.ReadAnsiString(Stream,block.block_hash)<0 then exit;
         If Stream.Read(block.accumulatedWork,SizeOf(block.accumulatedWork)) < SizeOf(block.accumulatedWork) then exit;
+
         if checkAll then begin
           if (Not do_check_blockchain_info) then begin
             // Only check if block not found on previous or different block
@@ -2801,14 +2816,19 @@ begin
               // For TESTNET increase speed purposes, will only check latests blocks
             if ((iblock + (CT_BankToDiskEveryNBlocks * 10)) >= sbHeader.blockscount) then begin
             {$ENDIF}
-              If not IsValidNewOperationsBlock(block.blockchainInfo,False,True,aux_errors) then begin
+              LAddToMultiThreadOperationsBlockValidator := (LUseMultiThreadOperationsBlockValidator) and (block.blockchainInfo.protocol_version>=CT_PROTOCOL_4) and (Assigned(LPCOperationsBlockValidator));
+              If not IsValidNewOperationsBlock(block.blockchainInfo,False,Not LAddToMultiThreadOperationsBlockValidator,aux_errors) then begin
                 errors := errors + ' > ' + aux_errors;
                 exit;
               end;
+              if (LAddToMultiThreadOperationsBlockValidator) then begin
+                LPCOperationsBlockValidator.AddToValidate(block.blockchainInfo);
+              end;
             {$IFDEF TESTNET}
             end;
             {$ENDIF}
           end;
+
           // STEP 2: Check if valid block hash
           if (Not TBaseType.Equals(CalcBlockHash(block,FCurrentProtocol>=CT_PROTOCOL_2),block.block_hash)) then begin
             errors := errors + ' > Invalid block hash '+inttostr(iblock+1)+'/'+inttostr(sbHeader.blockscount);
@@ -2838,6 +2858,24 @@ begin
         if (block.blockchainInfo.protocol_version>FCurrentProtocol) And (block.blockchainInfo.protocol_version = CT_PROTOCOL_4) then begin
           FCurrentProtocol := CT_PROTOCOL_4;
         end;
+      end;
+        if Assigned(LPCOperationsBlockValidator) then begin
+          repeat
+            LPCOperationsBlockValidator.GetStatus(LValidatedOPOk, LValidatedOPError, LValidatedOPPending);
+            if LValidatedOPError>0 then begin
+              LPCOperationsBlockValidator.FillErrors(errors);
+              Exit;
+            end;
+            if LValidatedOPPending>0 then begin
+              if (Assigned(progressNotify)) and ((TPlatform.GetElapsedMilliseconds(tc)>=500)) then begin
+                tc := TPlatform.GetTickCount;
+                progressNotify(Self,Format('Validating OperationBlock info %d/%d',[LValidatedOPOk,LValidatedOPOk+LValidatedOPPending]),LValidatedOPOk,LValidatedOPOk+LValidatedOPPending);
+              end;
+
            end else Sleep(100);
+          until LValidatedOPPending<=0 ;
+        end;
+      finally
+        LPCOperationsBlockValidator.Free;
       end;
       if (Assigned(progressNotify)) then begin
         progressNotify(Self,'Checking Safebox integrity',sbHeader.blocksCount,sbHeader.blocksCount);
@@ -2881,6 +2919,8 @@ begin
   Finally
     EndThreadSave;
   end;
+  TLog.NewLog(ltdebug,ClassName,Format('Finalized read Safebox from blocks %d to %d (total %d blocks) in %.2f seconds',
+    [sbHeader.startBlock,sbHeader.endBlock,sbHeader.blocksCount,TPlatform.GetElapsedMilliseconds(LStartTickCount)/1000]));
 end;
 
 function TPCSafeBox.LoadSafeBoxFromStream(Stream: TStream; checkAll: Boolean; var LastReadBlock: TBlockAccount; var errors: String): Boolean;

src/core/UPCOperationsBlockValidator.pas

@@ -35,7 +35,7 @@ interface
 {$ENDIF}
 
 
-Uses UThread, UAccounts, UPCOrderedLists, UBlockChain,
+Uses UThread, UAccounts, UPCOrderedLists, UBlockChain, Classes,
   {$IFNDEF FPC}System.Generics.Collections{$ELSE}Generics.Collections{$ENDIF};
 
 type
@@ -53,20 +53,31 @@ type
 
   TPCOperationsBlockValidator = Class
   private
-    FLock : TPCCriticalSection;
+    FLockedList : TPCThreadList<TOperationBlock>;
+    FThreads : TList<TPCOperationsBlockValidatorThread>;
+    FErrorsList : TStringList;
     //
-    FPCOperationsCompList : TList<TPCOperationsComp>;
+    FPCOperationsCompList : TList<TPCOperationsComp>; // Optional field
     FLastIndexOperationsBlock : Integer;
     //
     FValidatedOkCount : Integer;
     FValidatedErrorCount : Integer;
   protected
     function GetNextOperationBlock(var ANextOperationBlock : TOperationBlock; var AIndex : Integer) : Boolean;
-    procedure SetOperationBlockResult(const AOperationBlock : TOperationBlock; AIndex : Integer; AValidated : Boolean);
+    procedure SetOperationBlockResult(const AOperationBlock : TOperationBlock; AIndex : Integer; AValidated : Boolean; const AErrorDetected : String);
   public
     Constructor Create;
     destructor Destroy; override;
-    function Validate(APCOperationsCompList : TList<TPCOperationsComp>; var AValidatedOkCount, AValidatedErrorCount : Integer) : Integer;
+    procedure AddToValidate(AList : TList<TOperationBlock>); overload;
+    procedure AddToValidate(AOperationBlock : TOperationBlock); overload;
+    procedure StartThreads;
+    procedure GetStatus(out AValidatedOk, AValidatedError, APendingToValidate : Integer);
+    procedure WaitUntilAllValidatedOrErrorFound;
+    procedure FillErrors(var AErrors : String);
+    procedure EndThreads;
+    property ValidatedOkCount : Integer read FValidatedOkCount;
+    property ValidatedErrorCount : Integer read FValidatedErrorCount;
+    function ValidateAndWaitUntilTerminate(APCOperationsCompList : TList<TPCOperationsComp>; var AValidatedOkCount, AValidatedErrorCount : Integer) : Integer;
     class function MultiThreadValidateOperationsBlock(APCOperationsCompList : TList<TPCOperationsComp>) : Boolean;
   End;
 
@@ -81,31 +92,116 @@ var _Cpus : Integer = 0;
 
 { TPCOperationsBlockValidator }
 
+procedure TPCOperationsBlockValidator.AddToValidate(AList: TList<TOperationBlock>);
+var i : Integer;
+  LList : TList<TOperationBlock>;
+begin
+  LList := FLockedList.LockList;
+  try
+    for i := 0 to AList.Count-1 do begin
+      LList.Add(AList[i]);
+    end;
+  finally
+    FLockedList.UnlockList;
+  end;
+end;
+
+procedure TPCOperationsBlockValidator.AddToValidate(AOperationBlock: TOperationBlock);
+var LList : TList<TOperationBlock>;
+begin
+  LList := FLockedList.LockList;
+  try
+    LList.Add(AOperationBlock);
+  finally
+    FLockedList.UnlockList;
+  end;
+end;
+
 constructor TPCOperationsBlockValidator.Create;
 begin
   FLastIndexOperationsBlock := -1;
-  FLock := TPCCriticalSection.Create('');
+  FLockedList := TPCThreadList<TOperationBlock>.Create(ClassName);
+  FPCOperationsCompList := Nil; // This field is external
+  FThreads := Nil;
+  FErrorsList := TStringList.Create;
 end;
 
 destructor TPCOperationsBlockValidator.Destroy;
 begin
-  FreeAndNil(FLock);
+  EndThreads;
+  FreeAndNil(FLockedList);
+  FreeAndNil(FErrorsList);
   inherited;
 end;
 
+procedure TPCOperationsBlockValidator.EndThreads;
+var i : Integer;
+begin
+  FLockedList.LockList;
+  try
+    if Not Assigned(FThreads) then Exit;
+
+    for i := 0 to FThreads.Count-1 do begin
+      FThreads[i].Terminate;
+    end;
+  finally
+    FLockedList.UnlockList;
+  end;
+  // WaitFor without locking
+  for i := 0 to FThreads.Count-1 do begin
+    FThreads[i].WaitFor;
+  end;
+  // Finished
+  FLockedList.LockList;
+  try
+    for i := 0 to FThreads.Count-1 do begin
+      FThreads[i].Free;
+    end;
+    FreeAndNil(FThreads);
+  finally
+    FLockedList.UnlockList;
+  end;
+end;
+
+procedure TPCOperationsBlockValidator.FillErrors(var AErrors: String);
+begin
+  FLockedList.LockList;
+  try
+    AErrors := FErrorsList.Text;
+  finally
+    FLockedList.UnlockList;
+  end;
+end;
+
 function TPCOperationsBlockValidator.GetNextOperationBlock(var ANextOperationBlock : TOperationBlock; var AIndex : Integer) : Boolean;
+var LList : TList<TOperationBlock>;
 begin
-  FLock.Acquire;
+  LList := FLockedList.LockList;
   try
     // Search new
     AIndex := FLastIndexOperationsBlock + 1; // Move to next
-    if (AIndex<FPCOperationsCompList.Count) then begin
-      ANextOperationBlock := FPCOperationsCompList[AIndex].OperationBlock;
+    if (AIndex<LList.Count) then begin
+      ANextOperationBlock := LList[AIndex];
       Result := True;
       FLastIndexOperationsBlock := AIndex;
     end else Result := False;
   finally
-    FLock.Release;
+    FLockedList.UnlockList;
+  end;
+end;
+
+procedure TPCOperationsBlockValidator.GetStatus(out AValidatedOk, AValidatedError, APendingToValidate: Integer);
+var LList : TList<TOperationBlock>;
+begin
+  LList := FLockedList.LockList;
+  try
+    AValidatedOk := FValidatedOkCount;
+    AValidatedError := FValidatedErrorCount;
+    if LList.Count>0 then
+      APendingToValidate := LList.Count - (FLastIndexOperationsBlock + 1)
+    else APendingToValidate := 0;
+  finally
+    FLockedList.UnlockList;
   end;
 end;
 
@@ -123,7 +219,7 @@ begin
   LTC := TPlatform.GetTickCount;
   LMultiThreadValidator := TPCOperationsBlockValidator.Create;
   try
-    LValidatedTotal := LMultiThreadValidator.Validate(APCOperationsCompList,LValidatedOk,LValidatedError);
+    LValidatedTotal := LMultiThreadValidator.ValidateAndWaitUntilTerminate(APCOperationsCompList,LValidatedOk,LValidatedError);
     LTC := TPlatform.GetElapsedMilliseconds(LTC);
     if (LValidatedTotal>0) and (LTC>0) then begin
       TLog.NewLog(ltdebug,ClassName,Format('Validated %d Operation blocks info with %d valids and %d Errors in %d miliseconds avg %.2f op/sec',[LValidatedTotal,LValidatedOk,LValidatedError,LTC,LValidatedTotal*1000/LTC]));
@@ -134,28 +230,35 @@ begin
   end;
 end;
 
-procedure TPCOperationsBlockValidator.SetOperationBlockResult(const AOperationBlock : TOperationBlock; AIndex : Integer; AValidated: Boolean);
+procedure TPCOperationsBlockValidator.SetOperationBlockResult(const AOperationBlock : TOperationBlock; AIndex : Integer; AValidated: Boolean; const AErrorDetected : String);
+var LList : TList<TOperationBlock>;
 begin
-  FLock.Acquire;
+  LList := FLockedList.LockList;
   try
     if AValidated then inc(FValidatedOkCount)
-    else inc(FValidatedErrorCount);
-    FPCOperationsCompList[AIndex].HasValidOperationBlockInfo := AValidated;
+    else begin
+      FErrorsList.Add(AErrorDetected + ' ' + TPCOperationsComp.OperationBlockToText(AOperationBlock) );
+      inc(FValidatedErrorCount);
+    end;
+    if Assigned(FPCOperationsCompList) then begin
+      FPCOperationsCompList[AIndex].HasValidOperationBlockInfo := AValidated;
+    end;
   finally
-    FLock.Release;
+    FLockedList.UnlockList;
   end;
 end;
 
-function TPCOperationsBlockValidator.Validate(APCOperationsCompList : TList<TPCOperationsComp>; var AValidatedOkCount, AValidatedErrorCount : Integer) : Integer;
+procedure TPCOperationsBlockValidator.StartThreads;
 var LMaxThreads : Integer;
-  LThreads : TList<TPCOperationsBlockValidatorThread>;
-  i,LTerminatedThreads : Integer;
+  i : Integer;
+  LList : TList<TOperationBlock>;
 begin
+  EndThreads; // Ensure no active threads
+
   FValidatedOkCount := 0;
   FValidatedErrorCount := 0;
-  if APCOperationsCompList.Count<=0 then Exit(0);
-
   FLastIndexOperationsBlock := -1;
+  FErrorsList.Clear;
 
   if _Cpus<=0 then begin
     _Cpus := TLogicalCPUCount.GetLogicalCPUCount;
@@ -165,41 +268,70 @@ begin
   if (LMaxThreads<=0) then LMaxThreads := 1;
   if (LMaxThreads>7) then LMaxThreads := 7;
 
-  LThreads := TList<TPCOperationsBlockValidatorThread>.Create;
-  Try
-    // Init values
-    FLastIndexOperationsBlock := -1;
-    FPCOperationsCompList := APCOperationsCompList;
-
-    // Step 1: Create the threads:
+  FThreads := TList<TPCOperationsBlockValidatorThread>.Create;
+  FLockedList.LockList;
+  try
     for i := 1 to LMaxThreads do begin
-      LThreads.Add( TPCOperationsBlockValidatorThread.Create(Self) );
+      FThreads.Add( TPCOperationsBlockValidatorThread.Create(Self) );
     end;
-    // Step 2: Start the threads
-    for i := 0 to LThreads.Count-1 do begin
-      LThreads[i].Suspended := False;
+    for i := 0 to FThreads.Count-1 do begin
+      FThreads[i].Suspended := False;
     end;
-    // Step 3: Wait until error of finalized
-    repeat
-      LTerminatedThreads := 0;
-      for i := 0 to LThreads.Count-1 do begin
-        if LThreads[i].Terminated then inc(LTerminatedThreads);
-      end;
-      Sleep(1);
-    until (LTerminatedThreads>=LThreads.Count);
-  Finally
-    for i := 0 to LThreads.Count-1 do begin
-      LThreads[i].Terminate;
-      LThreads[i].WaitFor;
-      LThreads[i].Free;
+  finally
+    FLockedList.UnlockList;
+  end;
+
+end;
+
+function TPCOperationsBlockValidator.ValidateAndWaitUntilTerminate(APCOperationsCompList : TList<TPCOperationsComp>; var AValidatedOkCount, AValidatedErrorCount : Integer) : Integer;
+var
+  i,LTerminatedThreads : Integer;
+  LList : TList<TOperationBlock>;
+begin
+  FValidatedOkCount := 0;
+  FValidatedErrorCount := 0;
+  if APCOperationsCompList.Count<=0 then Exit(0);
+
+  EndThreads;
+
+  LList := FLockedList.LockList;
+  try
+    for i := 0 to APCOperationsCompList.Count-1 do begin
+      LList.Add(APCOperationsCompList[i].OperationBlock);
     end;
-    LThreads.Free;
+  finally
+    FLockedList.UnlockList;
+  end;
+
+  FPCOperationsCompList := APCOperationsCompList;
+
+  StartThreads;
+  try
+    // Wait until error of finalized
+    WaitUntilAllValidatedOrErrorFound;
+  Finally
+    EndThreads;
   End;
   AValidatedOkCount := FValidatedOkCount;
   AValidatedErrorCount := FValidatedErrorCount;
   Result := FPCOperationsCompList.Count;
 end;
 
+procedure TPCOperationsBlockValidator.WaitUntilAllValidatedOrErrorFound;
+var LContinue : Boolean;
+  LList : TList<TOperationBlock>;
+begin
+  repeat
+    Sleep(10);
+    LList := FLockedList.LockList;
+    try
+      LContinue := (FValidatedErrorCount=0) And (FLastIndexOperationsBlock+1<LList.Count);
+    finally
+      FLockedList.UnlockList;
+    end;
+  until (Not LContinue);
+end;
+
 { TPCOperationsBlockValidatorThread }
 
 procedure TPCOperationsBlockValidatorThread.BCExecute;
@@ -211,8 +343,8 @@ begin
   repeat
     if FValidator.GetNextOperationBlock( LOperationBlock, LIndex ) then begin
       LValidated := TPCSafeBox.IsValidOperationBlock(LOperationBlock,LErrors);
-      FValidator.SetOperationBlockResult(LOperationBlock, LIndex, LValidated);
-    end else Terminate;
+      FValidator.SetOperationBlockResult(LOperationBlock, LIndex, LValidated, LErrors);
+    end else Sleep(1);
   until (Terminated);
 end;