如何知道 OmniThreadLibrary 中 Pipeline 阶段的状态？

Posted 2023-02-22

技术标签:

【中文标题】如何知道 OmniThreadLibrary 中 Pipeline 阶段的状态？【英文标题】：How to know the state of Pipeline stages in OmniThreadLibrary? 【发布时间】：2015-04-14 08:12:46 【问题描述】：

gabr's answer to another question 展示了使用 Parallel.Pipeline 进行数据处理的示例。目前我需要知道 Pipeline 何时启动以及其所有阶段何时完成。我阅读了其他 gabr 对此问题的回答 How to monitor Pipeline stages in OmniThreadLibrary?。我试着这样做（根据answer修改）：

unit Unit1;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, StdCtrls, superobject,
  OtlCommon, OtlCollections, OtlParallel, OtlComm, OtlTask, ExtCtrls;

const
  WM_STARTED = WM_USER;
  WM_ENDED = WM_USER + 1;

type
  TForm1 = class(TForm)
    btnStart: TButton;
    btnStop: TButton;
    lbLog: TListBox;
    procedure btnStartClick(Sender: TObject);
    procedure btnStopClick(Sender: TObject);
  private
    FCounterTotal: IOmniCounter;
    FCounterProcessed: IOmniCounter;
    FIsBusy: boolean;
    FPipeline: IOmniPipeline;
    procedure WMStarted(var msg: TOmniMessage); message WM_STARTED;
    procedure WMEnded(var msg: TOmniMessage); message WM_ENDED;
  strict protected
    procedure Async_Files(const input, output: IOmniBlockingCollection; const task: IOmniTask);
    procedure Async_Parse(const input: TOmniValue; var output: TOmniValue);
    procedure Async_JSON(const input, output: IOmniBlockingCollection; const task: IOmniTask);
  end;

var
  Form1: TForm1;

  procedure GetJSON_(const AData: PChar; var Output: WideString); stdcall; external 'my.dll';

implementation

uses IOUtils;

$R *.dfm

procedure TForm1.Async_Files(const input, output: IOmniBlockingCollection; const task: IOmniTask);
var
  i, cnt: integer;
  f: string;
begin
  while not input.IsCompleted do begin

    task.Comm.Send(WM_STARTED); // message is sent once every 1 min
    cnt := 0;

    for f in TDirectory.GetFiles(ExtractFilePath(Application.ExeName), '*.txt') do
    begin
      output.TryAdd(f);
      Inc(cnt);
      Sleep(1000); // simulate a work
    end;
    FCounterTotal.Value := cnt;

    // I need to continously check a specified folder for new files, with
    // a period of 1 minute (60 sec) for an unlimited period of time.
    i := 60;
    repeat
      Sleep(1000); // Check if we should stop every second (if Stop button is pushed)
      if input.IsCompleted then Break;
      dec(i);
    until i < 0;
  end;
end;

procedure TForm1.Async_Parse(const input: TOmniValue; var output: TOmniValue);
var
  sl: TStringList;
  ws: WideString;
begin
  sl := TStringList.Create;
  try
    sl.LoadFromFile(input.AsString);
    GetJSON_(PChar(sl.Text), ws); // output as ISuperObject ---  DLL procedure
    output := SO(ws);
//     TFile.Delete(input.AsString); // For testing purposes only - Continue without Deleting Processed File
  finally
    sl.Free;
  end;
end;

procedure TForm1.Async_JSON(const input, output: IOmniBlockingCollection; const task: IOmniTask);
var
  value: TOmniValue;
  JSON: ISuperObject;
  cnt: integer;
begin
  for value in input do begin
    JSON := value.AsInterface as ISuperObject;
    // do something with JSON

    cnt := FCounterProcessed.Increment;
    if FCounterTotal.Value = cnt then
      task.Comm.Send(WM_ENDED); // !!! message is not sent
  end;
end;

//
procedure TForm1.btnStartClick(Sender: TObject);
begin
  btnStart.Enabled := False;

  FCounterTotal := CreateCounter(-1);
  FCounterProcessed := CreateCounter(0);

  FPipeline := Parallel.Pipeline
    .Stage(Async_Files, Parallel.TaskConfig.OnMessage(Self))
    .Stage(Async_Parse)
    .Stage(Async_JSON, Parallel.TaskConfig.OnMessage(Self))
    .Run;
