Thread con il C++ in windows

[miondere]

Vi spiego il problema:
voglio creare due thread in C++ sotto Windows.
Questi 2 thread creati devono condividere un array (dove vi possono scrivere e leggere) e devono essere tra di loro sincronizzati, specialmente quando entrano nella sezione critica di aggiornare l'array.
Mi potete spiegare chiaramente come devo fare a risolvere questo problema in C++ con l'uso di Windows (xp)?
Magari, se siete così gentili, mi farebbe piacere vedere un piccolo abbozzo del programma, per capire quali funzioni devo usare, dove le devo usare e che parametri devo passare...
Sono alle prime armi con il C++!
Grazie mille! Ciao...

[cionci]

Dipende da quale compilatore usi e da quali librerie vuoi usare...

Se vuoi usare i thread delle API di Windows c'è CreateThread...

CreateThread può lanciare una qualsiasi funzione di questo determinato tipo:
DWORD WINAPI Funzione(
LPVOID lpParameter // thread data
);

Questa è CreateThread:

HANDLE CreateThread(
LPSECURITY_ATTRIBUTES lpThreadAttributes, // SD
DWORD dwStackSize, // initial stack size
LPTHREAD_START_ROUTINE lpStartAddress, // thread function
LPVOID lpParameter, // thread argument
DWORD dwCreationFlags, // creation option
LPDWORD lpThreadId // thread identifier
);

Inoltre devi utilizzare CreateMutex per creare le mutue esclusioni... waitForSingleObject per prendere l'accesso... ReleaseMutex per rilasciarlo...

Nota che le funzioni che vengono lanciate come thread, se sono membro di una classe, devono essere definite come static...

#include <windows.h>
#include <iostream>
#include <vector>
#include <ctime>

#define VSIZE 3

using namespace std;

class WorkerThreads {
	static DWORD WINAPI ThreadA(LPVOID lpParam);
	static DWORD WINAPI ThreadB(LPVOID lpParam);
	vector<int> v;
	HANDLE h[2];
	HANDLE StatusMutex;
	HANDLE VectorMutex;
	bool status;
public:
	WorkerThreads();
	const bool Start();
	const bool Stop();
	const bool Status() const { return status; };
};

WorkerThreads::WorkerThreads()
{
	status = false;
	srand((unsigned)time(NULL));
	for(int i=0; i<VSIZE; ++i)
		v.push_back(0);
	StatusMutex = CreateMutex(NULL, FALSE, "STATUS_MUTEX");
	VectorMutex = CreateMutex(NULL, FALSE, "STATUS_MUTEX");
}

const bool WorkerThreads::Start()
{
	if(WaitForSingleObject(StatusMutex, INFINITE) != WAIT_OBJECT_0)
		return false;

	if((h[0] = CreateThread(NULL, 0, ThreadA, this, 0, NULL)) == NULL)
		return false;
	if((h[1] = CreateThread(NULL, 0, ThreadB, this, 0, NULL)) == NULL)
		return false;

	status = true;

	ReleaseMutex(StatusMutex);
	return true;
}

const bool WorkerThreads::Stop()
{
	if(WaitForSingleObject(StatusMutex, INFINITE) != WAIT_OBJECT_0)
		return false;

	status = false;

	ReleaseMutex(StatusMutex);
	
	while(WaitForSingleObject(h[0], 1000) == WAIT_TIMEOUT);
	cout << "Uscito il thread 1" << endl;
	while(WaitForSingleObject(h[1], 1000) == WAIT_TIMEOUT);
	cout << "Uscito il thread 2" << endl;
	
	return true;
}

