C++ / Carreras de datos

Bienvenidos sean a este post, hoy veremos esta condicion un poco mas a fondo.

En este post mencionamos por primera vez la condicion de carrera o carrera de datos, y uno de nuestros objetivos es evitar esto. En el post anterior vimos, teoricamente, un patron de singleton que es thread-safe. Vamos a analizar un ejemplo simple para una conexion a una base de datos:

namespace Db {
  class AdminConexion 
  {
  public:
    static std::shared_ptr<AdminConexion> get_instancia()
    {
      if (instancia_ == nullptr) {
        instancia_.reset(new AdminConexion());
      }
      return instancia_;
    }

    // codigo relacionado a la conexion es omitido
  private:
    static std::shared_ptr<AdminConexion> instancia_{nullptr};
  };
}

En este simple codigo, establecemos una conexion a la base de datos mediante singleton. Esto es para cada vez que necesitemos conectarnos, usaremos siempre la misma y evitar tener una conexion por cada vez que la usemos. Ya comentamos que un patron singleton es para tener una instancia unica, para ello tenemos al constructor en la parte privada y en la parte publica un metodo para verificar si existe o no. En caso de no existir se procede a crearla, y devuelve el objeto de instancia. En caso contrario, devuelve el objeto ya generado.

Vamos a suponer que de esta creamos un thread, y mediante este accederemos al metodo get_instancia. Si lo denominamos como threadA, la primera instruccion que accedera es el condicional. Y si seguimos de forma secuencial, al chequear la instancia sera nula y por ende, procede a resetearla y genera una nueva instancia. Pero que sucede si creamos otra instancia? El thread pasara el condicional, como ya existe la instancia, y pasa a devolver la instancia asignada. En cambio, el segundo thread, llamemoslo threadB, al ser nulo cumple la condicion y procede a resetear la instancia, devolviendo este nuevo objeto. Si se dieron cuenta, este nuevo thread interfiere en el anterior haciendo todo lo contrario que se espera de un singleton. Hagamos el siguiente cambio en el codigo anterior:

namespace Db {
  class AdminConexion 
  {
  public:
    static std::shared_ptr<AdminConexion> get_instancia()
    {
      if (instancia_ == nullptr) {
		std::lock_guard lg{mutex_};
		if (instancia_ == nullptr) {
		   instancia_.reset(new AdminConexion());
		}
      }
	  return instancia_;
    }

    // codigo relacionado a la conexion es omitido
  private:
	static std::mutex mutex_;
    static std::shared_ptr<AdminConexion> instancia_{nullptr};
  };
}

Esta solucion la implementamos en este post. Aqui aplicamos un mutex para alternar los threads correctamente y un doble chequeo condicional. Esto cambiara a la hora de chequear todo, vamos a suponer que nuevamente primero actua threadA, pasa por el primer condicional y lo primero que hace es bloquear al mutex. Por lo tanto, cuando threadB revisa el primer condicional tambien pasa pero ahora debe esperar a que threadA lo desbloquee. threadA pasa al siguiente condicional y genera la instancia, ahora threadB bloquea al mutex y pasa al condicional pero no lo pasara porque ya no es nulo y por lo tanto, se desbloquea al mutex y devolvemos el objeto de threadA. Como regla general, siempre debe mirar entre líneas del código. Siempre hay un espacio entre dos declaraciones, y ese espacio hará que dos o más subprocesos interfieran entre sí. Pero para entender mejor este concepto, vamos a analizar el siguiente ejemplo:

#include <iostream>
#include <thread>
#include <string>

int contador = 0;

void foo(std::string id)
{
  contador++;
  std::cout << id << ": " << contador << std::endl;
}

int main()
{
  std::jthread A{foo, "A"};
  std::jthread B{foo, "B"};
  std::jthread C{[]{foo("C");}};
  std::jthread D{
    []{
      for (int ix = 0; ix < 10; ++ix) { foo("D"); }
    }
  };
}