end;

procedure TForm1.btnStopClick(Sender: TObject);
begin
  if Assigned(FPipeline) then begin
    FPipeline.Input.CompleteAdding;
    FPipeline := nil;
  end;

  btnStart.Enabled := True;
end;

//
procedure TForm1.WMEnded(var msg: TOmniMessage);
begin
  FIsBusy := False;
  lbLog.ItemIndex := lbLog.Items.Add(Format('%s - Pipeline stage 3 ended', [DateTimeToStr(Now)]));
end;

procedure TForm1.WMStarted(var msg: TOmniMessage);
begin
  FIsBusy := True;
  lbLog.ItemIndex := lbLog.Items.Add(Format('%s - Pipeline stage 1 starting', [DateTimeToStr(Now)]));
end;

end.

task.Comm.Send(WM_STARTED) 一切正常，但 task.Comm.Send(WM_ENDED) 行永远不会执行。我怎么知道最后一个阶段何时完成？正确的方法是什么？

【问题讨论】：

好吧，即使没有文件也可以每秒触发WM_STARTED 好吗？如果队列完成，您只会触发WM_ENDED。但是在处理一个项目后它还没有完成。如果你愿意，你必须在循环内触发消息。 @SirRufo 不，WM_STARTED 仅在我单击“开始”按钮后触发一次。如果是这样，请向我们展示您的真实代码。此代码无法编译，在对 uses 稍作更改后，WM_STARTED 每秒都会被触发。 【参考方案1】：

你的方法（我最初提出的）有一个竞争条件，阻止它工作。（抱歉，这是我最初设计的一个缺陷。）

基本上，发生的事情是：

Async_Files 将最后一个文件发送到管道。 Async_Files 块（模拟一些工作负载）。 Async_JSON 接收并处理最后一个文件。 Async_Files 现在设置 FCounterTotal 计数器。

此时，Async_JSON 已经在等待下一个数据，该数据永远不会到来，并且不再检查 FCounterTotal 值。

另一种方法是将特殊的sentinel 值作为最后一项发送到管道中。

异常也可以用作哨兵。如果您在第一阶段引发异常，它将通过管道“流动”到您可以处理它的末端。无需对任何特定阶段进行特殊工作 - 默认情况下，阶段只会重新引发异常。

【讨论】：

我试过了，但它不起作用或我做错了什么。你能看看真正的源代码，至少猜出是哪一个造成了这个问题吗？ test_OTL_SO_with_counter.zip @LuFang “真正的源代码”是什么意思？为什么您要询问一段代码，然后评论另一段代码。你应该在问题中提供一个 MCVE，然后没有人会浪费时间处理假代码。 @gabr 我改变了大卫所说的问题（将我的伪“假”代码替换为真实代码）。如果您能查看这段代码并提出我做错了什么，我将不胜感激。 @gabr 我看到了，谢谢。请你看看我关于异常的回答并建议我做错了什么？我的问题是这段代码（task.Comm.Send(WM_STARTED) 和task.Comm.Send(WM_ENDED)）只执行一次。【参考方案2】：

感谢 gabr，他的建议使用特殊的 sentinel 值帮助我找到了解决问题的方法。此代码按预期工作：

unit Unit1;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, StdCtrls, superobject,
  OtlCommon, OtlCollections, OtlParallel, OtlComm, OtlTask, ExtCtrls;

const
  WM_STARTED = WM_USER;
  WM_ENDED = WM_USER + 1;

type
  TForm1 = class(TForm)
    btnStart: TButton;
    btnStop: TButton;
    lbLog: TListBox;
    procedure btnStartClick(Sender: TObject);
    procedure btnStopClick(Sender: TObject);
  private
    FIsBusy: boolean;
    FPipeline: IOmniPipeline;
    procedure WMStarted(var msg: TOmniMessage); message WM_STARTED;
    procedure WMEnded(var msg: TOmniMessage); message WM_ENDED;
  strict protected
    procedure Async_Files(const input, output: IOmniBlockingCollection; const task: IOmniTask);
    procedure Async_Parse(const input: TOmniValue; var output: TOmniValue);
    procedure Async_JSON(const input, output: IOmniBlockingCollection; const task: IOmniTask);
  end;

var
  Form1: TForm1;

  procedure GetJSON_(const AData: PChar; var Output: WideString); stdcall; external 'my.dll';

