2

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

何らかの理由で、ステートマシンは同じ入力を2回受け入れています（状態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の続きです

2番目の並列タスクが受け入れるためにac：

s1はTask<Actor>
s2はs1の続きです
s3はs2の続きです（これはイプシロンの動きです）
s4はs3の続きです

これらのタスクは両方ともActor、メインの先行タスクから継続タスクに渡されるオブジェクトの独自のコピーを持っています。

私はもうすぐそこにいることを知っています、そして私はこの最後の謎を解く必要があります。

4

2 に答える 2

0

はるかに簡単な解決策を思いついた後、私は自分の質問に答えることができました。TPL DataFlow は、計算が完了したかどうかを判断する方法がない循環データ ネットワークを作成するため、これには適していないと推測しました。それで、私はそれを完全にビンに入れ、製図板に戻ることにしました。

私は最終的に、すべてのプロセッサコアを利用して各遷移を並行して実行する、まさに私が望んでいたことを行う Parallell.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 );
        } );
}
于 2012-11-17T14:48:38.377 に答える
0

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 クラス) を投稿する必要があります。

于 2012-11-15T15:34:35.160 に答える