thread safe writing to a stream in c++ by multiple thread

Question

I am trying to create multiple threads and try to find out the scheduling order and when they finish. Every time a thread got CPU it perform some computation , then it waits for some x time( 1000ns) as a busy wait and then go for sleep. Every time one thread got CPU it also prints the time so that I can later find the time when a particular thread got CPU and the order.

To make thread-safe , I am using mutex, but still it is not giving correct result. Where I am making mistakes ? Any help will be highly appreciated.

I am using g++ under linux.

Note: I can't use file as in that case file open/close incurs overhead.

Here is my program output

Output :

$ sort -t , -k4,4n -k5,5n -k6,6n >a.txt

$ head a.txt 
thread,   thread_no, iteration, time_min, time_sec, time_microsec    
foo,
foo,14,987
foo,32
foo,32,985,
foo,57,970
foo,71,933,
foo,71,933,
foo,71,933,
foo,71,933,

$ tail a.txt  
thread,   thread_no, iteration, time_min, time_sec, time_microsec
foo,98,991,40,05,935379 
foo,98,992,40,05,935442 
foo,98,993,40,05,935506 

foo,98,994,40,05,935569 
foo,98,995,40,05,935633 
foo,98,996,40,05,935697 
foo,98,997,40,05,935760 
foo,98,998,40,05,935824 
foo,98,999,40,05,937914 
foo,98,1000,40,05,937994

As per comment of Bart van Nierop after adding out.flush() before releasing mutex,

here is the result.$ head a.txt

foo,
foo,
foo,48,991,
foo,65,
foo,95,
foo,97
foo,10,1,15,59,288329 
foo,10,1,15,59,288329 
foo,10,1,15,59,288329

Here is my program

#include <pthread.h>
#include <stdio.h>
#include <iostream>
#include <cstdlib>
#include <fstream>
#include <sys/time.h>
#include <time.h>       /* time_t, struct tm, time, localtime */
#include <mutex>          // std::mutex
#include <sstream>

#define BILLION  1000000000L
#define NUM_THREADS 100

std::mutex mtx;           // mutex for critical section

using namespace std;


std::string now_str()
{
    struct timeval tv;
    struct timezone tz;
    struct tm *tm;
    gettimeofday(&tv, &tz);
    tm=localtime(&tv.tv_sec);
    char buf[40];
    sprintf(buf,"%02d,%02d,%ld ", tm->tm_min,tm->tm_sec, tv.tv_usec); 
    return buf;
}


std::ostringstream out;  
/* This is our thread function.  It is like main(), but for a thread*/
void *threadFunc(void *arg)
{ 
      int s,j;
      pthread_attr_t gattr;

      // Assigning SCHED_RR policy 

      j = SCHED_RR;
      s = pthread_attr_setschedpolicy(&gattr, j);
      if (s != 0) 
        printf( "pthread_attr_setschedpolicy");

      s = pthread_attr_getschedpolicy(&gattr, &j);
      if (s != 0)
        printf( "pthread_attr_getschedpolicy");


    struct timespec start, stop;
    double accum;

    char *str;
    int i = 0,k=0;

    str=(char*)arg;

    while(i < 1000)
    {

        ++i;

        // do something here

        mtx.lock(); 
        out << "\nfoo," <<str<<","<<i<<"," <<now_str(); // note the timing of thread
        mtx.unlock();  


    if( clock_gettime( CLOCK_REALTIME, &start) == -1 ) 
    {
          perror( "clock gettime" );
          exit( EXIT_FAILURE );
    }

    // busy wait for 1000ns
    do

    { 
        if( clock_gettime( CLOCK_REALTIME, &stop) == -1 ) 
        {
              perror( "clock gettime" );
              exit( EXIT_FAILURE );
            }

        accum = ( stop.tv_sec - start.tv_sec )* BILLION + ( stop.tv_nsec - start.tv_nsec ) ;

    }while(accum < 1000);

        // block the thread, to allow other thread to run
        usleep(1);
    }
    std::cout<<out.str();

    return NULL;
}

int main(void)
{

    pthread_t pth[NUM_THREADS];  
    int i = 0;
pthread_create(&pth[0],NULL, threadFunc,  (void *) "0"); 
pthread_create(&pth[1],NULL, threadFunc,  (void *) "1"); 
.
.
.

pthread_create(&pth[98],NULL, threadFunc,  (void *) "98"); 
pthread_create(&pth[99],NULL, threadFunc,  (void *) "99"); 

    for(int k=0;k<NUM_THREADS;k++)
    pthread_join(pth[k],NULL);

    return 0;
}

Answer 1

A little searching on google and it looks like ostringstream is not thread safe.

To get what you want you could roll your own logging.

One method would be to write to the end of a large buffer:

uint64_t offset = 0;
char buffer[ 200000 ];

Write to it using a mutex:

mtx.lock(); 
offset += sprintf( buffer + offset, "\nfoo,%s,%d,%s", str, i, now_str().c_str() );
mtx.unlock();  

Answer 2

You hold the mutex while populating std::ostringstream out, but not while streaming it to std::cout. So, one thread holding the mutex can be mutating out at the same time another without the mutex is reading from it.

The simplest fix would be protect your std::cout<<out.str(); line with the same mutex.

If you care about performance, you can make the global std::ostringstream out a function-local variable (I don't know why you have so many globals in the first place), so that each thread has its own copy. Then, while you should still protect the global cout, you don't need the mutex around formatting your own stream.