From: Andre Noll Date: Wed, 23 Jul 2014 15:29:29 +0000 (+0200) Subject: Merge branch 't/sched_improvements' X-Git-Tag: v0.5.3~8 X-Git-Url: http://git.tuebingen.mpg.de/?p=paraslash.git;a=commitdiff_plain;h=8aa9d0cddfefb882cfd8e7cadbeea5cce31b731f;hp=015183e5aa7858d2988ff0c0fb9c997aadeeca0c Merge branch 't/sched_improvements' Cooking for two months. This merge required to also patch gui.c due to semantic conflics against the changes introduced by the gui_sched branch which was merged to master in commit d15d8509 two weeks ago. Also a small fix for server.c is needed to squelch a compiler warning since the global "now" variable has become a const pointer. * t/sched_improvements: (36 commits) audiod: Fix use after free on exit. sched: Mark global now pointer as const. sched: Directly pass context pointer to pre/post_select(). sched: kill task->dead. sched: Do not shadow task_info in struct task. sched: Dont use fixed-size buffer for task names. sched: Rename task->error to tast->status. sched: Rename task->status to task->name. sched: Make struct task private to sched.c. sched: Introduce task_status(). sched: Remove ->owned_by_sched. sched: Remove register_task(). task_register() conversion: grab client task task_register() conversion: audiod status task task_register() conversion: audiod command task task_register() conversion: client task task_register() conversion: client supervisor task task_register() conversion: client exec task task_register() conversion: afs command task task_register() conversion: vss task ... --- diff --git a/NEWS b/NEWS index c613d672..311aca2b 100644 --- a/NEWS +++ b/NEWS @@ -5,8 +5,11 @@ NEWS 0.5.3 (to be released) "symbolic synchronization" ------------------------------------------------- +Not many new features, but lots of fixes and usability improvements. + - para_gui has been converted to use the paraslash scheduler. - Various alsa-related fixes, mostly for the raspberry pi. + - Many scheduler improvements and cleanups. - The test suite has been extended to include sanity checks for the generated man pages. - ao_writer fixes. This writer was in a quite bad shape. Many diff --git a/aacdec_filter.c b/aacdec_filter.c index d63236da..7988723b 100644 --- a/aacdec_filter.c +++ b/aacdec_filter.c @@ -80,9 +80,9 @@ static void aacdec_close(struct filter_node *fn) fn->private_data = NULL; } -static int aacdec_post_select(__a_unused struct sched *s, struct task *t) +static int aacdec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct btr_node *btrn = fn->btrn; struct private_aacdec_data *padd = fn->private_data; int i, ret; diff --git a/afh_recv.c b/afh_recv.c index 0c7b3946..657a2057 100644 --- a/afh_recv.c +++ b/afh_recv.c @@ -60,7 +60,6 @@ static int afh_execute(struct btr_node *btrn, const char *cmd, char **result) if (x >= pard->afhi.chunks_total) return -ERRNO_TO_PARA_ERROR(EINVAL); pard->first_chunk = pard->current_chunk = x; - rn->task.error = 0; return 1; } return -E_BTR_NAVAIL; @@ -151,14 +150,14 @@ static void afh_recv_close(struct receiver_node *rn) freep(&rn->private_data); } -static void afh_recv_pre_select(struct sched *s, struct task *t) +static void afh_recv_pre_select(struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; struct private_afh_recv_data *pard = rn->private_data; struct afh_info *afhi = &pard->afhi; struct afh_recv_args_info *conf = rn->conf; struct timeval chunk_time; - int state = generic_recv_pre_select(s, t); + int state = generic_recv_pre_select(s, rn); if (state <= 0) return; @@ -171,9 +170,9 @@ static void afh_recv_pre_select(struct sched *s, struct task *t) sched_request_barrier_or_min_delay(&chunk_time, s); } -static int afh_recv_post_select(__a_unused struct sched *s, struct task *t) +static int afh_recv_post_select(__a_unused struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; struct afh_recv_args_info *conf = rn->conf; struct private_afh_recv_data *pard = rn->private_data; struct btr_node *btrn = rn->btrn; diff --git a/afs.c b/afs.c index fa4f4326..a17da5b4 100644 --- a/afs.c +++ b/afs.c @@ -81,7 +81,7 @@ struct command_task { */ uint32_t cookie; /** The associated task structure. */ - struct task task; + struct task *task; }; extern int mmd_mutex; @@ -719,13 +719,13 @@ static int open_afs_tables(void) return ret; } -static void signal_pre_select(struct sched *s, struct task *t) +static void signal_pre_select(struct sched *s, void *context) { - struct signal_task *st = container_of(t, struct signal_task, task); + struct signal_task *st = context; para_fd_set(st->fd, &s->rfds, &s->max_fileno); } -static int afs_signal_post_select(struct sched *s, __a_unused struct task *t) +static int afs_signal_post_select(struct sched *s, __a_unused void *context) { int signum, ret; @@ -762,10 +762,13 @@ static void register_signal_task(struct sched *s) para_install_sighandler(SIGTERM); para_install_sighandler(SIGHUP); - st->task.pre_select = signal_pre_select; - st->task.post_select = afs_signal_post_select; - sprintf(st->task.status, "signal task"); - register_task(s, &st->task); + st->task = task_register(&(struct task_info) { + .name = "signal", + .pre_select = signal_pre_select, + .post_select = afs_signal_post_select, + .context = st, + + }, s); } static struct list_head afs_client_list; @@ -780,9 +783,9 @@ struct afs_client { struct timeval connect_time; }; -static void command_pre_select(struct sched *s, struct task *t) +static void command_pre_select(struct sched *s, void *context) { - struct command_task *ct = container_of(t, struct command_task, task); + struct command_task *ct = context; struct afs_client *client; para_fd_set(server_socket, &s->rfds, &s->max_fileno); @@ -917,14 +920,14 @@ err: /** Shutdown connection if query has not arrived until this many seconds. */ #define AFS_CLIENT_TIMEOUT 3 -static int command_post_select(struct sched *s, struct task *t) +static int command_post_select(struct sched *s, void *context) { - struct command_task *ct = container_of(t, struct command_task, task); + struct command_task *ct = context; struct sockaddr_un unix_addr; struct afs_client *client, *tmp; int fd, ret; - ret = task_get_notification(t); + ret = task_get_notification(ct->task); if (ret < 0) return ret; ret = execute_server_command(&s->rfds); @@ -972,10 +975,12 @@ static void register_command_task(uint32_t cookie, struct sched *s) ct->fd = setup_command_socket_or_die(); ct->cookie = cookie; - ct->task.pre_select = command_pre_select; - ct->task.post_select = command_post_select; - sprintf(ct->task.status, "afs command task"); - register_task(s, &ct->task); + ct->task = task_register(&(struct task_info) { + .name = "afs command", + .pre_select = command_pre_select, + .post_select = command_post_select, + .context = ct, + }, s); } /** @@ -1007,6 +1012,7 @@ __noreturn void afs_init(uint32_t cookie, int socket_fd) s.default_timeout.tv_sec = 0; s.default_timeout.tv_usec = 999 * 1000; ret = schedule(&s); + sched_shutdown(&s); out_close: close_afs_tables(); out: diff --git a/alsa_write.c b/alsa_write.c index 3759306e..32d8a50e 100644 --- a/alsa_write.c +++ b/alsa_write.c @@ -199,10 +199,10 @@ fail: return -E_ALSA; } -static void alsa_write_pre_select(struct sched *s, struct task *t) +static void alsa_write_pre_select(struct sched *s, void *context) { struct pollfd pfd; - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_alsa_write_data *pad = wn->private_data; int ret = btr_node_status(wn->btrn, wn->min_iqs, BTR_NT_LEAF); @@ -248,10 +248,9 @@ static void alsa_close(struct writer_node *wn) free(pad); } -static int alsa_write_post_select(__a_unused struct sched *s, - struct task *t) +static int alsa_write_post_select(__a_unused struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_alsa_write_data *pad = wn->private_data; struct btr_node *btrn = wn->btrn; char *data; @@ -259,7 +258,7 @@ static int alsa_write_post_select(__a_unused struct sched *s, snd_pcm_sframes_t frames; int ret; - ret = task_get_notification(t); + ret = task_get_notification(wn->task); if (ret < 0) goto err; again: diff --git a/amp_filter.c b/amp_filter.c index c3e55cf0..8b8db201 100644 --- a/amp_filter.c +++ b/amp_filter.c @@ -62,9 +62,9 @@ static void amp_open(struct filter_node *fn) pad->amp, pad->amp / 64.0 + 1.0); } -static int amp_post_select(__a_unused struct sched *s, struct task *t) +static int amp_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_amp_data *pad = fn->private_data; struct btr_node *btrn = fn->btrn; int ret, factor = 64 + pad->amp; diff --git a/ao_write.c b/ao_write.c index 63d18afa..4cecdfd8 100644 --- a/ao_write.c +++ b/ao_write.c @@ -46,9 +46,9 @@ static void aow_close(struct writer_node *wn) wn->private_data = NULL; } -static void aow_pre_select(struct sched *s, struct task *t) +static void aow_pre_select(struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_aow_data *pawd = wn->private_data; int ret; @@ -308,10 +308,9 @@ fail: return -E_AO_PTHREAD; } -static int aow_post_select(__a_unused struct sched *s, - struct task *t) +static int aow_post_select(__a_unused struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_aow_data *pawd = wn->private_data; int ret; diff --git a/audioc.c b/audioc.c index fe7165ed..83425a2e 100644 --- a/audioc.c +++ b/audioc.c @@ -79,7 +79,7 @@ static struct sched sched; struct audioc_task { int fd; struct btr_node *btrn; - struct task task; + struct task *task; }; static struct i9e_completer audiod_completers[]; @@ -130,9 +130,9 @@ static struct i9e_completer audiod_completers[] = { {.name = NULL} }; -static void audioc_pre_select(struct sched *s, struct task *t) +static void audioc_pre_select(struct sched *s, void *context) { - struct audioc_task *at = container_of(t, struct audioc_task, task); + struct audioc_task *at = context; int ret = btr_node_status(at->btrn, 0, BTR_NT_ROOT); if (ret < 0) @@ -140,10 +140,10 @@ static void audioc_pre_select(struct sched *s, struct task *t) para_fd_set(at->fd, &s->rfds, &s->max_fileno); } -static int audioc_post_select(struct sched *s, struct task *t) +static int audioc_post_select(struct sched *s, void *context) { char *buf = NULL; - struct audioc_task *at = container_of(t, struct audioc_task, task); + struct audioc_task *at = context; int ret = btr_node_status(at->btrn, 0, BTR_NT_ROOT); if (ret < 0) @@ -168,13 +168,7 @@ out: return ret; } -static struct audioc_task audioc_task = { - .task = { - .pre_select = audioc_pre_select, - .post_select = audioc_post_select, - .status = "audioc task" - }, -}, *at = &audioc_task; +static struct audioc_task audioc_task, *at = &audioc_task; static int audioc_i9e_line_handler(char *line) { @@ -202,8 +196,12 @@ static int audioc_i9e_line_handler(char *line) args = NULL; at->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "audioc line handler")); - at->task.error = 0; - register_task(&sched, &at->task); + at->task = task_register(&(struct task_info) { + .name = "audioc", + .pre_select = audioc_pre_select, + .post_select = audioc_post_select, + .context = at, + }, &sched); i9e_attach_to_stdout(at->btrn); return 1; close: @@ -248,6 +246,7 @@ __noreturn static void interactive_session(void) goto out; para_log = i9e_log; ret = schedule(&sched); + sched_shutdown(&sched); i9e_close(); para_log = stderr_log; out: diff --git a/audiod.c b/audiod.c index a8f40183..6fd101a0 100644 --- a/audiod.c +++ b/audiod.c @@ -93,14 +93,10 @@ enum vss_status_flags { */ struct sched sched = {.max_fileno = 0}; -/** - * The task for obtaining para_server's status (para_client stat). - * - * \sa struct task, struct sched. - */ +/* The task for obtaining para_server's status (para_client stat). */ struct status_task { /** The associated task structure of audiod. */ - struct task task; + struct task *task; /** Client data associated with the stat task. */ struct client_task *ct; /** Do not restart client command until this time. */ @@ -168,7 +164,7 @@ struct command_task { /** the local listening socket */ int fd; /** the associated task structure */ - struct task task; + struct task *task; }; /** iterate over all supported audio formats */ @@ -391,6 +387,7 @@ static void close_receiver(int slot_num) audio_formats[s->format], slot_num); a->receiver->close(s->receiver_node); btr_remove_node(&s->receiver_node->btrn); + task_reap(&s->receiver_node->task); free(s->receiver_node); s->receiver_node = NULL; tv_add(now, &(struct timeval)EMBRACE(0, 200 * 1000), @@ -407,6 +404,7 @@ static void writer_cleanup(struct writer_node *wn) PARA_INFO_LOG("closing %s\n", writer_names[wn->writer_num]); w->close(wn); btr_remove_node(&wn->btrn); + task_reap(&wn->task); } static void close_writers(struct slot_info *s) @@ -444,6 +442,7 @@ static void close_filters(struct slot_info *s) if (f->close) f->close(fn); btr_remove_node(&fn->btrn); + task_reap(&fn->task); } free(s->fns); s->fns = NULL; @@ -459,7 +458,7 @@ static void notify_receivers(int error) continue; if (!s->receiver_node) continue; - task_notify(&s->receiver_node->task, error); + task_notify(s->receiver_node->task, error); } } @@ -497,22 +496,26 @@ static void open_filters(struct slot_info *s) s->fns = para_calloc(nf * sizeof(struct filter_node)); parent = s->receiver_node->btrn; for (i = 0; i < nf; i++) { + char buf[20]; struct filter *f = filters + a->filter_nums[i]; fn = s->fns + i; fn->filter_num = a->filter_nums[i]; fn->conf = a->filter_conf[i]; - fn->task.pre_select = f->pre_select; - fn->task.post_select = f->post_select; fn->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = f->name, .parent = parent, .handler = f->execute, .context = fn)); f->open(fn); - register_task(&sched, &fn->task); + sprintf(buf, "%s (slot %d)", f->name, (int)(s - slot)); + fn->task = task_register(&(struct task_info) { + .name = buf, + .pre_select = f->pre_select, + .post_select = f->post_select, + .context = fn, + }, &sched); parent = fn->btrn; PARA_NOTICE_LOG("%s filter %d/%d (%s) started in slot %d\n", audio_formats[s->format], i, nf, f->name, (int)(s - slot)); - sprintf(fn->task.status, "%s (slot %d)", f->name, (int)(s - slot)); } } @@ -566,10 +569,12 @@ static int open_receiver(int format) s->receiver_node = rn; PARA_NOTICE_LOG("started %s: %s receiver in slot %d\n", audio_formats[format], r->name, slot_num); - rn->task.pre_select = r->pre_select; - rn->task.post_select = r->post_select; - sprintf(rn->task.status, "%s receiver node", r->name); - register_task(&sched, &rn->task); + rn->task = task_register(&(struct task_info) { + .name = r->name, + .pre_select = r->pre_select, + .post_select = r->post_select, + .context = rn, + }, &sched); return slot_num; } @@ -584,7 +589,7 @@ static bool receiver_running(void) if (!s->receiver_node) continue; - if (s->receiver_node->task.error >= 0) + if (task_status(s->receiver_node->task) >= 0) return true; if (ss1 == ss2) return true; @@ -611,7 +616,7 @@ struct btr_node *audiod_get_btr_root(void) struct timeval rstime; if (!s->receiver_node) continue; - if (s->receiver_node->task.error < 0) + if (task_status(s->receiver_node->task) < 0) continue; btr_get_node_start(s->receiver_node->btrn, &rstime); if (newest_slot >= 0 && tv_diff(&rstime, &newest_rstime, NULL) < 0) @@ -995,16 +1000,15 @@ err: exit(EXIT_FAILURE); } -static void signal_pre_select(struct sched *s, struct task *t) +static void signal_pre_select(struct sched *s, void *context) { - struct signal_task *st = container_of(t, struct signal_task, task); + struct signal_task *st = context; para_fd_set(st->fd, &s->rfds, &s->max_fileno); } -static int signal_post_select(struct sched *s, __a_unused struct task *t) +static int signal_post_select(struct sched *s, __a_unused void *context) { int signum; - signum = para_next_signal(&s->rfds); switch (signum) { case SIGINT: @@ -1016,23 +1020,16 @@ static int signal_post_select(struct sched *s, __a_unused struct task *t) return 0; } -static void signal_setup_default(struct signal_task *st) +static void command_pre_select(struct sched *s, void *context) { - st->task.pre_select = signal_pre_select; - st->task.post_select = signal_post_select; - sprintf(st->task.status, "signal task"); -} - -static void command_pre_select(struct sched *s, struct task *t) -{ - struct command_task *ct = container_of(t, struct command_task, task); + struct command_task *ct = context; para_fd_set(ct->fd, &s->rfds, &s->max_fileno); } -static int command_post_select(struct sched *s, struct task *t) +static int command_post_select(struct sched *s, void *context) { int ret; - struct command_task *ct = container_of(t, struct command_task, task); + struct command_task *ct = context; static struct timeval last_status_dump; struct timeval tmp, delay; bool force = true; @@ -1067,17 +1064,21 @@ dump: static void init_command_task(struct command_task *ct) { - ct->task.pre_select = command_pre_select; - ct->task.post_select = command_post_select; - ct->task.error = 0; ct->fd = audiod_get_socket(); /* doesn't return on errors */ - sprintf(ct->task.status, "command task"); + + ct->task = task_register(&(struct task_info) { + .name = "command", + .pre_select = command_pre_select, + .post_select = command_post_select, + .context = ct, + }, &sched); } static void close_stat_pipe(void) { if (!stat_task->ct) return; + task_reap(&stat_task->ct->task); client_close(stat_task->ct); stat_task->ct = NULL; clear_and_dump_items(); @@ -1103,17 +1104,17 @@ static bool must_close_slot(int slot_num) if (s->format < 0) return false; - if (s->receiver_node && s->receiver_node->task.error >= 0) + if (s->receiver_node && task_status(s->receiver_node->task) >= 0) return false; for (i = 0; i < a->num_filters; i++) - if (s->fns && s->fns[i].task.error >= 0) + if (s->fns && task_status(s->fns[i].task) >= 0) return false; if (a->num_writers > 0) { for (i = 0; i < a->num_writers; i++) - if (s->wns && s->wns[i].task.error >= 0) + if (s->wns && task_status(s->wns[i].task) >= 0) return false; } else { - if (s->wns && s->wns[0].task.error >= 0) + if (s->wns && task_status(s->wns[0].task) >= 0) return false; } return true; @@ -1171,7 +1172,6 @@ static void start_stop_decoders(void) { int ret; struct slot_info *sl; - struct audio_format_info *a; close_unused_slots(); if (audiod_status != AUDIOD_ON || @@ -1185,17 +1185,15 @@ static void start_stop_decoders(void) return; } sl = slot + ret; - a = afi + sl->format; - if (a->num_filters) - open_filters(sl); + open_filters(sl); open_writers(sl); activate_grab_clients(&sched); btr_log_tree(sl->receiver_node->btrn, LL_NOTICE); } -static void status_pre_select(struct sched *s, struct task *t) +static void status_pre_select(struct sched *s, void *context) { - struct status_task *st = container_of(t, struct status_task, task); + struct status_task *st = context; int i, ret, cafn = stat_task->current_audio_format_num; if (must_start_decoder()) @@ -1225,15 +1223,15 @@ min_delay: } /* restart the client task if necessary */ -static int status_post_select(struct sched *s, struct task *t) +static int status_post_select(struct sched *s, void *context) { - struct status_task *st = container_of(t, struct status_task, task); + struct status_task *st = context; if (audiod_status == AUDIOD_OFF) { if (!st->ct) goto out; - if (st->ct->task.error >= 0) { - task_notify(&st->ct->task, E_AUDIOD_OFF); + if (task_status(st->ct->task) >= 0) { + task_notify(st->ct->task, E_AUDIOD_OFF); goto out; } close_stat_pipe(); @@ -1256,14 +1254,14 @@ static int status_post_select(struct sched *s, struct task *t) struct timeval diff; tv_diff(now, &st->last_status_read, &diff); if (diff.tv_sec > 61) - task_notify(&st->ct->task, E_STATUS_TIMEOUT); + task_notify(st->ct->task, E_STATUS_TIMEOUT); goto out; } btr_merge(st->btrn, st->min_iqs); sz = btr_next_buffer(st->btrn, &buf); ret = for_each_stat_item(buf, sz, update_item); if (ret < 0) { - task_notify(&st->ct->task, -ret); + task_notify(st->ct->task, -ret); goto out; } if (sz != ret) { @@ -1305,14 +1303,18 @@ out: static void init_status_task(struct status_task *st) { memset(st, 0, sizeof(struct status_task)); - st->task.pre_select = status_pre_select; - st->task.post_select = status_post_select; st->sa_time_diff_sign = 1; st->clock_diff_count = conf.clock_diff_count_arg; st->current_audio_format_num = -1; - sprintf(st->task.status, "stat"); st->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "stat")); + + stat_task->task = task_register(&(struct task_info) { + .name = "stat", + .pre_select = status_pre_select, + .post_select = status_post_select, + .context = stat_task, + }, &sched); } static void set_initial_status(void) @@ -1416,7 +1418,6 @@ int main(int argc, char *argv[]) FOR_EACH_SLOT(i) clear_slot(i); setup_signal_handling(); - signal_setup_default(sig_task); init_status_task(stat_task); init_command_task(cmd_task); @@ -1424,12 +1425,17 @@ int main(int argc, char *argv[]) if (conf.daemon_given) daemonize(false /* parent exits immediately */); - register_task(&sched, &sig_task->task); - register_task(&sched, &cmd_task->task); - register_task(&sched, &stat_task->task); + sig_task->task = task_register(&(struct task_info) { + .name = "signal", + .pre_select = signal_pre_select, + .post_select = signal_post_select, + .context = sig_task, + }, &sched); + sched.default_timeout.tv_sec = 2; sched.default_timeout.tv_usec = 999 * 1000; ret = schedule(&sched); + sched_shutdown(&sched); PARA_EMERG_LOG("%s\n", para_strerror(-ret)); return EXIT_FAILURE; diff --git a/client.c b/client.c index b39a8b01..987a637f 100644 --- a/client.c +++ b/client.c @@ -39,24 +39,24 @@ __printf_2_3 void (*para_log)(int, const char*, ...) = stderr_log; #include "afs_completion.h" struct exec_task { - struct task task; + struct task *task; struct btr_node *btrn; char *result_buf; size_t result_size; }; -static void exec_pre_select(struct sched *s, struct task *t) +static void exec_pre_select(struct sched *s, void *context) { - struct exec_task *et = container_of(t, struct exec_task, task); + struct exec_task *et = context; int ret = btr_node_status(et->btrn, 0, BTR_NT_LEAF); if (ret != 0) sched_min_delay(s); } -static int exec_post_select(__a_unused struct sched *s, struct task *t) +static int exec_post_select(__a_unused struct sched *s, void *context) { - struct exec_task *et = container_of(t, struct exec_task, task); + struct exec_task *et = context; struct btr_node *btrn = et->btrn; char *buf; size_t sz; @@ -93,11 +93,6 @@ static int execute_client_command(const char *cmd, char **result) int ret; struct sched command_sched = {.default_timeout = {.tv_sec = 1}}; struct exec_task exec_task = { - .task = { - .pre_select = exec_pre_select, - .post_select = exec_post_select, - .status = "client exec task", - }, .result_buf = para_strdup(""), .result_size = 1, }; @@ -107,11 +102,17 @@ static int execute_client_command(const char *cmd, char **result) goto out; exec_task.btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "exec_collect")); - register_task(&command_sched, &exec_task.task); + exec_task.task = task_register(&(struct task_info) { + .name = "client exec", + .pre_select = exec_pre_select, + .post_select = exec_post_select, + .context = &exec_task, + }, &command_sched); ret = client_connect(ct, &command_sched, NULL, exec_task.btrn); if (ret < 0) goto out; schedule(&command_sched); + sched_shutdown(&command_sched); *result = exec_task.result_buf; btr_remove_node(&exec_task.btrn); ret = 1; @@ -495,6 +496,7 @@ __noreturn static void interactive_session(void) goto out; para_log = i9e_log; ret = schedule(&sched); + sched_shutdown(&sched); i9e_close(); para_log = stderr_log; out: @@ -528,36 +530,29 @@ __noreturn static void print_completions(void) struct supervisor_task { bool stdout_task_started; - struct task task; + struct task *task; }; -static int supervisor_post_select(struct sched *s, struct task *t) +static int supervisor_post_select(struct sched *s, void *context) { - struct supervisor_task *svt = container_of(t, struct supervisor_task, - task); + struct supervisor_task *svt = context; + int ret = task_status(ct->task); - if (ct->task.error < 0) - return ct->task.error; + if (ret < 0) + return ret; if (!svt->stdout_task_started && ct->status == CL_EXECUTING) { - stdout_set_defaults(&sot); - register_task(s, &sot.task); + stdout_task_register(&sot, s); svt->stdout_task_started = true; return 1; } if (ct->status == CL_SENDING) { - stdin_set_defaults(&sit); - register_task(s, &sit.task); + stdin_task_register(&sit, s); return -E_TASK_STARTED; } return 0; } -static struct supervisor_task supervisor_task = { - .task = { - .post_select = supervisor_post_select, - .status = "supervisor task" - } -}; +static struct supervisor_task supervisor_task; /** * The client program to connect to para_server. @@ -605,20 +600,29 @@ int main(int argc, char *argv[]) goto out; sot.btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "stdout", .parent = ct->btrn[0])); - register_task(&sched, &supervisor_task.task); + supervisor_task.task = task_register(&(struct task_info) { + .name = "supervisor", + .post_select = supervisor_post_select, + .context = &supervisor_task, + }, &sched); + ret = schedule(&sched); - if (ret >= 0 && ct->task.error < 0) { - switch(ct->task.error) { - /* these are not errors */ - case -E_SERVER_CMD_SUCCESS: - case -E_EOF: - case -E_SERVER_EOF: - case -E_BTR_EOF: - ret = 0; - break; - default: ret = -E_SERVER_CMD_FAILURE; + if (ret >= 0) { + ret = task_status(ct->task); + if (ret < 0) { + switch (ret) { + /* these are not errors */ + case -E_SERVER_CMD_SUCCESS: + case -E_EOF: + case -E_SERVER_EOF: + case -E_BTR_EOF: + ret = 0; + break; + default: ret = -E_SERVER_CMD_FAILURE; + } } } + sched_shutdown(&sched); out: if (ret < 0) PARA_ERROR_LOG("%s\n", para_strerror(-ret)); diff --git a/client.h b/client.h index e304f092..0a0c55cb 100644 --- a/client.h +++ b/client.h @@ -47,7 +47,7 @@ struct client_task { /** Paraslash user name. */ char *user; /** The client task structure. */ - struct task task; + struct task *task; /** List of features supported by the server. */ char **features; }; diff --git a/client_common.c b/client_common.c index 8212abb1..c111e351 100644 --- a/client_common.c +++ b/client_common.c @@ -53,24 +53,19 @@ void client_close(struct client_task *ct) free(ct); } -/** +/* * The preselect hook for server commands. * - * \param s Pointer to the scheduler. - * \param t Pointer to the task struct for this command. - * * The task pointer must contain a pointer to the initialized client data * structure as it is returned by client_open(). * * This function checks the state of the connection and adds the file descriptor - * of the connection to the read or write fd set of \a s accordingly. - * - * \sa register_task() client_open(), struct sched, struct task. + * of the connection to the read or write fd set of s accordingly. */ -static void client_pre_select(struct sched *s, struct task *t) +static void client_pre_select(struct sched *s, void *context) { int ret; - struct client_task *ct = container_of(t, struct client_task, task); + struct client_task *ct = context; if (ct->scc.fd < 0) return; @@ -268,27 +263,22 @@ static int send_sb_command(struct client_task *ct) return send_sb(ct, 0, command, len, SBD_COMMAND, false); } -/** +/* * The post select hook for client commands. * - * \param s Pointer to the scheduler. - * \param t Pointer to the task struct for this command. - * * Depending on the current state of the connection and the status of the read - * and write fd sets of \a s, this function performs the necessary steps to - * authenticate the connection, to send the command given by \a t->private_data + * and write fd sets of s, this function performs the necessary steps to + * authenticate the connection, to send the command given by t->private_data * and to receive para_server's output, if any. - * - * \sa struct sched, struct task. */ -static int client_post_select(struct sched *s, struct task *t) +static int client_post_select(struct sched *s, void *context) { - struct client_task *ct = container_of(t, struct client_task, task); + struct client_task *ct = context; int ret = 0; size_t n; char buf[CLIENT_BUFSIZE]; - ret = task_get_notification(t); + ret = task_get_notification(ct->task); if (ret < 0) goto out; if (ct->scc.fd < 0) @@ -496,11 +486,13 @@ int client_connect(struct client_task *ct, struct sched *s, EMBRACE(.name = "client recv", .parent = NULL, .child = child)); ct->btrn[1] = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "client send", .parent = parent, .child = NULL)); - ct->task.pre_select = client_pre_select; - ct->task.post_select = client_post_select; - ct->task.error = 0; - sprintf(ct->task.status, "client"); - register_task(s, &ct->task); + + ct->task = task_register(&(struct task_info) { + .name = "client", + .pre_select = client_pre_select, + .post_select = client_post_select, + .context = ct, + }, s); return 1; err_out: close(ct->scc.fd); diff --git a/compress_filter.c b/compress_filter.c index b488f55c..119e0169 100644 --- a/compress_filter.c +++ b/compress_filter.c @@ -41,9 +41,9 @@ static void compress_close(struct filter_node *fn) free(fn->private_data); } -static int compress_post_select(__a_unused struct sched *s, struct task *t) +static int compress_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_compress_data *pcd = fn->private_data; struct btr_node *btrn = fn->btrn; bool inplace = btr_inplace_ok(btrn); diff --git a/dccp_recv.c b/dccp_recv.c index 796a7e6b..a45b572d 100644 --- a/dccp_recv.c +++ b/dccp_recv.c @@ -119,25 +119,24 @@ static void *dccp_recv_parse_config(int argc, char **argv) return tmp; } -static void dccp_recv_pre_select(struct sched *s, struct task *t) +static void dccp_recv_pre_select(struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; - t->error = 0; - if (generic_recv_pre_select(s, t) <= 0) + if (generic_recv_pre_select(s, rn) <= 0) return; para_fd_set(rn->fd, &s->rfds, &s->max_fileno); } -static int dccp_recv_post_select(struct sched *s, struct task *t) +static int dccp_recv_post_select(struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; struct btr_node *btrn = rn->btrn; struct iovec iov[2]; int ret, iovcnt; size_t num_bytes; - ret = task_get_notification(t); + ret = task_get_notification(rn->task); if (ret < 0) goto out; ret = btr_node_status(btrn, 0, BTR_NT_ROOT); diff --git a/fecdec_filter.c b/fecdec_filter.c index b8e130d4..0909007a 100644 --- a/fecdec_filter.c +++ b/fecdec_filter.c @@ -436,9 +436,9 @@ static void fecdec_close(struct filter_node *fn) fn->private_data = NULL; } -static int fecdec_post_select(__a_unused struct sched *s, struct task *t) +static int fecdec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct btr_node *btrn = fn->btrn; int ret; struct fec_header h; diff --git a/file_write.c b/file_write.c index 7d63469f..5d43fe7d 100644 --- a/file_write.c +++ b/file_write.c @@ -74,9 +74,9 @@ out: return ret; } -static void file_write_pre_select(struct sched *s, struct task *t) +static void file_write_pre_select(struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_file_write_data *pfwd = wn->private_data; int ret = btr_node_status(wn->btrn, wn->min_iqs, BTR_NT_LEAF); @@ -97,17 +97,16 @@ static void file_write_close(struct writer_node *wn) free(pfwd); } -static int file_write_post_select(__a_unused struct sched *s, - struct task *t) +static int file_write_post_select(__a_unused struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_file_write_data *pfwd = wn->private_data; struct btr_node *btrn = wn->btrn; int ret; char *buf; size_t bytes; - ret = task_get_notification(t); + ret = task_get_notification(wn->task); if (ret < 0) goto out; ret = btr_node_status(btrn, wn->min_iqs, BTR_NT_LEAF); diff --git a/filter.c b/filter.c index b3dc022e..55afbd32 100644 --- a/filter.c +++ b/filter.c @@ -112,13 +112,13 @@ int main(int argc, char *argv[]) goto out; sit->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "stdin")); - stdin_set_defaults(sit); - register_task(&s, &sit->task); + stdin_task_register(sit, &s); fns = para_malloc(conf.filter_given * sizeof(*fns)); for (i = 0, parent = sit->btrn; i < conf.filter_given; i++) { char *fa = conf.filter_arg[i]; struct filter_node *fn; + struct task_info ti; fn = fns[i] = para_calloc(sizeof(*fn)); ret = check_filter_arg(fa, &fn->conf); @@ -128,26 +128,27 @@ int main(int argc, char *argv[]) } fn->filter_num = ret; f = filters + fn->filter_num; - sprintf(fn->task.status, "%s", f->name); PARA_DEBUG_LOG("filter #%d: %s\n", i, f->name); fn->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = f->name, .parent = parent, .handler = f->execute, .context = fn)); - fn->task.pre_select = f->pre_select; - fn->task.post_select = f->post_select; + ti.name = f->name; + ti.pre_select = f->pre_select; + ti.post_select = f->post_select; + ti.context = fn; f->open(fn); - register_task(&s, &fn->task); + fn->task = task_register(&ti, &s); parent = fn->btrn; } sot->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "stdout", .parent = parent)); - stdout_set_defaults(sot); - register_task(&s, &sot->task); + stdout_task_register(sot, &s); s.default_timeout.tv_sec = 1; s.default_timeout.tv_usec = 0; btr_log_tree(sit->btrn, LL_INFO); ret = schedule(&s); + sched_shutdown(&s); out_cleanup: for (i--; i >= 0; i--) { struct filter_node *fn = fns[i]; diff --git a/filter.h b/filter.h index de8ce8e8..7af0fbbb 100644 --- a/filter.h +++ b/filter.h @@ -27,7 +27,7 @@ struct filter_node { /** The buffer tree node. */ struct btr_node *btrn; /** The task corresponding to this filter node. */ - struct task task; + struct task *task; /** The minimal input queue size, see \ref btr_node_status(). */ size_t min_iqs; }; @@ -101,14 +101,14 @@ struct filter { * this function is to set file descriptors to be watched by the * subsequent select call to the two fd sets. */ - void (*pre_select)(struct sched *s, struct task *t); + void (*pre_select)(struct sched *s, void *context); /** * Convert (filter) the given data. * * Pointer to the converting function of the filter. On errors, the * post_select function is supposed to return a negative error code. */ - int (*post_select)(struct sched *s, struct task *t); + int (*post_select)(struct sched *s, void *context); /** * Answer a buffer tree query. * @@ -121,7 +121,7 @@ struct filter { void filter_init(void); int check_filter_arg(char *filter_arg, void **conf); void print_filter_helps(unsigned flags); -void generic_filter_pre_select(struct sched *s, struct task *t); +void generic_filter_pre_select(struct sched *s, void *context); int decoder_execute(const char *cmd, unsigned sample_rate, unsigned channels, char **result); diff --git a/filter_common.c b/filter_common.c index 8ce66ea8..616a7601 100644 --- a/filter_common.c +++ b/filter_common.c @@ -132,18 +132,17 @@ void print_filter_helps(unsigned flags) * Set select timeout of the scheduler. * * \param s The scheduler. - * \param t The task struct of this filter. + * \param context Pointer to the filter node (task context). * * This looks at the status of the btr node of the filter. If data is available * in the input queue of the filter, or if an error occurred, a minimal timeout * for the next select call is requested from the scheduler. Otherwise the * scheduler timeout is left unchanged. */ -void generic_filter_pre_select(struct sched *s, struct task *t) +void generic_filter_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; - t->error = 0; if (btr_node_status(fn->btrn, fn->min_iqs, BTR_NT_INTERNAL) != 0) sched_min_delay(s); } diff --git a/flacdec_filter.c b/flacdec_filter.c index 2a58044c..b741f6b2 100644 --- a/flacdec_filter.c +++ b/flacdec_filter.c @@ -210,9 +210,9 @@ static bool output_queue_full(struct btr_node *btrn) return btr_get_output_queue_size(btrn) > FLACDEC_MAX_OUTPUT_SIZE; } -static void flacdec_pre_select(struct sched *s, struct task *t) +static void flacdec_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_flacdec_data *pfd = fn->private_data; struct btr_node *btrn = fn->btrn; int ret; @@ -226,9 +226,9 @@ static void flacdec_pre_select(struct sched *s, struct task *t) return sched_min_delay(s); } -static int flacdec_post_select(__a_unused struct sched *s, struct task *t) +static int flacdec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_flacdec_data *pfd = fn->private_data; struct btr_node *btrn = fn->btrn; int ret; diff --git a/grab_client.c b/grab_client.c index 672c96f1..7948c449 100644 --- a/grab_client.c +++ b/grab_client.c @@ -54,7 +54,7 @@ struct grab_client { /** The point of the grab client's node in the buffer tree. */ struct btr_node *btrn; /* The task of this grab client. */ - struct task task; + struct task *task; /** Belongs to either the active or the inactive list. */ struct list_head node; }; @@ -92,9 +92,9 @@ err: return -E_GC_WRITE; } -static void gc_pre_select(struct sched *s, struct task *t) +static void gc_pre_select(struct sched *s, void *context) { - struct grab_client *gc = container_of(t, struct grab_client, task); + struct grab_client *gc = context; int ret = btr_node_status(gc->btrn, 0, BTR_NT_LEAF); if (ret == 0) @@ -108,7 +108,7 @@ static void gc_pre_select(struct sched *s, struct task *t) * We need this forward declaration as post_select() needs * activate_grab_client and vice versa. */ -static int gc_post_select(struct sched *s, struct task *t); +static int gc_post_select(struct sched *s, void *context); /** * Move a grab client to the active list and start it. @@ -129,12 +129,13 @@ static void gc_activate(struct grab_client *gc, struct sched *s) list_move(&gc->node, &active_grab_client_list); gc->btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = name, .parent = parent)); - gc->task.pre_select = gc_pre_select; - gc->task.post_select = gc_post_select; - snprintf(gc->task.status, sizeof(gc->task.status) - 1, "%s", name); - gc->task.status[sizeof(gc->task.status) - 1] = '\0'; - gc->task.error = 0; - register_task(s, &gc->task); + + gc->task = task_register(&(struct task_info) { + .name = name, + .pre_select = gc_pre_select, + .post_select = gc_post_select, + .context = gc, + }, s); } /** @@ -184,9 +185,9 @@ static int gc_close(struct grab_client *gc, int err) return 0; } -static int gc_post_select(__a_unused struct sched *s, struct task *t) +static int gc_post_select(__a_unused struct sched *s, void *context) { - struct grab_client *gc = container_of(t, struct grab_client, task); + struct grab_client *gc = context; struct btr_node *btrn = gc->btrn; int ret; size_t sz; diff --git a/gui.c b/gui.c index b077b00a..92b7f7cf 100644 --- a/gui.c +++ b/gui.c @@ -124,11 +124,11 @@ GUI_COMMANDS static struct gui_command command_list[] = {GUI_COMMANDS {.name = NULL}}; struct input_task { - struct task task; + struct task *task; }; struct status_task { - struct task task; + struct task *task; pid_t pid; char *buf; int bufsize, loaded; @@ -140,7 +140,7 @@ struct status_task { #define COMMAND_BUF_SIZE 32768 struct exec_task { - struct task task; + struct task *task; char command_buf[2][COMMAND_BUF_SIZE]; /* stdout/stderr of command */ int cbo[2]; /* command buf offsets */ unsigned flags[2]; /* passed to for_each_line() */ @@ -597,25 +597,25 @@ static void clear_all_items(void) } } -static void status_pre_select(struct sched *s, struct task *t) +static void status_pre_select(struct sched *s, void *context) { - struct status_task *st = container_of(t, struct status_task, task); + struct status_task *st = context; if (st->fd >= 0) para_fd_set(st->fd, &s->rfds, &s->max_fileno); - if (task_get_notification(t) < 0) + if (task_get_notification(st->task) < 0) return sched_min_delay(s); if (st->fd < 0) sched_request_barrier_or_min_delay(&st->next_exec, s); } -static int status_post_select(struct sched *s, struct task *t) +static int status_post_select(struct sched *s, void *context) { - struct status_task *st = container_of(t, struct status_task, task); + struct status_task *st = context; size_t sz; int ret, ret2; - ret = task_get_notification(t); + ret = task_get_notification(st->task); if (ret == -E_GUI_SIGCHLD && st->pid > 0) { int exit_status; if (waitpid(st->pid, &exit_status, WNOHANG) == st->pid) { @@ -897,7 +897,7 @@ static void reread_conf(void) /* * React to various signal-related events */ -static int signal_post_select(struct sched *s, __a_unused struct task *t) +static int signal_post_select(struct sched *s, __a_unused void *context) { int ret = para_next_signal(&s->rfds); @@ -933,22 +933,23 @@ static enum exec_status exec_status(void) return EXEC_IDLE; } -static void exec_pre_select(struct sched *s, struct task *t) +static void exec_pre_select(struct sched *s, void *context) { + struct exec_task *et = context; if (exec_fds[0] >= 0) para_fd_set(exec_fds[0], &s->rfds, &s->max_fileno); if (exec_fds[1] >= 0) para_fd_set(exec_fds[1], &s->rfds, &s->max_fileno); - if (task_get_notification(t) < 0) + if (task_get_notification(et->task) < 0) sched_min_delay(s); } -static int exec_post_select(struct sched *s, struct task *t) +static int exec_post_select(struct sched *s, void *context) { - struct exec_task *ct = container_of(t, struct exec_task, task); + struct exec_task *ct = context; int i, ret; - ret = task_get_notification(t); + ret = task_get_notification(ct->task); if (ret == -E_GUI_SIGCHLD && exec_pid > 0) { int exit_status; if (waitpid(exec_pid, &exit_status, WNOHANG) == exec_pid) { @@ -993,7 +994,7 @@ static int exec_post_select(struct sched *s, struct task *t) return 0; } -static void input_pre_select(struct sched *s, __a_unused struct task *t) +static void input_pre_select(struct sched *s, __a_unused void *context) { if (exec_status() != EXEC_XCMD) para_fd_set(STDIN_FILENO, &s->rfds, &s->max_fileno); @@ -1090,7 +1091,7 @@ static void handle_command(int c) km_keyname(c)); } -static int input_post_select(__a_unused struct sched *s, __a_unused struct task *t) +static int input_post_select(__a_unused struct sched *s, __a_unused void *context) { int ret; enum exec_status exs = exec_status(); @@ -1130,9 +1131,9 @@ static int input_post_select(__a_unused struct sched *s, __a_unused struct task return 0; } -static void signal_pre_select(struct sched *s, struct task *t) +static void signal_pre_select(struct sched *s, void *context) { - struct signal_task *st = container_of(t, struct signal_task, task); + struct signal_task *st = context; para_fd_set(st->fd, &s->rfds, &s->max_fileno); } @@ -1406,51 +1407,49 @@ __noreturn static void print_help_and_die(void) static int setup_tasks_and_schedule(void) { + struct exec_task exec_task = {.task = NULL}; + struct status_task status_task = {.fd = -1}; + struct input_task input_task = {.task = NULL}; + struct signal_task signal_task = {.task = NULL}; struct sched sched = { .default_timeout = { .tv_sec = conf.timeout_arg / 1000, .tv_usec = (conf.timeout_arg % 1000) * 1000, }, }; - struct exec_task exec_task = { - .task = { - .status = "exec", - .pre_select = exec_pre_select, - .post_select = exec_post_select, - }, - }; - struct status_task status_task = { - .task = { - .status = "status", - .pre_select = status_pre_select, - .post_select = status_post_select, - }, - .fd = -1 - }; - struct input_task input_task = { - .task = { - .status = "input", - .pre_select = input_pre_select, - .post_select = input_post_select, - }, - }; - struct signal_task signal_task = { - .task = { - .status = "signal", - .pre_select = signal_pre_select, - .post_select = signal_post_select, - }, - }; + + exec_task.task = task_register(&(struct task_info) { + .name = "exec", + .pre_select = exec_pre_select, + .post_select = exec_post_select, + .context = &exec_task, + }, &sched); + + status_task.task = task_register(&(struct task_info) { + .name = "status", + .pre_select = status_pre_select, + .post_select = status_post_select, + .context = &status_task, + }, &sched); + + input_task.task = task_register(&(struct task_info) { + .name = "input", + .pre_select = input_pre_select, + .post_select = input_post_select, + .context = &input_task, + }, &sched); + signal_task.fd = para_signal_init(); para_install_sighandler(SIGINT); para_install_sighandler(SIGTERM); para_install_sighandler(SIGCHLD); para_install_sighandler(SIGUSR1); - - register_task(&sched, &exec_task.task); - register_task(&sched, &status_task.task); - register_task(&sched, &input_task.task); - register_task(&sched, &signal_task.task); + signal_task.task = task_register(&(struct task_info) { + .name = "signal", + .pre_select = signal_pre_select, + .post_select = signal_post_select, + .context = &signal_task, + }, &sched); return schedule(&sched); } diff --git a/http_recv.c b/http_recv.c index 1f02e48d..9deea791 100644 --- a/http_recv.c +++ b/http_recv.c @@ -60,12 +60,12 @@ static char *make_request_msg(void) return ret; } -static void http_recv_pre_select(struct sched *s, struct task *t) +static void http_recv_pre_select(struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; struct private_http_recv_data *phd = rn->private_data; - if (generic_recv_pre_select(s, t) <= 0) + if (generic_recv_pre_select(s, rn) <= 0) return; if (phd->status == HTTP_CONNECTED) para_fd_set(rn->fd, &s->wfds, &s->max_fileno); @@ -78,16 +78,16 @@ static void http_recv_pre_select(struct sched *s, struct task *t) * area with data read from the socket. In any case, update the state of the * connection if necessary. */ -static int http_recv_post_select(struct sched *s, struct task *t) +static int http_recv_post_select(struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; struct private_http_recv_data *phd = rn->private_data; struct btr_node *btrn = rn->btrn; int ret, iovcnt; struct iovec iov[2]; size_t num_bytes; - ret = task_get_notification(t); + ret = task_get_notification(rn->task); if (ret < 0) goto out; ret = btr_node_status(btrn, 0, BTR_NT_ROOT); diff --git a/interactive.c b/interactive.c index 9f2b7195..3d2f6d68 100644 --- a/interactive.c +++ b/interactive.c @@ -29,7 +29,7 @@ struct i9e_private { FILE *stderr_stream; int num_columns; char empty_line[1000]; - struct task task; + struct task *task; struct btr_node *stdout_btrn; bool last_write_was_status; bool line_handler_running; @@ -52,7 +52,7 @@ static struct i9e_private i9e_private, *i9ep = &i9e_private; */ int i9e_get_error(void) { - return i9ep->task.error; + return task_status(i9ep->task); } static bool is_prefix(const char *partial, const char *full, size_t len) @@ -314,7 +314,7 @@ free_line: free(line); } -static int i9e_post_select(__a_unused struct sched *s, __a_unused struct task *t) +static int i9e_post_select(__a_unused struct sched *s, __a_unused void *context) { int ret; struct i9e_client_info *ici = i9ep->ici; @@ -371,7 +371,7 @@ out: return ret; } -static void i9e_pre_select(struct sched *s, __a_unused struct task *t) +static void i9e_pre_select(struct sched *s, __a_unused void *context) { int ret; @@ -439,7 +439,6 @@ static int dispatch_key(__a_unused int count, int key) * The caller must allocate and initialize the structure \a ici points to. * * \return Standard. - * \sa \ref register_task(). */ int i9e_open(struct i9e_client_info *ici, struct sched *s) { @@ -453,10 +452,13 @@ int i9e_open(struct i9e_client_info *ici, struct sched *s) ret = mark_fd_nonblocking(ici->fds[1]); if (ret < 0) return ret; - i9ep->task.pre_select = i9e_pre_select; - i9ep->task.post_select = i9e_post_select; - sprintf(i9ep->task.status, "i9e"); - register_task(s, &i9ep->task); + i9ep->task = task_register(&(struct task_info) { + .name = "i9e", + .pre_select = i9e_pre_select, + .post_select = i9e_post_select, + .context = i9ep, + }, s); + rl_readline_name = "para_i9e"; rl_basic_word_break_characters = " "; rl_attempted_completion_function = i9e_completer; diff --git a/mp3dec_filter.c b/mp3dec_filter.c index feaa1d1f..a51c77e2 100644 --- a/mp3dec_filter.c +++ b/mp3dec_filter.c @@ -77,9 +77,9 @@ static void mp3dec_close(struct filter_node *fn) #define MP3DEC_MAX_FRAME 8192 -static int mp3dec_post_select(__a_unused struct sched *s, struct task *t) +static int mp3dec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; int i, ret; struct private_mp3dec_data *pmd = fn->private_data; struct btr_node *btrn = fn->btrn; diff --git a/oggdec_filter.c b/oggdec_filter.c index 3222b4aa..8a3e261e 100644 --- a/oggdec_filter.c +++ b/oggdec_filter.c @@ -181,9 +181,9 @@ out: #define OGGDEC_MAX_OUTPUT_SIZE (96 * 1024) #define OGGDEC_OUTPUT_CHUNK_SIZE (32 * 1024) -static void ogg_pre_select(struct sched *s, struct task *t) +static void ogg_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_oggdec_data *pod = fn->private_data; struct btr_node *btrn = fn->btrn; int ret; @@ -198,9 +198,9 @@ static void ogg_pre_select(struct sched *s, struct task *t) sched_min_delay(s); } -static int ogg_post_select(__a_unused struct sched *s, struct task *t) +static int ogg_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_oggdec_data *pod = fn->private_data; struct btr_node *btrn = fn->btrn; int ret, have; diff --git a/opusdec_filter.c b/opusdec_filter.c index 9022fbab..4e69a46e 100644 --- a/opusdec_filter.c +++ b/opusdec_filter.c @@ -207,9 +207,9 @@ static int decode_packet(struct opusdec_context *ctx, ogg_packet *op, #define OPUSDEC_MAX_OUTPUT_SIZE (1024 * 1024) -static int opusdec_post_select(__a_unused struct sched *s, struct task *t) +static int opusdec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct opusdec_context *ctx = fn->private_data; struct btr_node *btrn = fn->btrn; int ret; @@ -269,9 +269,9 @@ out: return ret; } -static void opusdec_pre_select(struct sched *s, struct task *t) +static void opusdec_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct opusdec_context *ctx = fn->private_data; int ret = btr_node_status(fn->btrn, fn->min_iqs, BTR_NT_INTERNAL); diff --git a/oss_write.c b/oss_write.c index d547acf7..77fe2a51 100644 --- a/oss_write.c +++ b/oss_write.c @@ -44,9 +44,9 @@ static int get_oss_format(enum sample_format sf) } } -static void oss_pre_select(struct sched *s, struct task *t) +static void oss_pre_select(struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_oss_write_data *powd = wn->private_data; int ret = btr_node_status(wn->btrn, wn->min_iqs, BTR_NT_LEAF); @@ -157,17 +157,16 @@ err_free: return ret; } -static int oss_post_select(__a_unused struct sched *s, - struct task *t) +static int oss_post_select(__a_unused struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_oss_write_data *powd = wn->private_data; struct btr_node *btrn = wn->btrn; size_t frames, bytes; int ret; char *data; - ret = task_get_notification(t); + ret = task_get_notification(wn->task); if (ret < 0) goto out; ret = btr_node_status(btrn, wn->min_iqs, BTR_NT_LEAF); diff --git a/osx_write.c b/osx_write.c index 889854c2..229930be 100644 --- a/osx_write.c +++ b/osx_write.c @@ -274,9 +274,9 @@ static inline bool need_drain_delay(struct private_osx_write_data *powd) return btr_get_input_queue_size(powd->callback_btrn) != 0; } -static void osx_write_pre_select(struct sched *s, struct task *t) +static void osx_write_pre_select(struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_osx_write_data *powd = wn->private_data; int ret; bool drain_delay_nec = false; @@ -301,14 +301,14 @@ static void osx_write_pre_select(struct sched *s, struct task *t) sched_request_timeout_ms(50, s); } -static int osx_write_post_select(__a_unused struct sched *s, struct task *t) +static int osx_write_post_select(__a_unused struct sched *s, void *context) { - struct writer_node *wn = container_of(t, struct writer_node, task); + struct writer_node *wn = context; struct private_osx_write_data *powd = wn->private_data; struct btr_node *btrn = wn->btrn; int ret; - ret = task_get_notification(t); + ret = task_get_notification(wn->task); if (ret < 0) goto fail; if (!powd) { diff --git a/play.c b/play.c index 5736a2dd..c2a383c1 100644 --- a/play.c +++ b/play.c @@ -57,7 +57,7 @@ enum state_change_request_type { }; struct play_task { - struct task task; + struct task *task; /* A bit array of invalid files (those will be skipped). */ bool *invalid; /* The file which is currently open. */ @@ -243,13 +243,16 @@ static void wipe_receiver_node(struct play_task *pt) /* returns: 0 not eof, 1: eof, < 0: fatal error. */ static int get_playback_error(struct play_task *pt) { - int err = pt->wn.task.error; + int err; + if (!pt->wn.task) + return 0; + err = task_status(pt->wn.task); if (err >= 0) return 0; - if (pt->fn.task.error >= 0) + if (task_status(pt->fn.task) >= 0) return 0; - if (pt->rn.task.error >= 0) + if (task_status(pt->rn.task) >= 0) return 0; if (err == -E_BTR_EOF || err == -E_RECV_EOF || err == -E_EOF || err == -E_WRITE_COMMON_EOF) @@ -267,16 +270,19 @@ static int eof_cleanup(struct play_task *pt) if (ret == 0) return ret; PARA_NOTICE_LOG("cleaning up wn/fn nodes\n"); + task_reap(&pt->wn.task); w->close(&pt->wn); btr_remove_node(&pt->wn.btrn); w->free_config(pt->wn.conf); memset(&pt->wn, 0, sizeof(struct writer_node)); + task_reap(&pt->fn.task); decoder->close(&pt->fn); btr_remove_node(&pt->fn.btrn); free(pt->fn.conf); memset(&pt->fn, 0, sizeof(struct filter_node)); + task_reap(&pt->rn.task); btr_remove_node(&pt->rn.btrn); /* * On eof (ret > 0), we do not wipe the receiver node struct until a @@ -295,7 +301,7 @@ static int shuffle_compare(__a_unused const void *a, __a_unused const void *b) static void shuffle(char **base, size_t num) { - srandom(now->tv_sec); + srandom(time(NULL)); qsort(base, num, sizeof(char *), shuffle_compare); } @@ -351,9 +357,6 @@ static int open_new_file(struct play_task *pt) free(tmp); tmp = NULL; } - pt->rn.task.pre_select = afh_recv->pre_select; - pt->rn.task.post_select = afh_recv->post_select; - sprintf(pt->rn.task.status, "%s receiver node", afh_recv->name); return 1; fail: wipe_receiver_node(pt); @@ -363,7 +366,7 @@ fail: static int load_file(struct play_task *pt) { const char *af; - char *tmp; + char *tmp, buf[20]; int ret; struct filter *decoder; @@ -373,7 +376,6 @@ static int load_file(struct play_task *pt) if (ret < 0) return ret; } else { - char buf[20]; pt->rn.btrn = new_recv_btrn(&pt->rn); sprintf(buf, "repos %lu", pt->start_chunk); ret = btr_exec_up(pt->rn.btrn, buf, &tmp); @@ -392,9 +394,6 @@ static int load_file(struct play_task *pt) goto fail; pt->fn.filter_num = ret; decoder = filters + ret; - pt->fn.task.pre_select = decoder->pre_select; - pt->fn.task.post_select = decoder->post_select; - sprintf(pt->fn.task.status, "%s decoder", af); pt->fn.btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = decoder->name, .parent = pt->rn.btrn, .handler = decoder->execute, .context = &pt->fn)); @@ -402,11 +401,23 @@ static int load_file(struct play_task *pt) /* setup default writer */ pt->wn.conf = check_writer_arg_or_die(NULL, &pt->wn.writer_num); - pt->wn.task.error = 0; /* success, register tasks */ - register_task(&sched, &pt->rn.task); - register_task(&sched, &pt->fn.task); + pt->rn.task = task_register( + &(struct task_info) { + .name = afh_recv->name, + .pre_select = afh_recv->pre_select, + .post_select = afh_recv->post_select, + .context = &pt->rn + }, &sched); + sprintf(buf, "%s decoder", af); + pt->fn.task = task_register( + &(struct task_info) { + .name = buf, + .pre_select = decoder->pre_select, + .post_select = decoder->post_select, + .context = &pt->fn + }, &sched); register_writer_node(&pt->wn, pt->fn.btrn, &sched); return 1; fail: @@ -452,7 +463,8 @@ again: static void kill_stream(struct play_task *pt) { - task_notify(&pt->wn.task, E_EOF); + if (pt->wn.task) + task_notify(pt->wn.task, E_EOF); } #ifdef HAVE_READLINE @@ -1083,9 +1095,8 @@ static void session_update_time_string(struct play_task *pt, char *str, unsigned * terminates. Subsequent calls to i9e_get_error() then return negative and we * are allowed to call i9e_close() and terminate as well. */ -static int session_post_select(__a_unused struct sched *s, struct task *t) +static int session_post_select(__a_unused struct sched *s, struct play_task *pt) { - struct play_task *pt = container_of(t, struct play_task, task); int ret; if (pt->background) @@ -1107,9 +1118,8 @@ static int session_post_select(__a_unused struct sched *s, struct task *t) #else /* HAVE_READLINE */ -static int session_post_select(struct sched *s, struct task *t) +static int session_post_select(struct sched *s, struct play_task *pt) { - struct play_task *pt = container_of(t, struct play_task, task); char c; if (!FD_ISSET(STDIN_FILENO, &s->rfds)) @@ -1132,9 +1142,9 @@ static void session_update_time_string(__a_unused struct play_task *pt, } #endif /* HAVE_READLINE */ -static void play_pre_select(struct sched *s, struct task *t) +static void play_pre_select(struct sched *s, void *context) { - struct play_task *pt = container_of(t, struct play_task, task); + struct play_task *pt = context; char state; para_fd_set(STDIN_FILENO, &s->rfds, &s->max_fileno); @@ -1171,9 +1181,9 @@ static unsigned get_time_string(struct play_task *pt, char **result) ); } -static int play_post_select(struct sched *s, struct task *t) +static int play_post_select(struct sched *s, void *context) { - struct play_task *pt = container_of(t, struct play_task, task); + struct play_task *pt = context; int ret; ret = eof_cleanup(pt); @@ -1181,7 +1191,7 @@ static int play_post_select(struct sched *s, struct task *t) pt->rq = CRT_TERM_RQ; return 0; } - ret = session_post_select(s, t); + ret = session_post_select(s, pt); if (ret < 0) goto out; if (!pt->wn.btrn && !pt->fn.btrn) { @@ -1230,7 +1240,6 @@ int main(int argc, char *argv[]) filter_init(); writer_init(); - clock_get_realtime(now); sched.default_timeout.tv_sec = 5; parse_config_or_die(argc, argv); @@ -1245,11 +1254,14 @@ int main(int argc, char *argv[]) pt->rq = CRT_FILE_CHANGE; pt->current_file = conf.inputs_num - 1; pt->playing = true; - pt->task.pre_select = play_pre_select; - pt->task.post_select = play_post_select; - sprintf(pt->task.status, "play task"); - register_task(&sched, &pt->task); + pt->task = task_register(&(struct task_info){ + .name = "play", + .pre_select = play_pre_select, + .post_select = play_post_select, + .context = pt, + }, &sched); ret = schedule(&sched); + sched_shutdown(&sched); if (ret < 0) PARA_ERROR_LOG("%s\n", para_strerror(-ret)); return ret < 0? EXIT_FAILURE : EXIT_SUCCESS; diff --git a/prebuffer_filter.c b/prebuffer_filter.c index 6fac4cb3..356fb54c 100644 --- a/prebuffer_filter.c +++ b/prebuffer_filter.c @@ -28,9 +28,9 @@ struct private_prebuffer_data { struct timeval barrier; }; -static void prebuffer_pre_select(struct sched *s, struct task *t) +static void prebuffer_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct btr_node *btrn = fn->btrn; size_t iqs = btr_get_input_queue_size(btrn); struct private_prebuffer_data *ppd = fn->private_data; @@ -56,9 +56,9 @@ static void prebuffer_close(struct filter_node *fn) free(fn->private_data); } -static int prebuffer_post_select(__a_unused struct sched *s, struct task *t) +static int prebuffer_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct btr_node *btrn = fn->btrn; size_t iqs = btr_get_input_queue_size(btrn); struct private_prebuffer_data *ppd = fn->private_data; diff --git a/recv.c b/recv.c index babd1e30..0d7bb2ce 100644 --- a/recv.c +++ b/recv.c @@ -65,6 +65,7 @@ int main(int argc, char *argv[]) struct receiver_node rn; struct stdout_task sot; static struct sched s; + struct task_info ti; recv_cmdline_parser(argc, argv, &conf); loglevel = get_loglevel_by_name(conf.loglevel_arg); @@ -89,20 +90,20 @@ int main(int argc, char *argv[]) goto out; r_opened = 1; - memset(&sot, 0, sizeof(struct stdout_task)); sot.btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.parent = rn.btrn, .name = "stdout")); - stdout_set_defaults(&sot); - register_task(&s, &sot.task); + stdout_task_register(&sot, &s); - rn.task.pre_select = r->pre_select; - rn.task.post_select = r->post_select; - sprintf(rn.task.status, "%s", r->name); - register_task(&s, &rn.task); + ti.name = r->name; + ti.pre_select = r->pre_select; + ti.post_select = r->post_select; + ti.context = &rn; + rn.task = task_register(&ti, &s); s.default_timeout.tv_sec = 1; s.default_timeout.tv_usec = 0; ret = schedule(&s); + sched_shutdown(&s); out: if (r_opened) r->close(&rn); diff --git a/recv.h b/recv.h index a590aabd..4564337c 100644 --- a/recv.h +++ b/recv.h @@ -17,7 +17,7 @@ struct receiver_node { /** Pointer to the configuration data for this instance. */ void *conf; /** The task associated with this instance. */ - struct task task; + struct task *task; /** The receiver node is always the root of the buffer tree. */ struct btr_node *btrn; /** Each receiver node maintains a buffer pool for the received data. */ @@ -103,7 +103,7 @@ struct receiver { * * \sa select(2), time.c struct task, struct sched. */ - void (*pre_select)(struct sched *s, struct task *t); + void (*pre_select)(struct sched *s, void *context); /** * Evaluate the result from select(). * @@ -115,7 +115,7 @@ struct receiver { * * \sa select(2), struct receiver. */ - int (*post_select)(struct sched *s, struct task *t); + int (*post_select)(struct sched *s, void *context); /** The two help texts of this receiver. */ struct ggo_help help; @@ -142,7 +142,7 @@ struct receiver { void recv_init(void); void *check_receiver_arg(char *ra, int *receiver_num); void print_receiver_helps(unsigned flags); -int generic_recv_pre_select(struct sched *s, struct task *t); +int generic_recv_pre_select(struct sched *s, struct receiver_node *rn); /** \cond receiver */ extern void http_recv_init(struct receiver *r); diff --git a/recv_common.c b/recv_common.c index 2d6aa74a..280f5bc4 100644 --- a/recv_common.c +++ b/recv_common.c @@ -116,7 +116,7 @@ void print_receiver_helps(unsigned flags) * Simple pre-select hook, used by all receivers. * * \param s Scheduler info. - * \param t Determines the receiver node. + * \param rn The receiver node. * * This requests a minimal delay from the scheduler if the status of the buffer * tree node indicates an error/eof condition. No file descriptors are added to @@ -125,12 +125,10 @@ void print_receiver_helps(unsigned flags) * \return The status of the btr node of the receiver node, i.e. the return * value of the underlying call to \ref btr_node_status(). */ -int generic_recv_pre_select(struct sched *s, struct task *t) +int generic_recv_pre_select(struct sched *s, struct receiver_node *rn) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); int ret = btr_node_status(rn->btrn, 0, BTR_NT_ROOT); - t->error = 0; if (ret < 0) sched_min_delay(s); return ret; diff --git a/resample_filter.c b/resample_filter.c index 2caa1e21..61ffd2b6 100644 --- a/resample_filter.c +++ b/resample_filter.c @@ -65,9 +65,9 @@ static void resample_open(struct filter_node *fn) btr_log_tree(btr_parent(btr_parent(btrn)), LL_INFO); } -static void resample_pre_select(struct sched *s, struct task *t) +static void resample_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct resample_context *ctx = fn->private_data; int ret = btr_node_status(fn->btrn, fn->min_iqs, BTR_NT_INTERNAL); @@ -202,10 +202,10 @@ static int resample_frames(int16_t *in, size_t num_frames, bool have_more, return data.input_frames_used; } -static int resample_post_select(__a_unused struct sched *s, struct task *t) +static int resample_post_select(__a_unused struct sched *s, void *context) { int ret; - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct resample_context *ctx = fn->private_data; struct resample_filter_args_info *conf = fn->conf; struct btr_node *btrn = fn->btrn; diff --git a/sched.c b/sched.c index 8b68667b..44062a66 100644 --- a/sched.c +++ b/sched.c @@ -18,27 +18,38 @@ #include "time.h" #include "error.h" -static struct timeval now_struct; -struct timeval *now = &now_struct; - -/* - * Remove a task from the scheduler. - * - * \param t The task to remove. - * - * If the pre_select pointer of \a t is not \p NULL, it is removed from - * the pre_select list of the scheduler. Same goes for \a post_select. +/** + * The possible states of a task. + * + * In addition to the states listed here, a task may also enter zombie state. + * This happens when its ->post_select function returns negative, the ->status + * field is then set to this return value. Such tasks are not scheduled any + * more (i.e. ->pre_select() and ->post_select() are no longer called), but + * they stay on the scheduler task list until \ref task_reap() or + * \ref sched_shutdown() is called. */ -static void unregister_task(struct task *t) -{ - assert(t->error < 0); - PARA_INFO_LOG("unregistering %s (%s)\n", t->status, - para_strerror(-t->error)); - if (t->pre_select) - list_del(&t->pre_select_node); - if (t->post_select) - list_del(&t->post_select_node); -} +enum task_status { + /** Task has been reaped and may be removed from the task list. */ + TS_DEAD, + /** Task is active. */ + TS_RUNNING, +}; + +struct task { + /** A copy of the task name supplied when the task was registered. */ + char *name; + /** Copied during task_register(). */ + struct task_info info; + /* TS_RUNNING, TS_DEAD, or zombie (negative value). */ + int status; + /** Position of the task in the task list of the scheduler. */ + struct list_head node; + /** If less than zero, the task was notified by another task. */ + int notification; +}; + +static struct timeval now_struct; +const struct timeval *now = &now_struct; static inline bool timeout_is_zero(struct sched *s) { @@ -50,51 +61,67 @@ static void sched_preselect(struct sched *s) { struct task *t, *tmp; - list_for_each_entry_safe(t, tmp, &s->pre_select_list, pre_select_node) { + list_for_each_entry_safe(t, tmp, &s->task_list, node) { + if (t->status < 0) + continue; if (t->notification != 0) sched_min_delay(s); - if (t->pre_select) - t->pre_select(s, t); + if (t->info.pre_select) + t->info.pre_select(s, t->info.context); } } +static void unlink_and_free_task(struct task *t) +{ + PARA_INFO_LOG("freeing task %s\n", t->name); + list_del(&t->node); + free(t->name); + free(t); +} + //#define SCHED_DEBUG 1 static inline void call_post_select(struct sched *s, struct task *t) { + int ret; + #ifndef SCHED_DEBUG - t->error = t->post_select(s, t); + ret = t->info.post_select(s, t->info.context); #else struct timeval t1, t2, diff; unsigned long pst; clock_get_realtime(&t1); - t->error = t->post_select(s, t); + ret = t->info.post_select(s, t->info.context); clock_get_realtime(&t2); tv_diff(&t1, &t2, &diff); pst = tv2ms(&diff); if (pst > 50) PARA_WARNING_LOG("%s: post_select time: %lums\n", - t->status, pst); + t->name, pst); #endif + t->status = ret < 0? ret : TS_RUNNING; } -static void sched_post_select(struct sched *s) +static unsigned sched_post_select(struct sched *s) { struct task *t, *tmp; + unsigned num_running_tasks = 0; - list_for_each_entry_safe(t, tmp, &s->post_select_list, post_select_node) { - if (t->error >= 0) - call_post_select(s, t); -// PARA_INFO_LOG("%s: %d\n", t->status, t->ret); - t->notification = 0; - if (t->error >= 0) - continue; - unregister_task(t); + list_for_each_entry_safe(t, tmp, &s->task_list, node) { + if (t->status == TS_DEAD) /* task has been reaped */ + unlink_and_free_task(t); + else if (t->status == TS_RUNNING) { + call_post_select(s, t); /* sets t->status */ + t->notification = 0; + if (t->status == TS_RUNNING) + num_running_tasks++; + } } + return num_running_tasks; } /** - * The core function for all paraslash programs. + * The core function of all paraslash programs. * * \param s Pointer to the scheduler struct. * @@ -103,14 +130,15 @@ static void sched_post_select(struct sched *s) * the fd sets of \a s. Next, it calls para_select() and makes the result available * to the registered tasks by calling their post_select hook. * - * \return Zero if no more tasks are left in either of the two lists, negative - * if para_select returned an error. + * \return Zero if no more tasks are left in the task list, negative if the + * select function returned an error. * - * \sa task, now. + * \sa \ref now. */ int schedule(struct sched *s) { int ret; + unsigned num_running_tasks; if (!s->select_function) s->select_function = para_select; @@ -119,7 +147,7 @@ again: FD_ZERO(&s->wfds); s->select_timeout = s->default_timeout; s->max_fileno = -1; - clock_get_realtime(now); + clock_get_realtime(&now_struct); sched_preselect(s); ret = s->select_function(s->max_fileno + 1, &s->rfds, &s->wfds, &s->select_timeout); @@ -135,42 +163,105 @@ again: FD_ZERO(&s->rfds); FD_ZERO(&s->wfds); } - clock_get_realtime(now); - sched_post_select(s); - if (list_empty(&s->pre_select_list) && list_empty(&s->post_select_list)) + clock_get_realtime(&now_struct); + num_running_tasks = sched_post_select(s); + if (num_running_tasks == 0) return 0; goto again; } /** - * Add a task to the scheduler. + * Obtain the error status of a task and deallocate its resources. * - * \param t The task to add. - * \param s The scheduler instance to add the task to. + * \param tptr Identifies the task to reap. * - * If the pre_select pointer of \a t is not \p NULL, it is added to - * the pre_select list of the scheduler. Same goes for post_select. + * This function is similar to wait(2) in that it returns information about a + * terminated task and allows to release the resources associated with the + * task. Until this function is called, the terminated task remains in a zombie + * state. * - * \sa task::pre_select, task::post_select + * \return If \a tptr is \p NULL, or \a *tptr is \p NULL, the function does + * nothing and returns zero. Otherwise, it is checked whether the task + * identified by \a tptr is still running. If it is, the function returns zero + * and again, no action is taken. Otherwise the (negative) error code of the + * terminated task is returned and \a *tptr is set to \p NULL. The task will + * then be removed removed from the scheduler task list. + * + * \sa \ref sched_shutdown(), wait(2). */ -void register_task(struct sched *s, struct task *t) +int task_reap(struct task **tptr) { - PARA_INFO_LOG("registering %s (%p)\n", t->status, t); - t->notification = 0; - if (!s->pre_select_list.next) - INIT_LIST_HEAD(&s->pre_select_list); - if (!s->post_select_list.next) - INIT_LIST_HEAD(&s->post_select_list); - if (t->pre_select) { - PARA_DEBUG_LOG("pre_select: %p\n", &t->pre_select); - list_add_tail(&t->pre_select_node, &s->pre_select_list); - } - if (t->post_select) { - PARA_DEBUG_LOG("post_select: %p\n", &t->post_select); - list_add_tail(&t->post_select_node, &s->post_select_list); + struct task *t; + int ret; + + if (!tptr) + return 0; + t = *tptr; + if (!t) + return 0; + if (t->status >= 0) + return 0; + ret = t->status; + /* + * With list_for_each_entry_safe() it is only safe to remove the + * _current_ list item. Since we are being called from the loop in + * schedule() via some task's ->post_select() function, freeing the + * given task here would result in use-after-free bugs in schedule(). + * So we only set the task status to TS_DEAD which tells schedule() to + * free the task in the next iteration of its loop. + */ + t->status = TS_DEAD; + + *tptr = NULL; + return ret; +} + +/** + * Deallocate all resources of all tasks of a scheduler instance. + * + * \param s The scheduler instance. + * + * This should only be called after \ref schedule() has returned. + */ +void sched_shutdown(struct sched *s) +{ + struct task *t, *tmp; + + list_for_each_entry_safe(t, tmp, &s->task_list, node) { + if (t->status == TS_RUNNING) + /* The task list should contain only terminated tasks. */ + PARA_WARNING_LOG("shutting down running task %s\n", + t->name); + unlink_and_free_task(t); } } +/** + * Add a task to the scheduler task list. + * + * \param info Task information supplied by the caller. + * \param s The scheduler instance. + * + * \return A pointer to a newly allocated task structure. It will be + * freed by sched_shutdown(). + */ +struct task *task_register(struct task_info *info, struct sched *s) +{ + struct task *t = para_malloc(sizeof(*t)); + + assert(info->post_select); + + if (!s->task_list.next) + INIT_LIST_HEAD(&s->task_list); + + t->info = *info; + t->name = para_strdup(info->name); + t->notification = 0; + t->status = TS_RUNNING; + list_add_tail(&t->node, &s->task_list); + return t; +} + /** * Get the list of all registered tasks. * @@ -186,21 +277,15 @@ char *get_task_list(struct sched *s) struct task *t, *tmp; char *msg = NULL; - list_for_each_entry_safe(t, tmp, &s->pre_select_list, pre_select_node) { - char *tmp_msg; - tmp_msg = make_message("%s%p\tpre\t%s\n", msg? msg : "", t, t->status); - free(msg); - msg = tmp_msg; - } - list_for_each_entry_safe(t, tmp, &s->post_select_list, post_select_node) { + list_for_each_entry_safe(t, tmp, &s->task_list, node) { char *tmp_msg; -// if (t->pre_select) -// continue; - tmp_msg = make_message("%s%p\tpost\t%s\n", msg? msg : "", t, t->status); + tmp_msg = make_message("%s%p\t%s\t%s\n", msg? msg : "", t, + t->status == TS_DEAD? "dead" : + (t->status == TS_RUNNING? "running" : "zombie"), + t->name); free(msg); msg = tmp_msg; } - //PARA_DEBUG_LOG("task list:\n%s", msg); return msg; } @@ -228,7 +313,7 @@ void task_notify(struct task *t, int err) assert(err > 0); if (t->notification == -err) /* ignore subsequent notifications */ return; - PARA_INFO_LOG("notifying task %s: %s\n", t->status, para_strerror(err)); + PARA_INFO_LOG("notifying task %s: %s\n", t->name, para_strerror(err)); t->notification = -err; } @@ -243,11 +328,30 @@ void task_notify(struct task *t, int err) * * \sa \ref task_notify(). */ -int task_get_notification(struct task *t) +int task_get_notification(const struct task *t) { return t->notification; } +/** + * Return the status value of a task. + * + * \param t The task to get the status value from. + * + * \return Zero if task does not exist, one if task is running, negative error + * code if task has terminated. + */ +int task_status(const struct task *t) +{ + if (!t) + return 0; + if (t->status == TS_DEAD) /* pretend dead tasks don't exist */ + return 0; + if (t->status == TS_RUNNING) + return 1; + return t->status; +} + /** * Set the notification value of all tasks of a scheduler instance. * @@ -261,9 +365,7 @@ void task_notify_all(struct sched *s, int err) { struct task *t; - list_for_each_entry(t, &s->pre_select_list, pre_select_node) - task_notify(t, err); - list_for_each_entry(t, &s->post_select_list, post_select_node) + list_for_each_entry(t, &s->task_list, node) task_notify(t, err); } diff --git a/sched.h b/sched.h index f15e4efb..07877c46 100644 --- a/sched.h +++ b/sched.h @@ -10,11 +10,11 @@ /** * Paraslash's scheduler. * - * Designed with KISS in mind. It manages two lists of tasks. The pre_select - * list contains pointers to functions that are called before calling select() - * from the main loop. Similarly, \a post_select_list is a list of function - * pointers each of which is called after the select call. Tasks add hooks to - * these lists by registering themselves to the scheduler. + * Designed with KISS in mind. It maintains a list of task structures which is + * extended when a new task is registered. Each task may define a pre_select + * function which is called from the scheduler main loop before it calls + * select(). Similarly, each task must define a post_select function which is + * called after the select call. */ struct sched { /** Initial value before any pre_select call. */ @@ -29,46 +29,37 @@ struct sched { int max_fileno; /** If non-NULL, use this function instead of para_select. */ int (*select_function)(int, fd_set *, fd_set *, struct timeval *); - /** Currently active pre_select functions. */ - struct list_head pre_select_list; - /** Currently active post_select functions. */ - struct list_head post_select_list; + /** Tasks which have been registered to the scheduler. */ + struct list_head task_list; }; -/** - * Paraslash's task structure. - * - * Before registering a task to the scheduler, the task structure must be - * filled in properly by the caller. - * - * \sa \ref sched. - */ -struct task { +struct task; + +/** Information that must be supplied by callers of \ref task_register(). */ +struct task_info { + /** Used for log messages and by \ref get_task_list(). */ + const char *name; /** - * The pre select hook of \a t. + * The optional pre select method. * * Its purpose is to add file descriptors to the fd sets of the * scheduler and to decrease the select timeout if necessary. */ - void (*pre_select)(struct sched *s, struct task *t); + void (*pre_select)(struct sched *s, void *context); /** - * The post select hook of \a t. + * The mandatory post select method. * * Its purpose is to evaluate and act upon the results of the previous * select call. If this function returns a negative value, the * scheduler unregisters the task. */ - int (*post_select)(struct sched *s, struct task *t); - /** Whether this task is in error state. */ - int error; - /** Position of the task in the pre_select list of the scheduler. */ - struct list_head pre_select_node; - /** Position of the task in the post_select list of the scheduler. */ - struct list_head post_select_node; - /** Descriptive text and current status of the task. */ - char status[255]; - /** If less than zero, the task was notified by another task. */ - int notification; + int (*post_select)(struct sched *s, void *context); + /** + * This pointer is saved when the task is registered. It is passed to + * ->pre_select() and ->post_select(). Usually this is a pointer to the + * struct owned by the caller which contains the task pointer. + */ + void *context; }; /** @@ -77,14 +68,17 @@ struct task { * scheduler are allowed to block, this value should be accurate enough so that * there is no need to call clock_gettime() directly. */ -extern struct timeval *now; +extern const struct timeval *now; -void register_task(struct sched *s, struct task *t); +struct task *task_register(struct task_info *info, struct sched *s); int schedule(struct sched *s); +void sched_shutdown(struct sched *s); char *get_task_list(struct sched *s); void task_notify(struct task *t, int err); void task_notify_all(struct sched *s, int err); -int task_get_notification(struct task *t); +int task_get_notification(const struct task *t); +int task_status(const struct task *t); +int task_reap(struct task **tptr); void sched_min_delay(struct sched *s); void sched_request_timeout(struct timeval *to, struct sched *s); void sched_request_timeout_ms(long unsigned ms, struct sched *s); diff --git a/server.c b/server.c index fa19ce42..908e039b 100644 --- a/server.c +++ b/server.c @@ -110,7 +110,7 @@ struct server_command_task { /** Argument vector passed to para_server's main function. */ char **argv; /** The command task structure for scheduling. */ - struct task task; + struct task *task; }; static int want_colors(void) @@ -235,9 +235,9 @@ out: exit(EXIT_FAILURE); } -static void signal_pre_select(struct sched *s, struct task *t) +static void signal_pre_select(struct sched *s, void *context) { - struct signal_task *st = container_of(t, struct signal_task, task); + struct signal_task *st = context; para_fd_set(st->fd, &s->rfds, &s->max_fileno); } @@ -253,7 +253,7 @@ static void handle_sighup(void) kill(mmd->afs_pid, SIGHUP); } -static int signal_post_select(struct sched *s, __a_unused struct task *t) +static int signal_post_select(struct sched *s, __a_unused void *context) { int signum = para_next_signal(&s->rfds); @@ -312,10 +312,6 @@ static void init_signal_task(void) static struct signal_task signal_task_struct, *st = &signal_task_struct; - st->task.pre_select = signal_pre_select; - st->task.post_select = signal_post_select; - sprintf(st->task.status, "signal task"); - PARA_NOTICE_LOG("setting up signal handling\n"); st->fd = para_signal_init(); /* always successful */ para_install_sighandler(SIGINT); @@ -324,18 +320,24 @@ static void init_signal_task(void) para_install_sighandler(SIGCHLD); para_sigaction(SIGPIPE, SIG_IGN); add_close_on_fork_list(st->fd); - register_task(&sched, &st->task); + st->task = task_register(&(struct task_info) { + .name = "signal", + .pre_select = signal_pre_select, + .post_select = signal_post_select, + .context = st, + + }, &sched); } -static void command_pre_select(struct sched *s, struct task *t) +static void command_pre_select(struct sched *s, void *context) { - struct server_command_task *sct = container_of(t, struct server_command_task, task); + struct server_command_task *sct = context; para_fd_set(sct->listen_fd, &s->rfds, &s->max_fileno); } -static int command_post_select(struct sched *s, struct task *t) +static int command_post_select(struct sched *s, void *context) { - struct server_command_task *sct = container_of(t, struct server_command_task, task); + struct server_command_task *sct = context; int new_fd, ret, i; char *peer_name; @@ -399,8 +401,6 @@ static void init_server_command_task(int argc, char **argv) *sct = &server_command_task_struct; PARA_NOTICE_LOG("initializing tcp command socket\n"); - sct->task.pre_select = command_pre_select; - sct->task.post_select = command_post_select; sct->argc = argc; sct->argv = argv; ret = para_listen_simple(IPPROTO_TCP, conf.port_arg); @@ -411,8 +411,12 @@ static void init_server_command_task(int argc, char **argv) if (ret < 0) goto err; add_close_on_fork_list(sct->listen_fd); /* child doesn't need the listener */ - sprintf(sct->task.status, "server command task"); - register_task(&sched, &sct->task); + sct->task = task_register(&(struct task_info) { + .name = "server command", + .pre_select = command_pre_select, + .post_select = command_post_select, + .context = sct, + }, &sched); return; err: PARA_EMERG_LOG("%s\n", para_strerror(-ret)); @@ -484,9 +488,7 @@ static void server_init(int argc, char **argv) parse_config_or_die(0); log_welcome("para_server"); init_ipc_or_die(); /* init mmd struct and mmd->lock */ - /* make sure, the global now pointer is uptodate */ - clock_get_realtime(now); - set_server_start_time(now); + set_server_start_time(NULL); init_user_list(user_list_file); /* become daemon */ if (conf.daemon_given) @@ -573,6 +575,7 @@ int main(int argc, char *argv[]) server_init(argc, argv); mutex_lock(mmd_mutex); ret = schedule(&sched); + sched_shutdown(&sched); if (ret < 0) { PARA_EMERG_LOG("%s\n", para_strerror(-ret)); exit(EXIT_FAILURE); diff --git a/signal.h b/signal.h index 526f1502..a15349ca 100644 --- a/signal.h +++ b/signal.h @@ -13,7 +13,7 @@ struct signal_task { /** The signal pipe. */ int fd; /** The associated task structure. */ - struct task task; + struct task *task; }; int para_signal_init(void); diff --git a/spxdec_filter.c b/spxdec_filter.c index 72ecd62a..8b29007c 100644 --- a/spxdec_filter.c +++ b/spxdec_filter.c @@ -238,9 +238,9 @@ static int compute_skip_samples(ogg_page *og, struct private_spxdec_data *psd) return ret; } -static int speexdec_post_select(__a_unused struct sched *s, struct task *t) +static int speexdec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct private_spxdec_data *psd = fn->private_data; struct btr_node *btrn = fn->btrn; int ret, ns; diff --git a/stdin.c b/stdin.c index d5425b28..a66ad469 100644 --- a/stdin.c +++ b/stdin.c @@ -18,19 +18,13 @@ #include "buffer_tree.h" #include "string.h" -/** - * The pre_select function of the stdin task. - * - * \param s The scheduler this task was registered to. - * \param t The task structure of the stdin task. - * - * This function is always successful. If there is space left in the - * buffer of the stdin task, it adds \p STDIN_FILENO to the read fd set - * of \a s. +/* + * If there is space left in the buffer of the stdin task add STDIN_FILENO to + * the read fd set of s. */ -static void stdin_pre_select(struct sched *s, struct task *t) +static void stdin_pre_select(struct sched *s, void *context) { - struct stdin_task *sit = container_of(t, struct stdin_task, task); + struct stdin_task *sit = context; int ret; ret = btr_node_status(sit->btrn, 0, BTR_NT_ROOT); @@ -43,20 +37,14 @@ static void stdin_pre_select(struct sched *s, struct task *t) sched_request_timeout_ms(100, s); } -/** - * The post select function of the stdin task. - * - * \param s The scheduler this task was registered to. - * \param t The task structure of the stdin task. - * - * This function checks if \p STDIN_FILENO was included by in the read fd set - * of \a s during the previous pre_select call. If yes, and \p STDIN_FILENO - * appears to be readable, data is read from stdin and fed into the buffer - * tree. +/* + * This function checks if STDIN_FILENO was included by in the read fd set of s + * during the previous pre_select call. If so, and if STDIN_FILENO is readable, + * data is read from stdin and fed into the buffer tree. */ -static int stdin_post_select(struct sched *s, struct task *t) +static int stdin_post_select(struct sched *s, void *context) { - struct stdin_task *sit = container_of(t, struct stdin_task, task); + struct stdin_task *sit = context; ssize_t ret; size_t sz, n; char *buf = NULL; @@ -95,21 +83,25 @@ err: } /** - * Initialize a stdin task structure with default values. + * Register a stdin task structure. * - * \param sit The stdin task structure. + * \param sit The stdin task structure to register. + * \param s The task will be added to this scheduler's task list. * - * This fills in the pre/post select function pointers of the task structure - * given by \a sit and creates a buffer tree for I/O. + * This allocates a buffer tree pool for I/O, sets up \a sit and registers a + * task with \a sit as context pointer. */ -void stdin_set_defaults(struct stdin_task *sit) +void stdin_task_register(struct stdin_task *sit, struct sched *s) { int ret; + struct task_info ti = { + .name = "stdin", + .pre_select = stdin_pre_select, + .post_select = stdin_post_select, + .context = sit, + }; - sit->task.pre_select = stdin_pre_select; - sit->task.post_select = stdin_post_select; sit->btrp = btr_pool_new("stdin", 128 * 1024); - sprintf(sit->task.status, "stdin reader"); /* * Both STDIN_FILENO and STDOUT_FILENO may refer to the same open file * description (the terminal), and thus share the same file status @@ -124,4 +116,5 @@ void stdin_set_defaults(struct stdin_task *sit) } sit->fd_flags = ret; sit->must_set_nonblock_flag = (sit->fd_flags & O_NONBLOCK) == 0; + sit->task = task_register(&ti, s); } diff --git a/stdin.h b/stdin.h index 22ea7ed6..b40cf7eb 100644 --- a/stdin.h +++ b/stdin.h @@ -9,7 +9,7 @@ /** The task structure used for reading from stdin. */ struct stdin_task { /** The task structure. */ - struct task task; + struct task *task; /** Stdin is always the root of a buffer tree. */ struct btr_node *btrn; /** Use a buffer pool to minimize memcpy due to alignment problems. */ @@ -20,4 +20,4 @@ struct stdin_task { bool must_set_nonblock_flag; }; -void stdin_set_defaults(struct stdin_task *sit); +void stdin_task_register(struct stdin_task *sit, struct sched *s); diff --git a/stdout.c b/stdout.c index 4554145f..bb921ad1 100644 --- a/stdout.c +++ b/stdout.c @@ -16,18 +16,10 @@ #include "stdout.h" #include "buffer_tree.h" -/** - * The pre_select function of the stdout task. - * - * \param s The scheduler this task was registered to. - * \param t The task structure of the stdout task. - * - * This function is always successful. If there is input data available, it - * adds \p STDOUT_FILENO to the write fd set of \a s. - */ -static void stdout_pre_select(struct sched *s, struct task *t) +/* Add STDOUT_FILENO to the write fd set if there is input data available. */ +static void stdout_pre_select(struct sched *s, void *context) { - struct stdout_task *sot = container_of(t, struct stdout_task, task); + struct stdout_task *sot = context; int ret; ret = btr_node_status(sot->btrn, 0, BTR_NT_LEAF); @@ -37,18 +29,13 @@ static void stdout_pre_select(struct sched *s, struct task *t) sched_min_delay(s); } -/** - * The post select function of the stdout task. - * - * \param s The scheduler this task was registered to. - * \param t The task structure of the stdout task. - * - * This function writes input data from the buffer tree to stdout if \p +/* + * This function writes input data from the buffer tree to stdout if * STDOUT_FILENO is writable. */ -static int stdout_post_select(struct sched *s, struct task *t) +static int stdout_post_select(struct sched *s, void *context) { - struct stdout_task *sot = container_of(t, struct stdout_task, task); + struct stdout_task *sot = context; struct btr_node *btrn = sot->btrn; int ret; char *buf; @@ -85,21 +72,24 @@ out: } return ret; } + /** - * Initialize a stdout task structure with default values. + * Register a stdout task structure. * - * \param sot The stdout task structure. + * \param sot The stdout task structure to register. + * \param s The task will be added to this scheduler's task list. * - * This fills in the pre/post select function pointers of the task structure - * given by \a sot. + * This sets up \a sot and registers a task with \a sot as context pointer. */ -void stdout_set_defaults(struct stdout_task *sot) +void stdout_task_register(struct stdout_task *sot, struct sched *s) { int ret; - - sot->task.pre_select = stdout_pre_select; - sot->task.post_select = stdout_post_select; - sprintf(sot->task.status, "stdout"); + struct task_info ti = { + .pre_select = stdout_pre_select, + .post_select = stdout_post_select, + .context = sot, + .name = "stdout", + }; /* See stdin.c for details. */ ret = fcntl(STDOUT_FILENO, F_GETFL); @@ -109,4 +99,5 @@ void stdout_set_defaults(struct stdout_task *sot) } sot->fd_flags = ret; sot->must_set_nonblock_flag = (sot->fd_flags & O_NONBLOCK) == 0; + sot->task = task_register(&ti, s); } diff --git a/stdout.h b/stdout.h index dbf8866b..531fc7c2 100644 --- a/stdout.h +++ b/stdout.h @@ -13,7 +13,7 @@ */ struct stdout_task { /** The task structure used by the scheduler. */ - struct task task; + struct task *task; /** Stdout is always a leaf node in the buffer tree. */ struct btr_node *btrn; /** The descriptor flags of STDOUT at startup. */ @@ -22,4 +22,4 @@ struct stdout_task { bool must_set_nonblock_flag; }; -void stdout_set_defaults(struct stdout_task *sot); +void stdout_task_register(struct stdout_task *sot, struct sched *s); diff --git a/sync_filter.c b/sync_filter.c index 379b54b1..fceb1d1c 100644 --- a/sync_filter.c +++ b/sync_filter.c @@ -258,10 +258,10 @@ static void sync_set_timeout(struct sync_filter_context *ctx, tv_add(now, &to, &ctx->timeout); } -static void sync_pre_select(struct sched *s, struct task *t) +static void sync_pre_select(struct sched *s, void *context) { int ret; - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct sync_filter_context *ctx = fn->private_data; struct sync_filter_config *sfc = fn->conf; @@ -295,10 +295,10 @@ static struct sync_buddy *sync_find_buddy(struct sockaddr *addr, return NULL; } -static int sync_post_select(__a_unused struct sched *s, struct task *t) +static int sync_post_select(__a_unused struct sched *s, void *context) { int ret; - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct sync_filter_context *ctx = fn->private_data; struct sync_filter_config *sfc = fn->conf; struct sync_buddy *buddy, *tmp; diff --git a/udp_recv.c b/udp_recv.c index 15cf73eb..bde41243 100644 --- a/udp_recv.c +++ b/udp_recv.c @@ -27,11 +27,11 @@ #include "net.h" #include "fd.h" -static void udp_recv_pre_select(struct sched *s, struct task *t) +static void udp_recv_pre_select(struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; - if (generic_recv_pre_select(s, t) <= 0) + if (generic_recv_pre_select(s, rn) <= 0) return; para_fd_set(rn->fd, &s->rfds, &s->max_fileno); } @@ -54,15 +54,15 @@ static int udp_check_eof(size_t sz, struct iovec iov[2]) return -E_RECV_EOF; } -static int udp_recv_post_select(__a_unused struct sched *s, struct task *t) +static int udp_recv_post_select(__a_unused struct sched *s, void *context) { - struct receiver_node *rn = container_of(t, struct receiver_node, task); + struct receiver_node *rn = context; struct btr_node *btrn = rn->btrn; size_t num_bytes; struct iovec iov[2]; int ret, readv_ret, iovcnt; - ret = task_get_notification(t); + ret = task_get_notification(rn->task); if (ret < 0) goto out; ret = btr_node_status(btrn, 0, BTR_NT_ROOT); diff --git a/vss.c b/vss.c index 3ace49e9..928ef6c2 100644 --- a/vss.c +++ b/vss.c @@ -89,7 +89,7 @@ struct vss_task { /** The memory mapped audio file. */ char *map; /** Used by the scheduler. */ - struct task task; + struct task *task; /** Pointer to the header of the mapped audio file. */ char *header_buf; /** Length of the audio file header. */ @@ -885,12 +885,9 @@ static void set_mmd_offset(void) mmd->offset = tv2ms(&offset); } -/** +/* * Compute the timeout for the main select-loop of the scheduler. * - * \param s Pointer to the server scheduler. - * \param t Pointer to the vss task structure. - * * Before the timeout is computed, the current vss status flags are evaluated * and acted upon by calling appropriate functions from the lower layers. * Possible actions include @@ -899,10 +896,10 @@ static void set_mmd_offset(void) * - shutdown of all senders (stop/pause command), * - reposition the stream (ff/jmp command). */ -static void vss_pre_select(struct sched *s, struct task *t) +static void vss_pre_select(struct sched *s, void *context) { int i; - struct vss_task *vsst = container_of(t, struct vss_task, task); + struct vss_task *vsst = context; if (!vsst->map || vss_next() || vss_paused() || vss_repos()) { struct fec_client *fc, *tmp; @@ -1126,10 +1123,10 @@ static void vss_send(struct vss_task *vsst) } } -static int vss_post_select(struct sched *s, struct task *t) +static int vss_post_select(struct sched *s, void *context) { int ret, i; - struct vss_task *vsst = container_of(t, struct vss_task, task); + struct vss_task *vsst = context; if (mmd->sender_cmd_data.cmd_num >= 0) { int num = mmd->sender_cmd_data.cmd_num, @@ -1185,8 +1182,6 @@ void init_vss_task(int afs_socket, struct sched *s) conf.autoplay_delay_arg : 0; vsst->header_interval.tv_sec = 5; /* should this be configurable? */ vsst->afs_socket = afs_socket; - vsst->task.pre_select = vss_pre_select; - vsst->task.post_select = vss_post_select; ms2tv(announce_time, &vsst->announce_tv); PARA_INFO_LOG("announce timeval: %lums\n", tv2ms(&vsst->announce_tv)); INIT_LIST_HEAD(&fec_client_list); @@ -1206,6 +1201,10 @@ void init_vss_task(int afs_socket, struct sched *s) tv_add(&vsst->autoplay_barrier, &vsst->announce_tv, &vsst->data_send_barrier); } - sprintf(vsst->task.status, "vss task"); - register_task(s, &vsst->task); + vsst->task = task_register(&(struct task_info) { + .name = "vss task", + .pre_select = vss_pre_select, + .post_select = vss_post_select, + .context = vsst, + }, s); } diff --git a/wav_filter.c b/wav_filter.c index ea3236c7..199db45a 100644 --- a/wav_filter.c +++ b/wav_filter.c @@ -63,20 +63,19 @@ static void wav_open(struct filter_node *fn) *bof = 1; } -static void wav_pre_select(struct sched *s, struct task *t) +static void wav_pre_select(struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; size_t iqs = btr_get_input_queue_size(fn->btrn); - t->error = 0; if (iqs == 0) return; sched_min_delay(s); } -static int wav_post_select(__a_unused struct sched *s, struct task *t) +static int wav_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; struct btr_node *btrn = fn->btrn; size_t iqs = btr_get_input_queue_size(btrn); int ret; diff --git a/wmadec_filter.c b/wmadec_filter.c index fdca2814..188b0a41 100644 --- a/wmadec_filter.c +++ b/wmadec_filter.c @@ -1203,9 +1203,9 @@ static int wmadec_execute(struct btr_node *btrn, const char *cmd, char **result) #define WMA_OUTPUT_BUFFER_SIZE (128 * 1024) -static int wmadec_post_select(__a_unused struct sched *s, struct task *t) +static int wmadec_post_select(__a_unused struct sched *s, void *context) { - struct filter_node *fn = container_of(t, struct filter_node, task); + struct filter_node *fn = context; int ret, converted, out_size; struct private_wmadec_data *pwd = fn->private_data; struct btr_node *btrn = fn->btrn; diff --git a/write.c b/write.c index 4ed5f72e..eaf56f41 100644 --- a/write.c +++ b/write.c @@ -71,19 +71,19 @@ static void setup_writer_node(const char *arg, struct btr_node *parent, } struct write_task { - struct task task; + struct task *task; struct check_wav_context *cwc; }; -static void write_pre_select(struct sched *s, struct task *t) +static void write_pre_select(struct sched *s, void *context) { - struct write_task *wt = container_of(t, struct write_task, task); + struct write_task *wt = context; check_wav_pre_select(s, wt->cwc); } -static int write_post_select(__a_unused struct sched *s, struct task *t) +static int write_post_select(__a_unused struct sched *s, void *context) { - struct write_task *wt = container_of(t, struct write_task, task); + struct write_task *wt = context; return check_wav_post_select(wt->cwc); } @@ -94,22 +94,20 @@ static int setup_and_schedule(void) struct writer_node *wns; static struct sched s; struct wav_params wp; - struct write_task wt = { - .task = { - .pre_select = write_pre_select, - .post_select = write_post_select, - .status = "write task", - }, - }; + struct write_task wt; sit.btrn = btr_new_node(&(struct btr_node_description) EMBRACE(.name = "stdin")); - stdin_set_defaults(&sit); - register_task(&s, &sit.task); + stdin_task_register(&sit, &s); COPY_WAV_PARMS(&wp, &conf); wt.cwc = check_wav_init(sit.btrn, NULL, &wp, &cw_btrn); - register_task(&s, &wt.task); + wt.task = task_register(&(struct task_info) { + .name = "write", + .pre_select = write_pre_select, + .post_select = write_post_select, + .context = &wt, + }, &s); if (!conf.writer_given) { wns = para_calloc(sizeof(*wns)); setup_writer_node(NULL, cw_btrn, wns, &s); @@ -125,16 +123,17 @@ static int setup_and_schedule(void) s.default_timeout.tv_usec = 50000; ret = schedule(&s); if (ret >= 0) { - int j; + int j, ts; for (j = 0; j < i; j++) { - struct task *t = &wns[j].task; - assert(t->error < 0); - if (t->error != -E_WRITE_COMMON_EOF - && t->error != -E_BTR_EOF) { - PARA_ERROR_LOG("%s: %s\n", t->status, - para_strerror(-t->error)); + struct writer_node *wn = wns + j; + ts = task_status(wn->task); + assert(ts < 0); + if (ts != -E_WRITE_COMMON_EOF && ts != -E_BTR_EOF) { + const char *name = writer_names[wn->writer_num]; + PARA_ERROR_LOG("%s: %s\n", name, + para_strerror(-ts)); if (ret >= 0) - ret = t->error; + ret = ts; } } } @@ -149,6 +148,7 @@ static int setup_and_schedule(void) } free(wns); check_wav_shutdown(wt.cwc); + sched_shutdown(&s); return ret; } diff --git a/write.h b/write.h index 0cfcafda..cb734988 100644 --- a/write.h +++ b/write.h @@ -22,7 +22,7 @@ struct writer_node { /** The buffer tree node associated with this writer node. */ struct btr_node *btrn; /** The task of this writer node. */ - struct task task; + struct task *task; /** The minimal input queue size (size of one audio sample). */ size_t min_iqs; }; @@ -59,13 +59,13 @@ struct writer { * This is called from scheduler. It may use the sched pointer to add * any file descriptors or to decrease the select timeout. */ - void (*pre_select)(struct sched *s, struct task *t); + void (*pre_select)(struct sched *s, void *context); /** * Write audio data. * * Called from the post_select function of the writer node's task. */ - int (*post_select)(struct sched *s, struct task *t); + int (*post_select)(struct sched *s, void *context); /** * Close one instance of the writer. * diff --git a/write_common.c b/write_common.c index e191c49a..4c5ef4bc 100644 --- a/write_common.c +++ b/write_common.c @@ -107,16 +107,16 @@ void register_writer_node(struct writer_node *wn, struct btr_node *parent, struct sched *s) { struct writer *w = writers + wn->writer_num; - char *name = make_message("%s writer", writer_names[wn->writer_num]); wn->btrn = btr_new_node(&(struct btr_node_description) - EMBRACE(.name = name, .parent = parent, + EMBRACE(.name = writer_names[wn->writer_num], .parent = parent, .handler = w->execute, .context = wn)); - strcpy(wn->task.status, name); - free(name); - wn->task.pre_select = w->pre_select; - wn->task.post_select = w->post_select; - register_task(s, &wn->task); + wn->task = task_register(&(struct task_info) { + .name = writer_names[wn->writer_num], + .pre_select = w->pre_select, + .post_select = w->post_select, + .context = wn, + }, s); } /**