Передать данные в поток TThread

[ssnakess]

Делаю простой логер.
Хотел сделать его не зависящим от основного потока программы, а так же других потоков, которые порождаются в процессе выполнения.
Для этого в отдельном юните, сделал класс наследник от TTHread который должен получать строку с информацией для записи в лог-файл.

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

   TLogFile = class(TThread)
     private
       fMessage:String;
       fLogName:String;
       fLogDirectory:String;
       fMessages:TStringList;
       fFileLogName:String;
       fLogDate:TDateTime;
     protected
       function LogFileName:String;
       function SaveToLog:String;
       procedure Execute; override;
       procedure TerminateProcess(Sender: TObject);
     public
       constructor create(AlogDirectory:String=''; ALogName:String='');
       procedure Add(const AMessage:String);

       property LogDirectory:String read fLogDirectory;
       property Messages:TStringlist read fMessages;
   end;
.......

constructor TLogFile.create(AlogDirectory: String; ALogName: String);
begin
  inherited Create(true);
  OnTerminate:=@TerminateProcess;
  fMessages:=TStringList.Create;
  fLogName:=ALogName;
  fLogDirectory:=ALogDirectory;
  fFileLogName:='';
  FreeOnTerminate := True;
  fLogDate:=Date;
  start;
end;

procedure TLogFile.Add(const AMessage: String);
begin
  fMessages.Add(AMessage);
end; 

procedure TLogFile.Execute;
Begin
  While (not Terminated) or (fMessages.Count>0) do
   Begin
     if fMessages.Count>0 then
      Begin
         fMessage:=fMessages.Strings[0];
         fMessage:=SaveToLog;
         if fMessage<>'' then fMessages.Append(fMessage)
         else
         fMessages.Delete(0);
      end
     else sleep(1000);
   end;
end;

function TLogFile.SaveToLog: String;
begin
... тут просто запись в конец файла строки fMessage
... если ошибка вернет не записанную строку, если все ОК, то вернет пустую строку
end;


При инициализации юнита создаю глобальную переменную с экземпляром этого класса
и при финализации - освобождаю.

Здесь, я использовал fMessages:TStringList как своего рода очередь записей в лог.
Т.е. из любого места программы, использую вызов
MainLog.Add('супер сообщение в лог');
сообщение добавляется в "очередь" и поток обрабатывает её, при появлении там чего-либо, пишет в файл.

Но при работе с таким вот логом из дополнительных потоков, естестсвенно возникает проблема, т.к. это не потокобезопасное добавление нового сообщения в очередь на запись в лог.

Гуглил передачу данных в поток, но не смог найти решения.
Подскажите, как сделать эту очередь записей в лог - потокобезопасной.
syncronize или критические секции - не совсем то что надо, ибо оно будет тормозить основной поток, от чего собственно и пытаюсь уйти при помощи этого логера

[Сквозняк]

Подскажите, как сделать эту очередь записей в лог - потокобезопасной.

Не делать общую очередь записей, а каждому потоку иметь свою собственную. Потом обмениваться с записывающим в лог потоком однобайтовыми ключами: сюда можно писать, отсюда можно читать, запись обработана, можно удалять.

[ssnakess]

Подскажите, как сделать эту очередь записей в лог - потокобезопасной.

Не делать общую очередь записей, а каждому потоку иметь свою собственную. Потом обмениваться с записывающим в лог потоком однобайтовыми ключами: сюда можно писать, отсюда можно читать, запись обработана, можно удалять.

Т.е. утрирую, вы предлагаете в каждом потоке иметь свой TStringList для ведения лога, и потом его выдавать объекту TLogFile?
и в чем тогда разница?

тут как ни крути но где-то должен быть флажок - можно передать данные в TLogFile или жди, т.е. всеравно очередь хоть ожидающих потоков для передачи в TLogFile или очередь в самом TLogFile

[Sharfik]

