Class WorkerParent

A base class for WorkerSystem and WorkerBase with shared functionality for managing child workers, RunAsync(), etc.

Inheritance

Object

WorkerParent

WorkerBase

WorkerSystemBase

Implements

IDisposeOnFinished

Namespace: actionETL

Assembly: actionETL.dll

Syntax

public abstract class WorkerParent : IDisposeOnFinished

Properties

AggregateErrorOutputRows

Gets the total number of rows sent to any error output port in this or any descendant worker that has completed.

Note that this aggregates RowsReceived, i.e. all rows that were received by the error output ports, even if they were not necessarily sent to the downstream worker. Also see TotalRowsSent.

Note: This property is thread-safe.

Declaration

public long AggregateErrorOutputRows { get; }

Property Value

Type	Description
Int64	The total number of error output port rows. Returns 0 if this worker has not yet completed.

AggregateOutputRows

Gets the total number of rows sent to any (non-error) output port in this or any descendant worker that has completed.

Note: This property is thread-safe.

Declaration

public long AggregateOutputRows { get; }

Property Value

Type	Description
Int64	The total number of (non-error) output port rows. Returns 0 if this worker has not yet completed.

AggregateWorkersCompleted

Gets the total number of descendant workers plus this worker that ran and completed.

Note: This property is thread-safe.

Declaration

public long AggregateWorkersCompleted { get; }

Property Value

Type	Description
Int64	The number of completed workers. Returns 0 if this worker has not yet completed.

Children

Returns a list with all current worker children. It is often used in debugging for navigating the worker hierarchy, and in testing for checking worker status.

Note: This property is thread-safe. On each access, it returns a new list with a copy of all current children. In performance sensitive areas, avoid reading this property a large number of times (e.g. once for every dataflow row).

Declaration

public IReadOnlyList<WorkerBase> Children { get; }

Property Value

Type	Description
IReadOnlyList<WorkerBase>

DebugCommands

