1 /* radare - LGPL - Copyright 2014-2019 - pancake, thestr4ng3r */
2
3 #include <r_core.h>
4
r_core_task_scheduler_init(RCoreTaskScheduler * tasks,RCore * core)5 R_API void r_core_task_scheduler_init (RCoreTaskScheduler *tasks, RCore *core) {
6 tasks->task_id_next = 0;
7 tasks->tasks = r_list_newf ((RListFree)r_core_task_decref);
8 tasks->tasks_queue = r_list_new ();
9 tasks->oneshot_queue = r_list_newf (free);
10 tasks->oneshots_enqueued = 0;
11 tasks->lock = r_th_lock_new (true);
12 tasks->tasks_running = 0;
13 tasks->oneshot_running = false;
14 tasks->main_task = r_core_task_new (core, false, NULL, NULL, NULL);
15 r_list_append (tasks->tasks, tasks->main_task);
16 tasks->current_task = NULL;
17 }
18
r_core_task_scheduler_fini(RCoreTaskScheduler * tasks)19 R_API void r_core_task_scheduler_fini (RCoreTaskScheduler *tasks) {
20 r_list_free (tasks->tasks);
21 r_list_free (tasks->tasks_queue);
22 r_list_free (tasks->oneshot_queue);
23 r_th_lock_free (tasks->lock);
24 }
25
26 #if HAVE_PTHREAD
27 #define TASK_SIGSET_T sigset_t
tasks_lock_block_signals(sigset_t * old_sigset)28 static void tasks_lock_block_signals(sigset_t *old_sigset) {
29 sigset_t block_sigset;
30 sigemptyset (&block_sigset);
31 sigaddset (&block_sigset, SIGWINCH);
32 r_signal_sigmask (SIG_BLOCK, &block_sigset, old_sigset);
33 }
34
tasks_lock_block_signals_reset(sigset_t * old_sigset)35 static void tasks_lock_block_signals_reset(sigset_t *old_sigset) {
36 r_signal_sigmask (SIG_SETMASK, old_sigset, NULL);
37 }
38 #else
39 #define TASK_SIGSET_T void *
tasks_lock_block_signals(TASK_SIGSET_T * old_sigset)40 static void tasks_lock_block_signals(TASK_SIGSET_T *old_sigset) { (void)old_sigset; }
tasks_lock_block_signals_reset(TASK_SIGSET_T * old_sigset)41 static void tasks_lock_block_signals_reset(TASK_SIGSET_T *old_sigset) { (void)old_sigset; }
42 #endif
43
tasks_lock_enter(RCoreTaskScheduler * scheduler,TASK_SIGSET_T * old_sigset)44 static void tasks_lock_enter(RCoreTaskScheduler *scheduler, TASK_SIGSET_T *old_sigset) {
45 tasks_lock_block_signals (old_sigset);
46 r_th_lock_enter (scheduler->lock);
47 }
48
tasks_lock_leave(RCoreTaskScheduler * scheduler,TASK_SIGSET_T * old_sigset)49 static void tasks_lock_leave(RCoreTaskScheduler *scheduler, TASK_SIGSET_T *old_sigset) {
50 r_th_lock_leave (scheduler->lock);
51 tasks_lock_block_signals_reset (old_sigset);
52 }
53
54 typedef struct oneshot_t {
55 RCoreTaskOneShot func;
56 void *user;
57 } OneShot;
58
r_core_task_print(RCore * core,RCoreTask * task,PJ * pj,int mode)59 R_API void r_core_task_print (RCore *core, RCoreTask *task, PJ *pj, int mode) {
60 switch (mode) {
61 case 'j': {
62 pj_o (pj);
63 pj_ki (pj, "id", task->id);
64 pj_k (pj, "state");
65 switch (task->state) {
66 case R_CORE_TASK_STATE_BEFORE_START:
67 pj_s (pj, "before_start");
68 break;
69 case R_CORE_TASK_STATE_RUNNING:
70 pj_s (pj, "running");
71 break;
72 case R_CORE_TASK_STATE_SLEEPING:
73 pj_s (pj, "sleeping");
74 break;
75 case R_CORE_TASK_STATE_DONE:
76 pj_s (pj, "done");
77 break;
78 }
79 pj_kb (pj, "transient", task->transient);
80 pj_ks (pj, "cmd", r_str_get_fail (task->cmd, "null"));
81 pj_end (pj);
82 break;
83 }
84 default: {
85 const char *info = task->cmd;
86 if (task == core->tasks.main_task) {
87 info = "-- MAIN TASK --";
88 }
89 r_cons_printf ("%3d %3s %12s %s\n",
90 task->id,
91 task->transient ? "(t)" : "",
92 r_core_task_status (task),
93 r_str_get (info));
94 }
95 break;
96 }
97 }
98
r_core_task_list(RCore * core,int mode)99 R_API void r_core_task_list(RCore *core, int mode) {
100 RListIter *iter;
101 RCoreTask *task;
102 PJ *pj = NULL;
103 if (mode == 'j') {
104 pj = r_core_pj_new (core);
105 if (!pj) {
106 return;
107 }
108 pj_a (pj);
109 }
110 TASK_SIGSET_T old_sigset;
111 tasks_lock_enter (&core->tasks, &old_sigset);
112 r_list_foreach (core->tasks.tasks, iter, task) {
113 r_core_task_print (core, task, pj, mode);
114 }
115 if (mode == 'j') {
116 pj_end (pj);
117 r_cons_println (pj_string (pj));
118 pj_free (pj);
119 } else {
120 r_cons_printf ("--\ntotal running: %d\n", core->tasks.tasks_running);
121 }
122 tasks_lock_leave (&core->tasks, &old_sigset);
123 }
124
r_core_task_running_tasks_count(RCoreTaskScheduler * scheduler)125 R_API int r_core_task_running_tasks_count(RCoreTaskScheduler *scheduler) {
126 RListIter *iter;
127 RCoreTask *task;
128 int count = 0;
129 TASK_SIGSET_T old_sigset;
130 tasks_lock_enter (scheduler, &old_sigset);
131 r_list_foreach (scheduler->tasks, iter, task) {
132 if (task != scheduler->main_task && task->state != R_CORE_TASK_STATE_DONE) {
133 count++;
134 }
135 }
136 tasks_lock_leave (scheduler, &old_sigset);
137 return count;
138 }
139
task_join(RCoreTask * task)140 static void task_join(RCoreTask *task) {
141 RThreadSemaphore *sem = task->running_sem;
142 if (!sem) {
143 return;
144 }
145
146 r_th_sem_wait (sem);
147 r_th_sem_post (sem);
148 }
149
r_core_task_join(RCoreTaskScheduler * scheduler,RCoreTask * current,int id)150 R_API void r_core_task_join(RCoreTaskScheduler *scheduler, RCoreTask *current, int id) {
151 if (current && id == current->id) {
152 return;
153 }
154 if (id >= 0) {
155 RCoreTask *task = r_core_task_get_incref (scheduler, id);
156 if (!task) {
157 return;
158 }
159 if (current) {
160 r_core_task_sleep_begin (current);
161 }
162 task_join (task);
163 if (current) {
164 r_core_task_sleep_end (current);
165 }
166 r_core_task_decref (task);
167 } else {
168 TASK_SIGSET_T old_sigset;
169 tasks_lock_enter (scheduler, &old_sigset);
170 RList *tasks = r_list_clone (scheduler->tasks);
171 RListIter *iter;
172 RCoreTask *task;
173 r_list_foreach (tasks, iter, task) {
174 if (current == task) {
175 continue;
176 }
177 r_core_task_incref (task);
178 }
179 tasks_lock_leave (scheduler, &old_sigset);
180
181 r_list_foreach (tasks, iter, task) {
182 if (current == task) {
183 continue;
184 }
185 if (current) {
186 r_core_task_sleep_begin (current);
187 }
188 task_join (task);
189 if (current) {
190 r_core_task_sleep_end (current);
191 }
192 r_core_task_decref (task);
193 }
194 r_list_free (tasks);
195 }
196 }
197
task_free(RCoreTask * task)198 static void task_free (RCoreTask *task) {
199 if (!task) {
200 return;
201 }
202 free (task->cmd);
203 free (task->res);
204 r_th_free (task->thread);
205 r_th_sem_free (task->running_sem);
206 r_th_cond_free (task->dispatch_cond);
207 r_th_lock_free (task->dispatch_lock);
208 r_cons_context_free (task->cons_context);
209 free (task);
210 }
211
r_core_task_new(RCore * core,bool create_cons,const char * cmd,RCoreTaskCallback cb,void * user)212 R_API RCoreTask *r_core_task_new(RCore *core, bool create_cons, const char *cmd, RCoreTaskCallback cb, void *user) {
213 RCoreTask *task = R_NEW0 (RCoreTask);
214 if (!task) {
215 goto hell;
216 }
217
218 task->thread = NULL;
219 task->cmd = cmd ? strdup (cmd) : NULL;
220 task->cmd_log = false;
221 task->res = NULL;
222 task->running_sem = NULL;
223 task->dispatched = false;
224 task->dispatch_cond = r_th_cond_new ();
225 task->dispatch_lock = r_th_lock_new (false);
226 if (!task->dispatch_cond || !task->dispatch_lock) {
227 goto hell;
228 }
229
230 if (create_cons) {
231 task->cons_context = r_cons_context_new (r_cons_singleton ()->context);
232 if (!task->cons_context) {
233 goto hell;
234 }
235 task->cons_context->cmd_depth = core->max_cmd_depth;
236 }
237
238 task->id = core->tasks.task_id_next++;
239 task->state = R_CORE_TASK_STATE_BEFORE_START;
240 task->refcount = 1;
241 task->transient = false;
242 task->core = core;
243 task->user = user;
244 task->cb = cb;
245
246 return task;
247
248 hell:
249 task_free (task);
250 return NULL;
251 }
252
r_core_task_incref(RCoreTask * task)253 R_API void r_core_task_incref (RCoreTask *task) {
254 if (!task) {
255 return;
256 }
257 TASK_SIGSET_T old_sigset;
258 tasks_lock_enter (&task->core->tasks, &old_sigset);
259 task->refcount++;
260 tasks_lock_leave (&task->core->tasks, &old_sigset);
261 }
262
r_core_task_decref(RCoreTask * task)263 R_API void r_core_task_decref (RCoreTask *task) {
264 if (!task) {
265 return;
266 }
267 TASK_SIGSET_T old_sigset;
268 RCoreTaskScheduler *scheduler = &task->core->tasks;
269 tasks_lock_enter (scheduler, &old_sigset);
270 task->refcount--;
271 if (task->refcount <= 0) {
272 task_free (task);
273 }
274 tasks_lock_leave (scheduler, &old_sigset);
275 }
276
r_core_task_schedule(RCoreTask * current,RTaskState next_state)277 R_API void r_core_task_schedule(RCoreTask *current, RTaskState next_state) {
278 RCore *core = current->core;
279 RCoreTaskScheduler *scheduler = &core->tasks;
280 bool stop = next_state != R_CORE_TASK_STATE_RUNNING;
281
282 if (scheduler->oneshot_running || (!stop && scheduler->tasks_running == 1 && scheduler->oneshots_enqueued == 0)) {
283 return;
284 }
285
286 scheduler->current_task = NULL;
287
288 TASK_SIGSET_T old_sigset;
289 tasks_lock_enter (scheduler, &old_sigset);
290
291 current->state = next_state;
292
293 if (stop) {
294 scheduler->tasks_running--;
295 }
296
297 // oneshots always have priority.
298 // if there are any queued, run them immediately.
299 OneShot *oneshot;
300 while ((oneshot = r_list_pop_head (scheduler->oneshot_queue))) {
301 scheduler->oneshots_enqueued--;
302 scheduler->oneshot_running = true;
303 oneshot->func (oneshot->user);
304 scheduler->oneshot_running = false;
305 free (oneshot);
306 }
307
308 RCoreTask *next = r_list_pop_head (scheduler->tasks_queue);
309
310 if (next && !stop) {
311 r_list_append (scheduler->tasks_queue, current);
312 r_th_lock_enter (current->dispatch_lock);
313 }
314
315 tasks_lock_leave (scheduler, &old_sigset);
316
317 if (next) {
318 r_cons_context_reset ();
319 r_th_lock_enter (next->dispatch_lock);
320 next->dispatched = true;
321 r_th_lock_leave (next->dispatch_lock);
322 r_th_cond_signal (next->dispatch_cond);
323 if (!stop) {
324 while (!current->dispatched) {
325 r_th_cond_wait (current->dispatch_cond, current->dispatch_lock);
326 }
327 current->dispatched = false;
328 r_th_lock_leave (current->dispatch_lock);
329 }
330 }
331
332 if (!stop) {
333 scheduler->current_task = current;
334 if (current->cons_context) {
335 r_cons_context_load (current->cons_context);
336 } else {
337 r_cons_context_reset ();
338 }
339 }
340 }
341
task_wakeup(RCoreTask * current)342 static void task_wakeup(RCoreTask *current) {
343 RCore *core = current->core;
344 RCoreTaskScheduler *scheduler = &core->tasks;
345
346 TASK_SIGSET_T old_sigset;
347 tasks_lock_enter (scheduler, &old_sigset);
348
349 scheduler->tasks_running++;
350 current->state = R_CORE_TASK_STATE_RUNNING;
351
352 // check if there are other tasks running
353 bool single = scheduler->tasks_running == 1 || scheduler->tasks_running == 0;
354
355 r_th_lock_enter (current->dispatch_lock);
356
357 // if we are not the only task, we must wait until another task signals us.
358
359 if (!single) {
360 r_list_append (scheduler->tasks_queue, current);
361 }
362
363 tasks_lock_leave (scheduler, &old_sigset);
364
365 if (!single) {
366 while (!current->dispatched) {
367 r_th_cond_wait (current->dispatch_cond, current->dispatch_lock);
368 }
369 current->dispatched = false;
370 }
371
372 r_th_lock_leave (current->dispatch_lock);
373
374 scheduler->current_task = current;
375
376 if (current->cons_context) {
377 r_cons_context_load (current->cons_context);
378 } else {
379 r_cons_context_reset ();
380 }
381 }
382
r_core_task_yield(RCoreTaskScheduler * scheduler)383 R_API void r_core_task_yield(RCoreTaskScheduler *scheduler) {
384 RCoreTask *task = r_core_task_self (scheduler);
385 if (!task) {
386 return;
387 }
388 r_core_task_schedule (task, R_CORE_TASK_STATE_RUNNING);
389 }
390
task_end(RCoreTask * t)391 static void task_end(RCoreTask *t) {
392 r_core_task_schedule (t, R_CORE_TASK_STATE_DONE);
393 }
394
task_run(RCoreTask * task)395 static RThreadFunctionRet task_run(RCoreTask *task) {
396 RCore *core = task->core;
397 RCoreTaskScheduler *scheduler = &task->core->tasks;
398
399 task_wakeup (task);
400
401 if (task->cons_context && task->cons_context->breaked) {
402 // breaked in R_CORE_TASK_STATE_BEFORE_START
403 goto stillbirth;
404 }
405
406 char *res_str;
407 if (task == scheduler->main_task) {
408 r_core_cmd (core, task->cmd, task->cmd_log);
409 res_str = NULL;
410 } else {
411 res_str = r_core_cmd_str (core, task->cmd);
412 }
413
414 free (task->res);
415 task->res = res_str;
416
417 if (task != scheduler->main_task && r_cons_default_context_is_interactive ()) {
418 eprintf ("\nTask %d finished\n", task->id);
419 }
420
421 TASK_SIGSET_T old_sigset;
422 stillbirth:
423 tasks_lock_enter (scheduler, &old_sigset);
424
425 task_end (task);
426
427 if (task->cb) {
428 task->cb (task->user, task->res);
429 }
430
431 if (task->running_sem) {
432 r_th_sem_post (task->running_sem);
433 }
434
435 if (task->cons_context && task->cons_context->break_stack) {
436 r_cons_context_break_pop (task->cons_context, false);
437 }
438
439 int ret = R_TH_STOP;
440 if (task->transient) {
441 RCoreTask *ltask;
442 RListIter *iter;
443 r_list_foreach (scheduler->tasks, iter, ltask) {
444 if (ltask == task) {
445 r_list_delete (scheduler->tasks, iter);
446 ret = R_TH_FREED;
447 break;
448 }
449 }
450 }
451
452 tasks_lock_leave (scheduler, &old_sigset);
453 return ret;
454 }
455
task_run_thread(RThread * th)456 static RThreadFunctionRet task_run_thread(RThread *th) {
457 RCoreTask *task = (RCoreTask *)th->user;
458 return task_run (task);
459 }
460
r_core_task_enqueue(RCoreTaskScheduler * scheduler,RCoreTask * task)461 R_API void r_core_task_enqueue(RCoreTaskScheduler *scheduler, RCoreTask *task) {
462 if (!scheduler || !task) {
463 return;
464 }
465 TASK_SIGSET_T old_sigset;
466 tasks_lock_enter (scheduler, &old_sigset);
467 if (!task->running_sem) {
468 task->running_sem = r_th_sem_new (1);
469 }
470 if (task->running_sem) {
471 r_th_sem_wait (task->running_sem);
472 }
473 if (task->cons_context) {
474 r_cons_context_break_push (task->cons_context, NULL, NULL, false);
475 }
476 r_list_append (scheduler->tasks, task);
477 task->thread = r_th_new (task_run_thread, task, 0);
478 tasks_lock_leave (scheduler, &old_sigset);
479 }
480
r_core_task_enqueue_oneshot(RCoreTaskScheduler * scheduler,RCoreTaskOneShot func,void * user)481 R_API void r_core_task_enqueue_oneshot(RCoreTaskScheduler *scheduler, RCoreTaskOneShot func, void *user) {
482 if (!scheduler || !func) {
483 return;
484 }
485 TASK_SIGSET_T old_sigset;
486 tasks_lock_enter (scheduler, &old_sigset);
487 if (scheduler->tasks_running == 0) {
488 // nothing is running right now and no other task can be scheduled
489 // while core->tasks_lock is locked => just run it
490 scheduler->oneshot_running = true;
491 func (user);
492 scheduler->oneshot_running = false;
493 } else {
494 OneShot *oneshot = R_NEW (OneShot);
495 if (oneshot) {
496 oneshot->func = func;
497 oneshot->user = user;
498 r_list_append (scheduler->oneshot_queue, oneshot);
499 scheduler->oneshots_enqueued++;
500 }
501 }
502 tasks_lock_leave (scheduler, &old_sigset);
503 }
504
r_core_task_run_sync(RCoreTaskScheduler * scheduler,RCoreTask * task)505 R_API int r_core_task_run_sync(RCoreTaskScheduler *scheduler, RCoreTask *task) {
506 task->thread = NULL;
507 return task_run (task);
508 }
509
510 /* begin running stuff synchronously on the main task */
r_core_task_sync_begin(RCoreTaskScheduler * scheduler)511 R_API void r_core_task_sync_begin(RCoreTaskScheduler *scheduler) {
512 RCoreTask *task = scheduler->main_task;
513 TASK_SIGSET_T old_sigset;
514 tasks_lock_enter (scheduler, &old_sigset);
515 task->thread = NULL;
516 task->cmd = NULL;
517 task->cmd_log = false;
518 task->state = R_CORE_TASK_STATE_BEFORE_START;
519 tasks_lock_leave (scheduler, &old_sigset);
520 task_wakeup (task);
521 }
522
523 /* end running stuff synchronously, initially started with r_core_task_sync_begin() */
r_core_task_sync_end(RCoreTaskScheduler * scheduler)524 R_API void r_core_task_sync_end(RCoreTaskScheduler *scheduler) {
525 task_end (scheduler->main_task);
526 }
527
528 /* To be called from within a task.
529 * Begin sleeping and schedule other tasks until r_core_task_sleep_end() is called. */
r_core_task_sleep_begin(RCoreTask * task)530 R_API void r_core_task_sleep_begin(RCoreTask *task) {
531 r_core_task_schedule (task, R_CORE_TASK_STATE_SLEEPING);
532 }
533
r_core_task_sleep_end(RCoreTask * task)534 R_API void r_core_task_sleep_end(RCoreTask *task) {
535 task_wakeup (task);
536 }
537
r_core_task_status(RCoreTask * task)538 R_API const char *r_core_task_status (RCoreTask *task) {
539 switch (task->state) {
540 case R_CORE_TASK_STATE_RUNNING:
541 return "running";
542 case R_CORE_TASK_STATE_SLEEPING:
543 return "sleeping";
544 case R_CORE_TASK_STATE_DONE:
545 return "done";
546 case R_CORE_TASK_STATE_BEFORE_START:
547 return "before start";
548 default:
549 return "unknown";
550 }
551 }
552
r_core_task_self(RCoreTaskScheduler * scheduler)553 R_API RCoreTask *r_core_task_self (RCoreTaskScheduler *scheduler) {
554 return scheduler->current_task ? scheduler->current_task : scheduler->main_task;
555 }
556
task_get(RCoreTaskScheduler * scheduler,int id)557 static RCoreTask *task_get (RCoreTaskScheduler *scheduler, int id) {
558 RCoreTask *task;
559 RListIter *iter;
560 r_list_foreach (scheduler->tasks, iter, task) {
561 if (task->id == id) {
562 return task;
563 }
564 }
565 return NULL;
566 }
567
r_core_task_get_incref(RCoreTaskScheduler * scheduler,int id)568 R_API RCoreTask *r_core_task_get_incref(RCoreTaskScheduler *scheduler, int id) {
569 TASK_SIGSET_T old_sigset;
570 tasks_lock_enter (scheduler, &old_sigset);
571 RCoreTask *task = task_get (scheduler, id);
572 if (task) {
573 r_core_task_incref (task);
574 }
575 tasks_lock_leave (scheduler, &old_sigset);
576 return task;
577 }
578
r_core_task_break(RCoreTaskScheduler * scheduler,int id)579 R_API void r_core_task_break(RCoreTaskScheduler *scheduler, int id) {
580 TASK_SIGSET_T old_sigset;
581 tasks_lock_enter (scheduler, &old_sigset);
582 RCoreTask *task = task_get (scheduler, id);
583 if (!task || task->state == R_CORE_TASK_STATE_DONE) {
584 tasks_lock_leave (scheduler, &old_sigset);
585 return;
586 }
587 if (task->cons_context) {
588 r_cons_context_break (task->cons_context);
589 }
590 tasks_lock_leave (scheduler, &old_sigset);
591 }
592
r_core_task_break_all(RCoreTaskScheduler * scheduler)593 R_API void r_core_task_break_all(RCoreTaskScheduler *scheduler) {
594 TASK_SIGSET_T old_sigset;
595 tasks_lock_enter (scheduler, &old_sigset);
596 RCoreTask *task;
597 RListIter *iter;
598 r_list_foreach (scheduler->tasks, iter, task) {
599 if (task->state != R_CORE_TASK_STATE_DONE) {
600 r_cons_context_break (task->cons_context);
601 }
602 }
603 tasks_lock_leave (scheduler, &old_sigset);
604 }
605
r_core_task_del(RCoreTaskScheduler * scheduler,int id)606 R_API int r_core_task_del (RCoreTaskScheduler *scheduler, int id) {
607 RCoreTask *task;
608 RListIter *iter;
609 bool ret = false;
610 TASK_SIGSET_T old_sigset;
611 tasks_lock_enter (scheduler, &old_sigset);
612 r_list_foreach (scheduler->tasks, iter, task) {
613 if (task->id == id) {
614 if (task == scheduler->main_task) {
615 break;
616 }
617 if (task->state == R_CORE_TASK_STATE_DONE) {
618 r_list_delete (scheduler->tasks, iter);
619 } else {
620 task->transient = true;
621 }
622 ret = true;
623 break;
624 }
625 }
626 tasks_lock_leave (scheduler, &old_sigset);
627 return ret;
628 }
629
r_core_task_del_all_done(RCoreTaskScheduler * scheduler)630 R_API void r_core_task_del_all_done (RCoreTaskScheduler *scheduler) {
631 RCoreTask *task;
632 RListIter *iter, *iter2;
633 r_list_foreach_safe (scheduler->tasks, iter, iter2, task) {
634 if (task != scheduler->main_task && task->state == R_CORE_TASK_STATE_DONE) {
635 r_list_delete (scheduler->tasks, iter);
636 }
637 }
638 }
639