implementation

uses IOUtils;

$R *.dfm

procedure TForm1.Async_Files(const input, output: IOmniBlockingCollection; const task: IOmniTask);
var
  i: integer;
  f: string;
begin
  while not input.IsCompleted do begin

    task.Comm.Send(WM_STARTED); // message is sent once every 1 min

    for f in TDirectory.GetFiles(ExtractFilePath(Application.ExeName), '*.txt') do
    begin
      output.TryAdd(f);
      Sleep(1000); // simulate a work
    end;
    output.TryAdd(0); // to send a special 'sentinel' value

    // I need to continously check a specified folder for new files, with
    // a period of 1 minute (60 sec) for an unlimited period of time.
    i := 60;
    repeat
      Sleep(1000); // Check if we should stop every second (if Stop button is pushed)
      if input.IsCompleted then Break;
      dec(i);
    until i < 0;
  end;
end;

procedure TForm1.Async_Parse(const input: TOmniValue; var output: TOmniValue);
var
  sl: TStringList;
  ws: WideString;
begin
  if input.IsInteger and (input.AsInteger = 0) then begin
    output := 0; // if we got 'sentinel' value send it to the next stage
    Exit;
  end;

  sl := TStringList.Create;
  try
    sl.LoadFromFile(input.AsString);
    GetJSON_(PChar(sl.Text), ws); // output as ISuperObject ---  DLL procedure
    output := SO(ws);
//     TFile.Delete(input.AsString); // For testing purposes only - Continue without Deleting Processed File
  finally
    sl.Free;
  end;
end;

procedure TForm1.Async_JSON(const input, output: IOmniBlockingCollection; const task: IOmniTask);
var
  value: TOmniValue;
  JSON: ISuperObject;
begin
  for value in input do begin

    if value.IsInteger and (value.AsInteger = 0) then begin
      task.Comm.Send(WM_ENDED); // if we got 'sentinel' value
      Continue;
    end;

    JSON := value.AsInterface as ISuperObject;
    // do something with JSON
  end;
end;

//
procedure TForm1.btnStartClick(Sender: TObject);
begin
  btnStart.Enabled := False;

  FPipeline := Parallel.Pipeline
    .Stage(Async_Files, Parallel.TaskConfig.OnMessage(Self))
    .Stage(Async_Parse)
    .Stage(Async_JSON, Parallel.TaskConfig.OnMessage(Self))
    .Run;
end;

procedure TForm1.btnStopClick(Sender: TObject);
begin
  if Assigned(FPipeline) then begin
    FPipeline.Input.CompleteAdding;
    FPipeline := nil;
  end;

  btnStart.Enabled := True;
end;

//
procedure TForm1.WMEnded(var msg: TOmniMessage);
begin
  FIsBusy := False;
  lbLog.ItemIndex := lbLog.Items.Add(Format('%s - Pipeline stage 3 ended', [DateTimeToStr(Now)]));
end;

procedure TForm1.WMStarted(var msg: TOmniMessage);
begin
  FIsBusy := True;
  lbLog.ItemIndex := lbLog.Items.Add(Format('%s - Pipeline stage 1 starting', [DateTimeToStr(Now)]));
end;

end.

使用 Exception 作为哨兵的替代方法（尚未工作，但我可能做错了什么）：

unit Unit1;

interface

uses
  Windows, Messages, SysUtils, Variants, Classes, Graphics, Controls, Forms,
  Dialogs, StdCtrls, superobject,
  OtlCommon, OtlCollections, OtlParallel, OtlComm, OtlTask, ExtCtrls;

const
  WM_STARTED = WM_USER;
  WM_ENDED = WM_USER + 1;

type
  ESentinelException = class(Exception);

  TForm1 = class(TForm)
    btnStart: TButton;
    btnStop: TButton;
    lbLog: TListBox;
    procedure btnStartClick(Sender: TObject);
    procedure btnStopClick(Sender: TObject);
  private
    FIsBusy: boolean;
    FPipeline: IOmniPipeline;
    procedure WMStarted(var msg: TOmniMessage); message WM_STARTED;
    procedure WMEnded(var msg: TOmniMessage); message WM_ENDED;
  strict protected
    procedure Async_Files(const input, output: IOmniBlockingCollection; const task: IOmniTask);
    procedure Async_Parse(const input: TOmniValue; var output: TOmniValue);
    procedure Async_JSON(const input, output: IOmniBlockingCollection; const task: IOmniTask);
  end;

