/src/serenity/Userland/Libraries/LibThreading/WorkerThread.h

Source
/*
 * Copyright (c) 2022, Gregory Bertilson <zaggy1024@gmail.com>
 *
 * SPDX-License-Identifier: BSD-2-Clause
 */

#pragma once

#include <AK/Debug.h>
#include <AK/Variant.h>
#include <LibThreading/ConditionVariable.h>
#include <LibThreading/Mutex.h>
#include <LibThreading/Thread.h>

namespace Threading {

// Macro to allow single-line logging prints with fields that only exist in debug mode.
#if WORKER_THREAD_DEBUG
#    define WORKER_LOG(args...) ({ dbgln(args); })
#else
#    define WORKER_LOG(args...)
#endif

template<typename ErrorType>
class WorkerThread {
    enum class State {
        Idle,
        Working,
        Stopped,
    };
    using WorkerTask = Function<ErrorOr<void, ErrorType>()>;
    using WorkerState = Variant<State, WorkerTask, ErrorType>;

public:
    static ErrorOr<NonnullOwnPtr<WorkerThread>> create(StringView name)
    {
        auto worker_thread = TRY(adopt_nonnull_own_or_enomem(new (nothrow) WorkerThread()));
        worker_thread->m_thread = TRY(Threading::Thread::try_create([&self = *worker_thread]() {
            WORKER_LOG("Starting worker loop {}", self.m_id);

            while (true) {
                self.m_mutex.lock();
                if (self.m_stop) {
                    WORKER_LOG("Exiting {}", self.m_id);
                    self.m_state = State::Stopped;
                    self.m_condition.broadcast();
                    self.m_mutex.unlock();
                    return 0;
                }
                if (self.m_state.template has<WorkerTask>()) {
                    auto task = move(self.m_state.template get<WorkerTask>());
                    self.m_state = State::Working;
                    self.m_mutex.unlock();

                    WORKER_LOG("Starting task on {}", self.m_id);
                    auto result = task();
                    if (result.is_error()) {
                        WORKER_LOG("Task finished on {} with error", self.m_id);
                        self.m_mutex.lock();
                        self.m_state = result.release_error();
                        self.m_condition.broadcast();
                    } else {
                        WORKER_LOG("Task finished successfully on {}", self.m_id);
                        self.m_mutex.lock();
                        self.m_state = State::Idle;
                        self.m_condition.broadcast();
                    }
                }
                WORKER_LOG("Awaiting new task in {}...", self.m_id);
                self.m_condition.wait();
                WORKER_LOG("Worker thread awoken in {}", self.m_id);
                self.m_mutex.unlock();
            }

            return 0;
        },
            name));
        worker_thread->m_thread->start();
        return worker_thread;
    }

    ~WorkerThread()
    {
        m_mutex.lock();
        m_stop = true;
        m_condition.broadcast();
        while (!is_in_state(State::Stopped))
            m_condition.wait();
        m_mutex.unlock();
        (void)m_thread->join();
        WORKER_LOG("Worker thread {} joined successfully", m_id);
    }

    // Returns whether the task is starting.
    bool start_task(WorkerTask&& task)
    {
        m_mutex.lock();
        VERIFY(!is_in_state(State::Stopped));

        bool start_work = false;
        if (is_in_state(State::Idle)) {
            start_work = true;
        } else if (m_state.template has<ErrorType>()) {
            WORKER_LOG("Starting task and ignoring previous error: {}", m_state.template get<ErrorType>().string_literal());
            start_work = true;
        }
        if (start_work) {
            WORKER_LOG("Queuing task on {}", m_id);
            m_state = move(task);
            m_condition.broadcast();
        }

        m_mutex.unlock();
        return start_work;
    }

