Коли ReadWriteLock буває марний

Хочу поділитися досвідом використання ReadWriteLock - імплементації , що поставляється в JDK , нашим « розчаруванням » в ній і про те , як ми обійшли проблему обмеження scalability .

Але почнемо ми здалеку . Існує два типи блокувань - spin lock і lock , обрашается до OS для перемикання між потоками . Spin lock , як , я думаю , ви вже здогадалися , що не переводить потік в стан очікування , а чекає в циклі доти , поки необхідний ресурс не осовбодітся . Існує думка , що spin lock - це минуле століття , і що треба користуватися Блокирущие lock'амі . Насправді це не так. Якщо навіть не брати до уваги , що блокуючий lock викликає переключення контекстів потоків , у ньому також проводиться interpocessing method call - дорога операція . Виклик методу OS може привести до високої latency під час захоплення монітора об'єкта . Таким чином , якщо у вас час виконання самого методу дуже маленьке , то spin lock підвищуватиме швидкість виконанню програми сильніше , ніж блокуючий lock . Якщо ж у вас сам метод виконується дуже довго , то Блокирущие lock вигідніше , так як він дозволяє ефективніше перерозподілити ресурси OS. ReadWriteLock використовує в чомусь схожий підхід . Спочатку використовується spin lock , і якщо отримати доступ до ресурсу неможливо , викликається метод java.util.concurrent.locks.LockSupport # park ( java.lang.Object ) для передачі ресурсів OS іншому потоку . Причому даний підхід відноситься також до реалізації exclusive lock на рівні JDK ( java.util.concurrent.locks.ReentrantLock ) і JVM ( synchronized ) . Тепер давайте расмотрим в деталях , як це реалізовано . Почнемо з дуже примітивною ексклюзивної блокування (так званий test - and - set lock ): public class TestAndSetLock {     AtomicBoolean state = new AtomicBoolean ( false ) ;     public void lock ( ) {      while ( state.getAndSet ( true )) { }     }     public void unlock ( ) {      state.set ( false ) ;     } } Операція getAndSet є атомарної тобто між get і set значення змінної гарантовано не поміняється , що і дозволяє використовувати її для даного spin lock . Зробимо невеликий, далеко не ідеальний , але показовий бенчмарк .
Основний код цього бенчмарка тут : public final class Countdown implements Callable & lt ; Long & gt ; {      private final long iterations ;      public Countdown ( long iterations ) {       this.iterations = iterations ;      }      Override      public Long call ( ) throws Exception {       final long start = System.nanoTime ();       long cnt = iterations ;       while ( cnt & gt ; 0 ) {        spinLock.lock ();        cnt-- ;     spinLock.unlock ();       }       final long end = System.nanoTime ();       return end - start ;      }     } } Таким чином , ми виконуємо N операцій в кожному потоці ( в нашому випадку 2 ^ 24 ) , суміруем час виконання і дивимося середнє для однієї операції . Давайте спочатку подивимося час виконання операції для одного потоку : Average execution time : 1 ns per operation . Тепер подивимося результат для 8 потоків : Average execution time : 1163 ns per operation . Ого ! Різниця просто неймовірна . Таке збільшення часу операції пов'язане , грубо кажучи , з тим , що потоки постійно конкурують за один і той же ділянку пам'яті. Можна трохи уменшить рівень contention між потоками ( test - test - and - set - lock ): public class TestTestAndSetLock {     AtomicBoolean state = new AtomicBoolean ( false ) ;     public void lock ( ) {      while ( true ) {       while ( state.get ( )) { } ;       if (! state.getAndSet ( true ))        return ;      }     }     public void unlock ( ) {      state.set ( false ) ;     } } І результати його бенчмарка :
Average execution time : 948 ns per operation . Що, власне , і варто було очікувати. Загалом, я думаю , ви зрозуміли основну ідею оптимізації . В ідеалі кожен потік повинен мати свою власну клітинку пам'яті , яка міняється тільки раз для індикації того , що ресурс повинен захопити потік . Це приводить нас до CLH -черзі (як це часто буває , назва черги - це абревіатура імен її творців ) . Власне імплементація : public class CLHQueueLock {     private final AtomicReference & lt ; Qnode & gt ; tail = new AtomicReference & lt ; Qnode & gt ; ();     private final ThreadLocal & lt ; Qnode & gt ; myNode = new ThreadLocal & lt ; Qnode & gt ; ( ) {      Override      protected Qnode initialValue ( ) {       return new Qnode ();      }     } ;     private final ThreadLocal & lt ; Qnode & gt ; myPred = new ThreadLocal & lt ; Qnode & gt ; ();     public CLHQueueLock ( ) {      final Qnode qnode = new Qnode ();      qnode.locked = false ;      tail.set ( qnode ) ;     }     public void lock ( ) {      final Qnode localNode = myNode.get ();      localNode.locked = true ;      final Qnode pred = tail.getAndSet ( localNode ) ;      myPred.set ( pred ) ;      while ( pred.locked ) ;     }     public void unlock ( ) {      myNode.get ( ) . locked = false ;      myNode.set ( myPred.get ());     }     static final class Qnode {      volatile boolean locked = true ;     } } І результати бенчмарка : Average execution time : 681 ns per operation . Спочатку зробимо невелику таблицю порівняння результатів , а потім я дам пояснення по деталях реалізації CLH - блокування: Test - and - set test - test - and - set CLH Time/op , ns 1163 948 681 Взагалі - то наш бенчмарк зовсім Ідал - наприклад , ми не дивилися , як зміняться показники при зростанні кількості потоків , не дивилися на девіацію результатів . Хотілося б робити щось більш довгий , ніж декрементірованіе лічильника . Крім того , адже System.nanoTime ( ) теж не безкоштовний , однак його цілком достатньо , щоб зрозуміти основну ідею. Тепер давайте подивимося на прикладі , як працює CLH queue (хоча б тому , що саме її модифікація використовується в OpenJDK ) , на прикладі трьох потоків . Для кожного потоку у нас є QNode , яка показує , відпустив Чи даний потік необхідний ресурс . Стан цієї QNode моніториться потоком , який стоїть за ним у черзі . Тобто при захопленні ресурсу потоки шикуються в чергу , і потік в голові черги встановлює прапор захоплення ресурсу , який моніторить потік , що стоїть за ним , і визначає , чи може очікує потік використовувати необхідний ресурс . Отже , по кроках. Thread_1 викликає метод lock , отримує фейковий QNode передидущего потоку (попереднього потоку просто немає) і поміщає свій QNode в tail . І починає виконувати необхідні йому дії . Причому фейковий QNode стає QNode даного потоку при осовобожденіі блокування. Thread_2 отримує QNode попереднього потоку . І чекає , поки попередній потік не овободіт ресурс , встановивши свій прапор захоплення ресурсу ( locked ) в false . Так само в tail поміщається QNode з прапором locked , встановленим в true . Thread_3 Отримує QNode з tail і чекає вже другий потік . Thread_1 устанавалівает locked прапор в false .
Thread_2 виконує роботу і устанавалівает locked прапор в false .
Thread_3 виконує роботу і устанавалівает locked прапор в false . І на цьому виконання потоків закінчується. CLH queue насправді не завжди виграє в порівнянні з test - test - and - set lock , але більш детальний аналіз блокувань - це вже інша Истроия . До чого , власне, ми все це розглянули ? Хоча б тому , що OpenJDK використовує аналог CLH queue , але , як це не дивно , використовує його для пробудження заблокрірованних потоків , а не для spin lockов . Тобто коли потік звільняє ресурс , він пробуджує потоки , що стоять в черзі . Ми не будемо розглядати повністю , як работет RW lock , але в контексті цієї статті подивимося , як рабоать read частина даної блокування. Коли ми намагаємося заблокувати ресурс на читання , викликається метод public final void acquireShared ( int arg ) {      if ( tryAcquireShared ( arg ) & lt ; 0 )          doAcquireShared ( arg ) ; } У контексті ReadWriteLock метод працює таким чином ( будемо розглядати тільки окремі ділянки коду ) :
ReadWriteLock містить змінну , що зберігає в собі количесвто readers і writers , які захопили даний lock . Вона инкрементируется і декрементируется атомарно , тобто представляє собою , по суті , flag для spin lock . java.util.concurrent.locks.ReentrantReadWriteLock.Sync # fullTryAcquireShared for (;; ) {              int c = getState ();                         // отримали колличество writers              if ( exclusiveCount ( c ) ! = 0 ) {                         // якщо у нас хто то захопив exclusive lock , але не ми доведеться заснути                  if ( getExclusiveOwnerThread ( ) ! = current )                           return -1 ;               } Else if ( readerShouldBlock ( )) {    // все одно спимо наприклад для того , щоб writers могли отримати доступ     // при величезному колличестве readers                             return -1 ;              }                        // інкрементіруем колличество reades . якщо writers немає будемо робити до                        // перемоги              if ( compareAndSetState ( c , c + SHARED_UNIT )) {                               return 1 ;              } } doAcquireShared кладе потік в чергу очікування , намагається захопити read lock (як описано вище) , не виходить - він засинає , прокидається , і т.д . final Node node = addWaiter ( Node.SHARED ) ; for (;; ) {   final Node p = node.predecessor ();   if ( p == head ) {    int r = tryAcquireShared ( arg ) ;    if ( r & gt ; = 0 ) {      return ;   }   if ( shouldParkAfterFailedAcquire ( p , node ) & amp ; & amp ;    parkAndCheckInterrupt ( ))    interrupted = true ; }
Так от, якщо подивитися уважно , то можна побачити , що в разі наявності тільки readers наш ReadWriteLock дуже схожий на test - and - write lock з усіма витікаючими наслідками по втрати продуктивності. Тепер давайте трохи змінимо бенчмарк і зробимо наступне: порахуємо , скажімо , з 16 000 000 до 0 на одному потоці , потім виконаємо те ж число операцій ( 8 потоків по 2 000 000 операцій на кожному) на 8 потоках . А потім порахуємо сумарний час виконання для всіх потоків в наносекундах і час виконання всієї програми в мілісекундах . Адже читають потоки не блокують один одного , правда ? Отже , на 1 потоці : Count down for : 361172637 ns . Execution time is : 362 ms . На 8 потоках : Count down for : 21824811292 ns . Execution time is : 2780 ms . Тобто , грубо кажучи , 8 читаючих потоків працювали в сумі в 9 разів повільніше, ніж один. Тест кілька спекулятивний , але дуже показовий , тому що ми в своєму проекті ( NoSQL бази даних) зіткнулися з подібною ж ситуацією . Коли кілька потоків працювали на читання медленее , ніж один. Взагалі досить сумна перспектива , з урахуванням того , що нам дуже хотілося отримати масшатабірованіе системи на читання . І ми вирішили зробити свій власний ReadWriteLock .
У нас є деякі особливості: ми рідко використовуємо глобальні блокування , і глобальна блокування на запис нам потрібна була в тих , випадках коли :
1 . Швидкість запису була для нас не так критична , як читання .
2 . У нас було багато частих і швидких операцій читання .
3 . Ми працюємо з довгоживучими потоками (хоча , на мій погляд , це не зовсім обов'язково ) . Відразу наведу результат того сумашедшего бенчмарка , який ми прогнали : Count down for : 1221961202 ns . Execution time is : 166 ms . Що в два рази швидше , ніж на одному потоці , не кажучи вже про другому тесті . Якщо у вас ті ж умови , ви можете скористатися нашим досвідом і , власне, самої блокуванням com.orientechnologies.common.concur.lock.OReadersWriterSpinLock в проекті. Як працює даний lock : Write lock так само спочатку є CLH - lock . final WNode node = myNode.get (); node.locked = true ; final WNode pNode = tail.getAndSet ( myNode.get ()); predNode.set ( pNode ) ; while ( pNode.locked ) {   pNode.waitingWriter = Thread.currentThread ();   if ( pNode.locked )     LockSupport.park ( this ) ; } pNode.waitingWriter = null ; Далі нам доведеться перемкнутися на читання . Тут ситуація дещо інша :
кожен reader має совю собсвенно thread local змінну , яка вважає колличество readers для даного потоку . threadCountersHashTable.increment (); Потім ми перевіряємо , захоплений Чи write lock . Якщо так , то ми виконуємо back off , декрементіруя counter для поточного readerа . WNode wNode = tail.get (); while ( wNode.locked ) {     threadCountersHashTable.decrement ();     while ( wNode.locked ) {        wNode.waitingReaders.add ( Thread.currentThread ());        if ( wNode == tail.get ( ) & amp ; & amp ; wNode.locked )        LockSupport.park ( this ) ;         wNode = tail.get ();     }     threadCountersHashTable.increment ();      wNode = tail.get (); } Чекаємо , поки write lock звільниться , і знову інкрементіруем read lock counter для поточного потоку . Дуже важливо відзначити , що таким чином кожен reader инкрементируется свій read counter , і readers не конкурують між собою. Тепер повернемося до захоплення write lock . while (! threadCountersHashTable.isEmpty ( ))    ; setExclusiveOwnerThread ( Thread.currentThread ()); Власне кажучи , write lock чекає , поки всі readers не закінчать свою роботу. Звільнення write lock включає в себе скидання прапора і нотифікацію всіх чекаючих readers , які зареєстрували себе в даному потоці . До речі , ситуація , коли ми зареєстрували себе під wirter потоці , а він вже пробудив усіх зареєстрованих readers , не виникне , так як ми спочатку додаємо себе в чергу очікування wNode.waitingReaders.add ( Thread.currentThread ()); , а засипаємо тільки якщо контролюючий потік все ще не розблокував ресурс . if ( wNode == tail.get ( ) & amp ; & amp ; wNode.locked )        LockSupport.park ( this ) ; Самі по собі thread local counters містяться в HashMap , яка реалізована на основі алгоритму lock free cuckoo hashing . До того ж , це HashMap розглядає комірку з local thread counter , для якої значення Thread.isAlive ( ) = false як порожню , тобто у вас ніколи не буде переповнення пам'яті.

От і все . Якщо ви знайдете помилку в тому , що ми зробили або написали , ми вам будемо дуже вдячні . Якщо Вам цікаво дізнатися набагато більше про багатопотоковому програмуванні , реєструйтеся на наш курс . Крім того , успішно закінчивши курс отримають можливість працевлаштування в orientechnologies в якості core developer , а частина прибутку з курсу буде відправлена ??на благодійність. Все вихідні коди ви можете знайти тут .

Опубліковано: 08/10/14 @ 10:12
Розділ Різне

SEO Блог

Коли ReadWriteLock буває марний

Рекомендуємо: