Discussion :

[Andnotor]

Salut à tous !

Dernièrement je voulais paralléliser un traitement et en toute logique ai utilisé TParallel.For. Mais à l'utilisation le temps d'exécution me paraissait toujours très long (plus rapide évidemment mais toujours trop long à mon goût) ce qui m'étonnait.

J'ai donc codé une petit app de test et le résultat m'a plutôt surpris. Au démarrage j'ai bien le nombre de tâches égale au nombre de CPUs x2 (ce qui est prévu dans TParallel.For), soit 32 sur mon PC mais ça chute à 19 après 50% de traitement, puis à 3 (voire 2 comme ci-dessous !) à 90% avec pour conséquence que ces 10 derniers pourcents prennent 40 à 50% du temps total

Un test sur une VM 4 cœurs, et bien notre multi-tâches devient à terme du mono-tâche

Le résultat en image :
En rouge le nombre de tâches restantes (sur 300) et en bleu le nombre de tâches parallélisées. La liste à droite, pour info, est le nombre de tâches exécutées par cœur.



En augmentant le Stride à 10 (à quoi ça sert exactement, l'aide est plutôt laconique à ce sujet, voir ci-dessous) j'obtiens quelque chose de bien meilleur, 30 tâches du début à la fin. Chouette ! je reporte donc ça dans mon app et patatras ! les performances sont encore pires qu'avant !

Donc oui ce Stride peut augmenter les performances, encore faudrait-il savoir comment le définir.

Le paramètre AStride vous permet de régler la performance. La boucle "parallel for" utilise un thread-pool pour planifier le travail. Si les paquets de travail sont très petits, la surcharge de synchronisation dans le thread-pool peut diminuer la performance. Si le paramètre AStride est supérieur à 1, alors les indices AStride sont regroupés dans une seule partie du travail. Cela réduit le temps dédié à la surcharge de synchronisation au détriment de la réduction du parallélisme dans la boucle.

Dans mon test le "travail" est un Sleep(1000) alors que dans mon app le traitement est plus rapide que ça, j'observe l'inverse de ce que dit l'aide.

Est-ce que quelqu'un aurait une idée de comment calculer cela ? Ou faut-il y aller par tâtonnement (sic) ?

Merci pour toute info. Si vous n'en avez pas, je reviendrai à une gestion multithreads à l'ancienne


ps: j'ai quant même de la peine à comprendre le concept ; TParallel.For est bloquant, le but est qu'il se termine le plus vite possible. Pourquoi limiter le nombre de cœurs utilisés ? Ça fait un peu penser à ces barres de progression qui affichent très vite 99% mais restent coincées dessus pendant des plombes...

[Andnotor]

Je continue mes tests en ce dimanche mausade, principalement en comparant les différentes techniques à disposition : TParallel, N threads et le ThreadPool fournit par l'OS.

Le premier test est toujours en incluant un "travail" lent à base de Sleep (comme un accès disque, réseau, etc.).

• En rouge TParallel avec un Stride à 1.
• En vert un démarrage manuel de 32 threads.
• En bleu le ThreadPool Windows. Aucune indication particulière n'est donnée au pool, on le laisse faire.
• En violet le même ThreadPool mais en indiquant clairement cette fois que la tâche est lente.

Le trait épais est le nombre de tâches restantes et le fin le nombre de tâches parallèles.

• TParallel sert surtout de comparaison et comme on l'a déjà vu, augmenter son Stride améliorerait ses performances.
• 32 threads, sans surprise tourne à fond (est sous le trait violet).
• le ThreadPool sans indication n'est pas très performant ici, l'OS crée de nouveaux threads au fur et à mesure de l'avancement jusqu'à atteindre un nombre optimal. La différence s'amoindrit si le nombre de boucles augmente (passer de 300 à 30 000).
• Par contre si on lui indique que ce sont des tâches lentes (ça se définit par tâche), on est aussi rapide qu'avec 32 threads (les courbes se confondent).

Le deuxième test est un "travail" intensif à base de SpinWait (du calcul par exemple).

• TParallel est moins pénalisé ici.
• 32 threads, toujours sans surprise.
• le ThreadPool par contre est cette fois très performant quelque soit le réglage de lenteur. Ce qu'on remarque surtout est qu'il fait le travail dans le même temps que les 32 threads mais en en utilisant deux fois moins. La parallélisation/sérialisation est excellente avec pour effet un PC beaucoup moins stressé, l'option 32 threads perdant clairement du temps en commutation.

En conclusion, TParallel (et autres ITask) à la poubelle au profit du pool géré par l'OS ! L'avertir qu'une tâche est lente n'est pas un problème, on sait ce qu'on développe
Il propose aussi certaines fonctionnalités intéressantes comme le signalement automatique d'un event (mutex, sémaphore) en fin de routine, ce qui nous évite try finally SetEvent(Event) end pour un code plus léger.

Je vous mets la source. Il faut minimum un D10.2 pour les déclarations (sinon à vous de jouer) et D10.3 pour les variables en lignes (là aussi je vous laisse commuter le cas échéant).

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
unit ThreadPools;
 
interface
 
uses Winapi.Windows, System.SysUtils;
 
type
  TThreadPool = class
  public type
    TState = class
    private
      FThreadPool :TThreadPool;
      FAborted    :THandle;
      class function DoAbort(aThreadPool :TThreadPool) :cardinal; stdcall; static;
    public
      function    Aborted :boolean;
      procedure   Abort;
      constructor Create(aThreadPool :TThreadPool);
      destructor  Destroy; override;
    end;
 
    TCallbackProc = TProc<PTPCallbackInstance, pointer>;
    TParallelProc = TProc<integer, TState>;
 
    PCallbackData = ^TCallbackData;
    TCallbackData = record
      Proc     :TCallbackProc;
      Context  :pointer;
    end;
 
  private
    CBEnv        :TTPCallbackEnviron;
    Pool         :PTPPool;
    CleanupGroup :PTPCleanupGroup;
    FMaxThreads  :integer;
    procedure    SetMaxThreads(const Value: integer);
    procedure    CleanUp(aCancelPending :boolean);
 
    class procedure WorkCallback(aInstance :PTPCallbackInstance; aData :PCallbackData; aWork :PTPWork); stdcall; static;
  public
    function     Add(aProc :TCallbackProc; aContext :pointer = nil) :boolean;
    property     MaxThreads :integer read FMaxThreads write SetMaxThreads;
    constructor  Create(aMaxThreads :integer = -1);
    destructor   Destroy; override;
 
    class function Parallel(aMaxThreads, aLow, aHigh :integer; aProc :TParallelProc; aMayRunLong :boolean = FALSE) :boolean; overload;
    class function Parallel(aLow, aHigh :integer; aProc :TParallelProc; aMayRunLong :boolean = FALSE) :boolean; overload;
  end;
 
implementation
 
{ TThreadPool.TState }
 
class function TThreadPool.TState.DoAbort(aThreadPool: TThreadPool): cardinal;
begin
  aThreadPool.CleanUp(TRUE);
  Result := ERROR_SUCCESS;
end;
 
procedure TThreadPool.TState.Abort;
var
  ThreadID :cardinal;
begin
  if not Aborted then
  begin
    SetEvent(FAborted);
    CloseHandle(CreateThread(nil, 0, @DoAbort, FThreadPool, 0, ThreadID));
  end;
end;
 
function TThreadPool.TState.Aborted: boolean;
begin
  Result := WaitForSingleObject(FAborted, 0) = WAIT_OBJECT_0;
end;
 
constructor TThreadPool.TState.Create(aThreadPool :TThreadPool);
begin
  FThreadPool := aThreadPool;
  FAborted    := CreateEvent(nil, TRUE, FALSE, '');
end;
 
destructor TThreadPool.TState.Destroy;
begin
  CloseHandle(FAborted);
  inherited;
end;
 
{ TThreadPool }
 
function TThreadPool.Add(aProc :TCallbackProc; aContext :pointer): boolean;
var
  Data :PCallbackData;
begin
  Data := AllocMem(SizeOf(TCallbackData));
  Data.Proc := aProc;
  Data.Context := aContext;
 
  const Work = CreateThreadpoolWork(@WorkCallback, Data, CBEnv);
  Result := Assigned(Work);
 
  if Result
  then SubmitThreadpoolWork(Work)
  else FreeMem(Data);
end;
 
procedure TThreadPool.CleanUp(aCancelPending: boolean);
begin
  CloseThreadpoolCleanupGroupMembers(CleanupGroup, aCancelPending, nil);
end;
 
constructor TThreadPool.Create(aMaxThreads :integer);
begin
  InitializeThreadpoolEnvironment(CBEnv);
  Pool := CreateThreadpool(nil);
  CleanupGroup := CreateThreadpoolCleanupGroup;
 
  SetThreadpoolCallbackPool(CBEnv, Pool);
  CBEnv.CleanupGroup := CleanupGroup;
 
  SetThreadpoolThreadMinimum(Pool, 1);
  SetMaxThreads(aMaxThreads);
end;
 
destructor TThreadPool.Destroy;
begin
  CleanUp(TRUE);
  CloseThreadpoolCleanupGroup(CleanupGroup);
  CloseThreadpool(Pool);
 
  inherited;
end;
 
class function TThreadPool.Parallel(aMaxThreads, aLow, aHigh: integer; aProc: TParallelProc; aMayRunLong :boolean): boolean;
begin
  if aHigh < aLow then Exit(TRUE);
 
  const ThreadPool = TThreadPool.Create(aMaxThreads);
 
  try
    const State = TState.Create(ThreadPool);
 
    try
      const DoParallel = procedure(aIndex: integer)
                         begin
                           ThreadPool.Add(procedure(aInstance: PTPCallbackInstance; aContext :pointer)
                                          begin
                                            if not State.Aborted then
                                            begin
                                              if aMayRunLong then
                                                CallbackMayRunLong(aInstance);
 
                                              aProc(aIndex, State);
                                            end;
                                          end);
                         end;
 
      for var i := aLow to aHigh do
        DoParallel(i);
 
      ThreadPool.CleanUp(FALSE);
 
    finally
      Result := not State.Aborted;
      State.Free;
    end;
 
  finally
    ThreadPool.Free;
  end;
end;
 
class function TThreadPool.Parallel(aLow, aHigh: integer; aProc: TParallelProc; aMayRunLong :boolean): boolean;
begin
  Result := Parallel(0, aLow, aHigh, aProc, aMayRunLong);
end;
 
procedure TThreadPool.SetMaxThreads(const Value: integer);
begin
  if Value > 0
  then FMaxThreads := Value
  else FMaxThreads := CPUCount *2;
 
  SetThreadpoolThreadMaximum(Pool, FMaxThreads);
end;
 
class procedure TThreadPool.WorkCallback(aInstance: PTPCallbackInstance; aData :PCallbackData; aWork: PTPWork);
begin
  try
    aData.Proc(aInstance, aData.Context);
  finally
    FreeMem(aData);
  end;
end;
 
end.