From: Andre Noll Date: Sat, 12 Apr 2008 17:01:47 +0000 (+0200) Subject: Make para_server use the generic scheduling code. X-Git-Tag: v0.3.3~79 X-Git-Url: http://git.tue.mpg.de/?a=commitdiff_plain;h=9d9ef4f91346188baaef1656b861151ea23437d9;p=paraslash.git Make para_server use the generic scheduling code. --- diff --git a/NEWS b/NEWS index 82753cbb..a1146d19 100644 --- a/NEWS +++ b/NEWS @@ -5,6 +5,9 @@ NEWS 0.3.3 (to be announced) "axiomatic perspectivity" ------------------------------------------------- + - para_server uses the generic scheduling code. + - overhaul of the virtual streaming system. + ----------------------------------------- 0.3.2 (2008-04-11) "probabilistic parity" ----------------------------------------- diff --git a/afh.c b/afh.c index dd69d9e4..4c5ddae5 100644 --- a/afh.c +++ b/afh.c @@ -60,7 +60,7 @@ static int cat_file(void *audio_file_data, struct afh_info *afhi) int ret; struct timeval stream_start; long unsigned i, first_chunk, last_chunk; - char *buf; + const char *buf; size_t size; diff --git a/afh.h b/afh.h index a441d5fe..276452d8 100644 --- a/afh.h +++ b/afh.h @@ -104,5 +104,5 @@ int compute_afhi(const char *path, char *data, size_t size, struct afh_info *afhi); const char *audio_format_name(int); void afh_get_chunk(long unsigned chunk_num, struct afh_info *afhi, - void *map, char **buf, size_t *len); -void afh_get_header(struct afh_info *afhi, void *map, char **buf, size_t *len); + void *map, const char **buf, size_t *len); +void afh_get_header(struct afh_info *afhi, void *map, const char **buf, size_t *len); diff --git a/afh_common.c b/afh_common.c index feee4c8d..f57ca89f 100644 --- a/afh_common.c +++ b/afh_common.c @@ -182,7 +182,7 @@ const char *audio_format_name(int i) void afh_get_chunk(long unsigned chunk_num, struct afh_info *afhi, - void *map, char **buf, size_t *len) + void *map, const char **buf, size_t *len) { size_t pos = afhi->chunk_table[chunk_num]; *buf = map + pos; @@ -201,7 +201,7 @@ void afh_get_chunk(long unsigned chunk_num, struct afh_info *afhi, * afhi is \p NULL, or if the current audio format does not need special * header treamtment. */ -void afh_get_header(struct afh_info *afhi, void *map, char **buf, size_t *len) +void afh_get_header(struct afh_info *afhi, void *map, const char **buf, size_t *len) { if (!map || !afhi || ! afhi->header_len) { *buf = NULL; diff --git a/dccp_send.c b/dccp_send.c index d66b800f..bbef02e9 100644 --- a/dccp_send.c +++ b/dccp_send.c @@ -63,13 +63,14 @@ static void dccp_post_select(fd_set *rfds, __a_unused fd_set *wfds) } static void dccp_send(long unsigned current_chunk, - __a_unused long unsigned chunks_sent, const char *buf, size_t len) + __a_unused long unsigned chunks_sent, const char *buf, + size_t len, const char *header_buf, size_t header_len) { struct sender_client *sc, *tmp; list_for_each_entry_safe(sc, tmp, &dss->client_list, node) send_chunk(sc, dss, DCCP_MAX_BYTES_PER_WRITE, current_chunk, buf, - len); + len, header_buf, header_len); } static void dccp_shutdown_clients(void) diff --git a/http_send.c b/http_send.c index ccd441a2..8ebbbc9c 100644 --- a/http_send.c +++ b/http_send.c @@ -74,7 +74,8 @@ static void http_shutdown_clients(void) } static void http_send(long unsigned current_chunk, - __a_unused long unsigned chunks_sent, const char *buf, size_t len) + __a_unused long unsigned chunks_sent, const char *buf, size_t len, + const char *header_buf, size_t header_len) { struct sender_client *sc, *tmp; @@ -82,7 +83,8 @@ static void http_send(long unsigned current_chunk, struct private_http_sender_data *phsd = sc->private_data; if (phsd->status != HTTP_STREAMING) continue; - send_chunk(sc, hss, 0, current_chunk, buf, len); + send_chunk(sc, hss, 0, current_chunk, buf, len, header_buf, + header_len); } } diff --git a/ortp_send.c b/ortp_send.c index 9c166aee..17f14d17 100644 --- a/ortp_send.c +++ b/ortp_send.c @@ -159,13 +159,13 @@ static int need_extra_header(long unsigned current_chunk) } static void ortp_send(long unsigned current_chunk, long unsigned chunks_sent, - const char *buf, size_t len) + const char *buf, size_t len, const char *header_buf, + size_t header_len) { struct ortp_target *ot, *tmp; size_t sendbuf_len; - size_t header_len = 0; int packet_type = ORTP_DATA; - char *sendbuf, *header_buf = NULL; + char *sendbuf; struct timeval *chunk_tv; if (sender_status != SENDER_ON) @@ -190,7 +190,6 @@ static void ortp_send(long unsigned current_chunk, long unsigned chunks_sent, } if (list_empty(&targets)) return; - vss_get_header(&header_buf, &header_len); if (!need_extra_header(current_chunk)) header_len = 0; sendbuf_len = ORTP_AUDIO_HEADER_LEN + header_len + len; diff --git a/send.h b/send.h index 6550c701..270affd5 100644 --- a/send.h +++ b/send.h @@ -49,7 +49,8 @@ struct sender { * should be sent, and \a len is the length of this buffer. */ void (*send)(long unsigned current_chunk, long unsigned chunks_sent, - const char *buf, size_t len); + const char *buf, size_t len, const char *header_buf, + size_t header_len); /** * Add file descriptors to fd_sets. * @@ -126,7 +127,8 @@ void shutdown_client(struct sender_client *sc, struct sender_status *ss); void shutdown_clients(struct sender_status *ss); void send_chunk(struct sender_client *sc, struct sender_status *ss, size_t max_bytes_per_write, long unsigned current_chunk, - const char *buf, size_t len); + const char *buf, size_t len, const char *header_buf, + size_t header_len); void init_sender_status(struct sender_status *ss, char **access_arg, int num_access_args, int port, int max_clients, int default_deny); char *get_sender_info(struct sender_status *ss, char *name); diff --git a/send_common.c b/send_common.c index c542e7e6..c0098b98 100644 --- a/send_common.c +++ b/send_common.c @@ -161,21 +161,20 @@ static int send_queued_chunks(struct sender_client *sc, * \param current_chunk The number of the chunk to write. * \param buf The data to write. * \param len The number of bytes of \a buf. + * \param header_buf The audio file header. + * \param header_len The number of bytes of \a header_buf. * * On errors, the client is shut down. If only a part of the buffer could be * written, the remainder is put into the chunk queue for that client. */ void send_chunk(struct sender_client *sc, struct sender_status *ss, size_t max_bytes_per_write, long unsigned current_chunk, - const char *buf, size_t len) + const char *buf, size_t len, const char *header_buf, + size_t header_len) { int ret; if (!sc->header_sent && current_chunk) { - size_t header_len; - char *header_buf; - - vss_get_header(&header_buf, &header_len); if (header_buf && header_len > 0) { ret = queue_chunk_or_shutdown(sc, ss, header_buf, header_len); if (ret < 0) diff --git a/server.c b/server.c index caeeba42..244c1ecd 100644 --- a/server.c +++ b/server.c @@ -10,9 +10,7 @@ /** * \mainpage Paraslash API Reference * - * Starting points for getting an overview are - * - * probably: + * Starting points for getting an overview: * * - The main programs: \ref server.c, \ref audiod.c, \ref client.c, * \ref audioc.c, \ref fsck.c, @@ -64,7 +62,6 @@ */ #include -#include #include #include "para.h" @@ -87,10 +84,10 @@ #include "signal.h" #include "user_list.h" -/** define the array of error lists needed by para_server */ +/** Define the array of error lists needed by para_server. */ INIT_SERVER_ERRLISTS; -/** shut down non-authorized connections after that many seconds */ +/** Shut down non-authorized connections after that many seconds. */ #define ALARM_TIMEOUT 10 /** @@ -100,21 +97,25 @@ INIT_SERVER_ERRLISTS; struct misc_meta_data *mmd; /** - * the configuration of para_server + * The configuration of para_server * * It also contains the options for the audio file selector, audio format * handler and all supported senders. */ struct server_args_info conf; -/** the file containing user information (public key, permissions) */ -char *user_list_file = NULL; +/** A random value used in child context for authentication. */ +uint32_t afs_socket_cookie; /* global variables for server-internal use */ static FILE *logfile; +/** The file containing user information (public key, permissions). */ +static char *user_list_file = NULL; static int mmd_mutex, mmd_shm_id; -static int signal_pipe; +static pid_t afs_pid; + +/** The task resposible for server command handling. */ struct server_command_task { /** TCP port on which para_server listens for connections. */ int listen_fd; @@ -123,15 +124,14 @@ struct server_command_task { /** Argument vector passed to para_server's main function. */ char **argv; /** The command task structure for scheduling. */ - //struct task task; - char dummy; + struct task task; }; /** - * para_server's log function + * Para_server's log function. * - * \param ll the log level - * \param fmt the format string describing the log message + * \param ll The log level. + * \param fmt The format string describing the log message. */ void para_log(int ll, const char* fmt,...) { @@ -195,7 +195,7 @@ err_out: } /** - * lock the shared memory area containing the mmd struct + * Lock the shared memory area containing the mmd struct. * * \sa semop(2), struct misc_meta_data. */ @@ -205,9 +205,9 @@ void mmd_lock(void) } /** - * unlock the shared memory area containing the mmd struct + * Unlock the shared memory area containing the mmd struct. * - * \sa semop(2), struct misc_meta_data + * \sa semop(2), struct misc_meta_data. */ void mmd_unlock(void) @@ -262,9 +262,74 @@ out: exit(EXIT_FAILURE); } -static void setup_signal_handling(void) +static void signal_pre_select(struct sched *s, struct task *t) +{ + struct signal_task *st = container_of(t, struct signal_task, task); + para_fd_set(st->fd, &s->rfds, &s->max_fileno); +} + +/* + * called when server gets SIGHUP or when client invokes hup command. + */ +static void handle_sighup(void) +{ + PARA_NOTICE_LOG("SIGHUP\n"); + close_log(logfile); /* gets reopened if necessary by parse_config */ + logfile = NULL; + parse_config(1); /* reopens log */ + init_user_list(user_list_file); /* reload user list */ + if (afs_pid) + kill(afs_pid, SIGHUP); +} + +static void signal_post_select(struct sched *s, struct task *t) { - signal_pipe = para_signal_init(); /* always successful */ + struct signal_task *st = container_of(t, struct signal_task, task); + + if (!FD_ISSET(st->fd, &s->rfds)) + return; + + st->signum = para_next_signal(); + switch (st->signum) { + case SIGHUP: + handle_sighup(); + break; + case SIGCHLD: + for (;;) { + pid_t pid; + int ret = para_reap_child(&pid); + if (ret <= 0) + break; + if (pid != afs_pid) + continue; + PARA_EMERG_LOG("fatal: afs died\n"); + goto genocide; + } + break; + /* die on sigint/sigterm. Kill all children too. */ + case SIGINT: + case SIGTERM: + PARA_EMERG_LOG("terminating on signal %d\n", st->signum); +genocide: + kill(0, SIGTERM); + mutex_destroy(mmd_mutex); + shm_detach(mmd); + shm_destroy(mmd_shm_id); + + exit(EXIT_FAILURE); + } +} + +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"); + + st->fd = para_signal_init(); /* always successful */ PARA_NOTICE_LOG("setting up signal handlers\n"); if (para_install_sighandler(SIGINT) < 0) @@ -279,18 +344,76 @@ static void setup_signal_handling(void) goto err; if (signal(SIGUSR1, SIG_IGN) == SIG_ERR) goto err; - add_close_on_fork_list(signal_pipe); + add_close_on_fork_list(st->fd); + register_task(&st->task); return; err: PARA_EMERG_LOG("could not install signal handlers\n"); exit(EXIT_FAILURE); } -static void init_command_task(struct server_command_task *sct) +static void command_pre_select(struct sched *s, struct task *t) +{ + struct server_command_task *sct = container_of(t, struct server_command_task, task); + para_fd_set(sct->listen_fd, &s->rfds, &s->max_fileno); +} + +static void command_post_select(struct sched *s, struct task *t) +{ + struct server_command_task *sct = container_of(t, struct server_command_task, task); + + int new_fd, ret, i; + char *peer_name; + pid_t child_pid; + + if (!FD_ISSET(sct->listen_fd, &s->rfds)) + return; + ret = para_accept(sct->listen_fd, NULL, 0); + if (ret < 0) + goto out; + new_fd = ret; + peer_name = remote_name(new_fd); + PARA_INFO_LOG("got connection from %s, forking\n", peer_name); + mmd->num_connects++; + mmd->active_connections++; + random(); + child_pid = fork(); + if (child_pid < 0) { + ret = -ERRNO_TO_PARA_ERROR(errno); + goto out; + } + if (child_pid) { + close(new_fd); + /* parent keeps accepting connections */ + return; + } + alarm(ALARM_TIMEOUT); + close_listed_fds(); + para_signal_shutdown(); + /* + * put info on who we are serving into argv[0] to make + * client ip visible in top/ps + */ + for (i = sct->argc - 1; i >= 0; i--) + memset(sct->argv[i], 0, strlen(sct->argv[i])); + sprintf(sct->argv[0], "para_server (serving %s)", peer_name); + return handle_connect(new_fd, peer_name); +out: + if (ret < 0) + PARA_CRIT_LOG("%s\n", para_strerror(-ret)); +} + +static void init_server_command_task(int argc, char **argv) { int ret; + static struct server_command_task server_command_task_struct, + *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(AF_UNSPEC, IPPROTO_TCP, conf.port_arg); if (ret < 0) goto err; @@ -299,6 +422,7 @@ static void init_command_task(struct server_command_task *sct) if (ret < 0) goto err; add_close_on_fork_list(sct->listen_fd); /* child doesn't need the listener */ + register_task(&sct->task); return; err: PARA_EMERG_LOG("%s\n", para_strerror(-ret)); @@ -334,11 +458,7 @@ err: exit(EXIT_FAILURE); } -uint32_t afs_socket_cookie; -int afs_socket; -static pid_t afs_pid; - -static void init_afs(void) +static int init_afs(void) { int ret, afs_server_socket[2]; @@ -354,18 +474,17 @@ static void init_afs(void) afs_init(afs_socket_cookie, afs_server_socket[1]); } close(afs_server_socket[1]); - afs_socket = afs_server_socket[0]; - ret = mark_fd_nonblocking(afs_socket); + ret = mark_fd_nonblocking(afs_server_socket[0]); if (ret < 0) exit(EXIT_FAILURE); - add_close_on_fork_list(afs_socket); - PARA_INFO_LOG("afs_socket: %d, afs_socket_cookie: %u\n", afs_socket, - (unsigned) afs_socket_cookie); + add_close_on_fork_list(afs_server_socket[0]); + PARA_INFO_LOG("afs_socket: %d, afs_socket_cookie: %u\n", + afs_server_socket[0], (unsigned) afs_socket_cookie); + return afs_server_socket[0]; } static void server_init(int argc, char **argv) { - /* connector's address information */ struct server_cmdline_parser_params params = { .override = 0, .initialize = 1, @@ -373,6 +492,9 @@ static void server_init(int argc, char **argv) .check_ambiguity = 0, .print_errors = 1 }; + int afs_socket; + + valid_fd_012(); init_random_seed(); /* parse command line options */ server_cmdline_parser_ext(argc, argv, &conf, ¶ms); @@ -390,29 +512,15 @@ static void server_init(int argc, char **argv) PARA_NOTICE_LOG("initializing audio format handlers\n"); afh_init(); mmd->server_pid = getpid(); - setup_signal_handling(); + init_signal_task(); PARA_NOTICE_LOG("initializing the audio file selector\n"); - init_afs(); + afs_socket = init_afs(); PARA_NOTICE_LOG("initializing virtual streaming system\n"); - vss_init(); - mmd_lock(); + init_vss_task(afs_socket); + init_server_command_task(argc, argv); PARA_NOTICE_LOG("server init complete\n"); } -/* - * called when server gets SIGHUP or when client invokes hup command. - */ -static void handle_sighup(void) -{ - PARA_NOTICE_LOG("SIGHUP\n"); - close_log(logfile); /* gets reopened if necessary by parse_config */ - logfile = NULL; - parse_config(1); /* reopens log */ - init_user_list(user_list_file); /* reload user list */ - if (afs_pid) - kill(afs_pid, SIGHUP); -} - static void status_refresh(void) { static int prev_uptime = -1, prev_events = -1; @@ -421,10 +529,12 @@ static void status_refresh(void) if (prev_events != mmd->events) goto out; if (mmd->new_vss_status_flags != mmd->vss_status_flags) - goto out; + goto out_inc_events; if (uptime / 60 != prev_uptime / 60) - goto out; - ret = 0; + goto out_inc_events; + return; +out_inc_events: + mmd->events++; out: prev_uptime = uptime; prev_events = mmd->events; @@ -448,118 +558,30 @@ static int server_select(int max_fileno, fd_set *readfds, fd_set *writefds, return ret; } -static void command_pre_select(int *max_fileno, fd_set *rfds, char *dummy_ptr) -{ - struct server_command_task *sct = container_of(dummy_ptr, struct server_command_task, dummy); - para_fd_set(sct->listen_fd, rfds, max_fileno); -} - -static void command_post_select(fd_set *rfds, char *dummy_ptr) -{ - struct server_command_task *sct = container_of(dummy_ptr, struct server_command_task, dummy); - - int new_fd, ret; - char *peer_name; - pid_t child_pid; - - if (!FD_ISSET(sct->listen_fd, rfds)) - return; - ret = para_accept(sct->listen_fd, NULL, 0); - if (ret < 0) - goto out; - new_fd = ret; - peer_name = remote_name(new_fd); - PARA_INFO_LOG("got connection from %s, forking\n", peer_name); - mmd->num_connects++; - mmd->active_connections++; - random(); - child_pid = fork(); - if (child_pid < 0) { - ret = -ERRNO_TO_PARA_ERROR(errno); - goto out; - } - if (child_pid) { - close(new_fd); - /* parent keeps accepting connections */ - return; - } - alarm(ALARM_TIMEOUT); - close_listed_fds(); - para_signal_shutdown(); - /* - * put info on who we are serving into argv[0] to make - * client ip visible in top/ps - */ -// for (i = argc - 1; i >= 0; i--) -// memset(argv[i], 0, strlen(argv[i])); -// sprintf(argv[0], "para_server (serving %s)", peer_name); - return handle_connect(new_fd, peer_name); -out: - if (ret < 0) - PARA_CRIT_LOG("%s\n", para_strerror(-ret)); -} - /** - * the main function of para_server - * - * \param argc usual argument count - * \param argv usual argument vector + * The main function of para_server. * - * \return EXIT_SUCCESS or EXIT_FAILURE + * \param argc Usual argument count. + * \param argv Usual argument vector. * + * \return EXIT_SUCCESS or EXIT_FAILURE. */ int main(int argc, char *argv[]) { - int max_fileno, ret; - fd_set rfds, wfds; - struct timeval *timeout; - struct server_command_task server_command_task_struct; - - - valid_fd_012(); + int ret; + static struct sched s = { + .default_timeout = { + .tv_sec = 1, + .tv_usec = 0 + }, + .select_function = server_select + }; server_init(argc, argv); - init_command_task(&server_command_task_struct); -repeat: - FD_ZERO(&rfds); - FD_ZERO(&wfds); - max_fileno = -1; - command_pre_select(&max_fileno, &rfds, &server_command_task_struct.dummy); - para_fd_set(signal_pipe, &rfds, &max_fileno); - timeout = vss_preselect(&rfds, &wfds, &max_fileno); - server_select(max_fileno + 1, &rfds, &wfds, timeout); - vss_post_select(&rfds, &wfds); - if (FD_ISSET(signal_pipe, &rfds)) { - int sig; - pid_t pid; - sig = para_next_signal(); - switch (sig) { - case SIGHUP: - handle_sighup(); - break; - case SIGCHLD: - for (;;) { - ret = para_reap_child(&pid); - if (ret <= 0) - break; - if (pid != afs_pid) - continue; - PARA_EMERG_LOG("fatal: afs died\n"); - goto genocide; - } - break; - /* die on sigint/sigterm. Kill all children too. */ - case SIGINT: - case SIGTERM: - PARA_EMERG_LOG("terminating on signal %d\n", sig); -genocide: - kill(0, SIGTERM); - mutex_destroy(mmd_mutex); - shm_detach(mmd); - shm_destroy(mmd_shm_id); - - exit(EXIT_FAILURE); - } + mmd_lock(); + ret = schedule(&s); + if (ret < 0) { + PARA_EMERG_LOG("%s\n", para_strerror(-ret)); + exit(EXIT_FAILURE); } - command_post_select(&rfds, &server_command_task_struct.dummy); - goto repeat; + exit(EXIT_SUCCESS); } diff --git a/server.h b/server.h index 73f27c78..c1e670ad 100644 --- a/server.h +++ b/server.h @@ -105,9 +105,6 @@ struct misc_meta_data { /** Command line options for para_server. */ extern struct server_args_info conf; -/** Socket for afs-server communication. */ -extern int afs_socket; - __noreturn void handle_connect(int fd, const char *peername); void mmd_unlock(void); void mmd_lock(void); diff --git a/vss.c b/vss.c index 42d39db0..3e19b134 100644 --- a/vss.c +++ b/vss.c @@ -11,9 +11,6 @@ * senders. */ -#include /* mmap */ -#include /* gettimeofday */ -#include #include #include "para.h" @@ -29,6 +26,7 @@ #include "send.h" #include "ipc.h" #include "fd.h" +#include "sched.h" extern struct misc_meta_data *mmd; @@ -36,7 +34,7 @@ extern void dccp_send_init(struct sender *); extern void http_send_init(struct sender *); extern void ortp_send_init(struct sender *); -/** the list of supported senders */ +/** The list of supported senders. */ struct sender senders[] = { { .name = "http", @@ -57,28 +55,30 @@ struct sender senders[] = { } }; -/** The possible states of the afs socket. See \ref afs_socket. */ +/** The possible states of the afs socket. */ enum afs_socket_status { /** Socket is inactive. */ AFS_SOCKET_READY, /** Socket fd was included in the write fd set for select(). */ AFS_SOCKET_CHECK_FOR_WRITE, - /** vss wrote a request to the socket and waits for afs to reply. */ + /** vss wrote a request to the socket and waits for reply from afs. */ AFS_SOCKET_AFD_PENDING }; - +/** The task structure for the virtual streaming system. */ struct vss_task { struct timeval announce_tv; struct timeval data_send_barrier; struct timeval eof_barrier; struct timeval autoplay_barrier; + int afs_socket; enum afs_socket_status afsss; char *map; + struct task task; + const char *header_buf; + size_t header_len; }; -static struct vss_task vss_task_struct, *vsst = &vss_task_struct; - /** * Check if vss status flag \a P (playing) is set. * @@ -136,61 +136,8 @@ unsigned int vss_stopped(void) && !(mmd->new_vss_status_flags & VSS_PLAYING); } -/** - * Get the data of the given chunk. - * - * \param chunk_num The number of the desired chunk. - * \param buf Chunk data. - * \param len Chunk length in bytes. - * - * \return Standard. - */ -int vss_get_chunk(long unsigned chunk_num, char **buf, size_t *len) -{ - if (!vsst->map || !vss_playing()) - return -E_CHUNK; - if (chunk_num >= mmd->afd.afhi.chunks_total) - return -E_CHUNK; - afh_get_chunk(chunk_num, &mmd->afd.afhi, vsst->map, buf, len); - return 1; -} - -/** - * Initialize the virtual streaming system. - * - * This also initializes all supported senders and starts streaming - * if the --autoplay command line flag was given. - */ -void vss_init(void) -{ - int i; - char *hn = para_hostname(), *home = para_homedir(); - long unsigned announce_time = conf.announce_time_arg > 0? - conf.announce_time_arg : 300, - autoplay_delay = conf.autoplay_delay_arg > 0? - conf.autoplay_delay_arg : 0; - ms2tv(announce_time, &vsst->announce_tv); - PARA_INFO_LOG("announce timeval: %lums\n", tv2ms(&vsst->announce_tv)); - for (i = 0; senders[i].name; i++) { - PARA_NOTICE_LOG("initializing %s sender\n", senders[i].name); - senders[i].init(&senders[i]); - } - free(hn); - free(home); - mmd->sender_cmd_data.cmd_num = -1; - if (conf.autoplay_given) { - struct timeval now, tmp; - mmd->vss_status_flags |= VSS_PLAYING; - mmd->new_vss_status_flags |= VSS_PLAYING; - gettimeofday(&now, NULL); - ms2tv(autoplay_delay, &tmp); - tv_add(&now, &tmp, &vsst->autoplay_barrier); - } -} - -static int chk_barrier(const char *bname, const struct timeval *now, - const struct timeval *barrier, struct timeval *diff, - int print_log) +static int chk_barrier(const char *bname, const struct timeval *barrier, + struct timeval *diff, int print_log) { long ms; @@ -206,10 +153,10 @@ static int chk_barrier(const char *bname, const struct timeval *now, * != NULL: timeout for next chunk * NULL: nothing to do */ -static struct timeval *vss_compute_timeout(void) +static struct timeval *vss_compute_timeout(struct vss_task *vsst) { static struct timeval the_timeout; - struct timeval now, next_chunk; + struct timeval next_chunk; if (vss_next() && vsst->map) { /* only sleep a bit, nec*/ @@ -217,20 +164,19 @@ static struct timeval *vss_compute_timeout(void) the_timeout.tv_usec = 100; return &the_timeout; } - gettimeofday(&now, NULL); - if (chk_barrier("autoplay_delay", &now, &vsst->autoplay_barrier, + if (chk_barrier("autoplay_delay", &vsst->autoplay_barrier, &the_timeout, 1) < 0) return &the_timeout; - if (chk_barrier("eof", &now, &vsst->eof_barrier, &the_timeout, 1) < 0) + if (chk_barrier("eof", &vsst->eof_barrier, &the_timeout, 1) < 0) return &the_timeout; - if (chk_barrier("data send", &now, &vsst->data_send_barrier, + if (chk_barrier("data send", &vsst->data_send_barrier, &the_timeout, 1) < 0) return &the_timeout; if (!vss_playing() || !vsst->map) return NULL; compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv, &mmd->stream_start, &next_chunk); - if (chk_barrier("chunk", &now, &next_chunk, &the_timeout, 0) < 0) + if (chk_barrier("chunk", &next_chunk, &the_timeout, 0) < 0) return &the_timeout; /* chunk is due or bof */ the_timeout.tv_sec = 0; @@ -238,22 +184,23 @@ static struct timeval *vss_compute_timeout(void) return &the_timeout; } -static void vss_eof(void) +static void vss_eof(struct vss_task *vsst) { - struct timeval now; char *tmp; + mmd->stream_start = *now; if (!vsst->map) return; if (mmd->new_vss_status_flags & VSS_NOMORE) mmd->new_vss_status_flags = VSS_NEXT; - gettimeofday(&now, NULL); - tv_add(&mmd->afd.afhi.eof_tv, &now, &vsst->eof_barrier); + tv_add(&mmd->afd.afhi.eof_tv, now, &vsst->eof_barrier); para_munmap(vsst->map, mmd->size); vsst->map = NULL; mmd->chunks_sent = 0; mmd->offset = 0; mmd->afd.afhi.seconds_total = 0; + mmd->afd.afhi.chunk_tv.tv_sec = 0; + mmd->afd.afhi.chunk_tv.tv_usec = 0; free(mmd->afd.afhi.chunk_table); mmd->afd.afhi.chunk_table = NULL; tmp = make_message("%s:\n%s:\n%s:\n", status_item_list[SI_AUDIO_FILE_INFO], @@ -267,19 +214,6 @@ static void vss_eof(void) mmd->events++; } -/** - * Get the header of the current audio file. - * - * \param buf The length of the header is stored here. - * \param len Points to a buffer containing the header on return. - * - * \sa afh_get_header. - */ -void vss_get_header(char **buf, size_t *len) -{ - afh_get_header(&mmd->afd.afhi, vsst->map, buf, len); -} - /** * Get the list of all supported audio formats. * @@ -300,12 +234,13 @@ const char *supported_audio_formats(void) */ struct timeval *vss_chunk_time(void) { - if (!vsst->map) + if (mmd->afd.afhi.chunk_tv.tv_sec == 0 && + mmd->afd.afhi.chunk_tv.tv_usec == 0) return NULL; return &mmd->afd.afhi.chunk_tv; } -static int need_to_request_new_audio_file(void) +static int need_to_request_new_audio_file(struct vss_task *vsst) { if (vsst->map) /* have audio file */ return 0; @@ -324,9 +259,8 @@ static int need_to_request_new_audio_file(void) * This function gets called from para_server to determine the timeout value * for its main select loop. * - * \param rfds The set of file descriptors to be checked for reading. - * \param wfds The set of file descriptors to be checked for writing. - * \param max_fileno The highest-numbered file descriptor. + * \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. @@ -335,50 +269,47 @@ static int need_to_request_new_audio_file(void) * - request a new audio file from afs, * - shutdown of all senders (stop/pause command), * - reposition the stream (ff/jmp command). - * - * \return A pointer to a struct timeval containing the timeout for the next - * chunk of data to be sent, or NULL if we're not sending right now. */ -struct timeval *vss_preselect(fd_set *rfds, fd_set *wfds, int *max_fileno) +static void vss_pre_select(struct sched *s, struct task *t) { int i; - struct timeval now; + struct timeval *tv, diff; + struct vss_task *vsst = container_of(t, struct vss_task, task); if (!vsst->map || vss_next() || vss_paused() || vss_repos()) for (i = 0; senders[i].name; i++) senders[i].shutdown_clients(); if (vss_next()) - vss_eof(); + vss_eof(vsst); else if (vss_paused()) { - if (mmd->chunks_sent) { - gettimeofday(&now, NULL); - tv_add(&mmd->afd.afhi.eof_tv, &now, &vsst->eof_barrier); - } + if (mmd->chunks_sent) + tv_add(&mmd->afd.afhi.eof_tv, now, &vsst->eof_barrier); mmd->chunks_sent = 0; } else if (vss_repos()) { - gettimeofday(&now, NULL); - tv_add(&now, &vsst->announce_tv, &vsst->data_send_barrier); - tv_add(&mmd->afd.afhi.eof_tv, &now, &vsst->eof_barrier); + tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier); + tv_add(&mmd->afd.afhi.eof_tv, now, &vsst->eof_barrier); mmd->chunks_sent = 0; mmd->current_chunk = mmd->repos_request; mmd->new_vss_status_flags &= ~VSS_REPOS; } - if (need_to_request_new_audio_file()) { + if (need_to_request_new_audio_file(vsst)) { PARA_DEBUG_LOG("ready and playing, but no audio file\n"); - para_fd_set(afs_socket, wfds, max_fileno); + para_fd_set(vsst->afs_socket, &s->wfds, &s->max_fileno); vsst->afsss = AFS_SOCKET_CHECK_FOR_WRITE; } else - para_fd_set(afs_socket, rfds, max_fileno); + para_fd_set(vsst->afs_socket, &s->rfds, &s->max_fileno); for (i = 0; senders[i].name; i++) { if (!senders[i].pre_select) continue; - senders[i].pre_select(max_fileno, rfds, wfds); + senders[i].pre_select(&s->max_fileno, &s->rfds, &s->wfds); } - return vss_compute_timeout(); + tv = vss_compute_timeout(vsst); + if (tv && tv_diff(tv, &s->timeout, &diff) < 0) + s->timeout = *tv; } -static int recv_afs_msg(int *fd, uint32_t *code, uint32_t *data) +static int recv_afs_msg(int afs_socket, int *fd, uint32_t *code, uint32_t *data) { char control[255], buf[8]; struct msghdr msg = {.msg_iov = NULL}; @@ -397,7 +328,6 @@ static int recv_afs_msg(int *fd, uint32_t *code, uint32_t *data) ret = recvmsg(afs_socket, &msg, 0); if (ret < 0) return -ERRNO_TO_PARA_ERROR(errno); - vsst->afsss = AFS_SOCKET_READY; if (iov.iov_len != sizeof(buf)) return -E_AFS_SHORT_READ; *code = *(uint32_t*)buf; @@ -413,15 +343,15 @@ static int recv_afs_msg(int *fd, uint32_t *code, uint32_t *data) return 1; } -static void recv_afs_result(void) +static void recv_afs_result(struct vss_task *vsst) { int ret, passed_fd, shmid; uint32_t afs_code = 0, afs_data = 0; struct stat statbuf; - struct timeval now; + vsst->afsss = AFS_SOCKET_READY; mmd->afd.afhi.chunk_table = NULL; - ret = recv_afs_msg(&passed_fd, &afs_code, &afs_data); + ret = recv_afs_msg(vsst->afs_socket, &passed_fd, &afs_code, &afs_data); if (ret < 0) goto err; PARA_DEBUG_LOG("fd: %d, code: %u, shmid: %u\n", passed_fd, afs_code, @@ -453,8 +383,9 @@ static void recv_afs_result(void) mmd->events++; mmd->num_played++; mmd->new_vss_status_flags &= (~VSS_NEXT); - gettimeofday(&now, NULL); - tv_add(&now, &vsst->announce_tv, &vsst->data_send_barrier); + tv_add(now, &vsst->announce_tv, &vsst->data_send_barrier); + afh_get_header(&mmd->afd.afhi, vsst->map, &vsst->header_buf, + &vsst->header_len); return; err: free(mmd->afd.afhi.chunk_table); @@ -473,23 +404,22 @@ err: * each supported sender's send() function which is supposed to send out the data * to all connected clients. */ -static void vss_send_chunk(void) +static void vss_send_chunk(struct vss_task *vsst) { int i; - struct timeval now, due; - char *buf; + struct timeval due; + const char *buf; size_t len; if (!vsst->map || !vss_playing()) return; - gettimeofday(&now, NULL); compute_chunk_time(mmd->chunks_sent, &mmd->afd.afhi.chunk_tv, &mmd->stream_start, &due); - if (tv_diff(&due, &now, NULL) > 0) + if (tv_diff(&due, now, NULL) > 0) return; - if (chk_barrier("eof", &now, &vsst->eof_barrier, &due, 1) < 0) + if (chk_barrier("eof", &vsst->eof_barrier, &due, 1) < 0) return; - if (chk_barrier("data send", &now, &vsst->data_send_barrier, + if (chk_barrier("data send", &vsst->data_send_barrier, &due, 1) < 0) return; mmd->new_vss_status_flags &= ~VSS_REPOS; @@ -503,43 +433,85 @@ static void vss_send_chunk(void) */ if (!mmd->chunks_sent) { struct timeval tmp; - gettimeofday(&mmd->stream_start, NULL); + mmd->stream_start = *now; tv_scale(mmd->current_chunk, &mmd->afd.afhi.chunk_tv, &tmp); mmd->offset = tv2ms(&tmp); mmd->events++; } afh_get_chunk(mmd->current_chunk, &mmd->afd.afhi, vsst->map, &buf, &len); for (i = 0; senders[i].name; i++) - senders[i].send(mmd->current_chunk, mmd->chunks_sent, buf, len); + senders[i].send(mmd->current_chunk, mmd->chunks_sent, buf, len, + vsst->header_buf, vsst->header_len); mmd->new_vss_status_flags |= VSS_PLAYING; mmd->chunks_sent++; mmd->current_chunk++; } -void vss_post_select(fd_set *rfds, fd_set *wfds) +static void vss_post_select(struct sched *s, struct task *t) { int ret, i; + struct vss_task *vsst = container_of(t, struct vss_task, task); if (mmd->sender_cmd_data.cmd_num >= 0) { int num = mmd->sender_cmd_data.cmd_num, - s = mmd->sender_cmd_data.sender_num; + sender_num = mmd->sender_cmd_data.sender_num; - if (senders[s].client_cmds[num]) - senders[s].client_cmds[num](&mmd->sender_cmd_data); + if (senders[sender_num].client_cmds[num]) + senders[sender_num].client_cmds[num](&mmd->sender_cmd_data); mmd->sender_cmd_data.cmd_num = -1; } if (vsst->afsss != AFS_SOCKET_CHECK_FOR_WRITE) { - if (FD_ISSET(afs_socket, rfds)) - recv_afs_result(); - } else if (FD_ISSET(afs_socket, wfds)) { + if (FD_ISSET(vsst->afs_socket, &s->rfds)) + recv_afs_result(vsst); + } else if (FD_ISSET(vsst->afs_socket, &s->wfds)) { PARA_NOTICE_LOG("requesting new fd from afs\n"); - ret = send_buffer(afs_socket, "new"); + ret = send_buffer(vsst->afs_socket, "new"); vsst->afsss = AFS_SOCKET_AFD_PENDING; } for (i = 0; senders[i].name; i++) { if (!senders[i].post_select) continue; - senders[i].post_select(rfds, wfds); + senders[i].post_select(&s->rfds, &s->wfds); + } + vss_send_chunk(vsst); +} + +/** + * Initialize the virtual streaming system task. + * + * \param afs_socket The fd for communication with afs. + * + * This also initializes all supported senders and starts streaming + * if the --autoplay command line flag was given. + */ +void init_vss_task(int afs_socket) +{ + static struct vss_task vss_task_struct, *vsst = &vss_task_struct; + int i; + char *hn = para_hostname(), *home = para_homedir(); + long unsigned announce_time = conf.announce_time_arg > 0? + conf.announce_time_arg : 300, + autoplay_delay = conf.autoplay_delay_arg > 0? + conf.autoplay_delay_arg : 0; + + 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)); + for (i = 0; senders[i].name; i++) { + PARA_NOTICE_LOG("initializing %s sender\n", senders[i].name); + senders[i].init(&senders[i]); + } + free(hn); + free(home); + mmd->sender_cmd_data.cmd_num = -1; + if (conf.autoplay_given) { + struct timeval tmp; + mmd->vss_status_flags |= VSS_PLAYING; + mmd->new_vss_status_flags |= VSS_PLAYING; + ms2tv(autoplay_delay, &tmp); + tv_add(now, &tmp, &vsst->autoplay_barrier); } - vss_send_chunk(); + register_task(&vsst->task); } diff --git a/vss.h b/vss.h index 98af4220..f92fc809 100644 --- a/vss.h +++ b/vss.h @@ -4,27 +4,24 @@ * Licensed under the GPL v2. For licencing details see COPYING. */ -/** \file vss.h exported functions from vss.c (para_server) */ -void vss_init(void); -struct timeval *vss_preselect(fd_set *rfds, fd_set *wfds, int *max_fileno); -void vss_post_select(fd_set *rfds, fd_set *wfds); +/** \file vss.h Exported functions from vss.c (para_server). */ + +void init_vss_task(int afs_socket); unsigned int vss_playing(void); unsigned int vss_next(void); unsigned int vss_repos(void); unsigned int vss_paused(void); unsigned int vss_stopped(void); -void vss_get_header(char **buf, size_t *len); struct timeval *vss_chunk_time(void); const char *supported_audio_formats(void); -int vss_get_chunk(long unsigned chunk_num, char **buf, size_t *len); -/** stop playing after current audio file */ +/** Stop playing after current audio file. */ #define VSS_NOMORE 1 -/** skip remaining part of current audio file */ +/** Skip remaining part of current audio file. */ #define VSS_NEXT 2 -/** a reposition request was sent by a client */ +/** A reposition request was sent by a client. */ #define VSS_REPOS 4 -/** currently playing */ +/** Currently playing. */ #define VSS_PLAYING 8 -/** a client requested to change the audio file selector */ +/** A client requested to change the audio file selector. */ #define VSS_CHANGE 16