How to force threads to work in strict order?

Question

I'm writing a multi-thread console application using WinAPI's critical sections as the synchronizing mechanism. I need to create 5 threads, every thread has its own string to be displayed by the thread. Threads should output their strings sequentially, one by one.

So, I have thread function:

void thread_routine(void* data)
{
    std::string* st = (std::string*)data;
    while(1)
    {
        /* enter critical section */
        std::cout << st; // not exactly that way, but I hope that you understood what I've meant.
        Sleep(100);
        /* leave critical section */
    }
}

And in my main()

for(int i = 0; i < 10; ++i)
    _beginthread(thread_routine, 0, strings[i]);

Well I've expected to see something like [o1]:

1) First thread
2) Second thread
3) Third thread
4) Fourth thread
5) Fifth thread
... and so on

But instead of this output I saw something like [o2]:

1) First thread
1) First thread
3) Third thread
2) Second thread
5) Fifth thread
... and so on

I clearly understand what happens: the order in which the threads passes through the critical section is unknown, so the section will be captured by a random one, but I need to synchronize my threads to get an output like [o1]. So what do I have to do? Are there any patterns?

Answer 1

Critical sections are not the way to solve this problem. A critical section is nothing more than a mutual exclusion device. It's intended to ensure that only one thread can be executing a particular piece of code at a given time. It's not intended to be used for sequencing, nor is it particularly good for that task. The problem is that you can't wait on a critical section without acquiring it.

Consider a case in which Thread B has to wait for Thread A to finish before continuing with its work. If you use a critical section (call it cs1), then you have to ensure that Thread A acquires it before Thread B tries to acquire it. That means you have to make 100% sure that Thread A begins execution and acquires the critical section before Thread B can potentially acquire the critical section.

Event Objects, on the other hand, allow you to wait for an event or continue if the event has already occurred. So Thread B can wait on the event before Thread A even starts. And it will wait until Thread A (or somebody) sets the event. The problem you describe (thread B needing to wait for some event to happen) is exactly the type of thing that event objects were designed to solve.

Given 3 threads that process in parallel for some time, but at some point need to wait for some other event to happen, you'd write code to create and wait on the events. Briefly:

HANDLE event1_done = CreateEvent(...);
HANDLE event2_done = CreateEvent(...);
HANDLE event3_done = CreateEvent(...);

All events are manual reset, and their initial state is not signaled.

The main thread starts the three events. Then it waits on the third one to be finished:

WaitForSingleObject(event3_done, INFINITE);

The individual threads do their processing and wait on their respective events. Thread 1, of course, doesn't wait. It just does this:

// thread 1
// do processing
// then signal that it's finished
SetEvent(event1_done);

Thread 2 does its processing, waits on event1_done, and then sets event2_done:

// thread 2
// do processing
// wait for the first thread to complete
WaitForSingleObject(event1_done, INFINITE);
// do whatever
// and then signal that it's done
SetEvent(event2_done);

Thread 3 is just like Thread 2; only the event names have changed.

The difference between this method and using critical sections is that multiple threads could be waiting on the same event object. When that event is set, all of the threads waiting on that object are released. If you want only one of them released, then you'd use an auto reset event.

Also note that if you want to re-use those events, you'll need to reset them (call ResetEvent). When you do that is up to you.

Answer 2

You need to write your own scheduler. Just another thread which wakes up your threads in the specified order. In that case you have to pass to your threads more complex data including some waitable object (i.e. Semaphore). I'm not experienced in WinAPI, it's just an idea:

void scheduler_thread(void* data) {
  scheduler_data* threads = (scheduler_data*)data;
  int index = 0;
  while (true) {
    notify_waitable_object(threads->waitable_object[index]);
    sleep(timeout);
    index = (index + 1) % threads->thread_count;
  }
}

void thread_procedure(void* data) {
  some_thread_data* p = (some_thread_data*)data;
  while(true)
  {
    wait_for_notify(p->wait_object);
    std::cout << p->st;
  }  
}

Answer 3

1) You should not put I/O in separate threads. Do the output in your main thread after the threads have done their processing.

2) If you need the threads to execute sequentially, you don't need threads. Threads are useful for operating in parallel. If you need to do something sequentially, you just do it in a single thread (your main thread).

3) If you really want to do it anyway, you would want to create a series of critical sections where each thread is waiting on a different one. That is,

• Thread 2 is waiting for Thread 1 to clear Critical Section 1
• Thread 3 is waiting for Thread 2 to clear Critical Section 2
• Thread 4 is waiting for Thread 3 to clear Critical Section 3 etc.

Also, your data object (that is passed into each thread) will need to contain not only the std::string you want to print out, but also the handle to the critical sections it has to wait for, and clear (that is, you need a structure that at a minimum has 2 critical section handles and a string). An incomplete example to demonstrate the idea is below:

struct ThreadData
{
    std::string st;
    CRITICAL_SECTION* prevCS;
    CRITICAL_SECTION* nextCS;
};

class Mutex
{
public:
    Mutex(CRITICAL_SECTION* cs) : _cs(cs)
    {
        EnterCriticalSection(_cs);
    }

