Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > comp.lang.c++ > #68341 > unrolled thread

std::thread - DESPERATE NEED HELP!!!

Back to article view | Back to comp.lang.c++

Contents

  std::thread - DESPERATE NEED HELP!!! Szyk Cech <szykcech@spoko.pl> - 2019-09-11 20:16 +0200
    Re: std::thread - DESPERATE NEED HELP!!! "Chris M. Thomasson" <chris.m.thomasson.1@gmail.com> - 2019-09-11 23:01 -0700
    Re: std::thread - DESPERATE NEED HELP!!! Ralf Fassel <ralfixx@gmx.de> - 2019-09-12 11:14 +0200
    Re: std::thread - DESPERATE NEED HELP!!! "Chris M. Thomasson" <chris.m.thomasson.1@gmail.com> - 2019-09-13 12:53 -0700
    Re: std::thread - DESPERATE NEED HELP!!! Pavel <pauldontspamtolk@removeyourself.dontspam.yahoo> - 2019-09-13 23:38 -0400
      Re: std::thread - DESPERATE NEED HELP!!! "Chris M. Thomasson" <chris.m.thomasson.1@gmail.com> - 2019-09-13 22:07 -0700
        Re: std::thread - DESPERATE NEED HELP!!! "Chris M. Thomasson" <chris.m.thomasson.1@gmail.com> - 2019-09-13 22:17 -0700
          Re: std::thread - DESPERATE NEED HELP!!! "Chris M. Thomasson" <chris.m.thomasson.1@gmail.com> - 2019-09-13 23:01 -0700

#68341 — std::thread - DESPERATE NEED HELP!!!

Hi all!

[What I want]
I want write reliable and modern Semaphore class.
I want write unit test to validate it.

[What I do]
I visited some education institution pages and inspired by it
I rewrote their examples. Addresses are:
http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_h.html
http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_cpp.html

As you can see this class is simple and silly as hell.
But it not works for me.
I have race conditions for unknown reasons.
It works in debuger. It works when I add some:
cout << "message" << endl << flush;
But not works in stand alone (debug and release both).
My system is Linux KDE Neon (based on Ubuntu 18.04 LTS).
My processor is Intel Core Duo (released in 2009).

[How to reproduce]
I paste here my code with complete minimal example.
It will compile in release mode.
It is fully automatic you need only to create 3 small files
and run: run.sh form commad line.
After 2-3 executions it hangs for unknown reasons.

Please help me! Thanks in advance and best regards!
Szyk Cech

[Files]
=============================================================
BEGIN: run.sh

#!/bin/bash

./make.sh

lCounter=0

while [ 1 ]; do
     lCounter=$[lCounter + 1]
     echo "Run: $lCounter"
     ./semtest
done

END:  run.sh

=============================================================

BEGIN: make.sh

#!/bin/bash
g++ -std=c++17 -pthread -O2 -s -DNDEBUG ./Main.cpp -o semtest

END: make.sh

=============================================================

BEGIN: Main.cpp

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


using namespace std;

class Semaphore
{
public:
     Semaphore(int aCounter);

public:
     int value();
     void up();
     void down();

protected:
     int mCounter;
     std::mutex mAccessMutex;
     std::condition_variable mWaitingCondition;
};

Semaphore::Semaphore(int aCounter)
     : mCounter(aCounter)
{
}

int Semaphore::value()
{
     return mCounter;
}

void Semaphore::down()
{
     unique_lock lock(mAccessMutex);
     --mCounter;
     if(mCounter < 0)
         mWaitingCondition.wait(lock);
}

void Semaphore::up()
{
     unique_lock lock(mAccessMutex);
     ++mCounter;
     if(mCounter <= 0)
         mWaitingCondition.notify_one();
}

void SemTestThread(Semaphore* aSem)
{
     aSem->down();
}

