elsa-workflows / elsa-workflows/elsa-core
[BUG]In 3.2.1 ParallelForEach didn't work as expected
- Dominant language
- C#
- Stars
- 7.9k
- Forks
- 1.5k
- Avg merge
- 15h 22m
- Merged PRs (30d)
- 114
Description
## Description
I use `ParallelForEach ` in my workflow and I should get 3 lines of `start Hello from resource ID`, but only one line was output。
MyCode Like this:
```
var services = new ServiceCollection();
services.AddElsa();
var serviceProvider = services.BuildServiceProvider();
var runner = serviceProvider.GetRequiredService();
await runner.RunAsync(new VideoStreamProcessor());
class VideoStreamProcessor : WorkflowBase
{
protected override void Build(IWorkflowBuilder builder)
{
builder.Name = "Video Stream Processor";
var resourceIds = builder.WithVariable>().WithMemoryStorage();
var outputData = builder.WithVariable().WithMemoryStorage();
builder.Root = new Sequence
{
Activities =
{
new WriteLine("Fetching video stream addresses..."),
new FetchVideoStreams
{
Result = new(resourceIds)
},
new ParallelForEach
{
Items = new(resourceIds),
Body = new Sequence
{
Activities =
{
new WriteLine(context => $"Processing video stream for resource ID: {context.GetVariable("CurrentValue")!}"),
new HelloWorld
{
ResourceId = new(context => context.GetVariable("CurrentValue")!)
},
new WriteLine(context => $"Finished video stream for resource ID: {context.GetVariable("CurrentValue")!}"),
}
}
},
new WriteLine("All video streams processed!")
}
};
}
}
///
/// Fetches video stream addresses from a data source.
///
lass FetchVideoStreams : CodeActivity>
{
protected override void Execute(ActivityExecutionContext context)
{
var streamAddresses = GetVideoStreamAddresses().ToAsyncEnumerable();
Result.Set(context, streamAddresses);
}
private IEnumerable GetVideoStreamAddresses()
{
return new List
{
"http://example.com/stream1",
"http://example.com/stream2",
"http://example.com/stream3"
};
}
}
class HelloWorld : CodeActivity
{
[Input]
public Input ResourceId { get; set; } = default!;
protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var resourceId = ResourceId.Get(context);
Console.WriteLine($"start Hello from resource ID: {resourceId}");
// 模拟异步操作
await Task.Delay(1000000);
Console.WriteLine($"end Hello from resource ID: {resourceId}");
}
}
```
Help Me!
Contributor guide
Research direction
Start by reproducing the supplied workflow, focusing on ParallelForEach and the Execute methods in FetchVideoStreams and HelloWorld. Check why the resourceIds sequence produces only one HelloWorld start line, then verify that the workflow emits three start lines for the three listed resource IDs while preserving the expected completion behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- csharp
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100