Эффективность таймера

РЕДАКТИРОВАТЬ: я помню интересное интервью, которое определенно стоит посмотреть: Арун Кишан: Внутри Windows 7 — прощание с блокировкой диспетчера ядра Windows

Как заявил @Steven Sudit, я снова предупреждаю: используйте его только в качестве демонстрации того, как работает колесо таймера, и некоторых задач, о которых вам нужно заботиться при его реализации. Не как эталонная реализация. В реальном мире вам нужно написать гораздо более сложную логику, чтобы учесть доступные ресурсы, логику планирования и т. д.

Вот хорошие моменты, изложенные Стивеном Судитом (подробности читайте в комментариях к посту):

1) Выберите правильную структуру, чтобы сохранить список вакансий (как обычно зависит):

• SortedList‹> (или SortedDictionary‹>) хорошо потребляет память и индексирует, но должен реализовать синхронизированный доступ

• ConcurrentQueue‹> поможет вам избежать блокировки, но вы должны реализовать порядок. Это также очень эффективно памяти

• LinkedList‹> хорош для вставки и извлечения (в любом случае нам нужен только заголовок), но требует синхронизированного доступа (благодаря тому, что он легко реализуется через безблокировку) и не так эффективно использует память, поскольку хранит две ссылки (предыдущая/следующая). Но это становится проблемой, когда у вас есть миллионы заданий, поэтому все они занимают значительный объем памяти.

Но:

Я полностью согласен с @Steven:

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

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

2) Чтобы упростить логику обработки одновременных заданий, вы можете добавить список делегатов (например, через ConcurrentQueue, чтобы сделать его свободным от блокировки) в исходный класс Job, поэтому, когда вам нужно другое задание в то же время, вы просто добавляете другого делегата для запуска.

@Стивен:

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

3) Запуск/остановка диспетчера не так проста, как может быть, и поэтому может привести к ошибкам. Вместо этого вы можете ожидать события, используя тайм-аут.

@Стивен:

Таким образом, он либо проснется, когда будет готово следующее задание, либо когда новое задание будет вставлено перед головкой. В последнем случае может потребоваться запустить его сейчас или установить другое время ожидания. Если представить, скажем, 100 заданий, запланированных на одно и то же мгновение, лучшее, что мы можем сделать, — это поставить их все в очередь.

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

4) Об использовании тиков:

@Стивен:

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

Мои мысли:

Еще один хороший момент, я согласен с вами. Но иногда деление на константу становится затратным и не таким быстрым, как может показаться. Но когда мы говорим о 100 000 DateTimes, это не имеет значения, вы правы, спасибо за точку.

5) «Управление ресурсами»:

@Стивен:

Проблема, которую я пытаюсь подчеркнуть, заключается в том, что вызов GetAvailableThreads затратен и наивен; ответ устарел еще до того, как вы сможете его использовать. Если бы мы действительно хотели отслеживать, мы могли бы получить начальные значения и вести текущий подсчет, вызвав задание из оболочки, которая использует Interlocked.Increment/Decrement. Даже в этом случае предполагается, что остальная часть программы не использует пул потоков. Если нам действительно нужен точный контроль, то правильным ответом здесь будет создание собственного пула потоков.

Я абсолютно согласен с тем, что обращение к GetAvailableThreads — это наивный метод мониторинга доступных ресурсов через CorGetAvailableThreads, не такой дорогой. Я хочу продемонстрировать, что есть необходимость в управлении ресурсами и, кажется, выбрал плохой пример.

Любые средства, представленные в примере с исходным кодом, не должны рассматриваться как правильный способ мониторинга доступных ресурсов. Я просто хочу показать, что вы должны думать об этом. Возможно, закодирован не такой хороший фрагмент кода, как пример.

6) Использование Interlocked.CompareExchange:

@Стивен:

Нет, это не обычная схема. Самый распространенный шаблон — кратковременная блокировка. Менее распространено помечать переменную как изменчивую. Гораздо менее распространенным было бы использование VolatileRead или MemoryBarrier. Использование Interlocked.CompareExchange таким образом неясно, даже если это делает Рихтер. использование его без поясняющего комментария абсолютно гарантированно приведет к путанице, поскольку слово «Сравнить» подразумевает, что мы проводим сравнение, хотя на самом деле это не так.

Вы правы, я должен указать на его использование.

using System;
using System.Threading;

// Job.cs

// WARNING! Your jobs (tasks) have to be ASYNCHRONOUS or at least really short-living
// else it will ruin whole design and ThreadPool usage due to potentially run out of available worker threads in heavy concurrency

