Many applications contain objects whose methods are invoked concurrently by multiple client threads. The idea of a monitor object is to synchronize the access to an object's methods so that only one method can execute at any one time.
#include <iostream>
#include <thread>
#include <vector>
#include <future>
#include <mutex>
class Vehicle
{
public:
Vehicle(int id) : _id(id) {}
int getID() { return _id; }
private:
int _id;
};
class WaitingVehicles
{
public:
WaitingVehicles() : _numVehicles(0) {}
int getNumVehicles()
{
std::lock_guard<std::mutex> uLock(_mutex);
return _numVehicles;
}
bool dataIsAvailable()
{
std::lock_guard<std::mutex> myLock(_mutex);
return !_vehicles.empty();
}
Vehicle popBack()
{
// perform vector modification under the lock
std::lock_guard<std::mutex> uLock(_mutex);
// remove last vector element from queue
Vehicle v = std::move(_vehicles.back());
_vehicles.pop_back();
--_numVehicles;
return v; // will not be copied due to return value optimization (RVO) in C++
}
void pushBack(Vehicle &&v)
{
// simulate some work
std::this_thread::sleep_for(std::chrono::milliseconds(100));
// perform vector modification under the lock
std::lock_guard<std::mutex> uLock(_mutex);
// add vector to queue
std::cout << " Vehicle #" << v.getID() << " will be added to the queue" << std::endl;
_vehicles.emplace_back(std::move(v));
++_numVehicles;
}
private:
std::vector<Vehicle> _vehicles; // list of all vehicles waiting to enter this intersection
std::mutex _mutex;
int _numVehicles;
};
int main()
{
// create monitor object as a shared pointer to enable access by multiple threads
std::shared_ptr<WaitingVehicles> queue(new WaitingVehicles);
std::cout << "Spawning threads..." << std::endl;
std::vector<std::future<void>> futures;
for (int i = 0; i < 10; ++i)
{
// create a new Vehicle instance and move it into the queue
Vehicle v(i);
futures.emplace_back(std::async(std::launch::async, &WaitingVehicles::pushBack, queue, std::move(v)));
}
std::cout << "Collecting results..." << std::endl;
while (true)
{
if (queue->dataIsAvailable())
{
Vehicle v = queue->popBack();
std::cout << " Vehicle #" << v.getID() << " has been removed from the queue" << std::endl;
if(queue->getNumVehicles()<=0)
{
std::this_thread::sleep_for(std::chrono::milliseconds(200));
break;
}
}
}
std::for_each(futures.begin(), futures.end(), [](std::future<void> &ftr) {
ftr.wait();
});
std::cout << "Finished : " << queue->getNumVehicles() << " vehicle(s) left in the queue" << std::endl;
return 0;
}Output:
Spawning threads...
Collecting results...
Vehicle #0 will be added to the queue
Vehicle #3 will be added to the queue
Vehicle #2 will be added to the queue
Vehicle #1 will be added to the queue
Vehicle #5 will be added to the queue
Vehicle #4 will be added to the queue
Vehicle #6 will be added to the queue
Vehicle #7 will be added to the queue
Vehicle #8 will be added to the queue
Vehicle #9 will be added to the queue
Vehicle #9 has been removed from the queue
Vehicle #8 has been removed from the queue
Vehicle #7 has been removed from the queue
Vehicle #6 has been removed from the queue
Vehicle #4 has been removed from the queue
Vehicle #5 has been removed from the queue
Vehicle #1 has been removed from the queue
Vehicle #2 has been removed from the queue
Vehicle #3 has been removed from the queue
Vehicle #0 has been removed from the queue
Finished : 0 vehicle(s) left in the queueThis solution is not optimal because as long as the program is running, the while-loop will keep the processor busy, constantly asking whether new data is available.
The alternative to a polling loop is for the main thread to block and wait for a signal that new data is available. This would prevent the infinite loop from keeping the processor busy. We have already discussed a mechanism that would fulfill this purpose - the promise-future construct. The problem with futures is that they can only be used a single time. Once a future is ready and get() has been called, it can not be used any more. For our purpose, we need a signaling mechanism that can be re-used. The C++ standard offers such a construct in the form of "condition variables".
Let us pretend our shared variable was a boolean called dataIsAvailable. Now let’s discuss two scenarios for the protocol depending on who acts first, the producer or the consumer thread.
Scenario 1
The consumer thread checks dataIsAvailable() and since it is false, the consumer thread blocks and waits on the condition variable. Later in time, the producer thread sets dataIsAvailable to true and calls notify_one on the condition variable. At this point, the consumer wakes up and proceeds with its work.
Scenario 2
Here, the producer thread comes first, sets dataIsAvailable() to true and calls notify_one. Then, the consumer thread comes and checks dataIsAvailable() and finds it to be true - so it does not call wait and proceeds directly with its work. Even though the notification is lost, it does not cause a problem in this construct - the message has been passed successfully through dataIsAvailable and the wait-lock has been avoided.
Here is one combination that will cause the program to lock:
The consumer thread reads dataIsAvailable(), which is false in the example. Then, the producer sets dataIsAvailable() to true and calls notify. Due to this unlucky interleaving of actions, the consumer thread calls wait because it has seen dataIsAvailable() as false. This is possible because the consumer thread tasks are not a joint atomic operation but may be separated by the scheduler and interleaved with some other tasks - in this case the two actions performed by the producer thread. The problem here is that after calling wait, the consumer thread will never wake up again. Also, as you may have noticed, the shared variable dataReady is not protected by a mutex here - which makes it even more likely that something will go wrong.
One reason for discussing these failed scenarios in such depth is to make you aware of the complexity of concurrent behavior - even with a simple protocol like the one we are discussing right now.
So let us now look at the final solution to the above problems and thus a working version of our communication protocol.
As seen above, we are closing the gap between reading the state and entering the wait. We are reading the state under the lock (red bar) and we call wait still under the lock. Then, we let wait release the lock and enter the wait state in one atomic step. This is only possible because the wait() method is able to take a lock as an argument. The lock that we can pass to wait however is not the lock_guard we have been using so often until now but instead it has to be a lock that can be temporarily unlocked inside wait - a suitable lock for this purpose would be the unique_lock type which we have discussed in the previous section.
#include <iostream>
#include <thread>
#include <queue>
#include <future>
#include <mutex>
template <class T>
class MessageQueue
{
public:
T receive()
{
// perform queue modification under the lock
std::unique_lock<std::mutex> uLock(_mutex);
_cond.wait(uLock, [this] { return !_messages.empty(); }); // pass unique lock to condition variable
// remove last vector element from queue
T msg = std::move(_messages.back());
_messages.pop_back();
return msg; // will not be copied due to return value optimization (RVO) in C++
}
void send(T &&msg)
{
// simulate some work
std::this_thread::sleep_for(std::chrono::milliseconds(100));
// perform vector modification under the lock
std::lock_guard<std::mutex> uLock(_mutex);
// add vector to queue
std::cout << " Message " << msg << " has been sent to the queue" << std::endl;
_messages.push_back(std::move(msg));
_cond.notify_one(); // notify client after pushing new Vehicle into vector
}
private:
std::mutex _mutex;
std::condition_variable _cond;
std::deque<T> _messages;
};
int main()
{
// create monitor object as a shared pointer to enable access by multiple threads
std::shared_ptr<MessageQueue<int>> queue(new MessageQueue<int>);
std::cout << "Spawning threads..." << std::endl;
std::vector<std::future<void>> futures;
for (int i = 0; i < 10; ++i)
{
int message = i;
futures.emplace_back(std::async(std::launch::async, &MessageQueue<int>::send, queue, std::move(message)));
}
std::cout << "Collecting results..." << std::endl;
while (true)
{
int message = queue->receive();
std::cout << " Message #" << message << " has been removed from the queue" << std::endl;
}
std::for_each(futures.begin(), futures.end(), [](std::future<void> &ftr) {
ftr.wait();
});
std::cout << "Finished!" << std::endl;
return 0;
}