--- /dev/null
+#include <pthread.h>
+
+#include <linux/kthread.h>
+#include <linux/slab.h>
+#include <linux/workqueue.h>
+
+static pthread_mutex_t wq_lock = PTHREAD_MUTEX_INITIALIZER;
+static pthread_cond_t work_finished = PTHREAD_COND_INITIALIZER;
+static LIST_HEAD(wq_list);
+
+struct workqueue_struct {
+ struct list_head list;
+
+ struct work_struct *current_work;
+ struct list_head pending_work;
+
+ struct task_struct *worker;
+ char name[24];
+};
+
+enum {
+ WORK_PENDING_BIT,
+};
+
+static bool work_pending(struct work_struct *work)
+{
+ return test_bit(WORK_PENDING_BIT, work_data_bits(work));
+}
+
+static void clear_work_pending(struct work_struct *work)
+{
+ clear_bit(WORK_PENDING_BIT, work_data_bits(work));
+}
+
+static bool set_work_pending(struct work_struct *work)
+{
+ return !test_and_set_bit(WORK_PENDING_BIT, work_data_bits(work));
+}
+
+static void __queue_work(struct workqueue_struct *wq,
+ struct work_struct *work)
+{
+ BUG_ON(!work_pending(work));
+ BUG_ON(!list_empty(&work->entry));
+
+ list_add_tail(&work->entry, &wq->pending_work);
+ wake_up_process(wq->worker);
+}
+
+bool queue_work(struct workqueue_struct *wq, struct work_struct *work)
+{
+ bool ret;
+
+ pthread_mutex_lock(&wq_lock);
+ if ((ret = set_work_pending(work)))
+ __queue_work(wq, work);
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+void delayed_work_timer_fn(struct timer_list *timer)
+{
+ struct delayed_work *dwork =
+ container_of(timer, struct delayed_work, timer);
+
+ pthread_mutex_lock(&wq_lock);
+ __queue_work(dwork->wq, &dwork->work);
+ pthread_mutex_unlock(&wq_lock);
+}
+
+static void __queue_delayed_work(struct workqueue_struct *wq,
+ struct delayed_work *dwork,
+ unsigned long delay)
+{
+ struct timer_list *timer = &dwork->timer;
+ struct work_struct *work = &dwork->work;
+
+ BUG_ON(timer->function != delayed_work_timer_fn);
+ BUG_ON(timer_pending(timer));
+ BUG_ON(!list_empty(&work->entry));
+
+ if (!delay) {
+ __queue_work(wq, &dwork->work);
+ } else {
+ dwork->wq = wq;
+ timer->expires = jiffies + delay;
+ add_timer(timer);
+ }
+}
+
+bool queue_delayed_work(struct workqueue_struct *wq,
+ struct delayed_work *dwork,
+ unsigned long delay)
+{
+ struct work_struct *work = &dwork->work;
+ bool ret;
+
+ pthread_mutex_lock(&wq_lock);
+ if ((ret = set_work_pending(work)))
+ __queue_delayed_work(wq, dwork, delay);
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+static bool grab_pending(struct work_struct *work, bool is_dwork)
+{
+retry:
+ if (set_work_pending(work)) {
+ BUG_ON(!list_empty(&work->entry));
+ return false;
+ }
+
+ if (is_dwork) {
+ struct delayed_work *dwork = to_delayed_work(work);
+
+ if (likely(del_timer(&dwork->timer))) {
+ BUG_ON(!list_empty(&work->entry));
+ return true;
+ }
+ }
+
+ if (!list_empty(&work->entry)) {
+ list_del_init(&work->entry);
+ return true;
+ }
+
+ BUG_ON(!is_dwork);
+
+ pthread_mutex_unlock(&wq_lock);
+ flush_timers();
+ pthread_mutex_lock(&wq_lock);
+ goto retry;
+}
+
+static bool work_running(struct work_struct *work)
+{
+ struct workqueue_struct *wq;
+
+ list_for_each_entry(wq, &wq_list, list)
+ if (wq->current_work == work)
+ return true;
+
+ return false;
+}
+
+bool flush_work(struct work_struct *work)
+{
+ bool ret = false;
+
+ pthread_mutex_lock(&wq_lock);
+ while (work_pending(work) || work_running(work)) {
+ pthread_cond_wait(&work_finished, &wq_lock);
+ ret = true;
+ }
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+static bool __flush_work(struct work_struct *work)
+{
+ bool ret = false;
+
+ while (work_running(work)) {
+ pthread_cond_wait(&work_finished, &wq_lock);
+ ret = true;
+ }
+
+ return ret;
+}
+
+bool cancel_work_sync(struct work_struct *work)
+{
+ bool ret;
+
+ pthread_mutex_lock(&wq_lock);
+ ret = grab_pending(work, false);
+
+ __flush_work(work);
+ clear_work_pending(work);
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+bool mod_delayed_work(struct workqueue_struct *wq,
+ struct delayed_work *dwork,
+ unsigned long delay)
+{
+ struct work_struct *work = &dwork->work;
+ bool ret;
+
+ pthread_mutex_lock(&wq_lock);
+ ret = grab_pending(work, true);
+
+ __queue_delayed_work(wq, dwork, delay);
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+bool cancel_delayed_work(struct delayed_work *dwork)
+{
+ struct work_struct *work = &dwork->work;
+ bool ret;
+
+ pthread_mutex_lock(&wq_lock);
+ ret = grab_pending(work, true);
+
+ clear_work_pending(&dwork->work);
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+bool cancel_delayed_work_sync(struct delayed_work *dwork)
+{
+ struct work_struct *work = &dwork->work;
+ bool ret;
+
+ pthread_mutex_lock(&wq_lock);
+ ret = grab_pending(work, true);
+
+ __flush_work(work);
+ clear_work_pending(work);
+ pthread_mutex_unlock(&wq_lock);
+
+ return ret;
+}
+
+static int worker_thread(void *arg)
+{
+ struct workqueue_struct *wq = arg;
+ struct work_struct *work;
+
+ pthread_mutex_lock(&wq_lock);
+ while (1) {
+ __set_current_state(TASK_INTERRUPTIBLE);
+ work = list_first_entry_or_null(&wq->pending_work,
+ struct work_struct, entry);
+ wq->current_work = work;
+
+ if (kthread_should_stop()) {
+ BUG_ON(wq->current_work);
+ break;
+ }
+
+ if (!work) {
+ pthread_mutex_unlock(&wq_lock);
+ schedule();
+ pthread_mutex_lock(&wq_lock);
+ continue;
+ }
+
+ BUG_ON(!work_pending(work));
+ list_del_init(&work->entry);
+ clear_work_pending(work);
+
+ pthread_mutex_unlock(&wq_lock);
+ work->func(work);
+ pthread_mutex_lock(&wq_lock);
+
+ pthread_cond_broadcast(&work_finished);
+ }
+ pthread_mutex_unlock(&wq_lock);
+
+ return 0;
+}
+
+void destroy_workqueue(struct workqueue_struct *wq)
+{
+ kthread_stop(wq->worker);
+
+ pthread_mutex_lock(&wq_lock);
+ list_del(&wq->list);
+ pthread_mutex_unlock(&wq_lock);
+
+ kfree(wq);
+}
+
+struct workqueue_struct *alloc_workqueue(const char *fmt,
+ unsigned flags,
+ int max_active,
+ ...)
+{
+ va_list args;
+ struct workqueue_struct *wq;
+
+ wq = kzalloc(sizeof(*wq), GFP_KERNEL);
+ if (!wq)
+ return NULL;
+
+ INIT_LIST_HEAD(&wq->list);
+ INIT_LIST_HEAD(&wq->pending_work);
+
+ va_start(args, max_active);
+ vsnprintf(wq->name, sizeof(wq->name), fmt, args);
+ va_end(args);
+
+ wq->worker = kthread_run(worker_thread, wq, "%s", wq->name);
+ if (IS_ERR(wq->worker)) {
+ kfree(wq);
+ return NULL;
+ }
+
+ pthread_mutex_lock(&wq_lock);
+ list_add(&wq->list, &wq_list);
+ pthread_mutex_unlock(&wq_lock);
+
+ return wq;
+}
+
+struct workqueue_struct *system_wq;
+struct workqueue_struct *system_highpri_wq;
+struct workqueue_struct *system_long_wq;
+struct workqueue_struct *system_unbound_wq;
+struct workqueue_struct *system_freezable_wq;
+
+__attribute__((constructor(102)))
+static void wq_init(void)
+{
+ system_wq = alloc_workqueue("events", 0, 0);
+ system_highpri_wq = alloc_workqueue("events_highpri", WQ_HIGHPRI, 0);
+ system_long_wq = alloc_workqueue("events_long", 0, 0);
+ system_unbound_wq = alloc_workqueue("events_unbound", WQ_UNBOUND,
+ WQ_UNBOUND_MAX_ACTIVE);
+ system_freezable_wq = alloc_workqueue("events_freezable",
+ WQ_FREEZABLE, 0);
+ BUG_ON(!system_wq || !system_highpri_wq || !system_long_wq ||
+ !system_unbound_wq || !system_freezable_wq);
+}
+
+__attribute__((destructor(102)))
+static void wq_cleanup(void)
+{
+ destroy_workqueue(system_freezable_wq);
+ destroy_workqueue(system_unbound_wq);
+ destroy_workqueue(system_long_wq);
+ destroy_workqueue(system_highpri_wq);
+ destroy_workqueue(system_wq);
+
+ system_wq = system_highpri_wq = system_long_wq = system_unbound_wq =
+ system_freezable_wq = NULL;
+}
projects / bcachefs-tools-debian / blobdiff