std::thread - DESPERATE NEED HELP!!!

[Szyk Cech]

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

[Chris M. Thomasson]

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.

[Ralf Fassel]

* Szyk Cech <szyk...@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'

[Chris M. Thomasson]

On 9/11/2019 11:16 AM, Szyk Cech wrote:

^^^^^^^^^^^^^^^^^^^^^^^^^^^^
>         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?

[Pavel]

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

[Chris M. Thomasson]

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.

[...]

[Chris M. Thomasson]

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.


>
> [...]

[Chris M. Thomasson]

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);
}
};