DWORD WorkerThreads::ThreadA(LPVOID lpParam)
{
	WorkerThreads &obj = *(WorkerThreads *)lpParam;
	
	while(1)
	{
		if(WaitForSingleObject(obj.StatusMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(1);

		bool tmp = obj.status;

		ReleaseMutex(obj.StatusMutex);

		if(tmp == false)
			ExitThread(0);

		if(WaitForSingleObject(obj.VectorMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(2);

		int i = rand()%VSIZE;
		cout << "v[" << i << "]: " << obj.v[i] << " --> ";
		obj.v[i]++;
		cout << obj.v[i] << endl;
		cout.flush();

		Sleep(100); //con questo si rende l'output leggibile
		ReleaseMutex(obj.VectorMutex);
	}

	return 0;
}

DWORD WorkerThreads::ThreadB(LPVOID lpParam)
{
	WorkerThreads &obj = *(WorkerThreads *)lpParam;
	
	while(1)
	{
		if(WaitForSingleObject(obj.StatusMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(1);

		bool tmp = obj.status;

		ReleaseMutex(obj.StatusMutex);

		if(tmp == false)
			ExitThread(0);

		if(WaitForSingleObject(obj.VectorMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(2);

		int i = rand()%VSIZE;
		cout << "v[" << i << "]: " << obj.v[i] << " --> ";
		obj.v[i] = rand()%1000;
		cout << obj.v[i] << endl;

		Sleep(100); //con questo si rende l'output leggibile
		ReleaseMutex(obj.VectorMutex);
	}

	return 0;
}

void main()
{
	WorkerThreads t;
	if(t.Start() == false)
		return;

	Sleep(30000); //in questi 30 secondi i thread lavorano

	t.Stop();
}

[lombardp]

Originally posted by "cionci"

Inoltre devi utilizzare CreateMutex per creare le mutue esclusioni... waitForSingleObject per prendere l'accesso... ReleaseMutex per rilasciarlo...

Una domanda (ho fatto qualche applicazione multithread, però non conosco bene l'argomento): Quando si tratta di sincronizzare due thread per l'accesso a risorse comuni, la MSDN library consiglia l'uso di degli oggetti CSemaphore (e quindi i CSingleLock). Viceversa in caso di processi (applicazioni) diversi consiglia appunto i Mutex. Ho visto che nel tuo esempio hai usato i Mutex, c'è qualche ragione particolare?

[cionci]

Quello che ho fatto si basa sulle Win32 API... CSemaphore è MFC...
In ogni caso il semaforo è una generalizzazione della mutex...

[miondere]

Innanzi tutto...grazie!

Mi spieghi da quale controllo un thread capisce che deve finire la sua esecuzione? Come posso fare terminare l'esecuzione di UN SOLO thread quando lo voglio (quando si verifica una certa condizione)?
Metto una condizione al posto di while(1), cioè while(condizione=TRUE). In questo modo esce dal ciclo, ma attende i secondi specificati nello sleep nel main(). C'è un altro modo per realizzarlo? Indipendentemente dal tempo?
Spero di essermi spiegato...


Devo realizzare un programma dove vi è un thread produttore e uno consumatore.
Quando l'array è pieno il produttore deve fermarsi ed attendere che si liberi almeno un elemento dell'array, mentre il consumatore deve fermarsi se il vettore è vuoto.
Ho messo 2 while, come puoi vedere dal codice sottostante: uno while(count==VSIZE); e l'altro while(count==0);

//CODICE DEL THREAD A
DWORD WorkerThreads::ThreadA(LPVOID lpParam)
{
int p=0; //conta gli elementi prodotti

WorkerThreads &obj = *(WorkerThreads *)lpParam;

while(p<produrre) //produrre è il max numero di elementi da produrre
{
while(count==VSIZE);
if(WaitForSingleObject(obj.StatusMutex, INFINITE) != WAIT_OBJECT_0)
ExitThread(1);

bool tmp = obj.status;

ReleaseMutex(obj.StatusMutex);

if(tmp == false)
ExitThread(0);

if(WaitForSingleObject(obj.VectorMutex, INFINITE) != WAIT_OBJECT_0)
ExitThread(2);

cout << "SONO NEL THREAD A ";
cout << count;
count++; //elemento prodotto (count è variabile globale)
cout << "-->" << count << "PRODOTTO:" << p <<endl;

/* int i = rand()%VSIZE;
cout << "v[" << i << "]: " << obj.v[i] << " --> ";
obj.v[i]++;
cout << obj.v[i] << endl;
cout.flush();
*/
Sleep(100); //con questo si rende l'output leggibile
ReleaseMutex(obj.VectorMutex);
p++;

}

return 0;
}





//CODICE DEL THREAD B

DWORD WorkerThreads::ThreadB(LPVOID lpParam)
{
int consumato=0; //conta gli elementi consumati

WorkerThreads &obj = *(WorkerThreads *)lpParam;

while(consumato<produrre)
{
while(count==0);
if(WaitForSingleObject(obj.StatusMutex, INFINITE) != WAIT_OBJECT_0)
ExitThread(1);

bool tmp = obj.status;

ReleaseMutex(obj.StatusMutex);

if(tmp == false)
ExitThread(0);

if(WaitForSingleObject(obj.VectorMutex, INFINITE) != WAIT_OBJECT_0)
ExitThread(2);

cout << "SONO NEL THREAD B ";
cout << count;
count--; elemento consumato
cout << "-->" << count << "CONSUMATO: " << consumato <<endl;

/* int i = rand()%VSIZE;
cout << "v[" << i << "]: " << obj.v[i] << " --> ";
obj.v[i] = rand()%1000;
cout << obj.v[i] << endl;
*/
Sleep(100); //con questo si rende l'output leggibile
ReleaseMutex(obj.VectorMutex);
consumato++;
for (int f; f<10000000; f++); //faccio questo ciclo per far attendere un pò il consumatore...per prova

}

return 0;
}

void main()
{

WorkerThreads t;
if(t.Start() == false)
return;

Sleep(30000); //in questi 30 secondi i thread lavorano

t.Stop();
}

[cionci]

Certo...

const bool WorkerThreads::IsActive(DWORD timeout)
{
	//ritorno vero se almeno uno dei due è attivo
	if(WaitForSingleObject(h[0], timeout) == WAIT_TIMEOUT)
		return true;
	if(WaitForSingleObject(h[1], timeout) == WAIT_TIMEOUT)
		return true;
	status = false;
	return false;
}

Nel main:

void main()
{
	WorkerThreads t;
	if(t.Start() == false)
		return;

	while(t.IsActive(1000))
	{
		//qui fai le tue cose...
	}
}

Altrimenti se vuoi che il thread principale stia fermo:

void main()
{
	WorkerThreads t;
	if(t.Start() == false)
		return;

	t.IsActive(INFINITE); //qui aspetta fino a quando entrambi i thread sono finiti
}

Comunque per il caso del produttore-consumatore il modo migliore per farlo è quello di usare due semafori...

lombardp: logicamente non ci sono differenze fra semafori e mutex... I semafori si usano quando ci sono da gestire più risorse (code o stack ad esempio)...

[lombardp]

Originally posted by "cionci"

lombardp: logicamente non ci sono differenze fra semafori e mutex... I semafori si usano quando ci sono da gestire più risorse (code o stack ad esempio)...

Avevo intuito che gli oggetti di sincronizazione erano sostanzialmente la stessa cosa, però non ne avevo la certezza

Grazie per le risposte.

[cionci]

Questo è un esempio di sincronizzazione con semfori...

#include <windows.h>
#include <iostream>
#include <vector>
#include <ctime>

#define VSIZE 5

using namespace std;

class WorkerThreads {
	static DWORD WINAPI ThreadA(LPVOID lpParam); //Produttore
	static DWORD WINAPI ThreadB(LPVOID lpParam); //Consumatore
	int v[VSIZE];
	int start;
	int end;
	HANDLE h[2];
	HANDLE SemEmptySlots;	//Conta quanti elementi sono vuoti sul totale (VSIZE)
	HANDLE SemFullSlots;	//Conta quanti elementi sono pieni sul totale (VSIZE)
	HANDLE StatusMutex;
	HANDLE VectorMutex;
	bool status;
public:
	WorkerThreads();
	const bool Start();
	const bool Stop();
	const bool IsActive(DWORD timeout = 1);
	const bool Status() const { return status; };
};

WorkerThreads::WorkerThreads()
{
	start = end = 0;
	status = false;
	srand((unsigned)time(NULL));
	StatusMutex = CreateMutex(NULL, FALSE, "STATUS_MUTEX");
	VectorMutex = CreateMutex(NULL, FALSE, "VECTOR_MUTEX");
	SemEmptySlots = CreateSemaphore(NULL, VSIZE, VSIZE, "EMPTY_SLOTS");
	SemFullSlots = CreateSemaphore(NULL, 0, VSIZE, "FULL_SLOTS");
}

const bool WorkerThreads::Start()
{
	if(WaitForSingleObject(StatusMutex, INFINITE) != WAIT_OBJECT_0)
		return false;

	if((h[0] = CreateThread(NULL, 0, ThreadA, this, 0, NULL)) == NULL)
		return false;
	if((h[1] = CreateThread(NULL, 0, ThreadB, this, 0, NULL)) == NULL)
		return false;

	status = true;

	ReleaseMutex(StatusMutex);
	return true;
}

const bool WorkerThreads::Stop()
{
	if(WaitForSingleObject(StatusMutex, INFINITE) != WAIT_OBJECT_0)
		return false;

	status = false;

	ReleaseMutex(StatusMutex);
	
	while(WaitForSingleObject(h[0], 1000) == WAIT_TIMEOUT);
	cout << "E' uscito il produttore" << endl;
	while(WaitForSingleObject(h[1], 1000) == WAIT_TIMEOUT);
	cout << "E' uscito il consumatore" << endl;
	
	return true;
}

DWORD WorkerThreads::ThreadA(LPVOID lpParam)
{
	WorkerThreads &obj = *(WorkerThreads *)lpParam;
	
	while(1)
	{
		if(WaitForSingleObject(obj.StatusMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(1);

		bool tmp = obj.status;

		ReleaseMutex(obj.StatusMutex);

		if(tmp == false)
			ExitThread(0);

		//con questo si simula il lavoro del produttore per creare l'elemento da
		//produrre. Si ottiene anche di desincronizzare i due thread per rendere più
		//interessante l'evoluzione
		Sleep(rand()%3000); 

		//Se non ci sono slot vuoti resta in attesa
		if(WaitForSingleObject(obj.SemEmptySlots, INFINITE) != WAIT_OBJECT_0)
			ExitThread(2);

		if(WaitForSingleObject(obj.VectorMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(2);

		obj.v[obj.end++] = rand()%1000;
		cout << "Prodotto  v[" << obj.end-1 << "] = " << obj.v[obj.end-1];
		cout << " | " << ((obj.end <= obj.start)?obj.end+VSIZE:obj.end)-obj.start
			 << " slot occupati" << endl;
		obj.end %= VSIZE;
		cout.flush();

		ReleaseMutex(obj.VectorMutex);

		//C'è uno slot occupato in più quindi il valore del semaforo deve aumetare
		ReleaseSemaphore(obj.SemFullSlots, 1, NULL);
	}

	return 0;
}

DWORD WorkerThreads::ThreadB(LPVOID lpParam)
{
	WorkerThreads &obj = *(WorkerThreads *)lpParam;
	
	while(1)
	{
		if(WaitForSingleObject(obj.StatusMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(1);

		bool tmp = obj.status;

		ReleaseMutex(obj.StatusMutex);

		if(tmp == false)
			ExitThread(0);

		//Se non ci sono slot pieni resta in attesa
		if(WaitForSingleObject(obj.SemFullSlots, INFINITE) != WAIT_OBJECT_0)
			ExitThread(2);

		if(WaitForSingleObject(obj.VectorMutex, INFINITE) != WAIT_OBJECT_0)
			ExitThread(2);

		cout << "Prelevato v[" << obj.start << "] = " << obj.v[obj.start];
		cout << " | " << (((obj.end-1) < obj.start)?obj.end+VSIZE:obj.end)-obj.start-1 
			 << " slot ancora occupati" << endl;
		obj.start++;
		obj.start %= VSIZE;
		cout.flush();

		ReleaseMutex(obj.VectorMutex);

		//C'è uno slot libero in più quindi il valore del semaforo deve aumetare
		ReleaseSemaphore(obj.SemEmptySlots, 1, NULL);

		//con questo si simula il lavoro del consumatore per elaborare l'elemento 
		//prelevato. Si ottiene anche di desincronizzare i due thread per rendere più
		//interessante l'evoluzione
		Sleep(rand()%5000); 
	}

	return 0;
}

void main()
{
	WorkerThreads t;
	if(t.Start() == false)
		return;

	getchar();

	t.Stop();
}

Così come sono stati scritti ThreadA e ThreadB ai possono avere anche istanze multiple dei vari thread e tutto funziona...

[miondere]

Ciao
Ho un problemino...
come posso fare a passare ai due thread un oggetto comune?
Devo passare l'oggetto heap.


void round_robin();
{
Heap heap(n);

WorkerThreads t;

if(t.Start() == false)
return;

cout<< "ROUND ROBIN TEST" << endl;

t.IsActive(INFINITE);

}

[cionci]

Metti lo Heap all'intero della classe WorkerThread...

[miondere]

Originally posted by "cionci"

Metti lo Heap all'intero della classe WorkerThread...

Scusa l'ignoranza...
come faccio?

Metto l'intera classe Heap dentro WorkerThread?

Per esempio:

class WorkerThreads
{
static DWORD WINAPI ThreadA(LPVOID lpParam);
static DWORD WINAPI ThreadB(LPVOID lpParam);
HANDLE h[2];
HANDLE StatusMutex;
HANDLE VectorMutex;
bool status;

class Heap {

//qui ci metto il contenuto della classe Heap
};


public:
WorkerThreads();
const bool Start();
const bool Stop();
const bool Status() const { return status; };
const bool IsActive(DWORD timeout);
Heap heap(); //qui dichiaro l'oggetto heap
};

Devo fare così?
Se sì, nn c'è un modo + elegante? Cioè richiamare la classe Heap dentro la classe WorkerThread senza riscrivere tutta la classe Heap dentro?

Ciao

[cionci]

La scrivi fuori...
All'internod ella classe WorkerThread devi solo dichiarare la variabile di tipo Heap...

[miondere]

Originally posted by "cionci"

La scrivi fuori...
All'internod ella classe WorkerThread devi solo dichiarare la variabile di tipo Heap...

Infatti all'inizio lo avevo messo, ti faccio l'esempio così ci capiamo meglio:


class WorkerThreads
{
static DWORD WINAPI ThreadA(LPVOID lpParam);
static DWORD WINAPI ThreadB(LPVOID lpParam);
HANDLE h[2];
HANDLE StatusMutex;
HANDLE VectorMutex;
bool status;

public:
WorkerThreads();
const bool Start();
const bool Stop();
const bool Status() const { return status; };
const bool IsActive(DWORD timeout);
Heap pippo(); //QUI DICHIARO L'OGGETTO pippo: DEVO FARE COSI'?!?
};


Mentre supponiamo che la classe Heap sia la seguente:



class Heap
{
public:
// ecc... ecc...

private:

int now;
};


A questo punto voglio capire come in ThreadA richiamo la variabile "now"...
Io ho messo così: obj.pippo.now=10; (vedi sotto)


DWORD WorkerThreads::ThreadA(LPVOID lpParam)
{

WorkerThreads &obj = *(WorkerThreads *)lpParam;

while(1)
{

//ecc.. ecc...
obj.pippo.now=10;

}

Però mi da errore! E penso che il compilatore nn abbia tutti i torti... Cosa devo modificare?

[cionci]

Heap pippo(); //QUI DICHIARO L'OGGETTO pippo: DEVO FARE COSI'?!?

Heap pippo; //pippo è una variabile di tipo Heap...

[miondere]

Ho implementato un mio programma con 2 thread, grazie al tuo codice
Però ora volevo capire bene le istruzioni e il loro significato.
Ho 3 dubbi, che sotto ti espongo.

1) Non capisco cosa significhi questa riga:

WorkerThreads &obj = *(WorkerThreads *)lpParam;

Non si poteva scrivere solo WorkerThreads obj ?



2)

if(WaitForSingleObject(StatusMutex, INFINITE) != WAIT_OBJECT_0)
return false;

Mi spieghi per favore cos'è WAIT_OBJECT_0 ?



3) L'ultima domanda:
in ExitThread( ) all'interno c'è un argomento, ma a cosa serve? Ho visto che prima hai messo 0, poi 1 ed infine 2, perchè?
Ho letto sulla MSDN che è un codice d'uscita, ma a chi serve tale codice? E' una sorta di identificativo?


Ciao e naturalmente grazie

[cionci]

obj diventa un alias per il nostro oggetto...di conseguenza qualsiasi modifica tu faccia a obj la fai all'oggetto che viene passato al thread...

Non si possono usare direttamente i membri dell'oggetto perchè la funzione che svolge il thread è una funzione statica di conseguenza gli passo il puntatore all'oggetto...

Magari potevo fare WorkerThread *pObj = (WorkerThreads *)lpParam; e dopo dovevo accedere ai membri con pObj->membro;

WAIT_OBJECT_0 è il valore che ritorna quando riesco a prendere possesso dell'oggetto di sincronizzazione...

Quello in ExitThread è il valore che ritorna il thread al chi l'ha chiamato...
Quando il chiamante rileva che un thread è uscito può ottenere questo valore con GetExitCodeThread e capire di conseguenza il motivo per cui è uscito...

[miondere]

Originally posted by "cionci"

obj diventa un alias per il nostro oggetto...di conseguenza qualsiasi modifica tu faccia a obj la fai all'oggetto che viene passato al thread...

Non si possono usare direttamente i membri dell'oggetto perchè la funzione che svolge il thread è una funzione statica di conseguenza gli passo il puntatore all'oggetto...

Magari potevo fare WorkerThread *pObj = (WorkerThreads *)lpParam; e dopo dovevo accedere ai membri con pObj->membro;

Sulla seconda e terza domanda tutto ok, ho capito...
Sulla prima ho ancora dubbi:

lpParam contiene this, giusto? Quindi contiene l'oggetto che ha chiamato Start(), cioè WorkerThreads...


WorkerThreads &obj ; //alias con obj di WorkerThreads
obj = *(WorkerThreads *)lpParam; //non lo capisco: lpParam è definito come puntatore a WorkerThreads e...?!?


WorkerThreads *pObj; //ptr ad oggetti WorkerThreads
pObj = (WorkerThreads *)lpParam // definisce lpParm come un ptr a WorkerThreads?!?


Con tutti questi ptr e oggetti mi perdo...forse non ho ancora abbastanza pratica...

[cionci]

Ti faccio un esempio con un int...almeno si fa prima...

lpParam è un puntatore void che sappiamo dover essere convertito ad int...

(int *)lpParam mi converte il tipo del puntatore da void * a int *...

*((int *)lpParam) quindi mi permette di operare sul contenuto dell' intero puntato da lpParam...

*((int *)lpParam) = 10;

int &i;

A "i" gli posso assegnare solamente un altro intero di cui "i" diventerà un alias... *((int *)lpParam) è un intero di conseguenza:

int &i = *((int *)lpParam);

oppure:

int &i = *(int *)lpParam;