Не делать общую очередь записей, а каждому потоку иметь свою собственную. Потом обмениваться с записывающим в лог потоком однобайтовыми ключами: сюда можно писать, отсюда можно читать, запись обработана, можно удалять.

И получить логи разбитые по времени. В чем тогда их смысл?

syncronize или критические секции - не совсем то что надо, ибо оно будет тормозить основной поток, от чего собственно и пытаюсь уйти при помощи этого логера

Такого волшебства не бывает. Если нужна общая корзинка, значит все по очереди организованно к ней подходят.
Другое дело, что можно разделить. Сделать на базе TThreadList класс который будет принимать у каждого потока информацию, а отдельный поток когда ивходная корзина будет не занята, или по счетчику. или по времени, будет ее очищать предварительно оформив лог как надо и куда но. Фейс контроль - Очередь - Запись.
Зависания, как мои эксперименты сейчас показывают, по большей части из-за зацикленности потока, который вечно делает синхронизацию ненужную. Местами можно воткнуть Sleep() по какому то условию, чтобы иногда давать программе поделать что то более полезное, чем занятие наскальной живописью.

[Сквозняк]

Т.е. утрирую, вы предлагаете в каждом потоке иметь свой TStringList для ведения лога, и потом его выдавать объекту TLogFile?
и в чем тогда разница?

тут как ни крути но где-то должен быть флажок - можно передать данные в TLogFile или жди, т.е. всеравно очередь хоть ожидающих потоков для передачи в TLogFile или очередь в самом TLogFile

Можно так, а можно просто писать в ячейки расшаренного между потоками массива содержащего поля для записи и ключи/флаги. Идея в том, когда нужно записать лог, то поток ищет ячейку массива, в которую разрешена запись, после чего меняет значения ключа. Читающий поток ищет ячейки с нужным значением ключа (а их можно написать и больше чем 2, сколько для твоей логики понадобится, но это намного меньше выносит мозг, чем если для последовательных операций заведёшь несколько флагов с двумя значениями и будешь обрабатывать их комбинации), читает поле для чтения и пишет в ключ.

И получить логи разбитые по времени. В чем тогда их смысл?

Такая система работает, имеет право на жизнь и сглаживает возможные тормоза некоторых потоков, которые операционка или железо могут подвесить, и тогда очерёдность сообщений тоже собьётся. А если сбои есть и там и там, то нужно разобраться в том что логгировать и какие к логгированию требования, исходя из этого и нужно выбирать, какой алгоритм лучше. Когда writeln от разных потоков пишут сообщения в консоль, там иногда некоторые записи теряются или портятся, но если их маркировать, то разобраться в этой каше обычно можно. Если необходимо, то в лог можно писать номер и имя потока, общее время.

[Alex2013]

syncronize или критические секции - не совсем то что надо, ибо оно будет тормозить основной поток, от чего собственно и пытаюсь уйти при помощи этого логера

Кроме syncronize есть Queue(); (Синхронизирует поток с основным потоком в неблокирующем режиме)
В общем пробуй!
Зы
Есть еще немного странный механизм ForceQueue();

protected procedure TThread.ForceQueue( aMethod: TThreadMethod);
class procedure TThread.ForceQueue( aThread: TThread; aMethod: TThreadMethod);

ForceQueue планирует метод aMethod для выполнения в основном потоке, точно так же, как TThread.Queue. В отличие от TThread.Queue, если вызов выполняется из основного потока, функция не будет выполнена сразу, но фактически будет помещена в очередь.

(Удивила возможность использовать без создания отдельного потока и возможность делать что-то из основного потока но ЗА его пределами )

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

