Primož Gabrijelčič

For a timer, call Task.SetTimer and timer event will be called automatically. 

Something like this:

uses
  OtlTaskControl;

type
  TWorker = class(TOmniWorker)
  protected
    function  Initialize: boolean; override;
  public
    procedure TimerCallback;
  end;

function TWorker.Initialize: boolean;
begin
  Result := inherited Initialize;
  if Result then
    Task.SetTimer(1, 100, TimerCallback);
end;

procedure TWorker.TimerCallback;
begin
  // called every 100 ms
end;

var
  task: IOmniTaskControl;

task := CreateTask(TWorker.Create()).Run;

Please provide a minimal program that demonstrates the problem.

4 hours ago, Joseph MItzen said:

When did they hire a Java developer to name things?

Long, long time ago 🙂

A quick google search found out that it was introduced in Delphi 4: http://www.blong.com/Conferences/DCon99/VCLSourcery/VCLSourcery.htm.

Kudos to @blong who keeps 20 year old posts alive on the web!

On 9/2/2020 at 10:07 PM, Anders Melander said:

I just checked the source: On Windows it's just a wrapper around the Windows SRW lock.

https://docs.microsoft.com/en-us/windows/win32/sync/slim-reader-writer--srw--locks

And on other systems it's just a wrapper around pthread_rwlock.

This is actually a COM problem, I believe, and I know almost nothing about COM, so please consider that my thinking may be entirely wrong.

I believe that you are having problems because you initialize your server in tmApartment mode but then access it from multiple threads. http://docwiki.embarcadero.com/RADStudio/Sydney/en/Choosing_a_Threading_Model says for tmApartment: " All client calls use the thread in which the object was created." And the object is created in the main thread.

Does your code work if you use TThread instead of OTL to run async tasks?

I'm very sorry but I really don't speak any Frech. Can you please translate your question?

For Synchronize mechanism to work, console application should be calling function CheckSynchronize from System.Classes repeatedly. You can do that from the message processing loop.

