使用 TPL 并行运行任务

Question

我有一个简单的（只是一个测试）状态机，它接受以下输入字符串 abc 和 ac。状态机设置如下：

s1 --> 'a' --> s2
s2 --> 'b' --> s3
s3 --> 'c' --> s4
s2 --> s4 (Epsilon transition)

s1是开始状态
s4是接受状态

我想使用 TPL 并行执行s1->s2->s3->s4 和s1->s2->s3->s4（彼此独立）。

如果我传入“abc”作为机器接受的输入，即

> Thread 1 - Consumed: a, from State: 1 to State: 2
> Thread 2 - Consumed: b, from State: 2 to State: 3
> Thread 3 - Epsilon transition from State: 2 to State: 3
> Thread 4 - Consumed: c, from State: 3 to State: 4
> Thread 4 - Accepted in state 4

Time taken = 19

Input 'abc' is valid Press any key to exit

但是，如果我传入“ac”，我会得到：

> Thread 1 - Consumed: a, from State: 1 to State: 2
> Thread 2 - Epsilon transition from State: 2 to State: 3
> Thread 3 - Consumed: c, from State: 3 to State: 4
> Thread 3 - Accepted in state 4
> Thread 4 - Consumed: c, from State: 3 to State: 4
> Thread 4 - Accepted in state 4

Time taken = 39

Input 'ac' is not valid (Reason: RejectedAmbiguous) Press any key to exit

由于某种原因，状态机两次接受相同的输入（在状态 4 中接受），这是不可能的，因为并行执行的两条线都接受不同的输入。

由于代码太多，我不会发布所有代码，但我会发布主要部分，以便您了解我做错了什么。

public enum eResult
{
    Accepted = 0,
    RejectedAmbiguous,
    RejectedNoResults,
    RejectedNoInitialState
}

public eResult Execute()
{
    var startState = States.FirstOrDefault(s => s.Initial);
    if (startState == null) return eResult.RejectedNoInitialState;

    tasks.Clear();

    CancellationTokenSource cts = new CancellationTokenSource();
    Task t = new Task(() =>
        {
            foreach(Transition tr in getTransitions(startState))
            {
                var tr = trans[n];
                var actor = new Actor(tr.FromState, this.input);
                Task<Actor> task = Task<Actor>.Factory.StartNew(obj =>
                    {
                        return doTransitionFunction(tr, cts).Invoke((Actor)obj);
                    }, actor, cts.Token);
                buildContinuationTask(Transitions[tr], task, cts);
                tasks.Add(task);
            }
        }, cts.Token);

    t.RunSynchronously();

    try
    {
        Task.WaitAll(tasks.ToArray());
    }
    catch (AggregateException ae)
    {
        foreach (Exception e in ae.Flatten().InnerExceptions)
        {
            Console.WriteLine(e.Message);
        }
    }

    eResult result = eResult.Accepted;

    if (!results.Any()) result = eResult.RejectedNoResults;
    else if (results.Where(r => r.State.Accepted).Count() > 1) result = eResult.RejectedAmbiguous;

    return result;
}

IEnumerable<Transition> getTransitions(AtomicState state)
{
    return Transitions.Keys.Where(k => k.FromState == state);
}

bool isAccept(Actor parcel)
{
    return (parcel.State.Accepted && parcel.Cursor.EOF());
}

Func<object, Actor> doTransitionFunction(Transition transition, CancellationTokenSource cts)
{
    return new Func<object, Actor>(obj =>
    {
        var ts = (Actor)obj;
        var cur = ts.Cursor.Peek();
        if (transition.Epsilon || transition.Input.Invoke() == cur)
        {
            if (!transition.Epsilon) ts.Cursor.MoveNext();
            ts.State = Transitions[transition];
            OnTransitioned(this, new TransitionedEventArgs(transition.FromState, ts.State, cur, transition.Epsilon, Task.CurrentId));
            if (isAccept(ts))
            {
                OnAccepted(this, new AcceptedEventArgs(ts.State, Task.CurrentId));
                results.Add(ts);
                cts.Cancel();
            }
        }
        return ts;
    });
}

void buildContinuationTask(AtomicState s, Task<Actor> antecedentTask, CancellationTokenSource cts)
{
    var trans = getTransitions(s).ToArray();
    for (int n = 0; n < trans.Count(); n++)
    {
        Transition tr = trans[n];
        Task<Actor> continuation = antecedentTask.ContinueWith<Actor>(antecdent =>
            {
                if (!cts.IsCancellationRequested)
                    return doTransitionFunction(tr, cts).Invoke((Actor)antecdent.Result.Clone());
                else
                    return (Actor)antecdent.Result.Clone();
            }, cts.Token, TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Current);
        buildContinuationTask(Transitions[tr], continuation, cts);
        tasks.Add(continuation);
    }
}

如果这是不可能的，请纠正我，但我想要发生的是：

第一个并行任务接受abc 作为输入：

s1 是Task<Actor>
s2 是 s1 的延续
s3 是 s2 的延续
s4 是 s3 的延续

第二个并行任务接受ac：

s1 是Task<Actor>
s2 是 s1 的延续
s3 是 s2 的延续（这是 epsilon 移动）
s4 是 s3 的延续

这两个任务都有自己的 Actor 对象副本，该对象将从主要的先行任务传递到后续任务。

我知道我快到了，我只需要解开这最后一个谜。

Answer 1

在阅读了有关 TPL DataFlow 的信息后，这是我的尝试，它似乎按我想要的方式工作。

public interface IScrollableCursor
{
    void MoveNext();
    void MovePrevious();
    void MoveFirst();
    void MoveLast();
    bool BOF();
    bool EOF();
    char Peek();
    int CurrentPosition { get; }
}

