C ++ Thread Server: Kontrol & manajemen thread
Jadi saya membuat server utas yang membuat utas dan mengaitkannya dengan pegangan sehingga Anda dapat menyimpan utas tertentu untuk tugas tertentu misalnya menjalankan semua rendering grafik pada utas 0, menjalankan fisika menginjak utas 1 dan menggunakan utas 2 untuk mengunduh file dari internet dll. Secara pribadi saya merasa ini berguna dalam proyek saya dan saya sangat suka ini untuk sistem pekerjaan biasa di mana Anda tidak memiliki kendali atas utas apa tugas tersebut dijalankan.
Saya ingin tahu ruang perbaikan apa yang ada, terutama dalam pengoptimalan dan apakah menurut Anda biaya tambahan itu dapat dibenarkan atau tidak.
Setiap utas menjalankan tugas dalam antrian tugas, jadi saya membuat Thread_Safe_Queue yang hanya pembungkus untuk std :: queue tetapi dengan mutex sebelum baca / tulis:
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;
}
Seperti yang Anda lihat, ada juga fungsi untuk menjeda / memulai / mematikan dan menunggu utas tertentu. Ini dapat digunakan untuk menyinkronkan utas dan menunggu nilai kembali, dll.
Sebagai contoh kasus penggunaan, Anda dapat memiliki satu utas untuk membuat simbol pemuatan sampai utas lain selesai misalnya mengunduh file dari internet dan kemudian menggunakan file yang diunduh:
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);
}
}
Dan inilah tes singkat untuk memastikannya bekerja dengan benar:
#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();
}
Sunting: Saya cukup baru dalam multithreading jadi jika Anda memiliki sesuatu untuk dikatakan tentang keamanan kode saya, saya akan sangat senang mendengarnya.
Jawaban
Gambaran
Jadi saya membuat server utas yang membuat utas dan mengaitkannya dengan pegangan sehingga Anda dapat menyimpan utas tertentu untuk tugas tertentu misalnya menjalankan semua rendering grafis pada utas 0, menjalankan fisika menginjak utas 1 dan menggunakan utas 2 untuk mengunduh file dari internet dll.
Saya tidak berpikir itu ide yang bagus secara umum. Padahal ini adalah cara mudah untuk memikirkan berbagai hal dan mungkin menjadikannya pengalaman belajar yang baik.
Secara umum, utas 2 akan duduk diam tanpa melakukan apa pun. Thread 0 akan menjadi sangat sibuk atau tidak melakukan apa-apa dan thread 1 mungkin akan terhalang dengan semua pekerjaan.
Saya tidak tahu tentang fisika. Tidak tahu bagaimana itu akan berhasil. Tetapi dua lainnya (Grafik / Internet) keduanya adalah aktivitas yang digerakkan oleh peristiwa. Mereka akan lebih baik disajikan dengan sistem acara Kecuali jika Anda berencana untuk menulis ini sendiri (tidak sepele) kemudian menggunakan orang lain.
Sekarang katakan itu. Keduanya mungkin merupakan loop acara terpisah dengan utas masing-masing. Tapi yang biasanya terjadi adalah Anda memiliki utas master yang terus-menerus menjalankan loop peristiwa, lalu ketika tindakan terjadi, utas master membuat membuat job(item kerja) yang diserahkan ke antrean pekerjaan. Anda kemudian memiliki banyak utas dalam antrian pekerjaan yang mengambil pekerjaan seperti yang muncul di antrian dan langsung menjalankannya.
Secara pribadi saya merasa ini berguna dalam proyek saya dan saya sangat lebih suka ini daripada sistem pekerjaan biasa di mana Anda tidak memiliki kendali atas utas apa tugas tersebut dijalankan.
Ya itu akan lebih mudah. Tapi sepertinya Anda harus banyak keadaan global. Anda harus status pembungkusan dalam item pekerjaan yang tidak memiliki status global yang dapat dikacaukan oleh beberapa utas.
Saya ingin tahu ruang perbaikan apa yang ada, terutama dalam pengoptimalan dan apakah menurut Anda biaya tambahan itu dapat dibenarkan atau tidak.
Mari lihat :-)
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:
Peninjauan kode.
Tidakkah Anda ingin mengunci otherantrian di sini?
Thread_Safe_Queue(Thread_Safe_Queue<T>&& other) noexcept {
std::lock_guard<std::mutex> lock(mutex);
queue = std::move(other.queue);
}
Ini tidak benar-benar aman utas jika Anda mengunci tujuan (yang tidak sepenuhnya terbentuk sehingga tidak dapat diserahkan ke utas lain) tetapi sumber masih bermutasi mungkin utas lain.
Anda yakin ingin dapat menyalin antrian?
Thread_Safe_Queue(const Thread_Safe_Queue<T>& other) {
std::lock_guard<std::mutex> lock(mutex);
queue = other.queue;
}
Anda tetap harus mengunci sumbernya!
Jadi Anda membungkus antrian sehingga Anda dapat menambahkan lock_guardspada setiap metode. Baik. Tapi agak boros. Panggilan ke empty()akan memberi tahu Anda jika antrian kosong pada saat itu tetapi pop berikutnya tidak dapat menjamin bahwa itu masih kosong saat Anda melepaskan kunci antara panggilan ke kosong dan panggilan ke pop.
std::optional<T> pop() {
std::optional<T> front() {
void push(const T &item) {
bool empty() const {
void clear() {
Saya akan menulis antrian yang bekerja di tingkat yang lebih tinggi. Bagaimana dengan antrian pemblokiran. Ingin meletuskan item. Jika tidak ada item untuk meletuskan utas diblokir sampai ada satu. Atau akan menunggu waktu minimum hingga objek muncul.
Menarik:
constexpr thread_id_t NULL_THREAD = (thread_id_t)0 - (thread_id_t)1;
Apakah ini cara penulisan yang rumit:
constexpr thread_id_t NULL_THREAD = static_cast<thread_id_t>(-1);
Dua hal.
- Hindari semua pengenal huruf besar. Secara teknis ini disediakan untuk makro.
- Lebih suka menggunakan cast C ++ daripada cast C.
Mengapa Anda melewati pointer?
void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
Apakah Anda melewati kepemilikan (kemudian gunakan std::unique_ptr<>). Dapatkah objek yang lewat menjadi nullptr(tidak terlihat seperti Anda tidak memeriksanya untuk null).
Lewati referensi, bukan penunjuk. Kemudian secara eksplisit berarti Anda tidak melewatkan kepemilikan dan fungsi yang dipanggil tidak boleh menghapus penunjuk. Jika tidak, ada kebingungan tentang apakah work()fungsi tersebut harus atau tidak harus menghapus penunjuk.
Ini terlihat seperti bug
while (contexts[tid].pause); // Loop forever !
Ini sama dengan:
while (contexts[tid].pause)
{}
Anda berhasil. Badan tidak mengubah status sehingga loop tidak dapat keluar.
void work(thread_id_t tid, std::vector<Thread_Context>* pcontexts) {
Harus membaca ke depan untuk memahami ini.
Maaf ini rusak.
Anda mencoba dan menyiasati fakta bahwa vektor dapat mengalokasikan kembali ruangnya dengan meneruskan indeks ke item pekerjaan dalam vektor. Masalahnya di sini adalah bahwa tidak ada batasan akses setelah utas dibuat dan utas baru (dibuat dengan make_thread()) dapat menyebabkan pcontextsukurannya diubah kapan saja. Akses ke vektor bukanlah thread safe, jadi jika vektor sedang diubah ukurannya, akses ke anggotanya melalui operator[]tidak dijamin valid.
Menetapkan utas ke satu item pekerjaan tidak terlalu produktif. Utas adalah objek yang relatif kelas berat sehingga Anda tidak ingin membuatnya mau tidak mau saat item pekerjaan baru dibuat.
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;
}
Anda harus membuat sekumpulan utas kerja lalu membiarkan mereka mengambil item pekerjaan dari antrian. Setelah selesai, pegang dengan variabel kondisi hingga ada pekerjaan yang tersedia.
Harapan
void actionToUploadFile()
{
workEventLoop.addItem([url]{
guiEvenLoop.addAlert('Downloading');
download(url);
guiEvenLoop.delAlert();
workEvenLoop.addItem(do_something);
});
}