int main()
{
     Semaphore lSem(5);
     vector<thread> lArray;
     lArray.resize(15);

     for(unsigned long i(0); i < lArray.size(); ++i)
         lArray[i] = thread(SemTestThread, &lSem);

     bool lResult(true);

     for(int i(0); i < 5; ++i)
         lArray[i].join();

     lResult = lResult && (lSem.value() == -10);

     for(int i(10); i > 0; --i)
     {
         lSem.up();
         lArray[15 - i].join();
         lResult = lResult && (lSem.value() == -i + 1);
     }

     cout << "lResult: " << lResult << endl << flush;
//    return EXIT_SUCCESS;
}

END: Main.cpp

[toc] | [next] | [standalone]

#68351

On 9/11/2019 11:16 AM, Szyk Cech wrote:
> Hi all!
> 
> [What I want]
> I want write reliable and modern Semaphore class.
> I want write unit test to validate it.
[...]

Here is a _very_ basic, bare bones semaphore:
__________________________
#include <iostream>
#include <mutex>
#include <condition_variable>

// Basic, bare bones C++ Semaphore
struct cpp_sem
{
     unsigned int m_count;
     std::mutex m_mutex;
     std::condition_variable m_cond;

     cpp_sem(unsigned int count = 0) : m_count(count)
     {

     }

     void dec() // wait
     {
         std::unique_lock<std::mutex> lock(m_mutex);
         while (m_count == 0) m_cond.wait(lock);
         --m_count;
     }

     void inc() // post
     {
         {
             std::unique_lock<std::mutex> lock(m_mutex);
             ++m_count;
         }

         m_cond.notify_one();
     }
};

int main() {

   std::cout << "p0\n";

   {
     cpp_sem sem(2);
     sem.dec();
     sem.dec();

     std::cout << "  a\n";

     sem.inc();
     sem.inc();
     sem.inc();

     std::cout << "  b\n";

     sem.dec();
     sem.dec();
     sem.dec();
     sem.inc();
     sem.dec();

     std::cout << "  c\n";

     // sem.dec(); // uncomment for deadlock!
   }

   std::cout << "p1\n";

   return 0;
}
__________________________


This can be improved upon.

[toc] | [prev] | [next] | [standalone]

#68355

* Szyk Cech <szykcech@spoko.pl>
| void Semaphore::down()
| {
|     unique_lock lock(mAccessMutex);
|     --mCounter;
|     if(mCounter < 0)
|         mWaitingCondition.wait(lock);
| }

