How to run using multiple instances and running parallel activities

[robinabganpat]

I'm trying to use Elsa 3 to create a workflow for some import and processing activities of data in a multi-instance environment (microservice-architecture with multiple running stateless instances). I want to create a workflow which imports data from several external API's. These import activities should be scheduled and ran in parallel. After those data import activities are finished, it should then run several processing activities in parallel. It would be really nice if those activities are then ran across all the instances.

I've read this Medium article by @sfmskywalker which sparked my interest as to that this would be possible using Elsa 3. However, I've not been able to create a working prototype, using both MassTransit and EFCore.SqlServer.

For simplicity sake, I've created a dummy activity which would mimic a long running import task.
This is my current setup:

Startup.cs

services.AddElsa(elsa =>
            {
                // Configure Management layer to use EF Core.
                elsa.UseWorkflowManagement(
                    management => management.UseEntityFrameworkCore(c => c.UseSqlServer(<connection string>))
                );
                // Configure Runtime layer to use EF Core.
                elsa.UseWorkflowRuntime(runtime =>
                {
                    runtime.UseEntityFrameworkCore(c => c.UseSqlServer(<connection string>));
                    runtime.UseMassTransitDispatcher();
                });

                elsa.UseMassTransit(x =>
                {
                    x.UseAzureServiceBus(
                       <connection string>,
                        serviceBusFeature =>
                            serviceBusFeature.ConfigureServiceBus = bus =>
                            {
                                bus.PrefetchCount = 4;
                                bus.MaxConcurrentCalls = 32;
                                bus.MaxDeliveryCount = 8;
                            }
                    );
                });
                
                elsa.UseHttp();
                elsa.UseWorkflows();
            });
        }

DummyActivity.cs

[Activity("Elsa", Kind = ActivityKind.Task)]
    public class DummyActivity : CodeActivity
    {
        protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
        {
            Console.WriteLine("Start!");
            for (int i = 0; i < 20; i++)
            {
                await Task.Delay(500);
                Console.WriteLine("Hello World!");
            }

            Console.WriteLine("Finished!");
        }
    }

WorkflowController.cs

[Route("workflow")]
    public class WorkflowController
    {
        private readonly IWorkflowRunner _workflowRunner;

        public WorkflowController(IWorkflowRunner workflowRunner)
        {
            _workflowRunner = workflowRunner;
        }

        [HttpGet]
        public async Task<IActionResult> Get()
        {
            var workflow = new Workflow()
            {
                Root = new Sequence
                {
                    Activities =
                    {
                        new Parallel()
                        {
                            RunAsynchronously = true,
                            Activities = new List<IActivity>()
                            {

                                new DummyActivity() { RunAsynchronously = true, },
                                new DummyActivity() { RunAsynchronously = true, },
                                new DummyActivity() { RunAsynchronously = true, },
                            }
                        },
                        new WriteLine("Done!"),
                    }
                }
            };
            await _workflowRunner.RunAsync(workflow);

            return new OkResult();
        }
    }

Current outcome and questions:

• When running the workflow, I see that the DummyActivities are suspended immediately. I can't really seem to make it work. What am I doing wrong? I would expect the 3 dummy activities to run in parallel, preferably across multiple instances of the running .NET service.
• Does it make sense to both use EfCore and MassTransit? Or should I use one or the other to support running activities across multiple instances of the same service?

I hope someone can help me shed some light on whether this is possible and how I could accomplish this.

[glime-ai[bot]]

Based on the information provided and the context from the sources, it seems you are trying to set up a distributed workflow using Elsa 3 with parallel execution of activities across multiple instances of a microservice architecture. Your setup includes Entity Framework Core for persistence and MassTransit with Azure Service Bus for message dispatching.

Here are some points to consider and steps to potentially resolve your issues:

1. Parallel Activity Execution: Elsa 3 supports parallel activity execution, which you are trying to use with the Parallel activity. However, the activities being suspended immediately might indicate a configuration issue or a misunderstanding of how the activities should be scheduled and executed.

2. MassTransit and EF Core: Using both MassTransit and EF Core makes sense in your scenario. EF Core is used for persistence, while MassTransit with a message broker like Azure Service Bus enables the distribution of work across multiple instances. This setup should support running activities across multiple service instances.

