Монитор (синхронизация)

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

Другое определение монитора — это потокобезопасный класс , объект или модуль , который оборачивается вокруг мьютекса , чтобы безопасно разрешить доступ к методу или переменной более чем одному потоку . Определяющей характеристикой монитора является то, что его методы выполняются с принципом взаимного исключения : в каждый момент времени не более одного потока может выполнять любой из его методов . Используя одну или несколько переменных условия, он также может предоставить потокам возможность ожидать определенного условия (таким образом, используя приведенное выше определение «монитора»). В оставшейся части статьи этот смысл «монитора» будет называться «потокобезопасным объектом/классом/модулем».

Мониторы были изобретены Пером Бринчом Хансеном. и АВТОМОБИЛЬ Хоар , и впервые были реализованы на Бринча Хансена языке Concurrent Pascal .

Взаимное исключение

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

После вызова одного из методов поток должен дождаться, пока ни один другой поток не выполнит ни один из методов потокобезопасного объекта, прежде чем начать выполнение своего метода. Обратите внимание, что без этого взаимного исключения два потока могут привести к потере или получению денег без всякой причины. Например, два потока, снимающие 1000 со счета, могут оба вернуть true, в то время как баланс упадет только на 1000, следующим образом: сначала оба потока извлекают текущий баланс, обнаруживают, что он больше 1000, и вычитают из него 1000; затем оба потока сохраняют баланс и возвращаются.

Условные переменные

Постановка задачи

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

while not ( P ) do skip

не будет работать, поскольку взаимное исключение не позволит любому другому потоку войти в монитор, чтобы условие стало истинным. Существуют и другие «решения», такие как цикл, который разблокирует монитор, ждет определенное время, блокирует монитор и проверяет условие P . Теоретически работает и тупика не будет, но возникают проблемы. Трудно определить подходящее время ожидания: слишком маленькое — и поток перегружает процессор, слишком большое — и поток явно не будет отвечать. Что необходимо, так это способ сигнализировать потоку, когда условие P истинно (или может быть истинным).

Практический пример: классическая ограниченная проблема производителя/потребителя

Классическая проблема параллелизма — это проблема ограниченного производителя/потребителя , в которой существует очередь или кольцевой буфер задач максимального размера, причем один или несколько потоков являются потоками «производителей», которые добавляют задачи в очередь, а один или несколько потоков — «производителей», которые добавляют задачи в очередь. другие потоки являются «потребительскими» потоками, которые извлекают задачи из очереди. Предполагается, что очередь сама по себе не является потокобезопасной и может быть пустой, полной или между пустой и полной. Всякий раз, когда очередь полна задач, нам нужно, чтобы потоки-производители блокировались до тех пор, пока не останется место для задач, выводящих из очереди потоки-потребители. С другой стороны, всякий раз, когда очередь пуста, нам нужно, чтобы потребительские потоки блокировались до тех пор, пока не станут доступны новые задачи из-за их добавления потоками-производителями.

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

Неверно без синхронизации

Наивный подход заключается в разработке кода с ожиданием занятости и отсутствием синхронизации, что делает код подверженным условиям гонки:

global RingBuffer queue; // A thread-unsafe ring-buffer of tasks.

// Method representing each producer thread's behavior:
public method producer() {
    while (true) {
        task myTask = ...; // Producer makes some new task to be added.
        while (queue.isFull()) {} // Busy-wait until the queue is non-full.
        queue.enqueue(myTask); // Add the task to the queue.
    }
}

// Method representing each consumer thread's behavior:
public method consumer() {
    while (true) {
        while (queue.isEmpty()) {} // Busy-wait until the queue is non-empty.
        myTask = queue.dequeue(); // Take a task off of the queue.
        doStuff(myTask); // Go off and do something with the task.
    }
}

В этом коде есть серьезная проблема: доступ к очереди может прерываться и чередоваться с доступом к очереди других потоков. Методыqueue.enqueue , иqueue.dequeue вероятно , содержат инструкции по обновлению переменных-членов очереди, таких как ее размер, начальная и конечная позиции, назначение и размещение элементов очереди и т. д. Кроме того, очереди.isEmpty() и очередь.isFull () также читают это общее состояние. Если потокам-производителям и потребителям разрешено чередование во время вызовов постановки в очередь или удаления из очереди, то может проявиться несогласованное состояние очереди, что приведет к условиям гонки. Кроме того, если один потребитель опустошает очередь между выходом другого потребителя из режима ожидания занятости и вызовом «удаления из очереди», то второй потребитель попытается исключить из очереди пустую очередь, что приведет к ошибке. Аналогично, если производитель заполняет очередь между выходом другого производителя из режима ожидания занятости и вызовом «очереди», тогда второй производитель попытается добавить к полной очереди, что приведет к ошибке.

Ожидание вращения

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

global RingBuffer queue; // A thread-unsafe ring-buffer of tasks.
global Lock queueLock; // A mutex for the ring-buffer of tasks.

// Method representing each producer thread's behavior:
public method producer() {
    while (true) {
        task myTask = ...; // Producer makes some new task to be added.

        queueLock.acquire(); // Acquire lock for initial busy-wait check.
        while (queue.isFull()) { // Busy-wait until the queue is non-full.
            queueLock.release();
            // Drop the lock temporarily to allow a chance for other threads
            // needing queueLock to run so that a consumer might take a task.
            queueLock.acquire(); // Re-acquire the lock for the next call to "queue.isFull()".
        }

        queue.enqueue(myTask); // Add the task to the queue.
        queueLock.release(); // Drop the queue lock until we need it again to add the next task.
    }
}

