Concurrencia y ParalelismoC++11
Carlos Javier Fernández Candel
Introducción
C++98 C++03 C++11 C++14 C++17
Arreglos ArreglosInicio Threading
Estable Pruebas
Índice
• Nuevas características C++11
• Concurrencia C++11
• Threads
• Exclusión Mutua
• Futuros y Promesas
• Tareas asíncronas
• Paralelismo C++11
• Librería Parallel STL
Nuevas características
• Delegación de Constructores y nullptr
• Deducción automática de tipos y Expresiones Landas
• Inicializaciónuniforme
• Semántica del movimiento
Delegación de Constructores y nullptr
• Un constructor puede delegar en otro
• nullptr sustituye la macro NULL
class Punto{
int x, y;public:M(int _x, int _y) : x(_x), y(_y) {}M(): M(0,0) {}
};
void f(int) { } void f(char *) { }
f(0);f(NULL); f(nullptr);
Deducción automática de tipos y Expresiones Landas
• auto determina el tipo en tiempo de compilación
• Expresiones Landas
auto entero = 0; auto flotante = 0.5f; auto doble = 0.5; auto it = vi.begin();
[capture](parameters)->return-type { body }
for_each(s, s+sizeof(s), [&mayusculas] (char c) {
if (isupper(c)) { mayusculas++;
} }
);
for (auto x : s) { if (isupper(x)) {
mayusculas++;}
}
char s[] = "Hello World!";int mayusculas = 0;
Inicialización uniforme
• La inicialización de cualquier elemento se realiza de la misma forma
• Tipos Primitivos y Objetos
C++ 98/03 C++11
int i = 3; int j = 0; string s("hello");
int i{3};int j{}; string s{"hello"};
ClassA objA1;ClassA objA2 = { 1, 2.0}; ClassB objB1(1, 2.0);
ClassA objA1 {}; ClassA objA2 {1, 2.0};ClassB objB1 {1, 2.0};
Vector<ClassB> vectorOfBs;vectorOfBs.push_back(ClassB(1,1.0));vectorOfBs.push_back(ClassB(2,2.0));
vector<ClassB> vectorOfBs = {{1, 1.0}, {2, 2.0}
};
Inicialización uniforme
• La inicialización de cualquier elemento se realiza de la misma forma
• Tipos Primitivos y Objetos
C++ 98/03 C++11
int arr[] = { 1, 2, 3, 4, 5}; int arr[] { 1, 2, 3, 4, 5};
map<int, string> m;m[0] = "zero"; m[1] = "one”;
map<int, string> m {{0,"zero"}, {1,"one”}
};
pair<double, double> result = multiplyVectors(
pair<double, double>(1.0, 2.0), pair<double, double>(3.0, 4.0));
auto result = multiplyVectors({1.0, 2.0}, {3.0, 4.0}
);
Semántica del Movimiento: Rvalue y Lvalue
• Lvalue
• Es posible extraer una dirección
• Rvalue
• Objeto temporalint x; int getVal(){
return x;}getVal();
int x; int & getVal(){
return x;} getVal() = 4;
string getName(){
return "Alex";}getName();
Semántica del Movimiento: Constructores
class ArrayWrapper{
private:int * _p_vals;int _size;
public:// move constructorArrayWrapper (ArrayWrapper&& other)
: _p_vals(other._p_vals), _size(other._size)
{other._p_vals = NULL;other._size = 0;
}
// copy constructorArrayWrapper (const ArrayWrapper& other)
: _p_vals(new int[other._size]), _size(other._size)
{for (int i = 0; i < _size; ++i){
_p_vals[i] = other._p_vals[i];}
} };
Semántica del Movimiento: Constructores
class ArrayWrapper{
private:int * _p_vals;int _size;
public:// move constructorArrayWrapper (ArrayWrapper&& other)
: _p_vals(other._p_vals), _size(other._size)
{other._p_vals = NULL;other._size = 0;
}};
• Posibilidad de establecer el operador ‘=‘
• Los recursos se transfieren
• Se evita reservar memoria
• Se deja al origen en un estado destruible.
• Haciendo punteros nullptr.
[const] ArrayWrapper getArrayWrapper(){
return arrayWrapper;}
Semántica del Movimiento: Ejemplo
vector<string> concat(const vector<string> & l1, const vector<string> & l2) {
vector<string> r;
for (auto i = l1.begin(); i != l1.end(); i++) { r.push_back(*i);
} for (auto i = l2.begin(); i != l2.end(); i++) {
r.push_back(*i); }
return r; }
vector<string> a{"Hola”, "Mundo”};vector<string> b{"a”, "b”, "c”};
vector<string> sol = concat(a, b);
Semántica del Movimiento: Move
#include <utility>
vector<string> a{"a”, "b”, "c”};vector<string> c = move(a);
• Operador ‘move’
• Convierte un lvalue en rvalue
• No se ejecuta ningún procedimiento. Es una marca
• No realiza ningún movimiento
• Lo realiza el operador de asignación
Compilación
• g++ -std=c++11 file_name.cpp
• clang++ -std=c++11 -stdlib=libc++ file_name.cpp
Índice
• Nuevas características C++11
• Concurrencia C++11
• Threads
• Exclusión Mutua
• Futuros y Promesas
• Tareas asíncronas
• Paralelismo C++11
• Librería Parallel STL
Concurrencia C++11
• Modelo propio, forma parte:
• Lenguaje
• Biblioteca Estándar
• Código Portable
• Windows
• POSIX
• Nueva librería
• Threads
• Cerrojos
• Tasks
Threading: Thread Class
• Clase thread
• Un hilo del sistema operativo
• Mismo espacio de memoria
• Cada uno tiene pila
void tarea1(){
cout << “Thread: 1” << endl;}
thread thread1{tarea1};
void tarea(int i){
cout << “Thread: ” << i << endl;}
thread thread2{tarea, 2};
Threading: Thread Class
• Clase thread
• Un hilo del sistema operativo
• Mismo espacio de memoria
• Cada uno tiene pila
void tarea(int & i){
cout << “Thread: ” << i << endl;}
int i {1};thread thread1{tarea, i}; thread thread2{tarea, ref(i)};thread thread3{tarea, move(i)};thread thread4{[] () { f(i); }};
Threading: Joining
• join() para esperar a que un thread finalice
thread1.join();
if (thread2.joinable()) {thread2.join();
}
void tarea(int & i){
cout << “Thread: ” << i << endl;}
int i {1};thread thread1{tarea, i}; thread thread2{tarea, ref(i)};
Threading: Detaching
• deatch() para esperar a que un thread finalice
void tarea(int & i){
cout << “Thread: ” << i << endl;}
int i {1};thread thread{tarea, i};thread.detach();
thread.join() Error
Threading: id
• get_id() para obtener el identificador del thread
• sleep_until(x) Espera hasta un instante de tiempo.
• sleep_for(x) Espera durante un determinado de tiempo.
void tarea(){
cout << “Thread: ” << this_thread::get_id() << endl;}
int i {1};thread thread{tarea, i};thread::id thread.get_id();
Threading: Datos
• thread_local define variables no compartidas
thread_local i = 0;
void tarea(){
i = rand();cout << “Thread: ” << i << endl;
}
thread thread{tarea};
Threading: Ejemplo
int num_cpus = thread::hardware_concurrency();cout << "Threads: " << num_cpus << endl;
vector<thread> threads(num_cpus);for (int i = 0; i < num_cpus; i++) {
threads[i] = thread{ [](int i) {
cout << "Thread num: " << i << endl;}, i
}; }
for (auto& t : threads) { t.join(); }
Threading: Smart Pointers
• Unique_ptr
unique_ptr<int> p1(new int{1});
auto unique = make_unique<int>{1};
• Shared Pointer
shared_ptr<int> p1(new int{1});
auto shared = make_shared<int>{1};
int main () {
unique_ptr<int> p(new int{1});for (int i = 0; i < 5; i++) {
p[i] = i; cout << p[i] << endl;
}return 0;
}
P2P1
Exclusión Mutua
T1Load
T2Load
T3Load
T4Load
T1Add
T2Add
T3Add
T4Add
T1Assign
T2Assign
T3Assign
T4Assign
int i{0};
void add() {
i = i + 1;
}
thread t1{add};
thread t2{add};
thread t3{add};
thread t4{add};
t1.join(); t2.join();
t3.join(); t4.join();
Exclusión Mutua
• Soluciones
• mutex
• recursive_mutex
• timed_mutex
• recursive_timed_mutex
• Operaciones
• m.lock() Adquirir cerrojo
• m.unlock(): Liberar cerrojo
• m.try_lock(): Intenta adquirir el cerrojo
• m.try_lock_for(d): Intenta adquirir el cerrojo un tiempo
• m.try_lock_until(t); Intenta adquirir el cerrojo hasta un instante
Exclusión Mutua: Mutex
mutex m;
int i{0};
void add() {
m.lock();
i = i + 1;
add();
m.unlock();
}
thread t1{add};
thread t2{add};
t1.join();
t2.join();
Recursive_mutex m;
int i{0};
void add() {
m.lock();
i = i + 1;
add();
m.unlock();
}
thread t1{add};
thread t2{add};
t1.join();
t2.join();
Exclusión Mutua: Guards
mutex m;
void add() {
lock_guard<mutex> guard{m};
i = i + 1;
}
• lock_guard<M> {m[, adopt_lock]}
• Objeto que libera el mutex cuando se ejecuta el destructor de la guarda
• unique_lock<M> {m[, method]}
• Control mayor sobre mutex
• Methods
• defer_lock
• try_lock
• adop_lock, ya adquirido
• unique_lock<M> l{m,t}
• m.try_lock_until(t).
• unique_lock<M> l{m,d}
• m.try_lock_for(d).
Exclusión Mutua: Ejemplo
int num_cpus = thread::hardware_concurrency();cout << "Threads: " << num_cpus << endl;
mutex iomutex;vector<thread> threads(num_cpus);for (int i = 0; i < num_cpus; i++) {
threads[i] = thread{ [&iomutex](int i) {
lock_guard<mutex> iolock(iomutex); cout << "Thread num: " << i << endl;
}, i }; }
for (auto& t : threads) { t.join(); }
Exclusión Mutua: Adquisición múltiple
void process(mutex & a, mutex & b) {unique_lock<mutex> l1{a}; unique_lock<mutex> l2{b};
}
mutex a, b;thread t1{process, ref{a}, ref{b}};thread t2{process, ref{b}, ref{b}};
void process(mutex & a, mutex & b) {unique_lock<mutex> l1{a, defer_lock}; unique_lock<mutex> l2{b, defer_lock};lock(l1, l2);
}
mutex a, b;thread t1{process, ref{a}, ref{b}};thread t2{process, ref{b}, ref{b}};
Exclusión Mutua: Variables Condición
condition_variable c{};
• Operaciones:
• c.notify_one(): Despierta un hilo
• c.notify_all(): Despierta a todos los hilos
• c.wait(l): Se bloquea hasta que adquiere el cerrojo
• c.wait_until(l,t): Intenta adquirir el cerrojo. Si llega al tiempo sale sin cerrojo.
• c.wait_for(l,t): Intenta adquirir el cerrojo durante un tiempo.
Exclusión Mutua: Variables Condiciónqueue<int> nums;mutex m;condition_variable cond_var;bool done = false;bool notified = false;
thread producer{[]() {
for (int i = 0; i < 5; i++) {unique_lock<mutex> lock(m); nums.push(i);notified = true;cond_var.notify_one();
} done = true;cond_var.notify_one();
}}
thread consumer{[]() {
unique_lock<mutex> lock(m);while (!done) {
while (!notified) {cond_var.wait(lock);
} while (!nums.empty()) {
cout << nums.front();nums.pop();
} notified = false;
} }
};
Exclusión Mutua: Cerrojos Compartidos c++14
shared_mutex mshared_lock<M>
• Operaciones:
• m.lock(): Bloquea el cerrojo para uso exclusivo
• m.lock_shared(): Bloquea el cerrojo para uso compartido
void escritor() {m.lock();i++;
}
void lector() {m.lock_shared();cout << i << endl;
}
Futuros y Promesas• Forma de comunicar datos entre dos tareas
• Promesa es un envío de datos
• Mediante el método set_value()
• Es posible establecer excepciones como valor
• Futuro es la recepción de un valor.
• Mediante el método get()
• Solo es posible realizar un llamada
• Comprobar el valor con valid()
• Mediante el método share() se permite comprobar repetidamente
• Es posible esperar: wait_for() y wait_until()
future<T> f;
promise p;
Futuros y Promesas: Ejemplovoid initiazer(promise<int> * promObj){
cout << "Inside Thread” << endl; promObj->set_value(35);
}
promise<int> promiseObj;future<int> futureObj = promiseObj.get_future();
thread th(initiazer, &promiseObj);
cout << futureObj.get() << endl;
th.join();
Tareas asíncronas
• Política
• async Se lanza un nuevo hilo que ejecuta la función
• deferred El hilo se lanzará cuando se llame al método get()
asnync(launc::async | launch::deferred, funcion, argumentos);
vector<future<int>> tasks;
for (int i = 0; i < tam; i++) { tasks.push_back(async(launch::deferred, sum, x[i], y[i]));
}
Índice
• Nuevas características C++11
• Concurrencia C++11
• Threads
• Exclusión Mutua
• Futuros y Promesas
• Tareas asíncronas
• Paralelismo C++11
• Librería Parallel STL
Paralelismo
• STL: Standard Template Library
• Conjunto de algoritmos
• Los algoritmos paralelos presentan una política como argumento
• Secuencial: seq
• Parallel: par
• Parallel + Vector: parvec
• Dinámica
Paralelismousing namespace std::experimental::parallel;
vector<double> v = read_values();
sort(v.begin(), v.end()); sort(seq, v.begin(), v.end());sort(par, v.begin(), v.end());sort(parvec, v.begin(), v.end());
constexpr size_t threshold = 1000; execution_policy p = seq;
if (v.size() > threshold) {
p = par; }
sort(p, v.begin(), v.end());
Paralelismo• all_of: Comprueba si un predicado es cierto para todos los elementos• count: Cuenta el número de elementos iguales a un valor. • equal: Compara dos secuencias elemento a elemento
• find: Busca el primer elemento que tiene un valor• for_each: Aplica una función a los elementos de una secuencia
• search: Busca una subsecuencia dentro de una secuencia• move: Mueve una secuencia de elementos • copy: Copia una secuencia de elementos
• remove: Elimina los elementos que son iguales a un valor• partition: Divide una secuencia en dos usando un predicado como criterio
• sort: Ordena una secuencia ascendentemente• reduce: Realiza una operación de reducción sobre una secuencia
Paralelismovector<int> v {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14};auto q = find_if(par, begin(v), end(v),
[](int x) { return (x > 10) && (x < 15);
}
long primos(const vector<int> & v) {
atomic<long> count; for_each(par, v.begin(), v.end(),
[](int x) { if (esprimo(x)) {
count++; }
});
return count; }
Paralelismovector<int> v {1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14};
reduce(par, v.begin(), v.end(), [](int a, int b) {
return a + b; }
);
auto sumaCuadrados = transform_reduce(par, v.begin(), v.end(), [](double x) {
return x * x;}, 0, [](double x, double y) {
return x + y; }
);
Modelo de Memoria
• C++11 define un modelo de concurrencia propio como parte del propio lenguaje
• Tipos atómicos.
• Mecanismos de sincronización de bajo nivel
• Dos hilos pueden acceder a posiciones de memoria distintas de forma simultánea.
• Dos hilos pueden acceder a la misma posición de memoria de forma simultánea si ambos accesos son de lectura.
• Si dos hilos intentan acceder de forma simultánea a la misma posición de memoria y alguno de los accesos es de escritura existe una condición de carrera potencial.
• Mecanismo de sincronización
• Accesos atómicos
Accesos Atómicos
atomic<T>
• Operaciones atómicas
• Se ejecuta todo o nada
• Restricciones con los datos
• Puntero
• Bool
• No está definido para reales
atomic_flag
• Dos estados: activo o desactivado
atomic_address
• Acceso atómica a una dirección de memoria.
Concurrencia y ParalelismoC++11
Carlos Javier Fernández Candel