Thread design for sending data to multiple servers

Question

In the following code, checkServerExists function checks if the server exists in the vector. If it does, then the new message is directly pushed in the vector, otherwise a new thread is created and then the message is pushed in the vector.

I need to know whether it makes sense to write the code, as I have done.

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
#include <vector>
#include <string>
#include <iostream>

pthread_mutex_t demoMutex         = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t  conditionVariable = PTHREAD_COND_INITIALIZER;
unsigned int    condition         = 0;

struct serverInfo
{
    unsigned int                  serverId;
    pthread_t                     threadId;
    std :: vector <std :: string> queue;
};
std :: vector <serverInfo> serverInfoVector;

void * printHello (void* threadId)
{
    pthread_t *my_tid = (pthread_t *)threadId;

    pthread_mutex_lock (&demoMutex);

    while (condition == 0)
        pthread_cond_wait (&conditionVariable, &demoMutex);

    unsigned int i     = 0;
    char         found = false;

    if (serverInfoVector.size () > 0)
    {
        while ((i < serverInfoVector.size ()) && (found == false))
        {
            if (pthread_equal (pthread_self(), serverInfoVector [i].threadId))
            {
                found = true;
                break;
            }
            else
                i++;
        }
    }

    while ((found == true) && (!serverInfoVector [i].queue.empty ()))
    {
        std :: cout << "\nThread: " << pthread_self () << ", poped from queue: " << serverInfoVector [i].queue.front () << "\n";
        serverInfoVector [i].queue.pop_back ();
    }

    pthread_mutex_unlock (&demoMutex);
    pthread_exit (NULL);
}

void checkServerExists (unsigned int serverNumber, std :: string message)
{
    unsigned int i         = 0;
    char               found = false;

    pthread_mutex_lock (&demoMutex);

    if (serverInfoVector.size () > 0)
    {
        while ((i < serverInfoVector.size ()) && (found == false))
        {
            if (serverNumber == serverInfoVector [i].serverId)
            {
                found = true;
                break;
            }
            else
                i++;
        }
    }

    if (found == false)
    {
        // This server doesn't exist, so create a thread for it, create a queue for it, push the message in the corresponding queue.
        // Push the server number in the serverNumberArray.

        // Create a thread for it.
        pthread_t newThread;
        int       returnValue;
        if ((returnValue = pthread_create (&newThread,
                                    NULL,
                                    printHello,
                                        (void*) &newThread)) != 0)
        {
            printf ("\nerror: pthread_create failed with error number %d", returnValue);
        }
        printf ("\nIn checkServerExists ()`: thread id %ld\n", newThread);

        // Push the message in its queue.
        serverInfo obj;
        obj.serverId  = serverNumber;
        obj.threadId = newThread;
        obj.queue.push_back (message);
        serverInfoVector.push_back (obj);

        condition++;
        pthread_cond_signal (&conditionVariable);
        pthread_mutex_unlock (&demoMutex);
    }
    else
    {
        // This server exists, so lookup its thread and queue, push the message in the corresponding queue.
        printf ("\nIn else ()`: thread id %ld\n", serverInfoVector [i].threadId);
        serverInfoVector [i].queue.push_back (message);

        condition++;
        pthread_cond_signal (&conditionVariable);
        pthread_mutex_unlock (&demoMutex);
    }
}

int main ()
{
    checkServerExists (1, "anisha");
    checkServerExists (2, "kaul");
    checkServerExists (1, "sanjeev");
    checkServerExists (2, "sharma");

    for (unsigned int i = 0; i < serverInfoVector.size (); i++)
        pthread_join (serverInfoVector [i].threadId, NULL);

    return 0;
}

Answer 1

Any function that is a callback from a C library must use the C ABI.
If your code is C++ (as this is) then you MUST declare the callback function appropriately to make sure the compiler gets the correct ABI:

extern "C" void* printHello (void* threadId);

Correct usage of the signal:

while (condition == 0)
    pthread_cond_wait (&conditionVariable, &demoMutex);

But you never decrement the variable condition. Thus after the first thread is created it will always be true. Thus I would do this:

while (condition == 0)
{
    // Note I always enclose the statement in {} what happens if pthread_cond_wait()
    //      had been a macro? You can never trust all third party libraries so it
    //      is better to be safe than struggle to find it with debugger.
    pthread_cond_wait (&conditionVariable, &demoMutex);
}
// Once you know it is your decrement.
--condition;

