C# - отложено выполнение для коллекции предметов

Question

У меня есть коллекция из 100 000 предметов. Предмет имеет 2 свойства. (1) метка времени и (2) данные. Каждый элемент в коллекции имеет отметку времени, которая на несколько (меняющихся) миллисекунд равна или больше, чем предыдущие элементы. Мне нужно отправить эти элементы в конечную точку TCP, поддерживая задержку между ними. ie, если разница во времени между 1-м и 2-м пунктом составляет 300 мс, то после отправки первого элемента мне следует подождать 300 мс, а затем отправить второй элемент.

После нахождения разницы во времени между последовательными элементами я попытался Thread.Sleep. Для списка с разницей во времени между первым и последним элементом, равным 40 секундам, с Thread.Sleep для отправки всех элементов потребовалось около 70-80 секунд.

DateTime currenTimeStamp;
DateTime nextTimeStamp;
TimeSpan timeDiff;

using (NetworkStream stream = client.GetStream())
{
    for (int i = 0; i < count; i++)
    {
        //some lines here to do the TCP transfer

        currenTimeStamp = //get current item's time stamp;
        nextTimeStamp = //get next item's time stamp;
        timeDiff = nextTimeStamp - currenTimeStamp;
        Thread.Sleep(timeDiff);
    }
}

Позже я попытался подождать некоторое время l oop, сравнивая текущее время и ожидаемое время отправки. На обработку всех предметов уходило около 50-55 секунд. Также процессор все время занят, пока l oop.

DateTime currenTimeStamp;
DateTime nextTimeStamp;
DateTime TimeToSendNextItem;
TimeSpan timeDiff;

using (NetworkStream stream = client.GetStream())
{
    for (int i = 0; i < count; i++)
    {
        //some lines here to do the TCP transfer

        currenTimeStamp = //get current item's time stamp;
        nextTimeStamp = //get next item's time stamp;
        timeDiff = nextTimeStamp - currenTimeStamp;
        TimeToSendNextItem = DateTime.UtcNow.Add(timeDiff);
        while (DateTime.UtcNow < TimeToSendNextItem) { }
    }
}

Как я могу справиться с этим с минимальной задержкой и эффективно? Под минимальной задержкой я имею в виду отправку этих элементов в течение времени, близкого к исходной разнице во времени между первым и последним элементом, равной 40 секундам.

Answer 1

Если вы используете самую последнюю версию C# 8.0, вы можете использовать IAsyncEnumerable для предоставления вам необходимых вам предметов:

private async IAsyncEnumerable<Item> GetItemAfterDelay(IEnumerable<Item> items)
{
    DateTime previousItemTimestamp = items.First().Timestamp;

    foreach(var i in items)
    {
        var delay = i.Timestamp - previousItemTimestamp;

        await Task.Delay(delay);

        yield return i;

        previousItemTimestamp = i.Timestamp;
    }
}

ПРИМЕЧАНИЕ. Я создал класс под названием Item для представления элемента вашей коллекции.

struct Item
{
    public DateTime Timestamp { get; set; }
    public string Data { get; set; }
}

Этот метод возвращает один элемент из массива после разницы между текущей отметкой времени и предыдущей.

Затем можно использовать итерацию по элементам вернулся и выполните передачу TCP.

using (NetworkStream stream = client.GetStream())
{
    // assuming your collection is called `myCollection`
    await foreach(var item in GetItemAfterDelay(myCollection))
    {
        //some lines here to do the TCP transfer
    }
}

Answer 2

Ключ в том, чтобы измерить прошедшее время с первого отправленного вами элемента, а затем убедиться, что ни один элемент не был обработан досрочно. Поскольку другие факторы вне вашего контроля могут вызвать задержки обработки. (при условии c # 8);

private async IAsyncEnumerable<Item> GetItemAfterDelay(IEnumerable<Item> items)
{
    var e = items.GetEnumerator();
    if (e.MoveNext()){
       var started = DateTime.Now;
       var firstTime = e.Current.Timestamp;
       yield return e.Current;
       while(e.MoveNext()){
          var delay = (e.Current.Timestamp - firstTime) - (DateTime.Now - started);
          if (delay >0)
             await Task.Delay(delay);
          yield return e.Current;
       }
    }
}

await foreach(var item in GetItemAfterDelay(items)){
   // ...
}

Answer 3

Возможно, вам придется учитывать //some lines here to do the TCP transfer время работы. (Если они синхронны.)

Я буду использовать ваш первый фрагмент кода, например.

DateTime currenTimeStamp;
DateTime nextTimeStamp;
TimeSpan timeDiff;

using (NetworkStream stream = client.GetStream())
{
    for (int i = 0; i < count; i++)
    {
        var start = DateTime.Now;
        //some lines here to do the TCP transfer
        var operationDiff = DateTime.Now - start;
        currenTimeStamp = //get current item's time stamp;
        nextTimeStamp = //get next item's time stamp;
        timeDiff = nextTimeStamp - currenTimeStamp - operationDiff;
        Thread.Sleep(timeDiff);
    }
}

Редактировать: Поскольку у вас есть 100 000 элементов, фактическое время работы на 10 ~ 15 больше чем ожидалось. Это 0,1 мс на элемент, отредактируйте ваше сообщение для //some lines here to do the TCP transfer, если вам нужна дополнительная помощь.

Также точность Thread.Sleep зависит от аппаратных часов. Если вам нужна более точная синхронизация, вам может потребоваться библиотека в реальном времени и определенное c оборудование (для часов).

Использование while задает процесс «не останавливайтесь, продолжайте проверять», поэтому он будет займет меньше времени на работу.

Answer 4

Попробуйте этот код, используя таймеры.

        class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine("Adding elements to list");
            List<WorkItem> workItems = new List<WorkItem>(100);
            for (int i = 0; i < 1000; i++)
            {
                workItems.Add(new WorkItem
                {
                    TimeStamp = DateTime.Now.AddMilliseconds(i * 300)
                });
            }
            var tw = new TimedWorker();
            tw.Process(workItems);

            Console.ReadLine();
        }
    }
    class TimedWorker
    {
        private Timer _timer;
        private Queue<WorkItem> _workItems;
        public void Process(List<WorkItem> workItems)
        {
            _timer = new Timer
            {
                AutoReset = false,
                Interval = 1
            };
            _workItems = new Queue<WorkItem>();
            _timer.Elapsed += Elapsed;
            foreach (var item in workItems)
            {
                _workItems.Enqueue(item);
            }
            _timer.Start();
        }

        private void Elapsed(object sender, ElapsedEventArgs e)
        {
            ProcessNext();
        }

        private void ProcessNext()
        {
            var item = _workItems.Dequeue();
            var nextItem = _workItems.Peek();
            _timer.Interval = (nextItem.TimeStamp - item.TimeStamp).TotalMilliseconds;
            Console.WriteLine(item.TimeStamp.ToString("hh/mm/ss:FFFFFF"));
            _timer.Start();
        }
    }

    class WorkItem
    {
        public DateTime TimeStamp;
    }

Answer 5

Помните, что когда вы используете класс Thread, операционная система контролирует процесс, который выполняется, вы не можете контролировать время выполнения, сказав, что я не думаю, что вы можете решить проблему с помощью класса Thread.