3. Configuration Check: Ensure that your MassTransit configuration is correct and that the message broker (Azure Service Bus) is properly set up to handle the messages. The prefetch count, concurrent calls, and delivery count should be configured according to your workload and expected concurrency.

4. Activity Execution: The DummyActivity should be configured to run as a Job rather than a Task if you want it to be dispatched to the queue and potentially executed on different nodes. This is because Task activities are designed to execute synchronously within the same process, while Job activities can be dispatched to a queue for distributed execution.

5. Workflow Definition: Ensure that your workflow definition is correct and that the Parallel activity is properly configured to run its child activities in parallel. You might need to adjust the RunAsynchronously property or other settings to achieve the desired behavior.

6. Documentation and Examples: Refer to the official Elsa documentation and any available examples that demonstrate the setup of distributed workflows using MassTransit and EF Core. The documentation might provide insights into any missing configuration or steps in your setup.

7. Debugging and Logging: Add logging to your application to capture more details about the workflow execution and any errors that occur. This can help identify why the activities are being suspended immediately.

8. Community and Support: If you continue to face issues, consider reaching out to the Elsa community or support channels. The maintainers and community members might have encountered similar scenarios and can provide guidance.

In summary, your approach is valid, but you may need to adjust your configuration and ensure that your activities are set up for distributed execution. Review the documentation and examples for distributed workflows with Elsa 3, and consider reaching out to the community for additional support.

[robinabganpat]

Unfortunately, changing the Activity to a Job made no difference. Also, I've tried to remove the configuration for EFCore and MassTransit and attempted to make it work with the bare minimum required configuration of Elsa 3, but still, the activities don't seem to get executed.

[douglasg14b]

You ever figure this out? This bot sometimes seems less than helpful in the discussions I'm reading :/

[Dom1w]

@robinabganpat did you ever find a solution for this?

[robinabganpat]

Unfortunately not. I ended up using Quartz for job scheduling and doing things in parallel using a MSSQL Store.

[GeorgeS2019]

Parallel Task Orchestration with ELSA 3.0
https://sipkeschoorstra.medium.com/parallel-task-orchestration-with-elsa-3-0-e387f863655f

Parallel Task Orchestration is not similar to Microservice Orchestration

[robverhoef-united]

I've got this working. The main thing i found out the the use of "WorkflowStorage" variables.
var variable= builder.WithVariable<string>().WithWorkflowStorage();

But you code is not using any variables so that probably not the problem.

What i also found out, the await _workflowRunner.RunAsync(workflow); will return immediately when one of the activities is marked with RunAsynchronously = true . So the Get in your example will return immediately and will not await the completing of the workflow

Lastely, i prefer to use the Flowchart instead of the Sequence. That will also omit the use op Parallel:

var dummy1  = new DummyActivity() { RunAsynchronously = true, };
var dummy2  = new DummyActivity() { RunAsynchronously = true, };
var dummy3  = new DummyActivity() { RunAsynchronously = true, };
var join = var join = new FlowJoin { Mode = new(FlowJoinMode.WaitAll) };
var startWriteLine = new WriteLine("Starting!");
var endWriteLine = new WriteLine("Done!");

new Flowchart
        {
           Variables = { },
            Activities = { startWriteLine, dummy1, dummy2, dummy3, join, endWriteLine},
            Connections =
            {
                new Connection (new Endpoint(startWriteLine), new Endpoint(dummy1)),
                new Connection (new Endpoint(startWriteLine), new Endpoint(dummy2)),
                new Connection (new Endpoint(startWriteLine), new Endpoint(dummy3)),
                new Connection (new Endpoint(dummy1), new Endpoint(join)),
                new Connection (new Endpoint(dummy2), new Endpoint(join)),
                new Connection (new Endpoint(dummy3), new Endpoint(join)),
                new Connection (new Endpoint(join), new Endpoint(enWriteLine)),
            }
         }

Annoying thing that remains is that await _workflowRunner.RunAsync(workflow); will return immidiatly!! You can grap the result of this and retrieve the workflow result in a loop and check if its completed.

I also use: [Activity("Elsa", "Copy", "Copy files from SourceParh to DestinationPath", Kind = ActivityKind.Task)] in my activities. But i do not think this is of main importance.

Good luck, Rob