Откуда берутся лаги при 100% загрузке ЦП?

[CRobin]

Здравствуйте. Имеется код

Код: Выделить всё

iteration:
  signal := false;
  repeat until signal;
  do_work;
goto iteration;


Смысл в том, чтобы держать поток, который будет делать работу асинхронно с основным потоком, для этого он должен ждать сигнала от первого потока, считывая постоянно переменную signal. Задержка критична.
Проблема даже не в том, что такой код сжирает 100% ядра выданого треду, это еще терпимо. Проблема в том, что задержка между изменением значения signal = true и началом do_work, отнють не моментальна. Такое ощущение что ЦП захлебывается в пустом цикле и очень долго выходит из него. Есть ли способы побороть проблему, или возможно есть механизм в Lazarus по типу критических секций, который умеет замораживать и размораживать потоки не на уровне логики, а на уровне ОС?

[runewalsh]

А чего ты хотел от пустых циклов, так нельзя делать вообще никогда. RTLEventCreate / TSimpleEvent (uses syncobjs) / чтонибудьещё (pthread), вместо пустого цикла — Wait, в другом потоке вместе с изменением условия дёргаешь событие.

[MysticCoder]

я предполагаю, что после проверки условия repeat until signal; когда выяснилось, что signal = false, но еще не выполнилось присвоение signal := false другой поток может поставить этот signal в true, который в этом потоке сразу же в false выставится, пропустив тем самым сигнал впустую, теоретически таких пропусков подряд может быть неограниченно много)
пфф... этот repeat until signal; в одну строку сбил с толку) забудь все что я написал выше)

есть механизм в Lazarus по типу критических секций

да, есть, так и называются "критические секции", вроде как в syncobjs, TCriticalSection

Вообще интересно, захлебывание цпу пустым циклом смахивает на бред, да и тормозить там нечему, переменная поменялась - цикл прервался, если переменная не кешируется, а кешируется она вряд ли. по идее приоритет потока может быть низкий или система загружена сильно => долго выполняется, но это тоже вряд ли. сколько времени задержка? как определяешь? может это на самом деле лаги интерфейса?

[CRobin]

MysticCoder у приложения нет интерфейса, задержку меряю rdtsc. Я понимаю, что формально все выглядит как максимально реалтаймовый вариант, но практически поток, который грузит 100% ядра всегда лагает и показывает худшую среднюю по сравнению с тем, который отдает управление.

runewalsh нельзя ли пример кода с RTLEventCreate? Гугл морозится, информация очень скудная.

[runewalsh]

Чот много букв, ну да ладно. Работа с разными обёртками событий принципиально одинаковая: Set/Reset/Wait, поток выходит из ожидания, когда ты Set'аешь событие (также Wait не ждёт, если вызвана на ранее сетнутом, etc.). Т. о. в своём коде ты должен извне вместе с присваиванием signal дёрнуть событие, а вместо холостого цикла с проверкой — ждать этого события.

Код: Выделить всё

{$mode objfpc}
uses
  SysUtils, Classes;