    ~Mutex()
    {
        LeaveCriticalSection(_cs);
    }
private:
    CRITICAL_SECTION* _cs;
};

void thread_routine(void* data)
{
    ThreadData* d = (ThreadData*)data;
    std::unique_ptr<Mutex> current; // this is the one my predecessor must clear
    std::unique_ptr<Mutex> next; // this is the one I have to clear
    if (d->nextCS)
    {
        next = std::make_unique<Mutex>(d->nextCS);
    }

    if (d->prevCS)
    {
        current = std::make_unique<Mutex>(d->prevCS);
    }

    std::cout << d->st << std::endl;
}

int main()
{
    CRITICAL_SECTION cs1;
    CRITICAL_SECTION cs2;

    InitializeCriticalSection(&cs1);  
    InitializeCriticalSection(&cs2);

    ThreadData td1;
    ThreadData td2;
    ThreadData td3;

    td1.nextCS = &cs1;
    td1.prevCS = nullptr;
    td1.st = "Thread 1";

    td2.nextCS = &cs2;
    td2.prevCS = &cs1;
    td2.st = "Thread 2";

    td3.nextCS = nullptr;
    td3.prevCS = &cs2;
    td3.st = "Thread 3";     

    _beginthread(thread_routine, 0, &td1);
    _beginthread(thread_routine, 0, &td2);
    _beginthread(thread_routine, 0, &td3);

   // NOTE:  you also need to add an event handle to wait for in the main thread.  It would go here   

    return 0;
}

Alternate solution

(which may or may not meet the needs for your assignment):

You could only ever have 1 thread running at a time and let your main thread wait for a single thread to finish before letting the next one start. That is, you create all of your threads and leave then in a suspended state. Then main kicks off your first thread and waits for it to complete before kicking off the second, and repeats for the number of threads you have. Depending on how the requirements were written, this would probably be a clever way to skirt the silliness of a professor's assignment.

Answer 4

You probably have an design problem and should fix that one!

But if you really one to sync threads, here is one approach. Probably will get down voted for this as this is extremely inefficient and will deadlock if any of the threads skips the important part (like by try-catch), but still an approach:

#include <thread>
#include <atomic>
#include <cstdlib>
#include <iostream>
#include <vector>

void myFunc(int idx,std::atomic<int> * counter){

    std::this_thread::sleep_for(std::chrono::milliseconds(std::rand()%100));
    // Don't know about memory_order stuff
    // If you want to use this approach, you should read about it.
    while(counter->load()!=idx){
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    }
    std::cout << idx << " thread here\n";
    counter->store(idx+1);
}

int main(){
    std::srand(std::time(0));
    std::atomic<int> counter(0);
    std::vector<std::thread> thread_vector;
    for(int i=0;i<10;i++)
        thread_vector.push_back(std::thread(myFunc,i,&counter));

    for(int i=0;i<10;i++)
        thread_vector[i].join();
}

Answer 5

I have created a simple producer, consumer scenario, where I have a function "producer" producing data and N (N is Configurable) consumers, consuming the data sequentially(in strict order) i.e. 1st thread should consume data before 2nd thread. and 2nd thread before 3rd and so on. When each consumer consumed the data it will again start from the first thread.

To run it in sequence I am using condition variables and "notify_one()" function.

#include <iostream>
#include <queue>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>

#define NUMBER_OF_THREADS 5
#define MAX_QUEUE_ELEM 500

std::mutex m;
std::condition_variable producerMutex;
std::condition_variable consumerMutex[NUMBER_OF_THREADS];
std::queue<int> Q;
static int elemCounter = 1;
static int threadCounter = 1;

void producer()
{
    while (true)
    {
        // lock thread
        std::unique_lock<std::mutex> lock(m);

        while (Q.size() == MAX_QUEUE_ELEM)
        {
            std::cout << "Producer waiting" << std::endl;
            producerMutex.wait(lock);
        }

        Q.push(elemCounter++);

        // unlock thread
        lock.unlock();

        // notify next waiting consumer thread
        consumerMutex[threadCounter].notify_one();
    }
}

void consumer()
{
    while (true)
    {
        //lock thread
        std::unique_lock<std::mutex> lock(m);

        while (Q.empty())
        {
            std::cout << "Consumer waiting : %d "<< threadCounter << std::endl;
            consumerMutex[threadCounter].wait(lock);
        }

        int val = Q.front();
        Q.pop();

        // Printing in lock to print in sequnce
        std::cout << "val: %d " << val << " , thread number: %d " << threadCounter << std::endl;

        // unloack thread
        lock.unlock();

        // Notify to next waiting thread in sequnce
        if (threadCounter == NUMBER_OF_THREADS)
        {
            // it means this is last thread so notify first thread
            threadCounter = 1;
            consumerMutex[threadCounter].notify_one();
        }
        else
        {
            consumerMutex[++threadCounter].notify_one();
        }
    }
}

int main()
{
    std::thread thrds[NUMBER_OF_THREADS];

    for (int i = 0; i < NUMBER_OF_THREADS; i++)
    {
        thrds[i] = std::thread(consumer);
    }

    producer();

    for (int i = 0; i < NUMBER_OF_THREADS; i++)
    {
        thrds[i].join();
    }

    return 0;
}