If Synchronize is not working (can't tell why without your code), use Queue with a record containing (data, result, flag) and wait for main thread to set the flag. Like this:

var
  res, data: ...
  flag: boolean;

flag := false;

TThread.Queue(nil,
  procedure
  begin
    res := Process(data);
    flag := true;
  end);

while not flag do Sleep(0);

Or make 'flag' an event and wait on it. 

Indeed. IOmniBlockingCollection implements an enumerator which waits for the next available value. The only way to terminate such for loop is to call input.CompleteAdding which signals the enumerator that no new values can ever be produced.

SimplePipeline is, well, simple. This is the TButton event handler:

procedure TfrmSimplePipeline.btnPipelineClick(Sender: TObject);
var
  pipeline: IOmniPipeline;
  i: Integer;
  v: TOmniValue;
begin
  pipeline := Parallel.Pipeline;
  pipeline.Stage(AddOne);
  pipeline.Stage(Invert);
  pipeline.Run;

  for i := 1 to 5 do
    pipeline.Input.Add(i);
  pipeline.Input.CompleteAdding;

  for v in pipeline.Output do
    lbLog.Items.Add(Format('%.3f', [v.AsDouble]));

  pipeline := nil;
end;

As you can see, the code starts the pipeline and then runs the loop that processes the results. This loop is run in a main thread and therefore blocks the UI. If you slow down the pipeline, it needs more time to generate the results and UI is blocked.

UniqueFilter does not go out of scope. If defined in this format (with input, output: IOmniBlockingCollection parameters), pipeline only calls it once.

The 'for url in input' loop is the one that processes all elements that arrive via the input pipeline.

No, WaitFor is not the right choice.

Just use TThread.Queue and execute the code in the main thread. Something like:

res := CalculateTheResult();
TThread.Queue(nil,
  procedure
  begin
    // This procedure will execute in the main thread.
    // The address of the 'res' variable will be automatically captured so you can use its value here.
    ProcessInMainThread(res);
  end);

Can I get a minimal reproducible example project, please?

Sorry, impossible to guess 😞 Would need a test app that shows that behaviour to say more.

DSiWaitForTwoObjects is just a shortcut which calls WaitForMultipleObjects Windows API with two handles. In this case it will return with status WAIT_TIMEOUT when a timeout elapses, WAIT_OBJECT_1 when task.Comm.NewMessageEvent is signalled or WAIT_OBJECT_0 when task.TerminateEvent is signalled.

Parallel.For by default works in "blocking" mode. IOW, while the Parallel.For is being executed, owner thread (main thread in your case) does nothing else. Most importantly - it does not process messsages.

There are two ways around this problem.

1.  Use the .NoWait qualifier: FFor := Parallel.For(...).TaskConfig(...).NumTasks(...).NoWait.Execute(...). (http://www.omnithreadlibrary.com/book/chap06.html#highlevel-for, section 3.11.1) Read here why you have to store the result of the Parallel.For call into a field: http://www.omnithreadlibrary.com/book/chap06.html#leanpub-auto-a-life-cycle-of-an-abstraction.
2. Execute the Parallel.For asynchronously via Parallel.Async mechanism (http://www.omnithreadlibrary.com/book/chap06.html#highlevel-async):

   Parallel.Async(
     procedure begin
       Parallel.For....Execute();
     end);

    

Your way of killing children is fine. It should not cause the behaviour you are observing.

If the OnTerminated is not called, then the task owner (MainTask) is not processing messages. If MainTask is an OTL task, then you can just call it's function MsgWait when you create it.

MainTask := CreateTask(TOmniWorkerDescendant.Create()).MsgWait;

4 minutes ago, Sonjli said:

any idea for the OnTerminated event in the child task?

Missed that one, sorry.

If a OnTerminated is not called (but is called when the task does not die), then the task is probably killed with TerminateThread somewhere.

If you run your program in debugger, it should pop up when the exception occurs - unless you have that exception type listed on the 'ignored exception' list.

Alternatively, open OtlTaskControl unit and search for TOmniTask.InternalExecute. Inside you'll find:

        try
          if otSharedInfo_ref.ProcessorGroup >= 0 then
            otExecutor_ref.SetProcessorGroup(otSharedInfo_ref.ProcessorGroup);
          if otSharedInfo_ref.NUMANode >= 0 then
            otExecutor_ref.SetNUMANode(otSharedInfo_ref.NUMANode);
          otExecutor_ref.Asy_Execute(Self);
        except
          on E: Exception do begin
            taskException := Exception(AcquireExceptionObject);
            FilterException(taskException);
            if assigned(taskException) then
              SetException(taskException);
          end;
        end;

This is the main exception handler for a task. You can log information inside the exception handler or put a breakpoint there.

I see no obvious reasons why this could cause problems. IOW, I think it should work, yes.

Please go fight in private and stop killing this great thread.

Your pipeline is a local variable in `TForm1.PipelineExecute`. When this method exits, `ppl` goes out of scope and is destroyed. 

Put the pipeline in a form field.

Did some testing on that recently and most of the time wait functions work as expected while on some virtual machines they return early. The test cases were running on few Windows boxes and on two VMWare Fusion/Mac VMs. Worked OK on all but one. One Fusion/Mac combo consistently returned early from wait.

OmniThreadLibrary 3.07.8 has just been released. It contains few small fixes + support for Delphi 10.4 Sydney.

Changelog:

• New features:
  • Implemented IOmniTask.RegisterWaitObject with an anonymous method callback.
  • [Jacek Laskowski] donated a new OTL icon (res\OTL.ico).
  • Added Delphi 10.4 Syndey packages.
• Bug fixes:
  • [sglienke] A reference to an anonymous method executor in IOmniTask is cleaned up as soon as possible. This allows OTL tasks to be executed from a package. [issue #132]
  • TOmniMREW.TryEnterReadLock and .TryEnterWriteLock were returning True on timeout.
  • SetOnMessage(nil) works correctly.
  • Fixed invalid FreeAndNil of an interface in TOmniFuture<T>.Execute.
  • Compiles with Delphi 10.4 Sydney.

https://github.com/gabr42/OmniThreadLibrary/releases/tag/release-3.07.8