type
  Queue = class
    constructor Create(producers: cardinal);
    destructor Destroy; override;
    procedure Put(const item: string);
    function Get(out item: string): boolean; // ждёт; либо возвращает false, если очередь пуста и не осталось производителей
    procedure KillProducer;

  private type
    PItem = ^ItemRec;
    ItemRec = record
      next: PItem;
      data: string;
    end;

  var
    mutex: TRTLCriticalSection;
    nonemptyOrFinished: PRTLEvent;
    first, last: PItem;
    activeProducers: cardinal;
    procedure Lock;
    procedure Unlock;
    procedure Clear;
  end;

  constructor Queue.Create(producers: cardinal);
  begin
    inherited Create;
    InitCriticalSection(mutex);
    nonemptyOrFinished := RTLEventCreate;
    first := nil; last := nil;
    activeProducers := producers;
  end;

  destructor Queue.Destroy;
  begin
    Clear;
    RTLEventDestroy(nonemptyOrFinished);
    DoneCriticalSection(mutex);
    inherited Destroy;
  end;

  procedure Queue.Put(const item: string);
  var
    it: PItem;
  begin
    new(it);
    it^.next := nil;
    it^.data := item;

    Lock;
    if Assigned(last) then
    begin
      last^.next := it;
      last := it;
    end else
    begin
      first := it;
      last := it;
      RTLEventSetEvent(nonemptyOrFinished); // очередь стала непустой, ждущие могут этим заинтересоваться
    end;
    Unlock;
  end;

  function Queue.Get(out item: string): boolean;
  label again;
  var
    finished: boolean;
    it: PItem;
  begin
    again: Lock;
    it := first;
    if not Assigned(it) then
    begin
      finished := activeProducers = 0;
      Unlock;
      if finished then exit(false);

      // ждём, пока в очереди не появятся элементы, или не убьются все производители
      RTLEventWaitFor(nonemptyOrFinished);

      // Внимание. Обычно у событий настраивается режим auto-reset или manual reset.
      // Для RTLEvent это не так, они всегда auto-reset, т. е. пропускают ровно один ожидающий поток и автоматически сбрасываются.
      // Если нужно поведение как у manual reset, придётся выставить назад вручную (перепроверив условие).
      // Если заменить RTLEvent на тот же TEvent с ManualReset, три строчки ниже можно убрать.
      Lock;
      if Assigned(first) or (activeProducers = 0) then RTLEventSetEvent(nonemptyOrFinished);
      Unlock;

      // <- т. к. код не под критической секцией (ждать, не выходя из критической секции - плохая идея), сюда мог вклиниться
      //    другой потребитель, так что нужно перепроверить
      goto again;
    end;

    item := it^.data;
    first := it^.next;
    if not Assigned(first) then
    begin
      last := nil;
      if activeProducers > 0 then
        // очередь пуста, но ещё могут прийти новые элементы, так что ждущие пусть ждут
        RTLEventResetEvent(nonemptyOrFinished);
    end;
    Unlock;
    dispose(it);
    result := true;
  end;

  procedure Queue.KillProducer;
  begin
    Lock;
    dec(activeProducers);
    if activeProducers = 0 then RTLEventSetEvent(nonemptyOrFinished); // все производители умерли, новых элементов не будет, ждущие могут заинтересоваться
    Unlock;
  end;

  procedure Queue.Lock;
  begin
    EnterCriticalSection(mutex);
  end;

  procedure Queue.Unlock;
  begin
    LeaveCriticalSection(mutex);
  end;

  procedure Queue.Clear;
  var
    cur, t: PItem;
  begin
    Lock;
    cur := first;
    while Assigned(cur) do
    begin
      t := cur;
      cur := cur^.next;
      dispose(t);
    end;
    first := nil; last := nil;
    Unlock;
  end;

type
  ProducerConsumerThread = class(TThread)
    constructor Create(newQ: Queue; newId: cardinal);
  protected
    q: Queue;
    id: cardinal;
  end;

  ProducerThread = class(ProducerConsumerThread)
  protected
    procedure Execute; override;
  end;

  ConsumerThread = class(ProducerConsumerThread)
  protected
    procedure Execute; override;
  end;

  constructor ProducerConsumerThread.Create(newQ: Queue; newId: cardinal);
  begin
    q := newQ;
    id := newId;
    inherited Create(false); // CreateSuspended
  end;

  procedure ProducerThread.Execute;
  const
    Items: array[0 .. 2] of string = ('Item1', 'Item2', 'Item3');
  var
    i: integer;
    item: string;
  begin
    writeln(Format('Producer%d started', [id]));
    for i := 0 to High(Items) do
    begin
      Sleep(200);
      item := Format('P%d-%s', [id, Items[i]]);
      writeln(Format('Producer%d created %s', [id, item]));
      q.Put(item);
    end;
    q.KillProducer;
    writeln(Format('Producer%d finished', [id]));
  end;

  procedure ConsumerThread.Execute;
  var
    item: string;
  begin
    writeln(Format('Consumer%d started', [id]));
    while q.Get(item) do
    begin
      Sleep(300);
      writeln(Format('Consumer%d got %s', [id, item]));
    end;
    writeln(Format('Consumer%d finished', [id]));
  end;

