C++ 11 shared_ptr in multi-threads

Question

Kürzlich denke ich ein leistungsfähiges Event-driven Multithreads-Framework mit C++ 11. Und es werden hauptsächlich C++ 11 Einrichtungen wie std::thread, std::condition_variable, std::mutex, std::shared_ptr usw. in Betracht gezogen. Im Allgemeinen hat dieser Rahmen drei grundlegende Komponenten: Arbeit, Arbeit und rationalisieren, nun, es scheint eine echte Fabrik zu sein. Wenn der Benutzer sein Geschäftsmodell am Server erstellt, muss er nur die Daten und den Prozessor berücksichtigen. Sobald das Modell erstellt ist, muss der Benutzer nur den vererbten Worker der Datenklasse geerbter Job und Prozessorklasse erstellen.

Zum Beispiel:

class Data : public job {}; 
class Processsor : public worker {}; 

Wenn Serverdaten erhalten, es ist nur ein Datenobjekt durch auto data = std::make_shared<Data>() in der Datenquelle Rückruf-Thread neu und die Stromlinie nennen. job_dispatch, um den Prozessor und die Daten in einen anderen Thread zu übertragen. Natürlich muss der Benutzer nicht daran denken, Speicher freizugeben. Die Stromlinienform. job_dispatch tun hauptsächlich unter stuff:

void evd_thread_pool::job_dispatch(std::shared_ptr<evd_thread_job> job) { 
    auto task = std::make_shared<evd_task_wrap>(job); 
    task->worker = streamline.worker; 
    // worker has been registered in streamline first of all 
    { 
     std::unique_lock<std::mutex> lck(streamline.mutex); 
     streamline.task_list.push_back(std::move(task)); 
    } 
    streamline.cv.notify_all(); 
} 

Die evd_task_wrap im job_dispatch verwendet wie folgt definiert:

struct evd_task_wrap { 
    std::shared_ptr<evd_thread_job> order; 
    std::shared_ptr<evd_thread_processor> worker; 
    evd_task_wrap(std::shared_ptr<evd_thread_job>& o) 
    :order(o) {} 
}; 

Schließlich wird die task_wrap wird durch task_list in die Verarbeitungsthread versandt werden, daß ein std::list Objekt ist. Und der Verarbeitungs-Thread vor allem tut das Zeug wie:

void evd_factory_impl::thread_proc() { 
    std::shared_ptr<evd_task_wrap> wrap = nullptr; 
    while (true) { 
     { 
      std::unique_lock<std::mutex> lck(streamline.mutex); 
      if (streamline.task_list.empty()) 
       streamline.cv.wait(lck, 
       [&]()->bool{return !streamline.task_list.empty();}); 
      wrap = std::move(streamline.task_list.front()); 
      streamline.task_list.pop_front(); 
     } 
     if (-1 == wrap->order->get_type()) 
      break; 
     wrap->worker->process_task(wrap->order); 
     wrap.reset(); 
    } 
} 

Aber ich weiß nicht, warum der Prozess oft in der thread_proc Funktion abstürzt. Und der Coredump-Prompt, dass manchmal der Umbruch ist ein leerer shared_ptr oder Segmentfehler passiert in _Sp_counted_ptr_inplace::_M_dispose, die in wrap.reset() aufgerufen wird. Und ich nehme an, dass shared_ptr das threadsynchrone Problem in diesem Szenario hat, während ich weiß, dass der Kontrollblock in shared_ptr Thread-Sicherheit ist. Und natürlich ist die shared_ptr in job_dispatch und thread_proc andere shared_ptr Objekt, obwohl sie auf den gleichen Speicher verweisen. Hat jemand einen spezifischeren Vorschlag zur Lösung dieses Problems? Oder wenn es vorhanden ist ähnlich leichten Rahmen mit automatischer Speicherverwaltung mit C++ 11

---

Das Beispiel process_task wie: void log_handle::process_task(std::shared_ptr<crx::evd_thread_job> job) { auto j = std::dynamic_pointer_cast<log_job>(job); j->log->Printf(0, j->print_str.c_str()); write(STDOUT_FILENO, j->print_str.c_str(), j->print_str.size()); } class log_factory { public: log_factory(const std::string& name); virtual ~log_factory(); void print_ts(const char *format, ...) { //here dispatch the job char log_buf[4096] = {0}; va_list args; va_start(args, format); vsprintf(log_buf, format, args); va_end(args); auto job = std::make_shared<log_job>(log_buf, &m_log); m_log_th.job_dispatch(job); } public: E15_Log m_log; std::shared_ptr<log_handle> m_log_handle; crx::evd_thread_pool m_log_th; };

Answer 1

ich ein Problem in Ihrem Code festgestellt, die nicht sein können oder verbunden:

Sie verwenden notify_all von Ihrer Zustandsvariablen. Das wird ALLE Fäden aus dem Schlaf wecken. Es ist OK, wenn Sie Ihre wait in einer while-Schleife wickeln, wie:

while (streamline.task_list.empty()) 
    streamline.cv.wait(lck, [&]()->bool{return !streamline.task_list.empty();}); 

Aber da Sie eine if verwenden, alle Fäden verlassen die wait. Wenn Sie ein einzelnes Produkt versenden und mehrere Consumer-Threads haben, rufen alle außer einem Thread wrap = std::move(streamline.task_list.front()); auf, während die Taskliste leer ist und UB verursacht.