procedure TfrmFormMain.btnThread_QueueClick(Sender: TObject);
begin
  //
  Memo1.Clear;
  //
  // forcing a "Queue" be used
  //
  // for test it, mark 2 break-point in this code:
  //
  TThread.ForceQueue(nil,
    procedure
    begin
      // command 0
      btnThread_Queue.Caption := 'btnThread_Queue ' + TimeToStr(now); { break point - later here! }
      //
      prcMyLog(format('%s - Queue - command 1', [TimeToStr(now)]));
      //
      prcMyLog(format('%s - Queue - command 2', [TimeToStr(now)]));
      //
      // Button1.Repaint;
    end);
  //
  prcMyLog(format('%s - Queue - command 3', [TimeToStr(now)]));
  //
  prcMyLog(format('%s - Queue - command 4', [TimeToStr(now)]));
  //
  prcMyLog(format('%s - Queue - command 5 = %s', [TimeToStr(now), StringOfChar('-', 40)]));
end; { break point - first occurrs here }


УПС ! ( оказывается Queue(nil,procedure.. ) тоже может работать из основного потока )
https://www.programmersought.com/article/52676419008/

То есть получается что во многих случаях создавать отдельный поток не нужно ? Хм!

[ssnakess]

Сделал так, уж не знаю, правильно или нет, но пока работает.

добавил к поле property IsBusy:Boolean read fIsBusy write fIsBusy;
и вынес fMessages в свойство для чтения

к классу TlogFile
и создал спец класс для добавления новых данных в очередь на запись в лог
при создании объекта он получает сообщения для логирования, а также объект класса TLogFile
этот поток, ждет когда IsBusy станет равным false
и если это счастливое событие наступило, то ставит занятость в true и добавляет в очередь Messages, свое сообщение.
после чего возвращает isBusy в false и самоубивается

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

  TAppendMessage = class (TThread)
     private
       fMEssage:String;
       fCategory:String;
       fLog:TLogFile;
     protected
       procedure Execute; override;
     public
       constructor create(LogFile:TLogFile; AMessage:String; ACategory:String='');
   end;


procedure TAppendMessage.Execute;
var success:Boolean;
    msg:String;
begin
  success:=false;
  if fCategory<>'' Then msg:='['+fCategory+'] '+fMessage
  else msg:=fMessage;
  while (not Terminated) and (not success) do
   Begin
     if not fLog.IsBusy Then
      Begin
       fLog.IsBusy:=true;
       fLog.Messages.Append(msg);
       fLog.IsBusy:=false;
       success:=true
      end;
   end;
  Terminate;
end;

constructor TAppendMessage.create(LogFile: TLogFile; AMessage: String;
  ACategory: String);
begin
  inherited create(true);
  FreeOnTerminate:=true;
  fLog:=LogFile;
  fMEssage:=AMEssage;
  fCategory:=ACategory;
  Start;
end;


в потоках программы и основном коде добавляю данные в лог таким образом:

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

....
var msg:TAppendMessage;
begin
   ....
   msg:=TAppendMessage.Create(MainLog,'сообщение в лог');
   ....


Я спецально вынес очередь сообщений Messages для доступа в public, чтобы код добавления новых данных в очередь, был вне потока объекта класса TLogFile.

Но, меня смущает вот это

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

     if not fLog.IsBusy Then
      Begin
       fLog.IsBusy:=true;
       fLog.Messages.Append(msg);
       fLog.IsBusy:=false;



т.е. если два параллельно работающих потока вдруг одномоментно решат записать чтото в лог, то возникнет коллизия, и какоето из сообщений потеряется.
гипотетически такое возможно, только если параллельных потоков много и они очень часто пишут в лог, т.е. ситуация может возникнуть, но ИМХО, редко.
Но всетаки может

Добавлено спустя 12 минут 20 секунд:

syncronize или критические секции - не совсем то что надо, ибо оно будет тормозить основной поток, от чего собственно и пытаюсь уйти при помощи этого логера

Кроме syncronize есть Queue(); (Синхронизирует поток с основным потоком в неблокирующем режиме)

Есть такая штука, но тут засада

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