// Method representing each consumer thread's behavior:
public method consumer() {
    while (true) {
        queueLock.acquire(); // Acquire lock for initial busy-wait check.
        while (queue.isEmpty()) { // Busy-wait until the queue is non-empty.
            queueLock.release();
            // Drop the lock temporarily to allow a chance for other threads
            // needing queueLock to run so that a producer might add a task.
            queueLock.acquire(); // Re-acquire the lock for the next call to "queue.isEmpty()".
        }
        myTask = queue.dequeue(); // Take a task off of the queue.
        queueLock.release(); // Drop the queue lock until we need it again to take off the next task.
        doStuff(myTask); // Go off and do something with the task.
    }
}

Этот метод гарантирует, что несогласованное состояние не возникнет, но тратит ресурсы ЦП из-за ненужного ожидания занятости. Даже если очередь пуста и потокам-производителям нечего добавить в течение длительного времени, потоки-потребители всегда заняты ожиданием без необходимости. Аналогично, даже если потребители на долгое время заблокированы при обработке своих текущих задач и очередь заполнена, производители всегда заняты ожиданием. Это расточительный механизм. Что необходимо, так это способ блокировать потоки-производители до тех пор, пока очередь не переполнится, и способ блокировать потоки-потребители до тех пор, пока очередь не станет непустой.

(Примечание: мьютексы сами по себе также могут представлять собой спин-блокировки , которые включают ожидание занятости для получения блокировки, но чтобы решить проблему нерациональной траты ресурсов ЦП, мы предполагаем, что очередь QueueLock не является спин-блокировкой и правильно использует блокировку. заблокировать саму очередь.)

Условные переменные

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

Таким образом, над условными переменными выполняются три основные операции:

• wait c, m, где c является условной переменной и m — это мьютекс (блокировка), связанный с монитором. Эта операция вызывается потоком, которому необходимо дождаться, пока утверждение P _c станет истинным, прежде чем продолжить. Пока поток ожидает, он не занимает монитор. Функция и фундаментальный контракт операции ожидания заключается в выполнении следующих шагов:
  1. Атомно :

     1. освободить мьютекс m,
     2. перенесите эту тему из "работающей" в c«очередь ожидания» (также известная как «очередь сна») потоков и
     3. спи эту тему. (Контекст синхронно передается другому потоку.)

  2. Как только этот поток впоследствии будет уведомлен/сигнализирован (см. ниже) и возобновлен, он автоматически повторно захватит мьютекс. m.

  Шаги 1a и 1b могут выполняться в любом порядке, после них обычно выполняется шаг 1c. Пока поток спит и находится в cочереди ожидания, следующий счетчик программы, который будет выполнен, находится на шаге 2, в середине функции/ подпрограммы ожидания . Таким образом, поток засыпает, а затем просыпается в середине операции ожидания.

  Атомарность операций на шаге 1 важна, чтобы избежать состояний гонки, которые могут быть вызваны упреждающим переключением потоков между ними. Один из режимов сбоя, который мог бы возникнуть, если бы они не были атомарными, — это пропущенное пробуждение , при котором поток мог бы быть включен. cочередь сна и освободили мьютекс, но упреждающее переключение потока произошло до того, как поток перешел в спящий режим, и другой поток вызвал сигнальную операцию (см. ниже) в c перемещение первой нити обратно из cочередь. Как только первый поток, о котором идет речь, переключится обратно, его программный счетчик окажется на шаге 1c, и он заснет и не сможет снова разбудиться, нарушив инвариант, в котором он должен был быть включен. cспит-очередь, когда она спала. Другие условия гонки зависят от порядка шагов 1a и 1b и от того, где происходит переключение контекста.

• signal c, также известный как notify c, вызывается потоком, чтобы указать, что утверждение P _c истинно. В зависимости от типа и реализации монитора это перемещает один или несколько потоков из cочередь сна в «очередь готовности» или другую очередь для ее выполнения. Обычно считается лучшей практикой выполнять операцию «сигнала» перед освобождением мьютекса. m что связано с c, но если код правильно спроектирован для параллелизма и в зависимости от реализации потоков, часто также приемлемо снять блокировку перед передачей сигнала. В зависимости от реализации потоков, их порядок может иметь последствия с приоритетом планирования. (Некоторые авторы вместо этого отстаивайте предпочтение снятию блокировки перед передачей сигнала.) Реализация потоков должна документировать любые специальные ограничения на этот порядок.
• broadcast c, также известный как notifyAll c, — аналогичная операция, которая пробуждает все потоки в очереди ожидания c. Это очищает очередь ожидания. Как правило, когда с одной и той же переменной условия связано более одного предикатного условия, приложению потребуется широковещательная рассылка вместо сигнала , поскольку поток, ожидающий неправильного условия, может быть разбужен, а затем немедленно вернуться в спящий режим, не пробуждая поток, ожидающий правильное условие, которое только что стало истинным. В противном случае, если условие предиката однозначно соответствует связанной с ним переменной условия, сигнал может быть более эффективным, чем Broadcast .

