□ Пул потоков автоматически распределяет задания среди доступных процессоров.

Дата: 2025-05-25; Просмотры: 4

Оценка:

0.00 из 5.00 — оценок

Скачиваний: 0

Скачать

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

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

Блокировки популярны, но при удержании в течение долгого времени они создают серьезные проблемы с масштабированием. Было бы очень полезно иметь асинхронные конструкции синхронизации, в которых ваш код сообщает о том, что он хочет получить блокировку. Если получить ее не удалось, он просто возвращает управление для выполнения другой работы (вместо блокирования на неопределенное время). Затем, когда блокировка станет доступной, выполнение кода возобновляется, и он может получить доступ к ресурсу, защищенному блокировкой. Эта идея появилась у меня в процессе решения серьезных проблем масштабируемости у одного из наших клиентов. Затем я продал патентные права Microsoft. В 2009 году Патентное управление выдало патент номер 7 603 502.

Класс SemaphoreSlim реализует эту идею в своем методе WaitAsync. Сигнатура самой сложной перегруженной версии этого метода выглядит так:

public Task<Boolean> WaitAsync(Int32 millisecondsTimeout,

CancellationToken cancellationToken);

С ней вы можете синхронизировать доступ к ресурсу в асинхронном режиме (то есть без блокирования каких-либо потоков).

private static async Task

AccessResourceViaAsyncSynchronization(SemaphoreSlim asyncLock) {

// TODO: Разместите здесь любой код на ваше усмотрение...

await asyncLock.WaitAsyncQ j // Запрос монопольного доступа к ресурсу.

// Когда управление попадает в эту точку, мы знаем, что никакой другой // поток не обращается к ресурсу.

// T0D0: Работа с ресурсом (в монопольном режиме)...

// Завершив работу с ресурсом, снимаем блокировку, чтобы ресурс // стал доступным для других потоков. asyncLock.Release();

// T0D0: Разместите здесь любой код на ваше усмотрение...

}

Метод WaitAsync класса SemaphoreSlim чрезвычайно полезен, но, конечно, он реализует семантику семафора. Обычно объект SemaphoreSlim создается со счетчиком 1, что обеспечивает взаимоисключающий доступ к защищаемому ресурсу. Таким образом, реализуемое поведение сходно с тем, которое достигается при использовании Monitor — не считая того, что SemaphoreSlim не предоставляет семантики рекурсии и владения потоками (впрочем, это хорошо).

А что делать с семантикой чтения/записи? В .NET Framework входит класс ConcurrentExclusiveSchedulerPair, который выглядит примерно так:

public class ConcurrentExclusiveSchedulerPair {
public ConcurrentExclusiveSchedulerPair();

public TaskScheduler ExclusiveScheduler { get; } public TaskScheduler ConcurrentScheduler { get; }

// Другие методы не показаны

}

Экземпляр этого класса содержит два объекта TaskScheduler, которые совместно реализуют семантику чтения/записи при планировании заданий. Все задания, запланированные с использованием ExclusiveScheduler, выполняются по одному — при отсутствии выполняемых задач, запланированных с использованием ConcurrentScheduler. И конечно, все задачи, запланированные с использованием ConcurrentScheduler, могут выполняться одновременно — при отсутствии выпол-

няемых заданий, запланированных ExclusiveSchedulen. Пример использования класса ConcunnentExclusiveSchedulenPain представлен ниже:

private static void ConcurrentExclusiveSchedulerDemo() { var cesp = new ConcurrentExclusiveSchedulerPairQ; var tfExclusive = new TaskFactory(cesp.ExclusiveScheduler); var tfConcurrent = new TaskFactory(cesp.ConcurrentScheduler); for (Int32 operation = 0; operation < 5; operation++) {

var exclusive = operation < 2; // Для демонстрации создаются

// 2 монопольных и 3 параллельных задания

(exclusive ? tfExclusive : tfConcurrent).StartNew(() => {

Console.WriteLine("{0} access", exclusive ? "exclusive" : "concurrent");

// TODO: Здесь выполняется монопольная запись или параллельное чтение...

});

}

}