// BTW, amount of worker threads != amount of jobs scheduled via ThreadPool
// job may waits for any IO (via async call to Begin/End) at some point 
// and so free its worker thread to another waiting runner

// If you can't achieve this requirements then just use usual Thread class
// but you will lose all ThreadPool's advantages and will get noticeable overhead

// Read http://msdn.microsoft.com/en-us/magazine/cc164139.aspx for some details

// I named class "Job" instead of "Task" to avoid confusion with .NET 4 Task 
public class Job
{
    public DateTime FireTime { get; private set; }

    public WaitCallback DoAction { get; private set; }
    public object Param { get; private set; }

    // Please use UTC datetimes to avoid different timezones problem
    // Also consider to _never_ use DateTime.Now in repeat tasks because it significantly slower 
    // than DateTime.UtcNow (due to using TimeZone and converting time according to it)

    // Here we always work with with UTC
    // It will save you a lot of time when your project will get jobs (tasks) posted from different timezones
    public static Job At(DateTime fireTime, WaitCallback doAction, object param = null)
    {
        return new Job {FireTime = fireTime.ToUniversalTime(), DoAction = doAction, Param = param};
    }

    public override string ToString()
    {
        return string.Format("{0}({1}) at {2}", DoAction != null ? DoAction.Method.Name : string.Empty, Param,
                             FireTime.ToLocalTime().ToString("o"));
    }
}

 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
 using System.Threading;

// Dispatcher.cs

// Take a look at System.Runtime IOThreadTimer.cs and IOThreadScheduler.cs
// in Microsoft Reference Source, its interesting reading

public class Dispatcher
{
    // You need sorted tasks by fire time. I use Ticks as a key to gain some speed improvements during checks
    // There are maybe more than one task in same time
    private readonly SortedList<long, List<Job>> _jobs;

    // Synchronization object to access _jobs (and _timer) and make it thread-safe
    // See comment in ScheduleJob about locking
    private readonly object _syncRoot;

    // Queue (RunJobs method) is running flag
    private int _queueRun;

    // Flag to prevent pollute ThreadPool with many times scheduled JobsRun
    private int _jobsRunQueuedInThreadPool;

    // I'll use Stopwatch to measure elapsed interval. It is wrapper around QueryPerformanceCounter
    // It does not consume any additional resources from OS to count

    // Used to check how many OS ticks (not DateTime.Ticks!) elapsed already
    private readonly Stopwatch _curTime;

    // Scheduler start time. It used to build time delta for job
    private readonly long _startTime;

    // System.Threading.Timer to schedule next active time
    // You have to implement syncronized access as it not thread-safe
    // http://msdn.microsoft.com/en-us/magazine/cc164015.aspx
    private readonly Timer _timer;

    // Minimum timer increment to schedule next call via timer instead ThreadPool
    // Read http://www.microsoft.com/whdc/system/pnppwr/powermgmt/Timer-Resolution.mspx
    // By default it around 15 ms
    // If you want to know it exactly use GetSystemTimeAdjustment via Interop ( http://msdn.microsoft.com/en-us/library/ms724394(VS.85).aspx )
    // You want TimeIncrement parameter from there
    private const long MinIncrement = 15 * TimeSpan.TicksPerMillisecond;

    // Maximum scheduled jobs allowed per queue run (specify your own suitable value!)
    // Scheduler will add to ThreadPool queue (and hence count them as processed) no more than this constant

    // This is balance between how quick job will be scheduled after it time elapsed in one side, and 
    // how long JobsList will be blocked and RunJobs owns its thread from ThreadPool
    private const int MaxJobsToSchedulePerCheck = 10;

    // Queue length
    public int Length
    {
        get
        {
            lock (_syncRoot)
            {
                return _jobs.Count;
            }
        }
    }

    public Dispatcher()
    {
        _syncRoot = new object();

        _timer = new Timer(RunJobs);

        _startTime = DateTime.UtcNow.Ticks;
        _curTime = Stopwatch.StartNew();

        _jobs = new SortedList<long, List<Job>>();
    }


    // Is dispatcher still working
    // Warning! Queue ends its work when no more jobs to schedule but started jobs can be still working
    public bool IsWorking()
    {
        return Interlocked.CompareExchange(ref _queueRun, 0, 0) == 1;
    }

    // Just handy method to get current jobs list
    public IEnumerable<Job> GetJobs()
    {
        lock (_syncRoot)
        {
            // We copy original values and return as read-only collection (thread-safety reasons)
            return _jobs.Values.SelectMany(list => list).ToList().AsReadOnly();
        }
    }

