Najlepszy sposób synchronizowania dwóch wątków ze sobą w Delphi

Question

Obecnie próbuję znaleźć najlepszy (*) sposób, aby dwa wątki działały naprzemiennie i sprawiały, że czekałyby na siebie nawzajem.

(*) najlepsze połączenie jest szybkie, mając jednocześnie tani CPU

Znalazłem trzy sposoby dotychczas która ułożyła w niektórych aplikacji demo, aby pokazać problemy znalazłem.

Używanie TMonitora zgodnie z klasycznym wzorcem oczekiwania/pulsu działa niezbyt dobrze ze względu na wszystkie blokady (zgodnie z SamplingProfiler jest spalany przez większość czasu w tych funkcjach). Próbowałem tego samego przy użyciu zdarzeń systemu Windows (SyncObjs.TEvent), ale wykonano podobne (tj. Złe).

Używanie pętli oczekiwania, która wywołuje TThread.Yield, działa najlepiej, ale oczywiście spala cykle procesora jak szalone. To nie ma znaczenia, jeśli przełączanie odbywa się bardzo szybko, ale boli, gdy wątek rzeczywiście czeka (widać to w wersji demo).

Korzystanie z TSpinWait działa świetnie (jeśli nie jest najlepszym z nich), ale tylko wtedy, gdy przełączniki są wykonywane bardzo szybko. Im dłużej trwa przełączanie, tym gorsza wydajność wynika z działania TSpinWait.

Ponieważ wielowątkowość nie jest jedną z moich mocnych stron, zastanawiałem się, czy istnieje jakaś kombinacja tych sposobów lub zupełnie inne podejście do osiągnięcia dobrej wydajności w obu scenariuszach (szybkie i wolne przełączniki).

program PingPongThreads; 

{$APPTYPE CONSOLE} 

{$R *.res} 

uses 
    Classes, 
    Diagnostics, 
    SyncObjs, 
    SysUtils; 

type 
    TPingPongThread = class(TThread) 
    private 
    fCount: Integer; 
    protected 
    procedure Execute; override; 
    procedure Pong; virtual; 
    public 
    procedure Ping; virtual; 
    property Count: Integer read fCount; 
    end; 

    TPingPongThreadClass = class of TPingPongThread; 

    TMonitorThread = class(TPingPongThread) 
    protected 
    procedure Pong; override; 
    procedure TerminatedSet; override; 
    public 
    procedure Ping; override; 
    end; 

    TYieldThread = class(TPingPongThread) 
    private 
    fState: Integer; 
    protected 
    procedure Pong; override; 
    public 
    procedure Ping; override; 
    end; 

    TSpinWaitThread = class(TPingPongThread) 
    private 
    fState: Integer; 
    protected 
    procedure Pong; override; 
    public 
    procedure Ping; override; 
    end; 

{ TPingPongThread } 

procedure TPingPongThread.Execute; 
begin 
    while not Terminated do 
    Pong; 
end; 

procedure TPingPongThread.Ping; 
begin 
    TInterlocked.Increment(fCount); 
end; 

procedure TPingPongThread.Pong; 
begin 
    TInterlocked.Increment(fCount); 
end; 

{ TMonitorThread } 

procedure TMonitorThread.Ping; 
begin 
    inherited; 
    TMonitor.Enter(Self); 
    try 
    if Suspended then 
     Start 
    else 
     TMonitor.Pulse(Self); 
    TMonitor.Wait(Self, INFINITE); 
    finally 
    TMonitor.Exit(Self); 
    end; 
end; 

procedure TMonitorThread.Pong; 
begin 
    inherited; 
    TMonitor.Enter(Self); 
    try 
    TMonitor.Pulse(Self); 
    if not Terminated then 
     TMonitor.Wait(Self, INFINITE); 
    finally 
    TMonitor.Exit(Self); 
    end; 
end; 

procedure TMonitorThread.TerminatedSet; 
begin 
    TMonitor.Enter(Self); 
    try 
    TMonitor.Pulse(Self); 
    finally 
    TMonitor.Exit(Self); 
    end; 
end; 

{ TYieldThread } 

procedure TYieldThread.Ping; 
begin 
    inherited; 
    if Suspended then 
    Start 
    else 
    fState := 3; 
    while TInterlocked.CompareExchange(fState, 2, 1) <> 1 do 
    TThread.Yield; 
end; 

procedure TYieldThread.Pong; 
begin 
    inherited; 
    fState := 1; 
    while TInterlocked.CompareExchange(fState, 0, 3) <> 3 do 
    if Terminated then 
     Abort 
    else 
     TThread.Yield; 
