From 3dbe63b7dfacba4a46f3641781681e82c8091846 Mon Sep 17 00:00:00 2001 From: Marcus Holland-Moritz Date: Sun, 14 Jan 2024 19:11:36 +0100 Subject: [PATCH] chore: support logging in worker_group --- include/dwarfs/worker_group.h | 4 +++- src/dwarfs/block_cache.cpp | 12 ++++++------ src/dwarfs/filesystem_extractor.cpp | 2 +- src/dwarfs/filesystem_v2.cpp | 2 +- src/dwarfs/scanner.cpp | 4 ++-- src/dwarfs/worker_group.cpp | 29 ++++++++++++++++++----------- src/dwarfsbench_main.cpp | 2 +- src/mkdwarfs_main.cpp | 10 +++++----- test/compat_test.cpp | 2 +- test/dwarfs_benchmark.cpp | 6 +++--- test/dwarfs_test.cpp | 4 ++-- 11 files changed, 43 insertions(+), 34 deletions(-) diff --git a/include/dwarfs/worker_group.h b/include/dwarfs/worker_group.h index 65164bf2..bc2aa32a 100644 --- a/include/dwarfs/worker_group.h +++ b/include/dwarfs/worker_group.h @@ -31,6 +31,8 @@ namespace dwarfs { +class logger; + /** * A group of worker threads * @@ -48,7 +50,7 @@ class worker_group { * \param num_workers Number of worker threads. */ explicit worker_group( - const char* group_name, size_t num_workers = 1, + logger& lgr, const char* group_name, size_t num_workers = 1, size_t max_queue_len = std::numeric_limits::max(), int niceness = 0); diff --git a/src/dwarfs/block_cache.cpp b/src/dwarfs/block_cache.cpp index cb0670ec..9b7fb141 100644 --- a/src/dwarfs/block_cache.cpp +++ b/src/dwarfs/block_cache.cpp @@ -144,11 +144,11 @@ class block_cache_ final : public block_cache::impl { , LOG_PROXY_INIT(lgr) , options_(options) { if (options.init_workers) { - wg_ = - worker_group("blkcache", std::max(options.num_workers > 0 - ? options.num_workers - : folly::hardware_concurrency(), - static_cast(1))); + wg_ = worker_group(lgr, "blkcache", + std::max(options.num_workers > 0 + ? options.num_workers + : folly::hardware_concurrency(), + static_cast(1))); } } @@ -247,7 +247,7 @@ class block_cache_ final : public block_cache::impl { wg_.stop(); } - wg_ = worker_group("blkcache", num); + wg_ = worker_group(LOG_GET_LOGGER, "blkcache", num); } void set_tidy_config(cache_tidy_config const& cfg) override { diff --git a/src/dwarfs/filesystem_extractor.cpp b/src/dwarfs/filesystem_extractor.cpp index 5526f1c3..398fec1e 100644 --- a/src/dwarfs/filesystem_extractor.cpp +++ b/src/dwarfs/filesystem_extractor.cpp @@ -235,7 +235,7 @@ bool filesystem_extractor_::extract( ::archive_entry* spare = nullptr; - worker_group archiver("archiver", 1); + worker_group archiver(LOG_GET_LOGGER, "archiver", 1); cache_semaphore sem; LOG_DEBUG << "extractor semaphore size: " << opts.max_queued_bytes diff --git a/src/dwarfs/filesystem_v2.cpp b/src/dwarfs/filesystem_v2.cpp index 835ec913..a82ffc96 100644 --- a/src/dwarfs/filesystem_v2.cpp +++ b/src/dwarfs/filesystem_v2.cpp @@ -736,7 +736,7 @@ int filesystem_::check(filesystem_check_level level, size_t num_threads) const { filesystem_parser parser(mm_, image_offset_); - worker_group wg("fscheck", num_threads); + worker_group wg(LOG_GET_LOGGER, "fscheck", num_threads); std::vector> sections; while (auto sp = parser.next_section()) { diff --git a/src/dwarfs/scanner.cpp b/src/dwarfs/scanner.cpp index 6420b653..5ab49560 100644 --- a/src/dwarfs/scanner.cpp +++ b/src/dwarfs/scanner.cpp @@ -692,8 +692,8 @@ void scanner_::scan( { size_t const num_threads = options_.num_segmenter_workers; - worker_group wg_ordering("ordering", num_threads); - worker_group wg_blockify("blockify", num_threads); + worker_group wg_ordering(LOG_GET_LOGGER, "ordering", num_threads); + worker_group wg_blockify(LOG_GET_LOGGER, "blockify", num_threads); fsw.configure(frag_info.categories, num_threads); diff --git a/src/dwarfs/worker_group.cpp b/src/dwarfs/worker_group.cpp index f640b7f7..8c0beb4f 100644 --- a/src/dwarfs/worker_group.cpp +++ b/src/dwarfs/worker_group.cpp @@ -41,6 +41,7 @@ #include #include "dwarfs/error.h" +#include "dwarfs/logger.h" #include "dwarfs/util.h" #include "dwarfs/worker_group.h" @@ -91,16 +92,15 @@ double get_thread_cpu_time(std::thread const& t) { #endif -} // namespace - -template +template class basic_worker_group final : public worker_group::impl, private Policy { public: template - basic_worker_group(const char* group_name, size_t num_workers, + basic_worker_group(logger& lgr, const char* group_name, size_t num_workers, size_t max_queue_len, int niceness [[maybe_unused]], Args&&... args) : Policy(std::forward(args)...) + , LOG_PROXY_INIT(lgr) , running_(true) , pending_(0) , max_queue_len_(max_queue_len) { @@ -331,9 +331,8 @@ class basic_worker_group final : public worker_group::impl, private Policy { try { job(); } catch (...) { - std::cerr << "FATAL ERROR: exception thrown in worker thread: " - << folly::exceptionStr(std::current_exception()) << '\n'; - std::abort(); + LOG_FATAL << "exception thrown in worker thread: " + << folly::exceptionStr(std::current_exception()); } #ifdef _WIN32 if (is_background) { @@ -352,6 +351,7 @@ class basic_worker_group final : public worker_group::impl, private Policy { } } + LOG_PROXY_DECL(LoggerPolicy); std::vector workers_; jobs_t jobs_; std::condition_variable cond_; @@ -371,9 +371,16 @@ class no_policy { }; }; -worker_group::worker_group(const char* group_name, size_t num_workers, - size_t max_queue_len, int niceness) - : impl_{std::make_unique>( - group_name, num_workers, max_queue_len, niceness)} {} +template +using default_worker_group = basic_worker_group; + +} // namespace + +worker_group::worker_group(logger& lgr, const char* group_name, + size_t num_workers, size_t max_queue_len, + int niceness) + : impl_{make_unique_logging_object( + lgr, group_name, num_workers, max_queue_len, niceness)} {} } // namespace dwarfs diff --git a/src/dwarfsbench_main.cpp b/src/dwarfsbench_main.cpp index 92a74687..4333822e 100644 --- a/src/dwarfsbench_main.cpp +++ b/src/dwarfsbench_main.cpp @@ -104,7 +104,7 @@ int dwarfsbench_main(int argc, sys_char** argv, iolayer const& iol) { dwarfs::filesystem_v2 fs(lgr, std::make_shared(filesystem), fsopts); - worker_group wg("reader", num_readers); + worker_group wg(lgr, "reader", num_readers); fs.walk([&](auto entry) { auto inode_data = entry.inode(); diff --git a/src/mkdwarfs_main.cpp b/src/mkdwarfs_main.cpp index 3395ab77..e3e558cc 100644 --- a/src/mkdwarfs_main.cpp +++ b/src/mkdwarfs_main.cpp @@ -865,10 +865,6 @@ int mkdwarfs_main(int argc, sys_char** argv, iolayer const& iol) { num_segmenter_workers = num_workers; } - worker_group wg_compress("compress", num_workers, - std::numeric_limits::max(), - compress_niceness); - options.num_segmenter_workers = num_segmenter_workers; if (vm.count("debug-filter")) { @@ -1243,6 +1239,10 @@ int mkdwarfs_main(int argc, sys_char** argv, iolayer const& iol) { block_compressor metadata_bc(metadata_compression); block_compressor history_bc(history_compression); + worker_group wg_compress(lgr, "compress", num_workers, + std::numeric_limits::max(), + compress_niceness); + std::unique_ptr fsw; try { @@ -1316,7 +1316,7 @@ int mkdwarfs_main(int argc, sys_char** argv, iolayer const& iol) { auto sf = std::make_shared( lgr, prog, options.inode.categorizer_mgr, sf_config); - worker_group wg_scanner("scanner", num_scanner_workers); + worker_group wg_scanner(lgr, "scanner", num_scanner_workers); scanner s(lgr, wg_scanner, std::move(sf), entry_factory::create(), iol.os, std::move(script), options); diff --git a/test/compat_test.cpp b/test/compat_test.cpp index 95811484..44325dee 100644 --- a/test/compat_test.cpp +++ b/test/compat_test.cpp @@ -1109,7 +1109,7 @@ TEST_P(rewrite, filesystem_rewrite) { opts.recompress_block = recompress_block; opts.recompress_metadata = recompress_metadata; - worker_group wg("rewriter", 2); + worker_group wg(lgr, "rewriter", 2); block_compressor bc("null"); progress prog([](const progress&, bool) {}, 1000); std::ostringstream rewritten, idss; diff --git a/test/dwarfs_benchmark.cpp b/test/dwarfs_benchmark.cpp index b04191fc..3bb7df52 100644 --- a/test/dwarfs_benchmark.cpp +++ b/test/dwarfs_benchmark.cpp @@ -115,11 +115,11 @@ std::string make_filesystem(::benchmark::State const& state) { options.plain_names_table = state.range(1); options.plain_symlinks_table = state.range(1); - worker_group wg("writer", 4); - progress prog([](const progress&, bool) {}, 1000); - test::test_logger lgr; + worker_group wg(lgr, "writer", 4); + progress prog([](const progress&, bool) {}, 1000); + auto sf = std::make_shared(lgr, prog, cfg); scanner s(lgr, wg, sf, entry_factory::create(), diff --git a/test/dwarfs_test.cpp b/test/dwarfs_test.cpp index ff129807..ad0864e2 100644 --- a/test/dwarfs_test.cpp +++ b/test/dwarfs_test.cpp @@ -74,7 +74,7 @@ build_dwarfs(logger& lgr, std::shared_ptr input, std::optional> input_list = std::nullopt) { // force multithreading - worker_group wg("worker", 4); + worker_group wg(lgr, "worker", 4); std::unique_ptr local_prog; if (!prog) { @@ -908,7 +908,7 @@ class filter_test }; progress prog([](const progress&, bool) {}, 1000); - worker_group wg("worker", 1); + worker_group wg(lgr, "worker", 1); auto sf = std::make_shared(lgr, prog, segmenter_factory::config{}); scanner s(lgr, wg, sf, entry_factory::create(), input, scr, options);