Как правило, несколько условных переменных могут быть связаны с одним и тем же мьютексом, но не наоборот. (Это соответствие «один ко многим» .) Это связано с тем, что предикат P _c одинаков для всех потоков, использующих монитор, и должен быть защищен взаимным исключением от всех других потоков, которые могут вызвать изменение условия или которые могут прочитайте его, пока рассматриваемый поток вызывает его изменение, но могут быть разные потоки, которые хотят дождаться другого условия для одной и той же переменной, требующей использования одного и того же мьютекса. примере производитель-потребитель В описанном выше очередь должна быть защищена уникальным объектом мьютекса, m. Потоки-производители захотят ожидать на мониторе, используя блокировку m и переменная условия $c_{full}$ который блокируется до тех пор, пока очередь не переполнится. «Потребительские» потоки захотят ожидать на другом мониторе, используя тот же мьютекс. m но другая переменная условия $c_{empty}$ который блокируется до тех пор, пока очередь не станет непустой. (Обычно) никогда не имеет смысла иметь разные мьютексы для одной и той же условной переменной, но этот классический пример показывает, почему часто имеет смысл иметь несколько условных переменных, использующих один и тот же мьютекс. Мьютекс, используемый одной или несколькими условными переменными (одним или несколькими мониторами), также может использоваться совместно с кодом, который не использует условные переменные (и который просто получает/освобождает его без каких-либо операций ожидания/сигнала), если эти критические разделы не происходят. требовать ожидания определенного условия для одновременных данных.

Мониторинг использования

Правильное базовое использование монитора:

acquire(m); // Acquire this monitor's lock.
while (!p) { // While the condition/predicate/assertion that we are waiting for is not true...
	wait(m, cv); // Wait on this monitor's lock and condition variable.
}
// ... Critical section of code goes here ...
signal(cv2); // Or: broadcast(cv2);
             // cv2 might be the same as cv or different.
release(m); // Release this monitor's lock.

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

// ... (previous code)
// About to enter the monitor.
// Acquire the advisory mutex (lock) associated with the concurrent
// data that is shared between threads, 
// to ensure that no two threads can be preemptively interleaved or
// run simultaneously on different cores while executing in critical
// sections that read or write this same concurrent data. If another
// thread is holding this mutex, then this thread will be put to sleep
// (blocked) and placed on m's sleep queue.  (Mutex "m" shall not be
// a spin-lock.)
acquire(m);
// Now, we are holding the lock and can check the condition for the
// first time.

// The first time we execute the while loop condition after the above
// "acquire", we are asking, "Does the condition/predicate/assertion
// we are waiting for happen to already be true?"

while (!p()) 	// "p" is any expression (e.g. variable or 
		// function-call) that checks the condition and
		// evaluates to boolean.  This itself is a critical
		// section, so you *MUST* be holding the lock when
		// executing this "while" loop condition!
				
// If this is not the first time the "while" condition is being checked,
// then we are asking the question, "Now that another thread using this
// monitor has notified me and woken me up and I have been context-switched
// back to, did the condition/predicate/assertion we are waiting on stay
// true between the time that I was woken up and the time that I re-acquired
// the lock inside the "wait" call in the last iteration of this loop, or
// did some other thread cause the condition to become false again in the
// meantime thus making this a spurious wakeup?

{
	// If this is the first iteration of the loop, then the answer is
	// "no" -- the condition is not ready yet. Otherwise, the answer is:
	// the latter.  This was a spurious wakeup, some other thread occurred
	// first and caused the condition to become false again, and we must
	// wait again.

	wait(m, cv);
		// Temporarily prevent any other thread on any core from doing
		// operations on m or cv.
		// release(m) 		// Atomically release lock "m" so other
		//			// code using this concurrent data
		// 			// can operate, move this thread to cv's
		//			// wait-queue so that it will be notified
		//			// sometime when the condition becomes
		// 			// true, and sleep this thread. Re-enable
		//			// other threads and cores to do 
		//			// operations on m and cv.
		//
		// Context switch occurs on this core.
		//
		// At some future time, the condition we are waiting for becomes
		// true, and another thread using this monitor (m, cv) does either
		// a signal that happens to wake this thread up, or a
		// broadcast that wakes us up, meaning that we have been taken out
		// of cv's wait-queue.
		//
		// During this time, other threads may cause the condition to
		// become false again, or the condition may toggle one or more
		// times, or it may happen to stay true.
		//
		// This thread is switched back to on some core.
		//
		// acquire(m)		// Lock "m" is re-acquired.
		
	// End this loop iteration and re-check the "while" loop condition to make
	// sure the predicate is still true.
	
}

// The condition we are waiting for is true!
// We are still holding the lock, either from before entering the monitor or from
// the last execution of "wait".

// Critical section of code goes here, which has a precondition that our predicate
// must be true.
// This code might make cv's condition false, and/or make other condition variables'
// predicates true.

// Call signal or broadcast, depending on which condition variables'
// predicates (who share mutex m) have been made true or may have been made true,
// and the monitor semantic type being used.

for (cv_x in cvs_to_signal) {
	signal(cv_x); // Or: broadcast(cv_x);
}
// One or more threads have been woken up but will block as soon as they try
// to acquire m.

// Release the mutex so that notified thread(s) and others can enter their critical
// sections.
release(m);

Решение ограниченной задачи производителя/потребителя

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

