Skip to content

do not block messages during processing loop #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hkollmann wants to merge 1 commit into
base: master
Choose a base branch
from
Open

do not block messages during processing loop #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 30 additions & 1 deletion OtlParallel.pas
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -507,6 +507,7 @@ interface
IOmniParallelSimpleLoop = interface
function CancelWith(const token: IOmniCancellationToken): IOmniParallelSimpleLoop;
function NoWait: IOmniParallelSimpleLoop;
function NoMsgBlock: IOmniParallelSimpleLoop;
function NumTasks(taskCount : integer): IOmniParallelSimpleLoop;
function OnStop(stopCode: TProc): IOmniParallelSimpleLoop; overload;
function OnStop(stopCode: TOmniTaskStopDelegate): IOmniParallelSimpleLoop; overload;
Expand All @@ -530,6 +531,7 @@ interface
IOmniParallelSimpleLoop<T> = interface
function CancelWith(const token: IOmniCancellationToken): IOmniParallelSimpleLoop<T>;
function NoWait: IOmniParallelSimpleLoop<T>;
function NoMsgBlock: IOmniParallelSimpleLoop<T>;
function NumTasks(taskCount : integer): IOmniParallelSimpleLoop<T>;
function OnStop(stopCode: TProc): IOmniParallelSimpleLoop<T>; overload;
function OnStop(stopCode: TOmniTaskStopDelegate): IOmniParallelSimpleLoop<T>; overload;
Expand Down Expand Up @@ -902,6 +904,7 @@ TPartitionInfo = record
FFirst : integer;
FInitializerDelegate: TOmniSimpleTaskInitializerTaskDelegate;
FLast : integer;
FNoMsgBlock : boolean;
FNoWait : boolean;
FNumTasks : integer;
FNumTasksManual : boolean;
Expand All @@ -918,6 +921,7 @@ TPartitionInfo = record
constructor Create(first, last: integer; step: integer = 1);
destructor Destroy; override;
function CancelWith(const token: IOmniCancellationToken): IOmniParallelSimpleLoop;
function NoMsgBlock: IOmniParallelSimpleLoop;
function NoWait: IOmniParallelSimpleLoop;
function NumTasks(taskCount : integer): IOmniParallelSimpleLoop;
function OnStop(stopCode: TProc): IOmniParallelSimpleLoop; overload;
Expand All @@ -942,6 +946,7 @@ TOmniParallelSimpleLoop<T> = class(TInterfacedObject, IOmniParallelSimpleLoop<
public
constructor Create(const arr: TArray<T>);
function CancelWith(const token: IOmniCancellationToken): IOmniParallelSimpleLoop<T>; inline;
function NoMsgBlock: IOmniParallelSimpleLoop<T>; inline;
function NoWait: IOmniParallelSimpleLoop<T>; inline;
function NumTasks(taskCount : integer): IOmniParallelSimpleLoop<T>; inline;
function OnStop(stopCode: TProc): IOmniParallelSimpleLoop<T>; overload; inline;
Expand Down Expand Up @@ -3510,6 +3515,8 @@ procedure TOmniParallelSimpleLoop.InternalExecute(const taskDelegate: TTaskDeleg
lockAggregate: IOmniCriticalSection;
task : IOmniTaskControl;
taskCount : integer;
awaited : DWORD;
handles: array [0 .. 0] of THandle;
begin
dmOptions := [];
taskCount := FNumTasks;
Expand All @@ -3531,7 +3538,17 @@ procedure TOmniParallelSimpleLoop.InternalExecute(const taskDelegate: TTaskDeleg
FCountStopped.Allocate //all done
else
{$IFDEF MSWINDOWS}
WaitForSingleObject(FCountStopped.Handle, INFINITE);
if FNoMsgBlock then begin
while true do begin
handles[0] := FCountStopped.Handle;
awaited := MsgWaitForMultipleObjects(1, handles, false, INFINITE, QS_ALLINPUT);
case awaited of
WAIT_OBJECT_0: break;
WAIT_OBJECT_0 + 1: DSiProcessThreadMessages;
end;
end;
end else
WaitForSingleObject(FCountStopped.Handle, INFINITE);
{$ELSE}
FCountStopped.Synchro.WaitFor(INFINITE);
{$ENDIF ~MSWINDOWS}
Expand All @@ -3540,6 +3557,12 @@ procedure TOmniParallelSimpleLoop.InternalExecute(const taskDelegate: TTaskDeleg
end;
end; { TOmniParallelSimpleLoop.InternalExecute }

function TOmniParallelSimpleLoop.NoMsgBlock: IOmniParallelSimpleLoop;
begin
FNoMsgBlock := true;
Result := Self;
end; { TOmniParallelSimpleLoop.NoMsgBlock }

function TOmniParallelSimpleLoop.NoWait: IOmniParallelSimpleLoop;
begin
FNoWait := true;
Expand Down Expand Up @@ -3676,6 +3699,12 @@ function TOmniParallelSimpleLoop<T>.Initialize(
Result := Self;
end; { TOmniParallelSimpleLoop<T>.Initialize }

function TOmniParallelSimpleLoop<T>.NoMsgBlock: IOmniParallelSimpleLoop<T>;
begin
FIterator.NoMsgBlock;
Result := Self;
end; { TOmniParallelSimpleLoop<T>.NoMsgBlock }

function TOmniParallelSimpleLoop<T>.NoWait: IOmniParallelSimpleLoop<T>;
begin
FIterator.NoWait;
Expand Down
Loading