...
procedure Queue(aThread: TThread; aMethod: TThreadMethod);
...


Т.е. надо передать ссылку на метод, в виде @MethodForQueue
но не вызвать MethodForQueue(Data); при вызове метода Queue

и мне кажется что этот метод Queue работает с PostMessage и SendMesage
т.е. тогда надо для потока TLogFile создавать хендл окна и делать обработку сообщений
а тогда можно и напрямую слать сообщения этому объекту используя системную очередь сообщений
типа получил сообщение WM_USER+1 и забрал данные из WParam
и так по кругу, пока терминате не наступит

[Alex2013]

Но всетаки может

Хм ! Еще раз ... может быть можно подстраховаться через очередь? (По идее основной поток это тоже поток и если фокус
TThread.ForceQueue (или Queue) (nil,procedure... ) работает в основном потоке то что мешает ему работать в отдельном потоке ? )
.

[ssnakess]

Хм ! Еще раз ... может быть можно подстраховаться через очередь? (По идее основной поток это тоже поток и если фокус
TThread.ForceQueue (или Queue) (nil,procedure... ) работает в основном потоке то что мешает ему работать в отдельном потоке ? )

Ктож против то
но в линуксовом лазарусе нет варианта вызова

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

TThread.Queue(Mainlog, procedure ()
     begin
        MainLog.Add('сообщение в лог');
     end
);


А есть только

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

TThread.Queue(Mainlog, @MethodForAppendMessage);


и еще, читал что эта очередь работает как обертка над PostMessage
т.е. поток MainLog должен иметь хендл окна и соответственно обработчик сообщений от системной очереди.

Добавлено спустя 1 минуту 5 секунд:
а TThread такого хендла не имеет

[Сквозняк]

этот поток, ждет когда IsBusy станет равным false
и если это счастливое событие наступило

А вот в этом месте надо поподробнее. Результат операции fLog.IsBusy:=true; дойдёт до потоков с задержкой. И потому, когда наступает =false, то нужно у спецуправленца, который один на всю программу, и сидит в самом быстром потоке, получить состояние "в натуре чисто конкретно false, ты начал запись, жду окончания" То есть двух состояний мало, нужно поключаться к записи в два этапа, а потом сообщить управленцу об окончании записи, а уже потом писать fLog.IsBusy:=false;

[jsa]

Для передачи строк из много поточного idHTTPServer в основной поток для записи лога, создал структуру "очередь".
Там всего две процедуры put и get
потоки сервера используют только put, процедура записи в log в основном потоке использует только get
Проверено, работает сейчас в нескольких местах.
Пока доходило одновременно до 12 потоков.

[ssnakess]

А вот в этом месте надо поподробнее. Результат операции fLog.IsBusy:=true; дойдёт до потоков с задержкой. И потому, когда наступает =false, то нужно у спецуправленца, который один на всю программу, и сидит в самом быстром потоке, получить состояние "в натуре чисто конкретно false, ты начал запись, жду окончания" То есть двух состояний мало, нужно поключаться к записи в два этапа, а потом сообщить управленцу об окончании записи, а уже потом писать fLog.IsBusy:=false;

Вот тут не уловил связь. Можно примером кода показать?
или както по другому пояснить.

Из этого я понял. что Вы предлагаете не только в пишущем потоке ставить IsBusy в false, но и уведомлять еще когото.
кого?
Я сделал поток у которого Tstringlist это своеобразный буфер. Оне берет из ютого буфера строку с индексом 0 и пишет в файл.
затем строку с индексом 0 удаляет.
И так пока не количество строк в этом буффере не станет равным 0.
т.е. он не зависит от isBusy от слова совсем. Он просто постоянно смотрит количество строк в TStringList и если есть то пишет и опустошает.

