L'Événement / La Tâche De La File D'Attente De Multithreading C++

Il y a Les contrats à terme bibliothèque fait son chemin dans Boost et le C++ de la bibliothèque standard. Il y a aussi quelque chose du même genre dans ACE, mais je n'aimerais pas à le recommander à n'importe qui (comme @lothar déjà souligné, c'est l'Objet Actif.)

OriginalL'auteur Nikolai Fetissov

Vous pouvez résoudre ce problème en utilisant Boost Fil de la bibliothèque. Quelque chose comme ceci (demi-pseudo):

class GThreadObject
{
...
public:
GThreadObject()
: _done(false)
, _newJob(false)
, _thread(boost::bind(>hreadObject::workerThread, this))
{
}
~GThreadObject()
{
_done = true;
_thread.join();
}
void functionOne(char *argOne, int argTwo)
{
...
_jobQueue.push(myEvent);
{
boost::lock_guard l(_mutex);
_newJob = true;
}
_cond.notify_one();
}
private:
void workerThread()
{
while (!_done) {
boost::unique_lock l(_mutex);
while (!_newJob) {
cond.wait(l);
}
Event *receivedEvent = _jobQueue.front();
...
}
}
private:
volatile bool             _done;
volatile bool             _newJob;
boost::thread             _thread;
boost::mutex              _mutex;
boost::condition_variable _cond;
std::queue<Event*>        _jobQueue;
};

Veuillez également noter comment RAII nous permettre de nous rendre ce code plus compact et mieux gérer.

OriginalL'auteur nhaa123

