Unverified Commit 30f8b58e authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #496 from knowledge4igor/fix_issue_473

Fix issue #473: threads problem
parents 43bdf4eb 7437a23b
...@@ -8,37 +8,78 @@ ...@@ -8,37 +8,78 @@
#include <curl/curl.h> #include <curl/curl.h>
#include <curl/easy.h> #include <curl/easy.h>
#include <vector>
#include <queue>
#include <mutex>
#include <thread>
using namespace std; using namespace std;
using namespace Pistache; using namespace Pistache;
static const size_t N_LETTERS = 26; static constexpr size_t N_LETTERS = 26;
static const size_t LETTER_REPEATS = 100000; static constexpr size_t LETTER_REPEATS = 100000;
static const size_t SET_REPEATS = 10; static constexpr size_t SET_REPEATS = 10;
static constexpr size_t N_WORKERS = 10;
void dumpData(const Rest::Request&req, Http::ResponseWriter response) {
UNUSED(req);
auto stream = response.stream(Http::Code::Ok); void dumpData(const Rest::Request& /*req*/, Http::ResponseWriter response)
char letter = 'A'; {
using Lock = std::mutex;
using Guard = std::lock_guard<Lock>;
std::mutex responseGuard; Lock responseLock;
std::vector<std::thread> workers; std::vector<std::thread> workers;
std::condition_variable cv;
std::queue<std::function<void()>> jobs;
Lock jobLock;
std::atomic<size_t> jobCounter(0);
constexpr size_t JOB_LIMIT = SET_REPEATS * N_LETTERS;
for (size_t j = 0; j < N_WORKERS; ++j)
{
workers.push_back(std::thread([&jobCounter, &cv, &jobLock, &jobs]()
{
while(jobCounter < JOB_LIMIT)
{
std::unique_lock<Lock> l(jobLock);
cv.wait(l, [&jobCounter, &jobs]{ return jobs.size() || !(jobCounter < JOB_LIMIT);});
if (!jobs.empty())
{
auto f = std::move(jobs.front());
jobs.pop();
l.unlock();
f();
++jobCounter;
l.lock();
}
}
cv.notify_all();
}));
}
auto stream = response.stream(Http::Code::Ok);
const char letter = 'A';
for (size_t s = 0; s < SET_REPEATS; ++s) { for (size_t s = 0; s < SET_REPEATS; ++s) {
for (size_t i = 0; i < N_LETTERS; ++i ) { for (size_t i = 0; i < N_LETTERS; ++i ) {
std::thread job([&,i]() -> void { auto job = [&stream, &responseLock, i]() -> void {
const size_t nchunks = 10; constexpr size_t nchunks = 10;
size_t chunk_size = LETTER_REPEATS / nchunks; constexpr size_t chunk_size = LETTER_REPEATS / nchunks;
std::string payload(chunk_size, letter + i); const std::string payload(chunk_size, letter + i);
{ {
std::unique_lock<std::mutex> lock(responseGuard); Guard guard(responseLock);
for (size_t chunk = 0; chunk < nchunks; ++chunk) { for (size_t chunk = 0; chunk < nchunks; ++chunk) {
stream.write(payload.c_str(), chunk_size); stream.write(payload.c_str(), chunk_size);
stream.flush(); stream.flush();
} }
} }
}); };
workers.push_back(std::move(job)); std::unique_lock<Lock> l(jobLock);
jobs.push(std::move(job));
l.unlock();
cv.notify_all();
} }
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment