parallel task about workflow-core HOT 16 CLOSED

At the moment, the best way is with a paralell foreach... however, would it help if I added a basic .Fork() method?

from workflow-core.

                           StepC  - StepD
                         /                          \
StepA - StepB                            (AND or OR) - StepG
                         \                          /
                           StepE - StepF

I'd like to run StepC and StepE in parallel, and run StepG when StepD and StepF (both/or any one) finished.

How do I do? Thanks a lot.

from workflow-core.

hi @qazq I'm currently working on a feature to support this scenario.
I will try do a deploy to NuGet on the weekend, will comment here when done.

from workflow-core.

Hi @qazq , unfortunately I have not had as much time available as I had hoped to work on this last week... will try for next week. Will keep you posted.

from workflow-core.

Hi @danielgerlag , thank you :)

from workflow-core.

@alexdoan102 @qazq please update to 1.2.6 and see this sample
https://github.com/danielgerlag/workflow-core/tree/master/src/samples/WorkflowCore.Sample13

from workflow-core.

@danielgerlag Thanks a lot.
Except for Join() [WaitAll()], is it possible to add WaitAny()? If I have many sub-tasks do in parallel, I'd like to exit the Parallel() when any one task finished.

from workflow-core.

@qazq Let me see what I can do

from workflow-core.

@qazq
Was WaitAny() method added? Or just Join()?

from workflow-core.

Hi,

I'm also interested in such a feature. For example, I want to have a workflow that creates a task and, if 24h after the assignee has not approved/rejected the task, another one must be create for his manager. At first, I wanted to use "WaitFor" with a timeout but as specified here, it is currently not possible. Therefore, I use the proposed solution:

.Parallel()
    .Do(b1 => b1
        .StartWith<AssignTaskToManagerStep>()
            .Input(step => step.User, wf => wf.UserId)
        .WaitFor(Constants.WorkflowEvent.WaitForApproval, (data, context) => context.Workflow.Id)
            .Output(data => data.MainTaskState, step => step.EventData != null ? (ApprovalState)Enum.Parse(typeof(ApprovalState), step.EventData.ToString()) : ApprovalState.WaitingForApproval)
    )
    .Do(b2 => b2
        .StartWith(c => ExecutionResult.Next())
        .Delay(_ => TimeSpan.FromHours(24))
        .Then<AssignTaskToManagerStep>()
            .Input(step => step.User, wf => wf.UserId)
            .Input(step => step.Level, _ => 2)
        .WaitFor(Constants.WorkflowEvent.WaitForApproval, (data, context) => context.Workflow.Id, null, wf => wf.MainTaskState != ApprovalState.WaitingForApproval)
            .Output(data => data.MainTaskState, step => (ApprovalState)Enum.Parse(typeof(ApprovalState), step.EventData.ToString()))
    )
.Join()

The issue is that the "Parallel" will wait for the "Delay" to be executed meaning that if the initial assignee approves the task 5 min after its creation, the workflow will only continue 23:55 later.

Therefore, it would be nice that when a branch gets executed, all the other get cancelled if we use something else than "Join" (or maybe use "Join" with a "JoinType" enum argument that could be "Any" or "All").

Note that, using a "Any" join type should really cancel the other branch. Indeed, in the example above, if the initial assigne approves the task, another task must not be created 24h even though the workflow continued.

from workflow-core.

@ssougnez I have some ideas on how to solve this, I will post a PR soon

from workflow-core.

@ssougnez in the meantime, try playing around with the "Cancel Condition" that can be applied to a WaitFor as well as the Schedule

https://github.com/danielgerlag/workflow-core/releases/tag/1.3.3

https://github.com/danielgerlag/workflow-core/releases/tag/1.2.8

https://github.com/danielgerlag/workflow-core/blob/master/src/WorkflowCore/Interface/IStepBuilder.cs#L95

from workflow-core.

I could solve it with the approach you mentioned:

.StartWith<AssignTaskToManagerStep>()
    .Input(step => step.User, wf => wf.UserId)