Далее я сделал еще один класс, наследника от TThread, который смотрит когда у пищущего потока будет isBusy равным false.
Увидел false - поставил в true, т.е тем самым говорит другим потокам которые пытаются добавить свою инфу в буффер пишущего потока, что кто-то добавляет инфу, а вы ждите
Когда дописал в буффер, то ставит признак isBusy у пишущещго потока в False, и убивает себя. Тем самым, следующий поток который ждет на запись в буффер, видит что писатель освободился и продолжает свой жизненый цикл.
т.е. как-то так

вот не совсем понял Вас. куда мне в этой схеме добавить уведомление и кого?

Добавлено спустя 6 минут 9 секунд:

Для передачи строк из много поточного idHTTPServer в основной поток для записи лога, создал структуру "очередь".
Там всего две процедуры put и get
потоки сервера используют только put, процедура записи в log в основном потоке использует только get
Проверено, работает сейчас в нескольких местах.
Пока доходило одновременно до 12 потоков.

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

[jsa]

покажите реализацию. если не жалко конечно.

модуль с очередью

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

unit QueueBlock;

{$mode objfpc}{$H+}

interface

uses
  Classes, SysUtils;

type PNode = ^Node;
     Node = record
            Block: string;
            next: PNode;
            end;

procedure Put_Block_In_Queue( blk: string ); //Постановка блока в очередь
function Get_Block_From_Queue(): String; //Выемка блока из очереди

var Qhead, Qtail: PNode;

implementation

uses main;

{Постановка блока в очередь}
procedure Put_Block_In_Queue(blk: string );
var NewNode: PNode;
begin
     New(NewNode);
     NewNode^.Block := blk;
     NewNode^.next := nil;
     if Qtail <> nil then Qtail^.next := NewNode;
     Qtail := NewNode;
     if Qhead = nil then Qhead := Qtail;
     sleep(1);
end;

{Выемка блока из очереди}
function Get_Block_From_Queue(): String;
var top: PNode;
begin
     Result := '$$$$$';
     if Qhead <> nil then
     begin
          top := Qhead;
          Result := top^.Block;
          Qhead := top^.next;
          if Qhead = nil then Qtail := nil;
          Dispose(top);
          sleep(1);
     end;
end;

end.


использование очереди

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

procedure TFormMain.IdHTTPServer1CommandGet(AContext: TIdContext; ARequestInfo: TIdHTTPRequestInfo; AResponseInfo: TIdHTTPResponseInfo);
var p, delay, cnt, i : integer;
      rRemoteIP, rCommand, rDocument, rParams, rContentType, rType  : string;
      ....

    //процедура внутри потока - помещение строки в структуру очередь для записи в Log файл
    procedure InsertIntoQ(str:string);
      begin
         CrSection.Enter;
         QueueBlock.Put_Block_In_Queue(str);
         CrSection.Leave;
      end;

begin // обработка входящего запроса
rDocument:= AnsiUpperCase(ARequestInfo.Document);
rRemoteIP:= ARequestInfo.RemoteIP;
rCommand := ARequestInfo.Command; 
InsertIntoQ('RemoteIP = '+rRemoteIP+' / rCommand = '+rCommand+' / rDocument = '+rDocument );

....

end;




срабатывание таймера лога

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

{срабатывание таймера лога}
procedure TFormMain.TimerLogTimer(Sender: TObject);
var str:widestring;
begin
     TimerLog.Enabled:=false;

     repeat
     str:=QueueBlock.Get_Block_From_Queue();
     writelog.WrLog( str );
     until str='$$$$$';

     TimerLog.Interval:=200; TimerLog.Enabled:=true;
end;


writelog.WrLog

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

{Запись в лог}
procedure WrLog( txt:string );
var dt:string;
begin
     Append(FLOG);
     Writeln(FLOG, txt);
     Close(FLOG);
end;

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

Теоретически такая одновременность возможна.
Но на практике пока не встречалась.
Для того и сделал работу с очередью. т.к. она очень малозатратна по времени, вызовы Put_Block_In_Queue выполняются быстро и за счет этого не должны пересекаться.

