Как использовать TaskCompletionSource - PullRequest
Новая
       76

Как использовать TaskCompletionSource

0 голосов
/ 29 декабря 2018

У меня ситуация с несколькими производителями и одним потребителем. Я выбрал общий ресурс thread-safe, в котором все производители Enqueue товаров.Однако я не знаю, как эффективно сделать производителя await для новых предметов при Dequeue -ing из этого ресурса.

POCO 

struct Sample
    {
        public int Id { get; set; }
        public  double Value { get; set; }
    }

Производители 

class ProducerGroup
    {
        StorageQueue sink;
        int producerGroupSize;

        public ProducerGroup(StorageQueue _sink,int producers)
        {
            this.sink = _sink;
            this.producerGroupSize = producers;
        }
        public void StartProducers()
        {
            Task[] producers = new Task[producerGroupSize];
            int currentProducer;
            for (int i = 0; i < producerGroupSize; i++)
            {
                currentProducer = i;
                producers[i] = Task.Run(async () =>
                  {
                      int cycle = 0;
                      while (true)
                      {
                          if (cycle > 5)
                          {
                              cycle = 0;
                          }
                          Sample localSample = new Sample { Id = currentProducer, Value = cycle++ };
                          await Task.Delay(1000);
                          this.sink.Enqueue(localSample);
                      }
                  });
            }

        }
    }

Хранение 

class StorageQueue
    {
        private TaskCompletionSource<Sample> tcs;
        private object @lock = new object();

        private Queue<Sample> queue;

        public static StorageQueue CreateQueue(int?size=null)
        {
            return new StorageQueue(size);
        }

        public StorageQueue(int ?size)
        {
            if (size.HasValue)
            {
                this.queue = new Queue<Sample>(size.Value);
            }
            else
                this.queue = new Queue<Sample>();
        }

        public void Enqueue(Sample value)
        {
            lock (this.@lock)
            {
                this.queue.Enqueue(value);
                tcs = new TaskCompletionSource<Sample>();
                tcs.SetResult(this.queue.Peek());
            }
        }
        public async Task<Sample> DequeueAsync()
        {
            var result=await this.tcs.Task;
            this.queue.Dequeue();
            return result;
        }
    }

Потребитель 

class Consumer
    {
        private StorageQueue source;
        public Consumer(StorageQueue _source)
        {
            this.source = _source;
        }
        public async Task ConsumeAsync()
        {
            while (true)
            {
                Sample arrivedSample = await this.source.DequeueAsync();  //how should i implement this ?
                Console.WriteLine("Just Arrived");
            }
        }
    }

Как вы можетесм. в Consumer классе, я хотел бы обернуть мою Storage's method Dequeue in a Task so that i can await it in my Consumer . The only reason i used TaskCompletionSource is to be able to communicate between the Dequeue and Enqueue methods in the Storage`.

Я не знаю, нужно ли мне повторно инициализировать tcs, но я полагаю, что я делаю, потому что я хочу новый Task после каждой Enqueue операции.

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

Как мне поступить с этим?Реализация в порядке?System.Reactive предложит лучший вариант?

1 Ответ

0 голосов
/ 30 декабря 2018

Мне кажется, я вижу несколько проблем с вашей реализацией:

  1. Если вы наберете Enqueue() несколько раз без вызова DequeueAsync(), вы потеряете TaskCompletionSource и получите только последнюю.Тогда вызов DequeueAsync() не даст правильных результатов после первого вызова.

Чтобы это исправить, вам понадобится очередь TaskCompletionSource для этого.Взгляните здесь и далее здесь .

Лучше используйте SemaphoreSlim.WaitAsync () в DequeueAsync для правильного ожидания, если очередьпуст.

Вы также не защищаете свою очередь в DequeueAsync().
...