    // Add job to scheduler queue (schedule it)
    public void ScheduleJob(Job job)
    {
        // WARNING! This will introduce bottleneck if you have heavy concurrency. 
        // You have to implement lock-free solution to avoid botleneck but this is another complex topic.
        // Also you can avoid lock by using Jeffrey Richter's ReaderWriterGateLock (http://msdn.microsoft.com/en-us/magazine/cc163532.aspx)
        // But it can introduce significant delay under heavy load (due to nature of ThreadPool)
        // I recommend to implement or reuse suitable lock-free algorithm. 
        // It will be best solution in heavy concurrency (if you have to schedule large enough job count per second)
        // otherwise lock or maybe ReaderWriterLockSlim is cheap enough
        lock (_syncRoot)
        {
            // We'll shift start time to quick check when it pasts our _curTime
            var shiftedTime = job.FireTime.Ticks - _startTime;

            List<Job> jobs;
            if (!_jobs.TryGetValue(shiftedTime, out jobs))
            {
                jobs = new List<Job> {job};
                _jobs.Add(shiftedTime, jobs);
            }
            else jobs.Add(job);


            if (Interlocked.CompareExchange(ref _queueRun, 1, 0) == 0)
            {
                // Queue not run, schedule start
                Interlocked.CompareExchange(ref _jobsRunQueuedInThreadPool, 1, 0);
                ThreadPool.QueueUserWorkItem(RunJobs);
            }
            else 
            {
                // else queue already up and running but maybe we need to ajust start time
                // See detailed comment in RunJobs

                long firetime = _jobs.Keys[0];
                long delta = firetime - _curTime.Elapsed.Ticks;

                if (delta < MinIncrement)
                {
                    if (Interlocked.CompareExchange(ref _jobsRunQueuedInThreadPool, 1, 0) == 0)
                    {
                        _timer.Change(Timeout.Infinite, Timeout.Infinite);
                        ThreadPool.QueueUserWorkItem(RunJobs);
                    }
                }
                else 
                {
                    Console.WriteLine("DEBUG: Wake up time changed. Next event in {0}", TimeSpan.FromTicks(delta));
                    _timer.Change(delta/TimeSpan.TicksPerMillisecond, Timeout.Infinite);
                }
            }

        }
    }


    // Job runner
    private void RunJobs(object state)
    {
        // Warning! Here I block list until entire process done, 
        // maybe better will use ReadWriterLockSlim or somewhat (e.g. lock-free)
        // as usually "it depends..."

        // Here processing is really fast (a few operation only) so until you have to schedule many jobs per seconds it does not matter
        lock (_syncRoot)
        {
            // We ready to rerun RunJobs if needed
            Interlocked.CompareExchange(ref _jobsRunQueuedInThreadPool, 0, 1);

            int availWorkerThreads;
            int availCompletionPortThreads;

            // Current thread stats
            ThreadPool.GetAvailableThreads(out availWorkerThreads, out availCompletionPortThreads);

            // You can check max thread limits by
            // ThreadPool.GetMaxThreads(out maxWorkerThreads, out maxCompletionPortThreads);

            int jobsAdded = 0;

            while (jobsAdded < MaxJobsToSchedulePerCheck && availWorkerThreads > MaxJobsToSchedulePerCheck + 1 && _jobs.Count > 0)
            {
                // SortedList<> implemented as two arrays for keys and values so indexing on key/value will be fast
                // First element
                List<Job> curJobs = _jobs.Values[0];
                long firetime = _jobs.Keys[0];

                // WARNING! Stopwatch ticks are different from DateTime.Ticks
                // so we use _curTime.Elapsed.Ticks instead of _curTime.ElapsedTicks

                // Each tick in the DateTime.Ticks value represents one 100-nanosecond interval. 
                // Each tick in the ElapsedTicks value represents the time interval equal to 1 second divided by the Frequency.
                if (_curTime.Elapsed.Ticks <= firetime) break;

                while (curJobs.Count > 0 &&  jobsAdded < MaxJobsToSchedulePerCheck && availWorkerThreads > MaxJobsToSchedulePerCheck + 1)
                {
                    var job = curJobs[0];

                    // Time elapsed and we ready to start job
                    if (job.DoAction != null)
                    {
                        // Schedule new run

                        // I strongly recommend to look at new .NET 4 Task class because it give superior solution for managing Tasks
                        // e.g. cancel run, exception handling, continuation, etc
                        ThreadPool.QueueUserWorkItem(job.DoAction, job);
                        ++jobsAdded;

                        // It may seems that we can just decrease availWorkerThreads by 1 
                        // but don't forget about started jobs they can also consume ThreadPool's threads
                        ThreadPool.GetAvailableThreads(out availWorkerThreads, out availCompletionPortThreads);
                    }

                    // Remove job from list of simultaneous jobs
                    curJobs.Remove(job);
                }

                // Remove whole list if its empty
                if (curJobs.Count < 1) _jobs.RemoveAt(0);
            }

            if (_jobs.Count > 0)
            {
                long firetime = _jobs.Keys[0];

                // Time to next event
                long delta = firetime - _curTime.Elapsed.Ticks; 

                if (delta < MinIncrement) 
                {
                    // Schedule next queue check via ThreadPool (immediately)
                    // It may seems we start to consume all resouces when we run out of available threads (due to "infinite" reschdule)
                    // because we pass thru our while loop and just reschedule RunJobs
                    // but this is not right because before RunJobs will be started again
                    // all other thread will advance a bit and maybe even complete its task
                    // so it safe just reschedule RunJobs and hence wait when we get some resources
                    if (Interlocked.CompareExchange(ref _jobsRunQueuedInThreadPool, 1, 0) == 0)
                    {
                        _timer.Change(Timeout.Infinite, Timeout.Infinite);
                        ThreadPool.QueueUserWorkItem(RunJobs);
                    }
                }
                else // Schedule next check via timer callback
                {
                    Console.WriteLine("DEBUG: Next event in {0}", TimeSpan.FromTicks(delta)); // just some debug output
                    _timer.Change(delta / TimeSpan.TicksPerMillisecond, Timeout.Infinite);
                }
            }
            else // Shutdown the queue, no more jobs
            {
                Console.WriteLine("DEBUG: Queue ends");
                Interlocked.CompareExchange(ref _queueRun, 0, 1); 
            }
        }
    }
}