Normally you use pthread_exit(); if you are killing the thread deep inside the code. If it is directly inside the callback function then you can just return NULL. Also you must guard against exceptions propagating out of the callback (if they escape the application will terminate (usally: Its actually undefined what happens but I have found it usally terminates the whole application (not just the thread))).

void * printHello (void* threadId)
{
    void*  result = NULL;
    try
    {
        result = // Your code here
    }
    catch(std::exception const& e) // Optional
    {
        // Log Exception
        std::cerr << "Exception: " << e.what() << "\n";
    }
    catch(...) // MUST have this one.
    {
        // Log Exception
        std::cerr << "Exception: Unknown(...)\n";
    }
    return result;
}

You use lock/unlock pairs for the mutex. This is not exception safe. So I would create an object that will do the lock in the constructor and unlock in the destructor then use this to lock your mutexs. This will make your code more exception safe.

class MutexLocker
{
    pthread_mutex_t&  mutex;
    MutextLocker(pthread_mutex_t& mutex)
        : mutex(mutex)
    {
        pthread_mutex_lock(&mutex);
    }
    ~MutexLocker()
    {
        pthread_mutex_unlock(&mutex);
    }
};

Your code may seem threeaded but it is still inherently serial. This is because each thread maintain the mutex for its whole life this it prevents new threads from being created. You should unlock the mutext as soon as you have finished initialization (or just use the lock when accessing global objects that may be mutated by the master thread).

