c++ - Producer-consumer based multi-threading for image processing -

update: have provided reason of problem , solution in answer below.

i want implement multi-threading based upon producer-consumer approach image processing task. case, producer thread should grabs images , put them container whereas consumer thread should extract images container thread. think should use queue implementation of container.

i want use following code suggested in so answer. have become quite confused implementation of container , putting incoming image in producer thread.

problem: image displayed first consumer thread not contain full data. and, second consumer thread never displays image. may be, there race situation or lock situation due second thread not able access data of queue @ all. have tried use mutex.

#include <vector> #include <thread> #include <memory> #include <queue>  #include <opencv2/highgui.hpp> #include <opencv2/core.hpp> #include <opencv2/imgproc.hpp>  mutex mu;  struct threadsafecontainer {      queue<unsigned char*> safecontainer;  };  struct producer {     producer(std::shared_ptr<threadsafecontainer> c) : container(c)     {      }      void run()     {         while(true)         {             // grab image camera             // store image in container             mat image(400, 400, cv_8uc3, scalar(10, 100,180) );             unsigned char *pt_src = image.data;             mu.lock();             container->safecontainer.push(pt_src);             mu.unlock();         }     }      std::shared_ptr<threadsafecontainer> container; };  struct consumer {     consumer(std::shared_ptr<threadsafecontainer> c) : container(c)     {      }     ~consumer()     {      }      void run()     {         while(true)         {             // read next image container         mu.lock();         if (!container->safecontainer.empty())             {                 unsigned char *ptr_consumer_image;                 ptr_consumer_image = container->safecontainer.front(); //the front of queue contain pointer image data                 container->safecontainer.pop();                  mat image(400, 400, cv_8uc3);                 image.data = ptr_consumer_image;                  imshow("consumer image", image);                 waitkey(33);             }                    mu.unlock();         }     }      std::shared_ptr<threadsafecontainer> container; };    int main() {     //pointer object class containing "container" "producer" , "consumer" put , take images     auto ptrobject_container = make_shared<threadsafecontainer>();      //pointer object producer...intialize "container" variable of "struct producer" above created common "container"     auto ptrobject_producer = make_shared<producer>(ptrobject_container);       //first pointer object consumer...intialize "container" variable of "struct consumer" above created common "container"     auto first_ptrobject_consumer = make_shared<consumer>(ptrobject_container);      //second pointer object consumer...intialize "container" variable of "struct consumer" above created common "container"     auto second_ptrobject_consumer = make_shared<consumer>(ptrobject_container);      //run producer thread     thread producerthread(&producer::run, ptrobject_producer);       //run first thread of consumer     thread first_consumerthread(&consumer::run, first_ptrobject_consumer);      //run second thread of consumer     thread second_consumerthread(&consumer::run, second_ptrobject_consumer);      //join threads     producerthread.join();     first_consumerthread.join();     second_consumerthread.join();      return 0; } 

ok, make simple possible. need 2 threads, mutex, queue , 2 thread processing functions.

header.h   static dword winapi threadfunc_prod(lpvoid lpparam); static dword winapi threadfunc_con(lpvoid lpparam);  handle m_hthread[2]; queue<int> m_q; mutex m_m; 

add needed stuff, these core parts need

source.cpp   dword dwthreadid; m_hthread[0] = createthread(null, 0, this->threadfunc_prod, this, 0, &dwthreadid); // same 2nd thread  dword winapi server::threadfunc_prod(lpvoid lpparam) {     cyourclass* o = (cyourclass*) lpparam;     int ndata2q = getdata();    // whatever use data     m_m.lock();     m_q.push(ndata2q);     m_m.unlock(); }  dword winapi server::threadfunc_con(lpvoid lpparam) {     cyourclass* o = (cyourclass*) lpparam;     int res;     m_m.lock();     if (m_q.empty())     {         // bad, no data, escape or wait or whatever, don't block context     }     else      {         res = m_q.front();         m_q.pop();     }     m_m.unlock();     // magic res here } 

and in end of main - don't forget use waitformultipleobjects possible examples can found directly in msdn there quite nice commentary that.

part2: ok, believe header self-explainable, give little bit more description source. somewhere in source (can in constructor) create threads - way how create thread may differ idea same (in win - thread run right after creation in posix u have join). believe u shall have somewhere function starts magic, lets call magickicker() in case of posix, create thread in constructor , join em in magickicker(), win - create in magickicker()

than need declare (in header) 2 function thread function implemented threadfunc_prod , threadfunc_prod , important magic here pass reference object function (coz thread static) u can easy access shared resources queues, mutexes, etc... these function doing work. have u need in code, use adding routine in producer:

int ndata2q = getdata();    // whatever use data m_m.lock();                 // locks mutex nobody cant enter mutex m_q.push(ndata2q);          // puts data producer share queue m_m.unlock();               // unlock mutex u can access mutex in consumer 

and add consumer:

int res; m_m.lock();                 // locks mutex u cant access wrapped mutex in producer if (m_q.empty())            // check if there in queue {                             // nothing in queue yet or                             // skip thread run, can i.e. sleep time build queue     sleep(100);     continue; // in case of while wrap     return;   // in case u r running framework threadloop } else                        // there {     res = m_q.front();      // oldest element of queue     m_q.pop();              // delete element queue } m_m.unlock();               // unlock mutex producer can add new items queue // magic res here