Gets or sets commands for when to launch or break a debugger when debugging workers and the worker system (i.e. WorkerParent instances). Multiple commands can be set at the same time by combining them with bitwise OR (| in C#).

The various "On..." commands defines when to break or launch the debugger, while Launch and Break defines what to do, i.e. launching and breaking the debugger. In most debugging scenarios, both one or more "On..." commands, as well as either Launch or Break, should be set.

There are also several predefined combinations, e.g. BreakOnCompleted that is very useful for viewing WorkerParents after they complete.

Note: Any Launch command will be automatically downgraded to a Break command after the first attempt to launch a debugger for this WorkerParent.

Note: The WorkerParent state is checked for a match both when this property is set, and when the WorkerParent state changes.

Note: This property is thread-safe.

Declaration

public DebugWorkerParentCommands DebugCommands { get; set; }

Property Value

Type	Description
DebugWorkerParentCommands

See Also

WorkerParentState

HasChildren

True if the worker has any children.

Note: This property is thread-safe.

Declaration

public bool HasChildren { get; }

Property Value

Type	Description
Boolean

InstantCompleted

Gets the nullable instant the worker completed execution.

Note: This property is thread-safe.

Declaration

public Instant? InstantCompleted { get; }

Property Value

Type	Description
Nullable<Instant>	The nullable instant. Use InstantCompleted.HasValue to determine if the property has a value. HasValue will be false if the worker has not started or has not finished executing.

InstantCreated

Gets the instant the worker or worker system was created (which for workers is also when it was added to its parent).

Note: This property is thread-safe.

Declaration

public Instant InstantCreated { get; }

Property Value

Type	Description
Instant

InstantStarted

Gets the nullable instant the worker started execution.

Note: This property is thread-safe.

Declaration

public Instant? InstantStarted { get; }

Property Value

Type	Description
Nullable<Instant>	The nullable instant. Use InstantStarted.HasValue to determine if the property has a value. HasValue will be false if the worker has not started executing.

IsCanceled

Gets a value indicating whether this instance is canceled.

Note: This property is thread-safe.

Declaration

public bool IsCanceled { get; }

Property Value

Type	Description
Boolean	true if this instance is canceled; otherwise, false.

IsCompleted

Gets a value indicating whether this instance is completed, i.e. has stopped running.

Note: This property is thread-safe.

Declaration

public bool IsCompleted { get; }

Property Value

Type	Description
Boolean	true if this instance is completed; otherwise, false.

IsCreated

Gets a value indicating whether this instance is created, but has not progressed to any later state.

Note: This property is thread-safe.

Declaration

public bool IsCreated { get; }

Property Value

Type	Description
Boolean	true if this instance is created but has not progressed to any later state; otherwise, false.

IsError

Gets a value indicating whether this instance has errored.

Note: This property is thread-safe.

Declaration

public bool IsError { get; }

Property Value

Type	Description
Boolean	true if this instance is errored; otherwise, false.

IsFailed

Gets a value indicating whether this instance has completed unsuccessfully.

Note: This property is thread-safe.

Declaration

public bool IsFailed { get; }

Property Value

Type	Description
Boolean	true if this instance is Canceled, Error, or Fatal; otherwise, false.

IsFatal

Gets a value indicating whether this instance has fatally faulted.

Note: This property is thread-safe.

Declaration

public bool IsFatal { get; }

Property Value

Type	Description
Boolean	true if this instance has fatally faulted; otherwise, false.

IsRunning

Gets a value indicating whether this instance is running.

Note: This property is thread-safe.

Declaration

public bool IsRunning { get; }

Property Value

Type	Description
Boolean	true if this instance is running; otherwise, false.

IsSucceeded

Gets a value indicating whether this instance has succeeded.

Note: This property is thread-safe.

Declaration

public bool IsSucceeded { get; }

Property Value

Type	Description
Boolean	true if this instance has succeeded; otherwise, false.

Item[String]

Gets the child worker with the specified worker name.

There is a small overhead when retrieving the worker by name. Consider storing and using the worker instance in a variable in performance sensitive areas.

Note: This property is thread-safe.

Declaration

public WorkerBase this[string workerName] { get; }

Parameters

Type	Name	Description
String	workerName	Name of the child worker.

Property Value

Type	Description
WorkerBase	The child worker. To access members from derived workers, the returned value must first be cast to the appropriate worker type.

Exceptions

Type	Condition
ArgumentException	Child worker not found.

KeepChildrenLevels

Gets or sets a value indicating whether to remove any children and descendant workers after completion of the parent. This is useful when you need to access a property on a descendant worker after the parent has completed, e.g. to check its OutcomeStatus or other property explicitly.

The value is inherited recursively from parent to all its child workers when they start, but can also be set using AConfig with a suitable ApplyTo parameter.

Set to -1 to retain all descendants. 0 (the default for workers) to remove descendants after the parent completes. A positive number to retain descendant workers that many levels down.

The worker system gets a value of 1 by default, i.e. it will retain its direct children when it completes (which allows checking properties of top level workers after the worker system completes), but its grandchildren workers will by default be removed on completion.

This property can be set on workers in the middle of the worker hierarchy, i.e. not just on the worker system.

Note: This property is thread-safe, and it can be set at any time, including after the worker has started running or completed.

Note that sibling and ancestor workers are always available.

Declaration

public int KeepChildrenLevels { get; set; }

Property Value

Type

Description

Int32

-1 to retain all descendants 0 (the default for workers) to remove descendants after the parent completes. This can also be set after the parent has completed, and any descendants will be removed at that time. A positive number to retain descendant workers that many levels down. The top level worker system defaults to 1. A value of 2 would retain direct children and grand children when the parent completes, but worker descendants beyond that would be removed.

Locator

Gets the locator string, which represents where in the hierarchy this worker or worker system resides. It consists of the names of all ancestors separated by /, e.g.: /Root/Extract Products/Read CSV Product File. The worker system at the top of the hierarchy defaults to the name "Root", but can also be given a custom name when being created, which is useful when running different worker systems together, e.g. during testing.

Note: This property is thread-safe.

Declaration

public string Locator { get; }

Property Value

Type	Description
String	The locator string.

LogFactory

Gets the log factory for the worker system. It allows acquiring new loggers (if the preexisting ones on workers, ports etc. are not sufficient), flushing log messages etc.

Note: This property is thread-safe.

Declaration

public IALogFactory LogFactory { get; }

Property Value

Type	Description
IALogFactory

Logger

Gets the logger instance for this worker or worker system, which should be used for logging messages related to this worker system or worker. The logger's name is set to the Locator string of this worker or worker system, which will appear in the logged output.

Note: This property is thread-safe.

Declaration

public ALog Logger { get; }

Property Value

Type	Description
ALog

MaxRunningChildren

Gets or sets the maximum number of simultaneously running children allowed for this worker, or a negative number to not set a limit.

Note however that to avoid scheduling deadlocks, workers with input ports, i.e. transforms and targets, will be started irrespective of this setting. Transforms and targets still count towards the number of running children, and can therefore contribute to blocking workers without input ports from being started.

Defaults to -1, i.e. unlimited.

If greater than zero, no new children will be started until the number of running children is lower than this property.

If zero, no children will be started. Note that if no children are currently running, and no children can be started, the worker will have finished running its children.

As a more fine-grained alternative to setting this property on the parent, also consider setting IsStartable constraints on the children. See Worker Execution for more details.

Note: This property is thread-safe, and it can be set at any time, including after the worker or its children have started running.

Declaration

public int MaxRunningChildren { get; set; }

Property Value

Type	Description
Int32

See Also

IsStartable

Name

The name of the worker or worker system. All children of a parent worker must have unique names.

The worker system at the top of the hierarchy defaults to the name "Root", but can also be given a custom name when being created, which is useful when running different worker systems together, e.g. during testing. Other worker names default to their type name, followed by a number, e.g. "FileExistsWorker 1".

Note: This property is thread-safe.

Declaration

public string Name { get; }

Property Value

Type	Description
String

RunningDuration

Gets the running duration - how long the worker has been running or did run.

Note: This property is thread-safe.

Declaration

public Duration? RunningDuration { get; }

Property Value

Type	Description
Nullable<Duration>	The nullable running duration. Use RunningDuration.HasValue to determine if the property has a value. HasValue will be false if the worker has not started executing.

Status

Gets the execution status of the WorkerParent.

Note: This property is thread-safe.

Declaration

public WorkerParentStatus Status { get; }

Property Value

Type	Description
WorkerParentStatus

WorkerSystem

Gets the worker system. Note that in the worker system instance, this will point to itself.

Note: This property is thread-safe.

Declaration

public WorkerSystem WorkerSystem { get; }

Property Value

Type	Description
WorkerSystem

Methods

AddChildCompletedCallback(Action<WorkerBase>)

Adds a callback which will be called after each child worker that completes.

If the child worker is started, these callbacks will run, even if the child fails. If multiple callbacks have been added, they will all be called (even on failure) in the order they were added.

Each callback takes the worker that completed (typed as WorkerBase) as a parameter.

These callbacks are useful for taking a specific action after each child worker completes, e.g. to track the progress of the child workers. The callback can of course check e.g. the name or type of the child worker to decide what action to take.

Alternatives to AddChildCompletedCallback() include calling AddCompletedCallback(Func<WorkerBase, OutcomeStatus, Task<OutcomeStatus>>) on all or selected children, adding additional child workers with start constraints, or calling RunChildrenAsync() explicitly and awaiting individual child workers.

Note: Adding the callback is thread-safe. The callback will later run on the parent's scheduling thread. The callback must be synchronous, and should return quickly to avoid delaying the scheduling of other worker children of this parent.

Declaration

public void AddChildCompletedCallback(Action<WorkerBase> childCompletedAction)

Parameters

Type	Name	Description
Action<WorkerBase>	childCompletedAction	The synchronous callback that will be called after each child worker that completes. See "Remarks" below for details.

Remarks

Each callback takes the worker itself (typed as WorkerBase) as a parameter. If you need to access members of the derived worker type, there are two options:

• If you have access to the source of the worker, implement the callback as a (non-static) method in that worker. This allows accessing worker members without casting the worker parameter to the derived worker type.
• If you don't have access to the source of the worker, or you want to implement the callback using a lambda, cast the worker parameter to the actual derived type of the worker.

Examples

In this example we track how many child workers have failed:

AddChildCompletedCallback(w =>
{
    if (w.IsFailed)
        _numberOfChildrenFailed++;
});

Exceptions

Type	Condition
ArgumentNullException	childCompletedAction
InvalidOperationException	Cannot add 'ChildCompleted' callback after the worker or worker system has completed.

AddStartingChildrenCallback(Func<WorkerParent, Task<ProgressStatus>>)

Adds a callback which will be called every time just before the children run. These callbacks won't run if there are no children, or if the children won't run due to a AddStartingCallback(Func<WorkerBase, Task<ProgressStatus>>) callback.

When the callbacks do run, no further child workers or ports can be added, and ports cannot be linked (until optionally calling RemoveChildren()). Typical uses include:

• Configuring child workers and ports, even if they have been added externally. UnionAllTransform<TInputOutput> e.g. uses this to set port PortBufferingMode for its input ports, since it doesn't add them itself.
• Inspecting all child workers before they run

If multiple callbacks have been added, they will all be called (even on failure) in the order they were added, e.g. callbacks added in base classes will be called before callbacks added in derived classes.

The callback takes the worker or worker system itself as a parameter, typed as WorkerParent.

Note: Adding the callback is thread-safe. Children won't be started until the callback completes, so ensure the callback finishes quickly.

Declaration

public void AddStartingChildrenCallback(Func<WorkerParent, Task<ProgressStatus>> startingChildrenFuncAsync)

Parameters

Type	Name	Description
Func<WorkerParent, Task<ProgressStatus>>	startingChildrenFuncAsync	The asynchronous callback that will be called every time before the children run. See "Remarks" below for details.

Remarks

The callback takes the worker parent itself as a parameter. If you need to access members of the derived worker parent type, there are two options:

• If you have access to the source of the worker parent, implement the callback as a (non-static) method in that worker parent. This allows accessing worker parent members without casting the worker parent parameter to the derived worker parent type.
• If you don't have access to the source of the worker parent, or you want to implement the callback using a lambda, cast the worker parent parameter to the actual derived type of the worker parent.

The callback implementation is either synchronous and returns a Task<ProgressStatus> explicitly, or is async and returns a ProgressStatus.

The callback should return:

• ProgressStatus.NotCompletedTask (or ProgressStatus.NotCompleted) to continue normal processing (which will run the child workers).
• ProgressStatus.SucceededTask (or ProgressStatus.Succeeded) to neither run any child workers nor any AddChildCompletedCallback(Action<WorkerBase>) callbacks, and pass a Succeeded status to any worker or worker system AddCompletedCallback() callbacks.
• A failure ProgressStatus on failure, which will give the worker parent and any AddCompletedCallback() callbacks that failure status. Neither child workers nor any AddChildCompletedCallback() callbacks will run.

Examples

In this example we change any port BufferingMode from Default to Limited, which is e.g. what UnionAllTransform<TInputOutput> does, since it doesn't require Full buffering, despite having multiple input ports.

AddStartingChildrenCallback(w =>
{
    foreach (var port in myWorker.InputPorts)
        if (port.BufferingMode == PortBufferingMode.Default)
            port.BufferingMode = PortBufferingMode.Limited;
    return ProgressStatus.NotCompletedTask;
});

DisposeOnFinished<TDisposable>(TDisposable)

Adds the specified disposable to the list of objects to be disposed when either of the following occur, whichever comes first:

A: The worker (or WorkerSystem) completes after having run, or B: the worker never runs, and either the worker is removed from its parent, or the worker parent completes.

If either of the above has already taken place, the object will be disposed immediately.

For workers that have been started, the disposal will always happen after its AddCompletedCallback(Func<WorkerBase, OutcomeStatus, Task<OutcomeStatus>>) callbacks (if any) run, and for the WorkerSystem, just before Start() / StartAsync() finishes. Note however that if a worker is never started, any AddCompletedCallback() callbacks won't be called.

Do not add any worker result property instance (e.g. Rows if it had been disposable) as a disposable, since that would dispose the object before any ancestor could retrieve the result.

An alternative is to use UsingActionWorker<TDisposable>. Also see Disposing Disposables.

Note: This method is thread-safe.

Declaration

public TDisposable DisposeOnFinished<TDisposable>(TDisposable disposable)
    where TDisposable : IDisposable

Parameters

Type	Name	Description
TDisposable	disposable	The disposable object. Ignored if null.

Returns

Type	Description
TDisposable	The specified disposable. This allows using or assigning the disposable directly, e.g.: var connection = DisposeOnFinished(connectionBuilder.Create());

Type Parameters

Name	Description
TDisposable	The type of the disposable.

Examples

Adb Keep Open Connection has examples of how to use this class.

GetDownstreamFactory<TInput>()

Gets the dataflow downstream factory for creating transforms and targets that are initially not linked to an upstream worker. This is only rarely needed, e.g. when a transform or target is created before their upstream worker has been created.

Note that in the returned factory, the UpstreamPort will always be null. The newly created transform or target must therefore be linked (using LinkTo(InputPort<TOutput>) or LinkFrom(OutputPortBase<TInput>)) to an upstream worker before it and its siblings are run.

Also see Link, which is the more common way to create and link transforms and targets. See Worker Instantiation and Transform and Target Factory Methods for further details.

Note: This method is thread-safe.

Declaration

public DownstreamFactory<TInput> GetDownstreamFactory<TInput>()
    where TInput : class

Returns

Type	Description
DownstreamFactory<TInput>	The dataflow downstream factory for creating transforms and targets.

Type Parameters

Name	Description
TInput	The type of the rows for the "first" input of the transform or target to create.

Examples

var target = parentWorker.GetDownstreamFactory<MyRow>().CollectionTarget();

RemoveChildren()

Removes all children. Must not be called when any child is running. Logs any errors.

This method is used by looping workers (in conjunction with RunChildrenAsync that add, run, and then remove children multiple times.

Declaration

public void RemoveChildren()

Exceptions

Type	Condition
InvalidOperationException	Children active (i.e. started but not completed) when removing children.

RescheduleChildren()

Triggers the child worker scheduling loop, evaluating their IsStartable property for workers to start.

Note: Only call this method if the default scheduling events are not sufficient, see IsStartable for further details.

Note: This method is thread-safe.

Declaration

public void RescheduleChildren()

RunAsync()

This method is called by the system during the worker or worker system Running phase. It typically contains all the worker or worker system specific logic, including initialization and cleanup. Override this method to implement any custom functionality required by the derived worker or worker system, in particular when deriving from many of the out-of-box workers or worker system with a "Base" suffix.

E.g., WorkerSystem uses it to implement the functionality of the Root(Action<WorkerSystem>) overloads, and SortTransform<TInputOutput> uses it to sort dataflow rows.

Note that if this method returns a Succeeded status and did not call a RunChildrenAsync() overload, then the system will run any child workers.

Also note that the Running phase includes additional places where logic can optionally be inserted via callbacks, to e.g. customize the initialization, cleanup, and error handling of existing workers. This is mostly used when customizing workers or worker systems that are not designed to be derived from (i.e. without a "Base" suffix). See Worker Life-cycle for details.

Declaration

protected abstract Task<OutcomeStatus> RunAsync()

Returns

Type	Description
Task<OutcomeStatus>	An OutcomeStatus describing the success or failure of the worker or worker system.

RunChildrenAsync()

Run child workers asynchronously. They will not be automatically removed after they have finished running.

This method is used when the parent needs to await the running of the child workers, so that it can take action afterwards. This is used by e.g. looping workers that add, run, and then remove children multiple times.

Note that the caller should always check the return value, and normally return any failure status as its own failure status. The system will however escalate any child Fatal status to the parent, if the caller doesn't do it.

Most workers will however not call RunChildrenAsync themselves, in which case the library will call it automatically once the worker RunAsync method has completed.

Declaration

public Task<OutcomeStatus> RunChildrenAsync()

Returns

Type	Description
Task<OutcomeStatus>	A Task{OutcomeStatus} that completes when no more children are running or will be started. The OutcomeStatus shows the worst escalated status of any descendant workers (respecting the children EscalateError settings), or OutcomeStatus.Succeeded on success.

RunChildrenAsync(Boolean)

Run child workers asynchronously.

This method is used when the parent needs to await the running of the child workers, so that it can take action afterwards. This is used by e.g. looping workers that add, run, and then remove children multiple times.

Note that the caller must await the completion of the returned task, and should always check the result, and normally return any failure status as its own failure status. The system will however escalate any child Fatal status to the parent, if the caller doesn't do it.

Most workers will however not call RunChildrenAsync themselves, in which case the library will call it automatically once the worker RunAsync method and any AddRanCallback(Func<WorkerBase, OutcomeStatus, WorkerParentChildrenState, Task<OutcomeStatus>>) callbacks have completed.

Declaration

public Task<OutcomeStatus> RunChildrenAsync(bool removeChildren)

Parameters

Type	Name	Description
Boolean	removeChildren	If set to true, any children will be automatically removed after all children have finished running. This is commonly used by looping workers. Note that this is different vs. KeepChildrenLevels (which removes the children after the parent completes).

Returns

Type	Description
Task<OutcomeStatus>	A Task{OutcomeStatus} that completes when no more children are running or will be started. The OutcomeStatus shows the worst escalated status of any descendant workers (respecting the children EscalateError settings), or OutcomeStatus.Succeeded on success.

ToString()

Returns a String that represents this instance.

Note: This method is thread-safe.

Declaration

public override string ToString()

Returns

Type	Description
String

Overrides

Object.ToString()

Implements

IDisposeOnFinished

See Also

IDisposeOnFinished