C ++ Thread Server: controllo e gestione dei thread
Quindi ho creato un thread server che crea thread e li associa a un handle in modo da poter mantenere thread specifici per attività specifiche, ad esempio eseguire tutti i rendering grafici sul thread 0, eseguire i passaggi fisici sul thread 1 e utilizzare il thread 2 per scaricare un file del Internet ecc. Personalmente l'ho trovato utile in un mio progetto e lo preferisco di gran lunga a un tipico sistema di lavoro in cui non hai alcun controllo su quale thread viene eseguita l'attività.
Vorrei sapere quali margini di miglioramento ci sono, specialmente nell'ottimizzazione e se ritieni che il sovraccarico sia giustificato o meno.
Ogni thread esegue attività in una coda di attività, quindi ho creato un Thread_Safe_Queue che è solo un wrapper per std :: queue ma con un mutex prima di leggere / scrivere:
template<typename T>
struct Thread_Safe_Queue {
Thread_Safe_Queue() = default;
Thread_Safe_Queue(Thread_Safe_Queue<T>&& other) noexcept {
std::lock_guard<std::mutex> lock(mutex);
queue = std::move(other.queue);
}
Thread_Safe_Queue(const Thread_Safe_Queue<T>& other) {
std::lock_guard<std::mutex> lock(mutex);
queue = other.queue;
}
virtual ~Thread_Safe_Queue() { }
size_t size() const {
std::lock_guard<std::mutex> lock(mutex);
return queue.size();
}
std::optional<T> pop() {
std::lock_guard<std::mutex> lock(mutex);
if (queue.empty()) {
return {};
}
T tmp = queue.front();
queue.pop();
return tmp;
}
std::optional<T> front() {
std::lock_guard<std::mutex> lock(mutex);
if (queue.empty()) {
return {};
}
return queue.front();
}
void push(const T &item) {
std::lock_guard<std::mutex> lock(mutex);
queue.push(item);
}
bool empty() const {
std::lock_guard<std::mutex> lock(mutex);
return queue.empty();
}
void clear() {
std::lock_guard<std::mutex> lock(mutex);
queue = std::queue<T>();
}
std::queue<T> queue;
mutable std::mutex mutex;
};
Thread_Server.h:
#include "thread_safe_queue.h"
#include <thread>
#include <functional>
typedef unsigned int thread_id_t;
constexpr thread_id_t NULL_THREAD = (thread_id_t)0 - (thread_id_t)1;
typedef std::function<void()> Thread_Task;
struct Thread_Context {
Thread_Safe_Queue<Thread_Task> task_queue;
bool pause;
bool kill = false;
bool dead = false;
};
struct Thread_Server {
Thread_Server();
~Thread_Server();
thread_id_t make_thread(bool start = true);
void pause_thread(thread_id_t tid);
void start_thread(thread_id_t tid);
void kill_thread(thread_id_t tid);
void queue_task(thread_id_t tid, const Thread_Task& task);
void wait_for_thread(thread_id_t tid);
bool is_thread_busy(thread_id_t tid);
std::vector<Thread_Context> _thread_contexts;
};
thread_server.cpp:
#include "thread_server.h"
void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
auto& contexts = *pcontexts;
while (!contexts[tid].kill) {
while (contexts[tid].pause);
auto cmd = contexts[tid].task_queue.front();
if (cmd.has_value()) {
cmd.value()();
contexts[tid].task_queue.pop();
}
}
contexts[tid].dead = true;
}
Thread_Server::Thread_Server() {
}
Thread_Server::~Thread_Server() {
for (int i = 0; i < _thread_contexts.size(); i++) {
wait_for_thread(i);
_thread_contexts[i].kill = true;
}
}
thread_id_t Thread_Server::make_thread(bool start) {
thread_id_t tid = NULL_THREAD;
for (thread_id_t i = 0; i < _thread_contexts.size(); i++) {
if (_thread_contexts[i].dead) {
_thread_contexts[i].dead = false;
_thread_contexts[i].kill = false;
_thread_contexts[i].pause = !start;
_thread_contexts[i].task_queue.clear();
tid = i;
break;
}
}
if (tid == NULL_THREAD) {
tid = (thread_id_t)_thread_contexts.size();
Thread_Context ctx;
ctx.pause = !start;
_thread_contexts.push_back(ctx);
}
std::thread(work, tid, &_thread_contexts).detach();
return tid;
}
void Thread_Server::pause_thread(thread_id_t tid) {
_thread_contexts[tid].pause = true;
}
void Thread_Server::start_thread(thread_id_t tid) {
_thread_contexts[tid].pause = false;
}
void Thread_Server::kill_thread(thread_id_t tid) {
_thread_contexts[tid].kill = true;
}
void Thread_Server::queue_task(thread_id_t tid, const Thread_Task& task) {
auto& ctx = _thread_contexts[tid];
ctx.task_queue.push(task);
}
void Thread_Server::wait_for_thread(thread_id_t tid) {
auto& ctx = _thread_contexts[tid];
while (ctx.task_queue.size() > 0);
}
bool Thread_Server::is_thread_busy(thread_id_t tid) {
return _thread_contexts[tid].task_queue.size() > 0;
}
Come puoi vedere ci sono anche funzioni per mettere in pausa / avviare / uccidere e attendere un thread specifico. Questo può essere utilizzato per sincronizzare i thread e attendere i valori restituiti, ecc.
Come esempio di caso d'uso, potresti avere un thread per il rendering di un simbolo di caricamento fino al termine di un altro thread, ad esempio il download di un file da Internet e quindi l'utilizzo del file scaricato:
void update() {
if (want_to_download_something) {
thread_server.queue_task(download_thread, [url]() {
download_result = download(url);
});
}
if (thread_server.is_thread_busy(download_thread)) {
render_loading_icon("Downloading file");
} else if (download_result) {
do_something(download_result);
}
}
Ed ecco un rapido test per assicurarti che funzioni correttamente:
#include <iostream>
#include "thread_server.h"
int main()
{
Thread_Server thread_server;
auto t1 = thread_server.make_thread();
auto t2 = thread_server.make_thread();
auto t3 = thread_server.make_thread();
for (int i = 0; i < 10; i++) {
thread_server.queue_task(t1, []() {
std::cout << "\nHello from thread 1 (std::this_thread::get_id(): " << std::this_thread::get_id() << " )\n";
});
thread_server.queue_task(t2, []() {
std::cout << "\nHello from thread 2 (std::this_thread::get_id(): " << std::this_thread::get_id() << " )\n";
});
thread_server.queue_task(t3, []() {
std::cout << "\nHello from thread 3 (std::this_thread::get_id(): " << std::this_thread::get_id() << " )\n";
});
}
std::cin.get();
}
Modifica: sono abbastanza nuovo nel multithreading, quindi se hai qualcosa da dire sulla sicurezza del mio codice sarò più che felice di ascoltarlo.
Risposte
Panoramica
Quindi ho creato un thread server che crea thread e li associa a un handle in modo da poter mantenere thread specifici per attività specifiche, ad esempio eseguire tutti i rendering grafici sul thread 0, eseguire passaggi fisici sul thread 1 e utilizzare il thread 2 per scaricare un file internet ecc.
Non penso che sia una buona idea in generale. Anche se è un modo semplice per pensare alle cose e può renderlo una buona esperienza di apprendimento.
In generale, il thread 2 si siederà senza fare nulla. Il thread 0 sarà eccessivamente occupato o non farà nulla e il thread 1 sarà probabilmente bloccato con tutto il lavoro.
Non so nulla della fisica. Non ho idea di come funzionerebbe. Ma le altre due (Grafica / Internet) sono entrambe attività guidate da eventi. Sarebbero meglio serviti con un sistema di eventi A meno che tu non abbia intenzione di scriverlo da solo (non banale), allora usa qualcun altro.
Ora lo dico. Entrambi possono essere i propri loop di eventi separati con un thread ciascuno. Ma ciò che di solito accade è che hai un thread principale che esegue costantemente il ciclo di eventi, quindi quando si verifica un'azione il thread principale crea un job(elemento di lavoro) che viene passato a una coda di lavoro. Hai quindi un mucchio di thread nella coda di lavoro che acquisiscono i lavori come appaiono nella coda e semplicemente li eseguono.
Personalmente l'ho trovato utile in un mio progetto e lo preferisco di gran lunga a un tipico sistema di lavoro in cui non hai alcun controllo su quale thread viene eseguita l'attività.
Sì, sarà più facile. Ma sembra che tu abbia molto stato globale. Dovresti avvolgere lo stato negli elementi di lavoro che non hanno uno stato globale che può essere incasinato da più thread.
Vorrei sapere quali margini di miglioramento ci sono, specialmente nell'ottimizzazione e se ritieni che il sovraccarico sia giustificato o meno.
Diamo un'occhiata :-)
Each thread executes tasks in a task queue, so I made a Thread_Safe_Queue which is just a wrapper for std::queue but with a mutex before read/writes:
Revisione del codice.
Non vuoi bloccare le othercode qui?
Thread_Safe_Queue(Thread_Safe_Queue<T>&& other) noexcept {
std::lock_guard<std::mutex> lock(mutex);
queue = std::move(other.queue);
}
Non è realmente thread-safe se blocchi la destinazione (che non è completamente formata, quindi non può essere stata passata a un altro thread) ma l'origine è ancora in fase di modifica, forse un altro thread.
Vuoi davvero poter copiare le code?
Thread_Safe_Queue(const Thread_Safe_Queue<T>& other) {
std::lock_guard<std::mutex> lock(mutex);
queue = other.queue;
}
Dovresti comunque bloccare la fonte!
Quindi hai avvolto la coda in modo da poter aggiungere lock_guardsogni metodo. Bene. Ma un po 'dispendioso. Una chiamata a empty()ti dirà se la coda è vuota in quel punto, ma un pop successivo non può garantire che sia ancora vuota quando hai rilasciato il blocco tra la chiamata a vuoto e la chiamata a pop.
std::optional<T> pop() {
std::optional<T> front() {
void push(const T &item) {
bool empty() const {
void clear() {
Vorrei scrivere una coda che funziona a un livello superiore. Che ne dici di una coda di blocco. Vuoi far scoppiare un oggetto. Se non è presente alcun elemento da far apparire, il thread viene bloccato finché non ce n'è uno. Oppure attenderà una quantità minima di tempo prima che l'oggetto appaia.
Interessante:
constexpr thread_id_t NULL_THREAD = (thread_id_t)0 - (thread_id_t)1;
È un modo di scrivere complesso:
constexpr thread_id_t NULL_THREAD = static_cast<thread_id_t>(-1);
Due cose.
- Evita tutti gli identificatori maiuscoli. Tecnicamente questi sono riservati alle macro.
- Preferisco usare cast C ++ piuttosto che cast C.
Perché stai passando per pointer?
void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
Stai passando la proprietà (quindi usa std::unique_ptr<>). Può essere l'oggetto passato nullptr(non sembra che non lo controlli per null).
Passa per riferimento anziché per puntatore. Quindi significa esplicitamente che non stai passando la proprietà e la funzione chiamata non dovrebbe eliminare il puntatore. Altrimenti c'è confusione sul fatto che la work()funzione debba o meno cancellare il puntatore.
Questo sembra un bug
while (contexts[tid].pause); // Loop forever !
Questo è equivalente a:
while (contexts[tid].pause)
{}
Hai raggiunto il loop. Il corpo non cambia stato, quindi non è possibile uscire dal ciclo.
void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
Ho dovuto leggere in avanti per capire questo.
Spiacente, questo è rotto.
Si cerca di aggirare il fatto che un vettore può riallocare il proprio spazio passando un indice all'elemento di lavoro nel vettore. Il problema qui è che non ci sono restrizioni di accesso una volta che il thread viene creato e un nuovo thread (creato con make_thread()) può causare il pcontextsridimensionamento in qualsiasi momento. L'accesso a un vettore non è thread-safe, quindi se il vettore sta per essere ridimensionato, gli accessi ai suoi membri tramite operator[]non sono garantiti per essere validi.
Assegnare un thread a un singolo elemento di lavoro non è molto produttivo. Un filo è un oggetto relativamente pesante, quindi non vuoi crearli volenti o nolenti quando vengono creati nuovi elementi di lavoro.
void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
auto& contexts = *pcontexts;
while (!contexts[tid].kill) {
while (contexts[tid].pause);
auto cmd = contexts[tid].task_queue.front();
if (cmd.has_value()) {
cmd.value()();
contexts[tid].task_queue.pop();
}
}
contexts[tid].dead = true;
}
Dovresti creare un gruppo di thread di lavoro e poi lasciare che raccolgano gli elementi di lavoro dalla coda. Quando hanno finito, tienili con una variabile di condizione finché non c'è lavoro disponibile.
Aspettative
void actionToUploadFile()
{
workEventLoop.addItem([url]{
guiEvenLoop.addAlert('Downloading');
download(url);
guiEvenLoop.delAlert();
workEvenLoop.addItem(do_something);
});
}