var
  q: Queue;
  producers: array[0 .. 1] of TThread;
  consumers: array[0 .. 2] of TThread;
  i: integer;

begin
  q := Queue.Create(length(producers));

  for i := 0 to High(producers) do producers[i] := ProducerThread.Create(q, 1 + i);
  for i := 0 to High(consumers) do consumers[i] := ConsumerThread.Create(q, 1 + i);

  for i := 0 to High(producers) do producers[i].WaitFor;
  for i := 0 to High(consumers) do consumers[i].WaitFor;

  for i := 0 to High(producers) do producers[i].Free;
  for i := 0 to High(consumers) do consumers[i].Free;

  q.Free;
end.

[скалогрыз]

Чот много букв, ну да ладно

Продакшен код публикуешь?!
Вот такой пример, чуть короче.

Что делает:
1) основной поток считывает с клавиатуры числа. Если считано число, то передаёт него в рабочий поток, который напечатает это число 11 раз.
2) рабочий поток, ждёт сигнала о начале работы. После того как сигнал получен, печатает число 11-раз и начинает ожидать следующий сигнал.
3) если следующей число набрано во время работы потока, то оно сразу же передаётся в поток (и печатается на экране).
4) если с клавиатуры введено не число, программа завершается.

Код: Выделить всё

program project1;

{$mode delphi}{$H+}

uses
  {$IFDEF UNIX}cthreads,{$endif} Classes, SysUtils;

type

  { TCountThread }

  TCountThread = class(TThread)
  protected
    procedure Execute; override;
  public
    ev : PRTLEvent;
    num: integer;
    constructor Create;
    destructor Destroy; override;
  end;

{ TCountThread }

procedure TCountThread.Execute;
var
  i : integer;
begin
  while not Terminated do begin
    RTLeventWaitFor(ev);
    if not Terminated then begin
      for i:=0 to 10 do begin
        write(num,'...');
        sleep(200);
        if Terminated then Break;
      end;
      writeln;
      RTLeventResetEvent(ev);
    end;
  end;
end;

constructor TCountThread.Create;
begin
  inherited Create(false);
  ev:=RTLEventCreate;
end;

destructor TCountThread.Destroy;
begin
  RTLeventdestroy(ev);
  inherited Destroy;
end;

var
  thr : TCountThread;
  s   : string;
  err : integer;
  k   : integer;
begin
  writeln('starting');
  thr := TCountThread.Create;
  repeat
    readln(s);
    Val(s, k, err);
    if err=0 then begin
      thr.num:=k;
      RTLeventSetEvent(thr.ev);
    end;
  until err <> 0;
  writeln('stopping');

  thr.Terminate;
  RTLeventSetEvent(thr.ev);
  thr.WaitFor;
  thr.Free;
end.


ЗЫ: со времён TP такое принято реализовывать в одном потоке с функцией keypressed и модулем crt. Помню меня уже тогда интересовал вопрос, а как сделать так, чтобы keypressed не писать в каждой строчке.

Добавлено спустя 3 часа 11 минут 32 секунды:

Гугл морозится, информация очень скудная

а вот в вики...

[CRobin]

runewalsh, скалогрыз премного благодарен за примеры, это как раз то, что мне нужно.

Добавлено спустя 10 часов 20 минут 28 секунд:
скалогрыз насколько безопасно использовать

Код: Выделить всё

RTLeventSetEvent(self.ev);

из разных потоков? Нужно ли помещать счетчик события в критическую секцию или же этот механизм уже предусмотрен?

[runewalsh]

Сами по себе Set/Reset/Wait атомарны, но ты наверняка захочешь вместе с выставлением события выставить что-то ещё, как в коротком примере thr.num, поэтому в общем случае блокировка понадобится.