vaclavt/usql
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

thread pool changed

Browse Source
This commit is contained in:
VaclavT
2022-01-16 17:50:15 +01:00
parent 43d45e5bca
commit 471f218bdf
4 changed files with 154 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CMakeLists.txt
View File

@@ -5,10 +5,7 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
set(CMAKE_CXX_EXTENSIONS OFF) set(CMAKE_CXX_EXTENSIONS OFF)
# set(CMAKE_CXX_FLAGS "-O3 -Wall -Wextra") # set(CMAKE_CXX_FLAGS "-O3 -Wall -Wextra")
# set(CMAKE_CXX_FLAGS_DEBUG "-Wall -d")
set(CMAKE_OSX_DEPLOYMENT_TARGET "10.14")
set(CMAKE_CXX_FLAGS_DEBUG "-Wall")
project(usql) project(usql)

206
csvreader.cpp
View File

@@ -8,138 +8,136 @@
namespace usql { namespace usql {
CsvReader::CsvReader(bool skip_hdr, char field_sep, char quote_ch, char line_sep, char line_sep2) { CsvReader::CsvReader(bool skip_hdr, char field_sep, char quote_ch, char line_sep, char line_sep2) {
skip_header = skip_hdr; skip_header = skip_hdr;
field_separator = field_sep; field_separator = field_sep;
quote_character = quote_ch; quote_character = quote_ch;
line_separator = line_sep; line_separator = line_sep;
line_separator2 = line_sep2; line_separator2 = line_sep2;
header_skiped = !skip_hdr; header_skiped = !skip_hdr;
} }
size_t CsvReader::parseCSVFile(const std::string &filename, std::vector<ColDefNode> &cols_def, Table &table) { size_t CsvReader::parseCSVFile(const std::string &filename, std::vector<ColDefNode> &cols_def, Table &table) {
size_t row_cnt = 0; size_t row_cnt = 0;
errno = 0; errno = 0;
FILE* fp = fopen(filename.c_str(), "r"); FILE *fp = fopen(filename.c_str(), "r");
if (fp == NULL) if (fp == NULL)
throw Exception("load from csv, cannot read from file(" + std::to_string(errno) + ")"); throw Exception("load from csv, cannot read from file(" + std::to_string(errno) + ")");
char* line_str = NULL; char *line_str = NULL;
size_t len = 0; size_t len = 0;
try {
// TODO handle it by settings // TODO handle it by settings
const std::size_t hw_concurrency = std::max(0, (int)(std::thread::hardware_concurrency() - 2)); const std::size_t hw_concurrency = std::max(0, (int)(std::thread::hardware_concurrency() - 2));
// std::cout << "pool size: " << hw_concurrency << "\n";
const bool use_threadpool = hw_concurrency > 1; const bool use_threadpool = hw_concurrency > 1;
thread_pool tp{hw_concurrency}; thread_pool tp{hw_concurrency};
std::mutex row_cnt_mutex; std::mutex row_cnt_mutex;
long read_chars; try {
while ((read_chars = getline(&line_str, &len, fp)) != -1) { long read_chars;
if (skip_header && !header_skiped) { while ((read_chars = getline(&line_str, &len, fp)) != -1) {
header_skiped = true; if (skip_header && !header_skiped) {
continue; header_skiped = true;
} continue;
if (read_chars > 0 && line_str[read_chars - 1] == '\n') { }
line_str[read_chars - 1] = '\0'; if (read_chars > 0 && line_str[read_chars - 1] == '\n') {
--read_chars; line_str[read_chars - 1] = '\0';
--read_chars;
}
if (use_threadpool) {
std::string str{line_str};
dispatch(tp, std::function<void()>
([this, str, &cols_def, &table, &row_cnt, &row_cnt_mutex]() {
auto parsed = parseCSVString(str, cols_def, table);
{
std::unique_lock<std::mutex> lock(row_cnt_mutex);
row_cnt += parsed;
}
}
));
} else
row_cnt += parseCSVString(line_str, cols_def, table);
} }
if (use_threadpool) { fclose(fp);
dispatch(tp, std::function<void()> tp.join();
([this, line_str, &cols_def, &table, &row_cnt, &row_cnt_mutex]() {
// std::cout << "thread: " << std::this_thread::get_id() << " rownum " << row_cnt << "\n"; } catch (const std::exception &e) {
auto parsed = parseCSVString(line_str, cols_def, table); if (line_str)
{ free(line_str);
std::unique_lock<std::mutex> lock(row_cnt_mutex);
row_cnt += parsed; throw e;
}
}
));
} else
row_cnt += parseCSVString(line_str, cols_def, table);
} }
if (use_threadpool) tp.finish();
fclose(fp);
} catch (const std::exception &e) {
if (line_str) if (line_str)
free(line_str); free(line_str);
throw e; return row_cnt;
} }
if (line_str) size_t CsvReader::parseCSVString(const std::string &csvSource, std::vector<ColDefNode> &cols_def, Table &table) {
free(line_str); size_t row_cnt = 0;
bool inQuote(false);
bool newLine(false);
return row_cnt; std::vector<std::string> columns;
} std::string field;
columns.reserve(256);
field.reserve(64);
size_t CsvReader::parseCSVString(const std::string csvSource, std::vector<ColDefNode> &cols_def, Table& table) { std::string::const_iterator aChar = csvSource.begin();
size_t row_cnt = 0; std::string::const_iterator aEnd = csvSource.end();
bool inQuote(false); while (aChar != aEnd) {
bool newLine(false); if (*aChar == quote_character) {
newLine = false;
std::vector<std::string> columns; inQuote = !inQuote;
std::string field; } else if (*aChar == field_separator) {
columns.reserve(256); newLine = false;
field.reserve(64); if (inQuote) {
field += *aChar;
std::string::const_iterator aChar = csvSource.begin(); } else {
std::string::const_iterator aEnd = csvSource.end();
while (aChar != aEnd) {
if (*aChar == quote_character) {
newLine = false;
inQuote = !inQuote;
} else if (*aChar == field_separator) {
newLine = false;
if (inQuote) {
field += *aChar;
} else {
columns.push_back(field);
field.clear();
}
} else if (*aChar == line_separator || *aChar == line_separator2) {
if (inQuote) {
field += *aChar;
} else {
if (!newLine) {
columns.push_back(field); columns.push_back(field);
if (header_skiped) {
table.create_row_from_vector(cols_def, columns);
row_cnt++;
}
header_skiped = true;
field.clear(); field.clear();
columns.clear();
newLine = true;
} }
} else if (*aChar == line_separator || *aChar == line_separator2) {
if (inQuote) {
field += *aChar;
} else {
if (!newLine) {
columns.push_back(field);
if (header_skiped) {
table.create_row_from_vector(cols_def, columns);
row_cnt++;
}
header_skiped = true;
field.clear();
columns.clear();
newLine = true;
}
}
} else {
newLine = false;
field.push_back(*aChar);
} }
} else {
newLine = false; aChar++;
field.push_back(*aChar);
} }
aChar++; if (!field.empty())
columns.push_back(field);
if (header_skiped) {
table.create_row_from_vector(cols_def, columns);
row_cnt++;
} else
header_skiped = true;
return row_cnt;
} }
if (!field.empty()) } // namespace
columns.push_back(field);
if (header_skiped) {
table.create_row_from_vector(cols_def, columns);
row_cnt++;
} else
header_skiped = true;
return row_cnt;
}
} // namespace