global volatile RingBuffer queue; // A thread-unsafe ring-buffer of tasks.
global Lock queueLock; // A mutex for the ring-buffer of tasks. (Not a spin-lock.)
global CV queueEmptyCV; // A condition variable for consumer threads waiting for the queue to 
				        // become non-empty. Its associated lock is "queueLock".
global CV queueFullCV; // A condition variable for producer threads waiting for the queue to
				       // become non-full. Its associated lock is also "queueLock".

// Method representing each producer thread's behavior:
public method producer() {
    while (true) {
        // Producer makes some new task to be added.
        task myTask = ...;

        // Acquire "queueLock" for the initial predicate check.
        queueLock.acquire();

        // Critical section that checks if the queue is non-full.
        while (queue.isFull()) {
            // Release "queueLock", enqueue this thread onto "queueFullCV" and sleep this thread.
            wait(queueLock, queueFullCV);
            // When this thread is awoken, re-acquire "queueLock" for the next predicate check.
        }

        // Critical section that adds the task to the queue (note that we are holding "queueLock").
        queue.enqueue(myTask);

        // Wake up one or all consumer threads that are waiting for the queue to be non-empty
        // now that it is guaranteed, so that a consumer thread will take the task.
        signal(queueEmptyCV); // Or: broadcast(queueEmptyCV);
        // End of critical sections.

        // Release "queueLock" until we need it again to add the next task.
        queueLock.release();
    }
}

// Method representing each consumer thread's behavior:
public method consumer() {
    while (true) {
        // Acquire "queueLock" for the initial predicate check.
        queueLock.acquire();

        // Critical section that checks if the queue is non-empty.
        while (queue.isEmpty()) {
            // Release "queueLock", enqueue this thread onto "queueEmptyCV" and sleep this thread.
            wait(queueLock, queueEmptyCV);
            // When this thread is awoken, re-acquire "queueLock" for the next predicate check.
        }

        // Critical section that takes a task off of the queue (note that we are holding "queueLock").
        myTask = queue.dequeue();

        // Wake up one or all producer threads that are waiting for the queue to be non-full
        // now that it is guaranteed, so that a producer thread will add a task.
        signal(queueFullCV); // Or: broadcast(queueFullCV);
        // End of critical sections.

        // Release "queueLock" until we need it again to take the next task.
        queueLock.release();

        // Go off and do something with the task.
        doStuff(myTask);
    }
}

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

Вариант этого решения может использовать одну переменную условия как для производителей, так и для потребителей, возможно, с именем «queueFullOrEmptyCV» или «queueSizeChangedCV». В этом случае с переменной условия связано более одного условия, так что переменная условия представляет собой более слабое условие, чем условия, проверяемые отдельными потоками. Условная переменная представляет потоки, ожидающие, пока очередь не заполнится , и потоки, ожидающие, пока она не станет пустой. Однако для этого потребуется использовать широковещательную рассылку во всех потоках с использованием условной переменной и нельзя использовать обычный сигнал . Это связано с тем, что обычный сигнал может разбудить поток неправильного типа, условие которого еще не выполнено, и этот поток снова заснет без получения сигнала потока правильного типа. Например, производитель может заполнить очередь и разбудить другого производителя вместо потребителя, а проснувшийся производитель снова перейдет в режим сна. В дополнительном случае потребитель может опустошить очередь и разбудить другого потребителя вместо производителя, и потребитель снова перейдет в режим сна. С использованием Broadcast гарантирует, что некоторый поток правильного типа будет работать так, как ожидается в условии задачи.

Вот вариант, использующий только одну условную переменную и трансляцию:

global volatile RingBuffer queue; // A thread-unsafe ring-buffer of tasks.
global Lock queueLock; // A mutex for the ring-buffer of tasks.  (Not a spin-lock.)
global CV queueFullOrEmptyCV; // A single condition variable for when the queue is not ready for any thread
                              // i.e. for producer threads waiting for the queue to become non-full 
                              // and consumer threads waiting for the queue to become non-empty.
                              // Its associated lock is "queueLock".
                              // Not safe to use regular "signal" because it is associated with
                              // multiple predicate conditions (assertions).

// Method representing each producer thread's behavior:
public method producer() {
    while (true) {
        // Producer makes some new task to be added.
        task myTask = ...;

        // Acquire "queueLock" for the initial predicate check.
        queueLock.acquire();

        // Critical section that checks if the queue is non-full.
        while (queue.isFull()) {
            // Release "queueLock", enqueue this thread onto "queueFullOrEmptyCV" and sleep this thread.
            wait(queueLock, queueFullOrEmptyCV);
            // When this thread is awoken, re-acquire "queueLock" for the next predicate check.
        }

        // Critical section that adds the task to the queue (note that we are holding "queueLock").
        queue.enqueue(myTask);

        // Wake up all producer and consumer threads that are waiting for the queue to be respectively
        // non-full and non-empty now that the latter is guaranteed, so that a consumer thread will take the task.
        broadcast(queueFullOrEmptyCV); // Do not use "signal" (as it might wake up another producer thread only).
        // End of critical sections.

        // Release "queueLock" until we need it again to add the next task.
        queueLock.release();
    }
}