    ErrorOr<void, ErrorType> wait_until_task_is_finished()
    {
        WORKER_LOG("Waiting for task to finish on {}...", m_id);
        m_mutex.lock();
        while (true) {
            if (m_state.template has<WorkerTask>() || is_in_state(State::Working)) {
                m_condition.wait();
            } else if (m_state.template has<ErrorType>()) {
                auto error = move(m_state.template get<ErrorType>());
                m_state = State::Idle;
                m_mutex.unlock();
                WORKER_LOG("Finished waiting with error on {}: {}", m_id, error.string_literal());
                return error;
            } else {
                m_mutex.unlock();
                WORKER_LOG("Finished waiting on {}", m_id);
                return {};
            }
        }
        m_mutex.unlock();
    }

private:
#if WORKER_THREAD_DEBUG
    static inline size_t current_id = 0;
#endif

    WorkerThread()
        : m_condition(m_mutex)
#if WORKER_THREAD_DEBUG
        , m_id(current_id++)
#endif
    {
    }
    WorkerThread(WorkerThread const&) = delete;
    WorkerThread(WorkerThread&&) = delete;

    // Must be called with the mutex locked.
    bool is_in_state(State state)
    {
        return m_state.template has<State>() && m_state.template get<State>() == state;
    }

    RefPtr<Threading::Thread> m_thread;
    Threading::Mutex m_mutex;
    Threading::ConditionVariable m_condition;
    WorkerState m_state { State::Idle };
    bool m_stop { false };
#if WORKER_THREAD_DEBUG
    size_t m_id;
#endif
};

#undef WORKER_LOG

}

