ParallelEnd

Category: Common Steps
Namespace: CMS.Plugin.FlowManagement.Abstractions.FlowBusiness.Activitys
Base Class: ParallelEndActivity (SYC.Flow.Kernel)
Module: FlowManagement.Abstractions

Overview

ParallelEnd is a special node used to end parallel flows. When the flow runs to the parallel flow end step, it converges all parallel branches and ends the entire parallel flow through its internal logic operation rules. It inherits from SYC.Flow.Kernel.ParallelEndActivity, providing flexible waiting modes and automatic completion of unfinished branches.

ParallelEnd must be paired with ParallelStart. It is responsible for waiting for the completion of parallel branches and deciding when to continue executing the subsequent flow based on the configured waiting mode.

Business Scenarios

Applicable Scenarios

• Converge Parallel Detection Results: Wait for all detection branches to complete and summarize results
• Sync Parallel Processing: Ensure all parallel tasks are completed before continuing
• Fast Response: Continue execution as soon as one branch completes (WaitOne mode)
• Timeout Control: Combine with timeout mechanism to avoid infinite waiting
• Branch Result Integration: Collect output data from all parallel branches

Role in System

ParallelEnd plays the role of a parallel flow synchronizer in the LMES flow system:

• Converges all parallel branches
• Decides when to continue execution based on waiting mode
• Automatically completes unfinished parallel branches
• Manages level codes for parallel flows
• Provides completion signal for parallel flows

Collaboration with Other Nodes

• ParallelStart: Must be paired, ParallelStart is responsible for starting parallel branches
• BusinessActivity: Business nodes in parallel branches, flow to ParallelEnd after execution
• Subsequent Nodes: Flow continues to execute subsequent nodes after ParallelEnd completes

Configuration Description

Basic Configuration

Property Name	Type	Required	Default	Description
WaitMode	WaitMode	Yes	WaitAll	Waiting Mode (WaitOne/WaitAll)
ExtendedProperty	BusinessPropertyCollection	No	Empty Collection	Step extended properties

Configuration Details

WaitMode

Description: Waiting mode, decides when ParallelEnd continues to execute the subsequent flow.

Value Range:

• WaitOne: Continue execution as soon as one parallel branch completes
• WaitAll: Wait for all parallel branches to complete before continuing execution

Notes:

• WaitOne Mode: Suitable for fast response scenarios, the first completed branch triggers continuation, other branches will be automatically completed
• WaitAll Mode: Suitable for scenarios requiring all results, must wait for all branches to complete
• Choosing the appropriate waiting mode is crucial for flow performance and correctness

ExtendedProperty

Description: Extended property collection, allowing custom properties to be added to the node, which can be written to the flow context.

Value Range: BusinessPropertyCollection object

Notes:

• Extended properties will be written to the flow context when the parallel flow ends
• Can be used to record summary information of the parallel flow

Flow Context

Input Parameters

Parameter Name	Type	Description
Parallel Branch Output	any	Data written by each parallel branch
LevelCode	string	Level code of parallel branches
LevelKey	string	Level key information
LevelLabel	string	Level label information

Output Parameters

Parameter Name	Type	Description
Extended Property Value	any	Property values configured in ExtendedProperty
Summary Data	any	Summary data written in ProcessAsync

Data Flow Description

1. Data Convergence: ParallelEnd can access flow context data written by all parallel branches
2. Level Code: Uses level code to identify and manage parallel branches
3. Auto Complete: Unfinished branches will be automatically marked as completed
4. Data Integration: Results from all branches can be integrated in nodes following ParallelEnd

Business Logic Description

Processing Flow

The execution flow of ParallelEnd is as follows:

1. CanEnterAsync: Determine if node can be entered

   • Get the number of parallel branches (from the number of outgoing transitions of ParallelStart)
   • Judge based on waiting mode:
     • WaitOne Mode:
       • The first arriving branch can enter
       • Subsequent arriving branches cannot enter
       • Reset counter when all branches arrive
     • WaitAll Mode:
       • Can enter only when all branches arrive
       • Use atomic operation to count arriving branches
       • The last arriving branch triggers entry
   • Use SemaphoreSlim for thread synchronization
   • Record logs for waiting mode and branch count

