multithreading on OSX

[si_icarus]

Hi all, I have a query about multithreading. What I would like to do is, at the start of my main update() function, start a couple of threads in parallel, once they are all complete carry on with my main update function.

<br /> void update() {<br /> thread1->update(); // fluid solver<br /> thread2->update(); // particle system<br /> thread3->update(); // camera1 optical flow<br /> thread4->update(); // camera2 optical flow<br /> // all of the above threads have their own data and work independantly<br /> <br /> etc.<br /> .<br /> .. wait till all the above threads are done.<br /> .. once they are done, feed the data from each of them into other ones<br /> .. e.g. fluid velocity info is added to particle systems reedy for next frame's update<br /> .. e.g. camera optical flow data is fed into fluid solver, ready for next frame's update<br /> <br /> render all<br /> }<br />

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

<br /> void update() {<br /> thread1->update();  // fluid solver<br /> thread2->update();  // particle system<br /> thread3->update();  // camera1 optical flow<br /> thread4->update();  // camera2 optical flow<br /> // all of the above threads have their own data and work independantly<br /> <br /> etc.<br /> .<br /> .. wait till all the above threads are done.<br /> .. once they are done, feed the data from each of them into other ones<br /> .. e.g. fluid velocity info is added to particle systems reedy for next frame's update<br /> .. e.g. camera optical flow data is fed into fluid solver, ready for next frame's update<br /> <br /> render all<br /> }<br />

I am having trouble getting this working the way I want. I don’t want my threads to loop continuously independent of my main update loop, I want them all to start at the beginning of my update loop, and then once they are all finished, feed the data around, then render, then repeat.

I tried creating a thread every update loop which runs only once (no while() in the function) – it works and behaves fine for a few minutes, then the threads just stop being created – no crash or hang, just no updates from threads. I also tried using an infinte while() but with a flag that determined whether the code ran that frame or not, and only set that flag at the beginning of the main update loop (code for this below) – that one is very stable, but behaviour was weird, I think the thread updates weren’t being called when I wanted them to.

I’ve read a lot of documentation on multithreading in general and all the pthread commands. I understand the fundamentals of protecting data and all that.. but I’m just lacking a lot of experience and strategy .. so any suggestions welcome!

This is my thread class I’m using which has the infinite loop in the threaded function, but runs the update function if the flag is set. I initialize the flag at the beginning of each main update with thread1->runOnce(); thread2->runOnce(); etc…. but it doesn’t run smoothly.

<br /> <br /> #include "ofxThread.h"<br /> <br /> class MSAThread : public ofxThread {<br /> <br /> bool bHasRunThisFrame;<br /> bool bAutoLoop;<br /> int interval;<br /> <br /> public:<br /> <br /> // create the thread, if autoLoop is true, start running straight away and loop,<br /> // otherwise wait for the runOnce command<br /> void start(int frameRate = 30, bool initAutoLoop = false){<br /> interval = 1000/frameRate;<br /> bAutoLoop = initAutoLoop;<br /> if(!bAutoLoop) bHasRunThisFrame = true; // don't run straight away<br /> startThread(true, false); // blocking, verbose<br /> }<br /> <br /> // run the update function once in another thread<br /> void runOnce() {<br /> if(lock()) {<br /> bAutoLoop = false;<br /> bHasRunThisFrame = false;<br /> unlock();<br /> }<br /> }<br /> <br /> // run the update function in another thread<br /> void threadedFunction() {<br /> while(isThreadRunning()) {<br /> if( lock() ){<br /> if(bAutoLoop || !bHasRunThisFrame) {<br /> update();<br /> bHasRunThisFrame = true;<br /> unlock();<br /> ofSleepMillis(interval);<br /> } else {<br /> unlock();<br /> }<br /> }<br /> }<br /> }<br /> <br /> // main thread waits for the update function to finish<br /> void waitForFinish() { if(lock()) { unlock(); } }<br /> <br /> // this is the function that gets overvidden<br /> virtual void update() = 0;<br /> };<br />

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53

<br /> <br /> #include "ofxThread.h"<br /> <br /> class MSAThread : public ofxThread {<br /> <br /> bool bHasRunThisFrame;<br /> bool bAutoLoop;<br /> int interval;<br /> <br /> public:<br /> <br /> // create the thread, if autoLoop is true, start running straight away and loop,<br /> // otherwise wait for the runOnce command<br /> void start(int frameRate = 30, bool initAutoLoop = false){<br /> interval = 1000/frameRate;<br /> bAutoLoop = initAutoLoop;<br /> if(!bAutoLoop) bHasRunThisFrame = true;          // don't run straight away<br /> startThread(true, false);   // blocking, verbose<br /> }<br /> <br /> // run the update function once in another thread<br /> void runOnce() {<br /> if(lock()) {<br /> bAutoLoop = false;<br /> bHasRunThisFrame = false;<br /> unlock();<br /> }<br /> }<br /> <br /> // run the update function in another thread<br /> void threadedFunction() {<br /> while(isThreadRunning()) {<br /> if( lock() ){<br /> if(bAutoLoop || !bHasRunThisFrame) {<br /> update();<br /> bHasRunThisFrame = true;<br /> unlock();<br /> ofSleepMillis(interval);<br /> } else {<br /> unlock();<br /> }<br /> }<br /> }<br /> }<br /> <br /> // main thread waits for the update function to finish<br /> void waitForFinish() { if(lock()) { unlock(); } }<br /> <br /> // this is the function that gets overvidden<br /> virtual void update() = 0;<br /> };<br />

and it’s based on this OpenFrameworks class:

<br /> class ofxThread{<br /> <br /> public:<br /> ofxThread();<br /> virtual ~ofxThread();<br /> bool isThreadRunning();<br /> void startThread(bool _blocking = true, bool _verbose = true);<br /> bool lock();<br /> bool unlock();<br /> void stopThread();<br /> <br /> protected:<br /> <br /> // <hr class="bbcode_rule" /> //you need to overide this with the function you want to thread<br /> virtual void threadedFunction(){<br /> if(verbose)printf("ofxThread: overide threadedFunction with your ownn");<br /> }<br /> <br /> // <hr class="bbcode_rule" /> <br /> #ifdef TARGET_WIN32<br /> static unsigned int __stdcall thread(void * objPtr){<br /> ofxThread* me = (ofxThread*)objPtr;<br /> me->threadedFunction();<br /> return 0;<br /> }<br /> <br /> #else<br /> static void * thread(void * objPtr){<br /> ofxThread* me = (ofxThread*)objPtr;<br /> me->threadedFunction();<br /> return 0;<br /> }<br /> #endif<br /> <br /> <br /> #ifdef TARGET_WIN32<br /> HANDLE myThread;<br /> CRITICAL_SECTION critSec; //same as a mutex<br /> #else<br /> pthread_t myThread;<br /> pthread_mutex_t myMutex;<br /> #endif<br /> <br /> bool threadRunning;<br /> bool blocking;<br /> bool verbose;<br /> };<br /> <br /> <br /> // <hr class="bbcode_rule" /> ofxThread::ofxThread(){<br /> threadRunning = false;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> ofxThread::~ofxThread(){<br /> stopThread();<br /> }<br /> <br /> // <hr class="bbcode_rule" /> bool ofxThread::isThreadRunning(){<br /> //should be thread safe - no writing of vars here<br /> return threadRunning;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> void ofxThread::startThread(bool _blocking, bool _verbose){<br /> <br /> //have to put this here because the thread can be running<br /> //before the call to create it returns<br /> threadRunning = true;<br /> <br /> #ifdef TARGET_WIN32<br /> InitializeCriticalSection(&#038;critSec);<br /> myThread = (HANDLE)_beginthreadex(NULL, 0, this->thread, (void *)this, 0, NULL);<br /> #else<br /> pthread_mutex_init(&#038;myMutex, NULL);<br /> pthread_create(&#038;myThread, NULL, thread, (void *)this);<br /> #endif<br /> <br /> blocking = _blocking;<br /> verbose = _verbose;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> //returns false if it can't lock<br /> bool ofxThread::lock(){<br /> if(!threadRunning){<br /> if(verbose)printf("ofxThread: need to call startThread firstn");<br /> return false;<br /> }<br /> <br /> #ifdef TARGET_WIN32<br /> if(blocking)EnterCriticalSection(&#038;critSec);<br /> else {<br /> if(!TryEnterCriticalSection(&#038;critSec)){<br /> if(verbose)printf("ofxThread: mutext is busy n");<br /> return false;<br /> }<br /> }<br /> if(verbose)printf("ofxThread: we are in -- mutext is now locked n");<br /> #else<br /> <br /> if(blocking){<br /> if(verbose)printf("ofxThread: waiting till mutext is unlockedn");<br /> pthread_mutex_lock(&#038;myMutex);<br /> if(verbose)printf("ofxThread: we are in -- mutext is now locked n");<br /> }else{<br /> int value = pthread_mutex_trylock(&#038;myMutex);<br /> if(value == 0){<br /> if(verbose)printf("ofxThread: we are in -- mutext is now locked n");<br /> }<br /> else{<br /> if(verbose)printf("ofxThread: mutext is busy - already lockedn");<br /> return false;<br /> }<br /> }<br /> #endif<br /> <br /> return true;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> bool ofxThread::unlock(){<br /> <br /> if(!threadRunning){<br /> if(verbose)printf("ofxThread: need to call startThread firstn");<br /> return false;<br /> }<br /> <br /> #ifdef TARGET_WIN32<br /> LeaveCriticalSection(&#038;critSec);<br /> #else<br /> pthread_mutex_unlock(&#038;myMutex);<br /> #endif<br /> <br /> if(verbose)printf("ofxThread: we are out -- mutext is now unlocked n");<br /> <br /> return true;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> void ofxThread::stopThread(){<br /> if(threadRunning){<br /> #ifdef TARGET_WIN32<br /> CloseHandle(myThread);<br /> #else<br /> pthread_mutex_destroy(&#038;myMutex);<br /> pthread_detach(myThread);<br /> #endif<br /> if(verbose)printf("ofxThread: thread stoppedn");<br /> threadRunning = false;<br /> }else{<br /> if(verbose)printf("ofxThread: thread already stoppedn");<br /> }<br /> }<br />

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169

<br /> class ofxThread{<br /> <br /> public:<br /> ofxThread();<br /> virtual ~ofxThread();<br /> bool isThreadRunning();<br /> void startThread(bool _blocking = true, bool _verbose = true);<br /> bool lock();<br /> bool unlock();<br /> void stopThread();<br /> <br /> protected:<br /> <br /> // <hr class="bbcode_rule" /> //you need to overide this with the function you want to thread<br /> virtual void threadedFunction(){<br /> if(verbose)printf("ofxThread: overide threadedFunction with your ownn");<br /> }<br /> <br /> // <hr class="bbcode_rule" /> <br /> #ifdef TARGET_WIN32<br /> static unsigned int __stdcall thread(void * objPtr){<br /> ofxThread* me     = (ofxThread*)objPtr;<br /> me->threadedFunction();<br /> return 0;<br /> }<br /> <br /> #else<br /> static void * thread(void * objPtr){<br /> ofxThread* me     = (ofxThread*)objPtr;<br /> me->threadedFunction();<br /> return 0;<br /> }<br /> #endif<br /> <br /> <br /> #ifdef TARGET_WIN32<br /> HANDLE            myThread;<br /> CRITICAL_SECTION  critSec;       //same as a mutex<br /> #else<br /> pthread_t        myThread;<br /> pthread_mutex_t  myMutex;<br /> #endif<br /> <br /> bool threadRunning;<br /> bool blocking;<br /> bool verbose;<br /> };<br /> <br /> <br /> // <hr class="bbcode_rule" /> ofxThread::ofxThread(){<br /> threadRunning = false;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> ofxThread::~ofxThread(){<br /> stopThread();<br /> }<br /> <br /> // <hr class="bbcode_rule" /> bool ofxThread::isThreadRunning(){<br /> //should be thread safe - no writing of vars here<br /> return threadRunning;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> void ofxThread::startThread(bool _blocking, bool _verbose){<br /> <br /> //have to put this here because the thread can be running<br /> //before the call to create it returns<br /> threadRunning     = true;<br /> <br /> #ifdef TARGET_WIN32<br /> InitializeCriticalSection(&#038;critSec);<br /> myThread = (HANDLE)_beginthreadex(NULL, 0, this->thread,  (void *)this, 0, NULL);<br /> #else<br /> pthread_mutex_init(&#038;myMutex, NULL);<br /> pthread_create(&#038;myThread, NULL, thread, (void *)this);<br /> #endif<br /> <br /> blocking          =     _blocking;<br /> verbose               = _verbose;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> //returns false if it can't lock<br /> bool ofxThread::lock(){<br /> if(!threadRunning){<br /> if(verbose)printf("ofxThread: need to call startThread firstn");<br /> return false;<br /> }<br /> <br /> #ifdef TARGET_WIN32<br /> if(blocking)EnterCriticalSection(&#038;critSec);<br /> else {<br /> if(!TryEnterCriticalSection(&#038;critSec)){<br /> if(verbose)printf("ofxThread: mutext is busy n");<br /> return false;<br /> }<br /> }<br /> if(verbose)printf("ofxThread: we are in -- mutext is now locked n");<br /> #else<br /> <br /> if(blocking){<br /> if(verbose)printf("ofxThread: waiting till mutext is unlockedn");<br /> pthread_mutex_lock(&#038;myMutex);<br /> if(verbose)printf("ofxThread: we are in -- mutext is now locked n");<br /> }else{<br /> int value = pthread_mutex_trylock(&#038;myMutex);<br /> if(value == 0){<br /> if(verbose)printf("ofxThread: we are in -- mutext is now locked n");<br /> }<br /> else{<br /> if(verbose)printf("ofxThread: mutext is busy - already lockedn");<br /> return false;<br /> }<br /> }<br /> #endif<br /> <br /> return true;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> bool ofxThread::unlock(){<br /> <br /> if(!threadRunning){<br /> if(verbose)printf("ofxThread: need to call startThread firstn");<br /> return false;<br /> }<br /> <br /> #ifdef TARGET_WIN32<br /> LeaveCriticalSection(&#038;critSec);<br /> #else<br /> pthread_mutex_unlock(&#038;myMutex);<br /> #endif<br /> <br /> if(verbose)printf("ofxThread: we are out -- mutext is now unlocked n");<br /> <br /> return true;<br /> }<br /> <br /> // <hr class="bbcode_rule" /> void ofxThread::stopThread(){<br /> if(threadRunning){<br /> #ifdef TARGET_WIN32<br /> CloseHandle(myThread);<br /> #else<br /> pthread_mutex_destroy(&#038;myMutex);<br /> pthread_detach(myThread);<br /> #endif<br /> if(verbose)printf("ofxThread: thread stoppedn");<br /> threadRunning = false;<br /> }else{<br /> if(verbose)printf("ofxThread: thread already stoppedn");<br /> }<br /> }<br />

[si_icarus]

Actually, I think I’ve actually identified the problem. I think it was the sleep() function in my thread loop. I understand why I need that if the loop is continuously running and doing heavy stuff, but do I need it in my case?

When I comment out the sleep line (As below), the behaviour looks a lot smoother and correct, maybe because as soon as I set the bHasRunThisFrame flag in the thread instance to false, the update function is called almost immediately (which is what I want), whereas with the sleep() function, maybe there was a delay of upto interval before the update kicked in?…. or something :P Is it bad to not have a sleep function in there?

// run the update function in another thread

void MSAThread::threadedFunction() {<br /> while(isThreadRunning()) {<br /> if( lock() ){<br /> if(bAutoLoop || !bHasRunThisFrame) {<br /> update();<br /> bHasRunThisFrame = true;<br /> unlock();<br /> // ofSleepMillis(interval);<br /> } else {<br /> unlock();<br /> }<br /> }<br /> }<br /> }<br />

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

void MSAThread::threadedFunction() {<br /> while(isThreadRunning()) {<br /> if( lock() ){<br /> if(bAutoLoop || !bHasRunThisFrame) {<br /> update();<br /> bHasRunThisFrame = true;<br /> unlock();<br /> //                    ofSleepMillis(interval);<br /> } else {<br /> unlock();<br /> }<br /> }<br /> }<br /> }<br />

[Author]

Posts