c # — Шаблон ZeroMQ PUB / SUB с многопоточным отменой поллера

У меня есть два приложения, сервер C ++ и пользовательский интерфейс WPF C #. Код C ++ принимает запросы (откуда угодно / кто угодно) через службу сообщений ZeroMQ [PUB / SUB]. Я использую свой код C # для бэк-тестирования и для создания «бэк-тестов» и их выполнения. Эти обратные тесты могут состоять из множества «модульных тестов», каждый из которых отправляет / получает тысячи сообщений с сервера C ++.

В настоящее время отдельные бэк-тесты работают хорошо и могут отправлять N модульных тестов, каждый из которых содержит тысячи запросов и захватов. Моя проблема — архитектура; когда я отправляю еще один бэк-тест (следующий за первым), я получаю проблему с повторной подпиской на события из-за того, что поток опроса не отменен и не удален. Это приводит к ошибочному выводу. Это может показаться тривиальной проблемой (возможно, для некоторых из вас), но отмена этой задачи опроса в моей текущей конфигурации оказывается проблематичной. Какой-то код …

Мой класс брокера сообщений прост и выглядит

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);
}
}

«Движок» тестирования на истории, используемый для выполнения каждого обратного теста, сначала создает Dictionary содержащий каждый Test (модульный тест) и сообщения для отправки в приложение C ++ для каждого теста.

DispatchTests метод, вот он

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 сообщение в конце, это сказать C ++, что мы закончили. Затем мы форсируем ожидание, чтобы следующий тест [unit] не отправлялся до получения всех возвратов из кода C ++ — мы делаем это, используя ManualResetEvent,

Когда C ++ получает сообщение PING, он отправляет сообщение обратно. Мы обрабатываем полученные сообщения через OnMessageRecieved и ПИН говорит нам, чтобы установить ManualResetEvent.Set() чтобы мы могли продолжить юнит-тестирование; «Следующий, пожалуйста»…

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();
}
}

Моя проблема в том, что broker.Dispose() в конце концов выше никогда не ударил. Я ценю, что, наконец, блоки, которые выполняются в фоновых потоках, не гарантированно будут выполнены.

Вычеркнутый текст выше был из-за того, что я возился с кодом; Я останавливал родительский поток до того, как ребенок закончил. Тем не менее, есть еще проблемы …

Сейчас broker.Dispose() называется правильно, и broker.Dispose() в этом методе я пытаюсь отменить поток опроса и избавиться от Task правильно, чтобы избежать каких-либо нескольких подписок.

Чтобы отменить тему я использую CancelPolling() метод

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

но в StartPolling() метод

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

ThrowIfCancellationRequested() никогда не вызывается и поток никогда не отменяется, поэтому никогда не удаляется должным образом. Поток опроса блокируется subSocket.Receive() метод.

Теперь мне не ясно, как добиться того, чего я хочу, мне нужно вызвать broker.Dispose()/PollerCancel() в потоке, отличном от того, который использовался для опроса сообщений, и в некоторых случаях вызывал отмену. Прерывание потока — не то, во что я хочу попасть любой ценой.

По сути, я хочу правильно распоряжаться broker перед выполнением следующего обратного теста, как мне правильно обработать это, разделить опрос и запустить его в отдельном Домене Приложения?

Я пытался, избавляясь от OnMessageRecived обработчик, но он явно выполняется в том же потоке, что и модуль опроса, и это не способ сделать это, без вызова дополнительных потоков он блокируется.

Какой лучший способ добиться того, чего я хочу а также есть ли образец за такой случай, которым я могу следовать?

Спасибо за ваше время.