Краткий пример использования:

    // Test job worker
    static void SomeJob(object param)
    {
        var job = param as Job;
        if (job == null) return;

        Console.WriteLine("Job started: {0}, [scheduled to: {1}, param: {2}]", DateTime.Now.ToString("o"),
                          job.FireTime.ToLocalTime().ToString("o"), job.Param);
    }

    static void Main(string[] args)
    {
        var curTime = DateTime.UtcNow;
        Console.WriteLine("Current time: {0}", curTime.ToLocalTime().ToString("o"));
        Console.WriteLine();

        var dispatcher = new Dispatcher();

        // Schedule +10 seconds to future
        dispatcher.ScheduleJob(Job.At(curTime + TimeSpan.FromSeconds(10), SomeJob, "+10 sec:1"));
        dispatcher.ScheduleJob(Job.At(curTime + TimeSpan.FromSeconds(10), SomeJob, "+10 sec:2"));

        // Starts almost immediately
        dispatcher.ScheduleJob(Job.At(curTime - TimeSpan.FromMinutes(1), SomeJob, "past"));

        // And last job to test
        dispatcher.ScheduleJob(Job.At(curTime + TimeSpan.FromSeconds(25), SomeJob, "+25 sec"));

        Console.WriteLine("Queue length: {0}, {1}", dispatcher.Length, dispatcher.IsWorking()? "working": "done");
        Console.WriteLine();

        foreach (var job in dispatcher.GetJobs()) Console.WriteLine(job);
        Console.WriteLine();

        Console.ReadLine();

        Console.WriteLine(dispatcher.IsWorking()?"Dispatcher still working": "No more jobs in queue");

        Console.WriteLine();
        foreach (var job in dispatcher.GetJobs()) Console.WriteLine(job);

        Console.ReadLine();
    }

Надеюсь, это будет полезно.

@Steven Sudit указывает мне на некоторые проблемы, поэтому здесь я пытаюсь изложить свое видение.

1) Я бы не рекомендовал использовать SortedList здесь или где-либо еще, так как это устаревший класс .NET 1.1.

SortedList‹> никоим образом не устарел. Он по-прежнему существует в .NET 4.0 и появился в .NET 2.0, когда в язык были введены универсальные шаблоны. Я не вижу смысла удалять его из .NET.

Но реальный вопрос, на который я пытаюсь ответить: Какая структура данных может хранить значения в отсортированном порядке и будет эффективна при хранении и индексировании . Существуют две подходящие готовые к использованию структуры данных: SortedDictionary‹> и Отсортированный список‹>. здесь некоторая информация о том, как выбрать. Я просто не хочу тратить реализацию с моим собственным кодом и скрывать основной алгоритм. Здесь я могу реализовать массив приоритетов или что-то другое, но это займет больше строк кода. Не вижу причин не использовать здесь SortedList‹>...

// A sorted list internally maintains two arrays that store the keys and
// values of the entries.  

_jobs.Values.SelectMany(list => list).ToList().AsReadOnly();