Hallo,

mit den neuen Klassen kann man ja prima mal schnell eine procedure in einen Task schubsen. Aber, ist das auch problemlos mit mehreren Tasks möglich?

Hintergrund: Ich hab ein kleines Backup-Programm nur für den Eigenbedarf, das ich bei Gelegenheit anschubse, um die wichtigsten Daten in verschiedenen Ordnern (je Ordner eine separate Zip-Datei) auf mehreren Partitionen/Platten zu zippen. Das funktioniert schon jahrelang prima, mich hat nur die lange Ausführungszeit etwas gestört. Also hab ich den jeweiligen Zip-Vorgang in einen Thread ausgelagert. Nicht mit abgeleiteter Thread-Klasse sondern über die Funktionen der (für mich) neuen Unit "System.Threading". Ich will ganz einfach mal wissen/testen, was geht, und was nicht, weil ich ansonsten kein Threadprofi bin

Hardware: Windows 8.1 64Bit, 16 GB RAM, Boot-Partition auf einer SSD, und zwei HDD's mit 1x 2TB und 1x 1TB.

Es werden maximal 4 Threads gestartet. Das läuft am Anfang auch prima los, aber je länger das Programm läuft, umso langsamer wird mein PC. An der CPU-Auslastung kann es eigentlich nicht liegen. Also was läuft da schief?

zusammenfalten · markieren

Delphi-Quellcode:

unit uMyBackup;

interface

uses
  Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants,
  System.Classes, Vcl.Graphics,
  Vcl.Controls, Vcl.Forms, Vcl.Dialogs, ZipForge, Vcl.ExtCtrls, Vcl.ComCtrls,
  Vcl.StdCtrls, System.SyncObjs;

type
  TFrmMyDelphiBackup = class(TForm)
    StatusBar: TStatusBar;
    Panel1: TPanel;
    BtnStart: TButton;
    ZipForge1: TZipForge;
    lvZipList: TListView;
    procedure FormShow(Sender: TObject);
    procedure FormDestroy(Sender: TObject);
    procedure BtnStartClick(Sender: TObject);
    procedure ZipperOverallProgress(Sender: TObject; Progress: Double; Operation: TZFProcessOperation;
      ProgressPhase: TZFProgressPhase; var Cancel: Boolean);
    procedure ZipForge1ProcessFileFailure(Sender: TObject; FileName: string; Operation: TZFProcessOperation;
      NativeError, ErrorCode: Integer; ErrorMessage: string; var Action: TZFAction);
  private
    { Private-Deklarationen }
    FMaxThreads: Integer;
    FTempStrings: TStrings;
    FApplicationBaseDir: string;
    FCritSection: TCriticalSection;
    function IniFileName: string;
    function IniLesen: Boolean;
    procedure IniSchreiben;
    procedure LogSchreiben;
  public
    { Public-Deklarationen }
    procedure StatusMessage(aValue: string);
    procedure StringListToListView(const aItems: TStrings);
    procedure FileListZippen(aID: Integer; const aZipFileName: string; const aZipperBaseDir: string;
      const aZipFileMask: string);
  end;

var
  FrmMyDelphiBackup: TFrmMyDelphiBackup;

implementation

{$R *.dfm}

uses System.Threading, System.IOUtils, System.StrUtils, System.IniFiles;

const
  cCrLf = #13#10;
  cCrLf2 = cCrLf + cCrLf;
  cMaxThreadsDefault = 4;

resourcestring
  rsAppName = 'Mein kleines Zip-Backup';
  rsTrennChar = ' # ';
  { -Ini-File- }
  rsIniName = 'BackupZipper-Einstellungen.ini';
  rsSectionSettings = 'Einstellungen';
  rsSectionObjectList = 'Objekt-Liste';
  rsSectionObjectChecked = 'Objekt-Auswahl';
  rsAppBaseDir = 'Arbeitsverzeichnis';
  rsMaxThreads = 'MaxThreads';
  { -Messages- }
  rsAppBaseDirCreate = 'Das Verzeichnis ' + cCrLf2 + '%s' + cCrLf2 + 'ist nicht vorhanden. Soll es neu angelegt werden?';
  rsInvalidPath = 'Unzulässige Pfadangabe! Das zu sichernde Verzeichnis' + cCrLf2 + '%s ' + cCrLf2 +
    'darf nicht im Arbeitsverzeichnis' + cCrLf2 + '%s ' + cCrLf2 +
    'enthalten sein! Die erzeugtren ZIP-Dateien würden sich zur Laufzeit selber ' + 'sichern, und das geht nicht!';
  rsNoDirectory = 'Das Verzeichnis %s ist nicht vorhanden!';