// Method representing each consumer thread's behavior:
public method consumer() {
    while (true) {
        // Acquire "queueLock" for the initial predicate check.
        queueLock.acquire();

        // Critical section that checks if the queue is non-empty.
        while (queue.isEmpty()) {
            // Release "queueLock", enqueue this thread onto "queueFullOrEmptyCV" and sleep this thread.
            wait(queueLock, queueFullOrEmptyCV);
            // When this thread is awoken, re-acquire "queueLock" for the next predicate check.
        }

        // Critical section that takes a task off of the queue (note that we are holding "queueLock").
        myTask = queue.dequeue();

        // Wake up all producer and consumer threads that are waiting for the queue to be respectively
        // non-full and non-empty now that the former is guaranteed, so that a producer thread will add a task.
        broadcast(queueFullOrEmptyCV); // Do not use "signal" (as it might wake up another consumer thread only).
        // End of critical sections.

        // Release "queueLock" until we need it again to take the next task.
        queueLock.release();

        // Go off and do something with the task.
        doStuff(myTask);
    }
}

Примитивы синхронизации

Мониторы реализованы с использованием атомарного примитива чтения-изменения-записи и примитива ожидания. Примитив чтения-изменения-записи (обычно тест-и-установка или сравнение-и-замена) обычно имеет форму инструкции блокировки памяти, предоставляемой ISA , но также может состоять из неблокирующих инструкций на одиночном процессоре. процессорные устройства, когда прерывания отключены. Примитив ожидания может быть циклом ожидания занятости потока или примитивом, предоставляемым ОС, который предотвращает планирование до тех пор, пока он не будет готов продолжить работу.

Вот пример реализации псевдокода частей системы потоков, мьютексов и условных переменных в стиле Mesa с использованием политики «проверь и установи» и политики «первым пришел — первым обслужен»:

Пример реализации Mesa-монитора с помощью Test-and-Set

// Basic parts of threading system:
// Assume "ThreadQueue" supports random access.
public volatile ThreadQueue readyQueue; // Thread-unsafe queue of ready threads.  Elements are (Thread*).
public volatile global Thread* currentThread; // Assume this variable is per-core.  (Others are shared.)

// Implements a spin-lock on just the synchronized state of the threading system itself.
// This is used with test-and-set as the synchronization primitive.
public volatile global bool threadingSystemBusy = false;

// Context-switch interrupt service routine (ISR):
// On the current CPU core, preemptively switch to another thread.
public method contextSwitchISR() {
    if (testAndSet(threadingSystemBusy)) {
        return; // Can't switch context right now.
    }

    // Ensure this interrupt can't happen again which would foul up the context switch:
    systemCall_disableInterrupts();

    // Get all of the registers of the currently-running process.
    // For Program Counter (PC), we will need the instruction location of
    // the "resume" label below.  Getting the register values is platform-dependent and may involve
    // reading the current stack frame, JMP/CALL instructions, etc.  (The details are beyond this scope.)
    currentThread->registers = getAllRegisters(); // Store the registers in the "currentThread" object in memory.
    currentThread->registers.PC = resume; // Set the next PC to the "resume" label below in this method.

    readyQueue.enqueue(currentThread); // Put this thread back onto the ready queue for later execution.
    
    Thread* otherThread = readyQueue.dequeue(); // Remove and get the next thread to run from the ready queue.
    
    currentThread = otherThread; // Replace the global current-thread pointer value so it is ready for the next thread.

    // Restore the registers from currentThread/otherThread, including a jump to the stored PC of the other thread
    // (at "resume" below).  Again, the details of how this is done are beyond this scope.
    restoreRegisters(otherThread.registers);

    // *** Now running "otherThread" (which is now "currentThread")!  The original thread is now "sleeping". ***

    resume: // This is where another contextSwitch() call needs to set PC to when switching context back here.

    // Return to where otherThread left off.

    threadingSystemBusy = false; // Must be an atomic assignment.
    systemCall_enableInterrupts(); // Turn pre-emptive switching back on on this core.
}

// Thread sleep method:
// On current CPU core, a synchronous context switch to another thread without putting
// the current thread on the ready queue.
// Must be holding "threadingSystemBusy" and disabled interrupts so that this method
// doesn't get interrupted by the thread-switching timer which would call contextSwitchISR().
// After returning from this method, must clear "threadingSystemBusy".
public method threadSleep() {
    // Get all of the registers of the currently-running process.
    // For Program Counter (PC), we will need the instruction location of
    // the "resume" label below.  Getting the register values is platform-dependent and may involve
    // reading the current stack frame, JMP/CALL instructions, etc.  (The details are beyond this scope.)
    currentThread->registers = getAllRegisters(); // Store the registers in the "currentThread" object in memory.
    currentThread->registers.PC = resume; // Set the next PC to the "resume" label below in this method.

    // Unlike contextSwitchISR(), we will not place currentThread back into readyQueue.
    // Instead, it has already been placed onto a mutex's or condition variable's queue.
    
    Thread* otherThread = readyQueue.dequeue(); // Remove and get the next thread to run from the ready queue.
    
    currentThread = otherThread; // Replace the global current-thread pointer value so it is ready for the next thread.

    // Restore the registers from currentThread/otherThread, including a jump to the stored PC of the other thread
    // (at "resume" below).  Again, the details of how this is done are beyond this scope.
    restoreRegisters(otherThread.registers);

    // *** Now running "otherThread" (which is now "currentThread")!  The original thread is now "sleeping". ***

    resume: // This is where another contextSwitch() call needs to set PC to when switching context back here.

    // Return to where otherThread left off.
}

