Освобождение нескольких замков, не вызывающих инверсию приоритета

Question

Краткая версия: Как я могу освободить несколько блокировок из одного потока, не будучи выгруженным на полпути?

У меня есть программа, предназначенная для работы на N-ядерном компьютере. Он состоит из одного основного потока и N рабочих потоков. Каждый поток (включая основной поток) имеет семафор, на котором он может блокироваться. Обычно каждый рабочий поток блокируется при уменьшении его семафора, а основной поток работает. Время от времени основной поток должен разбудить рабочие потоки, чтобы делать свое дело в течение определенного времени, а затем блокировать на своем собственном семафоре, ожидая, что все они вернутся спать. Например:

def main_thread(n): 
    for i = 1 to n: 
     worker_semaphore[i] = semaphore(0) 
     spawn_thread(worker_thread, i) 
    main_semaphore = semaphore(0) 

    while True: 
     ...do some work... 
     workers_to_wake = foo() 
     for i in workers_to_wake: 
      worker_semaphore[i].increment() # wake up worker n 
     for i in workers_to_wake: 
      main_semaphore.decrement() # wait for all workers 

def worker_thread(i): 
    while True: 
     worker_semaphore(i).decrement() # wait to be woken 
     ...do some work... 
     main_semaphore.increment() # report done with step 

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

Возможно, я смогу найти для этого хакеры с ОС и планировщиками, такие как отключение повышения приоритета под Windows, а также поиск приоритетов потоков и сходства процессоров, но мне бы хотелось что-то межплатформенное и надежное и чистый. Итак: Как я могу разбудить кучу нитей атомарно?

Answer 1

Решение Петра Бриттена, а также предложение Антона о «древовидном пробуждении», привело меня к другому решению: «Цепные пробуждения». В основном, а не основной поток, выполняющий все пробуждения, он только просыпает один поток; и затем каждый поток отвечает за пробуждение следующего. Элегантный бит здесь заключается в том, что есть только один подвешенный поток, готовый к запуску, поэтому потоки редко заканчиваются коммутационными ядрами. На самом деле, это прекрасно работает со строгим отношением к процессору, даже если один из рабочих потоков имеет сходство с основным потоком.

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

workers_to_wake = [] 
main_semaphore = semaphore(0) 
num_woken_workers = atomic_integer() 

def main_thread(n): 
    for i = 1 to n: 
     worker_semaphore[i] = semaphore(0) 
     spawn_thread(worker_thread, i) 
    main_semaphore = semaphore(0) 

    while True: 
     ...do some work... 

     workers_to_wake = foo() 
     num_woken_workers.atomic_set(len(workers_to_wake)) # set completion countdown 
     one_to_wake = workers_to_wake.pop() 
     worker_semaphore[one_to_wake].increment() # wake the first worker 
     main_semaphore.decrement() # wait for all workers 

def worker_thread(i): 
    while True: 
     worker_semaphore[i].decrement() # wait to be woken 
     if workers_to_wake.len() > 0: # more pending wakeups 
      one_to_wake = workers_to_wake.pop() 
      worker_semaphore[one_to_wake].increment() # wake the next worker 

     ...do some work... 

     if num_woken_workers.atomic_decrement() == 0: # see whether we're the last one 
      main_semaphore.increment() # report all done with step 

Answer 2

Это невозможно, если вы используете несколько объектов синхронизации (семафоров), когда сложность алгоритма пробуждения равна O (n). Однако есть несколько способов решения проблемы.

релиз сразу

Я не уверен, есть ли Python нужный метод (ваш вопрос Python-конкретнее?), Но, как правило, семафоры имеют операции с аргументом, указывающим число до декрементах/приращений. Таким образом, вы просто помещаете все свои потоки в один и тот же семафор и разбудите их все вместе. Аналогичный подход заключается в использовании условной переменной и notify all.

событие петли

Если вы все еще хотите, чтобы иметь возможность контролировать каждый поток по отдельности, но, как подход с уведомлением один-ко-многим, попробуйте библиотеки для асинхронного ввода/вывода, как libuv (и its Python counterpart). Здесь вы можете создать одно событие, которое пробуждает все потоки сразу, а также создаст для каждого потока его индивидуальное событие, а затем просто подождите на обоих (или более) объектах событий в циклах событий в каждом потоке. Другая библиотека - pevents, которая реализует WaitForMultipleObjects поверх условных переменных pthreads.

делегат пробуждения

Другой подход заключается заменить алгоритм O (N) с древовидного алгоритма (O (журнал N)), где каждый поток просыпается только фиксированное число других потоков, но делегатов, чтобы они просыпаются - другие. В краевом случае основной поток может разбудить только один другой поток, который пробудит всех остальных или начнет цепную реакцию. Это может быть полезно, если вы хотите уменьшить латентность для основного потока за счет сеансов пробуждения других потоков.