var
  ThreadCount: Integer = -1;
  ThreadID: Integer = 0;

  { ------------------------------------------------------------------------------ }
  { - Formular-Events ------------------------------------------------------------ }
  { ------------------------------------------------------------------------------ }

procedure TFrmMyDelphiBackup.FormShow(Sender: TObject);
begin
  FCritSection := TCriticalSection.Create;
  FTempStrings := TStringList.Create;
  FMaxThreads := cMaxThreadsDefault;
  if IniLesen then
  begin
    if Pos(':', FApplicationBaseDir) = 2 then
    begin
      if not DirectoryExists(FApplicationBaseDir) then
      begin
        if MessageBox(Handle, PChar(Format(rsAppBaseDirCreate, [FApplicationBaseDir])), PChar(rsAppName),
          MB_YESNO + MB_ICONQUESTION) = IDYES then
          ForceDirectories(FApplicationBaseDir)
        else
          Close;
      end
      else
      begin
        StringListToListView(FTempStrings);
        FTempStrings.Clear;
        StatusMessage('Programmstart mit maximal ' + FMaxThreads.ToString + ' Threads');
      end;
    end
    else
      Close;
  end;
end;

procedure TFrmMyDelphiBackup.FormDestroy(Sender: TObject);
begin
  FCritSection.Free;
  FTempStrings.Free;
  IniSchreiben;
end;

{ ------------------------------------------------------------------------------ }
{ - Component-Events ----------------------------------------------------------- }
{ ------------------------------------------------------------------------------ }

procedure TFrmMyDelphiBackup.BtnStartClick(Sender: TObject);
var
  Ix, TimeStamp: Integer;

  function GetZipFileName(aPraefix: string): string;
  begin
    Result := aPraefix + '-' + IntToStr(TimeStamp) + '.zip';
  end;

begin
  BtnStart.Enabled := false;
  ThreadCount := 0;
  TimeStamp := DateTimeToFileDate(Now);
  for Ix := 0 to Pred(lvZipList.Items.Count) do
  begin
    if lvZipList.Items[Ix].Checked then
    begin
      lvZipList.ItemIndex := Ix;
      FileListZippen(Ix, GetZipFileName(lvZipList.Items[Ix].SubItems[0]), FApplicationBaseDir, lvZipList.Items[Ix].SubItems[1]);
    end;
  end;
end;

procedure TFrmMyDelphiBackup.ZipForge1ProcessFileFailure(Sender: TObject; FileName: string; Operation: TZFProcessOperation;
  NativeError, ErrorCode: Integer; ErrorMessage: string; var Action: TZFAction);
var
  x: Integer;
begin
  Action := fxaAbort;
  FCritSection.Enter;
  TThread.Queue(nil,
    procedure
    begin
      x := (Sender as TZipForge).Tag;
      StatusMessage('Thread Nr. ' + x.ToString + 'ZIP-Fehler bei ' + FileName);
      StatusMessage('ErrorMessage: ' + ErrorMessage);
    end);
  FCritSection.Leave;
end;

procedure TFrmMyDelphiBackup.ZipperOverallProgress(Sender: TObject; Progress: Double; Operation: TZFProcessOperation;
ProgressPhase: TZFProgressPhase; var Cancel: Boolean);
var
  x: Integer;