public method wait(Mutex m, ConditionVariable c) {
    // Internal spin-lock while other threads on any core are accessing this object's
    // "held" and "threadQueue", or "readyQueue".
    while (testAndSet(threadingSystemBusy)) {}
    // N.B.: "threadingSystemBusy" is now true.
    
    // System call to disable interrupts on this core so that threadSleep() doesn't get interrupted by
    // the thread-switching timer on this core which would call contextSwitchISR().
    // Done outside threadSleep() for more efficiency so that this thread will be sleeped
    // right after going on the condition-variable queue.
    systemCall_disableInterrupts();
 
    assert m.held; // (Specifically, this thread must be the one holding it.)
    
    m.release();
    c.waitingThreads.enqueue(currentThread);
    
    threadSleep();
    
    // Thread sleeps ... Thread gets woken up from a signal/broadcast.
    
    threadingSystemBusy = false; // Must be an atomic assignment.
    systemCall_enableInterrupts(); // Turn pre-emptive switching back on on this core.
    
    // Mesa style:
    // Context switches may now occur here, making the client caller's predicate false.
    
    m.acquire();
}

public method signal(ConditionVariable c) {
    // Internal spin-lock while other threads on any core are accessing this object's
    // "held" and "threadQueue", or "readyQueue".
    while (testAndSet(threadingSystemBusy)) {}
    // N.B.: "threadingSystemBusy" is now true.
    
    // System call to disable interrupts on this core so that threadSleep() doesn't get interrupted by
    // the thread-switching timer on this core which would call contextSwitchISR().
    // Done outside threadSleep() for more efficiency so that this thread will be sleeped
    // right after going on the condition-variable queue.
    systemCall_disableInterrupts();
    
    if (!c.waitingThreads.isEmpty()) {
        wokenThread = c.waitingThreads.dequeue();
        readyQueue.enqueue(wokenThread);
    }
    
    threadingSystemBusy = false; // Must be an atomic assignment.
    systemCall_enableInterrupts(); // Turn pre-emptive switching back on on this core.
    
    // Mesa style:
    // The woken thread is not given any priority.
}

public method broadcast(ConditionVariable c) {
    // Internal spin-lock while other threads on any core are accessing this object's
    // "held" and "threadQueue", or "readyQueue".
    while (testAndSet(threadingSystemBusy)) {}
    // N.B.: "threadingSystemBusy" is now true.
    
    // System call to disable interrupts on this core so that threadSleep() doesn't get interrupted by
    // the thread-switching timer on this core which would call contextSwitchISR().
    // Done outside threadSleep() for more efficiency so that this thread will be sleeped
    // right after going on the condition-variable queue.
    systemCall_disableInterrupts();
    
    while (!c.waitingThreads.isEmpty()) {
        wokenThread = c.waitingThreads.dequeue();
        readyQueue.enqueue(wokenThread);
    }
    
    threadingSystemBusy = false; // Must be an atomic assignment.
    systemCall_enableInterrupts(); // Turn pre-emptive switching back on on this core.
    
    // Mesa style:
    // The woken threads are not given any priority.
}

class Mutex {
    protected volatile bool held = false;
    private volatile ThreadQueue blockingThreads; // Thread-unsafe queue of blocked threads.  Elements are (Thread*).
    
    public method acquire() {
        // Internal spin-lock while other threads on any core are accessing this object's
        // "held" and "threadQueue", or "readyQueue".
        while (testAndSet(threadingSystemBusy)) {}
        // N.B.: "threadingSystemBusy" is now true.
        
        // System call to disable interrupts on this core so that threadSleep() doesn't get interrupted by
        // the thread-switching timer on this core which would call contextSwitchISR().
        // Done outside threadSleep() for more efficiency so that this thread will be sleeped
        // right after going on the lock queue.
        systemCall_disableInterrupts();

        assert !blockingThreads.contains(currentThread);

        if (held) {
            // Put "currentThread" on this lock's queue so that it will be
            // considered "sleeping" on this lock.
            // Note that "currentThread" still needs to be handled by threadSleep().
            readyQueue.remove(currentThread);
            blockingThreads.enqueue(currentThread);
            threadSleep();
            
            // Now we are woken up, which must be because "held" became false.
            assert !held;
            assert !blockingThreads.contains(currentThread);
        }
        
        held = true;
        
        threadingSystemBusy = false; // Must be an atomic assignment.
        systemCall_enableInterrupts(); // Turn pre-emptive switching back on on this core.
    }        
        
    public method release() {
        // Internal spin-lock while other threads on any core are accessing this object's
        // "held" and "threadQueue", or "readyQueue".
        while (testAndSet(threadingSystemBusy)) {}
        // N.B.: "threadingSystemBusy" is now true.
        
        // System call to disable interrupts on this core for efficiency.
        systemCall_disableInterrupts();
        
        assert held; // (Release should only be performed while the lock is held.)

        held = false;
        
        if (!blockingThreads.isEmpty()) {
            Thread* unblockedThread = blockingThreads.dequeue();
            readyQueue.enqueue(unblockedThread);
        }
        
        threadingSystemBusy = false; // Must be an atomic assignment.
        systemCall_enableInterrupts(); // Turn pre-emptive switching back on on this core.
    }
}

struct ConditionVariable {
    volatile ThreadQueue waitingThreads;
}

Блокирующие переменные условия

