/ray/src/lib/threads/thread.cc

Go to the documentation of this file.

/*
 * lib/threads/thread.cc
 * 
 * Implementing thread class for multi-threaded applications. 
 * 
 * Copyright (c) 2004 by Wolfgang Wieser ] wwieser (a) gmx <*> de [ 
 * 
 * This file may be distributed and/or modified under the terms of the 
 * GNU General Public License version 2 as published by the Free Software 
 * Foundation. (See COPYING.GPL for details.)
 * 
 * This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
 * WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
 * 
 */

#include "thread.h"


// Initialize static data: 
ThreadKernel *ThreadKernel::kernel=NULL;


int ThreadKernel::_NRunningThreads()
{
    mutex.lock();
    int n=nrunning;
    mutex.unlock();
    
    return(n);
}


void *ThreadKernel::_ThreadFuncWrapper(void *ptr)
{
//fprintf(stderr,"WRAP %p...\n",ptr);
    // This function is executed by the newly created thread in that new 
    // thread context. 
    ExecutionThread *ethread=(ExecutionThread*)ptr;
    
    // Add to the list of threads. 
    TNode *tn=new TNode();
    tn->thread=g_thread_self();
    tn->execthread=ethread;
    
    mutex.lock();
    threadlist.append(tn);
    ++nrunning;
    mutex.unlock();
    
    // Well, this should be thread-safe :)
    g_private_set(phook,tn);
    
    // Attach information to the ExecutionThread: 
    bool rv=g_atomic_pointer_compare_and_exchange(
        (gpointer*)&ethread->handle,NULL,tn);
    CritAssert(rv);  // critical for thread safety
    
    // Actually "run" the thread function: 
    ethread->run();
    
//fprintf(stderr,"RUN %p done.\n",ethread);
    return(NULL);
}

void *ThreadKernel::ThreadFuncWrapper(void *ptr)
{
    // (Static version.)
    return(ThreadKernel::kernel->_ThreadFuncWrapper(ptr));
}


void ThreadKernel::_PhookNotifier(gpointer ptr)
{
//fprintf(stderr,"_PhookNotifier(%p %s)",ptr,ptr==kernel ? "kernel" : "");
    
    // This means that any thread has exited. 
    // We are executed in this thread's context. 
    // Let's see which one. 
    if(ptr==this)
    {
        // Oops, the kernel exited? This should not happen. 
        CritAssert(0);
        return;
    }
    
    TNode *tn=(TNode*)ptr;
    CritAssert(tn->thread==g_thread_self());
    
    mutex.lock();
    
    // We must tell the ExecutionThread class about that, 
    // if it still exists. 
    if(tn->execthread)
    {
        bool rv=g_atomic_pointer_compare_and_exchange(
            (gpointer*)&tn->execthread->handle,tn,NULL);
        // If this assert fails, someone (probably ExecutionThread) must 
        // have modified the handle pointer which he is not allowed to. 
        CritAssert(rv);  // critical for thread safety
        
        // This is all. The ExecutionThread now knows that the thread exited. 
    }
    
    // The associated thread is no longer valid. 
    // (Well, currently it still is but only for a few CPU cycles :)
    tn->thread=NULL;
    
    // And remove the thread from the list: 
    threadlist.dequeue(tn);
    --nrunning;
    
    mutex.unlock();
    
    // Operator delete should be thread-save by itself :)
    delete tn;
}

void ThreadKernel::PhookNotifier(gpointer ptr)
{
    // (Static version.)
    ThreadKernel::kernel->_PhookNotifier(ptr);
}


void ThreadKernel::_unregister(ExecutionThread *et)
{
    // In case the thread kernel was already destroyed, there is nothing 
    // to do. 
    if(!g_atomic_pointer_get((gpointer*)&kernel))  return;
    
    // It is important that all this is protected by a mutex. 
    mutex.lock();
    
    if(this && et->handle)
    {
        // Clear handle in the execution thread. 
        TNode *tn=et->handle;
        et->handle=NULL;
        
        // Clear attached ExecutionThread since it does no longer exist. 
        tn->execthread=NULL;
    }
    
    mutex.unlock();
}