begin
  case ProgressPhase of
    ppStart:
      begin
        FCritSection.Enter;
        TThread.Synchronize(nil,
          procedure
          begin
            x := (Sender as TZipForge).Tag;
            StatusMessage('Thread Nr. ' + x.ToString + ' ZIP-Vorgang gestartet');
          end);
        FCritSection.Leave;
      end;
    ppProcess:
      begin
        if Progress > 99.0 then
          Progress := 100;
        FCritSection.Enter;
        TThread.Queue(nil,
          procedure
          begin
            x := (Sender as TZipForge).Tag;
            lvZipList.Items[x].SubItems[2] := IntToStr(Trunc(Progress)) + '%';
          end);
        FCritSection.Leave;
      end;
    ppEnd:
      begin
        FCritSection.Enter;
        TThread.Synchronize(nil,
          procedure
          begin
            Dec(ThreadCount);
            x := (Sender as TZipForge).Tag;
            StatusMessage('Thread Nr. ' + x.ToString + ' beendet');
          end);
        FCritSection.Leave;
      end;
  end;
end;

{ ------------------------------------------------------------------------------ }
{ - Private-Deklarationen ------------------------------------------------------ }
{ ------------------------------------------------------------------------------ }

function TFrmMyDelphiBackup.IniFileName: string;
begin
  Result := TPath.Combine(TPath.GetDocumentsPath, rsIniName);
end;

function TFrmMyDelphiBackup.IniLesen: Boolean;
var
  i: Integer;
  Value, Dir: string;
begin
  Result := false;
  if not FileExists(IniFileName) then
    with TIniFile.Create(IniFileName) do
      try // Standardwerte eintragen
        WriteString(rsSectionSettings, rsAppBaseDir, TPath.GetDocumentsPath);
        WriteString(rsSectionObjectList, 'Ini-File', IniFileName);
      finally
        Free;
      end;
  with TIniFile.Create(IniFileName) do
    try
      FMaxThreads := ReadInteger(rsSectionSettings, rsMaxThreads, cMaxThreadsDefault);
      FApplicationBaseDir := ReadString(rsSectionSettings, rsAppBaseDir, TPath.GetDocumentsPath);
      ReadSection(rsSectionObjectList, FTempStrings);
      for i := Pred(FTempStrings.Count) downto 0 do
      begin
        Value := ReadString(rsSectionObjectList, FTempStrings[i], '');
        Dir := ExtractFileDir(Value);
        if StartsText(Dir, FApplicationBaseDir) then
        begin
          ShowMessage(Format(rsInvalidPath, [Dir, FApplicationBaseDir]));
          FTempStrings.Delete(i)
        end
        else
        begin
          if DirectoryExists(Dir) then
            FTempStrings[i] := FTempStrings[i] + rsTrennChar + Value
          else
          begin
            ShowMessage(Format(rsNoDirectory, [Dir]));
            FTempStrings.Delete(i);
          end;
        end;
      end;
    finally
      Free;
    end;
  Result := FTempStrings.Count > 0;
end;

procedure TFrmMyDelphiBackup.IniSchreiben;
var
  i: Integer;
begin
  with TIniFile.Create(IniFileName) do
    try
      for i := 0 to Pred(lvZipList.Items.Count) do
        WriteBool(rsSectionObjectChecked, lvZipList.Items[i].SubItems[0], lvZipList.Items[i].Checked);
    finally
      Free;
    end;
end;

procedure TFrmMyDelphiBackup.LogSchreiben;
begin
  if FTempStrings.Count > 0 then
    FTempStrings.SaveToFile(ChangeFileExt(IniFileName, '.log'));
end;

{ ------------------------------------------------------------------------------ }
{ - Public-Deklarationen ------------------------------------------------------- }
{ ------------------------------------------------------------------------------ }

procedure TFrmMyDelphiBackup.StatusMessage(aValue: string);
begin
  StatusBar.Panels[0].Text := TimeToStr(Now);
  StatusBar.Panels[1].Text := aValue;
  FTempStrings.Add(StatusBar.Panels[0].Text + rsTrennChar + StatusBar.Panels[1].Text);
end;

procedure TFrmMyDelphiBackup.StringListToListView(const aItems: TStrings);
var
  Item: TListItem;
  i, p: Integer;
  Ini: TIniFile;
begin
  Ini := TIniFile.Create(IniFileName);
  try
    for i := 0 to Pred(aItems.Count) do
    begin
      p := Pos(rsTrennChar, aItems[i]);
      Item := lvZipList.Items.Add;
      Item.Caption := IntToStr(i + 1);
      Item.SubItems.Add(copy(aItems[i], 1, p - 1));
      p := p + Length(rsTrennChar);
      Item.SubItems.Add(copy(aItems[i], p, Length(aItems[i])));
      Item.SubItems.Add('-');
      Item.Checked := Ini.ReadBool(rsSectionObjectChecked, Item.SubItems[0], false);
    end;
  finally
    Ini.Free;
  end;
