try to fix some eThread races

diff --git a/lib/base/thread.cpp b/lib/base/thread.cpp
index b64ba1365bfb2376e69e36ce1da73ffb1614c4bc..ceb50bfe350c3dbab479f9150f066ebdfff3f024 100644 (file)
--- a/lib/base/thread.cpp
+++ b/lib/base/thread.cpp
@@ -7,17 +7,20 @@
 void eThread::thread_completed(void *ptr)
 {
        eThread *p = (eThread*) ptr;
 void eThread::thread_completed(void *ptr)
 {
        eThread *p = (eThread*) ptr;

-       eDebug("thread has completed..");
-       p->alive=0;
-       p->thread_finished();
+       p->m_alive = 0;
+
+               /* recover state */
+       if (!p->m_state.value())
+       {
+               p->m_state.up();
+               assert(p->m_state.value() == 1);
+       }

 }
 
 void *eThread::wrapper(void *ptr)
 {
        eThread *p = (eThread*)ptr;
 }
 
 void *eThread::wrapper(void *ptr)
 {
        eThread *p = (eThread*)ptr;

-       p->before_set_thread_alive();
-       p->alive=1;
-       pthread_cleanup_push( thread_completed, (void*)p );
+       pthread_cleanup_push(thread_completed, (void*)p);

        p->thread();
        pthread_exit(0);
        pthread_cleanup_pop(1);
        p->thread();
        pthread_exit(0);
        pthread_cleanup_pop(1);


@@ -25,65 +28,95 @@ void *eThread::wrapper(void *ptr)
 }
 
 eThread::eThread()
 }
 
 eThread::eThread()

-       :the_thread(0), alive(0)
+       : the_thread(0), m_alive(0)

 {
 }
 
 {
 }
 

-void eThread::run( int prio, int policy )
+int eThread::runAsync(int prio, int policy)

 {
 {

-       if (alive)
-       {
-               eDebug("thread already running !!");
-               return;
-       }
+       eDebug("before: %d", m_state.value());
+               /* the thread might already run. */
+       if (sync())
+               return -1;
+       
+       eDebug("after: %d", m_state.value());
+       assert(m_state.value() == 1); /* sync postconditions */
+       assert(!m_alive);
+       m_state.down();
+       
+       m_alive = 1;
+
+               /* start thread. */

        pthread_attr_t attr;
        pthread_attr_init(&attr);
        pthread_attr_t attr;
        pthread_attr_init(&attr);

-       if (prio||policy)
+       
+       if (prio || policy)

        {
                struct sched_param p;
                p.__sched_priority=prio;
        {
                struct sched_param p;
                p.__sched_priority=prio;

-               pthread_attr_setschedpolicy(&attr, policy );
+               pthread_attr_setschedpolicy(&attr, policy);

                pthread_attr_setschedparam(&attr, &p);
        }
                pthread_attr_setschedparam(&attr, &p);
        }

-       if ( pthread_create(&the_thread, &attr, wrapper, this) )
+       
+       if (pthread_create(&the_thread, &attr, wrapper, this))

        {
        {

+               pthread_attr_destroy(&attr);
+               m_alive = 0;

                eDebug("couldn't create new thread");
                eDebug("couldn't create new thread");

-               return;
+               return -1;

        }
        }

+       

        pthread_attr_destroy(&attr);
        pthread_attr_destroy(&attr);

-       usleep(1000);
-       int timeout=50;
-       while(!alive && timeout--)
-       {
-//             eDebug("waiting for thread start...");
-               usleep(1000*10);
-       }
-       if ( !timeout )
-               eDebug("thread couldn't be started !!!");
+       return 0;

 }                     
 
 }                     
 

+int eThread::run(int prio, int policy)
+{
+       if (runAsync(prio, policy))
+               return -1;
+       sync();
+       return 0;
+}
+

 eThread::~eThread()
 {
        kill();
 }
 
 eThread::~eThread()
 {
        kill();
 }
 

