Scheduler

Планировщик (англ. Scheduler) — Шаблон проектирования обеспечивающий механизм реализации политики планирования, но при этом не зависит ни от одной конкретной политики. Управляет порядком, в соответствии с которым потокам предстоит выполнить последовательный код, используя для этого объект, который явным образом задаёт последовательность ожидающих потоков.

Мотивы

• Несколько потоков могут потребовать доступа к ресурсу одновременно, и только один поток в какой-то момент времени может осуществить доступ к ресурсу.
• Согласуясь с требованиями программы, потоки должны осуществлять доступ к ресурсу в определенном порядке.

Пример реализации

Пример C#

<source lang="csharp"> using System;

namespace Digital_Patterns.Concurrency.Sheduler {

   class Printer
   {
       private static Int32 mID = 0;

       private Scheduler _scheduler = new Scheduler();

       public void Print(JournalEntry journalEntry)
       {
           Int32 id = ++mID;
           try
           {
               Console.WriteLine(String.Format(@"{0}: enter scheduler", id));
               // вызов не выполниться до тех пор, пока объект Scheduler не решит,
               // что подошла очередь распечатать этот объект JournalEntry
               _scheduler.Enter(journalEntry);
               Console.WriteLine(String.Format(@"{0}: start printing", id));
               try
               {
                   //TODO Something
                   journalEntry.Do(id);
               }
               finally
               {
                   // вызов метода Done говорит Scheduler о том, что объект JournalEntry
                   // распечатан, и может подойти очередь вывода на печать другого объекта
                   // JournalEntry
                   _scheduler.Done();
                   Console.WriteLine(String.Format(@"{0}: done scheduler", id));
               }
           }
           catch (Exception) {}
       }
   }

} </source>
<source lang="csharp"> using System; using System.Collections.Generic; using System.Threading;

namespace Digital_Patterns.Concurrency.Sheduler {

   /// <summary>
   /// Экземпляры классов в этой роли управляют обработкой объектов Request <see cref="JournalEntry"/>,
   /// выполняемой объектом Processor <see cref="Printer"/>. Чтобы быть независимыми от типов
   /// запросов, класс <see cref="Scheduler"/> не должен ничего знать об управляемом им классе Requect.
   /// Вместо этого он осущуствляет доступ к объектам Request через реализуемый ими интерфейс <see cref="ISchedulerOrdering"/>
   /// </summary>
   class Scheduler
   {
       /// <summary>
       /// Объект синхронизации потоков
       /// </summary>
       private AutoResetEvent _event = new AutoResetEvent(false);

       /// <summary>
       /// Устанавливается в null, если управляемый объектом Scheduler ресурс не занят.
       /// </summary>
       private Thread _runningThread;

       /// <summary>
       /// Потоки и их запросы ожидающие выполнения
       /// </summary>
       private Dictionary<Thread, ISchedulerOrdering> _waiting = new Dictionary<Thread, ISchedulerOrdering>();

       /// <summary>
       /// Метод <see cref="Enter"/> вызывается перед тем, как поток начнет использовать уравляемый ресурс.
       /// Метод не выполняется до тех пор пока управляемый ресур не освободиться и объект <see cref="Sheduler"/>
       /// не примет решение, что подошла очередь выполнения этого запроса
       /// </summary>
       /// <param name="s"></param>
       public void Enter(ISchedulerOrdering s)
       {
           var thisThread = Thread.CurrentThread;

           lock(this)
           {
               // Определяем не занат ли планировщик
               if(_runningThread == null)
               {
                   // Немедленно начинмем выполненин поступившего запроса
                   _runningThread = thisThread;
                   return;
               }
               _waiting.Add(thisThread, s);
           }
           
           lock(thisThread)
           {
               //Блокируем поток до тех пор, пока планировщик не решит сделать его текущим
               while(thisThread != _runningThread)
               {
                   _event.WaitOne();
                   _event.Set();   // даем возможность другим потокам проверить своё состояние
                   Thread.Sleep(1);
               }
               _event.Reset();
           }

           lock (this)
           {
               _waiting.Remove(thisThread);
           }
       }

       /// <summary>
       /// Вызов метода <see cref="Done"/> указывает на то, что текущий поток завершил работу
       /// и управляемый ресурс освободился
       /// </summary>
       public void Done()
       {
           lock (this)
           {
               if (_runningThread != Thread.CurrentThread)
                   throw new ThreadStateException(@"Wrong Thread");

               Int32 waitCount = _waiting.Count;
               if (waitCount <= 0)
               {
                   _runningThread = null;
               }
               else if (waitCount == 1)
               {
                   _runningThread = _waiting.First().Key;
                   _waiting.Remove(_runningThread);
                   _event.Set();
               }
               else
               {
                   var next = _waiting.First();
                   foreach (var wait in _waiting)
                   {
                       if(wait.Value.ScheduleBefore(next.Value))
                       {
                           next = wait;
                       }
                   }

                   _runningThread = next.Key;
                   _event.Set();
               }
           }
       }
   }