end;

procedure TFrmMyDelphiBackup.FileListZippen(aID: Integer; const aZipFileName: string; const aZipperBaseDir: string;
const aZipFileMask: string);
begin
  { -warten, falls MaxThreads überschritten- }
  if ThreadCount > FMaxThreads then
  begin
    StatusMessage('Thread-ID(' + aID.ToString + ') warte auf nächsten Thread...');
    while ThreadCount > FMaxThreads do
    begin
      Sleep(10);
      Application.ProcessMessages;
    end;
  end;
  { -Task starten- }
  StatusMessage('Thread-ID(' + aID.ToString + ') [' + aZipFileMask + '] gestartet');
  TTask.Run(
    procedure
    var
      LZipper: TZipForge;
    begin
      Inc(ThreadCount);
      LZipper := TZipForge.Create(Application);
      try
        LZipper.Tag := aID;
        LZipper.OnOverallProgress := ZipperOverallProgress;
        LZipper.Zip64Mode := zmAuto;
        LZipper.BaseDir := aZipperBaseDir;
        LZipper.FileName := TPath.Combine(aZipperBaseDir, aZipFileName);
        LZipper.OpenArchive(fmCreate);
        try
          LZipper.AddFiles(aZipFileMask);
        except
          on E: Exception do
          begin
            FCritSection.Enter;
            TThread.Synchronize(nil,
              procedure
              begin
                Dec(ThreadCount);
                StatusMessage('Thread Nr. ' + aID.ToString + ' Exception: ' + E.Message);
              end);
            FCritSection.Leave;
          end;
        end;
        LZipper.CloseArchive;
      finally
        LZipper.Free;
      end;
    end);
end;

end.

An der verwendeten Zipper-Komponente von ZipForge kann es normalerweise nicht liegen, denn die ist threadsave.

Fragen:
Kann oder sollte man überhaupt über

markieren

Code:

TTask.Run()

mehrere Tasks laufen lassen?
Oder hab ich irgendwo einen Design- oder Denk-Fehler?

Ich kann auch gerne die Unit incl. DFM posten.

Vorab:
Du startest keine Threads, sondern erzeugst Tasks (Aufgaben), die von einem ThreadPool dann in einem Thread ausgeführt werden. Wieviele deiner Tasks dann parallel laufen, darauf hast du nur begrenzt Einfluss (wenn überhaupt).

Zur Performance:
Nimm mal dieses seltsame Log-Geschreibsel heraus und teste dann nochmal

markieren

Delphi-Quellcode:

procedure TFrmMyDelphiBackup.LogSchreiben;
begin
// if FTempStrings.Count > 0 then
// FTempStrings.SaveToFile(ChangeFileExt(IniFileName, '.log'));
end;

Nachtrag:
Deine ganzen CtriticalSections sehen aus, als ob du einfach mal pauschal da rein gemacht hast, weil besser ist besser ... so wie ich das sehe, sind die aber unsinning und stören schlimmstenfalls sogar. Auf jeden Fall bremsen die dich auch noch aus.

Nachtrag 2:
Ob ZipForge threadsafe ist oder nicht spielt hier absolut keine Geige, denn du verwendest die erzeugte Instanz immer nur von einem Thread-Kontext aus ... wo soll es hier also zu Problemen beim Zugriff von unterschiedlichen Thread-Kontexten kommen - die ja eben nicht erfolgen - und nur dann muss ich mir Gedanken um threadsafe machen.

Zitat von Sir Rufo:

Vorab: Du startest keine Threads, sondern erzeugst Tasks (Aufgaben), die von einem ThreadPool dann in einem Thread ausgeführt werden. Wieviele deiner Tasks dann parallel laufen, darauf hast du nur begrenzt Einfluss (wenn überhaupt).

Oooops...Danke! Wieder was dazu gelernt. Aber, mehr wie 4 Tasks können es ja nicht werden. Ist das schon zu viel?

Zitat von Sir Rufo:

