Subj : Re: why my threads do not wake up
To   : comp.programming.threads
From : Jingbo
Date : Wed Aug 24 2005 06:18 pm

OK, here is a program to demostrate the problem.
Compile with:  CC -mt ttt.C -lpthread
After I ran it, only one thread is on CPU. After several times of
running "pstack" command, all 8 threads is running.

#include <string>
#include <vector>
#include <deque>
#include <string.h>
#include <strings.h>
#include <errno.h>
#include <pthread.h>

class Engine
{
  public:

    Engine();
    // ~Engine() does not metter, ignore

    void blockIn(int blkId);
    bool blockOut(int blkId);
    void Thread_run( int threadId );

    // Thread specific data
    pthread_cond_t      updated;
    pthread_mutex_t     mutex;
    void notifyUpdate()
    {
        printf("(th %d) call pthread_cond_broadcast()\n",
(int)pthread_self());
        pthread_cond_broadcast( &updated );
    }
    void waitUpdate()
    {
        printf("(th %d) call pthread_cond_wait()\n",
(int)pthread_self());
        pthread_cond_wait( &updated, &mutex );
        printf("(th %d) wake up from pthread_cond_wait()\n",
(int)pthread_self());
    }

    struct ThreadInfo {
        Engine  *eng;
        pthread_t       th_id;
        int             id;
    };
    std::vector<ThreadInfo> threads;

    // task information. Each task is for one block.
    struct TaskInfo {
        enum {NEW, COMP, COMP_DONE}; // task status
        int status;
        int blockNumber;         // which block
    };
    std::deque<TaskInfo> tasks;
};

class AutoLock
{

    pthread_mutex_t *pmutex;

    bool  locked;

  public:

    AutoLock( pthread_mutex_t *m ) : pmutex(m) { lock(); }
    ~AutoLock() { if( locked ) unlock(); }

    void lock()
    {
        pthread_mutex_lock( pmutex );
        locked = true;
    }
    void unlock()
    {
        locked = false;
        pthread_mutex_unlock( pmutex );
    }
};

extern "C" void *thread_run_C( void *p )
{
    Engine::ThreadInfo *pa = (Engine::ThreadInfo*)p;
    printf("(th %d) thread %d created\n", (int)pthread_self(),
(int)pa->id);
    pa->eng->Thread_run( pa->id );
    return 0;
}

Engine::Engine()
{
    pthread_mutex_init( &mutex, 0 );
    pthread_cond_init( &updated, 0 );

    int numCompThread = 8;
    threads.resize(numCompThread);

    pthread_attr_t Attr;
    pthread_attr_init(&Attr);
    pthread_attr_setdetachstate(&Attr, PTHREAD_CREATE_JOINABLE);
    int i;
    for(i=0; i<threads.size(); i++)
    {
        threads[i].eng = this;
        threads[i].id = i;
        int res = pthread_create( &threads[i].th_id, &Attr,
thread_run_C, &threads[i]);
        printf("(th %d) create thread %d, res=%d, errno=%d(%s)\n",
(int)pthread_self(), i, res, errno, strerror(errno));
    }
}

void Engine::Thread_run( int threadId )
{
    // acquire lock
    AutoLock m(&mutex);

    // a loop of waiting job and processing job
    while( true )
    {
        // wait while new task is not available
        int blockNumber;
        int i;
        while( true )
        {
            for(i=0; i<tasks.size(); i++)
                if( tasks[i].status == TaskInfo::NEW )
                    break;
            if( i < tasks.size() )
            {
                blockNumber = tasks[i].blockNumber;
                tasks[i].status = TaskInfo::COMP;
                break;
            }
            waitUpdate();
        }
        TaskInfo task = tasks[i]; // local copy of tasks[i]
        ThreadInfo &th = threads[threadId];

        m.unlock();
        int sum=0;
        for(i=0; i<100000000; i++)
            sum += i*i*i*i % 10;
        printf("(th %d) thread %d finish block %d, sum=%d\n",
(int)pthread_self(), threadId, blockNumber, sum);
        bool OK = true;
        m.lock();


        // Update task status
        for(i=0; i<tasks.size(); i++)
            if( tasks[i].blockNumber == blockNumber )
                break;

        tasks[i].status = TaskInfo::COMP_DONE;
        notifyUpdate();
    }
}

// provide input data for one block to compute
void Engine::blockIn(int blkId)
{
    AutoLock m(&mutex);
    TaskInfo t;
    t.status = TaskInfo::NEW;
    t.blockNumber = blkId;
    tasks.push_back(t);
    notifyUpdate();
}

// get output of one block
bool Engine::blockOut(int blkId)
{
    AutoLock m(&mutex);


    while( true )
    {
        // search the block
        int i;
        for(i=0; i<tasks.size(); i++)
            if( tasks[i].blockNumber == blkId )
                break;

        if( tasks[i].status != TaskInfo::COMP_DONE )
            waitUpdate();
        else
        {
            tasks.erase(tasks.begin()+i);
            break;
        }
    }
    return true;
}

main()
{
    Engine *eng = new Engine;
    int i;

    // add tasks
    for(i=0; i<100; i++)
        eng->blockIn(i);

    // wait for tasks done
    for(i=0; i<100; i++)
        eng->blockOut(i);
}

.