• source navigation  • diff markup  • identifier search  • freetext search  • 

Sources/libubox/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 static void _runqueue_task_add(struct runqueue *q, struct runqueue_task *t, bool running, bool first)
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         if (first)
170                 safe_list_add_first(&t->list, head);
171         else
172                 safe_list_add(&t->list, head);
173         t->cancelled = false;
174         t->queued = true;
175         t->running = running;
176         q->empty = false;
177 
178         runqueue_start_next(q);
179 }
180 
181 void runqueue_task_add(struct runqueue *q, struct runqueue_task *t, bool running)
182 {
183         _runqueue_task_add(q, t, running, 0);
184 }
185 
186 void runqueue_task_add_first(struct runqueue *q, struct runqueue_task *t, bool running)
187 {
188         _runqueue_task_add(q, t, running, 1);
189 }
190 
191 void runqueue_task_kill(struct runqueue_task *t)
192 {
193         struct runqueue *q = t->q;
194         bool running = t->running;
195 
196         if (!t->queued)
197                 return;
198 
199         runqueue_task_complete(t);
200         if (running && t->type->kill)
201                 t->type->kill(q, t);
202 
203         runqueue_start_next(q);
204 }
205 
206 void runqueue_stop(struct runqueue *q)
207 {
208         q->stopped = true;
209 }
210 
211 void runqueue_resume(struct runqueue *q)
212 {
213         q->stopped = false;
214         runqueue_start_next(q);
215 }
216 
217 void runqueue_task_complete(struct runqueue_task *t)
218 {
219         struct runqueue *q = t->q;
220 
221         if (!t->queued)
222                 return;
223 
224         if (t->running)
225                 t->q->running_tasks--;
226 
227         uloop_timeout_cancel(&t->timeout);
228 
229         safe_list_del(&t->list);
230         t->queued = false;
231         t->running = false;
232         t->cancelled = false;
233         if (t->complete)
234                 t->complete(q, t);
235         runqueue_start_next(q);
236 }
237 
238 static void
239 __runqueue_proc_cb(struct uloop_process *p, int ret)
240 {
241         struct runqueue_process *t = container_of(p, struct runqueue_process, proc);
242 
243         runqueue_task_complete(&t->task);
244 }
245 
246 void runqueue_process_cancel_cb(struct runqueue *q, struct runqueue_task *t, int type)
247 {
248         struct runqueue_process *p = container_of(t, struct runqueue_process, task);
249 
250         if (!type)
251                 type = SIGTERM;
252 
253         kill(p->proc.pid, type);
254 }
255 
256 void runqueue_process_kill_cb(struct runqueue *q, struct runqueue_task *t)
257 {
258         struct runqueue_process *p = container_of(t, struct runqueue_process, task);
259 
260         uloop_process_delete(&p->proc);
261         kill(p->proc.pid, SIGKILL);
262 }
263 
264 static const struct runqueue_task_type runqueue_proc_type = {
265         .name = "process",
266         .cancel = runqueue_process_cancel_cb,
267         .kill = runqueue_process_kill_cb,
268 };
269 
270 void runqueue_process_add(struct runqueue *q, struct runqueue_process *p, pid_t pid)
271 {
272         if (p->proc.pending)
273                 return;
274 
275         p->proc.pid = pid;
276         p->proc.cb = __runqueue_proc_cb;
277         if (!p->task.type)
278                 p->task.type = &runqueue_proc_type;
279         uloop_process_add(&p->proc);
280         if (!p->task.running)
281                 runqueue_task_add(q, &p->task, true);
282 }
283 

This page was automatically generated by LXR 0.3.1.  •  OpenWrt