К сожалению, .NET Framework не предоставляет асинхронных средств блокировки с семантикой чтения/записи. Впрочем, я создал такой класс, который назвал AsyncOneManyLock. Он используется по тем же принципам, что и SemaphoreSlim:

private static async Task

AccessResourceViaAsyncSynchronization(AsyncOneManyLock asyncLock) {

// TODO: Здесь выполняется любой код...

// Передайте OneManyMode.Exclusive или OneManyMode.Shared // в зависимости от нужного параллельного доступа

await asyncLock.AcquireAsync(OneManyMode.Shared); // Запросить общий доступ // Когда управление передается в эту точку, потоки, выполняющие // запись в ресурс, отсутствуют; другие потоки могут читать данные // TODO: Чтение из ресурса...

// Завершив работу с ресурсом, снимаем блокировку, чтобы ресурс // стал доступным для других потоков. asyncLock.Release();

// TODO: Здесь выполняется любой код...

}

Е1иже приведен код моей реализации AsyncOneManyLock.

public enutn OneManyMode { Exclusive, Shared }

public sealed class AsyncOneManyLock {

#region Lock code

private SpinLock m_lock = new SpinLock(true); // He используем

// readonly c SpinLock

private void Lock() { Boolean taken = false; m_lock.Enter(ref taken); } private void UnlockQ { m lock. Exit(); }

#endregion

#region Lock state and helper methods

продолжение #

private Int32 mstate = 0;

private Boolean IsFree { get { return m_state == 0; } }

private Boolean IsOwnedByWriter { get { return m_state == 1; } }

private Boolean IsOwnedByReaders { get { return m_state > 0; } }

private Int32 AddReaders(Int32 count) { return m_state += count; }

private Int32 SubtractReader() { return instate; }

private void MakeWriterQ { m_state = 1; }

private void MakeFreeQ { m_state = 0; }

#endregion

// Для отсутствия конкуренции (с целью улучшения производительности // и сокращения затрат памяти)

private readonly Task mnoContentionAccessGranter;

// Каждый ожидающий поток записи пробуждается через свой объект // TaskCompletionSourcej находящийся в очереди

private readonly Queue<TaskCompletionSource<Object>> m_qWaitingWriters = new Queue<TaskCompletionSource<Object>>();

// Все ожидающие потоки чтения пробуждаются по одному // объекту TaskCompletionSource

private TaskCompletionSource<Object> mwaitingReadersSignal = new TaskCompletionSource<Object>(); private Int32 mnuml/daitingReaders = 0;

public AsyncOneManyLock() {

m_noContentionAccessGranter = Task.FromResult<Object>(null);

}

LockQ;

switch (mode) {

case OneManyMode.Exclusive: if (IsFree) {

MakeWriterQ; // Без конкуренции } else {

// Конкуренция: ставим в очередь новое задание записи var tes = new TaskCompletionSource<Object>(); m_qWaitingWriters.Enqueue(tcs); accressGranter = tcs.Task;

}

break;

case OneManyMode.Shared:

if (IsFree || (IsOwnedByReaders && m_qWaitingWriters.Count == 0)) { AddReaders(l); // Отсутствие конкуренции } else { // Конкуренция

// Увеличиваем количество ожидающих заданий чтения m_numWaitingReaders++;

accressGranter =

m_waitingReadersSignal.Task.ContinueWith(t => t.Result);

}

break;

}

Unlock();

return accressGranter;

}