void ThreadKernel::init()
{
    // The TheadKernel will itself set the static data. 
    new ThreadKernel();
}

void ThreadKernel::cleanup()
{
    if(ThreadKernel::kernel)
    {  delete ThreadKernel::kernel;  }  // Destroctor will set to NULL. 
}


ThreadKernel::ThreadKernel() : 
    mutex(),
    threadlist(),
    nrunning(0)
{
    // Initialize the thread system. 
    g_thread_init(NULL);
    CritAssert(g_thread_supported());  // critical
    
    // Play with the mutex :)
    mutex.lock();
    
    // There may be a limited number of GPrivate but this should not fail 
    // since we've just initialized the thread system and hence there 
    // should not have ben much use of GPrivate yet. 
    phook=g_private_new(&PhookNotifier);
    CritAssert(phook);  // critical
    
    g_private_set(phook,this);
    
    Assert(!ThreadKernel::kernel);
    ThreadKernel::kernel=this;
    
    mutex.unlock();
}

ThreadKernel::~ThreadKernel()
{
    mutex.lock();
    
    // Free all elements of the thread list. There should be none. 
    // If this assert fails, there are still threads running. 
    // You may not clean up the thread kernel in such a case. 
    CritAssert(threadlist.IsEmpty());
    while(!threadlist.IsEmpty())
    {
        delete threadlist.PopFirst();
        --nrunning;
    }
    
    // GPrivate cannot be destroyed. 
    g_private_set(phook,NULL);
    phook=NULL;
    
    Assert(ThreadKernel::kernel==this);
    ThreadKernel::kernel=NULL;
    
    mutex.unlock();
}


//------------------------------------------------------------------------------

int ExecutionThread::start()
{
    if(g_atomic_pointer_get(&handle))  return(-2);
    
    // Only non-joinable threads are supported. We get to know when 
    // they exit in any case and can do better than simple joining 
    // using a condition or a semaphore. 
    //GError *err=NULL;
    GThread *thread=g_thread_create(
        &ThreadKernel::ThreadFuncWrapper,this,
        /*joinable=*/0,/*&err*/NULL);
    if(!thread)
    {
        // g_error_free(err);
        return(-3);
    }
    
    for(;;)
    {
        if(g_atomic_pointer_get(&handle)) break;
        g_thread_yield();
    }
    
    return(0);
}


void ExecutionThread::exit()
{
    g_thread_exit(NULL);
}


void ExecutionThread::yield()
{
    g_thread_yield();
}


void ExecutionThread::run()
{
    // Hmm... default is to do nothing :)
    //fprintf(stderr,"You'd better do something useful with the thread!\n");
    return;
}



ExecutionThread::ExecutionThread()
{
    handle=NULL;
}

ExecutionThread::~ExecutionThread()
{
    // Unregister at the ThreadKernel. 
    ThreadKernel::unregister(this);
}


#if 0
#include <stdio.h>  /* test program */
#include <sched.h>

// Test program. 
//  g++ lib_threads.a  -o test `pkg-config --libs gthread-2.0`

class MyThreadA : public ExecutionThread
{
    void run()
    {
        for(int i=0; i<100; i++)
        {
            fprintf(stderr,"runA\t");
            yield();
        }
    }
};

class MyThreadB : public ExecutionThread
{
    void run()
    {
        for(int i=0; i<100; i++)
        {
            fprintf(stderr,"runB\t");
            yield();
            if(i==40)  exit();
        }
    }
};


int main()
{
    ThreadKernel::init();
    
    MyThreadA A;
    MyThreadB B;
    fprintf(stderr,"A=%p, B=%p\n",&A,&B);
    
    A.start();
    B.start();
    
    for(int i=0; i<50; i++)
    {
        sched_yield();
        A.start();
        if(!B.running())
        {
            fprintf(stderr,"B-OUT!");
            B.start();
        }
    }
    
    fprintf(stderr,"Press any key..."); getchar();
    
    ThreadKernel::cleanup();
    
    fprintf(stderr,"Exiting\n");
    return(0);
}

#endif

---