C# Неизменный счетчик для нескольких полей

Question

У меня довольно высокая пропускная способность для счетчика сообщений (десятки тысяч в секунду), и я ищу эффективный способ получения счетчика без повсеместной блокировки или, в идеале, без блокировки каждого счетчика сообщений, когда я обновляю каждый 10 секунд.

Использование объекта неизменяемого счетчика

Я использую класс неизменяемого счетчика:

public class Counter
{
    public Counter(int quotes, int trades)
    {
        Quotes = quotes;
        Trades = trades;
    }

    readonly public int Quotes;
    readonly public int Trades;
    // and some other counter fields snipped
}

И будет обновлять это при каждом сообщении process l oop:

class MyProcessor
{
    System.Timers.Timer timer;
    Counter counter = new Counter(0,0);

    public MyProcessor()
    {
       // update ever 10 seconds
       this.timer = new System.Timers.Timer(10000);

       timer.Elapsed += (sender, e) => {
          var quotesPerSecond = this.counter.Quotes / 10.0;
          var tradesPerSecond = this.counter.Trades / 10.0;
          this.Counter = new Counter(0,0);
       });
    }

    public void ProcessMessages(Messages messages)
    {
       foreach(var message in messages) { /* */ }

       var oldCounter = counter;
       this.counter = new Counter(oldCounter.Quotes, oldCounter.Trades);   
    }
}

У меня много счетчиков (не все показаны), поэтому будет означать много отдельных вызовов Interlocked.Increment для отдельных полей счетчиков.

Единственное Другой способ, о котором я могу думать, это блокировка каждого запуска ProcessMessages (который будет обширным) и тяжелый для чего-то, что является утилитой, а не критическим, когда программа взломает sh.

Можно ли использовать неизменяемый объект-счетчик таким образом без жестких механизмов блокировки / потоков, когда нам нужно обновлять только раз в 10 секунд?

Идея проверки флага чтобы избежать блокировок

Может ли поток таймера установить флаг для ProcessMessages, чтобы проверить, и если он видит, что он установлен, снова начать отсчет с нуля, то есть

/* snipped the MyProcessor class, same as before */

System.Timers.Timer timer;
Counter counter = new Counter(0,0);
ManualResetEvent reset = new ManualResetEvent(false);

public MyProcessor()
{
   // update ever 10 seconds
   this.timer = new System.Timers.Timer(10000);

   timer.Elapsed += (sender, e) => {
      var quotesPerSecond = this.counter.Quotes / 10.0;
      var tradesPerSecond = this.counter.Trades / 10.0;
      // log
      this.reset.Set();
   });
}

// this should be called every second with a heartbeat message posted to queue
public void ProcessMessages(Messages messages)
{
   if (reset.WaitOne(0) == true)
   {
      this.counter = new Counter(this.counter.Quotes, this.counter.Trades, this.counter.Aggregates);
      reset.Reset();
   }
   else
   {
      this.counter = new Counter(
                        this.counter.Quotes + message.Quotes.Count,
                        this.counter.Trades + message.Trades.Count);
   }
}

/* end of MyProcessor class */

Это сработало бы, однако обновление «останавливается», когда сообщения процесса останавливаются (хотя пропускная способность очень высока, она делает паузу на несколько часов ночью, в идеале должно отображаться фактическое, а не последнее значение).

Одним из способов решения этой проблемы является отправка сообщения пульса на MyProcessor.ProcessMessages() каждую секунду для принудительного внутреннего обновления счетчиков сообщений и последующего сброса, когда установлен reset ManualResetEvent.

Answer 1

Вот три новых метода для вашего Counter класса. Один для чтения последнего значения из указанного c местоположения, один для безопасного обновления указанного c местоположения и один для простого создания нового Counter на основе существующего:

public static Counter Read(ref Counter counter)
{
    return Interlocked.CompareExchange(ref counter, null, null);
}

public static void Update(ref Counter counter, Func<Counter, Counter> updateFactory)
{
    var counter1 = counter;
    while (true)
    {
        var newCounter = updateFactory(counter1);
        var counter2 = Interlocked.CompareExchange(ref counter, newCounter, counter1);
        if (counter2 == counter1) break;
        counter1 = counter2;
    }
}

public Counter Add(int quotesDelta, int tradesDelta)
{
    return new Counter(Quotes + quotesDelta, Trades + tradesDelta);
}

Пример использования:

Counter latest = Counter.Read(ref this.counter);

Counter.Update(ref this.counter, existing => existing.Add(1, 1));

Доступ к полю MyProcessor.counter одновременно несколькими потоками одновременно не является поточно-ориентированным, поскольку он не volatile и не защищен lock. Вышеуказанные методы безопасны в использовании, потому что они получают доступ к полю через операции с блокировкой .

Answer 2

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

Все движется DequeueThread l oop, а в частности this.queue.ReceiveAsync(TimeSpan.FromSeconds(UpdateFrequencySeconds)) function.

Это либо вернет элемент из очереди, обработает его и обновит счетчики, либо истечет время ожидания, а затем обновит счетчики - другие потоки не задействованы, все, включая обновление частоты сообщений, выполняется внутри thread.