Answer 3

Reader/Writer Замок

Решения, которое я обычно использую в системах POSIX для одного до многих отношений является блокировкой чтения/записи.Это удивительно для меня, что они не являются полными универсальными, но в большинстве языков либо реализовать версию, или, по крайней мере, есть пакет, доступный для реализации их на любые примитивы существует, например, в Python prwlock:

from prwlock import RWLock 

def main_thread(n): 
    for i = 1 to n: 
     worker_semaphore[i] = semaphore(0) 
     spawn_thread(worker_thread, i) 
    main_lock = RWLock() 

    while True: 
     main_lock.acquire_write() 
     ...do some work... 
     workers_to_wake = foo() 
     # The above acquire could be moved as low as here, 
     # depending on how independent the above processing is..    
     for i in workers_to_wake: 
      worker_semaphore[i].increment() # wake up worker n 

     main_lock.release() 


def worker_thread(i): 
    while True: 
     worker_semaphore(i).decrement() # wait to be woken 
     main_lock.acquire_read() 
     ...do some work... 
     main_lock.release() # report done with step 

Барьеры

Barriers, кажется, как ближайший предназначенный встроенный механизм Python, чтобы держать все нити, пока все они не будут предупреждены, но:

1. они довольно необычное решение, таким образом, они бы вас r code/сложнее перевести на другие языки.

2. Я бы не хотел использовать их для этого случая, когда количество потоков, которые нужно просыпаться, продолжает меняться. Учитывая, что ваши n звучат мало, у меня возникнет соблазн использовать константу Barrier(n) и уведомлять обо всех потоках, чтобы проверить, работает ли этот цикл. Но:

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

Answer 4

TL; DR

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

Контекст

Сначала мне нужно подвести итог более широкий контекст, в нашей дискуссии ...

У вас есть графическое приложение Windows. Он имеет желаемую частоту кадров, поэтому вам нужно, чтобы основной поток выполнялся с такой скоростью, планируя всех ваших сотрудников точно с интервалом времени, чтобы они завершили свою работу в пределах интервала обновления. Это означает, что у вас очень плотные ограничения на время запуска и выполнения для каждого потока. Кроме того, ваши рабочие потоки не все одинаковы, поэтому вы не можете просто использовать одну рабочую очередь.

Проблема

Как и любой современной операционной системы, Windows имеет множество synchronization primitives. Тем не менее, ни один из них напрямую не обеспечивает механизм для уведомления нескольких примитивов сразу. Просматривая другие операционные системы, я вижу аналогичную модель; все они предоставляют способы ожидания на нескольких примитивах, но ни один из них не дает атомный способ , вызывая.

Так что же мы можем сделать вместо этого? Проблемы, которые вам необходимо решить, это:

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

Опция

Наиболее очевидное решение вопроса-просто использовать один семафор событий, но вы также можете использовать/блокировки чтения и записи (путем приобретения блокировки записи после того, как рабочие закончили и получение рабочим использовать блокировку чтения). Все остальные параметры больше не являются атомарными, поэтому потребуется дополнительная синхронизация, чтобы заставить потоки делать то, что вы хотите - например, предложение getleader для блокировок внутри ваших семафоров.

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

Ну ... Блокировка чтения/записи - отличный способ для одного потока записать некоторые важные данные в блок управления и для многих других читать с него. Почему бы просто не иметь простой массив булевых флагов (по одному для каждого рабочего потока), чтобы ваш основной поток обновлял каждый кадр? К сожалению, вам все равно нужно прекратить выполнение рабочих, пока не появится таймер. Короче говоря, мы снова возвращаемся к семафору и блокируем решение.

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

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

Раствор

Применяя выше, ваш псевдо-код будет выглядеть примерно так:

def main_thread(n): 
    main_event = event() 
    for i = 1 to n: 
     worker_scheduled[i] = False 
     spawn_thread(worker_thread, i) 
    main_barrier = barrier(n+1) 

    while True: 
     ...do some work... 
     workers_to_wake = foo() 
     for i in workers_to_wake: 
      worker_scheduled[i] = True 
     main_event.set() 
     main_barrier.enter() # wait for all workers 
     main_event.reset() 

def worker_thread(i): 
    while True: 
     main_event.wait() 
     if worker_scheduled[i]: 
      worker_scheduled[i] = False 
      ...do some work... 
     main_barrier.enter() # report finished for this frame. 
     main_event.reset() # to catch the case that a worker is scheduled before the main thread 

Поскольку нет явного полицейская из worker_scheduled массива, это гораздо более хрупкими решение ,

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