Pour l'extensibilité et la maintenabilité (et d'autres sur la responsabilité), vous pouvez définir une classe abstraite (ou d'interface) pour le "travail" que le thread est à effectuer. Utilisateur(s) de votre pool de threads mise en œuvre de cette interface et de faire référence à l'objet pour le pool de threads. Ceci est très similaire à Symbian Active de l'Objet de conception: chaque AO sous-classes CActive et ont à mettre en œuvre des méthodes telles que Run() et Cancel().

Pour des raisons de simplicité de votre interface (classe abstraite) pourrait être aussi simple que:

class IJob
{
virtual Run()=0;
};

Alors le fil de la piscine, ou seul thread à accepter les demandes aurait quelque chose comme:

class CThread
{
<...>
public:
void AddJob(IJob* iTask);
<...>
};

Naturellement, vous avez plusieurs tâches peuvent avoir toutes sortes d'extra-setters /getters /attributs et tout ce que vous devez dans tous les horizons de la vie. Cependant, la seule doit soit mettre en œuvre la méthode Run(), ce qui permettrait d'effectuer de longs calculs:

class CDumbLoop : public IJob
{
public:
CDumbJob(int iCount) : m_Count(iCount) {};
~CDumbJob() {};
void Run()
{
//Do anything you want here
}
private:
int m_Count;
};

OriginalL'auteur Ignas Limanauskas

Ci-dessous est une mise en œuvre qui ne nécessitent pas un "functionProxy" la méthode. Même si il est plus facile d'ajouter de nouvelles méthodes, il est toujours en désordre.

Boost::Bind et "Futures" ne semblent comme ils le feraient bien rangé, beaucoup de cela. Je crois que je vais avoir un coup d'oeil à l'élan de code et de voir comment il fonctionne. Merci pour vos suggestions de tout le monde.

GThreadObject.h

#include <queue>
using namespace std;
class GThreadObject
{
template <int size>
class VariableSizeContainter
{
char data[size];
};
class event
{
public:
void (GThreadObject::*funcPtr)(void *);
int dataSize;
char * data;
};
public:
void functionOne(char * argOne, int argTwo);
void functionTwo(int argTwo, int arg2);
private:
void newEvent(void (GThreadObject::*)(void*), unsigned int argStart, int argSize);
void workerThread();
queue<GThreadObject::event*> jobQueue;
void functionTwoInternal(int argTwo, int arg2);
void functionOneInternal(char * argOne, int argTwo);
};

GThreadObject.cpp

#include <iostream>
#include "GThreadObject.h"
using namespace std;
/* On a continuous loop, reading tasks from queue
* When a new event is received it executes the attached function pointer
* Thread code removed to decrease clutter
*/
void GThreadObject::workerThread()
{
//New Event added, process it
GThreadObject::event * receivedEvent = jobQueue.front();
/* Create an object the size of the stack the function is expecting, then cast the function to accept this object as an argument.
* This is the bit i would like to remove
* Only supports 8 byte argument size e.g 2 int's OR pointer + int OR myObject8bytesSize
* Subsequent data sizes would need to be added with an else if
* */
if (receivedEvent->dataSize == 8)
{
const int size = 8;
void (GThreadObject::*newFuncPtr)(VariableSizeContainter<size>);
newFuncPtr = (void (GThreadObject::*)(VariableSizeContainter<size>))receivedEvent->funcPtr;
//Execute the function
(*this.*newFuncPtr)(*((VariableSizeContainter<size>*)receivedEvent->data));
}
//Clean up
free(receivedEvent->data);
delete receivedEvent;
}
void GThreadObject::newEvent(void (GThreadObject::*funcPtr)(void*), unsigned int argStart, int argSize)
{
//Malloc an object the size of the function arguments
void * myData = malloc(argSize);
//Copy the data passed to this function into the buffer
memcpy(myData, (char*)argStart, argSize);
//Create the event and push it on to the queue
GThreadObject::event * myEvent = new event;
myEvent->data = (char*)myData;
myEvent->dataSize = argSize;
myEvent->funcPtr = funcPtr;
jobQueue.push(myEvent);
//This would be send a thread condition signal, replaced with a simple call here
this->workerThread();
}
/*
* This is the public interface, Can be called from child threads
* Instead of executing the event directly it adds it to a job queue
* Then the workerThread picks it up and executes all tasks on the same thread
*/
void GThreadObject::functionOne(char * argOne, int argTwo)
{
newEvent((void (GThreadObject::*)(void*))&GThreadObject::functionOneInternal, (unsigned int)&argOne, sizeof(char*)+sizeof(int));
}
/*
* This handles the actual event
*/
void GThreadObject::functionOneInternal(char * argOne, int argTwo)
{
cout << "We've made it to functionOne Internal char*:" << argOne << " int:" << argTwo << endl;
//Now do the work
}
void GThreadObject::functionTwo(int argOne, int argTwo)
{
newEvent((void (GThreadObject::*)(void*))&GThreadObject::functionTwoInternal, (unsigned int)&argOne, sizeof(int)+sizeof(int));
}
/*
* This handles the actual event
*/
void GThreadObject::functionTwoInternal(int argOne, int argTwo)
{
cout << "We've made it to functionTwo Internal arg1:" << argOne << " int:" << argTwo << endl;
}

main.cpp

#include <iostream>
#include "GThreadObject.h"
int main()
{
GThreadObject myObj;
myObj.functionOne("My Message", 23);
myObj.functionTwo(456, 23);
return 0;
}

Edit: Juste pour être complet, je n'ai qu'une mise en œuvre avec Boost::bind. Principales Différences:

queue<boost::function<void ()> > jobQueue;
void GThreadObjectBoost::functionOne(char * argOne, int argTwo)
{
jobQueue.push(boost::bind(&GThreadObjectBoost::functionOneInternal, this, argOne, argTwo));
workerThread();
}
void GThreadObjectBoost::workerThread()
{
boost::function<void ()> func = jobQueue.front();
func();
}

À l'aide de la poussée de la mise en œuvre de 10 000 000 d'Itérations de functionOne (), il a fallu ~19sec. Toutefois, la non améliorer la mise en œuvre a pris seulement ~6,5 sec. Si Environ 3x plus lent. Je devine trouver un bon non-fermeture de la file d'attente sera la plus grande performance de goulot de bouteille ici. Mais il est encore tout à fait une grande différence.

OriginalL'auteur Ben Reeves