var
  Form1: TForm1;

  procedure GetJSON_(const AData: PChar; var Output: WideString); stdcall; external 'my.dll';

implementation

uses IOUtils;

$R *.dfm

procedure TForm1.Async_Files(const input, output: IOmniBlockingCollection; const task: IOmniTask);
var
  i: integer;
  f: string;
begin
  while not input.IsCompleted do begin

    task.Comm.Send(WM_STARTED); // message is sent once every 1 min

    for f in TDirectory.GetFiles(ExtractFilePath(Application.ExeName), '*.txt') do
    begin
      output.TryAdd(f);
      Sleep(1000); // simulate a work
    end;

    raise ESentinelException.Create('sentinel');

    // I need to continously check a specified folder for new files, with
    // a period of 1 minute (60 sec) for an unlimited period of time.
    i := 60;
    repeat
      Sleep(1000); // Check if we should stop every second (if Stop button is pushed)
      if input.IsCompleted then Break;
      dec(i);
    until i < 0;
  end;
end;

procedure TForm1.Async_Parse(const input: TOmniValue; var output: TOmniValue);
var
  sl: TStringList;
  ws: WideString;
begin
  sl := TStringList.Create;
  try
    sl.LoadFromFile(input.AsString);
    GetJSON_(PChar(sl.Text), ws); // output as ISuperObject ---  DLL procedure
    output := SO(ws);
//     TFile.Delete(input.AsString); // For testing purposes only - Continue without Deleting Processed File
  finally
    sl.Free;
  end;
end;

procedure TForm1.Async_JSON(const input, output: IOmniBlockingCollection; const task: IOmniTask);
var
  value: TOmniValue;
  JSON: ISuperObject;
begin
  for value in input do begin

    if value.IsException and (value.AsException is ESentinelException) then begin
      task.Comm.Send(WM_ENDED); // if we got 'sentinel' Exception
      value.AsException.Free;
    end
    else begin
      JSON := value.AsInterface as ISuperObject;
      // do something with JSON
    end;
  end;
end;

//
procedure TForm1.btnStartClick(Sender: TObject);
begin
  btnStart.Enabled := False;

  FPipeline := Parallel.Pipeline
    .Stage(Async_Files, Parallel.TaskConfig.OnMessage(Self))
    .Stage(Async_Parse)
    .Stage(Async_JSON, Parallel.TaskConfig.OnMessage(Self))
    .HandleExceptions
    .Run;
end;

procedure TForm1.btnStopClick(Sender: TObject);
begin
  if Assigned(FPipeline) then begin
    FPipeline.Input.CompleteAdding;
    FPipeline := nil;
  end;

  btnStart.Enabled := True;
end;

//
procedure TForm1.WMEnded(var msg: TOmniMessage);
begin
  FIsBusy := False;
  lbLog.ItemIndex := lbLog.Items.Add(Format('%s - Pipeline stage 3 ended', [DateTimeToStr(Now)]));
end;

procedure TForm1.WMStarted(var msg: TOmniMessage);
begin
  FIsBusy := True;
  lbLog.ItemIndex := lbLog.Items.Add(Format('%s - Pipeline stage 1 starting', [DateTimeToStr(Now)]));
end;

end.

【讨论】：

如果我删除了对 GetJSON_ 的调用，该调用存在于我没有的 DLL 中，您的基于异常的解决方案在这里工作得很好。 @gabr 这是一件很奇怪的事情......我已经注释掉了所有与 dll&json 相关的行，但问题仍然存在。 WM_STARTED 和 WM_ENDED 消息仅在按下开始按钮后发送一次。我很困惑。

以上是关于如何知道 OmniThreadLibrary 中 Pipeline 阶段的状态？的主要内容，如果未能解决你的问题，请参考以下文章

Delphi OmniThreadLibrary + OPC 客户端

如何确定主线程没有响应 Omni 线程库的原因？

如何知道设备中 imageView 的实际大小？

我是不是需要知道如何手动创建数据结构以获得入门级工作，或者我应该只知道如何使用收集框架中的数据结构？ [关闭]

如何知道视图层次结构中是不是显示 UIActionSheet？

不知道如何在 C# 中自动按键