end; 

{ TSpinWaitThread } 

procedure TSpinWaitThread.Ping; 
var 
    w: TSpinWait; 
begin 
    inherited; 
    if Suspended then 
    Start 
    else 
    fState := 3; 
    w.Reset; 
    while TInterlocked.CompareExchange(fState, 2, 1) <> 1 do 
    w.SpinCycle; 
end; 

procedure TSpinWaitThread.Pong; 
var 
    w: TSpinWait; 
begin 
    inherited; 
    fState := 1; 
    w.Reset; 
    while TInterlocked.CompareExchange(fState, 0, 3) <> 3 do 
    if Terminated then 
     Abort 
    else 
     w.SpinCycle; 
end; 

procedure TestPingPongThread(threadClass: TPingPongThreadClass; quickSwitch: Boolean); 
const 
    MAXCOUNT = 10000; 
var 
    t: TPingPongThread; 
    i: Integer; 
    sw: TStopwatch; 
    w: TSpinWait; 
begin 
    t := threadClass.Create(True); 
    try 
    for i := 1 to MAXCOUNT do 
    begin 
     t.Ping; 

     if not quickSwitch then 
     begin 
     // simulate some work 
     w.Reset; 
     while w.Count < 20 do 
      w.SpinCycle; 
     end; 

     if i = 1 then 
     begin 
     if not quickSwitch then 
     begin 
      Writeln('Check CPU usage. Press <Enter> to continue'); 
      Readln; 
     end; 
     sw := TStopwatch.StartNew; 
     end; 
    end; 
    Writeln(threadClass.ClassName, ' quick switches: ', quickSwitch); 
    Writeln('Duration: ', sw.ElapsedMilliseconds, ' ms'); 
    Writeln('Call count: ', t.Count); 
    Writeln; 
    finally 
    t.Free; 
    end; 
end; 

procedure Main; 
begin 
    TestPingPongThread(TMonitorThread, False); 
    TestPingPongThread(TYieldThread, False); 
    TestPingPongThread(TSpinWaitThread, False); 

    TestPingPongThread(TMonitorThread, True); 
    TestPingPongThread(TYieldThread, True); 
    TestPingPongThread(TSpinWaitThread, True); 
end; 

begin 
    try 
    Main; 
    except 
    on E: Exception do 
     Writeln(E.ClassName, ': ', E.Message); 
    end; 
    Writeln('Press <Enter> to exit'); 
    Readln; 
end. 

Aktualizacja:

wymyśliłem kombinacji zdarzenia i spinwait:

constructor TSpinEvent.Create; 
begin 
    inherited Create(nil, False, False, ''); 
end; 

procedure TSpinEvent.SetEvent; 
begin 
    fState := 1; 
    inherited; 
end; 

procedure TSpinEvent.WaitFor; 
var 
    startCount: Cardinal; 
begin 
    startCount := TThread.GetTickCount; 
    while TInterlocked.CompareExchange(fState, 0, 1) <> 1 do 
    begin 
    if (TThread.GetTickCount - startCount) >= YieldTimeout then // YieldTimeout = 10 
     inherited WaitFor(INFINITE) 
    else 
     TThread.Yield; 
    end; 
end; 

Wykonuje tylko w przybliżeniu od 5 do 6 razy wolniej niż realizacji włókna oparte kiedy robi szybkie przełączanie i mniej niż 1% wolniej, gdy dodajesz trochę pracy między wywołaniami Ping. To oczywiście działa na 2 rdzenie zamiast tylko jednego przy użyciu światłowodu.

Answer 1

Kiedy znajduję się w takich sytuacjach, lubię używać zdarzeń systemu Windows. Są one wyświetlane w Delphi przy użyciu klasy TEvent, którą można uzyskać od WaitForSingleObject.

Można więc użyć dwóch zdarzeń: Thread1NotActive i Thread2NotActive. Po zakończeniu wątku1 ustawia flagę Thread1NotActive, na którą czeka wątek2. I odwrotnie, jeśli wątek2 przestaje przetwarzać, ustawia Thread2NotActive, który jest monitorowany przez wątek1.

Powinno to pozwolić na uniknięcie warunków wyścigu (dlatego sugeruję użycie dwóch zdarzeń zamiast 1) i powinno być przy zdrowych zmysłach, nie pochłaniając nadmiernych ilości czasu procesora.

Jeśli potrzebujesz bardziej kompletnego przykładu, będziesz musiał poczekać jutro :)