diff --git a/.github/workflows/build_dependencies.yml b/.github/workflows/build_dependencies.yml index 180c5f9b7..3b565295d 100644 --- a/.github/workflows/build_dependencies.yml +++ b/.github/workflows/build_dependencies.yml @@ -228,17 +228,31 @@ jobs: key_prefix: HomestoreDeps-${{ inputs.platform }}-${{ inputs.build-type }}-${{ inputs.malloc-impl }} if: ${{ github.event_name != 'pull_request' && steps.restore-cache.outputs.cache-hit != 'true' && inputs.tooling != 'Coverage'}} + - name: Install gdb + # src/tests/stack_dump_launcher.sh uses it to print the stacks of a hung test, and of a crashed one from the + # core. file it leaves in the test directory + run: | + command -v gdb || (sudo apt-get update -q && sudo apt-get install -y -q gdb) + sudo sysctl -w kernel.core_pattern=core.%p + if: ${{ inputs.testing == 'True' }} + - name: Create and Test Package run: | if [[ "${{ inputs.tooling }}" == "AddressSanitize" ]]; then sanitize="address" elif [[ "${{ inputs.tooling }}" == "ThreadSanitize" ]]; then sanitize="thread" else sanitize="False"; fi - conan create \ + # ASan makes the templated btree tests much bigger to compile: four parallel compiles have run the runner out of + # memory, killing the job mid-build. + build_jobs="" + if [[ "${sanitize}" == "address" ]]; then build_jobs="-c tools.build:jobs=2"; fi + conan create ${build_jobs} \ -o sisl/*:malloc_impl=${{ inputs.malloc-impl }} \ -o homestore/*:sanitize=${sanitize} \ -s:h build_type=${{ inputs.build-type }} \ ${{ inputs.conan-channel != '' && format('--user oss --channel {0}', inputs.conan-channel) || '' }} \ . + env: + CTEST_OUTPUT_ON_FAILURE: 1 if: ${{ inputs.testing == 'True' && inputs.tooling != 'Coverage' }} - name: Code Coverage Run @@ -251,6 +265,8 @@ jobs: ${{ inputs.conan-channel != '' && format('--user oss --channel {0}', inputs.conan-channel) || '' }} \ . gcovr --cobertura ./coverage.xml + env: + CTEST_OUTPUT_ON_FAILURE: 1 if: ${{ inputs.testing == 'True' && inputs.tooling == 'Coverage' }} - name: Upload coverage reports to Codecov @@ -259,3 +275,22 @@ jobs: files: ./coverage.xml disable_search: true if: ${{ inputs.tooling == 'Coverage' }} + + - name: Collect Test Logs + # Also after a failed, timed out or cancelled test step: the test logs are the only record of what went wrong. + # Pack them first: the log directories are named by time (logs/2026-10-20_19:45), and upload-artifact rejects ':'. + run: | + shopt -s nullglob + dirs=(~/.conan2/p/b/homes*/b/build/*/Testing/Temporary ~/.conan2/p/b/homes*/b/build/*/src/tests/logs + build/*/Testing/Temporary build/*/src/tests/logs) + if (( ${#dirs[@]} )); then tar -czf test-logs.tgz --ignore-failed-read "${dirs[@]}" || true; fi + if: ${{ always() && inputs.testing == 'True' }} + + - name: Upload Test Logs + uses: actions/upload-artifact@v4 + with: + name: test-logs-${{ inputs.platform }}-${{ inputs.build-type }}-${{ inputs.malloc-impl }}-${{ inputs.tooling }} + path: test-logs.tgz + if-no-files-found: ignore + retention-days: 7 + if: ${{ always() && inputs.testing == 'True' }} diff --git a/CMakeLists.txt b/CMakeLists.txt index 1a1cda169..954e4a614 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -32,15 +32,13 @@ if ($ENV{ASAN_OPTIONS} MATCHES "detect_leaks=1") set(ASAN_ENABLED_BUILD true) endif() -if (DEFINED MEMORY_SANITIZER_ON) - if (${MEMORY_SANITIZER_ON}) - message(STATUS "Build with Memory Sanitizer ON") - include (cmake/mem_sanitizer.cmake) - else () - message(STATUS "Build with Memory Sanitizer OFF") - endif() +# conanfile.py sets ADDRESS_SANITIZER_ON from the sanitize option; MEMORY_SANITIZER_ON is the older name for it. +# THREAD_SANITIZER_ON isn't supported yet: with uninstrumented dependencies TSan reports thousands of false races. +if (ADDRESS_SANITIZER_ON OR MEMORY_SANITIZER_ON) + message(STATUS "Build with Address Sanitizer ON") + include (cmake/mem_sanitizer.cmake) else () - message(STATUS "Build with Memory Sanitizer OFF") + message(STATUS "Build with Address Sanitizer OFF") endif () if (${CMAKE_BUILD_TYPE} STREQUAL "Debug") diff --git a/conanfile.py b/conanfile.py index c32bf3697..d247c65b5 100644 --- a/conanfile.py +++ b/conanfile.py @@ -9,7 +9,7 @@ class HomestoreConan(ConanFile): name = "homestore" - version = "8.3.5" + version = "8.3.6" homepage = "https://github.com/eBay/Homestore" description = "HomeStore Storage Engine" diff --git a/src/include/homestore/logstore/log_store.hpp b/src/include/homestore/logstore/log_store.hpp index dac2935fd..995065b8e 100644 --- a/src/include/homestore/logstore/log_store.hpp +++ b/src/include/homestore/logstore/log_store.hpp @@ -302,6 +302,8 @@ class home_log_store : public std::enable_shared_from_this< home_log_store > { LogStoreServiceMetrics& m_metrics; logdev_key m_trunc_ld_key{0, 0}; + // Serializes truncate(); see the lock order note there. + std::mutex m_truncate_mtx; private: // graceful shutdown related fields diff --git a/src/include/homestore/replication/repl_dev.hpp b/src/include/homestore/replication/repl_dev.hpp index 03b5c24e7..959c06ef7 100644 --- a/src/include/homestore/replication/repl_dev.hpp +++ b/src/include/homestore/replication/repl_dev.hpp @@ -662,10 +662,16 @@ class repl_dev { // we have no shutdown for repl_dev, since shutdown repl_dev is done by repl_service void stop() { start_stopping(); - while (true) { + for (uint64_t waited_sec{0};; ++waited_sec) { auto pending_request_num = get_pending_request_num(); if (!pending_request_num) break; + // No deadline: callers free resources (e.g. the raft server in leave_group()) that in-flight requests + // still use. Report instead, so a leaked count shows up in the logs rather than as a silent hang. + if (waited_sec && (waited_sec % 30 == 0)) { + LOGWARNMOD(replication, "repl_dev group_id={} still waiting for {} pending requests after {} sec", + boost::uuids::to_string(group_id()), pending_request_num, waited_sec); + } std::this_thread::sleep_for(std::chrono::milliseconds(1000)); } } diff --git a/src/lib/blkdata_svc/blk_read_tracker.cpp b/src/lib/blkdata_svc/blk_read_tracker.cpp index e5192fd6d..a9d6e8fdf 100644 --- a/src/lib/blkdata_svc/blk_read_tracker.cpp +++ b/src/lib/blkdata_svc/blk_read_tracker.cpp @@ -52,11 +52,12 @@ void BlkReadTracker::merge(const blk_id& blkid, int64_t new_ref_count, // trying to acquire the same write lock in its IO completion path. boost::container::small_vector< blk_track_waiter_ptr, 8 > fired_waiters; m_pending_reads_map.upsert_or_delete( - base_blkid, [new_ref_count, &base_blkid, &fired_waiters](BlkTrackRecord& rec, bool existing) { + base_blkid, [this, new_ref_count, &base_blkid, &fired_waiters](BlkTrackRecord& rec, bool existing) { HS_DBG_ASSERT_EQ(existing, true, "Decrement a ref count (blk: {}) which does not exist in map", base_blkid.to_string()); rec.m_ref_cnt += new_ref_count; if (rec.m_ref_cnt == 0) { + std::lock_guard lg{m_waiters_mtx}; fired_waiters = std::move(rec.m_waiters); return true; } @@ -65,7 +66,8 @@ void BlkReadTracker::merge(const blk_id& blkid, int64_t new_ref_count, // fired_waiters destructs here, outside the write lock -> m_cb() fires safely } else { // this is wait_on operation - m_pending_reads_map.update(base_blkid, [&waiter_rescheduled, &waiter](BlkTrackRecord& rec) { + m_pending_reads_map.update(base_blkid, [this, &waiter_rescheduled, &waiter](BlkTrackRecord& rec) { + std::lock_guard lg{m_waiters_mtx}; rec.m_waiters.push_back(waiter); waiter_rescheduled = true; }); diff --git a/src/lib/blkdata_svc/blk_read_tracker.hpp b/src/lib/blkdata_svc/blk_read_tracker.hpp index 5b5fafae7..e89844b47 100644 --- a/src/lib/blkdata_svc/blk_read_tracker.hpp +++ b/src/lib/blkdata_svc/blk_read_tracker.hpp @@ -14,6 +14,7 @@ *********************************************************************************/ #pragma once #include +#include #include #include @@ -115,6 +116,9 @@ class BlkReadTracker { private: sisl::SimpleHashMap< blk_id, BlkTrackRecord > m_pending_reads_map; + // wait_on() appends to BlkTrackRecord::m_waiters through SimpleHashMap::update(), which only holds the bucket's + // shared lock, so concurrent wait_on()s on the same record need this to modify m_waiters. + std::mutex m_waiters_mtx; BlkReadTrackerMetrics m_metrics; uint32_t m_entries_per_record{s_entries_per_record}; diff --git a/src/lib/checkpoint/cp_internal.hpp b/src/lib/checkpoint/cp_internal.hpp index f17a26952..ddab8a2f8 100644 --- a/src/lib/checkpoint/cp_internal.hpp +++ b/src/lib/checkpoint/cp_internal.hpp @@ -36,7 +36,9 @@ class CPWatchdog { CP* m_cp; CPManager* m_cp_mgr; std::shared_mutex m_cp_mtx; - Clock::time_point m_last_state_ch_time; + Clock::time_point m_last_progress_time; // start of the current stall, reported if the cp is declared stuck + Clock::time_point m_last_repair_time; // last repair_slow_cp() round, gives the repairs time to take effect + uint32_t m_repair_attempts{0}; // repair_slow_cp() rounds in the current stall uint64_t m_timer_sec{0}; iomgr::timer_handle_t m_timer_hdl; uint32_t m_progress_pct{0}; diff --git a/src/lib/checkpoint/cp_mgr.cpp b/src/lib/checkpoint/cp_mgr.cpp index efaf15a1c..46919533e 100644 --- a/src/lib/checkpoint/cp_mgr.cpp +++ b/src/lib/checkpoint/cp_mgr.cpp @@ -448,7 +448,9 @@ void CPWatchdog::reset_cp() { void CPWatchdog::set_cp(CP* cp) { std::unique_lock< std::shared_mutex > lk{m_cp_mtx}; m_cp = cp; - m_last_state_ch_time = Clock::now(); + m_last_progress_time = Clock::now(); + m_last_repair_time = m_last_progress_time; + m_repair_attempts = 0; } void CPWatchdog::stop() { @@ -466,45 +468,52 @@ void CPWatchdog::cp_watchdog_timer() { // check if any cp to track if (m_cp == nullptr) { return; } const auto status = m_cp->get_status(); - if ((status != cp_status_t::cp_flush_prepare) || (status != cp_status_t::cp_flushing)) { return; } + if ((status != cp_status_t::cp_flush_prepare) && (status != cp_status_t::cp_flushing)) { return; } - uint32_t cum_pct{0}; - uint32_t count{0}; + // Track the slowest consumer, so one wedged consumer can't be masked by its peers advancing. + // NOTE: every consumer currently reports a constant 100 (they flush synchronously), so in practice this is a + // time-based check: a cp that stays in flush for 12 timer periods is declared stuck. + uint32_t pct{100}; for (auto& consumer : m_cp_mgr->consumer_list()) { - if (consumer) { - ++count; - cum_pct += consumer->cp_progress_percent(); - } + if (consumer) { pct = std::min(pct, static_cast< uint32_t >(consumer->cp_progress_percent())); } } - if (m_progress_pct > cum_pct / count) { + if (pct > m_progress_pct) { // We are making progress in flushing the data. - m_progress_pct = cum_pct / count; + m_progress_pct = pct; + m_last_progress_time = Clock::now(); + m_repair_attempts = 0; return; } - if (get_elapsed_time_ms(m_last_state_ch_time) >= m_timer_sec * 1000) { - LOGINFO("cp progress percent {} is not changed. time elapsed {}, cp state={} ", m_progress_pct, - get_elapsed_time_ms(m_last_state_ch_time), m_cp->to_string()); - } - - // check if enough time passed since last state change + // check if enough time passed since the last progress, and since the last repair attempt + const auto elapsed_ms = get_elapsed_time_ms(m_last_progress_time); uint32_t max_time_multiplier = 12; - if (get_elapsed_time_ms(m_last_state_ch_time) < max_time_multiplier * m_timer_sec * 1000) { + const auto max_stall_ms = max_time_multiplier * m_timer_sec * 1000; + if ((elapsed_ms < max_stall_ms) || (get_elapsed_time_ms(m_last_repair_time) < max_stall_ms)) { return; } + + // Ask the consumers that are behind to push harder, but only a limited number of times: a consumer that is + // deadlocked never reaches 100, and repairing it forever would leave the cp stuck with just warn logs. + const uint32_t max_repair_attempts = 3; + if (m_repair_attempts < max_repair_attempts) { uint32_t repair_attempted{0}; for (auto& consumer : m_cp_mgr->consumer_list()) { - if (consumer) { - const auto pct = consumer->cp_progress_percent(); - if (pct != 100) { - consumer->repair_slow_cp(); - ++repair_attempted; - } + if (consumer && (consumer->cp_progress_percent() != 100)) { + consumer->repair_slow_cp(); + ++repair_attempted; } - if (repair_attempted) { return; } } - - HS_REL_ASSERT(0, "cp seems to be stuck. CP State={} total time elapsed {}", m_cp->to_string(), - get_elapsed_time_ms(m_last_state_ch_time)); + if (repair_attempted) { + ++m_repair_attempts; + LOGWARN("cp seems to be stuck, no progress for {} ms, repair attempt {}/{} on {} consumers. CP State={}", + elapsed_ms, m_repair_attempts, max_repair_attempts, repair_attempted, m_cp->to_string()); + // give the repairs time to take effect + m_last_repair_time = Clock::now(); + return; + } } + + HS_REL_ASSERT(0, "cp seems to be stuck, no progress for {} ms after {} repair attempts. CP State={}", elapsed_ms, + m_repair_attempts, m_cp->to_string()); } cp_id_t CPContext::id() const { return m_cp->id(); } diff --git a/src/lib/device/journal_vdev.cpp b/src/lib/device/journal_vdev.cpp index 6602d67f9..86fc66d70 100644 --- a/src/lib/device/journal_vdev.cpp +++ b/src/lib/device/journal_vdev.cpp @@ -99,7 +99,10 @@ void JournalVirtualDev::init() { for (auto& [logdev_id, head] : logdev_head_map) { // Create descriptor for each logdev_id auto journal_desc = std::make_shared< JournalVirtualDev::Descriptor >(*this, logdev_id); - m_journal_descriptors.emplace(logdev_id, journal_desc); + { + std::lock_guard lock{m_mutex}; + m_journal_descriptors.emplace(logdev_id, journal_desc); + } LOGINFOMOD(journalvdev, "Loading descriptor log_dev={}", logdev_id); // Traverse the list starting from the head and add those chunks // in order to the journal descriptor. next_chunk is stored in private_data. @@ -178,6 +181,8 @@ uint64_t JournalVirtualDev::get_end_of_chunk(shared< Chunk >& chunk) const { } shared< JournalVirtualDev::Descriptor > JournalVirtualDev::open(logdev_id_t logdev_id) { + // Descriptors can be opened from several threads at once (VDevJournalIOTest.LongRunning opens one per thread) + std::lock_guard lock{m_mutex}; auto it = m_journal_descriptors.find(logdev_id); if (it == m_journal_descriptors.end()) { auto journal_desc = std::make_shared< JournalVirtualDev::Descriptor >(*this, logdev_id); @@ -194,6 +199,7 @@ shared< JournalVirtualDev::Descriptor > JournalVirtualDev::open(logdev_id_t logd } void JournalVirtualDev::destroy(logdev_id_t logdev_id) { + std::lock_guard lock{m_mutex}; auto it = m_journal_descriptors.find(logdev_id); if (it == m_journal_descriptors.end()) { LOGERROR("logdev not found log_dev={}", logdev_id); diff --git a/src/lib/logstore/log_store.cpp b/src/lib/logstore/log_store.cpp index 043c22802..47b71b99b 100644 --- a/src/lib/logstore/log_store.cpp +++ b/src/lib/logstore/log_store.cpp @@ -214,6 +214,16 @@ bool home_log_store::truncate(logstore_seq_num_t upto_lsn, bool in_memory_trunca m_start_lsn.load(std::memory_order_relaxed)); return false; } + // Truncates of this store must not interleave: another one can move m_start_lsn and m_records past upto_lsn + // between the range check and the record lookup below. Lock order is m_truncate_mtx, then the logdev's flush guard + // and store map lock (taken by flush() and m_logdev->truncate()). LogDev::start() calls truncate() while holding + // the store map lock, but only up to the already truncated lsn, which returns above without taking m_truncate_mtx. + std::lock_guard truncate_lg{m_truncate_mtx}; + if (upto_lsn < m_start_lsn) { + // a concurrent truncate got here first + decr_pending_request_num(); + return false; + } flush(); #ifndef NDEBUG auto cs = get_contiguous_completed_seq_num(0); diff --git a/src/lib/replication/repl_dev/raft_repl_dev.cpp b/src/lib/replication/repl_dev/raft_repl_dev.cpp index 187d6f8c4..a7e61b87f 100644 --- a/src/lib/replication/repl_dev/raft_repl_dev.cpp +++ b/src/lib/replication/repl_dev/raft_repl_dev.cpp @@ -1383,6 +1383,17 @@ bool RaftReplDev::wait_for_data_receive(std::vector< repl_req_ptr_t > const& rre } void RaftReplDev::check_and_fetch_remote_data(std::vector< repl_req_ptr_t > rreqs) { + // Same gate as on_fetch_data_received, on the sending side: once this dev is no longer ACTIVE, the reaper calls + // leave_group(), which frees the raft server that group_msg_service() resolves the originator through. Register + // the request before checking, so the reaper's stop() either waits for it or this check sees is_stopping(). + incr_pending_request_num(); + if (is_stopping() || (get_stage() != repl_dev_stage_t::ACTIVE)) { + RD_LOGD(NO_TRACE_ID, "Data Channel: not fetching data while not active (stopping={} stage={})", is_stopping(), + static_cast< int >(get_stage())); + decr_pending_request_num(); + return; + } + auto total_size_to_fetch = 0ul; std::vector< repl_req_ptr_t > next_batch_rreqs; auto const max_batch_size = HS_DYNAMIC_CONFIG(consensus.data_fetch_max_size_kb) * 1024ull; @@ -1415,6 +1426,7 @@ void RaftReplDev::check_and_fetch_remote_data(std::vector< repl_req_ptr_t > rreq next_batch_rreqs.emplace_back(rreq); } fetch_data_from_remote(std::move(next_batch_rreqs)); + decr_pending_request_num(); } void RaftReplDev::fetch_data_from_remote(std::vector< repl_req_ptr_t > rreqs) { @@ -1458,6 +1470,7 @@ void RaftReplDev::fetch_data_from_remote(std::vector< repl_req_ptr_t > rreqs) { // Fetch is fire-and-forget: detach() starts the coroutine and returns. Copy originator out of // rreqs.front() before rreqs is moved into the coroutine frame -- the reference would otherwise dangle. nuraft_mesg::svr_id_t const originator_id = originator; + incr_pending_request_num(); // released by fetch_data_coro once the fetch is fully handled sisl::async::detach(fetch_data_coro(std::move(builder), originator_id, std::move(rreqs))); } @@ -1498,6 +1511,7 @@ sisl::async::task< void > RaftReplDev::fetch_data_coro(shared< flatbuffers::Flat } } COUNTER_INCREMENT(m_metrics, fetch_err_cnt, 1); + decr_pending_request_num(); co_return; } @@ -1506,6 +1520,7 @@ sisl::async::task< void > RaftReplDev::fetch_data_coro(shared< flatbuffers::Flat iomanager.run_on_forget(iomgr::reactor_regex::random_worker, [this, r = std::move(response.value()), rreqs = std::move(rreqs)]() { handle_fetch_data_response(std::move(r), std::move(rreqs)); + decr_pending_request_num(); }); co_return; } diff --git a/src/lib/replication/service/raft_repl_service.cpp b/src/lib/replication/service/raft_repl_service.cpp index a1ff8ad99..de73db021 100644 --- a/src/lib/replication/service/raft_repl_service.cpp +++ b/src/lib/replication/service/raft_repl_service.cpp @@ -739,7 +739,7 @@ void RaftReplService::gc_repl_devs() { return; } - std::vector< group_id_t > groups_to_leave; + std::vector< shared< RaftReplDev > > rdevs_to_leave; { std::shared_lock lg(m_rd_map_mtx); for (auto it = m_rd_map.begin(); it != m_rd_map.end(); ++it) { @@ -750,7 +750,7 @@ void RaftReplService::gc_repl_devs() { LOGINFOMOD(replication, "repl_dev group_id={} was destroyed, shutting down the raft group in delayed fashion now", rdev->group_id()); - groups_to_leave.push_back(rdev->group_id()); + rdevs_to_leave.push_back(rdev); } } } @@ -758,7 +758,11 @@ void RaftReplService::gc_repl_devs() { // Call leave_group to shut down the raft server and destroy all resources on the repl dev. // This operation may require acquiring the m_rd_map_mtx lock for some steps (e.g., trigger cp flush). // Therefore, we perform it outside the lock scope and then remove group from m_rd_map. - for (const auto& group_id : groups_to_leave) { + for (auto const& rdev : rdevs_to_leave) { + auto const group_id = rdev->group_id(); + // Wait out in-flight requests on this dev (e.g. data fetches, which resolve peers through the raft server) + // before leave_group() frees the raft server. + rdev->stop(); m_msg_mgr->leave_group(group_id); // notify consumer to cleanup any resources associated with the listener itself; m_repl_app->destroy_repl_dev_listener(group_id); diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index a5404f08a..d19b081a4 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -1,5 +1,9 @@ cmake_minimum_required(VERSION 3.13) +# Run every test through stack_dump_launcher.sh (CMake >= 3.29 applies CMAKE_TEST_LAUNCHER to the test executables +# created below): a test that hangs prints the stacks of all its threads before the ctest timeout kills it. +set(CMAKE_TEST_LAUNCHER ${CMAKE_CURRENT_SOURCE_DIR}/stack_dump_launcher.sh) + include_directories (BEFORE ../include/) include_directories (BEFORE ../lib/) include_directories (BEFORE ../tests/) @@ -174,3 +178,10 @@ if (${non_coverage_build}) target_link_libraries(index_btree_benchmark homestore ${COMMON_TEST_DEPS} benchmark::benchmark) endif() +# ctest has no default timeout: give every test one, so a hung test fails and prints its output instead of running until +# the CI job is killed. +get_property(all_tests DIRECTORY PROPERTY TESTS) +if (all_tests) + set_tests_properties(${all_tests} PROPERTIES TIMEOUT 1800) +endif () + diff --git a/src/tests/stack_dump_launcher.sh b/src/tests/stack_dump_launcher.sh new file mode 100755 index 000000000..855026308 --- /dev/null +++ b/src/tests/stack_dump_launcher.sh @@ -0,0 +1,58 @@ +#!/bin/bash +# ctest launcher: runs a test, and if it is still running after HS_TEST_STACK_DUMP_SEC seconds (default 1500, below the +# ctest timeout) prints the stack of every thread of the test and of the processes it spawned, so the output of a hung +# test shows where it is stuck. If the test or a process it spawned crashes and leaves a core. file (CI sets +# kernel.core_pattern=core.%p), prints the stacks from that core. + +ulimit -c unlimited 2> /dev/null +start_marker=$(mktemp) +"$@" & +test_pid=$! +limit=${HS_TEST_STACK_DUMP_SEC:-1500} + +descendants() { + local p + for p in $(ps -o pid= --ppid "$1" 2>/dev/null); do + echo "$p" + descendants "$p" + done +} + +dump_stacks() { + local gdb=(gdb) + # attaching to a process that isn't our child needs ptrace rights (yama ptrace_scope), which sudo provides on CI + if [[ $EUID -ne 0 ]] && sudo -n true 2>/dev/null; then gdb=(sudo -n gdb); fi + local p + for p in $test_pid $(descendants $test_pid); do + echo "==================== stacks of pid $p: $(tr '\0' ' ' < /proc/$p/cmdline 2>/dev/null)" + "${gdb[@]}" -p "$p" -batch -ex "set pagination off" -ex "thread apply all bt" 2>&1 | grep -v "^\[New LWP" + done +} + +( + waited=0 + while (( waited < limit )); do + sleep 5 > /dev/null 2>&1 + kill -0 $test_pid 2> /dev/null || exit 0 + waited=$((waited + 5)) + done + echo "==================== test still running after ${limit}s, dumping stacks" + if command -v gdb > /dev/null; then dump_stacks; else echo "gdb not found, cannot dump stacks"; fi +) & +watcher_pid=$! + +wait $test_pid +rc=$? +kill $watcher_pid 2> /dev/null + +if command -v gdb > /dev/null; then + for core in $(find . -maxdepth 1 -name 'core.*' -newer "$start_marker" 2> /dev/null); do + exe=$(gdb -batch -c "$core" 2>&1 | sed -n "s/^Core was generated by \`\([^ ']*\).*/\1/p") + [[ -x "$exe" ]] || exe=$1 + echo "==================== stacks from $core ($exe)" + gdb -batch -ex "set pagination off" -ex "thread apply all bt" "$exe" "$core" 2>&1 | grep -v "^\[New LWP" + rm -f "$core" + done +fi +rm -f "$start_marker" +exit $rc diff --git a/src/tests/test_common/homestore_test_common.hpp b/src/tests/test_common/homestore_test_common.hpp index a11ea6536..b17b68ebd 100644 --- a/src/tests/test_common/homestore_test_common.hpp +++ b/src/tests/test_common/homestore_test_common.hpp @@ -142,6 +142,7 @@ struct Waiter { std::atomic< uint64_t > expected_comp{0}; std::atomic< uint64_t > actual_comp{0}; std::promise< void > comp_promise; + std::atomic_bool completed{false}; Waiter(uint64_t num_op) : expected_comp{num_op} {} Waiter() : Waiter{SISL_OPTIONS["num_io"].as< uint64_t >()} {} @@ -155,7 +156,19 @@ struct Waiter { } void one_complete() { - if ((actual_comp.fetch_add(1) + 1) >= expected_comp.load()) { comp_promise.set_value(); } + if ((actual_comp.fetch_add(1) + 1) >= expected_comp.load()) { complete(); } + } + + // Changes the expected number of completions once it is known (e.g. how many journal entries were left to replay + // after truncation); completes the wait if that many have already finished. + void set_expected(uint64_t num_op) { + expected_comp.store(num_op); + if (actual_comp.load() >= num_op) { complete(); } + } + +private: + void complete() { + if (!completed.exchange(true)) { comp_promise.set_value(); } } }; diff --git a/src/tests/test_common/hs_repl_test_common.hpp b/src/tests/test_common/hs_repl_test_common.hpp index b23bd5c85..c0058c12d 100644 --- a/src/tests/test_common/hs_repl_test_common.hpp +++ b/src/tests/test_common/hs_repl_test_common.hpp @@ -29,6 +29,7 @@ #include #include #include +#include #include #include @@ -51,7 +52,10 @@ SISL_OPTION_GROUP(test_repl_common_setup, "Internal replica num (used to launch multi process) - don't override", ::cxxopts::value< uint16_t >()->default_value("0"), "number"), (replica_dev_list, "", "replica_dev_list", "Device list for all replicas", - ::cxxopts::value< std::vector< std::string > >(), "path [...]")); + ::cxxopts::value< std::vector< std::string > >(), "path [...]"), + (wait_timeout_sec, "", "wait_timeout_sec", + "Max seconds a replica waits on the others (barriers, leader, commits, group destroy); 0 = forever", + ::cxxopts::value< uint32_t >()->default_value("600"), "seconds")); using namespace homestore; namespace bip = boost::interprocess; @@ -61,6 +65,31 @@ namespace test_common { ENUM(ipc_packet_op_t, uint32_t, WAKE_UP = 0, CLEAN_EXIT = 1, UNCLEAN_EXIT = 2, PEER_GOING_DOWN = 3); ENUM(repl_test_phase_t, uint32_t, REGISTER, MEMBER_START, TEST_RUN, VALIDATE, CLEANUP); +// Bounds the harness's waits on other replicas. If a peer crashes or a group never converges, the waiting replica fails +// with a description of what it was waiting for, instead of hanging until the CI job is killed. --wait_timeout_sec=0 +// waits forever (e.g. for long manual runs). +class wait_deadline { +public: + explicit wait_deadline(std::string what) : + what_{std::move(what)}, + timeout_sec_{SISL_OPTIONS["wait_timeout_sec"].as< uint32_t >()}, + start_{std::chrono::steady_clock::now()} {} + + // Call on every iteration of the wait loop; state describes what this replica sees right now. + void check(std::string const& state = {}) const { + if ((timeout_sec_ == 0) || (std::chrono::steady_clock::now() - start_ < std::chrono::seconds{timeout_sec_})) { + return; + } + RELEASE_ASSERT(false, "Timed out after {}s waiting for {}{}{}", timeout_sec_, what_, state.empty() ? "" : ": ", + state); + } + +private: + std::string what_; + uint32_t timeout_sec_; + std::chrono::steady_clock::time_point start_; +}; + class HSReplTestHelper : public HSTestHelper { protected: struct IPCData { @@ -69,6 +98,7 @@ class HSReplTestHelper : public HSTestHelper { bip::interprocess_mutex exec_mtx_; repl_test_phase_t phase_{repl_test_phase_t::REGISTER}; + uint64_t generation_{0}; // number of completed barriers uint32_t registered_count_{0}; uint32_t test_start_count_{0}; uint32_t verify_start_count_{0}; @@ -92,15 +122,34 @@ class HSReplTestHelper : public HSTestHelper { void sync_for(uint32_t& count, repl_test_phase_t new_phase, uint32_t max_count = 0) { if (max_count == 0) { max_count = SISL_OPTIONS["replicas"].as< uint32_t >(); } std::unique_lock< bip::interprocess_mutex > lg(mtx_); - ++count; - if (count == max_count) { + // Wait for this barrier to complete, not for phase_: consecutive barriers can share a phase (e.g. several + // register_listener() calls in a row), and a replica must not pass one because the previous barrier already + // set that phase. Only the last arriver resets count, so a slow waker can't wipe out an early arrival at + // the next barrier. + auto const my_generation = generation_; + if (++count == max_count) { + count = 0; phase_ = new_phase; + ++generation_; cv_.notify_all(); } else { - cv_.wait(lg, [this, new_phase]() { return (phase_ == new_phase); }); + auto const reached = [this, my_generation]() { return (generation_ != my_generation); }; + auto const timeout_sec = SISL_OPTIONS["wait_timeout_sec"].as< uint32_t >(); + if (timeout_sec == 0) { + cv_.wait(lg, reached); + } else if (!cv_.timed_wait(lg, + boost::posix_time::microsec_clock::universal_time() + + boost::posix_time::seconds(timeout_sec), + reached)) { + auto const arrived = count; + // The mutex lives in shared memory and isn't robust: release it before failing, so the other + // replicas can still time out instead of blocking on it forever. + lg.unlock(); + RELEASE_ASSERT(false, + "Timed out after {}s at the barrier for phase={}: only {} of {} replicas arrived", + timeout_sec, enum_name(new_phase), arrived, max_count); + } } - - count = 0; } }; @@ -291,7 +340,8 @@ class HSReplTestHelper : public HSTestHelper { void register_listener(std::shared_ptr< repl_dev_listener > listener) { if (replica_num_ != 0) { pending_listeners_.emplace_back(std::move(listener)); } - ipc_data_->sync_for_member_start(); + // Every launched process registers its listener here, spare replicas included. + ipc_data_->sync_for_member_start(num_replicas_); if (replica_num_ == 0) { std::set< homestore::replica_id_t > members; diff --git a/src/tests/test_common/raft_repl_test_base.hpp b/src/tests/test_common/raft_repl_test_base.hpp index 647fb393c..b2b55ebf1 100644 --- a/src/tests/test_common/raft_repl_test_base.hpp +++ b/src/tests/test_common/raft_repl_test_base.hpp @@ -530,7 +530,10 @@ class RaftReplDevTestBase : public testing::Test { if (!repl_dev) continue; int i = 0; bool force_leave = false; + test_common::wait_deadline const deadline{ + fmt::format("repl dev of group={} to get destroyed", boost::uuids::to_string(repl_dev->group_id()))}; do { + deadline.check(); std::this_thread::sleep_for(std::chrono::seconds(1)); auto& raft_repl_svc = dynamic_cast< RaftReplService& >(hs()->repl_service()); raft_repl_svc.gc_repl_devs(); @@ -559,6 +562,7 @@ class RaftReplDevTestBase : public testing::Test { void wait_for_commits(uint64_t exp_writes) { uint64_t total_writes{0}; + test_common::wait_deadline const deadline{fmt::format("{} commits", exp_writes)}; while (true) { total_writes = 0; for (auto const& db : dbs_) { @@ -566,6 +570,7 @@ class RaftReplDevTestBase : public testing::Test { } if (total_writes >= exp_writes) { break; } + deadline.check(fmt::format("received {}", total_writes)); std::this_thread::sleep_for(std::chrono::milliseconds(1000)); LOGINFO("Replica={} received {} commits but expected {}", g_helper->replica_num(), total_writes, exp_writes); @@ -583,9 +588,11 @@ class RaftReplDevTestBase : public testing::Test { void assign_leader(uint16_t replica) { LOGINFO("Switch the leader to replica_num = {}", replica); + test_common::wait_deadline const deadline{fmt::format("replica={} to become leader", replica)}; if (g_helper->replica_num() == replica) { for (auto const& db : dbs_) { do { + deadline.check("become_leader() keeps failing on this replica"); auto result = sisl::async::sync_get(db->device()->become_leader()); if (!result) { std::this_thread::sleep_for(std::chrono::milliseconds(1000)); @@ -600,6 +607,9 @@ class RaftReplDevTestBase : public testing::Test { while (true) { leader_uuid = db->device()->get_leader_id(); if (!leader_uuid.is_nil() && (g_helper->member_id(leader_uuid) == replica)) { break; } + deadline.check(fmt::format("group={} leader seen by this replica={}", + boost::uuids::to_string(db->device()->group_id()), + boost::uuids::to_string(leader_uuid))); LOGINFO("Waiting for replica={} to become leader", replica); std::this_thread::sleep_for(std::chrono::milliseconds{500}); @@ -614,10 +624,13 @@ class RaftReplDevTestBase : public testing::Test { return; } + test_common::wait_deadline const deadline{ + fmt::format("a leader for group={}", boost::uuids::to_string(db->device()->group_id()))}; do { auto leader_uuid = db->device()->get_leader_id(); if (leader_uuid.is_nil()) { + deadline.check(); LOGINFO("Waiting for leader to be elected for group={}", db->device()->group_id()); std::this_thread::sleep_for(std::chrono::milliseconds{500}); } else if (leader_uuid == g_helper->my_replica_id()) { @@ -633,11 +646,13 @@ class RaftReplDevTestBase : public testing::Test { uint64_t* data_size = nullptr) { if (dbs_[0]->device() == nullptr) return; + test_common::wait_deadline const deadline{"a leader to write on"}; do { auto repl_dev = dbs_[0]->device(); auto leader_uuid = repl_dev->get_leader_id(); if (leader_uuid.is_nil()) { + deadline.check(); LOGINFO("Waiting for leader to be elected"); std::this_thread::sleep_for(std::chrono::milliseconds{500}); } else if (leader_uuid == g_helper->my_replica_id()) { @@ -670,9 +685,11 @@ class RaftReplDevTestBase : public testing::Test { if (wait_for_commit) { this->wait_for_all_commits(); } } replica_id_t wait_and_get_leader_id() { + test_common::wait_deadline const deadline{"a leader to be elected"}; do { auto leader_uuid = dbs_[0]->device()->get_leader_id(); if (leader_uuid.is_nil()) { + deadline.check(); LOGINFO("Waiting for leader to be elected"); std::this_thread::sleep_for(std::chrono::milliseconds{500}); } else { @@ -748,9 +765,11 @@ class RaftReplDevTestBase : public testing::Test { } void wait_for_listener_destroy(uint64_t exp_listeners) { + test_common::wait_deadline const deadline{fmt::format("{} listeners to remain", exp_listeners)}; while (true) { auto total_listeners = g_helper->num_listeners(); if (total_listeners == exp_listeners) { break; } + deadline.check(fmt::format("{} listeners", total_listeners)); std::this_thread::sleep_for(std::chrono::milliseconds(100)); } } @@ -819,7 +838,10 @@ class RaftReplDevTestBase : public testing::Test { void remove_member(std::shared_ptr< TestReplicatedDB > db, replica_id_t member_id) { this->run_on_leader(db, [this, db, member_id]() { LOGINFO("remove member, member={}", boost::uuids::to_string(member_id)); + test_common::wait_deadline const deadline{ + fmt::format("remove_member of member={}", boost::uuids::to_string(member_id))}; while (true) { + deadline.check(); auto result = sisl::async::sync_get(hs()->repl_service().remove_member(db->device()->group_id(), member_id, 0)); if (result.has_value() || result.error() == ReplServiceError::OK) { @@ -894,7 +916,9 @@ class RaftReplDevTestBase : public testing::Test { LOGINFO("check replace member rollback result, task_id={}, out={} in={}", task_id, boost::uuids::to_string(member_out), boost::uuids::to_string(member_in)); auto rdev = std::dynamic_pointer_cast< RaftReplDev >(db->device()); + test_common::wait_deadline const deadline{fmt::format("replace member task={} to roll back", task_id)}; while (true) { + deadline.check(); auto ret = rdev->get_ongoing_replace_member_task(); if (ret.has_value()) { LOGDEBUG("replace member task still exists"); diff --git a/src/tests/test_index_crash_recovery.cpp b/src/tests/test_index_crash_recovery.cpp index 40f908754..e7722a98f 100644 --- a/src/tests/test_index_crash_recovery.cpp +++ b/src/tests/test_index_crash_recovery.cpp @@ -1313,8 +1313,11 @@ TYPED_TEST(IndexCrashTest, DestroyTableWithPendingCpCrash) { // Step 5: Trigger the CP (do not wait). async_cp_flush will: // (a) write the txn_journal to disk (entries include the destroyed table's ordinal), // (b) start flushing nodes, hit the buf with m_crash_flag_on → simulated crash. + // A CP that started on its own after Step 3 may already have flushed the flagged buf and begun the simulated + // crash. A CP triggered now would then fail against the shutting-down homestore and abort the test, so only + // trigger one if the crash hasn't started yet; Step 6 waits for the recovery either way. LOGINFO("Step 5: Trigger CP (crash expected after journal is written)"); - test_common::HSTestHelper::trigger_cp(false /* no wait */); + if (!hs()->crash_simulator().is_crashed()) { test_common::HSTestHelper::trigger_cp(false /* no wait */); } // Step 6: Wait for the simulated crash and homestore recovery. // Without fix : repair_index_node / sanity_check asserts → process aborts (SIGABRT). diff --git a/src/tests/test_log_store.cpp b/src/tests/test_log_store.cpp index 172dd92a6..b53b55e52 100644 --- a/src/tests/test_log_store.cpp +++ b/src/tests/test_log_store.cpp @@ -620,8 +620,16 @@ class LogStoreTest : public ::testing::Test { void wait_for_inserts() { std::unique_lock< std::mutex > lk{m_pending_mtx}; - m_pending_cv.wait(lk, - [&] { return (m_nrecords_waiting_to_issue == 0) && (m_nrecords_waiting_to_complete == 0); }); + // The test logdevs only flush explicitly. Each batch asks for a flush with flush_if_necessary(), which skips + // the flush if another one holds the flush lock (e.g. a truncate running in parallel to the writes), and + // nothing would flush those records later. So keep flushing until all inserts complete. + while (!m_pending_cv.wait_for(lk, std::chrono::milliseconds{100}, [&] { + return (m_nrecords_waiting_to_issue == 0) && (m_nrecords_waiting_to_complete == 0); + })) { + lk.unlock(); + flush(); + lk.lock(); + } } void read_validate(bool expect_all_completed = false) { diff --git a/src/tests/test_solo_repl_dev.cpp b/src/tests/test_solo_repl_dev.cpp index 18498687d..449414d6b 100644 --- a/src/tests/test_solo_repl_dev.cpp +++ b/src/tests/test_solo_repl_dev.cpp @@ -14,6 +14,8 @@ * *********************************************************************************/ #include +#include +#include #include #include #include @@ -100,6 +102,7 @@ class SoloReplDevTest : public testing::Test { if (ctx == nullptr) { m_test.validate_replay(*device(), lsn, header, key, blkids); } else { + m_test.on_written(device()->group_id(), lsn); auto req = boost::static_pointer_cast< test_repl_req >(ctx); req->written_blkids = std::move(blkids); m_test.on_write_complete(*device(), req); @@ -169,6 +172,9 @@ class SoloReplDevTest : public testing::Test { protected: test_common::Runner m_io_runner; test_common::Waiter m_task_waiter; + std::atomic< uint64_t > m_replayed{0}; // journal entries replayed by the last restart + std::mutex m_written_mtx; + std::map< group_id_t, int64_t > m_last_written_lsn; // per repl dev shared< repl_dev > m_repl_dev1; shared< repl_dev > m_repl_dev2; uuid_t m_uuid1; @@ -203,10 +209,36 @@ class SoloReplDevTest : public testing::Test { void restart() { m_repl_dev1.reset(); m_repl_dev2.reset(); + m_replayed = 0; m_helper.restart_homestore(); m_repl_dev1 = hs()->repl_service().get_repl_dev(m_uuid1).value(); m_repl_dev2 = hs()->repl_service().get_repl_dev(m_uuid2).value(); + + // Journal entries are replayed while homestore starts, except the ones a CP already truncated: with checkpoints + // every second, a slow run truncates part of the journal before the restart. Every write after the truncation + // point must have been replayed. + uint64_t expected_replays{0}; + for (auto const& rdev : {m_repl_dev1, m_repl_dev2}) { + auto const [trunc_lsn, trunc_ld_key, tail_lsn] = + std::dynamic_pointer_cast< SoloReplDev >(rdev)->m_data_journal->truncate_info(); + expected_replays += std::max< int64_t >(last_written_lsn(rdev->group_id()) - trunc_lsn, 0); + } + EXPECT_EQ(m_replayed.load(), expected_replays) << "journal entries replayed after the restart"; + // validate the replayed entries, without waiting for ones that won't come if the counts differ + m_task_waiter.set_expected(m_replayed.load()); + } + + void on_written(group_id_t group_id, int64_t lsn) { + std::lock_guard lg{m_written_mtx}; + auto [it, inserted] = m_last_written_lsn.try_emplace(group_id, lsn); + if (!inserted) { it->second = std::max(it->second, lsn); } + } + + int64_t last_written_lsn(group_id_t group_id) { + std::lock_guard lg{m_written_mtx}; + auto const it = m_last_written_lsn.find(group_id); + return (it == m_last_written_lsn.end()) ? -1 : it->second; } void write_io(uint32_t key_size, uint64_t data_size, uint32_t max_size_per_iov) { @@ -275,6 +307,7 @@ class SoloReplDevTest : public testing::Test { void validate_replay(repl_dev& rdev, int64_t lsn, sisl::blob const& header, sisl::blob const& key, std::vector< multi_blk_id > const& blkids) { + ++m_replayed; if (blkids.empty()) { m_task_waiter.one_complete(); return;