pipeline and visual feedback

[Attila Kovacs 631]

What is the proper way to display the progress from a pipeline stage when the main thread is blocked by WaitFor()? VCL.

[Primož Gabrijelčič 223]

There is none. Only the main thread should update the UI in a VCL application so if it is blocked, you're out of options.

 

In theory, you could create a window purely by the Windows API and use it to show the progress, but that is probably much to much work. 

 

Better solution would be to not block the VCL application at all.

[Attila Kovacs 631]

Thanks, does this mean pipeline is the wrong approach here or can I run the pipeline without Waitfor? Could not find anything in docs.

 

Edit: ok, I think I just have to omit WaitFor. I'll give it a try.

Edited April 2, 2020 by Attila Kovacs

[Primož Gabrijelčič 223]

Of course you can run a pipeline without a WaitFor.

 

You have different options to detect when a pipeline has finished its work.

 

a) The main program can count the number of items sent to the pipeline and number of items returned from. (If there is a simple correspondence between two - for example if each input produces exactly one output.)

b) The pipeline itself can detect that it has no more work and then it can signal this to the main program.

 

When you detect a terminating condition, you can shut down the pipeline (with WaitFor) and you'll done.

 

See the "folder scanner" in OTL examples folder for an example of the b) technique or read this chapter of the book: http://www.omnithreadlibrary.com/book/chap10.html#howto-webDownload

[Attila Kovacs 631]

Thank you Primož. Awesome lib. 

[Attila Kovacs 631]

Sadly I'm stuck already, if I omit WaitFor(), nothing happens.

Occasionally one task will be executed if I'm hitting the start button like a maniac, but I have 4 values assigned to the input.

 

I'm playing with the example with retriever/inserter.

 

  pipeline := Parallel.pipeline //
    .Stage(Retriever) //
    .NumTasks(Environment.Process.Affinity.Count * 2) //
    .Stage(Inserter, Parallel.TaskConfig.OnMessage(Self)) //
    .Run //
    ;

  pipeline.input.Add(TThreadParam.Create(....));
  pipeline.input.Add(TThreadParam.Create(....));
  pipeline.input.Add(TThreadParam.Create(....));
  pipeline.input.Add(TThreadParam.Create(....));
  pipeline.input.CompleteAdding;
  // pipeline.WaitFor(INFINITE);

 

I've tried to put ".run" after completeadding as a 'trial and fail'attempt, and in this case on the first run 2 tasks are executed, and from the second run all tasks are executed.

 

I'm really missing the explanations  from the doc, there is also an example

"procedure TfrmOtlParallelExceptions.btnPipeline1Click(Sender: TObject);"

where is no waitfor() but processing output right after "CompleteAdding", which is also confusing.

 

// Provide input
37   with pipeline.Input do begin
38     // few normal elements
39     Add(1);
40     Add(2);
41     // then trigger the exception in the first stage;
42     // this exception should be 'corrected' in the second stage
43     Add('three');
44     Add(4);
45     CompleteAdding;
46   end;
47 
48   // Process output; there should be no exception in the output collection
49   for value in pipeline.Output do
50     Log(value.AsString);

 

 

[Primož Gabrijelčič 223]

I don't know. Show us the code. You are probably doing something after that "WaitFor" and I don't know what.

 

Quote

where is no waitfor() but processing output right after "CompleteAdding", which is also confusing.

 

What is confusing here? You don't have to wait for pipeline to terminate (WaitFor) to start processing its output.

[Attila Kovacs 631]

Ha-ha. ATM I entered the code here I just noticed that pipeline is a local variable, and also an interfaced obj. 🙈

[Primož Gabrijelčič 223]

Oh, a classic mistake 🙂 Been there, done that.

[Fr0sT.Brutal 900]

On 4/2/2020 at 1:52 PM, Primož Gabrijelčič said:

In theory, you could create a window purely by the Windows API and use it to show the progress, but that is probably much to much work. 

 

Not so much thanks to resources https://stackoverflow.com/questions/388506/displaying-splash-screen-in-delphi-when-main-thread-is-busy/45465592#45465592

[Attila Kovacs 631]

As I put more code into the pipeline stages some questions are arising.

For example, some of the exceptions are arriving as "string" into the final output collection instead as "exception".

One case is "Abort" and the other one is if I'm re-raising an exception in the stage code. Why is that?

 

Also, what is the proper way to re-run a task or add a new one after "CompleteAdding"?

I would like to return the task with the exception and be able to re-fire the same task. (Network communication)

Do I need a new pipeline?

 

Edited April 7, 2020 by Attila Kovacs

[Primož Gabrijelčič 223]

5 minutes ago, Attila Kovacs said:

For example, some of the exceptions are arriving as "string" into the final output collection instead as "exception".

Without having a reproducible example, I have no idea.

5 minutes ago, Attila Kovacs said:

Do I need a new pipeline?

Yes, you need a new pipeline. After a pipeline goes into CompleteAdding state, it cannot be "revived".

[Attila Kovacs 631]

On 4/7/2020 at 4:26 PM, Primož Gabrijelčič said:

Without having a reproducible example, I have no idea.

Never underestimate this kind of answers, this tells me always that there is a user error. It was. Thx.

 

I have here something what I'm not getting and very hard to debug.

(got it, not even multi threading problem, it just revealed it \o/)

 

btw, I still can't figure out what "Parallel.TaskConfig.OnMessage(Self)" does, as 2nd parameter to a stage.

 

thx

 

 

 

 

Edited April 9, 2020 by Attila Kovacs

[Attila Kovacs 631]

Ooookay, everything is working perfectly, it was a journey, transforming the legacy code to work with OTL, which I really enjoyed.

It's frightening how easy and how fast can you write complex and _working_ code with this lib.

The online documentation seems to be a bit behind the current release (didn't check the book yet), so I took a copy of the book at leanpub with a custom input in the box as a small donation and a big thank you.

[Primož Gabrijelčič 223]

Oh, that was you! Big thanks!

 

For the record - I'm trying to keep the book in sync with the OTL most of the time.