public void ReleaseQ {

TaskCompletionSource<ObJect> accessGranter = null;

LockQ;

if (IsOwnedByWriter) MakeFreeQ; // Ушло задание записи else SubtractReaderQ; // Ушло задание чтения

if (IsFree) {

// Если ресурс свободен; пробудить одно ожидающее задание записи // или все задания чтения if (m_qWaitingWriters.Count > 0) {

MakeWriterQ;

accessGranter = m_qWaitingWriters. DequeueQ ;

} else if (m_numWaitingReaders > 0) {

AddReaders(m_numWaitingReaders); m_numWaitingReaders = 0; accessGranter = mwaitingReadersSlgnal;

// Создание нового объекта TCS для будущих заданий;

// которым придется ожидать

m_waitingReadersSignal = new TaskCompletionSource<Object>();

>

>

Unlock();

// Пробуждение задания чтения/записи вне блокировки снижает // вероятность конкуренции и повышает производительность if (accessGranter != null) accessGranter.SetResult(null);

>

}

Как я уже упоминал, этот код вообще не блокирует выполнение потоков, поскольку в его внутренней реализации не используются конструкции ядра. В нем используется класс Spin Lock, в реализации которого задействованы конструкции пользовательского режима. Но если вы вспомните из обсуждения Spin Lock в главе 29, этот класс следует использовать только для секций кода, заведомо выполняемых за короткое и конечное время. Проанализировав мой метод WaitAsync, вы увидите, что во время удержания блокировки я ограничиваюсь незначительными целочисленными вычислениями и сравнениями и, возможно, созданием объекта TaskCompletionSource и его добавлением в очередь. Все это не займет много времени, поэтому блокировка заведомо будет удерживаться в течение очень короткого промежутка.

Аналогичным образом метод Release ограничивается целочисленными вычислениями, сравнением и, возможно, выведением объекта TaskCompletionSource из очереди или его созданием. Все это тоже происходит очень быстро. Все это позволило мне с уверенностью использовать Spin Lock для защиты доступа к Queue. Выполнение потоков никогда не блокируется, что способствует написанию масштабируемого, быстрого кода.

Классы коллекций

для параллельного доступа

В FCL существует четыре безопасных в отношении потоков класса коллекций, принадлежащих пространству имен System. Collections. Concurrent: ConcurrentQueue, ConcurrentStack, ConcurrentDictionary и ConcurrentBag. Вот как выглядят наиболее часто используемые члены:

// Обработка элементов по алгоритму FIFO

public class ConcurrentQueue<T> : IProducerConsumerCollection<T>,

IEnumerable<T>_J ICollection, IEnumerable {

public ConcurrentQueue();

public void Enqueue(T item);

public Boolean TryDequeue(out T result);

public Int32 Count { get; }

public IEnumerator<T> GetEnumerator();

>

// Обработка элементов по алгоритму LIFO

public class ConcurrentStack<T> : IProducerConsumerCollection<T>,

IEnumerable<T>, ICollection, IEnumerable {

public ConcurrentStack();

public void Push(T item);

public Boolean TryPop(out T result);

public Int32 Count { get; }

public IEnumerator<T> GetEnumeratorQ;

>

// Несортированный набор элементов с возможностью хранения дубликатов public class ConcurrentBag<T> : IProducerConsumerCollection<T>,

IEnumerable<T>, ICollection, IEnumerable {

public ConcurrentBag();

public void Add(T item);

public Boolean TryTake(out T result);

public Int32 Count { get; }

public IEnumerator<T> GetEnumeratorQ;

}

// Несортированный набор пар ключ/значение

public class ConcurrentDictionary<TKey, TValue> : IDictionary<TKey, TValue>, ICollection<KeyValuePair<TKey, TValue>>, IEnumerable<KeyValuePair<TKey, TValue>>, IDictionary, ICollection, IEnumerable {

public ConcurrentDictionary();

public Boolean TryAdd(TKey key, TValue value);

public Boolean TryGetValue(TKey key, out TValue value);

public TValue this[TKey key] { get; set; }

public Boolean TnyUpdate(

TKey key, TValue newValue, TValue comparisonValue); public Boolean TryRemove(TKey key, out TValue value); public TValue AddOnUpdate(

TKey key, TValue addValue, FunccTKey, TValue> updateValueFactory); public TValue GetOnAdd(TKey key, TValue value); public Int32 Count { get; }

public IEnumerator<KeyValuePair<TKey, TValue>> GetEnumerator();

Эти классы коллекций являются неблокирующими. При попытке извлечь несуществующий элемент поток немедленно возвращает управление, а не блокируется, ожидая появления элемента. Именно поэтому такие методы, как ТryDequeue, ТгуРор, ТryTake и ТryGetValue, при получении элемента возвращают значение true, а при

его невозможности — false.

Хотя эти коллекции являются неблокирующими, это вовсе не означает, что они обходятся без синхронизации. Класс ConcurrentDictionary внутренне использует класс Monitor, но блокировка удерживается только на короткое время, необходимое для работы с элементом коллекции. В то же время классы ConcurrentQueue и ConcurrentStack для манипулирования коллекцией используют методы Interlocked и поэтому обходятся вообще без блокирования. Один объект ConcurrentBag внутренне состоит из объекта мини-коллекций для каждого потока. При добавлении нового элемента методы Interlocked помещают его в мини-коллекцию вызывающего потока. При попытке извлечь элемент его наличие опять же проверяется в мини-коллекции вызывающего потока. При обнаружении элемента задействуется метод класса Interlocked. Если же элемент в рассматриваемой мини-коллекции отсутствует, методы класса Monitor извлекают его из мини-коллекции другого потока. Мы говорим, что имеет место захват (stealing) элемента у другого потока.

Обратите внимание, что все рассматриваемые классы обладают методом GetEnumerator, обычно используемым в инструкции C# foreach, но допустимым и в языке LINQ. Для классов ConcurrentStack, ConcurrentQueue и ConcurrentBag метод GetEnumerator создает снимок содержимого коллекции и возвращает зафиксированные элементы; при этом реальное содержимое коллекции уже может измениться. Метод GetEnumerator класса ConcurrentDictionary не фиксирует содержимое коллекции, а значит, в процессе просмотра словаря его вид может поменяться; об этом следует помнить. Свойство Count возвращает количество элементов в коллекции на момент запроса. Если другие потоки в это время добавляют элементы в коллекцию или извлекают их оттуда, возвращенное значение может оказаться неверным.

Классы ConcurrentStack, ConcurrentQueue и ConcurrentBag реализуют интерфейс IProducerConsumerCollection, который выглядит следующим образом:

public interface IProducerConsumerCollection<T> : IEnumerable<T>,

ICollection, IEnumerable {

Boolean TryAdd(T item);

Boolean TryTake(out T item);

T[] ToArray();

void CopyTo(T[] array, Int32 index);

}

Любой реализующий данный интерфейс класс может превратиться в блокирующую коллекцию. Поток, добавляющий элементы, блокируется, если коллекция уже заполнена, а поток, удаляющий элементы, блокируется, если она пуста. Разумеется, я по возможности стараюсь избегать таких коллекций, ведь они предназначены именно для блокировки потоков. Для преобразования коллекции в блокирующую создается класс System.Collections.Concurrent.BlockingCollection, конструктору которого передается ссылка на неблокирующую коллекцию. Этот класс выглядит следующим образом (некоторые методы не показаны):

public class BlockingCollection<T> : IEnumerable<T>, ICollection,

IEnumerable, IDisposable { public BlockingCollection(

IProducerConsumerCollection<T> collection, Int32 boundedCapacity);

public void Add(T item); public Boolean TryAdd(

T item, Int32 msTimeout, CancellationToken cancellationToken); public void CompleteAdding();

public T Take(); public Boolean TryTake(

out T item, Int32 msTimeout, CancellationToken cancellationToken);

public Int32 BoundedCapacity { get; } public Int32 Count { get; }

public Boolean IsAddingCompleted { get; } // true, если вызван метод

// AddingComplete

public Boolean IsCompleted { get; } // true, если вызван метод

// IsAddingComplete и Count==0

public IEnumerable<T> GetConsumingEnumerable(

CancellationToken cancellationToken);

public void CopyTo(T[] array, int index);

public Т[] ToArrayQ; public void DisposeQ;

}

При конструировании экземпляра BlockingCollection параметр bounded- Capacity показывает максимально допустимое количество элементов коллекции. Если поток вызывает метод Add для уже заполненной коллекции, он блокируется. Впрочем, поток может вызвать метод ТnyAdd, передав ему время задержки (в миллисекундах) и/или объект CancellationToken. В результате поток блокируется до добавления элемента, окончания времени ожидания или отмены объекта CancellationToken (класс CancellationToken подробно рассматривался в главе 27).

Класс BlockingCollection реализует интерфейс IDisposable. В итоге метод Dispose вызывается для внутренней коллекции и удаляет заодно два объекта SemaphoreSlim, используемые классом для блокировки потоков-производителей и потоков-потребителей.

Завершив добавление элементов в коллекцию, поток-производитель должен вызвать метод CompleteAdding. Это даст понять потоку-потребителю, что больше элементов не будет и цикл f oneach, использующий объект GetConsumingEnumerable, завершится. Показанный далее код демонстрирует, как организовать сценарий с участием производителя/потребителя и сигналом о завершении:

public static void Main() {

var Ы = new BlockingCollection<Int32>(new ConcurrentQueue<Int32>());

11 Поток пула получает элементы ThreadPool.QueueUserWorkrtem(ConsumeItems, Ы);

// Добавляем в коллекцию 5 элементов for (Int32 item = 0; item < 5; item++) {

Console.WriteLine("Producing: " + item); bl.Add(item);

>

// Информируем поток-потребитель, что больше элементов не будет Ь1.CompleteAdding();

Console.ReadLine(); // Для целей тестирования

>

private static void ConsumeItems(Object о) {
var Ы = (BlockingCollection<Int32>) о;

// Блокируем до получения элемента, затем обрабатываем его foreach (var item in bl.GetConsumingEnumerable()) {

Console.WriteLine("Consuming: " + item);

}

// Коллекция пуста и там больше не будет элементов Console.WriteLine("All items have been consumed");

}

После выполнения данного кода я получаю:

Producing: 0 Producing: 1 Producing: 2 Producing: 3 Producing: 4 Consuming: 0 Consuming: 1 Consuming: 2 Consuming: 3 Consuming: 4

All items have been consumed

Если вы попробуете запустить этот код, строчки Producing (производство) и Consuming (потребление) могут быть перемешаны, но строка All items have been consumed (все элементы потреблены) всегда будет замыкать список вывода.

Класс BlockingCollection обладает также статическими методами AddToAny, TryAddToAny, TakeFromAny и TryTakeFromAny. Все они принимают в качестве параметров коллекцию BlockingCollection<T>[ ], а кроме того, элемент, время ожидания и объект CancellationToken. Методы (Try)AddToAny циклически просматривают все коллекции в массиве, пока не обнаруживают коллекцию, способную принять новый элемент. Методы (Try)TakeFromAny циклически просматривают все коллекции до обнаружения той, из которой можно извлечь элемент.

Словарь соответствия ру сскояз ы иных и англоязычных терминов

Джеффри Рихтер

CLR via С#. Программирование на платформе
Microsoft .NET Framework 4.5 на языке C#
4-е издание

Перевел с английского Е. Матвеев

Заведующий редакцией Руководитель проекта Ведущий редактор Художественный редактор Корректор Верстка

ООО «Питер Пресс», 192102, Санкт-Петербург, ул. Андреевская (д. Волкова), д. 3, литер А, пом. 7Н.
Налоговая льгота — общероссийский классификатор продукции ОК 005-93,
том 2; 95 3005 — литература учебная.

Подписано в печать 13.06.13. Формат 70x100/16. Уел. п. л. 72,240. Тираж 2000. Заказ
Отпечатано по технологии CtP в ООО «Полиграфический комплекс «ЛЕНИЗДАТ».

194044, Санкт-Петербург, ул. Менделеевская, 9. Телефон / факс (812) 495-56-10.

¹ В C# внутри типов, помеченных атрибутом [Serializable], не стоит определять автоматически реализуемые свойства. Дело в том, что имена полей, генерируемые компилятором, могут меняться после каждой следующей компиляции, что сделает невозможной десериализацию экземпляров типа.

Рис. 1.3. Определение целевой платформы средствами Visual Studio

[2] la рис. 1.3 обратите внимание на флажок Prefer 32-Bit. Он доступен только в том случае, когда в списке Platform Target выбрана строка Any CPU. а для выбранного типа проекта создается исполняемый файл. Если установить флажок Prefer 32-Bit. то Visual Studio запускает компилятор C# с параметром командной строки /platform: anycpu32bitpreferred. Этот параметр указывает, что исполняемый файл должен выполняться как 32-разрядный даже на 64-разрядных машинах. Если вашему приложению не нужна дополнительная память, доступная для 64-разрядных процессов, обычно стоит выбрать именно этот режим, потому что Visual Studio не поддерживает функцию «Изменить и продолжить» (Edit-and-Continue) для приложений х64. Кроме того. 32-разрядные приложения могут взаимодействовать с 32-разрядными библиотеками DLL и компонентами СОМ, если этого потребует ваше приложение.

В зависимости от указанной целевой платформы C# генерирует заголовок — РЕ.32 или РЕ.32+. а также включает в него требуемую процессорную архитектуру (или признак независимости от архитектуры). Для анализа заголовочной информации. вставленной компилятором в управляемый модуль, Microsoft предоставляет две утилиты — DumpBin.exe и CorFlags.exe.

^[3] Программный код может запросить переменную окружения Is64 BitOperatingSystem для того, чтобы определить, выполняется ли данная программа в 64-разрядной системе Windows, а также запросить переменную окружения Is64Bil Process, чтобы определить, выполняется ли данная программа в 64-разрядном адресном пространстве.

^[4] При использовании параметра /t[arget]:winmdobj полученный файл .winmdobj должен быть передан программе WinMDExp.exe, который немного обрабатывает метаданные для представления открытых типов CLR сборки как типов Windows Runtime. Программа WinMDExp.exe никак не затрагивает код IL.

^[5] В этом примере используется механизм Enhanced Strong Naming, появившийся в .NET Framework 4.5. Чтобы создать сборку, совместимую с предыдущими версиями .NET Framework, вам также придется создать подпись другой стороны (counter-signature) с использованием атрибута AssemblySignatureKey. За подробностями обращайтесь по адресу http://msdn. microsoit.com/en-us/library/hh415055(v=vs.ll0).aspx.

^[6] CLR также позволяет распаковывать значимые типы в версию этого же типа, поддерживающую присвоение значений null (см. главу 19).

^[7] И как обычно, значимый тип будет упакован.

[8] вызова конструктора типа есть некоторые особенности. При компиляции метода JIT-компилятор обнаруживает типы, на которые есть ссылки из кода. Если в каком-либо из типов определен конструктор, JIT-компилятор проверяет, был ли исполнен конструктор типа в данном домене приложений. Если нет, JIT-

^[9] В настоящий момент в msdn.microsoft.com используется термин «необязательные и именованные аргументы», но автор в данной книге называет их «параметрами». Мы решили сохранить авторскую терминологию. — Примем, ред.

^[10] Термин «tuple» возник как «обобщение» последовательности: single, double, triple, quadruple, quintuple, n-tuple.

^[11] По этой причине класс IndexerNameAttribute не входит в описанные в ЕСМА стандарты CLI и языка С#.

^[12] П рерывание потока или выгрузка домена приложений является источником исключения Thread Abort Exception, обеспечивающего выполнение блока finally. Если же поток прерывается функцией TerminateThread или методом FailFast класса System.Environment, блок finally не выполняется. Разумеется, Windows производит очистку всех ресурсов, которые использовались прерванным процессом.

^[13] Класс System.Exception следовало бы объявить абстрактным, чтобы код, который пытается сгенерировать его, даже не компилировался.

^[14] Мне следовало сказать, что столь привлекательной для разработчиков платформу делают еще и редактор Visual Studio, механизм IntelliSense, возможность работы с фрагментами кода, шаблоны, расширяемость, отладчик и многое другое. Но я вынес это за пределы основного обсуждения, так как эти средства не влияют на поведение кода во время выполнения.

^[15] Кстати, конструкторы классов для типов System.Char, System.String, System.Type и System.10.Stream в этом приложении также могут стать причиной исключения TypelnitializationException, причем при аналогичных обстоятельствах.