INCLUDE_DIRECTORIES(${JSONC_INCLUDE_DIRS})
ENDIF()
-SET(SOURCES avl.c avl-cmp.c blob.c blobmsg.c uloop.c usock.c ustream.c ustream-fd.c vlist.c utils.c safe_list.c)
+SET(SOURCES avl.c avl-cmp.c blob.c blobmsg.c uloop.c usock.c ustream.c ustream-fd.c vlist.c utils.c safe_list.c runqueue.c)
ADD_LIBRARY(ubox SHARED ${SOURCES})
ADD_EXECUTABLE(ustream-example ustream-example.c)
TARGET_LINK_LIBRARIES(ustream-example ubox)
+ADD_EXECUTABLE(runqueue-example runqueue-example.c)
+TARGET_LINK_LIBRARIES(runqueue-example ubox)
+
--- /dev/null
+/*
+ * runqueue-example.c
+ *
+ * Copyright (C) 2013 Felix Fietkau <nbd@openwrt.org>
+ *
+ * Permission to use, copy, modify, and/or distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include <libubox/uloop.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <unistd.h>
+
+#include "runqueue.h"
+
+static struct runqueue q;
+
+struct sleeper {
+ struct runqueue_process proc;
+ int val;
+};
+
+static void q_empty(struct runqueue *q)
+{
+ fprintf(stderr, "All done!\n");
+ uloop_end();
+}
+
+static void q_sleep_run(struct runqueue *q, struct runqueue_task *t)
+{
+ struct sleeper *s = container_of(t, struct sleeper, proc.task);
+ char str[32];
+ pid_t pid;
+
+ fprintf(stderr, "[%d/%d] start 'sleep %d'\n", q->running_tasks, q->max_running_tasks, s->val);
+
+ pid = fork();
+ if (pid < 0)
+ return;
+
+ if (pid) {
+ runqueue_process_add(q, &s->proc, pid);
+ return;
+ }
+
+ sprintf(str, "%d", s->val);
+ execlp("sleep", "sleep", str, NULL);
+ exit(1);
+}
+
+static void q_sleep_cancel(struct runqueue *q, struct runqueue_task *t, int type)
+{
+ struct sleeper *s = container_of(t, struct sleeper, proc.task);
+
+ fprintf(stderr, "[%d/%d] cancel 'sleep %d'\n", q->running_tasks, q->max_running_tasks, s->val);
+ runqueue_process_cancel_cb(q, t, type);
+}
+
+static void q_sleep_complete(struct runqueue *q, struct runqueue_process *p, int ret)
+{
+ struct sleeper *s = container_of(p, struct sleeper, proc);
+
+ fprintf(stderr, "[%d/%d] finish 'sleep %d'\n", q->running_tasks, q->max_running_tasks, s->val);
+ free(s);
+}
+
+static void add_sleeper(int val)
+{
+ static const struct runqueue_task_type sleeper_type = {
+ .run = q_sleep_run,
+ .cancel = q_sleep_cancel,
+ .kill = runqueue_process_kill_cb,
+ };
+ struct sleeper *s;
+
+ s = calloc(1, sizeof(*s));
+ s->proc.task.type = &sleeper_type;
+ s->proc.task.run_timeout = 500;
+ s->proc.complete = q_sleep_complete;
+ s->val = val;
+ runqueue_task_add(&q, &s->proc.task, false);
+}
+
+int main(int argc, char **argv)
+{
+ uloop_init();
+
+ runqueue_init(&q);
+ q.empty_cb = q_empty;
+ q.max_running_tasks = 1;
+
+ if (argc > 1)
+ q.max_running_tasks = atoi(argv[1]);
+
+ add_sleeper(1);
+ add_sleeper(1);
+ add_sleeper(1);
+ uloop_run();
+ uloop_done();
+
+ return 0;
+}
--- /dev/null
+/*
+ * runqueue.c - a simple task queueing/completion tracking helper
+ *
+ * Copyright (C) 2013 Felix Fietkau <nbd@openwrt.org>
+ *
+ * Permission to use, copy, modify, and/or distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#include <string.h>
+#include <stdio.h>
+#include "runqueue.h"
+
+static void
+__runqueue_empty_cb(struct uloop_timeout *timeout)
+{
+ struct runqueue *q = container_of(timeout, struct runqueue, timeout);
+
+ q->empty_cb(q);
+}
+
+void runqueue_init(struct runqueue *q)
+{
+ INIT_SAFE_LIST(&q->tasks_active);
+ INIT_SAFE_LIST(&q->tasks_inactive);
+}
+
+static void __runqueue_start_next(struct uloop_timeout *timeout)
+{
+ struct runqueue *q = container_of(timeout, struct runqueue, timeout);
+ struct runqueue_task *t;
+
+ do {
+ if (q->stopped)
+ break;
+
+ if (list_empty(&q->tasks_inactive.list))
+ break;
+
+ if (q->max_running_tasks && q->running_tasks >= q->max_running_tasks)
+ break;
+
+ t = list_first_entry(&q->tasks_inactive.list, struct runqueue_task, list.list);
+ safe_list_del(&t->list);
+ safe_list_add(&t->list, &q->tasks_active);
+ t->running = true;
+ q->running_tasks++;
+ if (t->run_timeout)
+ uloop_timeout_set(&t->timeout, t->run_timeout);
+ t->type->run(q, t);
+ } while (1);
+
+ if (!q->empty &&
+ list_empty(&q->tasks_active.list) &&
+ list_empty(&q->tasks_inactive.list)) {
+ q->empty = true;
+ if (q->empty_cb) {
+ q->timeout.cb = __runqueue_empty_cb;
+ uloop_timeout_set(&q->timeout, 1);
+ }
+ }
+}
+
+static void runqueue_start_next(struct runqueue *q)
+{
+ if (q->empty)
+ return;
+
+ q->timeout.cb = __runqueue_start_next;
+ uloop_timeout_set(&q->timeout, 1);
+}
+
+static int __runqueue_cancel(void *ctx, struct safe_list *list)
+{
+ struct runqueue_task *t;
+
+ t = container_of(list, struct runqueue_task, list);
+ runqueue_task_cancel(t, 0);
+
+ return 0;
+}
+
+void runqueue_cancel_active(struct runqueue *q)
+{
+ safe_list_for_each(&q->tasks_active, __runqueue_cancel, NULL);
+}
+
+void runqueue_cancel_pending(struct runqueue *q)
+{
+ safe_list_for_each(&q->tasks_inactive, __runqueue_cancel, NULL);
+}
+
+void runqueue_cancel(struct runqueue *q)
+{
+ runqueue_cancel_pending(q);
+ runqueue_cancel_active(q);
+}
+
+void runqueue_kill(struct runqueue *q)
+{
+ struct runqueue_task *t;
+
+ while (!list_empty(&q->tasks_active.list)) {
+ t = list_first_entry(&q->tasks_active.list, struct runqueue_task, list.list);
+ runqueue_task_kill(t);
+ }
+ runqueue_cancel_pending(q);
+ uloop_timeout_cancel(&q->timeout);
+}
+
+void runqueue_task_cancel(struct runqueue_task *t, int type)
+{
+ if (!t->queued)
+ return;
+
+ if (!t->running) {
+ runqueue_task_complete(t);
+ return;
+ }
+
+ t->cancelled = true;
+ if (t->cancel_timeout)
+ uloop_timeout_set(&t->timeout, t->cancel_timeout);
+ if (t->type->cancel)
+ t->type->cancel(t->q, t, type);
+}
+
+static void
+__runqueue_task_timeout(struct uloop_timeout *timeout)
+{
+ struct runqueue_task *t = container_of(timeout, struct runqueue_task, timeout);
+
+ if (t->cancelled)
+ runqueue_task_kill(t);
+ else
+ runqueue_task_cancel(t, t->cancel_type);
+}
+
+void runqueue_task_add(struct runqueue *q, struct runqueue_task *t, bool running)
+{
+ struct safe_list *head;
+
+ if (t->queued)
+ return;
+
+ if (!t->type->run && !running) {
+ fprintf(stderr, "BUG: inactive task added without run() callback\n");
+ return;
+ }
+
+ if (running) {
+ q->running_tasks++;
+ head = &q->tasks_active;
+ } else {
+ head = &q->tasks_inactive;
+ }
+
+ t->timeout.cb = __runqueue_task_timeout;
+ t->q = q;
+ safe_list_add(&t->list, head);
+ t->cancelled = false;
+ t->queued = true;
+ t->running = running;
+ q->empty = false;
+
+ runqueue_start_next(q);
+}
+
+void runqueue_task_kill(struct runqueue_task *t)
+{
+ struct runqueue *q = t->q;
+ bool running = t->running;
+
+ if (!t->queued)
+ return;
+
+ runqueue_task_complete(t);
+ if (running && t->type->kill)
+ t->type->kill(q, t);
+
+ runqueue_start_next(q);
+}
+
+void runqueue_stop(struct runqueue *q)
+{
+ q->stopped = true;
+}
+
+void runqueue_resume(struct runqueue *q)
+{
+ q->stopped = false;
+ runqueue_start_next(q);
+}
+
+void runqueue_task_complete(struct runqueue_task *t)
+{
+ if (!t->queued)
+ return;
+
+ if (t->running)
+ t->q->running_tasks--;
+
+ safe_list_del(&t->list);
+ t->queued = false;
+ t->running = false;
+ t->cancelled = false;
+ runqueue_start_next(t->q);
+}
+
+static void
+__runqueue_proc_cb(struct uloop_process *p, int ret)
+{
+ struct runqueue_process *t = container_of(p, struct runqueue_process, proc);
+ struct runqueue *q = t->task.q;
+
+ runqueue_task_complete(&t->task);
+ if (t->complete)
+ t->complete(q, t, ret);
+}
+
+void runqueue_process_cancel_cb(struct runqueue *q, struct runqueue_task *t, int type)
+{
+ struct runqueue_process *p = container_of(t, struct runqueue_process, task);
+
+ if (!type)
+ type = SIGTERM;
+
+ kill(p->proc.pid, type);
+}
+
+void runqueue_process_kill_cb(struct runqueue *q, struct runqueue_task *t)
+{
+ struct runqueue_process *p = container_of(t, struct runqueue_process, task);
+
+ uloop_process_delete(&p->proc);
+ kill(p->proc.pid, SIGKILL);
+ __runqueue_proc_cb(&p->proc, -1);
+}
+
+static const struct runqueue_task_type runqueue_proc_type = {
+ .name = "process",
+ .cancel = runqueue_process_cancel_cb,
+ .kill = runqueue_process_kill_cb,
+};
+
+void runqueue_process_add(struct runqueue *q, struct runqueue_process *p, pid_t pid)
+{
+ if (p->proc.pending)
+ return;
+
+ p->proc.pid = pid;
+ p->proc.cb = __runqueue_proc_cb;
+ if (!p->task.type)
+ p->task.type = &runqueue_proc_type;
+ uloop_process_add(&p->proc);
+ if (!p->task.running)
+ runqueue_task_add(q, &p->task, true);
+}
--- /dev/null
+/*
+ * runqueue.c - a simple task queueing/completion tracking helper
+ *
+ * Copyright (C) 2013 Felix Fietkau <nbd@openwrt.org>
+ *
+ * Permission to use, copy, modify, and/or distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+#ifndef __LIBUBOX_RUNQUEUE_H
+#define __LIBUBOX_RUNQUEUE_H
+
+#include <libubox/list.h>
+#include <libubox/safe_list.h>
+#include <libubox/uloop.h>
+
+struct runqueue;
+struct runqueue_task;
+struct runqueue_task_type;
+
+struct runqueue {
+ struct safe_list tasks_active;
+ struct safe_list tasks_inactive;
+ struct uloop_timeout timeout;
+
+ int running_tasks;
+ int max_running_tasks;
+ bool stopped;
+ bool empty;
+
+ /* called when the runqueue is emptied */
+ void (*empty_cb)(struct runqueue *q);
+};
+
+struct runqueue_task_type {
+ const char *name;
+
+ /*
+ * called when a task is requested to run
+ *
+ * The task is removed from the list before this callback is run. It
+ * can re-arm itself using runqueue_task_add.
+ */
+ void (*run)(struct runqueue *q, struct runqueue_task *t);
+
+ /*
+ * called to request cancelling a task
+ *
+ * int type is used as an optional hint for the method to be used when
+ * cancelling the task, e.g. a signal number for processes. Calls
+ * runqueue_task_complete when done.
+ */
+ void (*cancel)(struct runqueue *q, struct runqueue_task *t, int type);
+
+ /*
+ * called to kill a task. must not make any calls to runqueue_task_complete,
+ * it has already been removed from the list.
+ */
+ void (*kill)(struct runqueue *q, struct runqueue_task *t);
+};
+
+struct runqueue_task {
+ struct safe_list list;
+ const struct runqueue_task_type *type;
+ struct runqueue *q;
+
+ struct uloop_timeout timeout;
+ int run_timeout;
+ int cancel_timeout;
+ int cancel_type;
+
+ bool queued;
+ bool running;
+ bool cancelled;
+};
+
+struct runqueue_process {
+ struct runqueue_task task;
+ struct uloop_process proc;
+ void (*complete)(struct runqueue *q, struct runqueue_process *p, int ret);
+};
+
+void runqueue_init(struct runqueue *q);
+void runqueue_cancel(struct runqueue *q);
+void runqueue_cancel_active(struct runqueue *q);
+void runqueue_cancel_pending(struct runqueue *q);
+void runqueue_kill(struct runqueue *q);
+
+void runqueue_stop(struct runqueue *q);
+void runqueue_resume(struct runqueue *q);
+
+void runqueue_task_add(struct runqueue *q, struct runqueue_task *t, bool running);
+void runqueue_task_complete(struct runqueue_task *t);
+
+void runqueue_task_cancel(struct runqueue_task *t, int type);
+void runqueue_task_kill(struct runqueue_task *t);
+
+void runqueue_process_add(struct runqueue *q, struct runqueue_process *p, pid_t pid);
+
+/* to be used only from runqueue_process callbacks */
+void runqueue_process_cancel_cb(struct runqueue *q, struct runqueue_task *t, int type);
+void runqueue_process_kill_cb(struct runqueue *q, struct runqueue_task *t);
+
+#endif
projects / oweals / libubox.git / commitdiff