2. EnterAsync: Enter parallel flow end node

   • Wait for semaphore release (controlled by CanEnterAsync)
   • Respond to flow cancellation token
   • Create new work item (FlowItem)
   • Set work item properties:
     • StartTime and FinishTime: Current time
     • TaskStat: FlowItemStatus.AutoFinished
     • AutoFinish: 1 (Auto finish flag)
     • EnableCalculate: 1 (Participate in calculation)
     • LevelCode: Current level code
     • KeyInfo and KeyLabel: Get from event args
     • TaskName: Node name
   • Add work item to flow instance
   • Update current work item ID
   • Record entry log

3. EndingParallelProcess: End parallel flow

   • Find all work items belonging to the current parallel flow
   • Filter work items by level code (LevelCode)
   • Mark unfinished work items as auto-finished:
     • Set AutoFinish = 1
     • Set TaskStat = FlowItemStatus.AutoFinished
     • Set FinishTime = Current time

4. ExitAsync: Exit parallel flow end node

   • Calculate execution time
   • Record exit log

Flow Chart

[Parallel Branch 1] ┐
[Parallel Branch 2] ├→ [ParallelEnd] → [Next Node]
[Parallel Branch 3] ┘

Waiting Mode Details

WaitOne Mode

Timeline: ─────────────────────────────→

Branch 1:  ──────●────────────────────  (1st completes, triggers continue)
Branch 2:  ────────────●──────────────  (2nd completes, auto-completed)
Branch 3:  ──────────────────●────────  (3rd completes, auto-completed)

ParallelEnd: ──────●────────────────  (Continues when 1st branch completes)

WaitAll Mode

Timeline: ─────────────────────────────→

Branch 1:  ──────●────────────────────  (1st completes, waits)
Branch 2:  ────────────●──────────────  (2nd completes, waits)
Branch 3:  ──────────────────●────────  (3rd completes, triggers continue)

ParallelEnd: ──────────────────●────  (Continues when all branches complete)

Level Code Mechanism

ParallelEnd uses level codes to identify and manage parallel branches:

• AllSubflowCode: Get level codes of all subflows

  • Search in flow instance work items
  • Filter by parent level code and current key
  • Return all matching subflow level codes

• GetParentInfo: Get parent flow information

  • Find parent flow work item by level code
  • Return KeyInfo and KeyLabel of parent flow

Dependency Services

Service Interface	Purpose	Description
Flow	Flow Instance	Access flow context and work items
Flow.Logger	Logger	Record execution logs of parallel flow
SemaphoreSlim	Semaphore	Used for thread synchronization and wait control

Exception Handling

Exception handling in ParallelEnd node:

• Cancellation Token: Respond to flow cancellation token while waiting for semaphore
• Thread Safety: Use atomic operations (Interlocked) to ensure thread safety of counters
• Null Reference: Check if flow instance and work items are null

Logging

ParallelEnd records the following key logs:

• Entry Judgment: CanEnter={canEnter}, WaitMode={WaitMode}, TransitionCount={transitionCount}
• Enter Node: Enter, TaskID={taskId}
• Subflow Info: CurrentLevel={currentLevel}, ParentLevel={parentLevel}, CurrentKey={key}, Subflows={subflows}
• Exit Node: Exit, Time Consumption={milliseconds}ms

Usage Examples

Basic Example: WaitAll Mode

{
  "Name": "Parallel Detection Flow",
  "Activities": [
    {
      "Type": "ParallelStart",
      "Name": "Start Parallel Detection",
      "Alias": "ParallelStart1"
    },
    {
      "Type": "BusinessActivity",
      "Name": "Visual Check",
      "Alias": "AppearanceCheck"
    },
    {
      "Type": "BusinessActivity",
      "Name": "Dimension Check",
      "Alias": "DimensionCheck"
    },
    {
      "Type": "BusinessActivity",
      "Name": "Function Check",
      "Alias": "FunctionCheck"
    },
    {
      "Type": "ParallelEnd",
      "Name": "End Parallel Detection",
      "Alias": "ParallelEnd1",
      "WaitMode": "WaitAll"
    },
    {
      "Type": "BusinessActivity",
      "Name": "Summarize Results",
      "Alias": "SummaryResult"
    }
  ],
  "Transitions": [
    {
      "From": "ParallelStart1",
      "To": "AppearanceCheck"
    },
    {
      "From": "ParallelStart1",
      "To": "DimensionCheck"
    },
    {
      "From": "ParallelStart1",
      "To": "FunctionCheck"
    },
    {
      "From": "AppearanceCheck",
      "To": "ParallelEnd1"
    },
    {
      "From": "DimensionCheck",
      "To": "ParallelEnd1"
    },
    {
      "From": "FunctionCheck",
      "To": "ParallelEnd1"
    },
    {
      "From": "ParallelEnd1",
      "To": "SummaryResult"
    }
  ]
}