Este codigo simple posee una funcion que simplemente incrementa a la variable contador pero tambien recibe un valor de tipo string que usaremos para identificar al thread desde donde es ejecutado. En la funcion mostramos no solamente el valor de contador sino tambien a su identificador, simplemente para ver quien incremento ese valor. En el main, creamos cuatro thread para ver como trabaja, en cada uno llamamos a la funcion y pasamos un identificador pero en el ultimo usamos un bucle para que cuente diez veces. Compilemos y veamos como es su salida:

$ ./race
D: 1
D: 2
D: 3
D: 4
D: 6
D: 7
D: 8
D: 10
D: B: 11
C: 11
11
D: 12
A: 13
$

Observen como se dio la prioridad, asi como tambien una superposicion de contadores y luego la mostro. Esto es lo que se buscca evitar. Si descomponemos a la accion de sumar a contador se veria de la siguiente manera:

auto res = contador;
contador = contador + 1;
return res;

Puede suceder que la primer linea sea tratada por un thread, la segunda por otra y la ultima por otra. Por esta razon, tenemos los problemas que vimos en la salida anterior. Tomemos el codigo anterior y hagamos el siguiente cambio:

#include <iostream>
#include <thread>
#include <string>
#include <mutex>

int contador = 0;
std::mutex m;

void foo(std::string id)
{
  std::lock_guard g{m};
  contador++;
  std::cout << id << ": " << contador << std::endl;
}

int main()
{
  std::jthread A{foo, "A"};
  std::jthread B{foo, "B"};
  std::jthread C{[]{foo("C");}};
  std::jthread D{
    []{
      for (int ix = 0; ix < 10; ++ix) { foo("D"); }
    }
  };
}

Simplemente agregamos al mutex para poder bloquear los accesos de los threads. Para ello, agregamos un lock_guard que sera el encargado de monitorear los bloqueos y desbloqueos en la funcion. El resto sigue siendo de la misma forma. Compilemos y veamos como es la salida:

$ ./race
D: 1
D: 2
D: 3
D: 4
D: 5
D: 6
D: 7
D: 8
D: 9
D: 10
B: 11
C: 12
A: 13
$

Observen como funciono perfectamente y nos mostro en orden los valores de contador cada vez que es llamada la funcion. Si lo vuelven a ejecutar siempre mostrara la misma salida. Analicemos como trabaja ahora la parte de la incrementacion de contador:

lock mutex;
auto res = contador;
contador = contador + 1;
unlock mutex;
return res;    

Si lo tomamos como ejemplo al inicio, en la primer linea los thread A, B y C esperan por el mutex bloqueado. Thread D es el que tiene bloqueado al mutex y es el que trabaja en la segunda linea. Procede a incrementarlo y luego desbloqueamos el mutex, Pero mientras D necesite hacer esto seguira bloqueanddolo pero cuando termine y lo desbloquee sera bloqueado por algunos de los otros threads.

El mayor problema que tenemos con el bloqueo es la performance. Pero paradojicamente los threads son para mejorar la ejecucion aunque en realidad es el proceso de la informacion. En los casos de grandes colecciones, el uso de multiple threads podria incrementar la performance drasticamente. Sin embargo, en un entorno multithreading debemos tener cuidado del acceso concurrente porque el acceso a la coleccion mediante multiples threads pueden derivar en una corrupcion. Pero en el proximo post veremos como trabaja sobre esto.

En resumen, hoy hemos vsto a carrera de datos, tambien conocida como condiciones de carrera, que son, porque se pueden producir, asi como tambien un ejemplo clasico para ver como es una conducta, asi como tambien como solucionarlo mediante mutex pero viendo un posible error que puede afectarnos. Espero les haya resultado de utilidad sigueme en tumblr, Twitter o Facebook para recibir una notificacion cada vez que subo un nuevo post en este blog, nos vemos en el proximo post.