Line	Count	Source
1		/*
2		* Copyright (c) 2022, Gregory Bertilson <zaggy1024@gmail.com>
3		*
4		* SPDX-License-Identifier: BSD-2-Clause
5		*/
6
7		#pragma once
8
9		#include <AK/Debug.h>
10		#include <AK/Variant.h>
11		#include <LibThreading/ConditionVariable.h>
12		#include <LibThreading/Mutex.h>
13		#include <LibThreading/Thread.h>
14
15		namespace Threading {
16
17		// Macro to allow single-line logging prints with fields that only exist in debug mode.
18		#if WORKER_THREAD_DEBUG
19		# define WORKER_LOG(args...) ({ dbgln(args); })
20		#else
21		# define WORKER_LOG(args...)
22		#endif
23
24		template<typename ErrorType>
25		class WorkerThread {
26		enum class State {
27		Idle,
28		Working,
29		Stopped,
30		};
31		using WorkerTask = Function<ErrorOr<void, ErrorType>()>;
32		using WorkerState = Variant<State, WorkerTask, ErrorType>;
33
34		public:
35		static ErrorOr<NonnullOwnPtr<WorkerThread>> create(StringView name)
36	109	{
37	109	auto worker_thread = TRY(adopt_nonnull_own_or_enomem(new (nothrow) WorkerThread()));
38	109	worker_thread->m_thread = TRY(Threading::Thread::try_create([&self = *worker_thread]() {
39	109	WORKER_LOG("Starting worker loop {}", self.m_id);
40
41	109	while (true) {
42	109	self.m_mutex.lock();
43	109	if (self.m_stop) {
44	109	WORKER_LOG("Exiting {}", self.m_id);
45	109	self.m_state = State::Stopped;
46	109	self.m_condition.broadcast();
47	109	self.m_mutex.unlock();
48	109	return 0;
49	109	}
50	109	if (self.m_state.template has<WorkerTask>()) {
51	109	auto task = move(self.m_state.template get<WorkerTask>());
52	109	self.m_state = State::Working;
53	109	self.m_mutex.unlock();
54
55	109	WORKER_LOG("Starting task on {}", self.m_id);
56	109	auto result = task();
57	109	if (result.is_error()) {
58	109	WORKER_LOG("Task finished on {} with error", self.m_id);
59	109	self.m_mutex.lock();
60	109	self.m_state = result.release_error();
61	109	self.m_condition.broadcast();
62	109	} else {
63	109	WORKER_LOG("Task finished successfully on {}", self.m_id);
64	109	self.m_mutex.lock();
65	109	self.m_state = State::Idle;
66	109	self.m_condition.broadcast();
67	109	}
68	109	}
69	109	WORKER_LOG("Awaiting new task in {}...", self.m_id);
70	109	self.m_condition.wait();
71	109	WORKER_LOG("Worker thread awoken in {}", self.m_id);
72	109	self.m_mutex.unlock();
73	109	}
74
75	109	return 0;
76	109	},
77	109	name));
78	109	worker_thread->m_thread->start();
79	109	return worker_thread;
80	109	}
81
82		~WorkerThread()
83	109	{
84	109	m_mutex.lock();
85	109	m_stop = true;
86	109	m_condition.broadcast();
87	218	while (!is_in_state(State::Stopped))
88	109	m_condition.wait();
89	109	m_mutex.unlock();
90	109	(void)m_thread->join();
91	109	WORKER_LOG("Worker thread {} joined successfully", m_id);
92	109	}
93
94		// Returns whether the task is starting.
95		bool start_task(WorkerTask&& task)
96	109	{
97	109	m_mutex.lock();
98	109	VERIFY(!is_in_state(State::Stopped));
99
100	109	bool start_work = false;
101	109	if (is_in_state(State::Idle)) {
102	109	start_work = true;
103	109	} else if (m_state.template has<ErrorType>()) {
104	0	WORKER_LOG("Starting task and ignoring previous error: {}", m_state.template get<ErrorType>().string_literal());
105	0	start_work = true;
106	0	}
107	109	if (start_work) {
108	109	WORKER_LOG("Queuing task on {}", m_id);
109	109	m_state = move(task);
110	109	m_condition.broadcast();
111	109	}
112
113	109	m_mutex.unlock();
114	109	return start_work;
115	109	}
116
117		ErrorOr<void, ErrorType> wait_until_task_is_finished()
118	109	{
119	109	WORKER_LOG("Waiting for task to finish on {}...", m_id);
120	109	m_mutex.lock();
121	143	while (true) {
122	143	if (m_state.template has<WorkerTask>() \|\| is_in_state(State::Working)) {
123	34	m_condition.wait();
124	109	} else if (m_state.template has<ErrorType>()) {
125	109	auto error = move(m_state.template get<ErrorType>());
126	109	m_state = State::Idle;
127	109	m_mutex.unlock();
128	109	WORKER_LOG("Finished waiting with error on {}: {}", m_id, error.string_literal());
129	109	return error;
130	109	} else {
131	0	m_mutex.unlock();
132	0	WORKER_LOG("Finished waiting on {}", m_id);
133	0	return {};
134	0	}
135	143	}
136	0	m_mutex.unlock();
137	0	}
138
139		private:
140		#if WORKER_THREAD_DEBUG
141		static inline size_t current_id = 0;
142		#endif
143
144		WorkerThread()
145	109	: m_condition(m_mutex)
146		#if WORKER_THREAD_DEBUG
147		, m_id(current_id++)
148		#endif
149	109	{
150	109	}
151		WorkerThread(WorkerThread const&) = delete;
152		WorkerThread(WorkerThread&&) = delete;
153
154		// Must be called with the mutex locked.
155		bool is_in_state(State state)
156	579	{
157	579	return m_state.template has<State>() && m_state.template get<State>() == state;
158	579	}
159
160		RefPtr<Threading::Thread> m_thread;
161		Threading::Mutex m_mutex;
162		Threading::ConditionVariable m_condition;
163		WorkerState m_state { State::Idle };
164		bool m_stop { false };
165		#if WORKER_THREAD_DEBUG
166		size_t m_id;
167		#endif
168		};
169
170		#undef WORKER_LOG
171
172		}

Coverage Report

Created: 2026-09-26 06:49