Advanced Example: WaitOne Mode

{
  "Name": "Fast Response Flow",
  "Activities": [
    {
      "Type": "ParallelStart",
      "Name": "Start Parallel Query",
      "Alias": "ParallelStart1"
    },
    {
      "Type": "ApiCallActivity",
      "Name": "Query Main DB",
      "Alias": "QueryMainDB"
    },
    {
      "Type": "ApiCallActivity",
      "Name": "Query Backup DB",
      "Alias": "QueryBackupDB"
    },
    {
      "Type": "ApiCallActivity",
      "Name": "Query Cache",
      "Alias": "QueryCache"
    },
    {
      "Type": "ParallelEnd",
      "Name": "End Parallel Query",
      "Alias": "ParallelEnd1",
      "WaitMode": "WaitOne"
    },
    {
      "Type": "BusinessActivity",
      "Name": "Process Data",
      "Alias": "ProcessData"
    }
  ],
  "Transitions": [
    {
      "From": "ParallelStart1",
      "To": "QueryMainDB"
    },
    {
      "From": "ParallelStart1",
      "To": "QueryBackupDB"
    },
    {
      "From": "ParallelStart1",
      "To": "QueryCache"
    },
    {
      "From": "QueryMainDB",
      "To": "ParallelEnd1"
    },
    {
      "From": "QueryBackupDB",
      "To": "ParallelEnd1"
    },
    {
      "From": "QueryCache",
      "To": "ParallelEnd1"
    },
    {
      "From": "ParallelEnd1",
      "To": "ProcessData"
    }
  ]
}

Complete Flow Example: Result Summary

// Summarize results in node following ParallelEnd
[Serializable]
[Design("Summarize Results", "Summarize results of all parallel checks", Sort = 1)]
[Category("Custom")]
public class SummaryResultActivity : BusinessActivity
{
    public override async Task ProcessAsync(ProcessflowEventArgs args)
    {
        // Read results from each parallel branch
        var appearanceResult = Flow.DataItems["AppearanceCheck_Result"]?.ToString();
        var dimensionResult = Flow.DataItems["DimensionCheck_Result"]?.ToString();
        var functionResult = Flow.DataItems["FunctionCheck_Result"]?.ToString();
        
        // Summarize results
        var allPassed = appearanceResult == "Pass" 
                     && dimensionResult == "Pass" 
                     && functionResult == "Pass";
        
        // Write summary result
        Flow.DataItems["FinalResult_Value"] = allPassed ? "Pass" : "Fail";
        
        Flow.Logger.LogMessage(
            $"Result Summary: Appearance={appearanceResult}, Dimension={dimensionResult}, Function={functionResult}, Final={allPassed}", 
            Name);
        
        await base.ProcessAsync(args);
    }
}

Extension Development Guide

Inheritance Hierarchy

Activity (SYC.Flow.Kernel)
  └── ParallelEndActivity (SYC.Flow.Kernel)
        └── ParallelEnd

Overridable Methods

Method Name	Purpose	When to Override
CanEnterAsync	Determine if node can be entered	Need custom waiting logic
EnterAsync	Executed when entering node	Need custom parallel flow end logic
ExitAsync	Executed when exiting node	Need custom post-parallel flow processing
EndingParallelProcess	End parallel flow	Need custom logic for unfinished branches
AllSubflowCode	Get all subflow codes	Need custom subflow identification logic
GetParentInfo	Get parent flow info	Need custom parent flow info retrieval logic

Custom Parallel End Node Example

[Serializable]
[Design("Custom Parallel End", "Parallel end node with result validation", Sort = 1)]
[Category("Custom")]
public class CustomParallelEnd : ParallelEnd
{
    [Design("Min Success Count", "Minimum number of successful branches required", Sort = 1)]
    [Category("Configuration")]
    [DataMember]
    public int MinSuccessCount { get; set; } = 2;
    
