pgrondek/polybar-dwm
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(eventloop): Use concurrent queue for events

Browse Source 
Events are now enqueued using a thread safe concurrent queue
which makes the previous eventloop locking redundant.
This commit is contained in:
Michael Carlberg 2016-10-29 06:48:51 +02:00
parent 92900e78d6
commit 1075144b00
15 changed files with 5263 additions and 384 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
.ycm_extra_conf.py
View File

@ -48,20 +48,12 @@ def DirectoryOfThisScript():
flags.append('-I'+ DirectoryOfThisScript() +'/src')
flags.append('-I'+ DirectoryOfThisScript() +'/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/gsl')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/cpp_freetype/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/boost/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/concurrentqueue/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/i3ipcpp/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/xpp/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/lemonbar/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/fastdelegate/include')
flags.append('-I'+ DirectoryOfThisScript() +'/lib/boost/include')
flags.append('-I'+ DirectoryOfThisScript() +'/tests')
flags.append('-I/usr/include/freetype2')
flags.append('-I/usr/include/pango-1.0')
flags.append('-I/usr/include/cairomm-1.0')
flags.append('-I/usr/include/pangomm-1.4')
flags.append('-I/usr/include/glibmm-2.4')
flags.append('-I/usr/lib/cairomm-1.0/include')
flags.append('-I/usr/include')
def MakeRelativePathsInFlagsAbsolute( flags, working_directory ):

13
include/common.hpp
View File

@ -44,6 +44,7 @@ LEMONBUDDY_NS
namespace di = boost::di;
namespace chrono = std::chrono;
namespace this_thread = std::this_thread;
namespace placeholders = std::placeholders;
using namespace std::chrono_literals;
@ -53,6 +54,7 @@ using std::size_t;
using std::move;
using std::bind;
using std::forward;
using std::pair;
using std::function;
using std::shared_ptr;
using std::unique_ptr;
@ -128,4 +130,15 @@ auto read_env = [](const char* var, string&& fallback = "") {
return value != nullptr ? value : fallback;
};
template <class T>
auto time_execution(const T& expr) noexcept {
auto start = std::chrono::high_resolution_clock::now();
expr();
auto finish = std::chrono::high_resolution_clock::now();
return std::chrono::duration_cast<std::chrono::milliseconds>(finish - start).count();
}
template <typename... Args>
using callback = function<void(Args...)>;
LEMONBUDDY_NS_END

24
include/components/bar.hpp
View File

@ -434,21 +434,21 @@ class bar : public xpp::event::sink<evt::button_press, evt::expose, evt::propert
// Connect signal handlers {{{
// clang-format off
g_signals::parser::alignment_change = bind(&bar::on_alignment_change, this, std::placeholders::_1);
g_signals::parser::attribute_set = bind(&bar::on_attribute_set, this, std::placeholders::_1);
g_signals::parser::attribute_unset = bind(&bar::on_attribute_unset, this, std::placeholders::_1);
g_signals::parser::attribute_toggle = bind(&bar::on_attribute_toggle, this, std::placeholders::_1);
g_signals::parser::action_block_open = bind(&bar::on_action_block_open, this, std::placeholders::_1, std::placeholders::_2);
g_signals::parser::action_block_close = bind(&bar::on_action_block_close, this, std::placeholders::_1);
g_signals::parser::color_change = bind(&bar::on_color_change, this, std::placeholders::_1, std::placeholders::_2);
g_signals::parser::font_change = bind(&bar::on_font_change, this, std::placeholders::_1);
g_signals::parser::pixel_offset = bind(&bar::on_pixel_offset, this, std::placeholders::_1);
g_signals::parser::ascii_text_write = bind(&bar::draw_character, this, std::placeholders::_1);
g_signals::parser::unicode_text_write = bind(&bar::draw_character, this, std::placeholders::_1);
g_signals::parser::alignment_change = bind(&bar::on_alignment_change, this, placeholders::_1);
g_signals::parser::attribute_set = bind(&bar::on_attribute_set, this, placeholders::_1);
g_signals::parser::attribute_unset = bind(&bar::on_attribute_unset, this, placeholders::_1);
g_signals::parser::attribute_toggle = bind(&bar::on_attribute_toggle, this, placeholders::_1);
g_signals::parser::action_block_open = bind(&bar::on_action_block_open, this, placeholders::_1, placeholders::_2);
g_signals::parser::action_block_close = bind(&bar::on_action_block_close, this, placeholders::_1);
g_signals::parser::color_change = bind(&bar::on_color_change, this, placeholders::_1, placeholders::_2);
g_signals::parser::font_change = bind(&bar::on_font_change, this, placeholders::_1);
g_signals::parser::pixel_offset = bind(&bar::on_pixel_offset, this, placeholders::_1);
g_signals::parser::ascii_text_write = bind(&bar::draw_character, this, placeholders::_1);
g_signals::parser::unicode_text_write = bind(&bar::draw_character, this, placeholders::_1);
// clang-format on
if (m_tray.align != alignment::NONE)
g_signals::tray::report_slotcount = bind(&bar::on_tray_report, this, std::placeholders::_1);
g_signals::tray::report_slotcount = bind(&bar::on_tray_report, this, placeholders::_1);
// }}}

501
include/components/controller.hpp
View File