Look up "spurious wakeups".  IMHO it is better to use the predicate-form
of cv.wait() (#2 in https://en.cppreference.com/w/cpp/thread/condition_variable/wait)

|     for(unsigned long i(0); i < lArray.size(); ++i)
|         lArray[i] = thread(SemTestThread, &lSem);
>
|     bool lResult(true);
>
|     for(int i(0); i < 5; ++i)
|         lArray[i].join();

There is no guarantee that the first five threads will *not* block in
sem->down().  Just that they are created first does not guarantee they
execute first.  If lArray[5] executes before lArray[4], lArray[4].join()
will block since it still waits in sem->down().

R'

[toc] | [prev] | [next] | [standalone]

#68400

On 9/11/2019 11:16 AM, Szyk Cech wrote:
> Hi all!
> 
> [What I want]
> I want write reliable and modern Semaphore class.
> I want write unit test to validate it.
> 
> [What I do]
> I visited some education institution pages and inspired by it
> I rewrote their examples. Addresses are:
> http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_h.html
> http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_cpp.html
> 
> As you can see this class is simple and silly as hell.
> But it not works for me.
> I have race conditions for unknown reasons.
> It works in debuger. It works when I add some:
> cout << "message" << endl << flush;
> But not works in stand alone (debug and release both).
> My system is Linux KDE Neon (based on Ubuntu 18.04 LTS).
> My processor is Intel Core Duo (released in 2009).
> 
> [How to reproduce]
> I paste here my code with complete minimal example.
> It will compile in release mode.
> It is fully automatic you need only to create 3 small files
> and run: run.sh form commad line.
> After 2-3 executions it hangs for unknown reasons.
> 
> Please help me! Thanks in advance and best regards!
> Szyk Cech
> 
> [Files]
> =============================================================
> BEGIN: run.sh
> 
> #!/bin/bash
> 
> ./make.sh
> 
> lCounter=0
> 
> while [ 1 ]; do
>      lCounter=$[lCounter + 1]
>      echo "Run: $lCounter"
>      ./semtest
> done
> 
> END:  run.sh
> 
> =============================================================
> 
> BEGIN: make.sh
> 
> #!/bin/bash
> g++ -std=c++17 -pthread -O2 -s -DNDEBUG ./Main.cpp -o semtest
> 
> END: make.sh
> 
> =============================================================
> 
> BEGIN: Main.cpp
> 
> #include <mutex>
> #include <condition_variable>
> #include <chrono>
> #include <thread>
> #include <vector>
> #include <iostream>
> 
> 
> using namespace std;
> 
> class Semaphore
> {
> public:
>      Semaphore(int aCounter);
> 
> public:
>      int value();
>      void up();
>      void down();
> 
> protected:
>      int mCounter;
>      std::mutex mAccessMutex;
>      std::condition_variable mWaitingCondition;
> };
> 
> Semaphore::Semaphore(int aCounter)
>      : mCounter(aCounter)
> {
> }
> 
> int Semaphore::value()
> {
>      return mCounter;
> }
> 
> void Semaphore::down()
> {
>      unique_lock lock(mAccessMutex);
>      --mCounter;
>      if(mCounter < 0)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^
>          mWaitingCondition.wait(lock);
> }

This is not how a condition variable was designed to be used. In fact, 
iirc, a conforming condvar impl does not even need to block, it can just 
yield the thread and continue on. You need to contain your wait 
predicate in a _loop_. The dec function from my example:
__________________________
void dec() // wait
{
     std::unique_lock<std::mutex> lock(m_mutex);
     while (m_count == 0) m_cond.wait(lock);
     --m_count;
}
__________________________

Notice the while loop?

[toc] | [prev] | [next] | [standalone]

#68404

Szyk Cech wrote:
> Hi all!
> 
> [What I want]
> I want write reliable and modern Semaphore class.
> I want write unit test to validate it.
> 
> [What I do]
> I visited some education institution pages and inspired by it
> I rewrote their examples. Addresses are:
> http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_h.html
> http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_cpp.html
> 
> As you can see this class is simple and silly as hell.
> But it not works for me.
> I have race conditions for unknown reasons.
> It works in debuger. It works when I add some:
> cout << "message" << endl << flush;
> But not works in stand alone (debug and release both).
> My system is Linux KDE Neon (based on Ubuntu 18.04 LTS).
> My processor is Intel Core Duo (released in 2009).
> 
> [How to reproduce]
> I paste here my code with complete minimal example.
> It will compile in release mode.
> It is fully automatic you need only to create 3 small files
> and run: run.sh form commad line.
> After 2-3 executions it hangs for unknown reasons.
> 
> Please help me! Thanks in advance and best regards!
> Szyk Cech
> 
> [Files]
> =============================================================
> BEGIN: run.sh
> 
> #!/bin/bash
> 
> ./make.sh
> 
> lCounter=0
> 
> while [ 1 ]; do
>     lCounter=$[lCounter + 1]
>     echo "Run: $lCounter"
>     ./semtest
> done
> 
> END:  run.sh
> 
> =============================================================
> 
> BEGIN: make.sh
> 
> #!/bin/bash
> g++ -std=c++17 -pthread -O2 -s -DNDEBUG ./Main.cpp -o semtest
> 
> END: make.sh
> 
> =============================================================
> 
> BEGIN: Main.cpp
> 
> #include <mutex>
> #include <condition_variable>
> #include <chrono>
> #include <thread>
> #include <vector>
> #include <iostream>
> 
> 
> using namespace std;
> 
> class Semaphore
> {
> public:
>     Semaphore(int aCounter);
> 
> public:
>     int value();
>     void up();
>     void down();
> 
> protected:
>     int mCounter;
>     std::mutex mAccessMutex;
>     std::condition_variable mWaitingCondition;
> };
> 
> Semaphore::Semaphore(int aCounter)
>     : mCounter(aCounter)
> {
> }
> 
> int Semaphore::value()
> {
>     return mCounter;
> }
> 
> void Semaphore::down()
> {
>     unique_lock lock(mAccessMutex);
>     --mCounter;
>     if(mCounter < 0)
>         mWaitingCondition.wait(lock);
> }
> 
> void Semaphore::up()
> {
>     unique_lock lock(mAccessMutex);
>     ++mCounter;
>     if(mCounter <= 0)
>         mWaitingCondition.notify_one();
> }
> 
> void SemTestThread(Semaphore* aSem)
> {
>     aSem->down();
> }
> 
> int main()
> {
>     Semaphore lSem(5);
>     vector<thread> lArray;
>     lArray.resize(15);
> 
>     for(unsigned long i(0); i < lArray.size(); ++i)
>         lArray[i] = thread(SemTestThread, &lSem);
> 
>     bool lResult(true);
> 
>     for(int i(0); i < 5; ++i)
>         lArray[i].join();
Code might hang here because these 5 threads are waiting for a signal
but the signal isonly sent by up() which is called below in the same thread.

On a side note, down() shall not really decrease the semaphore until it
is > 0 and it shall wait for it in a loop in this non-exact-waiting test.

Sub-Note: sometimes (but I think, not in your case) it is correct to
wait for signal not in a loop -- but this is only when the protocol is
"exact" i.e. you know exactly which signals (notifications) every thread
is supposed to get. I think I posted some example here while ago with an
example of exact waiting for dining philosophers problem -- see class
EventQueue in
https://groups.google.com/forum/#!original/comp.lang.c++/sV4WC_cBb9Q/PAIxbypUCAAJ
.

HTH -Pavel
> 
>     lResult = lResult && (lSem.value() == -10);
> 
>     for(int i(10); i > 0; --i)
>     {
>         lSem.up();
>         lArray[15 - i].join();
>         lResult = lResult && (lSem.value() == -i + 1);
>     }
> 
>     cout << "lResult: " << lResult << endl << flush;
> //    return EXIT_SUCCESS;
> }
> 
> END: Main.cpp
> 
> 
> 

[toc] | [prev] | [next] | [standalone]

#68406

On 9/13/2019 8:38 PM, Pavel wrote:
> Szyk Cech wrote:
>> Hi all!
>>
>> [What I want]
>> I want write reliable and modern Semaphore class.
>> I want write unit test to validate it.
>>
>> [What I do]
>> I visited some education institution pages and inspired by it
>> I rewrote their examples. Addresses are:
>> http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_h.html
>> http://sandbox.mc.edu/~bennet/cs422b/notes/semaphore_cpp.html
>>
>> As you can see this class is simple and silly as hell.
>> But it not works for me.
>> I have race conditions for unknown reasons.
>> It works in debuger. It works when I add some:
>> cout << "message" << endl << flush;
>> But not works in stand alone (debug and release both).
>> My system is Linux KDE Neon (based on Ubuntu 18.04 LTS).
>> My processor is Intel Core Duo (released in 2009).
>>
>> [How to reproduce]
>> I paste here my code with complete minimal example.
>> It will compile in release mode.
>> It is fully automatic you need only to create 3 small files
>> and run: run.sh form commad line.
>> After 2-3 executions it hangs for unknown reasons.
>>
>> Please help me! Thanks in advance and best regards!
>> Szyk Cech
>>
>> [Files]
>> =============================================================
>> BEGIN: run.sh
>>
>> #!/bin/bash
>>
>> ./make.sh
>>
>> lCounter=0
>>
>> while [ 1 ]; do
>>      lCounter=$[lCounter + 1]
>>      echo "Run: $lCounter"
>>      ./semtest
>> done
>>
>> END:  run.sh
>>
>> =============================================================
>>
>> BEGIN: make.sh
>>
>> #!/bin/bash
>> g++ -std=c++17 -pthread -O2 -s -DNDEBUG ./Main.cpp -o semtest
>>
>> END: make.sh
>>
>> =============================================================
>>
>> BEGIN: Main.cpp
>>
>> #include <mutex>
>> #include <condition_variable>
>> #include <chrono>
>> #include <thread>
>> #include <vector>
>> #include <iostream>
>>
>>
>> using namespace std;
>>
>> class Semaphore
>> {
>> public:
>>      Semaphore(int aCounter);
>>
>> public:
>>      int value();
>>      void up();
>>      void down();
>>
>> protected:
>>      int mCounter;
>>      std::mutex mAccessMutex;
>>      std::condition_variable mWaitingCondition;
>> };
>>
>> Semaphore::Semaphore(int aCounter)
>>      : mCounter(aCounter)
>> {
>> }
>>
>> int Semaphore::value()
>> {
>>      return mCounter;
>> }
>>
>> void Semaphore::down()
>> {
>>      unique_lock lock(mAccessMutex);
>>      --mCounter;
>>      if(mCounter < 0)
>>          mWaitingCondition.wait(lock);
>> }
>>
>> void Semaphore::up()
>> {
>>      unique_lock lock(mAccessMutex);
>>      ++mCounter;
>>      if(mCounter <= 0)
>>          mWaitingCondition.notify_one();
>> }
>>
>> void SemTestThread(Semaphore* aSem)
>> {
>>      aSem->down();
>> }
>>
>> int main()
>> {
>>      Semaphore lSem(5);
>>      vector<thread> lArray;
>>      lArray.resize(15);
>>
>>      for(unsigned long i(0); i < lArray.size(); ++i)
>>          lArray[i] = thread(SemTestThread, &lSem);
>>
>>      bool lResult(true);
>>
>>      for(int i(0); i < 5; ++i)
>>          lArray[i].join();
> Code might hang here because these 5 threads are waiting for a signal
> but the signal isonly sent by up() which is called below in the same thread.
> 
> On a side note, down() shall not really decrease the semaphore until it
> is > 0 and it shall wait for it in a loop in this non-exact-waiting test.
> 
> Sub-Note: sometimes (but I think, not in your case) it is correct to
> wait for signal not in a loop --

Using condition variables directly? If the code thinks that the return 
from a condvar wait means a signal was actually sent, well, this is 
mistaken. However, there is something called the eventcount, that can 
abuse a condvar wait. The predicate is atomic.

[...]

[toc] | [prev] | [next] | [standalone]

#68407

On 9/13/2019 10:07 PM, Chris M. Thomasson wrote:
> On 9/13/2019 8:38 PM, Pavel wrote:
[...]
>> Sub-Note: sometimes (but I think, not in your case) it is correct to
>> wait for signal not in a loop --
> 
> Using condition variables directly? If the code thinks that the return 
> from a condvar wait means a signal was actually sent, well, this is 
> mistaken. However, there is something called the eventcount, that can 
> abuse a condvar wait. The predicate is atomic.

The loop is different in an eventcount, iirc its something like:

imagine trying to pop from any lock-free collection, and you want to 
make it wait on an empty condition, well:

// quick pseudo-code
____________________________________________
eventcount ec;

node* n = null;

// FAST PATH
while ((n = try_pop()) == null)
{
     // SLOW PATH
     eventcount_key eckey = ec.begin();

     if ((n = try_pop()) != null)
     {
         ec.leave(eckey);
         break;
     }

     // wait for it!
     ec.wait(eckey);
}

// n is an actual node we just popped! nice. :^)
____________________________________________

The double check for the predicate is required here. An eventcount is 
like a condition variable for lock-free algorithms.


> 
> [...]

[toc] | [prev] | [next] | [standalone]

#68408

On 9/13/2019 10:17 PM, Chris M. Thomasson wrote:
> On 9/13/2019 10:07 PM, Chris M. Thomasson wrote:
>> On 9/13/2019 8:38 PM, Pavel wrote:
> [...]
>>> Sub-Note: sometimes (but I think, not in your case) it is correct to
>>> wait for signal not in a loop --
>>
>> Using condition variables directly? If the code thinks that the return 
>> from a condvar wait means a signal was actually sent, well, this is 
>> mistaken. However, there is something called the eventcount, that can 
>> abuse a condvar wait. The predicate is atomic.
> 
> The loop is different in an eventcount, iirc its something like:
> 
> imagine trying to pop from any lock-free collection, and you want to 
> make it wait on an empty condition, well:
> 
> // quick pseudo-code
> ____________________________________________
> eventcount ec;
> 
> node* n = null;
> 
> // FAST PATH
> while ((n = try_pop()) == null)
> {
>      // SLOW PATH
>      eventcount_key eckey = ec.begin();
> 
>      if ((n = try_pop()) != null)
>      {
>          ec.leave(eckey);
>          break;
>      }
> 
>      // wait for it!
>      ec.wait(eckey);
> }
> 
> // n is an actual node we just popped! nice. :^)
> ____________________________________________
> 
> The double check for the predicate is required here. An eventcount is 
> like a condition variable for lock-free algorithms.

Wrt the use case above, here is a crude eventcount (called waitset) 
implementation that uses Relacy Race Detector, but its easy to turn into 
pure C++11. I put the standalone membars in some macros, sorry for that!


#define mb_relaxed std::memory_order_relaxed
#define mb_consume std::memory_order_consume
#define mb_acquire std::memory_order_acquire
#define mb_release std::memory_order_release
#define mb_acq_rel std::memory_order_acq_rel
#define mb_seq_cst std::memory_order_seq_cst


#define mb_relaxed_fence() std::atomic_thread_fence(mb_relaxed)
#define mb_consume_fence() std::atomic_thread_fence(mb_consume)
#define mb_acquire_fence() std::atomic_thread_fence(mb_acquire)
#define mb_release_fence() std::atomic_thread_fence(mb_release)
#define mb_acq_rel_fence() std::atomic_thread_fence(mb_acq_rel)
#define mb_seq_cst_fence() std::atomic_thread_fence(mb_seq_cst)


class waitset
{
     std::mutex m_mutex;
     std::condition_variable m_cond;
     std::atomic<bool> m_waitbit;
     VAR_T(unsigned) m_waiters;



public:
     waitset()
     :   m_waitbit(false),
         m_waiters(0)
     {

     }


     ~waitset()
     {
         bool waitbit = m_waitbit.load(mb_relaxed);

         unsigned waiters = VAR(m_waiters);

         RL_ASSERT(! waitbit && ! waiters);
     }



private:
     void prv_signal(bool waitbit, bool broadcast)
     {
         if (! waitbit) return;

         m_mutex.lock($);

         unsigned waiters = VAR(m_waiters);

         if (waiters < 2 || broadcast)
         {
             m_waitbit.store(false, mb_relaxed);
         }

         m_mutex.unlock($);

         if (waiters)
         {
             if (! broadcast)
             {
                 m_cond.notify_one($);
             }

             else
             {
                 m_cond.notify_all($);
             }
         }
     }



public:
     unsigned wait_begin()
     {
         m_mutex.lock($);

         m_waitbit.store(true, mb_relaxed);

         mb_seq_cst_fence();

         return 0;
     }


     bool wait_try_begin(unsigned& key)
     {
         if (! m_mutex.try_lock($)) return false;

         m_waitbit.store(true, mb_relaxed);

         mb_seq_cst_fence();

         return true;
     }


     void wait_cancel(unsigned key)
     {
         unsigned waiters = VAR(m_waiters);

         if (! waiters)
         {
             m_waitbit.store(false, mb_relaxed);
         }

         m_mutex.unlock($);
     }


     void wait_commit(unsigned key)
     {
         ++VAR(m_waiters);

         m_cond.wait(m_mutex, $);

         if (! --VAR(m_waiters))
         {
             m_waitbit.store(false, mb_relaxed);
         }

         m_mutex.unlock($);
     }



public:
     void signal()
     {
         mb_seq_cst_fence();

         bool waitbit = m_waitbit.load(std::memory_order_relaxed);

         prv_signal(waitbit, false);
     }


     void broadcast()
     {
         mb_seq_cst_fence();

         bool waitbit = m_waitbit.load(std::memory_order_relaxed);

         prv_signal(waitbit, true);
     }
};

[toc] | [prev] | [standalone]

Back to top | Article view | comp.lang.c++

csiph-web