Also on thread startup you spend a lot of time finding your queue. Why not get the master thread to pass a pointer to the threads queue (via the void* parameter). Note the value you pass (the pthread_t pointer is not guranteed to be correctly filled out when the thread starts (it is only guranteed correct after pthread_create() returns which may be after the thread has started so it is not a good object to pass:

What I would have done is:

struct serverInfo
{
    unsigned int                  serverId;
    pthread_t                     threadId;

    // Make the queue a pointer
    // So even when this object is copied the queue is unaffected.
    // May want to use a smart pointer or something (needs slightly more thought).
    std::vector<std::string>*     queue;
};

 std::auto_ptr<std::vector<std::string>> queue = new std::vector<std::string>();
 // Pass a pointer to the queue to the thread.
 // Now the thread does not need to know or care about serverInfoVector
 // Which is good because this is being mutated by other people.
 pthread_create(&newThread, NULL, printHello, queue.get()); // Add error code

 if (/* Everything OK */)
 {
     serverInfoVector.push_back(serverInfo(id, newThread, queue.release());
 }

Minor Comments:

Add a constructor to you serverInfo class.

struct serverInfo
{
    unsigned int                  serverId;
    pthread_t                     threadId;
    std :: vector <std :: string> queue;

    // Add this:
    serverInfo(unsigned int serverId, pthread_t threadId, std::string const& message)
        : serverId(serverId),
        , threadId(threadId)
    {
        queue.push_back(message);
    }
};

This will allow more readable code below.

AS a personal preference I allways have my types begin with an uppercase letter. This helps me distinguish types from data in an easy way that is not intrusive. This becomes more important when you start playing with templates.

Try and limit global variables.

std :: vector <serverInfo> serverInfoVector;

Easier to create these at the top level and then pass references to them around. If you have underlying functions that depend on global state it becomes really hard to test the code. In the long run it is best to pass everything as parameters then you can see exactly what the function depends on.

This is a horrible variable name. Try searching for all occurrences of the variable i to see where it is. I understand it short for index. Why not use index or my favorite loop.

unsigned int i     = 0;

This test seems unnesacery. If size is 0 then the while loop will just fail.

if (serverInfoVector.size () > 0)
{

I don't like the extra space you add around :: makes the code seem less readable (probably because I am not used to the style). But I think std::string is just fine.

This bit of code looks very familiar (is this not the same as above)

if (serverInfoVector.size () > 0)
{
    while ((i < serverInfoVector.size ()) && (found == false))
    {
        if (serverNumber == serverInfoVector [i].serverId)
        {
            found = true;
            break;
        }
        else
            i++;
    }
}

Refactor common code into a function. This way if there is a bug in your code then you only need to fix the bug in a single location (this prevents cut/paste errors).

Here is the point where the constructor will help:

    // Push the message in its queue.
    serverInfo obj;
    obj.serverId  = serverNumber;
    obj.threadId = newThread;
    obj.queue.push_back (message);
    serverInfoVector.push_back (obj);

This can now be written as:

    serverInfoVector.push_back(serverInfo(serverNumber, newThread, message));

Seems like you have common code on both sides of the branch.

{
    // STUFF

    condition++;
    pthread_cond_signal (&conditionVariable);
    pthread_mutex_unlock (&demoMutex);
}
else
{
    // STUFF

    condition++;
    pthread_cond_signal (&conditionVariable);
    pthread_mutex_unlock (&demoMutex);
}

This should be refactored out of the branches.

You have implemented all this as a function. That is called multiple times and uses some global data.

checkServerExists (1, "anisha");
checkServerExists (2, "kaul");
checkServerExists (1, "sanjeev");
checkServerExists (2, "sharma");

Personally I would implement this as an object (all the data internal to the object). Then call a method on the object to activate the new thread. This way nobody else can mess around with my data structures.

MultiThreadServer  server;
server.AddJobToThreadWithID (1, "anisha");
server.AddJobToThreadWithID (2, "kaul");
server.AddJobToThreadWithID (1, "sanjeev");
server.AddJobToThreadWithID (2, "sharma");

Unfortunately I am not sure this is a good idea.

for (unsigned int i = 0; i < serverInfoVector.size (); i++)
    pthread_join (serverInfoVector [i].threadId, NULL);

This is because the threads remove there object from serverInfoVector. There is a race condition here where you might access an element just as a thread removes it from the structure.

Note: main() is special. If you do not provide a return X then C++ will implicitly add a return 0 for you. Thus I always leave this out of main. Note: This ONLY applies to main you must return the correct value from all other functions.

return 0;

Answers to follow comments:

you said: "Normally you use pthread_exit(); if you are killing the thread deep inside the code." Why not join? Any special reasons for that?

• Return (in the callback function) is used when the thread wants to die of old age.
• pthread_exit() is used when a thread wants to commit suicide.
• pthread_join() is when a thread wants to wait for a different thread to die.

You said: "Also you must guard against exceptions propagating out of the callback" You mean if there are errors the application may terminate, so we should properly handle exceptions?

Yes. You must catch ALL exceptions before they leave the thread.

You said: "You use lock/unlock pairs for the mutex. This is not exception safe" Couldn't understand this. You mean that I have written openly the lock API two times in two functions? Why is that not exception safe? – Anisha Kaul 9 hours ago

This means there is a potential for your code to leave a mutex locked forever and your application will get stuck.

 void myFunc()
 {
     pthread_mutex_lock(mutex)

     // WORK

     pthread_mutex_unlock(mutex);
 }

The above code is not exception safe. If the WORK part throws an exception then mutex will not be unlocked and other threads will become stuck waiting for it. The simple way to solve this is to use RAII (and the class I wrote above).

 void myFunc()
 {
     MutexLocker  lock(mutex)

     // WORK
 }

Now the mutex will always be unlocked when this method exits (even if there are exceptions) as the destructor of lock will be called when the function exits and this will unlock the mutex correctly.

You said: "This is because each thread maintain the mutex for its whole life this it prevents new threads from being created. You should unlock the mutext as soon as you have finished initialization (or just use the lock when accessing global objects that may be mutated by the master thread)" Would you please edit my code to reflect this? AFAIK, we are supposed to keep the locks on till we complete the access to the shared data? So, in the printHello function I have kept the lock on while dealing with the queue. Is there some othe rway of doing the same?

Yes you must keep the locks on while accessing the shared data.

But you brake your own rules here:

for (unsigned int i = 0; i < serverInfoVector.size (); i++)
    pthread_join (serverInfoVector [i].threadId, NULL);

Which as I noted above can lead to race conditions and crashes. But if you do lock it then the the application may never quit. Also your code locks the code so long that only one thread may ever execute at a time (so there seems no point in using threads). Sorry that was over blowing it a bit (you release the lock while waiting on the conditional so that is not completely true).

This is an indication that too much of your data is shared. In my description above I explained how to remove the queue from the shared part of the structure (so each thread only sees the queue) and how serverInfoVector does not even need to be shared with the threads. Re-writting this is a major task and will require some work. Thus you only need a lock when adding/removing work from the queue associated with the thread.

Also, in the checkserver function I have used mutex while inserting in the queue. How should I improve this?

Just make sure it is exception safe.

Answer 2

This is what I would have done:

#include <string>
#include <map>
#include <list>
#include <iostream>

#include <pthread.h>

Mutex Locker

Generic Exception safe mutext locker/unlocker

class MutextLocker
{
    pthread_mutex_t&    mutex;
    public:
    MutextLocker(pthread_mutex_t& mutex): mutex(mutex)
    {
        pthread_mutex_lock(&mutex);
    }
    ~MutextLocker()
    {
        pthread_mutex_unlock(&mutex);
    }
};

QueInfo:

Data used and manipulate by each thread.
Both the main thread and the owning thread access this object so all access needs to controlled via locks.

class QueInfo
{
    public:
    QueInfo()
        : noMoreWork(false)
        , threadStarted(false)
    {
        if (pthread_mutex_init(&mutex, NULL) != 0)
        {   throw int(1);
        }
        if (pthread_cond_init(&cond, NULL) != 0)
        {
            pthread_mutex_destroy(&mutex);
            throw int(2);
        }
    }
    ~QueInfo()
    {
        pthread_cond_destroy(&cond);
        pthread_mutex_destroy(&mutex);
    }
    bool getWorkItem(std::string& item)
    {
        MutextLocker    lock(mutex);

        while ((queue.size() == 0) && (!noMoreWork))
        {
            pthread_cond_wait (&cond, &mutex);
        }
        bool result = false;
        if (queue.size() != 0)
        {
            item    = queue.front();
            queue.pop_front();
            result  = true;
        }
        return result;
    }
    void addMessage(std::string const& item)
    {
        MutextLocker    lock(mutex);
        queue.push_back(item);
        pthread_cond_signal(&cond);
    }
    void finishedAdding()
    {
        MutextLocker    lock(mutex);
        noMoreWork  = true;
        pthread_cond_signal(&cond);
    }

    // These two are accessed by ServerInfo but
    // never by the thread. This means we do not need
    // to lock on their use. But I am being lazy here 
    // leaving them as public members.
    pthread_t                       threadId;       // Being lazy here
    bool                            threadStarted;  // these two are for use by ServerInfo
    private:
    pthread_mutex_t                 mutex;
    pthread_cond_t                  cond;
    bool                            noMoreWork;
    std::list<std::string>          queue;
};

ServerInfo

Object that controls all threads.
Keeps a map of QueInfo objects. Each one represents a running thread.
Only the main thread sees this data so no locking needed.

class ServerInfo
{
    public:
    ~ServerInfo()
    {
        for (Cont::iterator loop = queue.begin(); loop != queue.end(); ++loop)
        {
            loop->second.finishedAdding();
            void*   result;
            pthread_join(loop->second.threadId, &result);
        }
    }

    void checkServerExists(unsigned int serverNumber, std::string const& message);

    private:
    typedef std::map<unsigned int, QueInfo>  Cont;
    Cont                            queue;
};

Callback function

Notice we pass a pointer to its own QueInfo object as a parameter.
Only this thread and the main thread can see this object. But all access is guarded.

void* printHello(void* data)
{
    QueInfo*   myQueue = reinterpret_cast<QueInfo*>(data);

    std::string workItem;
    while(myQueue->getWorkItem(workItem))
    {
        std::cout << "\nThread: " << pthread_self () << ", poped from queue: " << workItem << "\n";
    }
    return NULL;  // Return NULL on exit.
}

The main server adding threads.

void ServerInfo::checkServerExists(unsigned int serverNumber, std::string const& message)
{
    QueInfo& item = queue[serverNumber];  // If it does not exist it is inserted.
    item.addMessage(message);

    if (!item.threadStarted)
    {
        int       returnValue;
        if ((returnValue = pthread_create (&item.threadId,
                        NULL,
                        printHello,
                        reinterpret_cast<void*>(&item))) != 0)
        {
            std::cout << "\nerror: pthread_create failed with error number "<< returnValue;
            queue.erase(serverNumber);
        }
        else
        {
            item.threadStarted = true;
        }
        std::cout << "\nIn checkServerExists ()`: thread id " << item.threadId << "\n";
    }
    else
    {
        std::cout << "\nIn else ()`: thread id " << item.threadId << "\n";
    }
}

Main

Main is simplified as it does not even know about the threads.

int main ()
{
    ServerInfo  server;
    server.checkServerExists (1, "anisha");
    server.checkServerExists (2, "kaul");
    server.checkServerExists (1, "sanjeev");
    server.checkServerExists (2, "sharma");

    // Note ServerInfo destructor will wait for all the threads.
}