Первоначальные предложения К.А.Р. Хоара и Пера Бринча Хансена касались блокировки условных переменных . При использовании переменной условия блокировки сигнальный поток должен ждать за пределами монитора (по крайней мере), пока сигнальный поток не освободит монитор, либо вернувшись, либо снова ожидая условную переменную. Мониторы, использующие переменные условия блокировки, часто называются мониторами в стиле Хоара или мониторами сигналов и срочного ожидания .

Монитор в стиле Хоара с двумя переменными состояния a и b. После того, как Бур и др.

Мы предполагаем, что с каждым объектом монитора связаны две очереди потоков.

• e очередь на вход
• s представляет собой очередь потоков, которые передали сигнал.

Кроме того, мы предполагаем, что для каждой условной переменной c существует очередь

• c.q, которая представляет собой очередь потоков, ожидающих условную переменную c

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

Реализация каждой операции осуществляется следующим образом. (Мы предполагаем, что каждая операция выполняется во взаимном исключении других; таким образом, перезапущенные потоки не начинают выполнение до тех пор, пока операция не завершится.)

enter the monitor:
    enter the method
    if the monitor is locked
        add this thread to e
        block this thread
    else
        lock the monitor

leave the monitor:
    schedule
    return from the method

wait c:
    add this thread to c.q
    schedule
    block this thread

signal c:
    if there is a thread waiting on c.q
        select and remove one such thread t from c.q
        (t is called "the signaled thread")
        add this thread to s
        restart t
        (so t will occupy the monitor next)
        block this thread

schedule:
    if there is a thread on s
        select and remove one thread from s and restart it
        (this thread will occupy the monitor next)
    else if there is a thread on e
        select and remove one thread from e and restart it
        (this thread will occupy the monitor next)
    else
        unlock the monitor
        (the monitor will become unoccupied)

The schedule подпрограмма выбирает следующий поток, который займет монитор или, при отсутствии потоков-кандидатов, разблокирует монитор.

Получающаяся в результате дисциплина сигнализации известна как «сигнальное и срочное ожидание», поскольку сигнализатор должен ждать, но ему дается приоритет над потоками во входной очереди. Альтернативой является «сигнал и ожидание», при котором нет s очередь и сигнальщик ждет на e вместо этого очередь.

Некоторые реализации предоставляют операцию сигнала и возврата , которая сочетает в себе сигнализацию и возврат из процедуры.

signal c and return:
    if there is a thread waiting on c.q
        select and remove one such thread t from c.q
        (t is called "the signaled thread")
        restart t
        (so t will occupy the monitor next)
    else
        schedule
    return from the method

В любом случае («сигнал и срочное ожидание» или «сигнал и ожидание»), когда сигнализируется условная переменная и есть хотя бы один поток, ожидающий условной переменной, сигнальный поток плавно передает занятость сигнальному потоку, поэтому что никакой другой поток не может занять место между ними. Если P _c истинно в начале каждой операции сигнала c , оно будет истинным и в конце каждой ожидания c операции . Это суммировано следующими контрактами . В этих контрактах I монитора является инвариантом .

enter the monitor:
    postcondition I

leave the monitor:
    precondition I

wait c:
    precondition I
    modifies the state of the monitor
    postcondition Pc and I

signal c:
    precondition Pc and I
    modifies the state of the monitor
    postcondition I

signal c and return:
    precondition Pc and I

В этих контрактах предполагается, что I и P _c не зависят от содержимое или длину любых очередей.