Zur Performance:
Nimm mal dieses seltsame Log-Geschreibsel heraus und teste dann nochmal

Schau mal den Code an. Das ist nirgendwo aktiviert, nur deklariert. Tut also aktuell gar nichts. Ist bzw. war beim FormDestroy als Debug-Info drin. Kann also auch wenn es wieder aktiviert sein sollte, hinsichtlich der Performance nichts beeinflussen.

markieren

Delphi-Quellcode:

procedure TFrmMyDelphiBackup.FormDestroy(Sender: TObject);
begin
  FCritSection.Free;
  FTempStrings.Free;
  IniSchreiben;
  LogSchreiben; // <-----
end;

Zur CriticalSection: Ich hatte die CriticalSection vorher nicht drin, und dann sporadisch Zugriffsverletzungen. Wie schließt man aus, dass mehrere Tasks zur gleichen Zeit im Formular schreiben?

Zitat von mm1256:

Zur CriticalSection: Ich hatte die CriticalSection vorher nicht drin, und dann sporadisch Zugriffsverletzungen. Wie schließt man aus, dass mehrere Tasks zur gleichen Zeit im Formular schreiben?

Indem man die Zugriffe synchronisiert mit TThread.Synchronize oder TThread.Queue

Wirf doch einfach mal einen Blick auf den BackgroundWorker. In den Quellen kannst du sehen, wie so etwas gemacht wird.

Eigentlich wäre der BW genau das, was du benötigst ...

Nichts anderes mache ich doch. Also ist die CriticalSection definitiv nicht erforderlich?

Zitat von SirRufo:

Eigentlich wäre der BW genau das, was du benötigst ...

War mir schon klar, dass das noch kommt Ich möchte aber in diesem Fall nicht irgendwelche fertigen Lösungen, sondern eine Anwendung für die eigene Praxis, um Erfahrungen zu gewinnen, wie ich System.Threading effektiv für meine Anwendungen nutzen kann.

Zitat von mm1256:

Ich will ganz einfach mal wissen/testen, was geht, und was nicht

Gerade gesehen: TZipForge.Create(Application);

Ganz üble Geschichte ... denk daran, du befindest dich da nicht im MainThread. Statt Application ganz simpel nil nehmen.

Dann wäre da noch Inc(ThreadCount); aufgerufen aus einem Thread-Kontext, das kann auch lustige Effekte erzielen und du rufst Dec(ThreadCount); nur dann auf, wenn in dem Task eine Exception geworfen wurde ... klingt irgendwie seltsam.

Für ein sicheres Increment/Decrement verwende ganz simpel System.SyncObjs.TInterlocked.Increment bzw. System.SyncObjs.TInterlocked.Decrement. Dafür sind die da

Hallo Sir Rufo,

erst mal vielen Dank für deine Bemühungen.

Die Hilfe bzw. Beschreibung zu

markieren

Code:

System.SyncObjs.TInterlocked.Increment / Decrement

im Wiki ist ja " wahnsinnig ausführlich". Was muss man denn tun, um von selbst auf sowas zu kommen

Zitat:

und du rufst Dec(ThreadCount); nur dann auf, wenn in dem Task eine Exception geworfen wurde ... klingt irgendwie seltsam.

Nicht "seltsam"....das hat schon seine Richtigkeit. Du hast lediglich in ZipperOverallProgress bei der ProgressPhase ppEnd (also wenn der Zipper fertig ist) was überlesen. Ist aber gut versteckt, ich geb's zu.

Weil ich in der Zwischenzeit den Unterschied zwischen TTask und TThread begriffen habe, stelle ich zurzeit auf die klassische Methode mit einem TThread um. Darin feuere ich dann "OnTerminate" und zähle darin ThreadCount runter. So wie es sich (so hoffe ich) eigentlich gehört

Das wird nur dann doof, wenn auf einmal eine Exception auftaucht nachdem du schon heruntergezählt hast.

Darum macht man das immer so

markieren

Delphi-Quellcode:

Inc(foo);
try
  // whatever
finally
  Dec(foo);
end;

Schon kann das gar nicht anders gehen.

Hier mal auf die Schnelle etwas zum Ansschauen

zusammenfalten · markieren

Delphi-Quellcode:

unit dp_186773.Forms.MainForm;

interface