+int eThread::sync(void)
+{
+       int res;
+       m_state.down(); /* this might block */
+       res = m_alive;
+       assert(m_state.value() == 0);
+       m_state.up();
+       return res; /* 0: thread is guaranteed not to run. 1: state unknown. */
+}
+

 void eThread::sendSignal(int sig)
 {
 void eThread::sendSignal(int sig)
 {

-       if ( alive )
-               pthread_kill( the_thread, sig );
+       if (m_alive)
+               pthread_kill(the_thread, sig);

        else
                eDebug("send signal to non running thread");
 }
 
 void eThread::kill(bool sendcancel)
 {
        else
                eDebug("send signal to non running thread");
 }
 
 void eThread::kill(bool sendcancel)
 {

-       if (!the_thread)
+       if (!the_thread) /* already joined */

                return;
                return;

-       if ( alive && sendcancel )
+
+       if (sync() && sendcancel)

        {
                eDebug("send cancel to thread");
                pthread_cancel(the_thread);
        }
        eDebug("thread joined %d", pthread_join(the_thread, 0));
        {
                eDebug("send cancel to thread");
                pthread_cancel(the_thread);
        }
        eDebug("thread joined %d", pthread_join(the_thread, 0));

-       the_thread=0;
+       the_thread = 0;
+}
+
+void eThread::hasStarted()
+{
+       assert(!m_state.value());
+       m_state.up();

 }
 }

diff --git a/lib/base/thread.h b/lib/base/thread.h
index 94cdd47e66c3721c43b27ec0146b8cbad6fff231..dad80424b526571efc72ecace1848fdc769457a8 100644 (file)
--- a/lib/base/thread.h
+++ b/lib/base/thread.h
@@ -3,26 +3,63 @@
 
 #include <pthread.h>
 #include <signal.h>
 
 #include <pthread.h>
 #include <signal.h>

+#include <lib/base/elock.h>
+
+/* the following states are possible:
+ 1 thread has not yet started
+ 2 thread has started, but not completed initialization (hadStarted not yet called)
+ 3 thread is running
+ 4 thread has finished, but not yet joined
+ 5 thread has joined (same as state 1)
+ 
+ sync() will return:
+       0 (="not alive") for 1, 4, 5
+       1 for 3, 4
+       
+       it will wait when state is 2. It can't differentiate between
+       state 3 and 4, because a thread could always finish.
+       
+       all other state transitions (1 -> 2, 4 -> 5) must be activately
+       changed by either calling run() or kill().
+ */

 
 class eThread
 {
 
 class eThread
 {

-       pthread_t the_thread;
-       static void *wrapper(void *ptr);
-       int alive;
-       static void thread_completed(void *p);

 public:
 public:

-       bool thread_running() { return alive; }

        eThread();
        virtual ~eThread();
 
        eThread();
        virtual ~eThread();
 

-       void run(int prio=0, int policy=0);
+               /* runAsync starts the thread.
+                  it assumes that the thread is not running,
+                  i.e. sync() *must* return 0.
+                  it will not wait for anything. */
+       int runAsync(int prio=0, int policy=0);
+
+               /* run will wait until the thread has either
+                  passed it's initialization, or it has
+                  died again. */
+       int run(int prio=0, int policy=0);

 
        virtual void thread()=0;
 
        virtual void thread()=0;

-       virtual void before_set_thread_alive() { }
-       virtual void thread_finished() { }
+       
+               /* waits until thread is in "run" state */
+               /* result: 0 - thread is not alive
+                          1 - thread state unknown */
+       int sync();

        void sendSignal(int sig);
 
        void sendSignal(int sig);
 

+               /* join the thread, i.e. busywait until thread has finnished. */

        void kill(bool sendcancel=false);
        void kill(bool sendcancel=false);

+private:
+       pthread_t the_thread;
+
+       static void *wrapper(void *ptr);
+       int m_alive;
+       static void thread_completed(void *p);
+
+       eSemaphore m_state;
+protected:
+       void hasStarted();

 };
 
 #endif
 };
 
 #endif