[ssnakess]

jsa » 18.07.2023 19:00:28

Теоретически такая одновременность возможна.
Но на практике пока не встречалась.
Для того и сделал работу с очередью. т.к. она очень малозатратна по времени, вызовы Put_Block_In_Queue выполняются быстро и за счет этого не должны пересекаться

судя по вот этому куску кода

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

   //процедура внутри потока - помещение строки в структуру очередь для записи в Log файл
    procedure InsertIntoQ(str:string);
      begin
         CrSection.Enter;
         QueueBlock.Put_Block_In_Queue(str);
         CrSection.Leave;
      end;


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

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

//процедура внутри потока - помещение строки в структуру очередь для записи в Log файл
    procedure InsertIntoQ(str:string);
      begin
         CrSection.Enter;
         writelog.WrLog( str );
         CrSection.Leave;
      end;


и у вас все будет работать, без таймера, очереди и прочего

[Сквозняк]

Из этого я понял. что Вы предлагаете не только в пишущем потоке ставить IsBusy в false, но и уведомлять еще когото.
кого?

Ну да. Если не надеетесь на потокобезопасность своего кода, то всегда можно завести внешнего бюрократа, которому будете оставлять заявки и получать или не получать разрешение на действие. Это займёт лишнее время, потому обращаться к нему нужно только когда велика вероятность его положительного ответа. Ну вот как-то так:

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

   
if not fLog.IsBusy Then if bjurokrat1(nomer_potoka) Then
      Begin
       fLog.IsBusy:=true;
       fLog.Messages.Append(msg);
       uvedomlenie_bjurokratu1(nomer_potoka);
       fLog.IsBusy:=false;

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

var
zajavki1: array [1..kolvo_potokov] of byte=(0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0);
zajavki2: array [1..kolvo_potokov] of byte=(0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0);


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

function bjurokrat1(nomer_potoka: longint);
var
q2: longint;
begin
function bjurokrat1:=false;
if fLog.IsBusy=true Then exit;
if zajavki1[nomer_potoka]<>0 Then exit;
zajavki1[nomer_potoka]:=1;
while zajavki2[nomer_potoka]=0 do begin end;
if zajavki2[nomer_potoka]=1 Then
  begin
  zajavki1[nomer_potoka]:=0;
  zajavki2[nomer_potoka]:=0;
  exit;
  end;
zajavki1[nomer_potoka]:=0;
function bjurokrat1:=true;
end;

procedure  uvedomlenie_bjurokratu1(nomer_potoka: longint);
begin
zajavki2[nomer_potoka]:=0;
zajavki1[nomer_potoka]:=0;
end;


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

procedure glavk1; //запускается часто и только из одного потока.
var
q2,w2: longint;
begin
w2:=0;
for q2:=1 to kolvo_potokov do if zajavki1[q2]=1 Then
  begin
  w2:=q2;
  break;
  end;
if w2=0 then exit;
for q2:=1 to kolvo_potokov do if zajavki2[q2]=2 Then
  begin
  if q2=w2 then exit;
  zajavki2[w2]:=1;
  exit;
end;
zajavki2[w2]:=2; //вот, оно, разрешение на действие, которого ждёт функция в каком-то потоке
end;


Я сделал поток у которого Tstringlist это своеобразный буфер. Оне берет из ютого буфера строку с индексом 0 и пишет в файл.
затем строку с индексом 0 удаляет.
И так пока не количество строк в этом буффере не станет равным 0.
т.е. он не зависит от isBusy от слова совсем. Он просто постоянно смотрит количество строк в TStringList и если есть то пишет и опустошает.

Эту часть кода на качество не проверял. Нужно ей улучшение или не нужно, это другой вопрос, выше уже ответили, что на халяву сложилось так, что не нужно. Но если понадобится не допустить возможную коллизию тормозя только один поток, то я показал как.