I want to write small server that has one thread that pushes some jobs into queue and several threads that pull the jobs and process it somehow.
I wrote (of course no, I found an example) of blocking queue. Then I wrote cFakeWorker
that is server prototype. In general it works ok, but I'm not sure that the way I implemented termination is very good. And, so far, I will appreciate for ideas how should be done.
Currently I have m_bForceStop
member of cBlockingQueue
class. pull()
function checks its state with condition_variable
and throws exception in case of termination.
cFakeWorker
catches this exception while waits for the new jos.
The moment I call cBlockingQueue::stop_waiting()
is cFakeWorker::Terminate()
. This implementation works, but I have bad feeling that it might be done better. So I will appreciate for the review.
UPDATE! Uploading all sources @Phrancis, thanks for the comment. I will appreciate for comments.
main.cpp
void main()
{
cFakeWorker fw;
std::thread thFW([&fw](){fw.StartFlow();});
while (true)
{
std::string s;
std::cin >> s;
if (s == "exit")
{
std::cout << __func__ << " Terminating .... " << std::endl;
fw.Terminate();
std::cout << __func__ << " Termination done" << std::endl;
thFW.join();
return;
}
}
}
cFakeWorker.h
#ifndef CFAKEWORKER_H
#define CFAKEWORKER_H
#include <chrono>
#include <string>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include "cBlockingQueue.h"
using namespace std::chrono_literals;
struct taskData
{
std::string str;
std::chrono::milliseconds sleep_period;
};
class cFakeWorker
{
public:
cFakeWorker();
virtual ~cFakeWorker();
void StartFlow();
void Terminate();
private:
cBlockingQueue<taskData> m_queue;
int m_NCores;
int m_NPullThreads;
std::vector<std::thread> m_vecPullThreads;
std::thread m_PushThread;
std::mutex m_mtxIO;
std::mutex m_mtxTerminate;
std::condition_variable m_cvTerminate;
bool m_bForceTermination;
void PushJob();
void PullJob();
};
#endif /* CFAKEWORKER_H */
cFakeWorker.cpp
#include "cFakeWorker.h"
#include <iostream>
#include <sstream>
#include <cstdlib> // std::rand()
#include <ctime>
#include <time.h>
#include <iomanip>
#include <sched.h>
cFakeWorker::cFakeWorker()
{
m_NCores = std::thread::hardware_concurrency();
m_NPullThreads = (m_NCores > 1) ? m_NCores - 1 : 1;
}
void cFakeWorker::StartFlow()
{
m_bForceTermination = false;
m_PushThread = std::thread([this](){this->PushJob();});
for (int i = 0; i < m_NPullThreads; i++)
{
{
std::lock_guard<std::mutex> lckIO(m_mtxIO);
std::cout << "Starting pull thread #" << i << std::endl;
m_vecPullThreads.push_back(std::thread([this](){this->PullJob();}));
}
}
}
void cFakeWorker::PushJob()
{
std::condition_variable cv;
std::srand(std::time(nullptr));
while (true)
{
{
std::lock_guard<std::mutex> lck(m_mtxTerminate);
if(m_bForceTermination)
{
return;
}
}
std::stringstream ss;
std::chrono::high_resolution_clock::time_point tp = std::chrono::high_resolution_clock::now();
std::time_t t = std::chrono::system_clock::to_time_t(tp);
ss << std::put_time(std::localtime(&t), "%Y %a %b %d %H:%M:%S");
int nSleepDuration = ( std::rand() + 1) % 999;
std::this_thread::sleep_for(100ms);
taskData dat;
dat.str = ss.str();
dat.sleep_period = std::chrono::milliseconds(nSleepDuration);
m_queue.push(dat);
}
}
void cFakeWorker::PullJob()
{
while (m_queue.is_waiting())
{
try
{
taskData dat;
try
{
dat = m_queue.pull();
}
catch (const cTerminationException & ex)
{
std::cout << __func__ << ": " << "cBlockingQueue::pull() was terminated" << std::endl;
return;
}
{
std::lock_guard<std::mutex> lckIO(m_mtxIO);
std::cout << " Start doing "
<< "------ >>> "
<< "CPU: " << sched_getcpu()
<< " [" << std::this_thread::get_id() << "] --"
<< " msg : " << dat.str
<< " sleep: " << dat.sleep_period.count()
<< std::endl;
}
// fake load
for(int i = 1; i < 50000; i++)
{
char * p = new char [i];
for (int j = 0; j < i; ++j)
{
p[j] = 2;
}
delete [] p;
}
std::cout << " Stop doing "
<< "------ >>> "
<< "CPU: " << sched_getcpu()
<< std::endl;
}
catch (const std::exception & ex)
{
std::cout << __func__ << ": " << "something has happened. Keep pulling" << std::endl;
}
}
}
void cFakeWorker::Terminate()
{
{
std::lock_guard<std::mutex> lck(m_mtxTerminate);
m_bForceTermination = true;
}
m_PushThread.join();
m_queue.stop_waiting();
for (auto& th : m_vecPullThreads)
{
th.join();
}
}
cFakeWorker::~cFakeWorker()
{
try
{
Terminate();
}
catch (const std::exception & ex)
{}
}
cBlockingQueue.h (example code I found)
#ifndef CBLOCKINGQUEUE_H #define CBLOCKINGQUEUE_H #include <mutex> #include <deque> #include <condition_variable> #include <thread> #include <iostream> #include <chrono> using namespace std::chrono_literals; class cTerminationException : public std::exception { }; template <typename T> class cBlockingQueue { public: cBlockingQueue() { m_bForceStop = false; } void push(const T & param) { { std::unique_lock<std::mutex> lock(m_mut); m_deque.push_front(param); } m_cv.notify_all(); } T pull() { std::unique_lock<std::mutex> lock(m_mut); m_cv.wait(lock, [=](){return !this->m_deque.empty() || m_bForceStop;}); if (m_bForceStop) { std::cout << __func__ << " Throwing cTerminationException" << std::endl; throw cTerminationException(); } T ret(std::move(m_deque.back())); m_deque.pop_back(); return ret; } void stop_waiting() { m_bForceStop = true; m_cv.notify_all(); } bool is_waiting() { return !m_bForceStop; } private: std::mutex m_mut; std::deque<T> m_deque; std::condition_variable m_cv; bool m_bForceStop; }; #endif /* CBLOCKINGQUEUE_H */
I found an example
so did you write it yourself or not? \$\endgroup\$ – yuri Jun 11 '18 at 12:31push()
method \$\endgroup\$ – Yura Jun 11 '18 at 13:07