.Schedule(_ => TimeSpan.FromHours(24))
.Do(then => then
    .StartWith(_ => ExecutionResult.Next())
    .If(data => data.MainTaskState == ApprovalState.WaitingForApproval).Do(then2 => then2
        .StartWith<AssignTaskToManagerStep>()
            .Input(step => step.User, wf => wf.UserId)
            .Input(step => step.Level, _ => 2)
        .WaitFor(Constants.WorkflowEvent.WaitForApproval, (data, context) => context.Workflow.Id, null, wf => wf.MainTaskState != ApprovalState.WaitingForApproval)
            .Output(data => data.MainTaskState, step => (ApprovalState)Enum.Parse(typeof(ApprovalState), step.EventData.ToString()))
    )
)
.WaitFor(Constants.WorkflowEvent.WaitForApproval, (data, context) => context.Workflow.Id)
    .Output(data => data.MainTaskState, step => step.EventData != null ? (ApprovalState)Enum.Parse(typeof(ApprovalState), step.EventData.ToString()) : ApprovalState.WaitingForApproval)

It seems to work although I'm not sure it's bullet-proof. Another concern is that it just works around the real issue. Indeed, what if I want to ensure that a task bubbles from manager to manager every 24h as long it is not approved and the top manager is not reached. I should use a "While" loop, however, I guess it wouldn't work with a "Schedule" as the "While" loop would keep on scheduling events (as the "Schedule" is not blocking). The idea would be to use a "While" with "Delay" but as the "Parallel" requires all branches to be terminated to continue, it would't work either.

I'm still learning workflowCore so I might be wrong but it seems that what I just said is not possible, is it?

Sorry if I keep asking questions but I really like the approach of this library and I really want to use it for my project but as it will be used by 2000+ users every day, I need to be sure that the library I use can handle almost every situations and I fear that workflowCore can't (in all due respect). So basically, I ask these questions in the hope that you'll prove me wrong ha ha

Congrats anyway for your work, it's already a pretty good library.

Edit

I tried to manually add an event to the "Event" table with a date in the future for "EventTime" and false for "IsProcessed". This worked but I'm not sure it's a good solution as it would require to create a new event but only first the first occurrence of the workflow (indeed, the event should not be create if the app is restarted). It's just a proposal as apparently you already have an idea how to solve it ;-)

from workflow-core.

see #237
Also, the problem in the above comment could possibly be solve using the new CancelCondition on a parallel branch, that iterates through a stack or queue on the custom Data property?

I would not advise manually inserting into the Event table... why can't you use 'PublishEvent?

from workflow-core.

It was just a test actually, to see if it's possible to use WaitFor with a expiration date. Creating the event upfront with a EventTime set in the future could achieve that but I understand why you wouldn't recommend this hehe

I'll check the new version of wfcore to see if I can achieve this expiration things with the new features ;-)

from workflow-core.

Hi,

I could do what I want with this:

builder
    .StartWith<AssignTaskToManagerStep>()
        .Input(step => step.User, wf => wf.UserId)
    .Parallel()
        .Do(then => then
            .StartWith(_ => ExecutionResult.Next())
            .While(data => data.MainTaskState == ApprovalState.WaitingForApproval)
                .Do(then2 => then2
                    .StartWith(_ => ExecutionResult.Next())
                    .Delay(data => TimeSpan.FromMinutes(2))
                    .Then<AssignTaskToManagerStep>()
                        .Input(step => step.User, wf => wf.UserId)
                        .Input(step => step.Level, _ => 2)
                )
        )
        .Do(then => then
            .StartWith(_ => ExecutionResult.Next())
            .WaitFor(Constants.WorkflowEvent.WaitForApproval, (data, context) => context.Workflow.Id)
                .Output(data => data.MainTaskState, step => step.EventData != null ? (ApprovalState)Enum.Parse(typeof(ApprovalState), step.EventData.ToString()) : ApprovalState.WaitingForApproval)
        )
    .Join()
        .CancelCondition(data => data.MainTaskState != ApprovalState.WaitingForApproval, true)
    .If(data => data.MainTaskState == ApprovalState.Approved).Do(then => then
        .StartWith<SendMailStep>()
            .Input(step => step.To, data => "[email protected]")
            .Input(step => step.Subject, data => "Task approved")
            .Input(step => step.Content, data => "The task has been approved")
    )
    .If(data => data.MainTaskState == ApprovalState.Rejected).Do(then => then
        .StartWith<SendMailStep>()
            .Input(step => step.To, data => "[email protected]")
            .Input(step => step.Subject, data => "Task reject")
            .Input(step => step.Content, data => "The task has been rejected")
    );

It can be improved a lot but it works so thanks a lot ;-)

Now, using CancelCondition is really interesting but wouldn't it be nice to also have an overload of that function that takes a TimeSpan in parameter? I don't know how complex it would be for you but it would allow the user to cancel any kind of step after a certain amount of time which, I think, could be very handy in some case. That's just an idea ;-)

Thanks anyway for the CancelCondition that improves a lot the library :-)

from workflow-core.