hanzoai/datastore
2
0
Fork 0
mirror of https://github.com/hanzoai/datastore.git synced 2026-09-26 13:55:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #106353 from ClickHouse/asya/revert_add_final

Browse Source 
Revert TaskTracker::addFinal changes
This commit is contained in:
Alexey Milovidov
2026-06-02 22:13:11 -07:00
committed by GitHub
parent 7272570617 eb5178b60d
commit 3fa1ed2d9d
5 changed files with 20 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Common/ThreadPoolTaskTracker.cpp
+6 -58
View File
@@ -104,12 +104,6 @@ void TaskTracker::waitIfAny()
void TaskTracker::add(Callback && func)
{
{
std::lock_guard lock(mutex);
chassert(!final_task_added && "add must not be called after addFinal");
++tasks_added;
}
/// All this fuzz is about 2 things. This is the most critical place of TaskTracker.
/// The first is not to fail insertion in the list `futures`.
/// In order to face it, the element is allocated at the end of the list `futures` in advance.
@@ -124,33 +118,14 @@ void TaskTracker::add(Callback && func)
/// preallocation for the second issue
FinishedList pre_allocated_finished {future_placeholder};
Callback func_with_notification = [this, my_func = std::move(func), my_pre_allocated_finished = std::move(pre_allocated_finished)]() mutable
Callback func_with_notification = [&, my_func = std::move(func), my_pre_allocated_finished = std::move(pre_allocated_finished)]() mutable
{
SCOPE_EXIT({
std::shared_ptr<std::packaged_task<void()>> maybe_final_task;
{
DENY_ALLOCATIONS_IN_SCOPE;
std::lock_guard lock(mutex);
finished_futures.splice(finished_futures.end(), my_pre_allocated_finished);
++tasks_finished;
if (final_task && tasks_finished == tasks_added)
{
maybe_final_task = std::move(final_task);
}
has_finished.notify_one();
}
if (maybe_final_task)
{
try
{
scheduler([pt = std::move(maybe_final_task)]() mutable { (*pt)(); }, Priority{});
}
catch (...)
{
/// SCOPE_EXIT is noexcept; let the final task's future report broken_promise.
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
DENY_ALLOCATIONS_IN_SCOPE;
std::lock_guard lock(mutex);
finished_futures.splice(finished_futures.end(), my_pre_allocated_finished);
has_finished.notify_one();
});
my_func();
@@ -162,33 +137,6 @@ void TaskTracker::add(Callback && func)
waitTilInflightShrink();
}
void TaskTracker::addFinal(Callback && func)
{
auto pt = std::make_shared<std::packaged_task<void()>>(std::move(func));
futures.emplace_back(pt->get_future());
bool run_final_task_now = false;
{
std::lock_guard lock(mutex);
chassert(!final_task_added && "addFinal must be called at most once");
final_task_added = true;
if (tasks_finished == tasks_added)
{
/// Every previously added task has already finished.
/// There will be no SCOPE_EXIT to trigger the final callback, so run it here.
run_final_task_now = true;
}
else
{
final_task = pt;
}
}
if (run_final_task_now)
{
scheduler([p = std::move(pt)]() mutable { (*p)(); }, Priority{});
}
}
void TaskTracker::waitTilInflightShrink()
{
if (!max_tasks_inflight)
src/Common/ThreadPoolTaskTracker.h
-17
View File
@@ -39,11 +39,6 @@ public:
void add(Callback && func);
/// Must be called at most once, after all `add()` calls.
/// The callback runs on a worker thread once every previously-added task has completed.
/// May or may not run if prior tasks threw.
void addFinal(Callback && func);
private:
/// waitTilInflightShrink waits til the number of in-flight tasks beyond the limit `max_tasks_inflight`.
void waitTilInflightShrink() TSA_NO_THREAD_SAFETY_ANALYSIS;
@@ -62,18 +57,6 @@ private:
std::condition_variable has_finished TSA_GUARDED_BY(mutex);
using FinishedList = std::list<FutureList::iterator>;
FinishedList finished_futures TSA_GUARDED_BY(mutex);
/// Monotonic counters of regular tasks submitted via add (excludes the final task).
/// Worker threads compare these to decide when the final task is ready, so they
/// never need to read `futures`.
size_t tasks_added TSA_GUARDED_BY(mutex) = 0;
size_t tasks_finished TSA_GUARDED_BY(mutex) = 0;
/// A packaged task for the callback added by addFinal. A non-null value means
/// the callback has been added, but not yet scheduled.
std::shared_ptr<std::packaged_task<void()>> final_task TSA_GUARDED_BY(mutex);
bool final_task_added = false;
};
}
src/Common/tests/gtest_thread_pool_task_tracker.cpp
-203
View File
@@ -1,203 +0,0 @@
#include <Common/ThreadPoolTaskTracker.h>
#include <Common/setThreadName.h>
#include <gtest/gtest.h>
namespace CurrentMetrics
{
extern const Metric LocalThread;
extern const Metric LocalThreadActive;
extern const Metric LocalThreadScheduled;
}
using namespace DB;
namespace
{
LogSeriesLimiterPtr makeTestLogger()
{
return std::make_shared<LogSeriesLimiter>(getLogger("TaskTrackerTest"), 1, 5);
}
struct AsyncTracker
{
ThreadPool pool;
TaskTracker tracker;
explicit AsyncTracker(size_t threads, size_t max_inflight = 0)
: pool(CurrentMetrics::LocalThread, CurrentMetrics::LocalThreadActive, CurrentMetrics::LocalThreadScheduled, threads)
, tracker(threadPoolCallbackRunnerUnsafe<void>(pool, ThreadName::UNKNOWN), max_inflight, makeTestLogger())
{}
};
}
TEST(TaskTrackerAddFinal, AsyncHappyPathRunsFinalAfterAllPriors)
{
AsyncTracker t{/*threads=*/4};
constexpr size_t N = 16;
std::atomic<size_t> ran{0};
std::atomic<bool> final_ran{false};
std::atomic<size_t> count_observed_by_final{0};
for (size_t i = 0; i < N; ++i)
t.tracker.add([&] {
std::this_thread::sleep_for(std::chrono::milliseconds(5));
ran.fetch_add(1);
});
t.tracker.addFinal([&] {
count_observed_by_final = ran.load();
final_ran = true;
});
t.tracker.waitAll();
EXPECT_EQ(ran.load(), N);
EXPECT_TRUE(final_ran.load());
EXPECT_EQ(count_observed_by_final.load(), N);
}
TEST(TaskTrackerAddFinal, AsyncZeroPriorsStillRunsFinal)
{
AsyncTracker t{/*threads=*/2};
std::atomic<bool> final_ran{false};
t.tracker.addFinal([&] { final_ran = true; });
t.tracker.waitAll();
EXPECT_TRUE(final_ran.load());
}
TEST(TaskTrackerAddFinal, AsyncAllPriorsAlreadyDoneBeforeAddFinal)
{
AsyncTracker t{/*threads=*/2};
constexpr size_t N = 4;
std::atomic<size_t> ran{0};
std::atomic<bool> final_ran{false};
for (size_t i = 0; i < N; ++i)
t.tracker.add([&] { ran.fetch_add(1); });
/// Give workers a chance to finish before we call addFinal — exercises the path where
/// `finished_futures.size() == futures.size()` is already true at the time addFinal is called.
while (ran.load() != N)
std::this_thread::sleep_for(std::chrono::milliseconds(1));
t.tracker.addFinal([&] { final_ran = true; });
t.tracker.waitAll();
EXPECT_TRUE(final_ran.load());
}
TEST(TaskTrackerAddFinal, AsyncPriorFailureDoesNotDeadlock)
{
AsyncTracker t{/*threads=*/2};
t.tracker.add([] { /* ok */ });
t.tracker.add([] { throw std::runtime_error("boom"); });
t.tracker.add([] { /* ok */ });
t.tracker.addFinal([] {});
EXPECT_THROW(t.tracker.waitAll(), std::runtime_error);
/// The final callback is best-effort on failures — we only assert no deadlock and clean shutdown.
t.tracker.safeWaitAll();
}
TEST(TaskTrackerAddFinal, AsyncFinalCallbackException)
{
AsyncTracker t{/*threads=*/2};
t.tracker.add([] { /* ok */ });
t.tracker.addFinal([] { throw std::runtime_error("final boom"); });
EXPECT_THROW(t.tracker.waitAll(), std::runtime_error);
t.tracker.safeWaitAll();
}
TEST(TaskTrackerAddFinal, WaitIfAnyBeforeFinalTask)
{
TaskTracker tracker(/*scheduler_=*/{}, /*max_tasks_inflight=*/0, makeTestLogger());
ASSERT_FALSE(tracker.isAsync());
constexpr size_t N = 2;
std::atomic<size_t> ran{0};
std::atomic<bool> final_ran{false};
for (size_t i = 0; i < N; ++i)
tracker.add([&] { ran.fetch_add(1); });
/// Sync runner has already run every task body and populated `finished_futures`.
ASSERT_EQ(ran.load(), N);
tracker.waitIfAny();
tracker.addFinal([&] { final_ran = true; });
tracker.waitAll();
EXPECT_EQ(ran.load(), N);
EXPECT_TRUE(final_ran.load());
}
TEST(TaskTrackerAddFinal, AsyncWithInflightLimit)
{
/// Exercise the path where `waitTilInflightShrink` is hit on the owner thread between adds.
AsyncTracker t{/*threads=*/4, /*max_inflight=*/3};
constexpr size_t N = 32;
std::atomic<size_t> ran{0};
std::atomic<bool> final_ran{false};
for (size_t i = 0; i < N; ++i)
t.tracker.add([&] {
std::this_thread::sleep_for(std::chrono::milliseconds(2));
ran.fetch_add(1);
});
size_t observed_in_final = 0;
t.tracker.addFinal([&] {
observed_in_final = ran.load();
final_ran = true;
});
t.tracker.waitAll();
EXPECT_EQ(ran.load(), N);
EXPECT_TRUE(final_ran.load());
EXPECT_EQ(observed_in_final, N);
}
TEST(TaskTrackerAddFinal, SyncRunnerHappyPath)
{
TaskTracker tracker(/*scheduler_=*/{}, /*max_tasks_inflight=*/0, makeTestLogger());
ASSERT_FALSE(tracker.isAsync());
constexpr size_t N = 5;
std::atomic<size_t> ran{0};
std::atomic<bool> final_ran{false};
size_t observed_in_final = 0;
for (size_t i = 0; i < N; ++i)
tracker.add([&] { ran.fetch_add(1); });
tracker.addFinal([&] {
observed_in_final = ran.load();
final_ran = true;
});
tracker.waitAll();
EXPECT_EQ(ran.load(), N);
EXPECT_TRUE(final_ran.load());
EXPECT_EQ(observed_in_final, N);
}
TEST(TaskTrackerAddFinal, SyncRunnerPriorFailureDoesNotDeadlock)
{
TaskTracker tracker(/*scheduler_=*/{}, /*max_tasks_inflight=*/0, makeTestLogger());
tracker.add([] { /* ok */ });
tracker.add([] { throw std::runtime_error("boom"); });
tracker.addFinal([] {});
EXPECT_THROW(tracker.waitAll(), std::runtime_error);
tracker.safeWaitAll();
}
src/IO/WriteBufferFromS3.cpp
+13 -12
View File
@@ -195,11 +195,6 @@ void WriteBufferFromS3::preFinalize()
else
{
writeMultipartUpload();
task_tracker->addFinal([this]()
{
if (completeMultipartUpload())
multipart_upload_finished = true;
});
}
}
@@ -224,6 +219,12 @@ void WriteBufferFromS3::finalizeImpl()
span.addAttributeIfNotZero("clickhouse.multipart_upload_parts", multipart_tags.size());
if (!multipart_upload_id.empty())
{
completeMultipartUpload();
multipart_upload_finished = true;
}
if (request_settings[S3RequestSetting::check_objects_after_upload])
{
S3::checkObjectExists(*client_ptr, bucket, key, {}, "Immediately after upload");
@@ -612,7 +613,7 @@ void WriteBufferFromS3::writePart(WriteBufferFromS3::PartData && data)
task_tracker->add(std::move(upload_worker));
}
bool WriteBufferFromS3::completeMultipartUpload()
void WriteBufferFromS3::completeMultipartUpload()
{
LOG_TEST(limited_log, "Completing multipart upload. {}, Parts: {}", getShortLogDetails(), multipart_tags.size());
@@ -621,13 +622,13 @@ bool WriteBufferFromS3::completeMultipartUpload()
ErrorCodes::LOGICAL_ERROR,
"Failed to complete multipart upload. No parts have uploaded");
for (const auto & tag : multipart_tags)
for (size_t i = 0; i < multipart_tags.size(); ++i)
{
const auto tag = multipart_tags.at(i);
if (tag.empty())
{
// One of the earlier uploads failed.
return false;
}
throw Exception(
ErrorCodes::LOGICAL_ERROR,
"Failed to complete multipart upload. Part {} haven't been uploaded.", i);
}
S3::CompleteMultipartUploadRequest req;
@@ -674,7 +675,7 @@ bool WriteBufferFromS3::completeMultipartUpload()
if (outcome.IsSuccess())
{
LOG_TRACE(limited_log, "Multipart upload has completed. {}, Parts: {}", getShortLogDetails(), multipart_tags.size());
return true;
return;
}
ProfileEvents::increment(ProfileEvents::WriteBufferFromS3RequestsErrors, 1);
src/IO/WriteBufferFromS3.h
+1 -1
View File
@@ -72,7 +72,7 @@ private:
void writePart(PartData && data);
void writeMultipartUpload();
void createMultipartUpload();
bool completeMultipartUpload();
void completeMultipartUpload();
void abortMultipartUpload();
void tryToAbortMultipartUpload() noexcept;