1bdb48e157e88b611846f91819e3426a69198f76
[project/libubox.git] / runqueue.c
1 /*
2  * runqueue.c - a simple task queueing/completion tracking helper
3  *
4  * Copyright (C) 2013 Felix Fietkau <nbd@openwrt.org>
5  *
6  * Permission to use, copy, modify, and/or distribute this software for any
7  * purpose with or without fee is hereby granted, provided that the above
8  * copyright notice and this permission notice appear in all copies.
9  *
10  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
11  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
12  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
13  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
14  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
15  * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
16  * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
17  */
18
19 #include <string.h>
20 #include <stdio.h>
21 #include "runqueue.h"
22
23 static void
24 __runqueue_empty_cb(struct uloop_timeout *timeout)
25 {
26         struct runqueue *q = container_of(timeout, struct runqueue, timeout);
27
28         q->empty_cb(q);
29 }
30
31 void runqueue_init(struct runqueue *q)
32 {
33         INIT_SAFE_LIST(&q->tasks_active);
34         INIT_SAFE_LIST(&q->tasks_inactive);
35 }
36
37 static void __runqueue_start_next(struct uloop_timeout *timeout)
38 {
39         struct runqueue *q = container_of(timeout, struct runqueue, timeout);
40         struct runqueue_task *t;
41
42         do {
43                 if (q->stopped)
44                         break;
45
46                 if (list_empty(&q->tasks_inactive.list))
47                         break;
48
49                 if (q->max_running_tasks && q->running_tasks >= q->max_running_tasks)
50                         break;
51
52                 t = list_first_entry(&q->tasks_inactive.list, struct runqueue_task, list.list);
53                 safe_list_del(&t->list);
54                 safe_list_add(&t->list, &q->tasks_active);
55                 t->running = true;
56                 q->running_tasks++;
57                 if (t->run_timeout)
58                         uloop_timeout_set(&t->timeout, t->run_timeout);
59                 t->type->run(q, t);
60         } while (1);
61
62         if (!q->empty &&
63             list_empty(&q->tasks_active.list) &&
64             list_empty(&q->tasks_inactive.list)) {
65                 q->empty = true;
66                 if (q->empty_cb) {
67                         q->timeout.cb = __runqueue_empty_cb;
68                         uloop_timeout_set(&q->timeout, 1);
69                 }
70         }
71 }
72
73 static void runqueue_start_next(struct runqueue *q)
74 {
75         if (q->empty)
76                 return;
77
78         q->timeout.cb = __runqueue_start_next;
79         uloop_timeout_set(&q->timeout, 1);
80 }
81
82 static int __runqueue_cancel(void *ctx, struct safe_list *list)
83 {
84         struct runqueue_task *t;
85
86         t = container_of(list, struct runqueue_task, list);
87         runqueue_task_cancel(t, 0);
88
89         return 0;
90 }
91
92 void runqueue_cancel_active(struct runqueue *q)
93 {
94         safe_list_for_each(&q->tasks_active, __runqueue_cancel, NULL);
95 }
96
97 void runqueue_cancel_pending(struct runqueue *q)
98 {
99         safe_list_for_each(&q->tasks_inactive, __runqueue_cancel, NULL);
100 }
101
102 void runqueue_cancel(struct runqueue *q)
103 {
104         runqueue_cancel_pending(q);
105         runqueue_cancel_active(q);
106 }
107
108 void runqueue_kill(struct runqueue *q)
109 {
110         struct runqueue_task *t;
111
112         while (!list_empty(&q->tasks_active.list)) {
113                 t = list_first_entry(&q->tasks_active.list, struct runqueue_task, list.list);
114                 runqueue_task_kill(t);
115         }
116         runqueue_cancel_pending(q);
117         uloop_timeout_cancel(&q->timeout);
118 }
119
120 void runqueue_task_cancel(struct runqueue_task *t, int type)
121 {
122         if (!t->queued)
123                 return;
124
125         if (!t->running) {
126                 runqueue_task_complete(t);
127                 return;
128         }
129
130         t->cancelled = true;
131         if (t->cancel_timeout)
132                 uloop_timeout_set(&t->timeout, t->cancel_timeout);
133         if (t->type->cancel)
134                 t->type->cancel(t->q, t, type);
135 }
136
137 static void
138 __runqueue_task_timeout(struct uloop_timeout *timeout)
139 {
140         struct runqueue_task *t = container_of(timeout, struct runqueue_task, timeout);
141
142         if (t->cancelled)
143                 runqueue_task_kill(t);
144         else
145                 runqueue_task_cancel(t, t->cancel_type);
146 }
147
148 void runqueue_task_add(struct runqueue *q, struct runqueue_task *t, bool running)
149 {
150         struct safe_list *head;
151
152         if (t->queued)
153                 return;
154
155         if (!t->type->run && !running) {
156                 fprintf(stderr, "BUG: inactive task added without run() callback\n");
157                 return;
158         }
159
160         if (running) {
161                 q->running_tasks++;
162                 head = &q->tasks_active;
163         } else {
164                 head = &q->tasks_inactive;
165         }
166
167         t->timeout.cb = __runqueue_task_timeout;
168         t->q = q;
169         safe_list_add(&t->list, head);
170         t->cancelled = false;
171         t->queued = true;
172         t->running = running;
173         q->empty = false;
174
175         runqueue_start_next(q);
176 }
177
178 void runqueue_task_kill(struct runqueue_task *t)
179 {
180         struct runqueue *q = t->q;
181         bool running = t->running;
182
183         if (!t->queued)
184                 return;
185
186         runqueue_task_complete(t);
187         if (running && t->type->kill)
188                 t->type->kill(q, t);
189         if (t->complete)
190                 t->complete(q, t);
191
192         runqueue_start_next(q);
193 }
194
195 void runqueue_stop(struct runqueue *q)
196 {
197         q->stopped = true;
198 }
199
200 void runqueue_resume(struct runqueue *q)
201 {
202         q->stopped = false;
203         runqueue_start_next(q);
204 }
205
206 void runqueue_task_complete(struct runqueue_task *t)
207 {
208         if (!t->queued)
209                 return;
210
211         if (t->running)
212                 t->q->running_tasks--;
213
214         safe_list_del(&t->list);
215         t->queued = false;
216         t->running = false;
217         t->cancelled = false;
218         runqueue_start_next(t->q);
219 }
220
221 static void
222 __runqueue_proc_cb(struct uloop_process *p, int ret)
223 {
224         struct runqueue_process *t = container_of(p, struct runqueue_process, proc);
225
226         runqueue_task_complete(&t->task);
227 }
228
229 void runqueue_process_cancel_cb(struct runqueue *q, struct runqueue_task *t, int type)
230 {
231         struct runqueue_process *p = container_of(t, struct runqueue_process, task);
232
233         if (!type)
234                 type = SIGTERM;
235
236         kill(p->proc.pid, type);
237 }
238
239 void runqueue_process_kill_cb(struct runqueue *q, struct runqueue_task *t)
240 {
241         struct runqueue_process *p = container_of(t, struct runqueue_process, task);
242
243         uloop_process_delete(&p->proc);
244         kill(p->proc.pid, SIGKILL);
245 }
246
247 static const struct runqueue_task_type runqueue_proc_type = {
248         .name = "process",
249         .cancel = runqueue_process_cancel_cb,
250         .kill = runqueue_process_kill_cb,
251 };
252
253 void runqueue_process_add(struct runqueue *q, struct runqueue_process *p, pid_t pid)
254 {
255         if (p->proc.pending)
256                 return;
257
258         p->proc.pid = pid;
259         p->proc.cb = __runqueue_proc_cb;
260         if (!p->task.type)
261                 p->task.type = &runqueue_proc_type;
262         uloop_process_add(&p->proc);
263         if (!p->task.running)
264                 runqueue_task_add(q, &p->task, true);
265 }