Answer 6

/** For Scheduling the thread execution order, Evens are enough, no synchronization objects like Mutex, Critical Section etc.. are not needed. Below is a code and which is scalable based on the number of threads entered. */

//Code for Schedule threads execution in a order T1->T2->T3-> ..->Tn->T1..
// where T1 is Thread-1, T2 is Thread-2 .. Tn is Thread-N

#include <iostream>
#include <Windows.h>

HANDLE  * EventsHandlesArray;    
HANDLE * ThreadHandlesArray;

int iMaxWaitTime = 10000; // 10 Secs    
int iThreadWaitTimeInterval = 1000; //1 sec

int iNumberOfThread = 3; // You can change this no. of thread value within the permitted limit by your system :)

DWORD WINAPI ThreadFun( LPVOID lpParam )

{
    int iThreadIndex = reinterpret_cast<int> (lpParam);
    int iCurrentEVentIndex = iThreadIndex == 1 ? iNumberOfThread-1: iThreadIndex-2;
        int iNewEVentIndex = iThreadIndex -1;

    while(1)
    {
        switch(::WaitForSingleObject(EventsHandlesArray[iCurrentEVentIndex],iMaxWaitTime))
        {
            case WAIT_OBJECT_0:
                    std::cout<<" I am a thread "<<iThreadIndex<<std::endl;
                    ResetEvent ( EventsHandlesArray[iCurrentEVentIndex]);
                    Sleep (iThreadWaitTimeInterval);
                    SetEvent ( EventsHandlesArray[iNewEVentIndex]);
                break;

            case WAIT_FAILED:
                    std::cout<<"Error in thread"<<iThreadIndex<<" Thread wait time failed "<<std::endl;
                   ResetEvent ( EventsHandlesArray[iCurrentEVentIndex]);
                   SetEvent ( EventsHandlesArray[iCurrentEVentIndex]);
                break;

            case WAIT_TIMEOUT:
                    std::cout<<"Error in thread"<<iThreadIndex<<" Thread wait timed out"<<std::endl;
                    ResetEvent ( EventsHandlesArray[iCurrentEVentIndex]);
                   SetEvent ( EventsHandlesArray[iCurrentEVentIndex]);
                break;
        }
    }
}

int main(void)

{       
__try
    {
       std::cout<<"Running Main, Creating Events"<<std::endl;
       EventsHandlesArray = new HANDLE[iNumberOfThread];
       ThreadHandlesArray = new HANDLE[iNumberOfThread];

        for(int iCount = 0; iCount<iNumberOfThread; iCount++)
        {
            EventsHandlesArray[iCount] = CreateEvent ( NULL , true , false , NULL );
        }

        for(int iCount = 0; iCount<iNumberOfThread; iCount++)
        {
            if( EventsHandlesArray[iCount] == INVALID_HANDLE_VALUE)
            {
                std::cout<<"Problem with Event Creation"<<std::endl;
                for(int iCount =0;iCount<iNumberOfThread;iCount++)
                {
                    CloseHandle(EventsHandlesArray[iCount]); 
                }
                return 0;
            }
        }

        DWORD Id=0;
        int iThreadIndex = 1;

        for(int iCount = 0; iCount<iNumberOfThread; iCount++)
        {
            iThreadIndex = iCount+1;
            ThreadHandlesArray[iCount] = CreateThread ( NULL, 0, (LPTHREAD_START_ROUTINE)ThreadFun,(LPVOID)iThreadIndex ,CREATE_SUSPENDED,&Id );
            std::cout<<"Thread Created : "<<Id<<std::endl;
        }

        bool bThreadCreatedSuccessfully = true;

        for(int iCount = 0; iCount<iNumberOfThread; iCount++)
        {
            if( ThreadHandlesArray[iCount] == INVALID_HANDLE_VALUE || ThreadHandlesArray[iCount] == NULL)
            {
                bThreadCreatedSuccessfully = false;
                break;
            }
        }

        if(bThreadCreatedSuccessfully)
        {
            std::cout<<"Resuming Threads "<<std::endl;
            for(int iCount =0;iCount<iNumberOfThread;iCount++)
            {
                //Try to close the event handles
                ResumeThread(ThreadHandlesArray[iCount]);
            }

            Sleep (iThreadWaitTimeInterval);
            SetEvent ( EventsHandlesArray[iNumberOfThread-1]);
            WaitForMultipleObjects(iNumberOfThread,ThreadHandlesArray, TRUE, INFINITE);
        }
        else
        {
            std::cout<<"Issue with Thread Creation"<<std::endl;
        }
    }

    __finally 
    {
        //Close Threads & Events
        for(int iCount=0;iCount<iNumberOfThread;iCount++)
        {
            //Try to close the event handles
            CloseHandle(ThreadHandlesArray[iCount]);
            CloseHandle(EventsHandlesArray[iCount]);
        }
        std::cout<<"  .... Exiting the program"<<std::endl;
    }
    return 0;
}