ZeroMQ PUB/SUB Modèle Multi-Thread Poller Annulation

j'ai deux applications, un serveur C++ et une interface utilisateur C# WPF. Le code C++ prend les requêtes (de n'importe où/n'importe qui) via un service de messagerie ZeroMQ [PUB/SUB]. J'utilise mon code C# pour le back testing et pour créer des "back tests" et les exécuter. Ces derniers peuvent être constitués de nombreux "tests unitaires" et chacun d'eux envoyant/recevant des milliers de messages du serveur C++.

actuellement, les tests de dos individuels fonctionnent bien et peuvent envoyer N tests unitaires chacun avec des milliers de demandes et capturer. Mon problème est l'architecture; quand j'envoie un autre test arrière (suivant le premier) j'ai un problème avec l'abonnement d'événement étant fait une deuxième fois en raison du fil de sondage n'étant pas annulé et disposé. Il en résulte une sortie erronée. Cela peut sembler une situation banale (c'est peut-être pour certains d'entre vous), mais l'annulation de ce scrutin Tâche sous ma configuration actuelle s'avère gênant. Un peu de code...

ma classe de courtier de message est simple et regarde comme

public class MessageBroker : IMessageBroker<Taurus.FeedMux>, IDisposable
{
    private Task pollingTask;
    private NetMQContext context;
    private PublisherSocket pubSocket;

    private CancellationTokenSource source;
    private CancellationToken token;
    private ManualResetEvent pollerCancelled;

    public MessageBroker()
    {
        this.source = new CancellationTokenSource();
        this.token = source.Token;

        StartPolling();
        context = NetMQContext.Create();
        pubSocket = context.CreatePublisherSocket();
        pubSocket.Connect(PublisherAddress);
    }

    public void Dispatch(Taurus.FeedMux message)
    {
        pubSocket.Send(message.ToByteArray<Taurus.FeedMux>());
    }

    private void StartPolling()
    {
        pollerCancelled = new ManualResetEvent(false);
        pollingTask = Task.Run(() =>
        {
            try
            {
                using (var context = NetMQContext.Create())
                using (var subSocket = context.CreateSubscriberSocket())
                {
                    byte[] buffer = null;
                    subSocket.Options.ReceiveHighWatermark = 1000;
                    subSocket.Connect(SubscriberAddress);
                    subSocket.Subscribe(String.Empty);
                    while (true)
                    {
                        buffer = subSocket.Receive();
                        MessageRecieved.Report(buffer.ToObject<Taurus.FeedMux>());
                        if (this.token.IsCancellationRequested)
                            this.token.ThrowIfCancellationRequested();
                    }
                }
            }
            catch (OperationCanceledException)
            {
                pollerCancelled.Set();
            }
        }, this.token);
    }

    private void CancelPolling()
    {
        source.Cancel();
        pollerCancelled.WaitOne();
        pollerCancelled.Close();
    }

    public IProgress<Taurus.FeedMux> MessageRecieved { get; set; }
    public string PublisherAddress { get { return "tcp://127.X.X.X:6500"; } }
    public string SubscriberAddress { get { return "tcp://127.X.X.X:6501"; } }

    private bool disposed = false;

