При участии Тима Перлса, Джошуа Блоха, Джозева Боубира, Дэвида Холмса и Дага Ли Параллельное программирование в java на
 Скачать 4.97 Mb.
|
|
Листинг 14.13 Каноническая форма захвата и освобождения в классе AQS Вторая часть включает в себя, возможно, обновление состояния синхронизатора; один поток, захвативший синхронизатор, может оказывать влияние на то, смогут ли другие потоки также захватить его. Например, захват блокировки изменяет состояние блокировки с “не удерживается” на “удерживается”, а получение разрешения от семафора уменьшает количество оставшихся разрешений. С другой стороны, захват защёлки одним потоком не оказывает влияние на возможность её захвата другими потоками, поэтому захват защёлки не изменяет её состояние. Синхронизатор, поддерживающий эксклюзивный захват, должен реализовывать защищенные методы tryAcquire , tryRelease , и isHeldExclusively , а тот, что поддерживает совместный захват, должен реализовывать методы tryAcquireShared и tryReleaseShared . Методы acquire , acquireShared , release и releaseShared класса AQS вызывают формы try этих методов в подклассе синхронизатора, чтобы определить, может ли операция продолжиться. Подкласс синхронизатора может использовать методы getState , setState и compareAndSetState для проверки и обновления состояния, в соответствии с семантикой захвата и освобождения и сообщить базовому классу через возврат статуса, была ли успешна попытка захвата или освобождения синхронизатора. Например, возврат отрицательного значения из класса tryAcquireShared свидетельствует о крахе захвата; возврат нулевое значения указывает на то, что синхронизатор был захвачен эксклюзивно; и возврат положительного значения указывает на то, что синхронизатор был захвачен не эксклюзивно. Методы tryRelease и tryReleaseShared должны возвращать значение true , если при освобождении, могут быть разблокированы потоки, пытающиеся захватить синхронизатор. Для упрощения реализации блокировок, поддерживающих очереди условий (например, класс ReentrantLock), класс AQS также предоставляет механизмы для построения условных переменных, связанных с синхронизаторами. 14.5.1 Простая защёлка Класс OneShotLatch из листинга 14.14, представляет собой бинарную защелку, реализованную с использованием класса AQS. Он имеет два публичных метода, await и signal , которые соответствуют захвату и освобождению. Изначально, защёлка закрыта; любой поток, вызвавший метод await, блокируется до тех пор, пока защелка не будет открыта. Как только защёлка открывается с помощью вызова метода signal , ожидающие потоки освобождаются, и потоки, последовательно прибывшие к защёлке, продолжают своё выполнение. @ThreadSafe public class OneShotLatch { private final Sync sync = new Sync(); public void signal() { sync.releaseShared(0); } public void await() throws InterruptedException { sync.acquireSharedInterruptibly(0); } private class Sync extends AbstractQueuedSynchronizer { protected int tryAcquireShared(int ignored) { // Succeed if latch is open (state == 1), else fail return (getState() == 1) ? 1 : -1; } protected boolean tryReleaseShared(int ignored) { setState(1); // Latch is now open return true; // Other threads may now be able to acquire } } } Листинг 14.14 Бинарная защёлка, реализованная с помощью класса AbstractQueuedSynchronizer В классе OneShotLatch , состояние защёлки определяется состоянием класса AQS – закрыто (ноль) или открыто (один). Метод await вызывает метод acquireSharedInterruptibly класса AQS, который в свою очередь консультируется с методом tryAcquireShared класса OneShotLatch . Реализация метода tryAcquireShared должна возвращать значение, указывающее, может ли быть выполнен захват защёлки или нет. Если защёлка уже была ранее открыта, метод tryAcquireShared возвращает значение “успешно”, позволяя потоку продолжить выполнение; в противном случае, он возвращает значение, указывающее на то, что попытка захвата защёлки была неудачной. Метод acquireSharedInterruptibly интерпретирует неудачу при захвате защёлки в качестве признака того, что поток должен быть помещен в очередь ожидающих потоков. Аналогично, метод signal вызывает метод releaseShared , который, в свою очередь, консультируется с методом tryReleaseShared . Реализация метода tryReleaseShared безоговорочно устанавливает состояние защелки в “открыто” и указывает (через своё возвращаемое значение), что синхронизатор находится в полностью освобождённом состоянии. Это приводит к тому, что класс AQS позволяет всем ожидающим потокам попытаться повторно захватить синхронизатор, и захват защёлки теперь будет выполнен успешно, потому что метод tryAcquireShared возвращает значение “успешно”. Класс OneShotLatch представляет собой полнофункциональный, удобный и производительный синхронизатор, реализованный всего в двадцати или около того строках кода. Конечно, ему не хватает некоторых полезных возможностей, - таких как ограниченного по времени захвата или возможности проверки состояния защёлки - но их также легко реализовать, поскольку класс AQS предоставляет ограниченные по времени версии методов захвата и методы-утилиты для распространённых операций проверок. Класс OneShotLatch мог бы быть реализован путем расширения класса AQS, а не делегирования ему, но это нежелательно по нескольким причинам [EJ пункт 14]. Расширение разрушит простотой (состоящий из двух методов) интерфейс класса OneShotLatch , и, несмотря на то, что открытые методы AQS не позволят вызывающим объектам нарушить состояние защёлки, вызывающие объекты могут легко использовать их неправильно. Ни один из синхронизаторов пакета java.util.concurrent не расширяет класс AQS напрямую - все они делегируют выполнение приватным внутренним подклассам AQS. 14.6 AQS в классах-синхронизаторах из пакета java.util.concurrent Множество блокирующих классов из пакета java.util.concurrent , таких как ReentrantLock , Semaphore , ReentrantReadWriteLock , CountDownLatch , SynchronousQueue и FutureTask, построено с использованием класса AQS. Не вдаваясь в детали слишком глубоко (исходный код является частью поставки JDK 159 ), давайте кратко рассмотрим, как каждый из этих классов использует AQS. 14.6.1 Класс ReentrantLock Класс ReentrantLock поддерживает только эксклюзивный захват блокировки, поэтому он реализует методы tryAcquire , tryRelease , и isHeldExclusively ; реализация метода tryAcquire для несправедливой версии, приведена в листинге 14.15. Класс ReentrantLock использует состояние синхронизации для хранения счетчика захватов блокировки и поддерживает переменную owner , предназначенную для хранения идентификатора потока-владельца, который изменяется только тогда, когда текущий поток только что захватил блокировку или только собирается ее освободить. 160 В методе tryRelease проверяется поле owner , чтобы гарантировать, что текущий поток завладеет блокировкой прежде, чем позволит выполниться методу unlock ; в методе tryAcquire это поле используется для различия между реентерабельным и конкурентным захватом. 159 Или с меньшим количеством лицензионных ограничений http://gee.cs.oswego.edu/dl/concurrency- interest. 160 Поскольку защищенные (protected) методы манипулирования состоянием, и при чтении, и при записи, имеют семантику памяти volatile , и класс ReentrantLock аккуратно считывает значение поля owner только после вызова метода getState , и записывает в него значение только перед вызовом метода setState , класс ReentrantLock может переключаться на семантику памяти синхронизированного состояния, и таким образом избежать дальнейшей синхронизации - см. раздел 16.1.4. protected boolean tryAcquire(int ignored) { final Thread current = Thread.currentThread(); int c = getState(); if (c == 0) { if (compareAndSetState(0, 1)) { owner = current; return true; } } else if (current == owner) { setState(c+1); return true; } return false; } Листинг 14.15 Несправедливая реализация метода tryAcquire в классе ReentrantLock Когда поток пытается захватить блокировку, метод tryAcquire сначала справляется о состоянии блокировки. Если она не удерживается, метод пытается обновить состояние блокировки, чтобы обозначить, что блокировка удерживается. В связи с тем, что состояние могло измениться из-за того, что сведения о нём были получены несколько инструкций назад, метод tryAcquire пытается использовать метод compareAndSetState для автоматического обновления состояния, чтобы указать, что блокировка удерживается и убедится, что оно не изменилось с момента прошлого наблюдения. (См. описание метода compareAndSetState в разделе 15.3.) Если состояние блокировки указывает на то, что она уже удерживается и текущий поток является владельцем блокировки - счетчик захватов увеличивается; если текущий поток не является владельцем блокировки, попытка захвата блокировки завершается неудачей. Класс ReentrantLock также использует преимущества встроенной поддержки классом AQS множества условных переменных и наборов ожидания. Метод Lock.newCondition возвращает новый экземпляр ConditionObject , представляющий собой внутренний класс AQS. 14.6.2 Классы Semaphore и CountDownLatch Класс Semaphore использует синхронизированное состояние класса AQS для хранения количества доступных разрешений. Метод tryAcquireShared (см. листинг 14.16) сначала вычисляет количество оставшихся разрешений, а если их недостаточно, возвращает значение, указывающее, что захват завершился неудачно. Если осталось достаточное количество разрешений, он пытается атомарно уменьшить количество разрешений с помощью метода compareAndSetState . Если операция выполнена успешно (это означает, что количество разрешений не изменилось с момента последнего просмотра), он возвращает значение, указывающее, что захват произведён успешно. Возвращаемое значение также содержит код успешности других попыток совместного захвата, в этом случае другие ожидающие потоки также будут разблокированы. protected int tryAcquireShared(int acquires) { while (true) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; } } protected boolean tryReleaseShared(int releases) { while (true) { int p = getState(); if (compareAndSetState(p, p + releases)) return true; } } Листинг 14.16 Методы tryAcquireShared и tryReleaseShared класса Semaphore Цикл while завершается тогда, когда нет достаточного количества разрешений или когда метод tryAcquireShared может атомарно обновить количество разрешений для соответствия количеству захватов. В то время как любой вызов метода compareAndSetState может завершиться неудачей из-за конкуренции с другими потоками (см. раздел 15.3), вынуждая его повторяться, один из двух критериев завершения станет истинным в пределах разумного количества повторных попыток. Аналогичным образом, метод tryReleaseShared увеличивает количество разрешений, потенциально разблокируя ожидающие потоки и повторяет попытки до успешного обновления. Возвращаемое методом tryReleaseShared значение указывает на то, могли ли другие потоки быть разблокированы освобождением. Класс CountDownLatch использует класс AQS аналогично классу Semaphore : синхронизированное состояние содержит текущее количество. Метод countDown вызывает метод release , который приводит к уменьшению счетчика и разблокирует ожидающие потоки, если счетчик достигает нуля; метод await вызывает метод acquire , который немедленно возвращает управление, если счетчик достигает нуля, в противном случае блокируется. 14.6.3 Класс FutureTask На первый взгляд класс FutureTask даже не выглядит как синхронизатор. Но метод Future.get имеет семантику, очень похожую на семантику защёлки - если произошло какое-либо событие (завершение или отмена задачи, представленной экземпляром FutureTask ), то потоки могут продолжать работу, иначе они помещаются в очередь до тех пор, пока это событие не произойдет. Класс FutureTask использует синхронизированное состояние класса AQS для хранения состояния задачи – выполняется (running), завершено (completed) или отменено (cancelled). Он также поддерживает дополнительные переменные состояния для хранения результата вычисления или вызванного им исключения. Кроме того, он поддерживает ссылку на поток, в котором выполняется вычисление (если он в настоящее время находится в состоянии выполнения), чтобы его можно было прервать, если задача будет отменена. 14.6.4 Класс ReentrantReadWriteLock Интерфейс ReadWriteLock предполагает наличие двух блокировок - блокировки на чтение и блокировки на запись, но в реализации класса ReentrantReadWriteLock на основе AQS, блокировкой на чтение и на запись управляет единственный подкласс AQS. Класс ReentrantReadWriteLock использует 16 бит состояния для подсчёта количества блокировок на запись, а остальные 16 бит для подсчёта количества блокировок на чтение. Операции блокировок на чтение совместно используют методы захвата и освобождения; операции блокировки на запись используют методы захвата и освобождения эксклюзивно. Внутри себя класс AQS поддерживает очередь ожидающих потоков, отслеживая, запрашивал ли поток эксклюзивный или совместный доступ. В случае класса ReentrantReadWriteLock , когда блокировка становится доступной, если поток, находящийся вначале очереди, ищет доступ на запись, он получит его, и если поток, находящийся в начале очереди, ищет доступ на чтение, все ожидающие в очереди потоки будут получать его, вплоть до первого запроса запись. 161 14.7 Итоги Если вам нужно реализовать класс, зависимый от состояния, - тот, чьи методы должны блокироваться, если предусловие на основе состояния не выполняется - лучшая стратегия, как правило, заключается в реализации на основе существующего библиотечного класса, подобного Semaphore , BlockingQueue или CountDownLatch , как в случае класса ValueLatch (из раздела 8.5.1 ). Однако иногда существующие библиотечные классы не обеспечивают достаточной базы; в этих случаях можно создать собственные синхронизаторы, используя внутренние очереди условий, явные объекты Condition или класс AbstractQueuedSynchronizer . Внутренние очереди условий тесно связаны с внутренней блокировкой, так как механизм управления зависимостью состояний обязательно связан с механизмом обеспечения согласованности состояний. Точно так же, явные экземпляры Condition тесно связаны с явными экземплярами Lock и предлагают расширенный набор функций по сравнению с внутренними очередями условий, включая несколько наборов ожидания для каждой блокировки, прерываемые или непрерывные условия ожидания, справедливое или несправедливое помещение в очередь и ожидание с ограничением по времени. 161 Этот механизм не позволяет выбирать политику предпочтения операциям чтения или записи, как это делают некоторые реализации блокировки на чтение-запись. Для этого или очередь ожидания класса AQS должна была бы быть отличной от очереди FIFO, или были бы необходимы две очереди. Однако, такая строгая политика упорядочивания редко нужна на практике; если несправедливая версия класса ReentrantReadWriteLock не предлагает приемлемой живучести, справедливая версия обычно предоставляет удовлетворительный порядок и гарантирует отсутствие голодания для читателей и писателей. Chapter 15 Атомарные переменные и неблокирующая синхронизация Многие из классов пакета java.util.concurrent , такие как Semaphore и ConcurrentLinkedQueue , обеспечивают лучшую производительность и масштабируемость по сравнению с альтернативами, использующими блок synchronized . В этой главе мы рассмотрим основной источник повышения производительности: атомарные переменные и неблокирующую синхронизацию. Большая часть недавних исследований в области параллельных алгоритмов была сосредоточена на неблокирующих алгоритмах, которые используют для обеспечения целостности данных, при параллельном доступе, низкоуровневые атомарные машинные инструкции, подобные сравнить-и-обменять (compare-and- swap), а не блокировки. Неблокирующие алгоритмы широко используются в операционных системах и среде JVM для планирования потоков и процессов, сборки мусора, а также для реализации блокировок и других параллельных структур данных. Неблокирующие алгоритмы значительно сложнее в разработке и реализации, чем основанные на блокировках альтернативы, но они могут предложить значительные преимущества в масштабируемости и живучести. Они координируются на более тонком уровне детализации и могут значительно сократить затраты на планирование, поскольку они не блокируются, когда несколько потоков конкурируют за одни и те же данные. Кроме того, они невосприимчивы к взаимоблокировкам и другим проблемам живучести. В алгоритмах, основанных на блокировках, другие потоки не могут прогрессировать, если поток переходит в спящий режим или вращается, пока удерживается блокировка, тогда как неблокирующие алгоритмы непроницаемы для сбоев отдельных потоков. Начиная с Java 5.0, можно построить эффективные неблокирующие алгоритмы в Java, используя классы атомарных переменных, такие как AtomicInteger и AtomicReference Атомарные переменные также могут быть использованы в качестве “лучших volatile переменных”, даже если вы не разрабатываете неблокирующие алгоритмы. Атомарные переменные предлагают ту же семантику памяти, что и переменные volatile , но с дополнительной поддержкой атомарных обновлений, что делает их идеально подходящими для счетчиков, генераторов последовательностей и сбора статистики, предлагая лучшую масштабируемость, чем альтернативы на основе блокировок. 15.1 Недостатки блокировки Координация доступа к совместно используемому состоянию с помощью согласованного протокола блокировки гарантирует, что любой поток, удерживающий блокировку, защищающую набор переменных, имеет монопольный доступ к этим переменным и что любые изменения, внесенные в эти переменные, видны другим потокам, которые впоследствии захватят блокировку. Современные среды JVM могут оптимизировать захват и освобождение незапланированных блокировок довольно эффективно, но если несколько потоков запрашивают блокировку одновременно, JVM обращается за помощью к операционной системе. Если он дойдет до этого момента, какой-то неудачливый поток будет приостановлен, и его выполнение будет возобновлено позже 162 . При возобновлении этого потока, может потребоваться дождаться завершения квантов планирования другими потоками, прежде чем он фактически будет запланирован к выполнению. Приостановка и возобновление потока приводит к большим издержкам и обычно влечет за собой длительное прерывание. Для классов на основе блокировок с хорошо детализированными операциями (подобных синхронизированным классам коллекций, в которых большинство методов содержат только несколько операций), отношение издержек планирования к полезной работе может быть довольно высоким, когда блокировка часто оспаривается. Переменные volatile - это более легкий механизм синхронизации, чем блокировка, поскольку они не приводят к переключению контекста или планированию потоков. Однако volatile переменные имеют некоторые ограничения по сравнению с блокировкой: хотя они предоставляют аналогичные гарантии видимости, их нельзя использовать для построения атомарных составных действий. Это означает, что переменные volatile нельзя использовать, когда одна переменная зависит от другой, или когда новое значение переменной зависит от ее старого значения. Это ограничивает спектр ситуаций, в которых допустимо применение переменных volatile , так как они не могут быть использованы для надежной реализации обычных инструментов, подобных счетчикам или мьютексам. 163 Например, в то время как операция инкремента ( ++i ) может выглядеть как атомарная операция, на самом деле это три различные операции - получить текущее значение переменной, добавить к ней единицу, а затем записать обновленное значение обратно. Чтобы не потерять обновление, вся операция чтение-изменение-запись должна быть атомарной. До сих пор мы видели только один способ, которым можно это сделать – с помощью блокировки, как в классе Counter из раздела 4.1 Класс Counter потокобезопасен, и при наличии небольшой конкуренции, или в её отсутствии, работает просто отлично. Но в условиях конфликта производительность страдает из-за накладных расходов на переключение контекста и задержек, вызванных планированием. Когда блокировки удерживаются на краткое время, приостановка выполнения потока является суровым наказанием за то, что блокировка была запрошена в неудачный момент времени. Блокировка имеет ещё несколько недостатков. Когда поток ожидает блокировку, он не может делать ничего другого. Если поток, удерживающий блокировку, задерживается (из-за ошибки страницы, задержки, вызванной планированием или тому подобного), то ни один поток, которому нужна эта блокировка, не сможет прогрессировать. Это может быть серьезной проблемой, если заблокированный поток является высокоприоритетным потоком, но поток, удерживающий блокировку, является низкоприоритетным потоком - угроза производительности, известная как инверсия приоритета (priority inversion). Даже при том, что поток с более высоким приоритетом должен иметь приоритет, он вынужден ожидать, пока блокировка не будет освобождена, и это эффективно понижает его приоритет до потока с более низким приоритетом. Если поток, удерживающий блокировку, постоянно блокируется (из-за бесконечного цикла, 162 Умная среда JVM не обязательно должна приостанавливать поток, если он конкурирует за блокировку; она может использовать данные профилирования, чтобы принять адаптивное решение о приостановке или блокировке на основе прокручивания, в зависимости от того, как долго блокировка удерживалась во время предыдущих захватов. 163 Теоретически возможно, хотя и совершенно непрактично, использовать семантику volatile для построения мьютексов и других синхронизаторов; см. (Raynal, 1986). взаимоблокировки, динамической блокировки или другого сбоя живучести), любые потоки, ожидающие этой блокировки, никогда не смогут добиться прогресса. Даже игнорируя эти угрозы, блокировка сама по себе является тяжеловесным механизмом для хорошо детализированных операций, подобных операции увеличения счётчика. Было бы неплохо иметь более детализированный подход для управления конкуренцией между потоками – что-то вроде volatile переменных, но предоставляющих возможность атомарных обновлений. К счастью, современные процессоры предлагают нам именно такой механизм. |