Returning values from pthread asynchronously at regular intervals

Question

Hi,

The main() function creates a thread that is supposed to live until the user wishes to exit the program. The thread needs to return values to the main functions at periodic intervals. I tried doing something like this, but hasn't worked well -

std::queue<std::string> q;

void start_thread(int num)
{
 std::string str;
 //Do some processing
 q.push(str);
}

int main()
{
 //Thread initialization
 int i;
 //Start thread
 pthread_create(&m_thread,NULL,start_thread,static_cast<void *>i);

 while(true)
 {
  if(q.front())
  {
      std::cout<<q.front();
      return 0;
  }
 }

 //Destroy thread.....  
 return 0;
}

Any suggestions?

Answer 1

try locking a mutex before calling push() / front() on the queue.

Answer 2

• It is not safe to read and write from STL containers concurrently. You need a lock to synchronize access (see pthread_mutex_t).
• Your thread pushes a single value into the queue. You seem to be expecting periodic values, so you'll want to modify start_thread to include a loop that calls queue.push.
• The return 0; in the consumer loop will exit main() when it finds a value in the queue. You'll always read a single value and exit your program. You should remove that return.
• Using if (q.front()) is not the way to test if your queue has values (front assumes at least one element exists). Try if (!q.empty()).
• Your while(true) loop is gonna spin your processor somethin' nasty. You should look at condition variables to wait for values in the queue in a nice manner.

Answer 3

Here is a working example of what it looks like you were trying to accomplish:

#include <iostream>
#include <queue>
#include <vector>
#include <semaphore.h>
#include <pthread.h>

struct ThreadData
{
    sem_t sem;
    pthread_mutex_t mut;
    std::queue<std::string> q;
};

void *start_thread(void *num)
{
    ThreadData *td = reinterpret_cast<ThreadData *>(num);
    std::vector<std::string> v;
    std::vector<std::string>::iterator i;

    // create some data
    v.push_back("one");
    v.push_back("two");
    v.push_back("three");
    v.push_back("four");

    i = v.begin();

    // pump strings out until no more data
    while (i != v.end())
    {
        // lock the resource and put string in the queue
        pthread_mutex_lock(&td->mut);
        td->q.push(*i);
        pthread_mutex_unlock(&td->mut);

        // signal activity
        sem_post(&td->sem);
        sleep(1);

        ++i;
    }

    // signal activity
    sem_post(&td->sem);
}

int main()
{
    bool exitFlag = false;
    pthread_t m_thread;
    ThreadData td;

    // initialize semaphore to empty
    sem_init(&td.sem, 0, 0);

    // initialize mutex
    pthread_mutex_init(&td.mut, NULL);

    //Start thread
    if (pthread_create(&m_thread, NULL, start_thread, static_cast<void *>(&td)) != 0)
    {
        exitFlag = true;
    }

    while (!exitFlag)
    {
        if (sem_wait(&td.sem) == 0)
        {
            pthread_mutex_lock(&td.mut);

            if (td.q.empty())
            {
                exitFlag = true;
            }
            else
            {
                std::cout << td.q.front() << std::endl;
                td.q.pop();
            }

            pthread_mutex_unlock(&td.mut);
        }
        else
        {
            // something bad happened
            exitFlag = true;
        }
    }

    return 0;
}