Таким образом, ничто не выполняется параллельно (с точки зрения запроса пакета), оно выбирает элементы по одному и обрабатывает их, а затем счетчики. Затем, наконец, возвращается цикл для обработки следующего элемента в очереди.

Это устраняет необходимость в синхронизации:

internal class Counter
{
    public Counter(Action<int,int,int,int> updateCallback, double updateEvery)
    {
        this.updateCallback = updateCallback;
        this.UpdateEvery = updateEvery;
    }

    public void Poll()
    {
        if (nextUpdate < DateTimeOffset.UtcNow)
        {
            // post the stats, and reset
            this.updateCallback(this.quotes, this.trades, this.aggregates, this.statuses);
            this.quotes = 0;
            this.trades = 0;
            this.aggregates = 0;
            this.statuses = 0;
            nextUpdate = DateTimeOffset.UtcNow.AddSeconds(this.UpdateEvery);
        }
    }

    public void AddQuotes(int count) => this.quotes += count;
    public void AddTrades(int count) => this.trades += count;
    public void AddAggregates(int count) => this.aggregates += count;
    public void AddStatuses(int count) => this.statuses += count;

    private int quotes;
    private int trades;
    private int aggregates;
    private int statuses;

    private readonly Action<int,int,int,int> updateCallback;
    public double UpdateEvery { get; private set; }
    private DateTimeOffset nextUpdate;
}

public class DeserializeWorker
{
    private readonly BufferBlock<byte[]> queue = new BufferBlock<byte[]>();
    private readonly IPolygonDeserializer polygonDeserializer;
    private readonly ILogger<DeserializeWorker> logger;

    private readonly Counter counter; 
    const double UpdateFrequencySeconds = 5.0;        
    long maxBacklog = 0;

    public DeserializeWorker(IPolygonDeserializer polygonDeserializer, ILogger<DeserializeWorker> logger)
    {
        this.polygonDeserializer = polygonDeserializer ?? throw new ArgumentNullException(nameof(polygonDeserializer));
        this.logger = logger;
        this.counter = new Counter(ProcesCounterUpdateCallback, UpdateFrequencySeconds);
    }

    public void Add(byte[] data)
    {
        this.queue.Post(data);
    }

    public Task Run(CancellationToken stoppingToken)
    {
        return Task
                .Factory
                .StartNew(
                    async () => await DequeueThread(stoppingToken),
                    stoppingToken,
                    TaskCreationOptions.LongRunning,
                    TaskScheduler.Default)
                .Unwrap();
    }

    private async Task DequeueThread(CancellationToken stoppingToken)
    {
        while (stoppingToken.IsCancellationRequested == false)
        {
            try
            {
                var item = await this.queue.ReceiveAsync(TimeSpan.FromSeconds(UpdateFrequencySeconds), stoppingToken);
                await ProcessAsync(item);
            }
            catch (TimeoutException)
            {
                // this is ok, timeout expired 
            }
            catch(TaskCanceledException)
            {
                break; // task cancelled, break from loop
            }
            catch (Exception e)
            {
                this.logger.LogError(e.ToString());
            }

            UpdateCounters();
        }

        await StopAsync();
    }


    protected async Task StopAsync()
    {
        this.queue.Complete();
        await this.queue.Completion;
    }

    protected void ProcessStatuses(IEnumerable<Status> statuses)
    {
        Parallel.ForEach(statuses, (current) =>
        {
            if (current.Result != "success")
                this.logger.LogInformation($"{current.Result}: {current.Message}");
        });
    }

    protected void ProcessMessages<T>(IEnumerable<T> messages)
    {
        Parallel.ForEach(messages, (current) =>
        {
            // serialize by type T
            // dispatch
        });
    }

    async Task ProcessAsync(byte[] item)
    {
        try
        {
            var memoryStream = new MemoryStream(item);
            var message = await this.polygonDeserializer.DeserializeAsync(memoryStream);

            var messagesTask = Task.Run(() => ProcessStatuses(message.Statuses));
            var quotesTask = Task.Run(() => ProcessMessages(message.Quotes));
            var tradesTask = Task.Run(() => ProcessMessages(message.Trades));
            var aggregatesTask = Task.Run(() => ProcessMessages(message.Aggregates));

            this.counter.AddStatuses(message.Statuses.Count);
            this.counter.AddQuotes(message.Quotes.Count);
            this.counter.AddTrades(message.Trades.Count);
            this.counter.AddAggregates(message.Aggregates.Count);

            Task.WaitAll(messagesTask, quotesTask, aggregatesTask, tradesTask);                                
        }
        catch (Exception e)
        {
            this.logger.LogError(e.ToString());
        }
    }

    void UpdateCounters()
    {
        var currentCount = this.queue.Count;
        if (currentCount > this.maxBacklog)
            this.maxBacklog = currentCount;

        this.counter.Poll();
    }

    void ProcesCounterUpdateCallback(int quotes, int trades, int aggregates, int statuses)
    {
        var updateFrequency = this.counter.UpdateEvery;
        logger.LogInformation(
            $"Queue current {this.queue.Count} (max {this.maxBacklog }), {quotes / updateFrequency} quotes/sec, {trades / updateFrequency} trades/sec, {aggregates / updateFrequency} aggregates/sec, {statuses / updateFrequency} status/sec");
    }
}