   /// <summary>
   /// Вспомогательный класс
   /// </summary>
   static partial class ConvertTo
   {
       /// <summary>
       /// Получить первый элемент коллекции
       /// </summary>
       /// <param name="collection"></param>
       /// <returns></returns>
       public static KeyValuePair<Thread, ISchedulerOrdering> First(this Dictionary<Thread, ISchedulerOrdering> collection)
       {
           foreach (var item in collection)
           {
               return item;
           }
           throw new ArgumentException();
       }
   }

} </source>
<source lang="csharp"> using System;

namespace Digital_Patterns.Concurrency.Sheduler {

   /// <summary>
   /// Если несколько операций ожидают доступа к ресурсу, класс<see cref="Scheduler"/> использует
   /// данный интерфейс для определения порядка выполнения операций.
   /// </summary>
   interface ISchedulerOrdering
   {
       Boolean ScheduleBefore(ISchedulerOrdering s);
   }

} </source>
<source lang="csharp"> using System; using System.Threading;

namespace Digital_Patterns.Concurrency.Sheduler {

   /// <summary>
   /// Примерный код класса <see cref="JournalEntry"/>, который должен быть
   /// распечатан классом <see cref="Printer"/>
   /// </summary>
   class JournalEntry : ISchedulerOrdering
   {
       private static DateTime mTime = DateTime.Now;

       private DateTime _time;

       /// <summary>
       /// Возвращает время создания этго объекта
       /// </summary>
       public DateTime Time { get { return _time; } }

       private String _msg;

       public JournalEntry(String msg)
       {
           mTime = mTime.AddSeconds(1);
           _time = mTime;
           _msg = msg;
       }

       public void Do(Int32 id)
       {
           Console.WriteLine(String.Format(@"{0}: Start doing : {1} : {2}", id, _time, _msg));
           Thread.Sleep(1000);
           Console.WriteLine(String.Format(@"{0}: Finish do : {1} : {2}", id, _time, _msg));
       }

       /// <summary>
       /// Возвращает true, если данный запрос должен
       /// обрабатываться перед этим запросом.
       /// </summary>
       /// <param name="s"></param>
       /// <returns></returns>
       public Boolean ScheduleBefore(ISchedulerOrdering s)
       {
           if(s is JournalEntry)
           {
               var otherJournalEntry = (JournalEntry) s;
               return (this.Time < otherJournalEntry.Time);
           }
           return false;
       }
   }

}

</source>
<source lang="csharp"> using System; using System.Threading;

namespace Digital_Patterns.Concurrency.Sheduler {

   public class Example01
   {
       private Printer _printer;

       public void Run()
       {
           Console.WriteLine(@"Press any key for start, and press again for finish");
           Console.ReadKey();
           
           _printer = new Printer();
           new Thread(Thread1).Start();
           new Thread(Thread2).Start();
           new Thread(Thread3).Start();

           Console.ReadKey();
       }

       private void Thread1()
       {
           var msg1 = new JournalEntry(@"Buy toll 5.45 USD");
           var msg2 = new JournalEntry(@"Buy candy 1.05 USD");
           var msg3 = new JournalEntry(@"Buy chocolate 3.25 USD");
           _printer.Print(msg1);
           _printer.Print(msg2);
           _printer.Print(msg3);
       }

       private void Thread2()
       {
           var msg4 = new JournalEntry(@"Buy postcard 2.05 USD");
           var msg5 = new JournalEntry(@"Buy gerland 37.78 USD");
           _printer.Print(msg4);
           _printer.Print(msg5);
       }

       private void Thread3()
       {
           var msg6 = new JournalEntry(@"Buy ball 30.06 USD");
           var msg7 = new JournalEntry(@"Buy pipe 1.83 USD");
           _printer.Print(msg6);
           _printer.Print(msg7);
       }
   }

} </source>
<source lang="csharp"> using System; using Digital_Patterns.Concurrency.Sheduler;

namespace Digital_Patterns {

   class Program
   {
       static void Main(string[] args)
       {
           new Example01().Run();

           Console.WriteLine(@"Press any key for end");
           Console.ReadKey();
       }
   }

}

</source>

Ссылки

• Mark Grand Patterns in Java Volume 1: A Catalog of Reusable Design Patterns Illustrated with UML. — Wiley & Sons, 1998. — 480 с. — ISBN 0471258393 (см. синопсис (англ.))