(Когда у условной переменной можно запросить количество потоков, ожидающих в ее очереди, можно задать более сложные контракты. Например, полезная пара контрактов, позволяющая передавать занятость без установления инварианта, такова:

wait c:
    precondition I
    modifies the state of the monitor
    postcondition Pc

signal c
    precondition (not empty(c) and Pc) or (empty(c) and I)
    modifies the state of the monitor
    postcondition I

(См. Говард и Бур и др. для большего.)

Здесь важно отметить, что утверждение P _c полностью зависит от программиста; ему или ей просто нужно быть последовательным в том, что это такое.

Мы завершаем этот раздел примером потокобезопасного класса, использующего блокирующий монитор, реализующий ограниченный поточно-безопасный стек .

monitor class SharedStack {
    private const capacity := 10
    private int[capacity] A
    private int size := 0
    invariant 0 <= size and size <= capacity
    private BlockingCondition theStackIsNotEmpty /* associated with 0 < size and size <= capacity */
    private BlockingCondition theStackIsNotFull /* associated with 0 <= size and size < capacity */

    public method push(int value)
    {
        if size = capacity then wait theStackIsNotFull
        assert 0 <= size and size < capacity
        A[size] := value ; size := size + 1
        assert 0 < size and size <= capacity
        signal theStackIsNotEmpty and return
    }

    public method int pop()
    {
        if size = 0 then wait theStackIsNotEmpty
        assert 0 < size and size <= capacity
        size := size - 1 ;
        assert 0 <= size and size < capacity
        signal theStackIsNotFull and return A[size]
    }
}

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

Неблокирующие переменные условия

При использовании неблокирующих условных переменных (также называемых в стиле Mesa условными переменными или условными переменными «сигнал и продолжение» ) передача сигналов не приводит к тому, что поток сигнализации теряет занятость монитора. Вместо этого сигнальные потоки перемещаются в e очередь. Нет необходимости в s очередь.

Монитор в стиле Mesa с двумя переменными состояния a и b

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

Здесь дается значение различных операций. (Мы предполагаем, что каждая операция выполняется во взаимном исключении других; таким образом, перезапущенные потоки не начинают выполнение до тех пор, пока операция не завершится.)

enter the monitor:
    enter the method
    if the monitor is locked
        add this thread to e
        block this thread
    else
        lock the monitor

leave the monitor:
    schedule
    return from the method

wait c:
    add this thread to c.q
    schedule
    block this thread

notify c:
    if there is a thread waiting on c.q
        select and remove one thread t from c.q
        (t is called "the notified thread")
        move t to e

notify all c:
    move all threads waiting on c.q to e

schedule :
    if there is a thread on e
        select and remove one thread from e and restart it
    else
        unlock the monitor

В качестве вариации этой схемы поток уведомлений может быть перемещен в очередь, называемую w, который имеет приоритет над e. См. Говарда и Бур и др. для дальнейшего обсуждения.

Можно связать утверждение P _c с каждой условной переменной c так, что P _c обязательно будет истинным по возвращении из wait c. Однако необходимо убедитесь, что P _c сохраняется с момента, когда уведомляющий поток покидает занятость, до тех пор, пока уведомляемый поток не будет выбран для повторного входа в монитор. В это время могут наблюдаться активности других жильцов. Таким образом, обычно P _c просто истинно .

По этой причине обычно необходимо заключать каждую операцию ожидания в такой цикл:

while not ( P ) do
    wait c

где P — некоторое условие, более сильное, чем P _c . Операции notify c и notify all c рассматриваются как «подсказки» о том, что P может быть истинным для некоторого ожидающего потока. Каждая итерация такого цикла после первой представляет собой потерю уведомления; таким образом, при использовании неблокирующих мониторов необходимо быть осторожным, чтобы не потерять слишком много уведомлений.

В качестве примера «намека» рассмотрим банковский счет, на котором поток вывода средств будет ждать, пока на счету не будет достаточно средств, прежде чем продолжить.

monitor class Account {
    private int balance := 0
    invariant balance >= 0
    private NonblockingCondition balanceMayBeBigEnough

    public method withdraw(int amount)
        precondition amount >= 0
    {
        while balance < amount do wait balanceMayBeBigEnough
        assert balance >= amount
        balance := balance - amount
    }

    public method deposit(int amount)
        precondition amount >= 0
    {
        balance := balance + amount
        notify all balanceMayBeBigEnough
    }
}

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

Мониторы неявных переменных условий

Монитор в стиле Java

В языке Java каждый объект может использоваться в качестве монитора. Методы, требующие взаимного исключения, должны быть явно отмечены ключевым словом Synchronized . Блоки кода также могут быть отмечены синхронизированными .

Вместо явных переменных условий каждый монитор (т. е. объект) оснащен одной очередью ожидания в дополнение к своей очереди на вход. Все ожидания выполняются в этой единственной очереди ожидания, и все операции notify и notifyAll применяются к этой очереди. Этот подход был принят в других языках, например C# .

Неявная сигнализация

Другой подход к передаче сигналов — опустить операцию сигнала . Всякий раз, когда поток покидает монитор (путем возврата или ожидания), утверждения всех ожидающих потоков оцениваются до тех пор, пока одно из них не окажется истинным. В такой системе условные переменные не нужны, но утверждения должны быть явно закодированы. Контракт на ожидание

wait P:
    precondition I
    modifies the state of the monitor
    postcondition P and I

История

Бринч Хансен и Хоар разработали концепцию монитора в начале 1970-х годов, основываясь на своих более ранних идеях и идеях Эдсгера Дейкстры . Бринч Хансен опубликовал первую нотацию монитора, приняв классов концепцию Simula 67 . и изобрел механизм очередей. Хоар уточнил правила возобновления процесса. Бринч Хансен создал первую реализацию мониторов в Concurrent Pascal . Хоар продемонстрировал их эквивалентность семафорам .

Мониторы (и Concurrent Pascal) вскоре стали использоваться для структурирования синхронизации процессов в операционной системе Solo .

Языки программирования, поддерживающие мониторы, включают:

• Ада начиная с Ады 95 (как защищенные объекты)
• C# (и другие языки, использующие .NET Framework )
• Параллельный Евклид
• Параллельный Паскаль
• Д
• Delphi (Delphi 2009 и выше, через TObject.Monitor)
• Java (через методы ожидания и уведомления)
• Идти
• Меса
• Модуль-3
• Python (через объект threading.Condition )
• Руби
• Писк ​
• Тьюринг , Тьюринг+ и объектно-ориентированный Тьюринг
• микроС++
• Визуальный Пролог

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

См. также

• Взаимное исключение
• Обмен последовательными процессами - более поздняя разработка мониторов CAR Hoare.
• Семафор (программирование)

Дальнейшее чтение

• Блох, Джошуа (2018). «Эффективная Java: Руководство по языку программирования» (третье изд.). Аддисон-Уэсли. ISBN  978-0134685991 .
• Мониторы: концепция структурирования операционной системы, CAR Hoare – Communications of the ACM , v.17 n.10, с. 549–557, октябрь 1974 г. [1]
• Классификация мониторов П. А. Бур, М. Фортье, М. Х. Коффин – ACM Computing Surveys , 1995 г. [2]

Внешние ссылки

• Java-мониторы (понятное объяснение)
• « Мониторы: концепция структурирования операционной системы », автор: К. А. Хоар
• « Сигнализация в мониторах »