    protected virtual void Dispose(bool disposing)
    {
        if (!disposed)
        {
            if (disposing)
            {
                if (this.pollingTask != null)
                {
                    CancelPolling();
                    if (this.pollingTask.Status == TaskStatus.RanToCompletion ||
                         this.pollingTask.Status == TaskStatus.Faulted ||
                         this.pollingTask.Status == TaskStatus.Canceled)
                    {
                        this.pollingTask.Dispose();
                        this.pollingTask = null;
                    }
                }
                if (this.context != null)
                {
                    this.context.Dispose();
                    this.context = null;
                }
                if (this.pubSocket != null)
                {
                    this.pubSocket.Dispose();
                    this.pubSocket = null;
                }
                if (this.source != null)
                {
                  this.source.Dispose();
                  this.source = null;
                }
            }
            disposed = true;
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    ~MessageBroker()
    {
        Dispose(false);
    }
}

l'utilisation du "moteur" d'essai arrière pour exécuter chaque essai arrière, construit d'abord un Dictionary contenant chaque Test (test unitaire) et les messages à envoyer à L'application C++ pour chaque test.

DispatchTests méthode, c'est ici

private void DispatchTests(ConcurrentDictionary<Test, List<Taurus.FeedMux>> feedMuxCollection)
{
    broker = new MessageBroker();
    broker.MessageRecieved = new Progress<Taurus.FeedMux>(OnMessageRecieved);
    testCompleted = new ManualResetEvent(false);

    try
    {
        // Loop through the tests. 
        foreach (var kvp in feedMuxCollection)
        {
            testCompleted.Reset();
            Test t = kvp.Key;
            t.Bets = new List<Taurus.Bet>();
            foreach (Taurus.FeedMux mux in kvp.Value)
            {
                token.ThrowIfCancellationRequested();
                broker.Dispatch(mux);
            }
            broker.Dispatch(new Taurus.FeedMux()
            {
                type = Taurus.FeedMux.Type.PING,
                ping = new Taurus.Ping() { event_id = t.EventID }
            });
            testCompleted.WaitOne(); // Wait until all messages are received for this test. 
        }
        testCompleted.Close();
    }
    finally
    {
        broker.Dispose(); // Dispose the broker.
    }
}

PING message à la fin, c'est pour dire au c++ que nous sommes finis. Nous forçons alors une attente, de sorte que le prochain test [unit] ne soit pas envoyé avant que tous les retours ne soient reçus du c++ code - nous faisons cela en utilisant un ManualResetEvent.

quand le c++ reçoit le message de PING, il renvoie le message directement. Nous traitons les messages reçus via OnMessageRecieved et le PING nous dit de mettre le ManualResetEvent.Set() afin que nous puissions poursuivre les tests de l'unité; "suivant S'il vous plaît"...

private async void OnMessageRecieved(Taurus.FeedMux mux)
{
    string errorMsg = String.Empty;
    if (mux.type == Taurus.FeedMux.Type.MSG)
    {
        // Do stuff.
    }
    else if (mux.type == Taurus.FeedMux.Type.PING)
    {
        // Do stuff.

        // We are finished reciving messages for this "unit test"
        testCompleted.Set(); 
    }
}

Mon problème est que, broker.Dispose() dans le a finalement ci-dessus n'est jamais atteint. J'apprécie que finalement les blocs qui sont exécutés sur les threads d'arrière-plan ne sont pas garantis pour obtenir exécuté.

le texte rayé ci-dessus était dû au fait que je m'amusais avec le code; j'arrêtais un thread parent avant que l'enfant n'ait terminé. Cependant, il y a encore des problèmes...

broker.Dispose() s'appelle correctement, et broker.Dispose() est appelé, dans cette méthode je tente d'annuler le fil de poller et dispose de la Task correctement pour éviter tout Abonnement multiple.

pour annuler le thread j'utilise le CancelPolling() méthode

private void CancelPolling()
{
    source.Cancel();
    pollerCancelled.WaitOne(); <- Blocks here waiting for cancellation.
    pollerCancelled.Close();
}

mais dans le StartPolling() méthode

while (true)
{
    buffer = subSocket.Receive();
    MessageRecieved.Report(buffer.ToObject<Taurus.FeedMux>());
    if (this.token.IsCancellationRequested)
        this.token.ThrowIfCancellationRequested();
}

ThrowIfCancellationRequested() n'est jamais appelé et le thread n'est jamais annulé, donc jamais correctement disposé. Le fil de poller est bloqué par le subSocket.Receive() méthode.

maintenant, il ne me semble pas clair comment atteindre ce que je veux, je dois invoquer le broker.Dispose()/PollerCancel() sur un fil autre que celui utilisé pour Poller les messages et certains comment forcer l'annulation. Abandon de Thread n'est pas ce que je veux entrer dans à tout coût.

Essentiellement, je veux disposer correctement des broker avant d'exécuter le prochain test arrière, comment gérer correctement ceci, séparer le sondage et l'exécuter dans un domaine D'Application séparé?

j'ai essayé, se débarrasser à l'intérieur de la OnMessageRecived gestionnaire, mais ce n'est clairement exécuté sur le même thread que le poller et n'est pas la façon de le faire, sans invoquer d'autres threads, il bloque.

Quelle est la meilleure façon d'atteindre ce que l' Je veux et est-il un modèle pour ce genre d'affaire que je peux suivre?

Merci pour votre temps.