    public override async Task<bool> CanEnterAsync(ProcessflowEventArgs args)
    {
        // Call base class waiting logic first
        var canEnter = await base.CanEnterAsync(args);
        
        if (canEnter)
        {
            // Check successful branch count
            var successCount = 0;
            var subflows = AllSubflowCode(args.LevelCode);
            
            foreach (var subflow in subflows)
            {
                var result = Flow.DataItems[$"{subflow}_Result"]?.ToString();
                if (result == "Success")
                {
                    successCount++;
                }
            }
            
            if (successCount < MinSuccessCount)
            {
                Flow.Logger.LogWarningMessage(
                    $"Success branch count {successCount} is less than minimum requirement {MinSuccessCount}", 
                    Name);
                return false;
            }
        }
        
        return canEnter;
    }
    
    public override async Task EnterAsync(ProcessflowEventArgs e)
    {
        // Record results of all branches
        var subflows = AllSubflowCode(e.LevelCode);
        var results = new List<string>();
        
        foreach (var subflow in subflows)
        {
            var result = Flow.DataItems[$"{subflow}_Result"]?.ToString();
            results.Add($"{subflow}={result}");
        }
        
        Flow.Logger.LogMessage($"Parallel branch results: {string.Join(", ", results)}", Name);
        
        await base.EnterAsync(e);
    }
}

Registration and Integration

1. Compile Node:

   • Inherit ParallelEnd class
   • Add Design and Category attributes
   • Compile to DLL

2. Deploy Node:

   • Place DLL in LMES plugin directory
   • Restart LMES service

3. Use Node:

   • Select custom parallel end node in flow designer
   • Configure node properties (especially WaitMode)
   • Connect all parallel branches to this node

Notes

• ⚠️ Must Pair: ParallelEnd must be paired with ParallelStart
• ⚠️ Waiting Mode: Choose appropriate waiting mode (WaitOne or WaitAll) based on business requirements
• ⚠️ Branch Count: Number of transitions flowing into ParallelEnd must match outgoing count of ParallelStart
• ⚠️ Thread Safety: ParallelEnd uses thread-safe mechanisms, but code in parallel branches must guarantee thread safety
• ⚠️ Data Conflict: Beware of parallel branches writing to the same flow context key, causing data overwrite
• ⚠️ Cancellation Token: ParallelEnd responds to flow cancellation token, can interrupt waiting
• ⚠️ Performance: WaitAll mode performance depends on the slowest branch
• 💡 Best Practices:
  • Use different output keys for each parallel branch
  • Summarize results of all branches in node following ParallelEnd
  • When using WaitOne mode, ensure the first completed branch provides sufficient info
  • Consider adding timeout mechanism to avoid infinite waiting
  • Record detailed logs for easier troubleshooting of parallel flows

Related Nodes

• ParallelStart: Parallel flow start node, must be paired
• BusinessActivity: Business node base class in parallel branches
• SubflowActivity: Subflow node, also supports parallel execution

FAQ

Q1: What is the difference between WaitOne and WaitAll?

A:

• WaitOne: Continue execution as soon as one parallel branch completes, suitable for fast response scenarios (e.g., multi-source query, take fastest result)
• WaitAll: Must wait for all parallel branches to complete before continuing, suitable for scenarios requiring all results (e.g., all checks must complete)

Q2: What if a parallel branch fails?

A:

• If a branch throws an exception, that branch stops execution
• WaitAll mode: Waits for other branches to complete, failed branch is automatically marked as completed
• WaitOne mode: If the first completed branch is the failed one, flow continues, check result in subsequent node
• Recommend capturing exceptions in parallel branches and writing result flags, check after ParallelEnd

Q3: How to get results from all parallel branches?

A: In the node following ParallelEnd, read data written by each branch via flow context:

var result1 = Flow.DataItems["Branch1_Result"];
var result2 = Flow.DataItems["Branch2_Result"];
var result3 = Flow.DataItems["Branch3_Result"];

Q4: Does ParallelEnd automatically complete unfinished branches?

A: Yes, ParallelEnd's EndingParallelProcess method automatically marks all unfinished parallel branches as auto-finished status (AutoFinished).

Q5: How to avoid data conflict between parallel branches?

A:

• Use different flow context key names for each branch
• Use branch identifier as key prefix, e.g., Branch1_Result, Branch2_Result
• Avoid modifying shared data in parallel branches
• If sharing data is necessary, use thread-safe data structures

Q6: Can parallel flows be nested?

A: Yes, but note:

• Each ParallelStart/ParallelEnd pair must be correctly matched
• Level codes automatically manage nesting levels
• Nested parallel flows increase complexity, use with caution

Update History

Date	Version	Description
2025-11-28	1.0	Initial Release

---

Last updated: 2025-11-28