This blog was last modified 416 days before.
Create A Semaphore Class
So we already known that C++ added concurrency support since C++11 standard, but it don't provide us a encapsulation of semaphore until C++20. So today we are going to make one ourself using the basic concurrency API provided by C++ like mutex and condition_variable.
Let's check out the implementation code first:
#include <thread>
#include <mutex>
#include <iostream>
#include <condition_variable>
using std::mutex, std::unique_lock, std::condition_variable, std::lock_guard;
class Semaphore
{
private:
std::mutex m_sema_mutex;
int m_available;
condition_variable m_positive;
public:
// Construct a new recording semaphore with specified initial value
Semaphore(int initial) : m_available(initial) {}
// V operation
void signal()
{
lock_guard lk(m_sema_mutex);
++m_available;
m_positive.notify_one();
}
// P operation
void wait()
{
unique_lock lk(m_sema_mutex);
m_positive.wait(lk, [this]()
{ return this->m_available > 0; });
m_available--;
}
};
The structure is really clear and quite simple. signal() means to increase the semaphore, and wait() means consume the semaphore.
lock_guard & unique_lock
One thing need to be noticed is that we have to use unique_lock inside wait() method. Since inside m_positive.wait() process, it may need to do lock and unlock to lk based on situation for multiple times, and simple lock_guard could not provide such flexibility .
For info about
lock_guard, check book P38
Example Using Semaphore
This part shows how we can use our custom semaphore class. Here we use provider/consumer model as an example.
Check the code below:
#include <thread>
#include <mutex>
#include <iostream>
#include <condition_variable>
using std::cout, std::cin;
constexpr char endq = '\n';
using std::mutex, std::unique_lock, std::condition_variable, std::lock_guard;
using std::thread;
class Semaphore
{
private:
std::mutex m_sema_mutex;
int m_available;
condition_variable m_positive;
public:
// Construct a new recording semaphore with specified initial value
Semaphore(int initial) : m_available(initial) {}
void signal()
{
lock_guard lk(m_sema_mutex);
m_available++;
m_positive.notify_one();
}
void wait()
{
unique_lock lk(m_sema_mutex);
m_positive.wait(lk, [this]()
{ return this->m_available > 0; });
m_available--;
}
};
// Capacity of repo
constexpr int CAPACITY = 10;
// use arr as repo
int arr[CAPACITY] = {0};
int begin = 0, end = 0;
int currentUsed()
{
int tmp = end - begin;
if (tmp < 0)
{
tmp += CAPACITY;
}
return tmp;
}
mutex repo_mutex;
// Create semaphore needed here
Semaphore avai(CAPACITY);
Semaphore used(0);
// function to produce goods
void product_goods()
{
cout << "-> P\n";
avai.wait();
repo_mutex.lock();
cout << " P ...\n";
arr[end] = 1;
end = (end + 1) % CAPACITY;
cout << " P OK ";
cout << currentUsed() << endq;
repo_mutex.unlock();
used.signal();
cout << "<- P\n";
}
// function to consume goods
void consume_goods()
{
cout << "-> C\n";
used.wait();
repo_mutex.lock();
cout << " C ...\n";
arr[begin] = 0;
begin = (begin + 1) % CAPACITY;
cout << " C OK ";
cout << currentUsed() << endq;
repo_mutex.unlock();
avai.signal();
cout << "<- C\n";
std::this_thread::sleep_for(std::chrono::microseconds(10));
}
class ActionThread
{
private:
int m_repeat;
// 0 is provider, 1 is comsumer
bool m_action_type;
public:
ActionThread(int repeat, bool action_type)
: m_repeat(repeat),
m_action_type(action_type) {}
void operator()()
{
for (int i = 0; i < m_repeat; ++i)
{
if (m_action_type)
{
consume_goods();
}
else
{
product_goods();
}
}
}
};
int main()
{
ActionThread provider(10, 0);
ActionThread consumer(10, 1);
thread consumer_thread(consumer);
thread provider_thread(provider);
consumer_thread.join();
provider_thread.join();
return 0;
}
Here we create two thread, one is responsible for provide goods and another is responsible for consuming goods.
A possible output is like below:
-> C
-> P
P ...
P OK 1
<- P
-> P
C ...
C OK 0
<- C
-> C
P ...
P OK 1
<- P
-> P
C ...
C OK 0
<- C
-> C
P ...
P OK 1
<- P
-> P
C ...
C OK 0
<- C
P ...
P OK 1
<- P
-> P
P ...
P OK 2
<- P
-> P
P ...
P OK 3
<- P
-> P
P ...
P OK 4
<- P
-> P
P ...
-> C
P OK 5
<- P
-> P
C ...
C OK 4
<- C
-> C
P ...
P OK 5
<- P
-> P
C ...
C OK 4
<- C
-> C
P ...
P OK 5
<- P
C ...
C OK 4
<- C
-> C
C ...
C OK 3
<- C
-> C
C ...
C OK 2
<- C
-> C
C ...
C OK 1
<- C
-> C
C ...
C OK 0
<- C
We can see that repo is protected, and there is only one single operation towards repo is allowed at the same time. (If not, you may see something like below)
C ...
P ...
C OK
P OK
// This means more than one process try to operate the repo at the same time
Just a Multi-thread Simulatation
The semaphore introduced in textbook focus on the resource race on multiple process, however in our C++ code simulation, we don't create multiple process, and we just use two thread to simulate the similar concurrency and resource race situation. But we should keep in mind that thread and process are two completely different things.
Adding Delay
Also I add some delay in the consuming function:
std::this_thread::sleep_for(std::chrono::microseconds(10));
This is just in order to increase the probability of the repo to store more goods. Try to delete this code line and see what will happen.
In my laptop, without this delay, the total goods number in repo will jump arround 0 and 1, which makes some output like below:
-> C
-> P
P ...
P OK 1
<- P
-> P
C ...
C OK 0
<- C
-> C
P ...
P OK 1
<- P
-> P
C ...
C OK 0
<- C
-> C
P ...
P OK 1
<- P
-> P
C ...
C OK 0
<- C
// ...
Not funny at all right lol. So adding a slightly delay could help you observe how locks work more clearly in this case.
Finally I have to say all code above is just my little experiment to try implementing a semaphore on older C++ which have no <semaphore> standard library, and the code my code may contains mistake and some structure may be not so well-formed, if there is any bug feel free to tell me lol.
No comment