LibThreading: Remove BackgroundAction
Remove BackgroundAction after ImageDecoder stopped using it. Drop its dedicated test target and sources from LibThreading. Replace the LibCore stream tests' remaining usage with explicit Thread objects so the tree no longer depends on the removed API.
This commit is contained in:
Andreas Kling
Andreas Kling
parent
88dfee5528
commit
3c0d0ecc28
6 changed files with 13 additions and 315 deletions
|
|
@ -1,86 +0,0 @@
|
|||
/*
|
||||
* Copyright (c) 2019-2020, Sergey Bugaev <bugaevc@serenityos.org>
|
||||
* Copyright (c) 2021, Andreas Kling <andreas@ladybird.org>
|
||||
*
|
||||
* SPDX-License-Identifier: BSD-2-Clause
|
||||
*/
|
||||
|
||||
#include <AK/Queue.h>
|
||||
#include <LibSync/Mutex.h>
|
||||
#include <LibThreading/BackgroundAction.h>
|
||||
#include <LibThreading/Thread.h>
|
||||
|
||||
static pthread_mutex_t s_mutex = PTHREAD_MUTEX_INITIALIZER;
|
||||
static pthread_cond_t s_condition = PTHREAD_COND_INITIALIZER;
|
||||
static Queue<Function<void()>>* s_all_actions;
|
||||
static Threading::Thread* s_background_thread;
|
||||
static Atomic<bool> s_background_thread_should_run = true;
|
||||
|
||||
static intptr_t background_thread_func()
|
||||
{
|
||||
Vector<Function<void()>> actions;
|
||||
while (s_background_thread_should_run.load(AK::MemoryOrder::memory_order_acquire)) {
|
||||
pthread_mutex_lock(&s_mutex);
|
||||
|
||||
while (s_all_actions->is_empty() && s_background_thread_should_run.load(AK::MemoryOrder::memory_order_acquire))
|
||||
pthread_cond_wait(&s_condition, &s_mutex);
|
||||
|
||||
while (!s_all_actions->is_empty())
|
||||
actions.append(s_all_actions->dequeue());
|
||||
|
||||
pthread_mutex_unlock(&s_mutex);
|
||||
|
||||
for (auto& action : actions) {
|
||||
if (s_background_thread_should_run.load(AK::MemoryOrder::memory_order_acquire))
|
||||
action();
|
||||
}
|
||||
actions.clear();
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void init()
|
||||
{
|
||||
s_all_actions = new Queue<Function<void()>>;
|
||||
s_background_thread = &Threading::Thread::construct("Background"sv, background_thread_func).leak_ref();
|
||||
s_background_thread->start();
|
||||
}
|
||||
|
||||
void Threading::quit_background_thread()
|
||||
{
|
||||
if (!s_background_thread)
|
||||
return;
|
||||
|
||||
s_background_thread_should_run.store(false, AK::MemoryOrder::memory_order_release);
|
||||
|
||||
pthread_mutex_lock(&s_mutex);
|
||||
pthread_cond_broadcast(&s_condition);
|
||||
pthread_mutex_unlock(&s_mutex);
|
||||
|
||||
MUST(s_background_thread->join());
|
||||
|
||||
delete s_all_actions;
|
||||
s_background_thread->unref();
|
||||
s_all_actions = nullptr;
|
||||
s_background_thread = nullptr;
|
||||
|
||||
s_background_thread_should_run.store(true, AK::MemoryOrder::memory_order_release);
|
||||
}
|
||||
|
||||
Threading::Thread& Threading::BackgroundActionBase::background_thread()
|
||||
{
|
||||
if (s_background_thread == nullptr)
|
||||
init();
|
||||
return *s_background_thread;
|
||||
}
|
||||
|
||||
void Threading::BackgroundActionBase::enqueue_work(Function<void()> work)
|
||||
{
|
||||
if (s_all_actions == nullptr)
|
||||
init();
|
||||
|
||||
pthread_mutex_lock(&s_mutex);
|
||||
s_all_actions->enqueue(move(work));
|
||||
pthread_cond_broadcast(&s_condition);
|
||||
pthread_mutex_unlock(&s_mutex);
|
||||
}
|
||||
|
|
@ -1,97 +0,0 @@
|
|||
/*
|
||||
* Copyright (c) 2019-2020, Sergey Bugaev <bugaevc@serenityos.org>
|
||||
* Copyright (c) 2021, Andreas Kling <andreas@ladybird.org>
|
||||
* Copyright (c) 2022-2023, the SerenityOS developers.
|
||||
*
|
||||
* SPDX-License-Identifier: BSD-2-Clause
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
#include <AK/Atomic.h>
|
||||
#include <AK/Function.h>
|
||||
#include <AK/NonnullRefPtr.h>
|
||||
#include <AK/Optional.h>
|
||||
#include <AK/Queue.h>
|
||||
#include <LibCore/Event.h>
|
||||
#include <LibCore/EventLoop.h>
|
||||
#include <LibCore/EventReceiver.h>
|
||||
#include <LibCore/Promise.h>
|
||||
#include <LibThreading/Forward.h>
|
||||
|
||||
namespace Threading {
|
||||
|
||||
template<typename Result>
|
||||
class BackgroundAction;
|
||||
|
||||
class BackgroundActionBase {
|
||||
template<typename Result>
|
||||
friend class BackgroundAction;
|
||||
|
||||
private:
|
||||
BackgroundActionBase() = default;
|
||||
|
||||
static void enqueue_work(ESCAPING Function<void()>);
|
||||
static Thread& background_thread();
|
||||
};
|
||||
|
||||
template<typename Result>
|
||||
class BackgroundAction final
|
||||
: public Core::EventReceiver
|
||||
, private BackgroundActionBase {
|
||||
C_OBJECT(BackgroundAction);
|
||||
|
||||
public:
|
||||
virtual ~BackgroundAction() = default;
|
||||
|
||||
Optional<Result> const& result() const { return m_result; }
|
||||
Optional<Result>& result() { return m_result; }
|
||||
|
||||
// Cancellation is a best-effort cross-thread signal. No other state is protected by this flag.
|
||||
// It is not used to synchronize access to any other state (m_result), so relaxed atomics are fine.
|
||||
void cancel() { m_canceled.store(true, AK::MemoryOrder::memory_order_relaxed); }
|
||||
// If your action is long-running, you should periodically check the cancel state and possibly return early.
|
||||
bool is_canceled() const { return m_canceled.load(AK::MemoryOrder::memory_order_relaxed); }
|
||||
|
||||
private:
|
||||
BackgroundAction(ESCAPING Function<ErrorOr<Result>(BackgroundAction&)> action, ESCAPING Function<void(Result)> on_complete, ESCAPING Function<void(Error)> on_error = {})
|
||||
: m_action(move(action))
|
||||
, m_on_complete(move(on_complete))
|
||||
, m_on_error(move(on_error))
|
||||
{
|
||||
enqueue_work([self = NonnullRefPtr(*this), origin_event_loop = Core::EventLoop::current_weak()]() mutable {
|
||||
auto result = self->m_action(*self);
|
||||
|
||||
auto event_loop = origin_event_loop->take();
|
||||
if (!event_loop) {
|
||||
dbgln("BackgroundAction {:p} was dropped, the origin loop is gone.", self.ptr());
|
||||
return;
|
||||
}
|
||||
event_loop->deferred_invoke([self = move(self), result = move(result)]() mutable {
|
||||
auto const canceled = self->m_canceled.load(AK::MemoryOrder::memory_order_relaxed);
|
||||
|
||||
if (canceled)
|
||||
return;
|
||||
|
||||
if (result.is_error()) {
|
||||
if (self->m_on_error)
|
||||
self->m_on_error(result.release_error());
|
||||
return;
|
||||
}
|
||||
|
||||
if (self->m_on_complete)
|
||||
self->m_on_complete(result.release_value());
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
Function<ErrorOr<Result>(BackgroundAction&)> m_action;
|
||||
Function<void(Result)> m_on_complete;
|
||||
Function<void(Error)> m_on_error;
|
||||
Optional<Result> m_result;
|
||||
Atomic<bool> m_canceled { false };
|
||||
};
|
||||
|
||||
void quit_background_thread();
|
||||
|
||||
}
|
||||
|
|
@ -1,5 +1,4 @@
|
|||
set(SOURCES
|
||||
BackgroundAction.cpp
|
||||
Thread.cpp
|
||||
ThreadPool.cpp
|
||||
)
|
||||
|
|
|
|||
|
|
@ -17,7 +17,7 @@
|
|||
#include <LibCore/Timer.h>
|
||||
#include <LibCore/UDPServer.h>
|
||||
#include <LibTest/TestCase.h>
|
||||
#include <LibThreading/BackgroundAction.h>
|
||||
#include <LibThreading/Thread.h>
|
||||
#include <fcntl.h>
|
||||
|
||||
#include <AK/Windows.h>
|
||||
|
|
@ -360,8 +360,9 @@ TEST_CASE(local_socket_read)
|
|||
// impasse. LocalSocket::connect blocks because there's nobody to
|
||||
// accept, and LocalServer::accept blocks because there's nobody
|
||||
// connected.
|
||||
auto background_action = Threading::BackgroundAction<int>::construct(
|
||||
[&socket_path](auto&) {
|
||||
auto client_thread = Threading::Thread::construct(
|
||||
"LocalSocketRead"sv,
|
||||
[socket_path] {
|
||||
Core::EventLoop event_loop;
|
||||
|
||||
auto client_socket = MUST(Core::LocalSocket::connect(socket_path));
|
||||
|
|
@ -381,10 +382,11 @@ TEST_CASE(local_socket_read)
|
|||
EXPECT_EQ(sent_data, received_data);
|
||||
|
||||
return 0;
|
||||
},
|
||||
nullptr);
|
||||
});
|
||||
client_thread->start();
|
||||
|
||||
event_loop.exec();
|
||||
MUST(client_thread->join());
|
||||
::unlink(socket_path.characters());
|
||||
}
|
||||
|
||||
|
|
@ -421,18 +423,20 @@ TEST_CASE(local_socket_write)
|
|||
};
|
||||
|
||||
// NOTE: Same reason as in the local_socket_read test.
|
||||
auto background_action = Threading::BackgroundAction<int>::construct(
|
||||
[&socket_path](auto&) {
|
||||
auto client_thread = Threading::Thread::construct(
|
||||
"LocalSocketWrite"sv,
|
||||
[socket_path] {
|
||||
auto client_socket = MUST(Core::LocalSocket::connect(socket_path));
|
||||
|
||||
MUST(client_socket->write_until_depleted({ sent_data.characters_without_null_termination(), sent_data.length() }));
|
||||
client_socket->close();
|
||||
|
||||
return 0;
|
||||
},
|
||||
nullptr);
|
||||
});
|
||||
client_thread->start();
|
||||
|
||||
event_loop.exec();
|
||||
MUST(client_thread->join());
|
||||
::unlink(socket_path.characters());
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,7 @@
|
|||
set(TEST_SOURCES
|
||||
TestBackgroundAction.cpp
|
||||
TestThread.cpp
|
||||
)
|
||||
|
||||
foreach(source IN LISTS TEST_SOURCES)
|
||||
ladybird_test("${source}" LibThreading LIBS LibThreading)
|
||||
endforeach()
|
||||
|
||||
target_link_libraries(TestBackgroundAction PRIVATE LibSync)
|
||||
|
|
|
|||
|
|
@ -1,119 +0,0 @@
|
|||
/*
|
||||
* Copyright (c) 2026, The Ladybird developers
|
||||
*
|
||||
* SPDX-License-Identifier: BSD-2-Clause
|
||||
*/
|
||||
|
||||
#include <AK/Atomic.h>
|
||||
#include <AK/Error.h>
|
||||
#include <AK/Function.h>
|
||||
#include <AK/Time.h>
|
||||
#include <AK/Types.h>
|
||||
#include <LibCore/EventLoop.h>
|
||||
#include <LibCore/System.h>
|
||||
#include <LibTest/TestCase.h>
|
||||
#include <LibThreading/BackgroundAction.h>
|
||||
#include <pthread.h>
|
||||
|
||||
using namespace AK::TimeLiterals;
|
||||
|
||||
static void spin_until(Core::EventLoop& loop, Function<bool()> condition, AK::Duration timeout = 2000_ms)
|
||||
{
|
||||
i64 const timeout_ms = timeout.to_milliseconds();
|
||||
for (i64 elapsed_ms = 0; elapsed_ms < timeout_ms; elapsed_ms += 5) {
|
||||
(void)loop.pump(Core::EventLoop::WaitMode::PollForEvents);
|
||||
if (condition())
|
||||
return;
|
||||
MUST(Core::System::sleep_ms(5));
|
||||
}
|
||||
|
||||
FAIL("Timed out waiting for condition");
|
||||
}
|
||||
|
||||
TEST_CASE(background_action_on_error_called_on_action_failure_and_on_origin_thread)
|
||||
{
|
||||
Core::EventLoop loop;
|
||||
|
||||
pthread_t const origin_thread_id = pthread_self();
|
||||
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> action_ran = false;
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> on_error_called = false;
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> on_complete_called = false;
|
||||
|
||||
Optional<pthread_t> action_thread_id;
|
||||
|
||||
auto background_action = Threading::BackgroundAction<int>::construct(
|
||||
[&](auto&) -> ErrorOr<int> {
|
||||
action_thread_id = pthread_self();
|
||||
action_ran.store(true, AK::MemoryOrder::memory_order_relaxed);
|
||||
return Error::from_string_literal("action failed");
|
||||
},
|
||||
[&](int) {
|
||||
on_complete_called.store(true, AK::MemoryOrder::memory_order_relaxed);
|
||||
loop.quit(1);
|
||||
},
|
||||
[&](Error error) {
|
||||
EXPECT(pthread_equal(origin_thread_id, pthread_self()));
|
||||
EXPECT_EQ(error.string_literal(), "action failed"sv);
|
||||
on_error_called.store(true, AK::MemoryOrder::memory_order_relaxed);
|
||||
loop.quit(0);
|
||||
});
|
||||
|
||||
loop.exec();
|
||||
|
||||
EXPECT(action_ran.load(AK::MemoryOrder::memory_order_relaxed));
|
||||
EXPECT(action_thread_id.has_value());
|
||||
EXPECT(!pthread_equal(action_thread_id.value(), origin_thread_id));
|
||||
EXPECT(on_error_called.load(AK::MemoryOrder::memory_order_relaxed));
|
||||
EXPECT(!on_complete_called.load(AK::MemoryOrder::memory_order_relaxed));
|
||||
|
||||
(void)background_action;
|
||||
}
|
||||
|
||||
TEST_CASE(background_action_cancel_suppresses_on_error_and_on_complete)
|
||||
{
|
||||
Core::EventLoop loop;
|
||||
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> started = false;
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> finished = false;
|
||||
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<int> on_error_count = 0;
|
||||
IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<int> on_complete_count = 0;
|
||||
|
||||
auto background_action = Threading::BackgroundAction<int>::construct(
|
||||
[&](auto& action) -> ErrorOr<int> {
|
||||
started.store(true, AK::MemoryOrder::memory_order_relaxed);
|
||||
|
||||
while (!action.is_canceled())
|
||||
MUST(Core::System::sleep_ms(1));
|
||||
|
||||
finished.store(true, AK::MemoryOrder::memory_order_relaxed);
|
||||
return Error::from_string_literal("error after cancel");
|
||||
},
|
||||
[&](int) {
|
||||
on_complete_count.fetch_add(1, AK::MemoryOrder::memory_order_relaxed);
|
||||
},
|
||||
[&](Error) {
|
||||
on_error_count.fetch_add(1, AK::MemoryOrder::memory_order_relaxed);
|
||||
});
|
||||
|
||||
spin_until(loop, [&] {
|
||||
return started.load(AK::MemoryOrder::memory_order_relaxed);
|
||||
});
|
||||
|
||||
background_action->cancel();
|
||||
|
||||
spin_until(loop, [&] {
|
||||
return finished.load(AK::MemoryOrder::memory_order_relaxed);
|
||||
});
|
||||
|
||||
// Run the loop a bit more to ensure any incorrectly-posted callbacks would execute.
|
||||
for (size_t i = 0; i < 50; ++i) {
|
||||
(void)loop.pump(Core::EventLoop::WaitMode::PollForEvents);
|
||||
MUST(Core::System::sleep_ms(1));
|
||||
}
|
||||
|
||||
EXPECT(background_action->is_canceled());
|
||||
EXPECT_EQ(on_complete_count.load(AK::MemoryOrder::memory_order_relaxed), 0);
|
||||
EXPECT_EQ(on_error_count.load(AK::MemoryOrder::memory_order_relaxed), 0);
|
||||
}
|
||||
Loading…
Reference in a new issue