[Serializable]
public abstract class AtomicState
{
    protected int stateId;
    protected bool accepted;

    public int StateId
    {
        get
        {
            return stateId;
        }
    }

    public AtomicState( int stateId )
    {
        this.stateId = stateId;
        this.accepted = false;
    }

    public AtomicState( int stateId, bool accepted )
        : this( stateId )
    {
        this.accepted = accepted;
    }

    public abstract bool Initial { get; }

    public bool Accepted
    {
        get
        {
            return accepted;
        }
    }

}


[Serializable]
public struct Actor : ICloneable
{
    private AtomicState state;
    private IScrollableCursor cursor;

    public AtomicState State
    {
        get
        {
            return state;
        }
        set
        {
            state = value;
        }
    }

    public IScrollableCursor Cursor
    {
        get
        {
            return cursor;
        }
    }

    public Actor( AtomicState state, IScrollableCursor cursor )
    {
        this.state = state;
        this.cursor = cursor;
    }

    public object Clone()
    {
        return this.DeepClone();
    }


}

public class Transition
{
    protected AtomicState fromState;
    protected Func<Char> input;
    protected bool epsilon;

    public AtomicState FromState
    {
        get
        {
            return fromState;
        }
    }

    public Func<Char> Input
    {
        get
        {
            return input;
        }
    }

    public bool Epsilon
    {
        get
        {
            return epsilon;
        }
    }

    public Transition( AtomicState fromState, Func<Char> input )
    {
        this.fromState = fromState;
        this.input = input;
    }

    public Transition( AtomicState fromState, bool epsilon )
        : this( fromState, null )
    {
        this.epsilon = epsilon;
    }


}

public class EpsilonTransition : Transition
{
    public EpsilonTransition( AtomicState fromState )
        : base( fromState, true )
    {
    }
}



public eResult Execute()
{
    var startState = States.FirstOrDefault( s => s.Initial );
    if ( startState == null ) return eResult.RejectedNoInitialState;

    tasks.Clear();

    CancellationTokenSource cts = new CancellationTokenSource();

    ExecutionDataflowBlockOptions options = new ExecutionDataflowBlockOptions();
    options.MaxDegreeOfParallelism = 4;
    options.CancellationToken = cts.Token;

    // transitions an actor onto it's next state
    TransformBlock<Tuple<Transition, Actor>, Actor> actorTransitioner = new TransformBlock<Tuple<Transition, Actor>, Actor>( tr =>
        {
            return doTransitionFunction( tr.Item1, cts ).Invoke( tr.Item2 );

        }, options );

    BroadcastBlock<Actor> actorTransitionerBroadcaster = new BroadcastBlock<Actor>( a => { return a; } );

    ActionBlock<Actor> actorProcessor = new ActionBlock<Actor>( a =>
        {
            foreach ( Transition t in getTransitions( a.State ) )
            {
                actorTransitioner.Post( new Tuple<Transition, Actor>( t, (Actor)a.Clone() ) );
            }
        } );

    // link blocks
    actorTransitioner.LinkTo( actorTransitionerBroadcaster );
    actorTransitionerBroadcaster.LinkTo( actorProcessor );

    actorTransitionerBroadcaster.Post( new Actor( startState, input ) );

    try
    {
        actorTransitioner.Completion.Wait();
    }
    catch ( AggregateException ex )
    {
        foreach ( Exception ae in ex.Flatten().InnerExceptions )
        {
            Console.WriteLine( ae.Message );
        }
    }

    eResult result = eResult.Accepted;

    if ( !results.Any() ) result = eResult.RejectedNoResults;
    else if ( results.Where( r => r.State.Accepted ).Count() > 1 ) result = eResult.RejectedAmbiguous;

    return result;
}

我已经添加了我尝试中使用的构造，因此更容易被复制。有必要我会发布整个代码（大约 12 个类）。

Answer 2

在想出一个更简单的解决方案后，我设法回答了我自己的问题。我推断 TPL DataFlow 不适合这种情况，因为它会创建循环数据网络，无法确定计算是否已完成。所以我决定把它完全装箱，然后回到绘图板上。

我最终发现了 Paralell.ForEach()，它完全符合我的要求，利用所有处理器内核并行运行每个转换：

public eResult Execute()
{
    var startState = States.FirstOrDefault( s => s.Initial );
    if ( startState == null ) return eResult.RejectedNoInitialState;

    CancellationTokenSource cts = new CancellationTokenSource();
    CancellationToken token = cts.Token;

    Task t = new Task( () =>
        {
            Parallel.ForEach( getTransitions( startState ), new ParallelOptions { MaxDegreeOfParallelism = 4 }, tr =>
            {
                var a0 = new Actor( tr.FromState, (IScrollableCursor)this.input.DeepClone() );
                var a1 = doTransitionFunction( tr, cts ).Invoke( a0 );
                if ( a0.State != a1.State )
                    processRecursively( a1.State, a0, cts );

            } );

        }, cts.Token );

    t.RunSynchronously();


    eResult result = eResult.Accepted;

    if ( !results.Any() ) result = eResult.RejectedNoResults;
    else if ( results.Where( r => r.State.Accepted ).Count() > 1 ) result = eResult.RejectedAmbiguous;

    return result;


}


void processRecursively( AtomicState s, Actor a0, CancellationTokenSource cts )
{
    Parallel.ForEach( getTransitions( s ), tr =>
        {
            var a1 = doTransitionFunction( tr, cts ).Invoke( a0 );
            if ( a0.State != a1.State )
                processRecursively( a1.State, a1, cts );
        } );
}