Programación concurrente (V)

 Artículos de la serie:

1. Introducción básica: clase std::thread y función std::async
2. Paralelización de algoritmos
3. Comprendiendo std::async en más detalle
4. ¿Cómo funciona std::packaged_task?
5. Mecanismo de exclusión mutua (mutex)
6. std::jthread: hilo cooperativamente interrumpible
7. Queue sincronizado
8. Ejercicio resuelto de paralelización

---

Mecanismo de exclusión mutua (mutex)

Consideremos un recurso mutable compartido por dos hilos de ejecución diferentes (por supuesto, su número pudiera ser mayor). Si ambos hilos sólo realizan acciones de lectura sobre el recurso, no corremos riesgo alguno de corromperlo. Sin embargo, si al menos uno de los hilos ejecuta acciones de escritura, modificando el recurso, es obvio que deberemos establecer un orden de acceso entre los hilos, de forma que mientras que uno de ellos esté operando con el recurso, el resto de hilos que intenten acceder a él queden momentáneamente bloqueados. Hablamos aquí, pues, de un mecanismo de exclusión mutua entre hilos.

Consideremos, por ejemplo, un vector estándar no-constante std::vector<> compartido por dos hilos. Si uno de dichos hilos tratara de obtener el tamaño del vector mediante una llamada a size() mientras que el otro inserta un nuevo elemento a través de push_back(), la interferencia de ambas acciones no sincronizadas podría corromper el proceso (data race). Denominaremos sección crítica a toda aquella área del código que deba ser accedida por un único hilo por vez. El mecanismo que evita el acceso a una sección crítica por parte de más de un hilo a la vez recibe el nombre de mutex (del Inglés mutual exclusion). Así, en nuestro ejemplo, procederemos a proteger el vector asociándole un objeto de la clase std::mutex, proporcionado por la cabecera <mutex> [1]. Este objeto deberá ser compartido también por los hilos:

#include <mutex> #include <vector> using namespace std; auto v = vector<int>{}; auto mtx = mutex{}; // mutex asociado al vector v void thread_1_function(int i) {     mtx.lock(); // adquiere el mutex     v.push_back(i);      mtx.unlock(); // libera el mutex } void thread_2_function() {     mtx.lock(); // adquiere el mutex     cout << v.size() << '\n';     mtx.unlock(); // libera el mutex }

Como podemos observar, el área de código a sincronizar (la sección crítica, ya sea la llamada a push_back() o a size()) se encierra entre llamadas a las funciones miembro lock() y unlock() de la clase std::mutex (en ese orden). Cuando un hilo invoca lock(), se dice que posee el mutex, pudiendo entrar y ejecutar la sección crítica. Durante ese tiempo, cualquier otro hilo que llame a lock() con el fin de adquirir el mutex quedará bloqueado, a la espera de que el hilo que lo posee lo libere invocando unlock(). Sólo un hilo por vez puede adquirir el mutex.

Con el fin de garantizar que la adquisición/liberación del mutex sea segura ante la posible emisión de excepciones en la sección crítica, el estándar del lenguaje ofrece la clase std::lock_guard<>, un envoltorio para el mutex que implementa la técnica RAII. El destructor de lock_guard llamará automáticamente a unlock() cuando el objeto salga fuera de su ámbito de definición:

// el siguiente ámbito define la sección crítica: {     auto _ = lock_guard{mtx};  // adquirimos el mutex (empleo de CTAD de C++17)     // código de la sección crítica... } // el mutex mtx es liberado automáticamente al finalizar la sección crítica

Por supuesto, no es posible saber a priori el orden en que los diferentes hilos adquirirán un mutex compartido. A modo de ejemplo, consideremos el siguiente código, en el que dos hilos compiten por introducir enteros al fondo de un vector hasta que éste alcance una longitud máxima de diez elementos:

#include <future> #include <iostream> #include <mutex> #include <vector> using namespace std; auto push(int thread_idx, vector<int>& v, mutex& mtx) -> void {     static auto i = 0;     while (true) {         auto _ = lock_guard{mtx};         if (i >= 10)            return;         cout << "Thread " << thread_idx << " inserts element " << v.size() << '\n';        v.push_back(i);         ++i;     } } auto main() -> int {     auto data = vector<int>{};     auto mtx = mutex{};      auto f_1 = async(launch::async, [&]{ push(1, data, mtx); });     auto f_2 = async(launch::async, [&]{ push(2, data, mtx); });     f_1.wait(); // bloqueo hasta que la primera operación asíncrona haya terminado     f_2.wait(); // bloqueo hasta que la segunda operación asíncrona haya terminado }

En el código hemos tenido en cuenta que, de acuerdo a las especificaciones del lenguaje, si varios hilos intentan inicializar una variable estática, sólo uno de ellos lo logrará (la inicialización de variables estáticas locales es thread-safe desde el estándar C++11 [2]). Observemos también que, al ser el contador i un recurso compartido por los hilos, su incremento y la sentencia de control if que comprueba su valor deben ser protegidos por el mutex. Un posible orden de ejecución podría mostrar una perfecta alternancia de hilos:

Mientras que una ejecución posterior podría generar un patrón completamente diferente:

De suprimirse el empleo del mutex y dejar desprotegido al vector (así como al flujo de salida std::cout), visualizaríamos una interferencia de operaciones imprevisible y caótica como la siguiente (observemos, en particular, cómo el contador de elementos estático i puede aparecer repetido y/o superar 9):

Para terminar, hacemos notar que es posible evitar el uso de la variable estática compartida en la función push() del ejemplo sin más que dotar a cada hilo con un bucle for propio. Dado que cada hilo iniciará y trabajará con un índice de iteración particular, no resulta necesario protegerlo con un mutex como en el caso anterior, lo que simplifica notablemente la comprensión del código:

auto push(int thread_idx, vector<int>& v, mutex& mtx) -> void {     // el índice i no es compartido; no es necesario protegerlo con un mutex:     for (auto i = 0; i < 5; ++i) {         auto _ = lock_guard{mtx};         cout << "Thread " << thread_idx << " inserts element " << v.size() << '\n';        v.push_back(i);     } }

---

Referencias bibliográficas

1. cppreference - std::mutex - https://en.cppreference.com/w/cpp/thread/mutex
2. cppreference - Static local variables - https://en.cppreference.com/w/cpp/language/storage_duration#Static_local_variables