uses
  Threading.ProcQueue,

  Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants, System.Classes, Vcl.Graphics,
  Vcl.Controls, Vcl.Forms, Vcl.Dialogs, Vcl.StdCtrls;

type
  TMainForm = class( TForm )
    Button1: TButton;
    ListBox1: TListBox;
    procedure Button1Click( Sender: TObject );
  private
    FProcQueue: TProcQueue;
    procedure Log( const AMsg: string );
    function CreateProc( const AID: string ): TProc;
  public
    procedure AfterConstruction; override;
    procedure BeforeDestruction; override;
  end;

var
  MainForm: TMainForm;

implementation

{$R *.dfm}

procedure TMainForm.AfterConstruction;
begin
  inherited;
  FProcQueue := TProcQueue.Create( 4 );
end;

procedure TMainForm.BeforeDestruction;
begin
  FProcQueue.Free;
  inherited;
end;

procedure TMainForm.Button1Click( Sender: TObject );
var
  I: Integer;
begin
  for I := 1 to 10 do
    begin
      FProcQueue.Add( CreateProc( TGUID.NewGuid.ToString ) );
    end;
end;

function TMainForm.CreateProc( const AID: string ): TProc;
begin
  Result :=
    procedure
    begin
      TThread.Synchronize( nil,
        procedure
        begin
          Log( AID + ' started' );
        end );
      try
        Sleep( 1000 );
      finally
        TThread.Synchronize( nil,
          procedure
          begin
            Log( AID + ' finished' );
          end );
      end;
    end;
end;

procedure TMainForm.Log( const AMsg: string );
begin
  ListBox1.ItemIndex := ListBox1.Items.Add( AMsg );
end;

end.

zusammenfalten · markieren

Delphi-Quellcode:

unit Threading.ProcQueue;

interface

uses
  System.Generics.Collections,
  System.SysUtils,
  System.Threading;

type
  TProcQueue = class
  private
    FShutdown : Boolean;
    FMaxParallel: Integer;
    FSync : TObject;
    FProcQueue : TQueue<TProc>;
    FTaskList : TList<ITask>;
    procedure Execute( const AProc: TProc );
    procedure TaskHasFinished( const ATask: ITask );
  public
    constructor Create( const MaxParallel: Integer );
    destructor Destroy; override;

    procedure Add( const AProc: TProc );
  end;

implementation

{ TProcQueue }

procedure TProcQueue.Add( const AProc: TProc );
begin
  if FShutdown
  then
    raise EInvalidOpException.Create( 'we are going down' );

  TMonitor.Enter( FSync );
  try
    if FTaskList.Count < FMaxParallel
    then
      Execute( AProc )
    else
      FProcQueue.Enqueue( AProc );
  finally
    TMonitor.Exit( FSync );
  end;
end;

constructor TProcQueue.Create( const MaxParallel: Integer );
begin
  inherited Create;
  FMaxParallel := MaxParallel;
  FSync := TObject.Create;
  FProcQueue := TQueue<TProc>.Create;
  FTaskList := TList<ITask>.Create;
end;

destructor TProcQueue.Destroy;
var
  task: ITask;
begin
  TMonitor.Enter( FSync );
  try
    FShutdown := True;
    FProcQueue.Clear;
  finally
    TMonitor.Exit( FSync );
  end;
  try
    TTask.WaitForAll( FTaskList.ToArray );
  except
    // we do not care about exceptions
  end;
  FTaskList.Free;
  FProcQueue.Free;
  inherited;
  FSync.Free;
end;

procedure TProcQueue.Execute( const AProc: TProc );
var
  task: ITask;
begin
  task := TTask.Create(
    procedure
    begin
      try
        AProc( );
      finally
        TaskHasFinished( task );
      end;
    end );
  FTaskList.Add( task );
  task.Start;
end;

procedure TProcQueue.TaskHasFinished( const ATask: ITask );
begin
  TMonitor.Enter( FSync );
  try
    FTaskList.Remove( ATask );
    if not FShutdown and ( FProcQueue.Count > 0 )
    then
      Execute( FProcQueue.Dequeue( ) );
  finally
    TMonitor.Exit( FSync );
  end;
end;

end.

PS: ITask.Cancel sollte man sich verkneifen