2
csvreader.h
View File

@@ -27,7 +27,7 @@ public:
explicit CsvReader(bool skip_hdr = true, char field_sep = ',', char quote_ch = '"', char line_sep = '\r', char line_sep2 = '\n'); explicit CsvReader(bool skip_hdr = true, char field_sep = ',', char quote_ch = '"', char line_sep = '\r', char line_sep2 = '\n');
size_t parseCSVFile(const std::string &filename, std::vector<ColDefNode> &cols_def, Table& table); size_t parseCSVFile(const std::string &filename, std::vector<ColDefNode> &cols_def, Table& table);
size_t parseCSVString(const std::string csvSource, std::vector<ColDefNode> &cols_def, Table& table); size_t parseCSVString(const std::string &csvSource, std::vector<ColDefNode> &cols_def, Table& table);
}; };
} // namespace } // namespace

125
threadpool.h
View File

@@ -1,104 +1,79 @@
#include <iostream>
#include <thread> #include <thread>
#include <mutex> #include <mutex>
#include <functional> #include <functional>
#include <future>
#include <queue> #include <queue>
#include <condition_variable> #include <condition_variable>
#include <vector> #include <vector>
struct thread_pool { class thread_pool {
explicit thread_pool(std::size_t size) { public:
start(size); thread_pool(std::size_t size) : stop(false) {
finished.reserve(1024); for (std::size_t i = 0; i < size; ++i) {
workers.emplace_back([this] { spawn(); });
}
} }
std::mutex mutex; virtual ~thread_pool() {
std::condition_variable condition; if (!stop) join();
std::deque<std::packaged_task<void()>> work; }
std::vector<std::future<void>> finished; void post(std::function<void()> f) {
// queue( lambda ) will enqueue the lambda into the tasks for the threads
template<class F, class R=std::result_of_t<F &()>>
std::future<R> queue(F &&f) {
// wrap the function object into a packaged task, splitting
// execution from the return value:
std::packaged_task<R()> p(std::forward<F>(f));
auto r = p.get_future();
{ {
std::unique_lock<std::mutex> l(mutex); std::unique_lock<std::mutex> lock(mutex);
work.emplace_back(std::move(p)); tasks.push(f);
} }
condition.notify_one(); condition.notify_one();
return r; // return the future result of the task
} }
// start threads_num threads in the thread pool. void join() {
void start(std::size_t threads_num = 1) {
for (std::size_t i = 0; i < threads_num; ++i) {
finished.push_back(
std::async(
std::launch::async,
[this] { thread_task(); }
)
);
}
}
// abort() cancels all non-started tasks, and tells every working thread
// stop running, and waits for them to finish up.
void abort() {
cancel_pending();
finish();
}
// cancel_pending() merely cancels all non-started tasks:
void cancel_pending() {
std::unique_lock<std::mutex> l(mutex);
work.clear();
}
// finish enques a "stop the thread" message for every thread, then waits for them:
void finish() {
{ {
std::unique_lock<std::mutex> l(mutex); std::unique_lock<std::mutex> lock(mutex);
for (auto &&unused:finished) { stop = true;
work.emplace_back();
}
} }
condition.notify_all();
finished.clear();
}
~thread_pool() { condition.notify_all();
finish();
for (std::size_t i = 0; i < workers.size(); ++i) {
workers[i].join();
}
} }
private: private:
void thread_task() { void spawn() {
while (true) { std::function<void()> task;
std::packaged_task<void()> f;
bool task_queue_empty = tasks.empty();
while (!stop || !task_queue_empty) {
bool task_valid = false;
{ {
std::unique_lock<std::mutex> l(mutex); std::unique_lock<std::mutex> lock(mutex);
if (work.empty()) { condition.wait(lock, [this]() {
condition.wait(l, [&] { return !work.empty(); }); return (!tasks.empty()) || (tasks.empty() && stop);
} });
f = std::move(work.front());
work.pop_front(); if (!tasks.empty()) {
task = std::move(tasks.front());
tasks.pop();
task_valid = true;
}
task_queue_empty = tasks.empty();
} }
// if the task is invalid, it means we are asked to abort:
if (!f.valid()) return; if (task_valid) task();
f();
} }
} }
public:
std::vector<std::thread> workers;
std::queue<std::function<void()>> tasks;
std::mutex mutex;
std::condition_variable condition;
bool stop;
}; };
inline void dispatch(thread_pool &pool, std::function<void()> f) { inline void dispatch(thread_pool &pool, std::function<void()> f) {
pool.queue(f); pool.post(f);
} }