@ -5,6 +5,7 @@
#include "common.hpp"
#include "components/bar.hpp"
#include "components/config.hpp"
#include "components/eventloop.hpp"
#include "components/logger.hpp"
#include "components/signals.hpp"
#include "components/x11/connection.hpp"
@ -12,11 +13,10 @@
#include "components/x11/tray.hpp"
#include "components/x11/types.hpp"
#include "config.hpp"
#include "utils/command.hpp"
#include "utils/inotify.hpp"
#include "utils/process.hpp"
#include "utils/socket.hpp"
#include "utils/throttle.hpp"
#include "utils/string.hpp"
#include "modules/backlight.hpp"
#include "modules/battery.hpp"
@ -46,7 +46,6 @@
LEMONBUDDY_NS
using namespace modules;
using module_t = unique_ptr<module_interface>;
class controller {
public:
@ -54,10 +53,12 @@ class controller {
* Construct controller
*/
explicit controller(connection& conn, const logger& logger, const config& config,
unique_ptr<bar> bar, unique_ptr<traymanager> tray, inotify_watch_t& confwatch)
unique_ptr<eventloop> eventloop, unique_ptr<bar> bar, unique_ptr<traymanager> tray,
inotify_watch_t& confwatch)
: m_connection(conn)
, m_log(logger)
, m_conf(config)
, m_eventloop(forward<decltype(eventloop)>(eventloop))
, m_bar(forward<decltype(bar)>(bar))
, m_traymanager(forward<decltype(tray)>(tray))
, m_confwatch(confwatch) {}
@ -67,34 +68,29 @@ class controller {
* threads and spawned processes
*/
~controller() noexcept {
if (!m_mutex.try_lock_for(5s)) {
m_log.warn("Failed to acquire lock for 5s... Forcing shutdown using SIGKILL");
raise(SIGKILL);
}
std::lock_guard<std::timed_mutex> guard(m_mutex, std::adopt_lock);
m_log.info("Stopping modules");
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
module->stop();
}
}
g_signals::bar::action_click = nullptr;
if (m_command) {
m_log.info("Terminating running shell command");
m_command->terminate();
}
if (m_traymanager)
m_traymanager.reset();
if (m_eventloop) {
m_log.info("Deconstructing eventloop");
m_eventloop->set_update_cb(nullptr);
m_eventloop->set_input_db(nullptr);
m_eventloop.reset();
}
if (m_bar) {
m_log.trace("controller: Deconstruct bar instance");
g_signals::bar::action_click = nullptr;
m_log.info("Deconstructing bar");
m_bar.reset();
}
if (m_traymanager) {
m_traymanager.reset();
}
m_log.info("Interrupting X event loop");
m_connection.send_dummy_event(m_connection.root());
@ -124,8 +120,8 @@ class controller {
/**
* Setup X environment
*/
auto bootstrap(bool to_stdout = false, bool dump_wmname = false) {
m_stdout = to_stdout;
auto bootstrap(bool writeback = false, bool dump_wmname = false) {
m_writeback = writeback;
m_log.trace("controller: Initialize X atom cache");
m_connection.preload_atoms();
@ -133,13 +129,14 @@ class controller {
m_log.trace("controller: Query X extension data");
m_connection.query_extensions();
// Disabled X extensions {{{
// const auto& damage_ext = m_connection.extension<xpp::damage::extension>();
// m_log.trace("controller: Found 'Damage' (first_event: %i, first_error: %i)",
// damage_ext->first_event, damage_ext->first_error);
// const auto& render_ext = m_connection.extension<xpp::render::extension>();
// m_log.trace("controller: Found 'Render' (first_event: %i, first_error: %i)",
// render_ext->first_event, render_ext->first_error);
// }}}
const auto& randr_ext = m_connection.extension<xpp::randr::extension>();
m_log.trace("controller: Found 'RandR' (first_event: %i, first_error: %i)",
@ -148,7 +145,6 @@ class controller {
// Listen for events on the root window to be able to
// break the blocking wait call when cleaning up
m_log.trace("controller: Listen for events on the root window");
try {
const uint32_t value_list[1]{XCB_EVENT_MASK_STRUCTURE_NOTIFY};
m_connection.change_window_attributes_checked(
@ -157,21 +153,26 @@ class controller {
throw application_error("Failed to change root window event mask: " + string{err.what()});
}
g_signals::bar::action_click = bind(&controller::on_mouse_event, this, placeholders::_1);
m_log.trace("controller: Attach eventloop callbacks");
m_eventloop->set_update_cb(bind(&controller::on_update, this));
m_eventloop->set_input_db(bind(&controller::on_unrecognized_action, this, placeholders::_1));
try {
m_log.trace("controller: Setup bar renderer");
m_bar->bootstrap(m_stdout || dump_wmname);
if (dump_wmname) {
std::cout << m_bar->settings().wmname << std::endl;
return;
} else if (!to_stdout) {
g_signals::bar::action_click = bind(&controller::on_module_click, this, std::placeholders::_1);
}
m_log.trace("controller: Setup bar");
m_bar->bootstrap(m_writeback || dump_wmname);
} catch (const std::exception& err) {
throw application_error("Failed to setup bar renderer: " + string{err.what()});
}
if (dump_wmname) {
std::cout << m_bar->settings().wmname << std::endl;
return;
}
try {
if (m_stdout) {
if (m_writeback) {
m_log.trace("controller: Disabling tray (reason: stdout mode)");
m_traymanager.reset();
} else if (m_bar->tray().align == alignment::NONE) {
@ -187,14 +188,8 @@ class controller {
m_traymanager.reset();
}
m_log.trace("main: Setup bar modules");
m_log.trace("controller: Setup user-defined modules");
bootstrap_modules();
// Allow <throttle_limit> ticks within <throttle_ms> timeframe
const auto throttle_limit = m_conf.get<unsigned int>("settings", "throttle-limit", 3);
const auto throttle_ms = chrono::duration<double, std::milli>(
m_conf.get<unsigned int>("settings", "throttle-ms", 60));
m_throttler = throttle_util::make_throttler(throttle_limit, throttle_ms);
}
/**
@ -209,73 +204,36 @@ class controller {
install_sigmask();
install_confwatch();
m_threads.emplace_back([this] {
m_connection.flush();
// Wait for term signal in separate thread
m_threads.emplace_back(thread(&controller::wait_for_signal, this));
m_log.trace("controller: Start modules");
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
try {
module->start();
} catch (const application_error& err) {
m_log.err("Failed to start '%s' (reason: %s)", module->name(), err.what());
}
// Activate traymanager in separate thread
if (!m_writeback && m_traymanager) {
m_threads.emplace_back(thread(&controller::activate_tray, this));
}
// Offset the initial broadcasts by 25ms to
// avoid the updates from being ignored by the throttler
this_thread::sleep_for(25ms);
}
}
// Listen for X events in separate thread
if (!m_writeback) {
m_threads.emplace_back(thread(&controller::wait_for_xevent, this));
}
if (m_stdout) {
m_log.trace("controller: Ignoring tray manager (reason: stdout mode)");
m_log.trace("controller: Ignoring X event loop (reason: stdout mode)");
return;
}
// Start event loop
if (m_eventloop) {
auto throttle_ms = m_conf.get<double>("settings", "throttle-ms", 10);
auto throttle_limit = m_conf.get<int>("settings", "throttle-limit", 5);
m_eventloop->run(chrono::duration<double, std::milli>(throttle_ms), throttle_limit);
}
if (m_traymanager) {
try {
m_log.trace("controller: Activate tray manager");
m_traymanager->activate();
} catch (const std::exception& err) {
m_log.err(err.what());
m_log.err("Failed to activate tray manager, disabling...");
m_traymanager.reset();
}
}
m_connection.flush();
m_log.trace("controller: Listen for X events");
while (m_running) {
auto evt = m_connection.wait_for_event();
if (evt != nullptr && m_running)
m_connection.dispatch_event(evt);
}
});
wait();
// Wake up signal thread
if (m_waiting) {
kill(getpid(), SIGTERM);
}
m_running = false;
return !m_reload;
}
/**
* Block execution until a defined signal is raised
*/
void wait() {
m_log.trace("controller: Wait for signal");
int caught_signal = 0;
sigwait(&m_waitmask, &caught_signal);
m_log.warn("Termination signal received, shutting down...");
m_log.trace("controller: Caught signal %d", caught_signal);
m_reload = (caught_signal == SIGUSR1);
}
protected:
/**
* Set signal mask for the current and future threads
@ -294,6 +252,12 @@ class controller {
if (pthread_sigmask(SIG_BLOCK, &m_waitmask, nullptr) == -1)
throw system_error();
sigemptyset(&m_ignmask);
sigaddset(&m_ignmask, SIGPIPE);
if (pthread_sigmask(SIG_BLOCK, &m_ignmask, nullptr) == -1)
throw system_error();
}
/**
@ -334,21 +298,65 @@ class controller {
});
}
void wait_for_signal() {
m_log.trace("controller: Wait for signal");
m_waiting = true;
int caught_signal = 0;
sigwait(&m_waitmask, &caught_signal);
m_log.warn("Termination signal received, shutting down...");
m_log.trace("controller: Caught signal %d", caught_signal);
if (m_eventloop) {
m_eventloop->stop();
}
m_reload = (caught_signal == SIGUSR1);
m_waiting = false;
}
void wait_for_xevent() {
m_log.trace("controller: Listen for X events");
m_connection.flush();
while (true) {
shared_ptr<xcb_generic_event_t> evt;
if ((evt = m_connection.wait_for_event()))
m_connection.dispatch_event(evt);
if (!m_running)
break;
}
}
void activate_tray() {
m_log.trace("controller: Activate tray manager");
try {
m_traymanager->activate();
} catch (const std::exception& err) {
m_log.err(err.what());
m_log.err("Failed to activate tray manager, disabling...");
m_traymanager.reset();
}
}
/**
* Create and initialize bar modules
*/
void bootstrap_modules() {
m_modules.emplace(alignment::LEFT, vector<module_t>{});
m_modules.emplace(alignment::CENTER, vector<module_t>{});
m_modules.emplace(alignment::RIGHT, vector<module_t>{});
const bar_settings bar{m_bar->settings()};
string bs{m_conf.bar_section()};
size_t module_count = 0;
for (auto& block : m_modules) {
string bs{m_conf.bar_section()};
for (int i = 0; i < 3; i++) {
alignment align = static_cast<alignment>(i + 1);
string confkey;
switch (block.first) {
switch (align) {
case alignment::LEFT:
confkey = "modules-left";
break;
@ -363,49 +371,56 @@ class controller {
}
for (auto& module_name : string_util::split(m_conf.get<string>(bs, confkey, ""), ' ')) {
auto type = m_conf.get<string>("module/" + module_name, "type");
auto bar = m_bar->settings();
auto& modules = block.second;
try {
auto type = m_conf.get<string>("module/" + module_name, "type");
module_t module;
if (type == "internal/counter")
modules.emplace_back(new counter_module(bar, m_log, m_conf, module_name));
else if (type == "internal/backlight")
modules.emplace_back(new backlight_module(bar, m_log, m_conf, module_name));
else if (type == "internal/xbacklight")
modules.emplace_back(new xbacklight_module(bar, m_log, m_conf, module_name));
else if (type == "internal/battery")
modules.emplace_back(new battery_module(bar, m_log, m_conf, module_name));
else if (type == "internal/bspwm")
modules.emplace_back(new bspwm_module(bar, m_log, m_conf, module_name));
else if (type == "internal/cpu")
modules.emplace_back(new cpu_module(bar, m_log, m_conf, module_name));
else if (type == "internal/date")
modules.emplace_back(new date_module(bar, m_log, m_conf, module_name));
else if (type == "internal/memory")
modules.emplace_back(new memory_module(bar, m_log, m_conf, module_name));
else if (type == "internal/i3")
modules.emplace_back(new i3_module(bar, m_log, m_conf, module_name));
else if (type == "internal/mpd")
modules.emplace_back(new mpd_module(bar, m_log, m_conf, module_name));
else if (type == "internal/volume")
modules.emplace_back(new volume_module(bar, m_log, m_conf, module_name));
else if (type == "internal/network")
modules.emplace_back(new network_module(bar, m_log, m_conf, module_name));
else if (type == "custom/text")
modules.emplace_back(new text_module(bar, m_log, m_conf, module_name));
else if (type == "custom/script")
modules.emplace_back(new script_module(bar, m_log, m_conf, module_name));
else if (type == "custom/menu")
modules.emplace_back(new menu_module(bar, m_log, m_conf, module_name));
else
throw application_error("Unknown module: " + module_name);
if (type == "internal/counter")
module.reset(new counter_module(bar, m_log, m_conf, module_name));
else if (type == "internal/backlight")
module.reset(new backlight_module(bar, m_log, m_conf, module_name));
else if (type == "internal/xbacklight")
module.reset(new xbacklight_module(bar, m_log, m_conf, module_name));
else if (type == "internal/battery")
module.reset(new battery_module(bar, m_log, m_conf, module_name));
else if (type == "internal/bspwm")
module.reset(new bspwm_module(bar, m_log, m_conf, module_name));
else if (type == "internal/cpu")
module.reset(new cpu_module(bar, m_log, m_conf, module_name));
else if (type == "internal/date")
module.reset(new date_module(bar, m_log, m_conf, module_name));
else if (type == "internal/memory")
module.reset(new memory_module(bar, m_log, m_conf, module_name));
else if (type == "internal/i3")
module.reset(new i3_module(bar, m_log, m_conf, module_name));
else if (type == "internal/mpd")
module.reset(new mpd_module(bar, m_log, m_conf, module_name));
else if (type == "internal/volume")
module.reset(new volume_module(bar, m_log, m_conf, module_name));
else if (type == "internal/network")
module.reset(new network_module(bar, m_log, m_conf, module_name));
else if (type == "custom/text")
module.reset(new text_module(bar, m_log, m_conf, module_name));
else if (type == "custom/script")
module.reset(new script_module(bar, m_log, m_conf, module_name));
else if (type == "custom/menu")
module.reset(new menu_module(bar, m_log, m_conf, module_name));
else
throw application_error("Unknown module: " + module_name);
auto& module = modules.back();
module->set_update_cb(bind(&eventloop::enqueue, m_eventloop.get(),
eventloop::entry_t{static_cast<int>(event_type::UPDATE)}));
module->set_stop_cb(bind(&eventloop::enqueue, m_eventloop.get(),
eventloop::entry_t{static_cast<int>(event_type::CHECK)}));
module->set_writer(bind(&controller::on_module_update, this, std::placeholders::_1));
module->set_terminator(bind(&controller::on_module_stop, this, std::placeholders::_1));
module->setup();
module_count++;
m_eventloop->add_module(align, move(module));
module_count++;
} catch (const module_error& err) {
continue;
}
}
}
@ -413,115 +428,21 @@ class controller {
throw application_error("No modules created");
}
void on_module_update(string /* module_name */) {
if (!m_mutex.try_lock_for(50ms)) {
this_thread::yield();
return;
/**
* Callback for clicked bar actions
*/
void on_mouse_event(string input) {
eventloop::entry_t evt{static_cast<int>(event_type::INPUT)};
if (input.length() > sizeof(evt.data)) {
m_log.warn("Ignoring input event (size)");
} else {
snprintf(evt.data, sizeof(evt.data), "%s", input.c_str());
m_eventloop->enqueue(evt);
}
std::lock_guard<std::timed_mutex> guard(m_mutex, std::adopt_lock);
if (!m_running)
return;
if (!m_throttler->passthrough(m_throttle_strategy)) {
m_log.trace("controller: Update event throttled");
return;
}
string contents{""};
string separator{m_bar->settings().separator};
string padding_left(m_bar->settings().padding_left, ' ');
string padding_right(m_bar->settings().padding_right, ' ');
auto margin_left = m_bar->settings().module_margin_left;
auto margin_right = m_bar->settings().module_margin_right;
for (auto&& block : m_modules) {
string block_contents;
for (auto&& module : block.second) {
auto module_contents = module->contents();
if (module_contents.empty())
continue;
if (!block_contents.empty() && !separator.empty())
block_contents += separator;
if (!(block.first == alignment::LEFT && module == block.second.front()))
block_contents += string(margin_left, ' ');
block_contents += module->contents();
if (!(block.first == alignment::RIGHT && module == block.second.back()))
block_contents += string(margin_right, ' ');
}
if (block_contents.empty())
continue;
switch (block.first) {
case alignment::LEFT:
contents += "%{l}";
contents += padding_left;
break;
case alignment::CENTER:
contents += "%{c}";
break;
case alignment::RIGHT:
contents += "%{r}";
block_contents += padding_right;
break;
case alignment::NONE:
break;
}
block_contents = string_util::replace_all(block_contents, "B-}%{B#", "B#");
block_contents = string_util::replace_all(block_contents, "F-}%{F#", "F#");
block_contents = string_util::replace_all(block_contents, "T-}%{T", "T");
contents += string_util::replace_all(block_contents, "}%{", " ");
}
if (m_stdout)
std::cout << contents << std::endl;
else
m_bar->parse(contents);
}
void on_module_stop(string /* module_name */) {
if (!m_running)
return;
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
if (module->running())
return;
}
}
m_log.warn("No running modules, raising SIGTERM");
kill(getpid(), SIGTERM);
}
void on_module_click(string input) {
if (!m_clickmtx.try_lock()) {
this_thread::yield();
return;
}
std::lock_guard<std::mutex> guard(m_clickmtx, std::adopt_lock);
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
if (!module->receive_events())
continue;
if (module->handle_event(input))
return;
}
}
m_clickmtx.unlock();
void on_unrecognized_action(string input) {
try {
if (m_command) {
m_log.warn("Terminating previous shell command");
@ -538,31 +459,96 @@ class controller {
}
}
void on_update() {
string contents{""};
string separator{m_bar->settings().separator};
string padding_left(m_bar->settings().padding_left, ' ');
string padding_right(m_bar->settings().padding_right, ' ');
auto margin_left = m_bar->settings().module_margin_left;
auto margin_right = m_bar->settings().module_margin_right;
for (const auto& block : m_eventloop->modules()) {
string block_contents;
bool is_left = false;
bool is_center = false;
bool is_right = false;
if (block.first == alignment::LEFT)
is_left = true;
else if (block.first == alignment::CENTER)
is_center = true;
else if (block.first == alignment::RIGHT)
is_right = true;
for (const auto& module : block.second) {
auto module_contents = module->contents();
if (module_contents.empty())
continue;
if (!block_contents.empty() && !separator.empty())
block_contents += separator;
if (!(is_left && module == block.second.front()))
block_contents += string(margin_left, ' ');
block_contents += module->contents();
if (!(is_right && module == block.second.back()))
block_contents += string(margin_right, ' ');
}
if (block_contents.empty())
continue;
if (is_left) {
contents += "%{l}";
contents += padding_left;
} else if (is_center) {
contents += "%{c}";
} else if (is_right) {
contents += "%{r}";
block_contents += padding_right;
}
block_contents = string_util::replace_all(block_contents, "B-}%{B#", "B#");
block_contents = string_util::replace_all(block_contents, "F-}%{F#", "F#");
block_contents = string_util::replace_all(block_contents, "T-}%{T", "T");
contents += string_util::replace_all(block_contents, "}%{", " ");
}
if (m_writeback) {
std::cout << contents << std::endl;
} else {
m_bar->parse(contents);
}
}
private:
connection& m_connection;
registry m_registry{m_connection};
const logger& m_log;
const config& m_conf;
unique_ptr<eventloop> m_eventloop;
unique_ptr<bar> m_bar;
unique_ptr<traymanager> m_traymanager;
std::timed_mutex m_mutex;
std::mutex m_clickmtx;
stateflag m_stdout{false};
stateflag m_running{false};
stateflag m_reload{false};
stateflag m_waiting{false};
sigset_t m_waitmask;
sigset_t m_ignmask;
inotify_watch_t& m_confwatch;
vector<thread> m_threads;
map<alignment, vector<module_t>> m_modules;
command_util::command_t m_command;
unique_ptr<throttle_util::event_throttler> m_throttler;
throttle_util::strategy::try_once_or_leave_yolo m_throttle_strategy;
bool m_writeback = false;
};
namespace {
@ -577,6 +563,7 @@ namespace {
configure_connection(),
configure_logger(),
configure_config(),
configure_eventloop(),
configure_bar(),
configure_traymanager());
// clang-format on

287
include/components/eventloop.hpp Normal file
View File

@ -0,0 +1,287 @@
#pragma once
#include <moodycamel/blockingconcurrentqueue.h>
#include "common.hpp"
#include "components/bar.hpp"
#include "components/logger.hpp"
#include "modules/meta.hpp"
#include "utils/command.hpp"
#include "utils/string.hpp"
LEMONBUDDY_NS
using module_t = unique_ptr<modules::module_interface>;
using modulemap_t = map<alignment, vector<module_t>>;
enum class event_type { NONE = 0, UPDATE, CHECK, INPUT, QUIT };
struct event {
int type;
char data[256]{'\0'};
};
class eventloop {
public:
/**
* Queue type
*/
using entry_t = event;
using queue_t = moodycamel::BlockingConcurrentQueue<entry_t>;
/**
* Construct eventloop
*/
explicit eventloop(const logger& logger) : m_log(logger) {}
/**
* Deconstruct eventloop
*/
~eventloop() noexcept {
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
auto module_name = module->name();
auto cleanup_ms = time_execution([&module] {
module->stop();
module.reset();
});
m_log.trace("eventloop: Deconstruction of %s took %lu ms.", module_name, cleanup_ms);
}
}
}
/**
* Set callback handler for UPDATE events
*/
void set_update_cb(callback<>&& cb) {
m_update_cb = forward<decltype(cb)>(cb);
}
/**
* Set callback handler for raw INPUT events
*/
void set_input_db(callback<string>&& cb) {
m_unrecognized_input_cb = forward<decltype(cb)>(cb);
}
/**
* Get reference to module map
*/
modulemap_t& modules() {
return m_modules;
}
/**
* Add module to alignment block
*/
void add_module(const alignment pos, module_t&& module) {
modulemap_t::iterator it = m_modules.lower_bound(pos);
if (it != m_modules.end() && !(m_modules.key_comp()(pos, it->first))) {
it->second.emplace_back(forward<module_t>(module));
} else {
vector<module_t> vec;
vec.emplace_back(forward<module_t>(module));
m_modules.insert(it, modulemap_t::value_type(pos, move(vec)));
}
}
/**
* Enqueue event
*/
bool enqueue(const entry_t& i) {
bool enqueued;
if ((enqueued = m_queue.enqueue(i)) == false) {
m_log.warn("Failed to queue event (%d)", i.type);
}
return enqueued;
}
/**
* Start module threads and wait for events on the queue
*
* @param timeframe Time to wait for subsequent events
* @param limit Maximum amount of subsequent events to swallow within timeframe
*/
template <typename Rep, typename Period>
void run(chrono::duration<Rep, Period> timeframe, int limit) {
m_log.info("Starting event loop");
m_running = true;
// Start module threads
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
try {
m_log.info("Starting %s", module->name());
module->start();
} catch (const application_error& err) {
m_log.err("Failed to start '%s' (reason: %s)", module->name(), err.what());
}
}
}
m_log.trace("eventloop: Enter loop");
while (m_running) {
entry_t evt, next{static_cast<int>(event_type::NONE)};
m_queue.wait_dequeue(evt);
if (!m_running) {
break;
}
if (match_event(evt, event_type::UPDATE)) {
int swallowed = 0;
while (swallowed++ < limit && m_queue.wait_dequeue_timed(next, timeframe)) {
if (match_event(next, event_type::QUIT)) {
evt = next;
break;
} else if (compare_events(evt, next)) {
m_log.trace("eventloop: Swallowing event within timeframe");
evt = next;
} else {
break;
}
}
}
forward_event(evt);
if (match_event(next, event_type::NONE))
continue;
if (compare_events(evt, next))
continue;
forward_event(next);
}
m_log.trace("eventloop: Loop ended");
}
/**
* Stop main loop by enqueuing a QUIT event
*/
void stop() {
m_log.info("Stopping event loop");
m_running = false;
enqueue({static_cast<int>(event_type::QUIT)});
}
protected:
/**
* Test if event matches given type
*/
bool match_event(entry_t evt, event_type type) {
return static_cast<int>(type) == evt.type;
}
/**
* Compare given events
*/
bool compare_events(entry_t evt, entry_t evt2) {
return evt.type == evt2.type;
}
/**
* Forward event to handler based on type
*/
void forward_event(entry_t evt) {
if (evt.type == static_cast<int>(event_type::UPDATE)) {
on_update();
} else if (evt.type == static_cast<int>(event_type::INPUT)) {
on_input(string{evt.data});
} else if (evt.type == static_cast<int>(event_type::CHECK)) {
on_check();
} else if (evt.type == static_cast<int>(event_type::QUIT)) {
on_quit();
} else {
m_log.warn("Unknown event type for enqueued event (%d)", evt.type);
}
}
/**
* Handler for enqueued UPDATE events
*/
void on_update() {
m_log.trace("eventloop: Received UPDATE event");
if (m_update_cb) {
m_update_cb();
} else {
m_log.warn("No callback to handle update");
}
}
/**
* Handler for enqueued INPUT events
*/
void on_input(string input) {
m_log.trace("eventloop: Received INPUT event");
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
if (!module->receive_events())
continue;
if (module->handle_event(input)) {
return;
}
}
}
if (m_unrecognized_input_cb) {
m_unrecognized_input_cb(input);
} else {
m_log.warn("No callback to handle unrecognized input");
}
}
/**
* Handler for enqueued CHECK events
*/
void on_check() {
if (!m_running) {
return;
}
for (auto&& block : m_modules) {
for (auto&& module : block.second) {
if (module->running())
return;
}
}
m_log.warn("No running modules...");
stop();
}
/**
* Handler for enqueued QUIT events
*/
void on_quit() {
m_log.trace("eventloop: Received QUIT event");
m_running = false;
}
private:
const logger& m_log;
queue_t m_queue;
modulemap_t m_modules;
stateflag m_running;
callback<> m_update_cb;
callback<string> m_unrecognized_input_cb;
};
namespace {
/**
* Configure injection module
*/
template <typename T = unique_ptr<eventloop>>
di::injector<T> configure_eventloop() {
return di::make_injector(configure_logger());
}
}
LEMONBUDDY_NS_END

2
include/modules/battery.hpp
View File

@ -178,7 +178,7 @@ namespace modules {
int i = 0;
const int poll_seconds = m_conf.get<float>(name(), "poll-interval", 3.0f) / dur.count();
while (enabled()) {
while (running()) {
// TODO(jaagr): Keep track of when the values were last read to determine
// if we need to trigger the event manually or not.
if (poll_seconds > 0 && (++i % poll_seconds) == 0) {

4
include/modules/bspwm.hpp
View File

@ -101,8 +101,10 @@ namespace modules {
}
void stop() {
if (m_subscriber)
if (m_subscriber) {
m_log.info("%s: Disconnecting from socket", name());
m_subscriber->disconnect();
}
event_module::stop();
}

185
include/modules/meta.hpp
View File

@ -101,14 +101,14 @@ namespace modules {
for (auto&& tag : string_util::split(format->value, ' ')) {
if (tag[0] != '<' || tag[tag.length() - 1] != '>')
continue;
if (std::find(format->tags.begin(), format->tags.end(), tag) != format->tags.end())
if (find(format->tags.begin(), format->tags.end(), tag) != format->tags.end())
continue;
if (std::find(whitelist.begin(), whitelist.end(), tag) != whitelist.end())
if (find(whitelist.begin(), whitelist.end(), tag) != whitelist.end())
continue;
throw undefined_format_tag("[" + m_modname + "] Undefined \"" + name + "\" tag: " + tag);
}
m_formats.insert(make_pair(name, std::move(format)));
m_formats.insert(make_pair(name, move(format)));
}
shared_ptr<module_format> get(string format_name) {
@ -158,8 +158,8 @@ namespace modules {
virtual bool handle_event(string cmd) = 0;
virtual bool receive_events() const = 0;
virtual void set_writer(std::function<void(string)>&& fn) = 0;
virtual void set_terminator(std::function<void(string)>&& fn) = 0;
virtual void set_update_cb(callback<>&& cb) = 0;
virtual void set_stop_cb(callback<>&& cb) = 0;
};
// }}}
@ -177,32 +177,23 @@ namespace modules {
, m_formatter(make_unique<module_formatter>(m_conf, m_name)) {}
~module() {
CAST_MOD(Impl)->stop();
m_log.trace("%s: Deconstructing", name());
m_updatelock.unlock();
assert(!running());
std::lock_guard<threading_util::spin_lock> lck(m_updatelock);
{
if (m_broadcast_thread.joinable())
m_broadcast_thread.join();
for (auto&& thread_ : m_threads) {
if (thread_.joinable())
thread_.join();
for (auto&& thread_ : m_threads) {
if (thread_.joinable()) {
thread_.join();
}
m_threads.clear();
}
m_log.trace("%s: Done cleaning up", name());
}
void set_writer(std::function<void(string)>&& fn) {
m_writer = forward<decltype(fn)>(fn);
void set_update_cb(callback<>&& cb) {
m_update_callback = forward<decltype(cb)>(cb);
}
void set_terminator(std::function<void(string)>&& fn) {
m_terminator = forward<decltype(fn)>(fn);
void set_stop_cb(callback<>&& cb) {
m_stop_callback = forward<decltype(cb)>(cb);
}
string name() const {
@ -210,36 +201,42 @@ namespace modules {
}
bool running() const {
return CONST_MOD(Impl).enabled();
return m_enabled.load(std::memory_order_relaxed);
}
void setup() {
m_log.trace("%s: Setup", name());
m_log.trace("%s: Setup", m_name);
try {
CAST_MOD(Impl)->setup();
} catch (const module_error& err) {
m_log.err("%s: Setup failed", name());
CAST_MOD(Impl)->halt(err.what());
} catch (const std::exception& err) {
m_log.err("%s: Setup failed", name());
CAST_MOD(Impl)->halt(err.what());
m_log.err("%s: Setup failed", m_name);
halt(err.what());
}
}
void stop() {
if (!enabled())
if (!running()) {
return;
std::unique_lock<threading_util::spin_lock> lck(m_updatelock);
{
enable(false);
CAST_MOD(Impl)->teardown();
m_log.trace("%s: Stop", name());
}
if (m_terminator)
m_terminator(name());
m_log.info("%s: Stopping", name());
m_enabled.store(false, std::memory_order_relaxed);
wakeup();
std::lock_guard<threading_util::spin_lock> guard(m_lock);
{
CAST_MOD(Impl)->teardown();
if (m_mainthread.joinable()) {
m_mainthread.join();
}
}
if (m_stop_callback) {
m_stop_callback();
}
}
void halt(string error_message) {
@ -248,9 +245,7 @@ namespace modules {
stop();
}
void teardown() {
CAST_MOD(Impl)->wakeup();
}
void teardown() {}
string contents() {
return m_cache;
@ -265,22 +260,15 @@ namespace modules {
}
protected:
bool enabled() const {
return m_enabled;
}
void enable(bool state) {
m_enabled = state;
}
void broadcast() {
if (!enabled())
if (!running()) {
return;
}
m_cache = CAST_MOD(Impl)->get_output();
if (m_writer)
m_writer(name());
if (m_update_callback)
m_update_callback();
else
m_log.warn("%s: No handler, ignoring broadcast...", name());
}
@ -302,7 +290,7 @@ namespace modules {
}
string get_output() {
if (!enabled()) {
if (!running()) {
m_log.trace("%s: Module is disabled", name());
return "";
}
@ -334,11 +322,10 @@ namespace modules {
}
protected:
function<void(string)> m_writer;
function<void(string)> m_terminator;
callback<> m_update_callback;
callback<> m_stop_callback;
threading_util::spin_lock m_updatelock;
// std::timed_mutex m_mutex;
threading_util::spin_lock m_lock;
const bar_settings m_bar;
const logger& m_log;
@ -351,11 +338,11 @@ namespace modules {
unique_ptr<builder> m_builder;
unique_ptr<module_formatter> m_formatter;
vector<thread> m_threads;
thread m_mainthread;
private:
stateflag m_enabled{false};
stateflag m_enabled{true};
string m_cache;
thread m_broadcast_thread;
};
// }}}
@ -368,8 +355,6 @@ namespace modules {
using module<Impl>::module;
void start() {
CAST_MOD(Impl)->enable(true);
CAST_MOD(Impl)->setup();
CAST_MOD(Impl)->broadcast();
}
@ -389,8 +374,7 @@ namespace modules {
using module<Impl>::module;
void start() {
CAST_MOD(Impl)->enable(true);
CAST_MOD(Impl)->m_threads.emplace_back(thread(&timer_module::runner, this));
CAST_MOD(Impl)->m_mainthread = thread(&timer_module::runner, this);
}
protected:
@ -398,11 +382,9 @@ namespace modules {
void runner() {
try {
CAST_MOD(Impl)->setup();
while (CONST_MOD(Impl).enabled()) {
while (CONST_MOD(Impl).running()) {
std::lock_guard<threading_util::spin_lock> guard(this->m_lock);
{
std::lock_guard<threading_util::spin_lock> lck(this->m_updatelock);
if (CAST_MOD(Impl)->update())
CAST_MOD(Impl)->broadcast();
}
@ -425,34 +407,36 @@ namespace modules {
using module<Impl>::module;
void start() {
CAST_MOD(Impl)->enable(true);
CAST_MOD(Impl)->m_threads.emplace_back(thread(&event_module::runner, this));
CAST_MOD(Impl)->m_mainthread = thread(&event_module::runner, this);
}
protected:
void runner() {
try {
CAST_MOD(Impl)->setup();
// Send initial broadcast to warmup cache
if (CONST_MOD(Impl).running()) {
CAST_MOD(Impl)->update();
CAST_MOD(Impl)->broadcast();
}
// warmup
CAST_MOD(Impl)->update();
CAST_MOD(Impl)->broadcast();
while (CONST_MOD(Impl).enabled()) {
while (CONST_MOD(Impl).running()) {
CAST_MOD(Impl)->idle();
if (!CONST_MOD(Impl).enabled())
if (!CONST_MOD(Impl).running())
break;
std::lock_guard<threading_util::spin_lock> lck(this->m_updatelock);
std::lock_guard<threading_util::spin_lock> guard(this->m_lock);
{
if (!CAST_MOD(Impl)->has_event())
continue;
else if (!CAST_MOD(Impl)->update())
if (!CONST_MOD(Impl).running())
break;
if (!CAST_MOD(Impl)->update())
continue;
else
CAST_MOD(Impl)->broadcast();
}
if (CONST_MOD(Impl).running())
CAST_MOD(Impl)->broadcast();
}
} catch (const module_error& err) {
CAST_MOD(Impl)->halt(err.what());
@ -471,18 +455,19 @@ namespace modules {
using module<Impl>::module;
void start() {
CAST_MOD(Impl)->enable(true);
CAST_MOD(Impl)->m_threads.emplace_back(thread(&inotify_module::runner, this));
CAST_MOD(Impl)->m_mainthread = thread(&inotify_module::runner, this);
}
protected:
void runner() {
try {
CAST_MOD(Impl)->setup();
CAST_MOD(Impl)->on_event(nullptr); // warmup
CAST_MOD(Impl)->broadcast();
// Send initial broadcast to warmup cache
if (CONST_MOD(Impl).running()) {
CAST_MOD(Impl)->on_event(nullptr);
CAST_MOD(Impl)->broadcast();
}
while (CAST_MOD(Impl)->enabled()) {
while (CONST_MOD(Impl).running()) {
CAST_MOD(Impl)->poll_events();
}
} catch (const module_error& err) {
@ -517,22 +502,28 @@ namespace modules {
return;
}
while (CONST_MOD(Impl).enabled()) {
for (auto&& w : watches) {
this->m_log.trace_x("%s: Poll inotify watch %s", CONST_MOD(Impl).name(), w->path());
std::lock_guard<threading_util::spin_lock> lck(this->m_updatelock);
while (CONST_MOD(Impl).running()) {
std::unique_lock<threading_util::spin_lock> guard(this->m_lock);
{
for (auto&& w : watches) {
this->m_log.trace_x("%s: Poll inotify watch %s", CONST_MOD(Impl).name(), w->path());
if (w->poll(1000 / watches.size())) {
auto event = w->get_event();
if (w->poll(1000 / watches.size())) {
auto event = w->get_event();
w->remove();
w->remove();
if (CAST_MOD(Impl)->on_event(event.get()))
CAST_MOD(Impl)->broadcast();
if (CAST_MOD(Impl)->on_event(event.get()))
CAST_MOD(Impl)->broadcast();
return;
return;
}
if (!CONST_MOD(Impl).running())
break;
}
}
guard.unlock();
CAST_MOD(Impl)->idle();
}
}

1
include/modules/mpd.hpp
View File

@ -83,7 +83,6 @@ namespace modules {
}
void teardown() {
wakeup();
m_mpd.reset();
}

2
include/modules/network.hpp
View File

@ -165,7 +165,7 @@ namespace modules {
const chrono::milliseconds framerate{m_animation_packetloss->framerate()};
const auto dur = chrono::duration<double>(framerate);
while (enabled()) {
while (running()) {
if (m_connected && m_packetloss)
broadcast();
sleep(dur);

3
include/modules/script.hpp
View File

@ -56,9 +56,6 @@ namespace modules {
if (!m_tail)
return true;
if (!enabled())
return false;
try {
if (!m_command || !m_command->is_running()) {
auto exec = string_util::replace_all(m_exec, "%counter%", to_string(++m_counter));

7
include/utils/string.hpp
View File

@ -171,10 +171,15 @@ namespace string_util {
return static_cast<const stringstream&>(os).str();
}
/**
* Hash type
*/
using hash_type = unsigned long;
/**
* Compute string hash
*/
inline auto hash(string src) {
inline hash_type hash(string src) {
return std::hash<string>()(src);
}
}

981
lib/concurrentqueue/include/moodycamel/blockingconcurrentqueue.h Normal file
View File

@ -0,0 +1,981 @@
// Provides an efficient blocking version of moodycamel::ConcurrentQueue.
// ©2015-2016 Cameron Desrochers. Distributed under the terms of the simplified
// BSD license, available at the top of concurrentqueue.h.
// Uses Jeff Preshing's semaphore implementation (under the terms of its
// separate zlib license, embedded below).
#pragma once
#include "concurrentqueue.h"
#include <type_traits>
#include <cerrno>
#include <memory>
#include <chrono>
#include <ctime>
#if defined(_WIN32)
// Avoid including windows.h in a header; we only need a handful of
// items, so we'll redeclare them here (this is relatively safe since
// the API generally has to remain stable between Windows versions).
// I know this is an ugly hack but it still beats polluting the global
// namespace with thousands of generic names or adding a .cpp for nothing.
extern "C" {
struct _SECURITY_ATTRIBUTES;
__declspec(dllimport) void* __stdcall CreateSemaphoreW(_SECURITY_ATTRIBUTES* lpSemaphoreAttributes, long lInitialCount, long lMaximumCount, const wchar_t* lpName);
__declspec(dllimport) int __stdcall CloseHandle(void* hObject);
__declspec(dllimport) unsigned long __stdcall WaitForSingleObject(void* hHandle, unsigned long dwMilliseconds);
__declspec(dllimport) int __stdcall ReleaseSemaphore(void* hSemaphore, long lReleaseCount, long* lpPreviousCount);
}
#elif defined(__MACH__)
#include <mach/mach.h>
#elif defined(__unix__)
#include <semaphore.h>
#endif
namespace moodycamel
{
namespace details
{
// Code in the mpmc_sema namespace below is an adaptation of Jeff Preshing's
// portable + lightweight semaphore implementations, originally from
// https://github.com/preshing/cpp11-on-multicore/blob/master/common/sema.h
// LICENSE:
// Copyright (c) 2015 Jeff Preshing
//
// This software is provided 'as-is', without any express or implied
// warranty. In no event will the authors be held liable for any damages
// arising from the use of this software.
//
// Permission is granted to anyone to use this software for any purpose,
// including commercial applications, and to alter it and redistribute it
// freely, subject to the following restrictions:
//
// 1. The origin of this software must not be misrepresented; you must not
// claim that you wrote the original software. If you use this software
// in a product, an acknowledgement in the product documentation would be
// appreciated but is not required.
// 2. Altered source versions must be plainly marked as such, and must not be
// misrepresented as being the original software.
// 3. This notice may not be removed or altered from any source distribution.
namespace mpmc_sema
{
#if defined(_WIN32)
class Semaphore
{
private:
void* m_hSema;
Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
public:
Semaphore(int initialCount = 0)
{
assert(initialCount >= 0);
const long maxLong = 0x7fffffff;
m_hSema = CreateSemaphoreW(nullptr, initialCount, maxLong, nullptr);
}
~Semaphore()
{
CloseHandle(m_hSema);
}
void wait()
{
const unsigned long infinite = 0xffffffff;
WaitForSingleObject(m_hSema, infinite);
}
bool try_wait()
{
const unsigned long RC_WAIT_TIMEOUT = 0x00000102;
return WaitForSingleObject(m_hSema, 0) != RC_WAIT_TIMEOUT;
}
bool timed_wait(std::uint64_t usecs)
{
const unsigned long RC_WAIT_TIMEOUT = 0x00000102;
return WaitForSingleObject(m_hSema, (unsigned long)(usecs / 1000)) != RC_WAIT_TIMEOUT;
}
void signal(int count = 1)
{
ReleaseSemaphore(m_hSema, count, nullptr);
}
};
#elif defined(__MACH__)
//---------------------------------------------------------
// Semaphore (Apple iOS and OSX)
// Can't use POSIX semaphores due to http://lists.apple.com/archives/darwin-kernel/2009/Apr/msg00010.html
//---------------------------------------------------------
class Semaphore
{
private:
semaphore_t m_sema;
Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
public:
Semaphore(int initialCount = 0)
{
assert(initialCount >= 0);
semaphore_create(mach_task_self(), &m_sema, SYNC_POLICY_FIFO, initialCount);
}
~Semaphore()
{
semaphore_destroy(mach_task_self(), m_sema);
}
void wait()
{
semaphore_wait(m_sema);
}
bool try_wait()
{
return timed_wait(0);
}
bool timed_wait(std::uint64_t timeout_usecs)
{
mach_timespec_t ts;
ts.tv_sec = timeout_usecs / 1000000;
ts.tv_nsec = (timeout_usecs % 1000000) * 1000;
// added in OSX 10.10: https://developer.apple.com/library/prerelease/mac/documentation/General/Reference/APIDiffsMacOSX10_10SeedDiff/modules/Darwin.html
kern_return_t rc = semaphore_timedwait(m_sema, ts);
return rc != KERN_OPERATION_TIMED_OUT;
}
void signal()
{
semaphore_signal(m_sema);
}
void signal(int count)
{
while (count-- > 0)
{
semaphore_signal(m_sema);
}
}
};
#elif defined(__unix__)
//---------------------------------------------------------
// Semaphore (POSIX, Linux)
//---------------------------------------------------------
class Semaphore
{
private:
sem_t m_sema;
Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION;
public:
Semaphore(int initialCount = 0)
{
assert(initialCount >= 0);
sem_init(&m_sema, 0, initialCount);
}
~Semaphore()
{
sem_destroy(&m_sema);
}
void wait()
{
// http://stackoverflow.com/questions/2013181/gdb-causes-sem-wait-to-fail-with-eintr-error
int rc;
do {
rc = sem_wait(&m_sema);
} while (rc == -1 && errno == EINTR);
}
bool try_wait()
{
int rc;
do {
rc = sem_trywait(&m_sema);
} while (rc == -1 && errno == EINTR);
return !(rc == -1 && errno == EAGAIN);
}
bool timed_wait(std::uint64_t usecs)
{
struct timespec ts;
const int usecs_in_1_sec = 1000000;
const int nsecs_in_1_sec = 1000000000;
clock_gettime(CLOCK_REALTIME, &ts);
ts.tv_sec += usecs / usecs_in_1_sec;
ts.tv_nsec += (usecs % usecs_in_1_sec) * 1000;
// sem_timedwait bombs if you have more than 1e9 in tv_nsec
// so we have to clean things up before passing it in
if (ts.tv_nsec > nsecs_in_1_sec) {
ts.tv_nsec -= nsecs_in_1_sec;
++ts.tv_sec;
}
int rc;
do {
rc = sem_timedwait(&m_sema, &ts);
} while (rc == -1 && errno == EINTR);
return !(rc == -1 && errno == ETIMEDOUT);
}
void signal()
{
sem_post(&m_sema);
}
void signal(int count)
{
while (count-- > 0)
{
sem_post(&m_sema);
}
}
};
#else
#error Unsupported platform! (No semaphore wrapper available)
#endif
//---------------------------------------------------------
// LightweightSemaphore
//---------------------------------------------------------
class LightweightSemaphore
{
public:
typedef std::make_signed<std::size_t>::type ssize_t;
private:
std::atomic<ssize_t> m_count;
Semaphore m_sema;
bool waitWithPartialSpinning(std::int64_t timeout_usecs = -1)
{
ssize_t oldCount;
// Is there a better way to set the initial spin count?
// If we lower it to 1000, testBenaphore becomes 15x slower on my Core i7-5930K Windows PC,
// as threads start hitting the kernel semaphore.
int spin = 10000;
while (--spin >= 0)
{
oldCount = m_count.load(std::memory_order_relaxed);
if ((oldCount > 0) && m_count.compare_exchange_strong(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed))
return true;
std::atomic_signal_fence(std::memory_order_acquire); // Prevent the compiler from collapsing the loop.
}
oldCount = m_count.fetch_sub(1, std::memory_order_acquire);
if (oldCount > 0)
return true;
if (timeout_usecs < 0)
{
m_sema.wait();
return true;
}
if (m_sema.timed_wait((std::uint64_t)timeout_usecs))
return true;
// At this point, we've timed out waiting for the semaphore, but the
// count is still decremented indicating we may still be waiting on
// it. So we have to re-adjust the count, but only if the semaphore
// wasn't signaled enough times for us too since then. If it was, we
// need to release the semaphore too.
while (true)
{
oldCount = m_count.load(std::memory_order_acquire);
if (oldCount >= 0 && m_sema.try_wait())
return true;
if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed))
return false;
}
}
ssize_t waitManyWithPartialSpinning(ssize_t max, std::int64_t timeout_usecs = -1)
{
assert(max > 0);
ssize_t oldCount;
int spin = 10000;
while (--spin >= 0)
{
oldCount = m_count.load(std::memory_order_relaxed);
if (oldCount > 0)
{
ssize_t newCount = oldCount > max ? oldCount - max : 0;
if (m_count.compare_exchange_strong(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed))
return oldCount - newCount;
}
std::atomic_signal_fence(std::memory_order_acquire);
}
oldCount = m_count.fetch_sub(1, std::memory_order_acquire);
if (oldCount <= 0)
{
if (timeout_usecs < 0)
m_sema.wait();
else if (!m_sema.timed_wait((std::uint64_t)timeout_usecs))
{
while (true)
{
oldCount = m_count.load(std::memory_order_acquire);
if (oldCount >= 0 && m_sema.try_wait())
break;
if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed))
return 0;
}
}
}
if (max > 1)
return 1 + tryWaitMany(max - 1);
return 1;
}
public:
LightweightSemaphore(ssize_t initialCount = 0) : m_count(initialCount)
{
assert(initialCount >= 0);
}
bool tryWait()
{
ssize_t oldCount = m_count.load(std::memory_order_relaxed);
while (oldCount > 0)
{
if (m_count.compare_exchange_weak(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed))
return true;
}
return false;
}
void wait()
{
if (!tryWait())
waitWithPartialSpinning();
}
bool wait(std::int64_t timeout_usecs)
{
return tryWait() || waitWithPartialSpinning(timeout_usecs);
}
// Acquires between 0 and (greedily) max, inclusive
ssize_t tryWaitMany(ssize_t max)
{
assert(max >= 0);
ssize_t oldCount = m_count.load(std::memory_order_relaxed);
while (oldCount > 0)
{
ssize_t newCount = oldCount > max ? oldCount - max : 0;
if (m_count.compare_exchange_weak(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed))
return oldCount - newCount;
}
return 0;
}
// Acquires at least one, and (greedily) at most max
ssize_t waitMany(ssize_t max, std::int64_t timeout_usecs)
{
assert(max >= 0);
ssize_t result = tryWaitMany(max);
if (result == 0 && max > 0)
result = waitManyWithPartialSpinning(max, timeout_usecs);
return result;
}
ssize_t waitMany(ssize_t max)
{
ssize_t result = waitMany(max, -1);
assert(result > 0);
return result;
}
void signal(ssize_t count = 1)
{
assert(count >= 0);
ssize_t oldCount = m_count.fetch_add(count, std::memory_order_release);
ssize_t toRelease = -oldCount < count ? -oldCount : count;
if (toRelease > 0)
{
m_sema.signal((int)toRelease);
}
}
ssize_t availableApprox() const
{
ssize_t count = m_count.load(std::memory_order_relaxed);
return count > 0 ? count : 0;
}
};
} // end namespace mpmc_sema
} // end namespace details
// This is a blocking version of the queue. It has an almost identical interface to
// the normal non-blocking version, with the addition of various wait_dequeue() methods
// and the removal of producer-specific dequeue methods.
template<typename T, typename Traits = ConcurrentQueueDefaultTraits>
class BlockingConcurrentQueue
{
private:
typedef ::moodycamel::ConcurrentQueue<T, Traits> ConcurrentQueue;
typedef details::mpmc_sema::LightweightSemaphore LightweightSemaphore;
public:
typedef typename ConcurrentQueue::producer_token_t producer_token_t;
typedef typename ConcurrentQueue::consumer_token_t consumer_token_t;
typedef typename ConcurrentQueue::index_t index_t;
typedef typename ConcurrentQueue::size_t size_t;
typedef typename std::make_signed<size_t>::type ssize_t;
static const size_t BLOCK_SIZE = ConcurrentQueue::BLOCK_SIZE;
static const size_t EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD = ConcurrentQueue::EXPLICIT_BLOCK_EMPTY_COUNTER_THRESHOLD;
static const size_t EXPLICIT_INITIAL_INDEX_SIZE = ConcurrentQueue::EXPLICIT_INITIAL_INDEX_SIZE;
static const size_t IMPLICIT_INITIAL_INDEX_SIZE = ConcurrentQueue::IMPLICIT_INITIAL_INDEX_SIZE;
static const size_t INITIAL_IMPLICIT_PRODUCER_HASH_SIZE = ConcurrentQueue::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE;
static const std::uint32_t EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE = ConcurrentQueue::EXPLICIT_CONSUMER_CONSUMPTION_QUOTA_BEFORE_ROTATE;
static const size_t MAX_SUBQUEUE_SIZE = ConcurrentQueue::MAX_SUBQUEUE_SIZE;
public:
// Creates a queue with at least `capacity` element slots; note that the
// actual number of elements that can be inserted without additional memory
// allocation depends on the number of producers and the block size (e.g. if
// the block size is equal to `capacity`, only a single block will be allocated
// up-front, which means only a single producer will be able to enqueue elements
// without an extra allocation -- blocks aren't shared between producers).
// This method is not thread safe -- it is up to the user to ensure that the
// queue is fully constructed before it starts being used by other threads (this
// includes making the memory effects of construction visible, possibly with a
// memory barrier).
explicit BlockingConcurrentQueue(size_t capacity = 6 * BLOCK_SIZE)
: inner(capacity), sema(create<LightweightSemaphore>(), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
{
assert(reinterpret_cast<ConcurrentQueue*>((BlockingConcurrentQueue*)1) == &((BlockingConcurrentQueue*)1)->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
if (!sema) {
MOODYCAMEL_THROW(std::bad_alloc());
}
}
BlockingConcurrentQueue(size_t minCapacity, size_t maxExplicitProducers, size_t maxImplicitProducers)
: inner(minCapacity, maxExplicitProducers, maxImplicitProducers), sema(create<LightweightSemaphore>(), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
{
assert(reinterpret_cast<ConcurrentQueue*>((BlockingConcurrentQueue*)1) == &((BlockingConcurrentQueue*)1)->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
if (!sema) {
MOODYCAMEL_THROW(std::bad_alloc());
}
}
// Disable copying and copy assignment
BlockingConcurrentQueue(BlockingConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
BlockingConcurrentQueue& operator=(BlockingConcurrentQueue const&) MOODYCAMEL_DELETE_FUNCTION;
// Moving is supported, but note that it is *not* a thread-safe operation.
// Nobody can use the queue while it's being moved, and the memory effects
// of that move must be propagated to other threads before they can use it.
// Note: When a queue is moved, its tokens are still valid but can only be
// used with the destination queue (i.e. semantically they are moved along
// with the queue itself).
BlockingConcurrentQueue(BlockingConcurrentQueue&& other) MOODYCAMEL_NOEXCEPT
: inner(std::move(other.inner)), sema(std::move(other.sema))
{ }
inline BlockingConcurrentQueue& operator=(BlockingConcurrentQueue&& other) MOODYCAMEL_NOEXCEPT
{
return swap_internal(other);
}
// Swaps this queue's state with the other's. Not thread-safe.
// Swapping two queues does not invalidate their tokens, however
// the tokens that were created for one queue must be used with
// only the swapped queue (i.e. the tokens are tied to the
// queue's movable state, not the object itself).
inline void swap(BlockingConcurrentQueue& other) MOODYCAMEL_NOEXCEPT
{
swap_internal(other);
}
private:
BlockingConcurrentQueue& swap_internal(BlockingConcurrentQueue& other)
{
if (this == &other) {
return *this;
}
inner.swap(other.inner);
sema.swap(other.sema);
return *this;
}
public:
// Enqueues a single item (by copying it).
// Allocates memory if required. Only fails if memory allocation fails (or implicit
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0,
// or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
// Thread-safe.
inline bool enqueue(T const& item)
{
if (details::likely(inner.enqueue(item))) {
sema->signal();
return true;
}
return false;
}
// Enqueues a single item (by moving it, if possible).
// Allocates memory if required. Only fails if memory allocation fails (or implicit
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0,
// or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
// Thread-safe.
inline bool enqueue(T&& item)
{
if (details::likely(inner.enqueue(std::move(item)))) {
sema->signal();
return true;
}
return false;
}
// Enqueues a single item (by copying it) using an explicit producer token.
// Allocates memory if required. Only fails if memory allocation fails (or
// Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
// Thread-safe.
inline bool enqueue(producer_token_t const& token, T const& item)
{
if (details::likely(inner.enqueue(token, item))) {
sema->signal();
return true;
}
return false;
}
// Enqueues a single item (by moving it, if possible) using an explicit producer token.
// Allocates memory if required. Only fails if memory allocation fails (or
// Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
// Thread-safe.
inline bool enqueue(producer_token_t const& token, T&& item)
{
if (details::likely(inner.enqueue(token, std::move(item)))) {
sema->signal();
return true;
}
return false;
}
// Enqueues several items.
// Allocates memory if required. Only fails if memory allocation fails (or
// implicit production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE
// is 0, or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
// Note: Use std::make_move_iterator if the elements should be moved instead of copied.
// Thread-safe.
template<typename It>
inline bool enqueue_bulk(It itemFirst, size_t count)
{
if (details::likely(inner.enqueue_bulk(std::forward<It>(itemFirst), count))) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
return true;
}
return false;
}
// Enqueues several items using an explicit producer token.
// Allocates memory if required. Only fails if memory allocation fails
// (or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
// Note: Use std::make_move_iterator if the elements should be moved
// instead of copied.
// Thread-safe.
template<typename It>
inline bool enqueue_bulk(producer_token_t const& token, It itemFirst, size_t count)
{
if (details::likely(inner.enqueue_bulk(token, std::forward<It>(itemFirst), count))) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
return true;
}
return false;
}
// Enqueues a single item (by copying it).
// Does not allocate memory. Fails if not enough room to enqueue (or implicit
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE
// is 0).
// Thread-safe.
inline bool try_enqueue(T const& item)
{
if (inner.try_enqueue(item)) {
sema->signal();
return true;
}
return false;
}
// Enqueues a single item (by moving it, if possible).
// Does not allocate memory (except for one-time implicit producer).
// Fails if not enough room to enqueue (or implicit production is
// disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0).
// Thread-safe.
inline bool try_enqueue(T&& item)
{
if (inner.try_enqueue(std::move(item))) {
sema->signal();
return true;
}
return false;
}
// Enqueues a single item (by copying it) using an explicit producer token.
// Does not allocate memory. Fails if not enough room to enqueue.
// Thread-safe.
inline bool try_enqueue(producer_token_t const& token, T const& item)
{
if (inner.try_enqueue(token, item)) {
sema->signal();
return true;
}
return false;
}
// Enqueues a single item (by moving it, if possible) using an explicit producer token.
// Does not allocate memory. Fails if not enough room to enqueue.
// Thread-safe.
inline bool try_enqueue(producer_token_t const& token, T&& item)
{
if (inner.try_enqueue(token, std::move(item))) {
sema->signal();
return true;
}
return false;
}
// Enqueues several items.
// Does not allocate memory (except for one-time implicit producer).
// Fails if not enough room to enqueue (or implicit production is
// disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0).
// Note: Use std::make_move_iterator if the elements should be moved
// instead of copied.
// Thread-safe.
template<typename It>
inline bool try_enqueue_bulk(It itemFirst, size_t count)
{
if (inner.try_enqueue_bulk(std::forward<It>(itemFirst), count)) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
return true;
}
return false;
}
// Enqueues several items using an explicit producer token.
// Does not allocate memory. Fails if not enough room to enqueue.
// Note: Use std::make_move_iterator if the elements should be moved
// instead of copied.
// Thread-safe.
template<typename It>
inline bool try_enqueue_bulk(producer_token_t const& token, It itemFirst, size_t count)
{
if (inner.try_enqueue_bulk(token, std::forward<It>(itemFirst), count)) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
return true;
}
return false;
}
// Attempts to dequeue from the queue.
// Returns false if all producer streams appeared empty at the time they
// were checked (so, the queue is likely but not guaranteed to be empty).
// Never allocates. Thread-safe.
template<typename U>
inline bool try_dequeue(U& item)
{
if (sema->tryWait()) {
while (!inner.try_dequeue(item)) {
continue;
}
return true;
}
return false;
}
// Attempts to dequeue from the queue using an explicit consumer token.
// Returns false if all producer streams appeared empty at the time they
// were checked (so, the queue is likely but not guaranteed to be empty).
// Never allocates. Thread-safe.
template<typename U>
inline bool try_dequeue(consumer_token_t& token, U& item)
{
if (sema->tryWait()) {
while (!inner.try_dequeue(token, item)) {
continue;
}
return true;
}
return false;
}
// Attempts to dequeue several elements from the queue.
// Returns the number of items actually dequeued.
// Returns 0 if all producer streams appeared empty at the time they
// were checked (so, the queue is likely but not guaranteed to be empty).
// Never allocates. Thread-safe.
template<typename It>
inline size_t try_dequeue_bulk(It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->tryWaitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue using an explicit consumer token.
// Returns the number of items actually dequeued.
// Returns 0 if all producer streams appeared empty at the time they
// were checked (so, the queue is likely but not guaranteed to be empty).
// Never allocates. Thread-safe.
template<typename It>
inline size_t try_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->tryWaitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
}
return count;
}
// Blocks the current thread until there's something to dequeue, then
// dequeues it.
// Never allocates. Thread-safe.
template<typename U>
inline void wait_dequeue(U& item)
{
sema->wait();
while (!inner.try_dequeue(item)) {
continue;
}
}
// Blocks the current thread until either there's something to dequeue
// or the timeout (specified in microseconds) expires. Returns false
// without setting `item` if the timeout expires, otherwise assigns
// to `item` and returns true.
// Using a negative timeout indicates an indefinite timeout,
// and is thus functionally equivalent to calling wait_dequeue.
// Never allocates. Thread-safe.
template<typename U>
inline bool wait_dequeue_timed(U& item, std::int64_t timeout_usecs)
{
if (!sema->wait(timeout_usecs)) {
return false;
}
while (!inner.try_dequeue(item)) {
continue;
}
return true;
}
// Blocks the current thread until either there's something to dequeue
// or the timeout expires. Returns false without setting `item` if the
// timeout expires, otherwise assigns to `item` and returns true.
// Never allocates. Thread-safe.
template<typename U, typename Rep, typename Period>
inline bool wait_dequeue_timed(U& item, std::chrono::duration<Rep, Period> const& timeout)
{
return wait_dequeue_timed(item, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
}
// Blocks the current thread until there's something to dequeue, then
// dequeues it using an explicit consumer token.
// Never allocates. Thread-safe.
template<typename U>
inline void wait_dequeue(consumer_token_t& token, U& item)
{
sema->wait();
while (!inner.try_dequeue(token, item)) {
continue;
}
}
// Blocks the current thread until either there's something to dequeue
// or the timeout (specified in microseconds) expires. Returns false
// without setting `item` if the timeout expires, otherwise assigns
// to `item` and returns true.
// Using a negative timeout indicates an indefinite timeout,
// and is thus functionally equivalent to calling wait_dequeue.
// Never allocates. Thread-safe.
template<typename U>
inline bool wait_dequeue_timed(consumer_token_t& token, U& item, std::int64_t timeout_usecs)
{
if (!sema->wait(timeout_usecs)) {
return false;
}
while (!inner.try_dequeue(token, item)) {
continue;
}
return true;
}
// Blocks the current thread until either there's something to dequeue
// or the timeout expires. Returns false without setting `item` if the
// timeout expires, otherwise assigns to `item` and returns true.
// Never allocates. Thread-safe.
template<typename U, typename Rep, typename Period>
inline bool wait_dequeue_timed(consumer_token_t& token, U& item, std::chrono::duration<Rep, Period> const& timeout)
{
return wait_dequeue_timed(token, item, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
}
// Attempts to dequeue several elements from the queue.
// Returns the number of items actually dequeued, which will
// always be at least one (this method blocks until the queue
// is non-empty) and at most max.
// Never allocates. Thread-safe.
template<typename It>
inline size_t wait_dequeue_bulk(It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue.
// Returns the number of items actually dequeued, which can
// be 0 if the timeout expires while waiting for elements,
// and at most max.
// Using a negative timeout indicates an indefinite timeout,
// and is thus functionally equivalent to calling wait_dequeue_bulk.
// Never allocates. Thread-safe.
template<typename It>
inline size_t wait_dequeue_bulk_timed(It itemFirst, size_t max, std::int64_t timeout_usecs)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max, timeout_usecs);
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue.
// Returns the number of items actually dequeued, which can
// be 0 if the timeout expires while waiting for elements,
// and at most max.
// Never allocates. Thread-safe.
template<typename It, typename Rep, typename Period>
inline size_t wait_dequeue_bulk_timed(It itemFirst, size_t max, std::chrono::duration<Rep, Period> const& timeout)
{
return wait_dequeue_bulk_timed<It&>(itemFirst, max, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
}
// Attempts to dequeue several elements from the queue using an explicit consumer token.
// Returns the number of items actually dequeued, which will
// always be at least one (this method blocks until the queue
// is non-empty) and at most max.
// Never allocates. Thread-safe.
template<typename It>
inline size_t wait_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue using an explicit consumer token.
// Returns the number of items actually dequeued, which can
// be 0 if the timeout expires while waiting for elements,
// and at most max.
// Using a negative timeout indicates an indefinite timeout,
// and is thus functionally equivalent to calling wait_dequeue_bulk.
// Never allocates. Thread-safe.
template<typename It>
inline size_t wait_dequeue_bulk_timed(consumer_token_t& token, It itemFirst, size_t max, std::int64_t timeout_usecs)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max, timeout_usecs);
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue using an explicit consumer token.
// Returns the number of items actually dequeued, which can
// be 0 if the timeout expires while waiting for elements,
// and at most max.
// Never allocates. Thread-safe.
template<typename It, typename Rep, typename Period>
inline size_t wait_dequeue_bulk_timed(consumer_token_t& token, It itemFirst, size_t max, std::chrono::duration<Rep, Period> const& timeout)
{
return wait_dequeue_bulk_timed<It&>(token, itemFirst, max, std::chrono::duration_cast<std::chrono::microseconds>(timeout).count());
}
// Returns an estimate of the total number of elements currently in the queue. This
// estimate is only accurate if the queue has completely stabilized before it is called
// (i.e. all enqueue and dequeue operations have completed and their memory effects are
// visible on the calling thread, and no further operations start while this method is
// being called).
// Thread-safe.
inline size_t size_approx() const
{
return (size_t)sema->availableApprox();
}
// Returns true if the underlying atomic variables used by
// the queue are lock-free (they should be on most platforms).
// Thread-safe.
static bool is_lock_free()
{
return ConcurrentQueue::is_lock_free();
}
private:
template<typename U>
static inline U* create()
{
auto p = (Traits::malloc)(sizeof(U));
return p != nullptr ? new (p) U : nullptr;
}
template<typename U, typename A1>
static inline U* create(A1&& a1)
{
auto p = (Traits::malloc)(sizeof(U));
return p != nullptr ? new (p) U(std::forward<A1>(a1)) : nullptr;
}
template<typename U>
static inline void destroy(U* p)
{
if (p != nullptr) {
p->~U();
}
(Traits::free)(p);
}
private:
ConcurrentQueue inner;
std::unique_ptr<LightweightSemaphore, void (*)(LightweightSemaphore*)> sema;
};
template<typename T, typename Traits>
inline void swap(BlockingConcurrentQueue<T, Traits>& a, BlockingConcurrentQueue<T, Traits>& b) MOODYCAMEL_NOEXCEPT
{
a.swap(b);
}
} // end namespace moodycamel

3623
lib/concurrentqueue/include/moodycamel/concurrentqueue.h Normal file
View File

File diff suppressed because it is too large Load Diff

2
src/CMakeLists.txt
View File

@ -62,6 +62,7 @@ target_include_directories(${LIBRARY_NAME}_static PUBLIC ${BOOST_INCLUDE_DIR})
target_include_directories(${LIBRARY_NAME}_static PUBLIC ${FONTCONFIG_INCLUDE_DIRS})
target_include_directories(${LIBRARY_NAME}_static PUBLIC ${PROJECT_SOURCE_DIR}/include)
target_include_directories(${LIBRARY_NAME}_static PUBLIC ${PROJECT_SOURCE_DIR}/lib/boost/include)
target_include_directories(${LIBRARY_NAME}_static PUBLIC ${PROJECT_SOURCE_DIR}/lib/concurrentqueue/include)
target_compile_definitions(${BINARY_NAME} PUBLIC
${X11_XCB_DEFINITIONS}
@ -116,6 +117,7 @@ set(APP_LIBRARIES ${LIBRARY_NAME}_static ${XPP_LIBRARY} PARENT_SCOPE)
set(APP_INCLUDE_DIRS
${PROJECT_SOURCE_DIR}/include
${PROJECT_SOURCE_DIR}/lib/boost/include
${PROJECT_SOURCE_DIR}/lib/concurrentqueue/include
${XPP_INCLUDE_DIRS}
PARENT_SCOPE)